fix: 解决 compact 命令断连问题 - 请求体超限413清晰错误 + /v2/compact分批压缩端点

根因: APIG请求体限制~1.25MB, 超限后JSON截断导致'model id缺失'400错误→客户端断连

修复:
1. 请求体超限校验: 返回413 + content_too_large清晰错误(而非上游的'model id缺失')
2. /v2/compact端点: 超长对话自动分批发送压缩,每批<800KB,最后合并总结
3. 动态超时: 根据请求体大小自动调整(60s→180s→300s)
4. 修复响应头转发: 剥离Content-Encoding避免客户端解压失败
5. 禁用flask-compress(与Waitress SSE兼容问题)

ai/huawei_gateway.py

@@ -14,6 +14,7 @@ import os
 import re
 import sys
 import time
+import gzip
 import logging
 import threading
 import traceback
@@ -42,6 +43,12 @@ MAX_MEM_SEGMENT = 200 * 1024 * 1024  # 单段最大扫描 200MB
 TOKEN_PATTERN = re.compile(b'Bearer ([A-Za-z0-9+/=_-]{100,})')
 TARGET_HOST = 'tokenhub.developer.huaweicloud.com'
 
+# ================= 请求体限制配置 =================
+APIG_BODY_LIMIT = 1200 * 1024    # APIG 请求体限制 ~1.2MB (实测边界1260KB)
+GZIP_THRESHOLD = 200 * 1024      # 超过 200KB 时启用 gzip 压缩转发
+UPSTREAM_TIMEOUT_MIN = 60        # 小请求超时 60s
+UPSTREAM_TIMEOUT_MAX = 300       # 大请求超时 300s
+
 # ================= 日志 =================
 logging.basicConfig(
     level=logging.INFO,
@@ -248,13 +255,12 @@ http_session.mount('http://', adapter)
 # ================= Flask 应用 =================
 app = Flask(__name__)
 
-# 启用响应压缩（gzip/brotli/zstd），对长文本响应压缩率可达 70-80%
+# 响应压缩（flask-compress 与 Waitress SSE 存在兼容问题，暂时禁用）
-if _compress_available:
+# 如需启用，需切换到 Gunicorn 或确保客户端不发送 Accept-Encoding
+if False and _compress_available:
     compress = Compress()
     compress.init_app(app)
-    # 设置压缩最小阈值（256字节以下不压缩）
     app.config['COMPRESS_MIN_SIZE'] = 256
-    # 压缩 MIME 类型白名单
     app.config['COMPRESS_MIMETYPES'] = [
         'text/html', 'text/plain', 'text/css', 'text/xml',
         'application/json', 'application/javascript',
@@ -291,6 +297,276 @@ def set_token():
     return {"status": "ok", "token_fingerprint": cache._fingerprint(token)}, 200
 
 
+# ================= Compact 分批压缩端点 =================
+@app.route('/v2/compact', methods=['POST', 'OPTIONS'])
+def compact_endpoint():
+    """
+    分批上下文压缩端点：
+    - 对话历史超长时，自动分批发送给模型压缩
+    - 每批独立总结，最后合并为完整压缩上下文
+    - 兼容 OpenAI chat completions 请求格式
+    """
+    if request.method == 'OPTIONS':
+        return Response(status=200, headers={
+            'Access-Control-Allow-Origin': '*',
+            'Access-Control-Allow-Methods': 'POST, OPTIONS',
+            'Access-Control-Allow-Headers': 'Content-Type, Authorization'
+        })
+
+    import json as _json
+
+    real_token = find_token_in_memory()
+    if not real_token:
+        return {"error": {"message": "未找到华为云Token", "type": "server_error"}}, 500
+
+    try:
+        payload = request.get_json(force=True)
+    except Exception:
+        return {"error": {"message": "无效的JSON请求体", "type": "invalid_request_error"}}, 400
+
+    model = payload.get('model', 'glm-5.1')
+    messages = payload.get('messages', [])
+    max_tokens = payload.get('max_tokens', 500)
+    stream = payload.get('stream', False)
+
+    if not messages:
+        return {"error": {"message": "messages 不能为空", "type": "invalid_request_error"}}, 400
+
+    # 计算实际请求体大小（使用原始请求体，而非重新序列化）
+    raw_body = request.get_data()
+    body_size = len(raw_body)
+
+    # 如果请求体在限制内，直接转发给上游（无需分批）
+    if body_size <= APIG_BODY_LIMIT:
+        logger.info(f"compact: 请求体{body_size//1024}KB在限制内，直接转发")
+        target_url = f'https://{TARGET_HOST}/v2/chat/completions'
+        headers = {
+            'Authorization': f'Bearer {real_token}',
+            'Host': TARGET_HOST,
+            'Content-Type': 'application/json',
+        }
+        # 保留客户端的其他头
+        for k, v in request.headers:
+            kl = k.lower()
+            if kl not in ('host', 'content-length', 'connection', 'accept-encoding',
+                          'transfer-encoding', 'authorization', 'content-type'):
+                headers[k] = v
+
+        try:
+            resp = http_session.request(
+                method='POST', url=target_url, headers=headers,
+                data=raw_body, allow_redirects=False,
+                timeout=UPSTREAM_TIMEOUT_MAX, stream=True
+            )
+        except Exception as e:
+            logger.error(f"compact 转发失败: {e}")
+            return {"error": {"message": f"转发失败: {str(e)}", "type": "server_error"}}, 502
+
+        # 流式转发
+        content_type = resp.headers.get('Content-Type', '')
+        if 'text/event-stream' in content_type or stream:
+            skip_h = {'transfer-encoding', 'content-encoding', 'content-length',
+                      'connection', 'keep-alive', 'upgrade'}
+            resp_headers = [(k, v) for k, v in resp.headers.items() if k.lower() not in skip_h]
+
+            def sse_stream():
+                try:
+                    for chunk in resp.iter_content(chunk_size=16384):
+                        if chunk:
+                            yield chunk
+                finally:
+                    resp.close()
+            return Response(sse_stream(), status=resp.status_code,
+                            headers=resp_headers, direct_passthrough=True)
+        else:
+            content = resp.content
+            resp.close()
+            skip_h2 = {'transfer-encoding', 'content-encoding', 'content-length',
+                       'connection', 'keep-alive', 'upgrade'}
+            return Response(content, status=resp.status_code,
+                            headers=[(k, v) for k, v in resp.headers.items()
+                                     if k.lower() not in skip_h2])
+
+    # ============ 分批压缩逻辑 ============
+    logger.info(f"compact: 请求体{body_size//1024}KB超限，启动分批压缩")
+
+    # 分离 system 消息和对话消息
+    system_msgs = [m for m in messages if m.get('role') == 'system']
+    convo_msgs = [m for m in messages if m.get('role') != 'system']
+
+    if not convo_msgs:
+        return {"error": {"message": "无对话内容可压缩", "type": "invalid_request_error"}}, 400
+
+    # 按批次分割对话：每批控制在安全大小内
+    SAFE_BATCH_BYTES = 800 * 1024  # 每批 800KB（留余量给 JSON 开销）
+    batches = []
+    current_batch = []
+    current_size = 0
+
+    for msg in convo_msgs:
+        msg_size = len(_json.dumps(msg, ensure_ascii=False).encode('utf-8'))
+        if current_size + msg_size > SAFE_BATCH_BYTES and current_batch:
+            batches.append(current_batch)
+            current_batch = [msg]
+            current_size = msg_size
+        else:
+            current_batch.append(msg)
+            current_size += msg_size
+
+    if current_batch:
+        batches.append(current_batch)
+
+    logger.info(f"compact: 分为{len(batches)}批, 每批~{SAFE_BATCH_BYTES//1024}KB")
+
+    # 逐批压缩
+    target_url = f'https://{TARGET_HOST}/v2/chat/completions'
+    summaries = []
+
+    for i, batch in enumerate(batches):
+        # 构建压缩请求
+        batch_messages = system_msgs + batch + [{
+            "role": "user",
+            "content": "请对以上对话内容进行简洁压缩总结，保留所有关键信息、决策和结论，去除冗余和重复。用简洁的条目式格式输出。"
+        }]
+
+        batch_payload = {
+            "model": model,
+            "messages": batch_messages,
+            "max_tokens": max_tokens,
+            "stream": False,
+            "temperature": 0.3
+        }
+
+        batch_body = _json.dumps(batch_payload, ensure_ascii=False).encode('utf-8')
+
+        headers = {
+            'Authorization': f'Bearer {real_token}',
+            'Host': TARGET_HOST,
+            'Content-Type': 'application/json',
+        }
+
+        try:
+            resp = http_session.request(
+                method='POST', url=target_url, headers=headers,
+                data=batch_body, allow_redirects=False,
+                timeout=UPSTREAM_TIMEOUT_MAX, stream=False
+            )
+            data = resp.json()
+
+            if resp.status_code == 200 and 'choices' in data:
+                summary = data['choices'][0].get('message', {}).get('content', '')
+                summaries.append(summary)
+                logger.info(f"compact: 第{i+1}/{len(batches)}批压缩完成, {len(summary)}字")
+            else:
+                err_msg = data.get('error', {}).get('message', str(data)[:200])
+                logger.warning(f"compact: 第{i+1}批失败: HTTP {resp.status_code} - {err_msg}")
+                # 失败的批次保留原始内容摘要
+                fallback = f"[第{i+1}批压缩失败，原始{len(batch)}条消息]"
+                summaries.append(fallback)
+        except Exception as e:
+            logger.error(f"compact: 第{i+1}批异常: {e}")
+            summaries.append(f"[第{i+1}批压缩异常]")
+
+        time.sleep(1)  # 避免触发限流
+
+    # 合并所有批次的压缩结果
+    if len(summaries) == 1:
+        final_summary = summaries[0]
+    else:
+        # 对多个摘要做最终合并压缩
+        merge_messages = system_msgs + [{
+            "role": "user",
+            "content": "以下是分批压缩的对话摘要，请合并为一个连贯的压缩总结，保留所有关键信息：\n\n" +
+                       "\n\n---\n\n".join(f"第{i+1}批摘要：\n{s}" for i, s in enumerate(summaries))
+        }]
+
+        merge_payload = {
+            "model": model,
+            "messages": merge_messages,
+            "max_tokens": max_tokens,
+            "stream": False,
+            "temperature": 0.3
+        }
+
+        try:
+            resp = http_session.request(
+                method='POST', url=target_url, headers=headers,
+                data=_json.dumps(merge_payload, ensure_ascii=False).encode('utf-8'),
+                allow_redirects=False, timeout=UPSTREAM_TIMEOUT_MAX
+            )
+            merge_data = resp.json()
+            if resp.status_code == 200 and 'choices' in merge_data:
+                final_summary = merge_data['choices'][0].get('message', {}).get('content', '')
+            else:
+                final_summary = "\n".join(summaries)
+        except Exception:
+            final_summary = "\n".join(summaries)
+
+    logger.info(f"compact: 压缩完成, 最终{len(final_summary)}字 (原始{body_size//1024}KB)")
+
+    # 返回 OpenAI 兼容格式
+    if stream:
+        # 流式返回：将压缩结果包装为 SSE 事件
+        import uuid
+        chat_id = str(uuid.uuid4())
+        created = int(time.time())
+
+        def compact_sse():
+            # 首个 chunk：role
+            first = {
+                "id": chat_id, "object": "chat.completion.chunk", "created": created,
+                "model": model,
+                "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]
+            }
+            yield f"data: {_json.dumps(first, ensure_ascii=False)}\n\n"
+
+            # 内容 chunk
+            content_chunk = {
+                "id": chat_id, "object": "chat.completion.chunk", "created": created,
+                "model": model,
+                "choices": [{"index": 0, "delta": {"content": final_summary}, "finish_reason": None}]
+            }
+            yield f"data: {_json.dumps(content_chunk, ensure_ascii=False)}\n\n"
+
+            # 结束 chunk
+            done_chunk = {
+                "id": chat_id, "object": "chat.completion.chunk", "created": created,
+                "model": model,
+                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]
+            }
+            yield f"data: {_json.dumps(done_chunk, ensure_ascii=False)}\n\n"
+            yield "data: [DONE]\n\n"
+
+        return Response(compact_sse(), status=200, headers={
+            'Content-Type': 'text/event-stream;charset=UTF-8',
+            'Cache-Control': 'no-cache',
+            'X-Compact-Batches': str(len(batches)),
+            'X-Compact-Original-KB': str(body_size // 1024),
+        }, direct_passthrough=True)
+    else:
+        # 非流式返回
+        return {
+            "id": f"compact-{int(time.time())}",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "model": model,
+            "choices": [{
+                "index": 0,
+                "message": {"role": "assistant", "content": final_summary},
+                "finish_reason": "stop"
+            }],
+            "usage": {
+                "prompt_tokens": body_size // 4,  # 估算
+                "completion_tokens": len(final_summary),
+                "total_tokens": body_size // 4 + len(final_summary)
+            },
+            "compact_meta": {
+                "batches": len(batches),
+                "original_kb": body_size // 1024,
+            }
+        }
+
+
 @app.route('/v2/<path:subpath>', methods=['POST', 'GET', 'OPTIONS', 'PUT', 'DELETE'])
 def proxy(subpath):
     if request.method == 'OPTIONS':
@@ -317,16 +593,40 @@ def proxy(subpath):
 
     target_url = f'https://{TARGET_HOST}/v2/{subpath}'
 
+    # ============ 请求体处理：大小校验 + 动态超时 ============
+    raw_body = request.get_data()
+    body_size = len(raw_body)
+
+    # 动态超时：根据请求体大小自动调整
+    if body_size > 800 * 1024:
+        upstream_timeout = UPSTREAM_TIMEOUT_MAX
+    elif body_size > 200 * 1024:
+        upstream_timeout = 180
+    else:
+        upstream_timeout = UPSTREAM_TIMEOUT_MIN
+
+    # 请求体超限校验：返回清晰错误而非上游的 "model id 缺失"
+    if body_size > APIG_BODY_LIMIT:
+        logger.warning(f"请求体超限: {body_size//1024}KB > {APIG_BODY_LIMIT//1024}KB (APIG限制)")
+        return {
+            "error": {
+                "message": f"请求体过大({body_size//1024}KB)，超过API网关限制({APIG_BODY_LIMIT//1024}KB)。请减少对话历史长度，或使用 /v2/compact 端点进行分批压缩。",
+                "type": "invalid_request_error",
+                "code": "content_too_large",
+                "param": None
+            }
+        }, 413
+
     try:
         # 使用 stream=True 支持 SSE 流式转发
         resp = http_session.request(
             method=request.method,
             url=target_url,
             headers=headers,
-            data=request.get_data(),
+            data=raw_body,
             cookies=request.cookies,
             allow_redirects=False,
-            timeout=60,
+            timeout=upstream_timeout,
             stream=True
         )
 
@@ -342,10 +642,10 @@ def proxy(subpath):
                     method=request.method,
                     url=target_url,
                     headers=headers,
-                    data=request.get_data(),
+                    data=raw_body,
                     cookies=request.cookies,
                     allow_redirects=False,
-                    timeout=60,
+                    timeout=upstream_timeout,
                     stream=True
                 )