diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/ai/huawei_gateway.py b/ai/huawei_gateway.py index 7dbef8c..23a7d33 100755 --- a/ai/huawei_gateway.py +++ b/ai/huawei_gateway.py @@ -267,6 +267,191 @@ if False and _compress_available: 'application/xml', 'text/event-stream', ] + +# ================= 自动分批压缩（透明，客户端无感知） ================= +def auto_compact(raw_body, real_token, orig_request): + """ + 当 chat/completions 请求体超过 APIG 限制时， + 自动分批压缩对话历史，返回标准 OpenAI 格式响应。 + 对客户端完全透明。 + """ + import json as _json + import uuid + + try: + payload = _json.loads(raw_body) + 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) + body_size = len(raw_body) + + if not messages: + return {"error": {"message": "messages 不能为空", "type": "invalid_request_error"}}, 400 + + # 分离 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 + 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) + + num_batches = len(batches) + logger.info(f"auto_compact: {body_size//1024}KB → {num_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"auto_compact: 第{i+1}/{num_batches}批完成, {len(summary)}字") + else: + err_msg = data.get('error', {}).get('message', str(data)[:200]) + logger.warning(f"auto_compact: 第{i+1}批失败: HTTP {resp.status_code} - {err_msg}") + summaries.append(f"[第{i+1}批压缩失败，原始{len(batch)}条消息]") + except Exception as e: + logger.error(f"auto_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"auto_compact: 完成, {body_size//1024}KB → {len(final_summary)}字 ({num_batches}批)") + + # 返回标准 OpenAI 格式（客户端无感知） + if stream: + chat_id = str(uuid.uuid4()) + created = int(time.time()) + + def compact_sse(): + 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" + + 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" + + 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-Auto-Compact': f'batches={num_batches},original_kb={body_size//1024}', + }) + else: + return { + "id": f"chatcmpl-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) + } + } + + @app.route('/health') def health(): """健康检查端点""" @@ -542,7 +727,7 @@ def compact_endpoint(): 'Cache-Control': 'no-cache', 'X-Compact-Batches': str(len(batches)), 'X-Compact-Original-KB': str(body_size // 1024), - }, direct_passthrough=True) + }) else: # 非流式返回 return { @@ -567,6 +752,7 @@ def compact_endpoint(): } +@app.route('/v1/', methods=['POST', 'GET', 'OPTIONS', 'PUT', 'DELETE']) @app.route('/v2/', methods=['POST', 'GET', 'OPTIONS', 'PUT', 'DELETE']) def proxy(subpath): if request.method == 'OPTIONS': @@ -605,12 +791,17 @@ def proxy(subpath): else: upstream_timeout = UPSTREAM_TIMEOUT_MIN - # 请求体超限校验：返回清晰错误而非上游的 "model id 缺失" + # ============ 超限自动分批压缩（仅对 chat/completions） ============ + if body_size > APIG_BODY_LIMIT and subpath in ('chat/completions', 'chat/completions/'): + logger.info(f"chat/completions 请求体超限: {body_size//1024}KB > {APIG_BODY_LIMIT//1024}KB, 自动触发分批压缩") + return auto_compact(raw_body, real_token, request) + + # 非 chat/completions 超限，返回清晰错误 if body_size > APIG_BODY_LIMIT: - logger.warning(f"请求体超限: {body_size//1024}KB > {APIG_BODY_LIMIT//1024}KB (APIG限制)") + logger.warning(f"非chat请求体超限: {body_size//1024}KB > {APIG_BODY_LIMIT//1024}KB") return { "error": { - "message": f"请求体过大({body_size//1024}KB)，超过API网关限制({APIG_BODY_LIMIT//1024}KB)。请减少对话历史长度，或使用 /v2/compact 端点进行分批压缩。", + "message": f"请求体过大({body_size//1024}KB)，超过API网关限制({APIG_BODY_LIMIT//1024}KB)。请减少请求内容长度。", "type": "invalid_request_error", "code": "content_too_large", "param": None