From 61bda8db7506c4150468dc527704a8f0762fdcb8 Mon Sep 17 00:00:00 2001
From: chaos <chaos@nomsg.cn>
Date: Wed, 22 Jul 2026 14:35:39 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20Anthropic=20/v1/me?=
 =?UTF-8?q?ssages=20=E6=8E=A5=E5=8F=A3=E9=80=82=E9=85=8D=E5=B1=82?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- anthropic_request_to_openai(): Anthropic 请求转 OpenAI 格式
  - system 提取到 messages 首条
  - content blocks 展开 (text/tool_use/tool_result)
  - tools 格式转换 (input_schema → parameters)
  - max_tokens/temperature/stop_sequences 等参数映射

- openai_response_to_anthropic(): OpenAI 响应转 Anthropic 格式
  - content 数组构建 (text/tool_use)
  - stop_reason 映射 (stop→end_turn, tool_calls→tool_use, length→max_tokens)
  - usage 字段映射 (input_tokens/output_tokens)

- openai_stream_to_anthropic_stream(): 流式 SSE 转换生成器
  - message_start/content_block_start/content_block_delta/content_block_stop
  - message_delta/message_stop 事件序列
  - 支持 text_delta 和 input_json_delta
  - 正确的 block index 分配

- /v1/messages 路由: 复用现有 token 获取/超时/重试逻辑
  - 非流式: 请求转换→上游转发→响应转换
  - 流式: 请求转换→上游流式转发→SSE 事件转换

测试通过: 非流式/流式文本、system prompt、tool_use、tool_result 多轮对话
---
 ai/huawei_gateway.py | 542 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 542 insertions(+)

diff --git a/ai/huawei_gateway.py b/ai/huawei_gateway.py
index 642df36..74204a3 100755
--- a/ai/huawei_gateway.py
+++ b/ai/huawei_gateway.py
@@ -452,6 +452,14 @@ def _maybe_trim_context(raw_body):
     target_limit = int((MAX_CONTEXT_TOKENS - RESPONSE_BUDGET) * TRIM_RATIO)
 
     if total_tokens <= target_limit:
+        logger.info(f"上下文估算: ~{total_tokens} tokens (≤{target_limit}), 无需截断 | model={body.get('model','?')} | max_tokens={body.get('max_tokens','?')} | stream={body.get('stream', False)} | messages={len(messages)}")
+        # 调试: 超过200KB的请求dump到文件分析
+        if len(raw_body) > 200 * 1024:
+            import os
+            dump_path = f'/tmp/gateway_request_{int(time.time())}.json'
+            with open(dump_path, 'wb') as f:
+                f.write(raw_body)
+            logger.info(f"请求已dump到 {dump_path} ({len(raw_body)//1024}KB)")
         return raw_body, None  # 无需截断
 
     original_count = len(messages)
@@ -478,6 +486,380 @@ def _maybe_trim_context(raw_body):
     return new_body, info
 
 
+# ================= Anthropic API 适配层 (/v1/messages) =================
+
+def _anthropic_request_to_openai(body):
+    """Anthropic /v1/messages 请求 → OpenAI /v1/chat/completions 请求
+
+    转换内容:
+    - system (top-level) → messages[0] role=system
+    - content blocks: text/image/tool_use/tool_result → OpenAI 格式
+    - tools: input_schema → parameters
+    - tool_choice: auto/any/tool → auto/required/function
+    - stop_sequences → stop
+    """
+    openai_body = {}
+    openai_body['model'] = body.get('model', '')
+
+    openai_messages = []
+
+    # 1. system → system message
+    system = body.get('system')
+    if system:
+        if isinstance(system, str):
+            openai_messages.append({'role': 'system', 'content': system})
+        elif isinstance(system, list):
+            parts = [b.get('text', '') for b in system
+                     if isinstance(b, dict) and b.get('type') == 'text']
+            if parts:
+                openai_messages.append({'role': 'system', 'content': '\n'.join(parts)})
+
+    # 2. 转换 messages
+    for msg in body.get('messages', []):
+        role = msg.get('role', 'user')
+        content = msg.get('content')
+
+        if isinstance(content, str):
+            openai_messages.append({'role': role, 'content': content})
+            continue
+
+        if not isinstance(content, list):
+            openai_messages.append({'role': role, 'content': str(content) if content else ''})
+            continue
+
+        # Content blocks 数组
+        text_parts = []
+        tool_calls = []
+        tool_results = []
+        has_image = False
+        multi_content = []
+
+        for block in content:
+            if not isinstance(block, dict):
+                continue
+            btype = block.get('type')
+
+            if btype == 'text':
+                text_parts.append(block.get('text', ''))
+                multi_content.append({'type': 'text', 'text': block.get('text', '')})
+            elif btype == 'image':
+                has_image = True
+                source = block.get('source', {})
+                if source.get('type') == 'base64':
+                    mt = source.get('media_type', 'image/png')
+                    multi_content.append({
+                        'type': 'image_url',
+                        'image_url': {'url': f'data:{mt};base64,{source.get("data", "")}'}
+                    })
+            elif btype == 'tool_use':
+                tool_calls.append({
+                    'id': block.get('id', ''),
+                    'type': 'function',
+                    'function': {
+                        'name': block.get('name', ''),
+                        'arguments': json.dumps(block.get('input', {}), ensure_ascii=False)
+                    }
+                })
+            elif btype == 'tool_result':
+                tool_results.append(block)
+
+        # tool_result → 独立的 tool 消息
+        if tool_results:
+            for tr in tool_results:
+                tr_content = tr.get('content', '')
+                if isinstance(tr_content, list):
+                    tr_text = '\n'.join(
+                        b.get('text', '') for b in tr_content
+                        if isinstance(b, dict) and b.get('type') == 'text'
+                    )
+                else:
+                    tr_text = str(tr_content) if tr_content else ''
+                openai_messages.append({
+                    'role': 'tool',
+                    'tool_call_id': tr.get('tool_use_id', ''),
+                    'content': tr_text
+                })
+            continue
+
+        # 构建 assistant/user 消息
+        msg_dict = {'role': role}
+        if has_image:
+            msg_dict['content'] = multi_content
+        else:
+            msg_dict['content'] = '\n'.join(text_parts) if text_parts else ''
+
+        if tool_calls:
+            msg_dict['tool_calls'] = tool_calls
+            if not msg_dict.get('content'):
+                msg_dict['content'] = None
+
+        openai_messages.append(msg_dict)
+
+    openai_body['messages'] = openai_messages
+
+    # 3. 参数映射
+    openai_body['max_tokens'] = body.get('max_tokens', 4096)
+    if 'temperature' in body:
+        openai_body['temperature'] = body['temperature']
+    if 'top_p' in body:
+        openai_body['top_p'] = body['top_p']
+    if 'stop_sequences' in body:
+        openai_body['stop'] = body['stop_sequences']
+    if body.get('stream'):
+        openai_body['stream'] = True
+
+    # metadata.user_id → user
+    if body.get('metadata', {}).get('user_id'):
+        openai_body['user'] = body['metadata']['user_id']
+
+    # 4. tools 转换
+    if body.get('tools'):
+        openai_body['tools'] = [{
+            'type': 'function',
+            'function': {
+                'name': t.get('name', ''),
+                'description': t.get('description', ''),
+                'parameters': t.get('input_schema', {'type': 'object', 'properties': {}})
+            }
+        } for t in body['tools']]
+
+    # 5. tool_choice 转换
+    tc = body.get('tool_choice')
+    if tc and isinstance(tc, dict):
+        tct = tc.get('type', 'auto')
+        if tct == 'auto':
+            openai_body['tool_choice'] = 'auto'
+        elif tct == 'any':
+            openai_body['tool_choice'] = 'required'
+        elif tct == 'tool':
+            openai_body['tool_choice'] = {
+                'type': 'function',
+                'function': {'name': tc.get('name', '')}
+            }
+
+    return openai_body
+
+
+def _openai_response_to_anthropic(resp_json, model):
+    """OpenAI 非流式响应 → Anthropic 响应格式
+
+    转换内容:
+    - choices[0].message.content → content[{type:text}]
+    - choices[0].message.tool_calls → content[{type:tool_use}]
+    - finish_reason → stop_reason (stop→end_turn, length→max_tokens, tool_calls→tool_use)
+    - usage.prompt_tokens → usage.input_tokens, usage.completion_tokens → usage.output_tokens
+    """
+    choices = resp_json.get('choices', [])
+    choice = choices[0] if choices else {}
+    message = choice.get('message', {})
+
+    # 构建 content 数组
+    content_blocks = []
+
+    # 文本内容
+    text = message.get('content')
+    if text:
+        content_blocks.append({'type': 'text', 'text': text})
+
+    # tool_calls → tool_use blocks
+    for tc in message.get('tool_calls', []):
+        func = tc.get('function', {})
+        try:
+            input_data = json.loads(func.get('arguments', '{}'))
+        except json.JSONDecodeError:
+            input_data = {}
+        content_blocks.append({
+            'type': 'tool_use',
+            'id': tc.get('id', ''),
+            'name': func.get('name', ''),
+            'input': input_data
+        })
+
+    if not content_blocks:
+        content_blocks.append({'type': 'text', 'text': ''})
+
+    # stop_reason 映射
+    fr_map = {
+        'stop': 'end_turn',
+        'length': 'max_tokens',
+        'tool_calls': 'tool_use',
+        'content_filter': 'end_turn',
+    }
+    stop_reason = fr_map.get(choice.get('finish_reason'), 'end_turn')
+
+    # usage 映射
+    usage = resp_json.get('usage', {})
+
+    return {
+        'id': 'msg_' + resp_json.get('id', str(int(time.time() * 1000))),
+        'type': 'message',
+        'role': 'assistant',
+        'model': model,
+        'content': content_blocks,
+        'stop_reason': stop_reason,
+        'stop_sequence': None,
+        'usage': {
+            'input_tokens': usage.get('prompt_tokens', 0),
+            'output_tokens': usage.get('completion_tokens', 0),
+        }
+    }
+
+
+def _openai_sse_to_anthropic_sse(resp, model):
+    """OpenAI SSE 流 → Anthropic SSE 流 (生成器，yield bytes)
+
+    事件序列:
+    message_start → content_block_start → content_block_delta* → content_block_stop
+    → (更多 content blocks...) → message_delta → message_stop
+    """
+    def sse(event_type, data):
+        return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode('utf-8')
+
+    msg_id = None
+    message_started = False
+    text_block_open = False
+    text_block_index = -1
+    next_block_index = 0       # 下一个 content block 的索引
+    tool_map = {}              # openai_tool_index -> anthropic_block_index
+    finish_reason = None
+    output_tokens = 0
+    input_tokens = 0
+    done = False
+    buffer = ''
+
+    for chunk in resp.iter_content(chunk_size=SSE_CHUNK_SIZE):
+        if not chunk:
+            continue
+        buffer += chunk.decode('utf-8', errors='replace')
+
+        while '\n' in buffer:
+            line, buffer = buffer.split('\n', 1)
+            line = line.strip()
+            if not line or not line.startswith('data:'):
+                continue
+
+            data_str = line[5:].strip()
+
+            if data_str == '[DONE]':
+                done = True
+                # 关闭未关闭的 content blocks
+                if text_block_open:
+                    yield sse('content_block_stop', {'type': 'content_block_stop', 'index': text_block_index})
+                    text_block_open = False
+                for idx in sorted(tool_map.values()):
+                    yield sse('content_block_stop', {'type': 'content_block_stop', 'index': idx})
+                tool_map.clear()
+
+                # message_delta + message_stop
+                fr_map = {'stop': 'end_turn', 'length': 'max_tokens',
+                          'tool_calls': 'tool_use', None: 'end_turn'}
+                sr = fr_map.get(finish_reason, 'end_turn')
+                yield sse('message_delta', {
+                    'type': 'message_delta',
+                    'delta': {'stop_reason': sr, 'stop_sequence': None},
+                    'usage': {'output_tokens': max(1, output_tokens)}
+                })
+                yield sse('message_stop', {'type': 'message_stop'})
+                continue
+
+            try:
+                data = json.loads(data_str)
+            except json.JSONDecodeError:
+                continue
+
+            # usage (有些 provider 在流式 chunk 中包含 usage)
+            if 'usage' in data:
+                u = data['usage']
+                output_tokens = u.get('completion_tokens', output_tokens)
+                input_tokens = u.get('prompt_tokens', input_tokens)
+
+            choices = data.get('choices', [])
+            if not choices:
+                continue
+            choice = choices[0]
+            delta = choice.get('delta', {})
+
+            # 首次 chunk: 发送 message_start
+            if not message_started:
+                message_started = True
+                msg_id = 'msg_' + data.get('id', str(int(time.time() * 1000)))
+                yield sse('message_start', {
+                    'type': 'message_start',
+                    'message': {
+                        'id': msg_id, 'type': 'message', 'role': 'assistant',
+                        'content': [], 'model': model,
+                        'stop_reason': None, 'stop_sequence': None,
+                        'usage': {'input_tokens': input_tokens, 'output_tokens': 1}
+                    }
+                })
+
+            fr = choice.get('finish_reason')
+            if fr:
+                finish_reason = fr
+
+            # 文本内容 delta
+            cd = delta.get('content')
+            if cd is not None and cd != '':
+                if not text_block_open:
+                    text_block_open = True
+                    text_block_index = next_block_index
+                    next_block_index += 1
+                    yield sse('content_block_start', {
+                        'type': 'content_block_start', 'index': text_block_index,
+                        'content_block': {'type': 'text', 'text': ''}
+                    })
+                yield sse('content_block_delta', {
+                    'type': 'content_block_delta', 'index': text_block_index,
+                    'delta': {'type': 'text_delta', 'text': cd}
+                })
+
+            # tool_calls delta
+            tcd = delta.get('tool_calls')
+            if tcd:
+                for tc in tcd:
+                    ti = tc.get('index', 0)
+                    if ti not in tool_map:
+                        # 关闭 text block（如果开着）
+                        if text_block_open:
+                            yield sse('content_block_stop', {'type': 'content_block_stop', 'index': text_block_index})
+                            text_block_open = False
+                        tool_map[ti] = next_block_index
+                        next_block_index += 1
+                        func = tc.get('function', {})
+                        yield sse('content_block_start', {
+                            'type': 'content_block_start', 'index': tool_map[ti],
+                            'content_block': {
+                                'type': 'tool_use',
+                                'id': tc.get('id', f'toolu_{tool_map[ti]}'),
+                                'name': func.get('name', ''),
+                                'input': {}
+                            }
+                        })
+                    func = tc.get('function', {})
+                    args = func.get('arguments', '')
+                    if args:
+                        yield sse('content_block_delta', {
+                            'type': 'content_block_delta', 'index': tool_map[ti],
+                            'delta': {'type': 'input_json_delta', 'partial_json': args}
+                        })
+
+    # 未收到 [DONE] 的兜底关闭
+    if not done and message_started:
+        if text_block_open:
+            yield sse('content_block_stop', {'type': 'content_block_stop', 'index': text_block_index})
+        for idx in sorted(tool_map.values()):
+            yield sse('content_block_stop', {'type': 'content_block_stop', 'index': idx})
+        fr_map = {'stop': 'end_turn', 'length': 'max_tokens',
+                  'tool_calls': 'tool_use', None: 'end_turn'}
+        sr = fr_map.get(finish_reason, 'end_turn')
+        yield sse('message_delta', {
+            'type': 'message_delta',
+            'delta': {'stop_reason': sr, 'stop_sequence': None},
+            'usage': {'output_tokens': max(1, output_tokens)}
+        })
+        yield sse('message_stop', {'type': 'message_stop'})
+
+
 # ================= Flask 应用 =================
 app = Flask(__name__)
 
@@ -737,6 +1119,166 @@ def proxy(subpath):
         return {"error": f"网关转发失败: {str(e)}"}, 500
 
 
+@app.route('/v1/messages', methods=['POST'])
+def anthropic_messages():
+    """Anthropic /v1/messages 端点
+
+    流程: Anthropic 请求 → OpenAI 请求 → 上游转发 → OpenAI 响应 → Anthropic 响应
+    复用现有的 token 获取、上下文截断、超时、重试逻辑
+    """
+    # 获取 token
+    real_token = find_token_in_memory()
+    if not real_token:
+        logger.error("Anthropic: 未找到 Token")
+        return {"type": "error", "error": {"type": "authentication_error", "message": "未找到有效Token"}}, 500
+
+    # 解析 Anthropic 请求
+    anthropic_body = request.get_json(force=True, silent=True)
+    if not anthropic_body:
+        return {"type": "error", "error": {"type": "invalid_request_error", "message": "无效的请求体"}}, 400
+
+    model = anthropic_body.get('model', '')
+    is_stream = anthropic_body.get('stream', False)
+
+    # 转换为 OpenAI 格式
+    try:
+        openai_body = _anthropic_request_to_openai(anthropic_body)
+    except Exception as e:
+        logger.error(f"Anthropic→OpenAI 请求转换失败: {traceback.format_exc()}")
+        return {"type": "error", "error": {"type": "invalid_request_error", "message": f"请求转换失败: {e}"}}, 400
+
+    openai_body_bytes = json.dumps(openai_body, ensure_ascii=False).encode('utf-8')
+
+    # 上下文自动截断（复用现有逻辑）
+    openai_body_bytes, trim_info = _maybe_trim_context(openai_body_bytes)
+    body_size = len(openai_body_bytes)
+
+    # 请求体大小校验
+    if body_size > UPSTREAM_BODY_LIMIT:
+        logger.warning(f"Anthropic: 请求体超限 {body_size // 1024}KB")
+        return {"type": "error", "error": {"type": "invalid_request_error", "message": f"请求体过大({body_size // 1024}KB)"}}, 413
+
+    # 构建上游请求头
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'Bearer {real_token}',
+        'Host': TARGET_HOST,
+        'Connection': 'keep-alive',
+    }
+    for k, v in request.headers:
+        kl = k.lower()
+        if kl not in ('host', 'content-length', 'connection', 'accept-encoding',
+                       'transfer-encoding', 'authorization', 'content-type',
+                       'x-api-key', 'anthropic-version', 'anthropic-beta'):
+            headers[k] = v
+
+    target_url = f'https://{TARGET_HOST}/v2/chat/completions'
+
+    # 动态超时
+    if body_size > 800 * 1024:
+        upstream_timeout = UPSTREAM_TIMEOUT_MAX
+    elif body_size > 200 * 1024:
+        upstream_timeout = 180
+    else:
+        upstream_timeout = UPSTREAM_TIMEOUT_MIN
+
+    try:
+        resp = _forward_upstream('POST', target_url, headers, openai_body_bytes,
+                                 request.cookies, upstream_timeout)
+
+        # 401 重试（复用 proxy 逻辑）
+        if resp.status_code == 401:
+            resp.close()
+            logger.warning("Anthropic: 401，刷新 Token 重试...")
+            cache.blacklist_current()
+            new_token = find_token_in_memory()
+            if new_token and new_token != real_token:
+                headers['Authorization'] = f'Bearer {new_token}'
+                resp = _forward_upstream('POST', target_url, headers, openai_body_bytes,
+                                         request.cookies, upstream_timeout)
+                if resp.status_code == 401:
+                    resp.close()
+                    cache.clear_blacklist()
+                    new_token2 = find_token_in_memory()
+                    if new_token2:
+                        headers['Authorization'] = f'Bearer {new_token2}'
+                        resp = _forward_upstream('POST', target_url, headers, openai_body_bytes,
+                                                 request.cookies, upstream_timeout)
+
+        # 429 限流重试
+        if resp.status_code == 429:
+            for i in range(1, RETRY_ON_429 + 1):
+                resp.close()
+                time.sleep(i * 5)
+                logger.warning(f"Anthropic: 429，重试 {i}/{RETRY_ON_429}")
+                resp = _forward_upstream('POST', target_url, headers, openai_body_bytes,
+                                         request.cookies, upstream_timeout)
+                if resp.status_code != 429:
+                    break
+
+        # 504 超时重试
+        if resp.status_code == 504:
+            for i in range(1, RETRY_ON_504 + 1):
+                resp.close()
+                time.sleep(i * 3)
+                logger.warning(f"Anthropic: 504，重试 {i}/{RETRY_ON_504}")
+                resp = _forward_upstream('POST', target_url, headers, openai_body_bytes,
+                                         request.cookies, upstream_timeout)
+                if resp.status_code != 504:
+                    break
+
+        # 错误处理
+        if resp.status_code >= 400:
+            content = resp.content
+            resp.close()
+            try:
+                err = json.loads(content)
+                msg = err.get('error', {}).get('message', '') or err.get('error_msg', str(err))
+            except Exception:
+                msg = f"上游返回 {resp.status_code}"
+            logger.warning(f"Anthropic: 上游错误 {resp.status_code}: {msg[:200]}")
+            return {"type": "error", "error": {"type": "api_error", "message": msg}}, resp.status_code
+
+        # 流式响应: OpenAI SSE → Anthropic SSE
+        if is_stream:
+            def stream_gen():
+                try:
+                    for chunk in _openai_sse_to_anthropic_sse(resp, model):
+                        yield chunk
+                finally:
+                    resp.close()
+
+            return Response(stream_gen(), status=200, headers={
+                'Content-Type': 'text/event-stream',
+                'Cache-Control': 'no-cache',
+            }, direct_passthrough=True)
+
+        # 非流式响应: OpenAI JSON → Anthropic JSON
+        else:
+            content = resp.content
+            resp.close()
+            try:
+                openai_resp = json.loads(content)
+                anthropic_resp = _openai_response_to_anthropic(openai_resp, model)
+                return Response(
+                    json.dumps(anthropic_resp, ensure_ascii=False).encode('utf-8'),
+                    status=200, headers={'Content-Type': 'application/json'}
+                )
+            except Exception as e:
+                logger.error(f"OpenAI→Anthropic 响应转换失败: {traceback.format_exc()}")
+                return {"type": "error", "error": {"type": "api_error", "message": f"响应转换失败: {e}"}}, 500
+
+    except requests.exceptions.Timeout:
+        logger.error("Anthropic: 请求超时")
+        return {"type": "error", "error": {"type": "api_error", "message": "请求超时"}}, 504
+    except requests.exceptions.ConnectionError:
+        logger.error("Anthropic: 无法连接上游")
+        return {"type": "error", "error": {"type": "api_error", "message": "无法连接上游服务"}}, 502
+    except Exception as e:
+        logger.error(f"Anthropic 端点错误: {traceback.format_exc()}")
+        return {"type": "error", "error": {"type": "api_error", "message": str(e)}}, 500
+
+
 def main():
     port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
     host = sys.argv[2] if len(sys.argv) > 2 else '127.0.0.1'