feat(core): 增强 OpenAI API 处理逻辑和工具支持

- 在 .env.example 中添加 HTTP 和 HTTPS 代理配置示例
- 更新 .gitignore，添加参考文档目录忽略
- 删除 pyproject.toml 文件
- 重构 openai.py，集成 ZAITransformer 和 SSEToolHandler，优化请求处理和流式响应逻辑
- 增强响应处理器，支持工具调用和思考内容处理

.env.example

@@ -39,3 +39,9 @@ TOOL_SUPPORT=true
 
 # 工具调用扫描限制（字符数）
 SCAN_LIMIT=200000
+
+# HTTP代理地址（可选）
+HTTP_PROXY="http://admin:sk-123456@192.168.10.100:8282"
+# HTTPS代理地址（可选）
+HTTPS_PROXY="http://admin:sk-123456@192.168.10.100:8282"
+# 示例: http://username:password@host:port

.gitignore

@@ -1,4 +1,5 @@
 # Custom
+参考文档/
 .vs/
 .vscode/
 .idea/

app/core/openai.py

@@ -3,22 +3,25 @@ OpenAI API endpoints
 """
 
 import time
+import json
+import asyncio
 from datetime import datetime
-from typing import List
+from typing import List, Dict, Any
 from fastapi import APIRouter, Header, HTTPException
 from fastapi.responses import StreamingResponse
+import httpx
 
 from app.core.config import settings
-from app.models.schemas import (
-    OpenAIRequest, Message, UpstreamRequest, ModelItem, 
-    ModelsResponse, Model
-)
-from app.utils.helpers import debug_log, generate_request_ids, get_auth_token
-from app.utils.tools import process_messages_with_tools, content_to_string
-from app.core.response_handlers import StreamResponseHandler, NonStreamResponseHandler
+from app.models.schemas import OpenAIRequest, Message, ModelsResponse, Model
+from app.utils.helpers import debug_log
+from app.core.zai_transformer import ZAITransformer, generate_uuid
+from app.utils.sse_tool_handler import SSEToolHandler
 
 router = APIRouter()
 
+# 全局转换器实例
+transformer = ZAITransformer()
+
 
 @router.get("/v1/models")
 async def list_models():
@@ -26,150 +29,453 @@ async def list_models():
     current_time = int(time.time())
     response = ModelsResponse(
         data=[
-            Model(
-                id=settings.PRIMARY_MODEL,
-                created=current_time,
-                owned_by="z.ai"
-            ),
-            Model(
-                id=settings.THINKING_MODEL,
-                created=current_time,
-                owned_by="z.ai"
-            ),
-            Model(
-                id=settings.SEARCH_MODEL,
-                created=current_time,
-                owned_by="z.ai"
-            ),
-            Model(
-                id=settings.AIR_MODEL,
-                created=current_time,
-                owned_by="z.ai"
-            ),
+            Model(id=settings.PRIMARY_MODEL, created=current_time, owned_by="z.ai"),
+            Model(id=settings.THINKING_MODEL, created=current_time, owned_by="z.ai"),
+            Model(id=settings.SEARCH_MODEL, created=current_time, owned_by="z.ai"),
+            Model(id=settings.AIR_MODEL, created=current_time, owned_by="z.ai"),
         ]
     )
     return response
 
 
 @router.post("/v1/chat/completions")
-async def chat_completions(
-    request: OpenAIRequest,
-    authorization: str = Header(...)
-):
-    """Handle chat completion requests"""
-    debug_log("收到chat completions请求")
+async def chat_completions(request: OpenAIRequest, authorization: str = Header(...)):
+    """Handle chat completion requests with ZAI transformer"""
+    role = request.messages[0].role if request.messages else "unknown"
+    debug_log(f"😶‍🌫️ 收到 客户端 请求 - 模型: {request.model}, 流式: {request.stream}, 消息数: {len(request.messages)}, 角色: {role}, 工具数: {len(request.tools) if request.tools else 0}")
     
     try:
         # Validate API key (skip if SKIP_AUTH_TOKEN is enabled)
         if not settings.SKIP_AUTH_TOKEN:
             if not authorization.startswith("Bearer "):
-                debug_log("缺少或无效的Authorization头")
                 raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
             
             api_key = authorization[7:]
             if api_key != settings.AUTH_TOKEN:
-                debug_log(f"无效的API key: {api_key}")
                 raise HTTPException(status_code=401, detail="Invalid API key")
             
-            debug_log(f"API key验证通过，AUTH_TOKEN={api_key[:8]}......")
-        else:
-            debug_log("SKIP_AUTH_TOKEN已启用，跳过API key验证")
-        debug_log(f"请求解析成功 - 模型: {request.model}, 流式: {request.stream}, 消息数: {len(request.messages)}")
-        
-        # Generate IDs
-        chat_id, msg_id = generate_request_ids()
-        
-        # Process messages with tools
-        processed_messages = process_messages_with_tools(
-            [m.model_dump() for m in request.messages],
-            request.tools,
-            request.tool_choice
-        )
-        
-        # Convert back to Message objects
-        upstream_messages: List[Message] = []
-        for msg in processed_messages:
-            content = content_to_string(msg.get("content"))
-            
-            upstream_messages.append(Message(
-                role=msg["role"],
-                content=content,
-                reasoning_content=msg.get("reasoning_content")
-            ))
+        # 使用新的转换器转换请求
+        request_dict = request.model_dump()
+        debug_log("🔄 开始转换请求格式: OpenAI -> Z.AI")
         
-        # Determine model features
-        is_thinking = request.model == settings.THINKING_MODEL
-        is_search = request.model == settings.SEARCH_MODEL
-        is_air = request.model == settings.AIR_MODEL
-        search_mcp = "deep-web-search" if is_search else ""
+        transformed = await transformer.transform_request_in(request_dict)
+
+        # 调用上游API
+        async def stream_response():
+            """流式响���生成器（包含重试机制）"""
+            retry_count = 0
+            last_error = None
+            current_token = transformed.get("token", "")  # 获取当前使用的token
+
+            while retry_count <= settings.MAX_RETRIES:
+                try:
+                    # 如果是重试，重新获取令牌并更新请求
+                    if retry_count > 0:
+                        delay = 2.0
+                        debug_log(f"重试请求 ({retry_count}/{settings.MAX_RETRIES}) - 等待 {delay:.1f}s")
+                        await asyncio.sleep(delay)
+
+                        # 标记前一个token失败
+                        if current_token:
+                            transformer.mark_token_failure(current_token, Exception(f"Retry {retry_count}: {last_error}"))
+
+                        # 重新获取令牌
+                        debug_log("🔑 重新获取令牌用于重试...")
+                        new_token = await transformer.get_token()
+                        if not new_token:
+                            debug_log("❌ 重试时无法获取有效的认证令牌")
+                            raise Exception("重试时无法获取有效的认证令牌")
+                        transformed["config"]["headers"]["Authorization"] = f"Bearer {new_token}"
+                        current_token = new_token
+
+                    async with httpx.AsyncClient(timeout=60.0) as client:
+                        # 发送请求到上游
+                        debug_log(f"🎯 发送请求到 Z.AI: {transformed['config']['url']}")
+                        async with client.stream(
+                            "POST",
+                            transformed["config"]["url"],
+                            json=transformed["body"],
+                            headers=transformed["config"]["headers"],
+                        ) as response:
+                            # 检查响应状态码
+                            if response.status_code == 400:
+                                # 400 错误，触发重试
+                                error_text = await response.aread()
+                                error_msg = error_text.decode('utf-8', errors='ignore')
+                                debug_log(f"❌ 上游返回 400 错误 (尝试 {retry_count + 1}/{settings.MAX_RETRIES + 1})")
+                                debug_log(f"上游错误响应: {error_msg}")
+
+                                retry_count += 1
+                                last_error = f"400 Bad Request: {error_msg}"
+
+                                # 如果还有重试机会，继续循环
+                                if retry_count <= settings.MAX_RETRIES:
+                                    continue
+                                else:
+                                    # 达到最大重试次数，抛出错误
+                                    debug_log(f"❌ 达到最大重试次数 ({settings.MAX_RETRIES})，请求失败")
+                                    error_response = {
+                                        "error": {
+                                            "message": f"Request failed after {settings.MAX_RETRIES} retries: {last_error}",
+                                            "type": "upstream_error",
+                                            "code": 400
+                                        }
+                                    }
+                                    yield f"data: {json.dumps(error_response)}\n\n"
+                                    yield "data: [DONE]\n\n"
+                                    return
+
+                            elif response.status_code != 200:
+                                # 其他错误，直接返回
+                                debug_log(f"❌ 上游返回错误: {response.status_code}")
+                                error_text = await response.aread()
+                                error_msg = error_text.decode('utf-8', errors='ignore')
+                                debug_log(f"❌ 错误详情: {error_msg}")
+
+                                error_response = {
+                                    "error": {
+                                        "message": f"Upstream error: {response.status_code}",
+                                        "type": "upstream_error",
+                                        "code": response.status_code
+                                    }
+                                }
+                                yield f"data: {json.dumps(error_response)}\n\n"
+                                yield "data: [DONE]\n\n"
+                                return
+
+                            # 200 成功，处理响应
+                            debug_log(f"✅ Z.AI 响应成功，开始处理 SSE 流")
+                            if retry_count > 0:
+                                debug_log(f"✨ 第 {retry_count} 次重试成功")
+
+                            # 标记token使用成功
+                            if current_token:
+                                transformer.mark_token_success(current_token)
+
+                            # 初始化工具处理器（如果需要）
+                            has_tools = transformed["body"].get("tools") is not None
+                            has_mcp_servers = bool(transformed["body"].get("mcp_servers"))
+                            tool_handler = None
+
+                            # 如果有工具定义或MCP服务器，都需要工具处理器
+                            if has_tools or has_mcp_servers:
+                                chat_id = transformed["body"]["chat_id"]
+                                model = request.model
+                                tool_handler = SSEToolHandler(chat_id, model)
+
+                                if has_tools and has_mcp_servers:
+                                    debug_log(f"🔧 初始化工具处理器: {len(transformed['body'].get('tools', []))} 个OpenAI工具 + {len(transformed['body'].get('mcp_servers', []))} 个MCP服务器")
+                                elif has_tools:
+                                    debug_log(f"🔧 初始化工具处理器: {len(transformed['body'].get('tools', []))} 个OpenAI工具")
+                                elif has_mcp_servers:
+                                    debug_log(f"🔧 初始化工具处理器: {len(transformed['body'].get('mcp_servers', []))} 个MCP服务器")
+
+                            # 处理状态
+                            has_thinking = False
+                            thinking_signature = None
+
+                            # 处理SSE流
+                            buffer = ""
+                            line_count = 0
+                            debug_log("📡 开始接收 SSE 流数据...")
+
+                            async for line in response.aiter_lines():
+                                line_count += 1
+                                if not line:
+                                    continue
+
+                                # 累积到buffer处理完整的数据行
+                                buffer += line + "\n"
+
+                                # 检查是否有完整的data行
+                                while "\n" in buffer:
+                                    current_line, buffer = buffer.split("\n", 1)
+                                    if not current_line.strip():
+                                        continue
+
+                                    if current_line.startswith("data:"):
+                                        chunk_str = current_line[5:].strip()
+                                        if not chunk_str or chunk_str == "[DONE]":
+                                            if chunk_str == "[DONE]":
+                                                yield "data: [DONE]\n\n"
+                                            continue
+
+                                        debug_log(f"📦 解析数据块: {chunk_str[:200]}..." if len(chunk_str) > 200 else f"📦 解析数据块: {chunk_str}")
+
+                                        try:
+                                            chunk = json.loads(chunk_str)
+
+                                            if chunk.get("type") == "chat:completion":
+                                                data = chunk.get("data", {})
+                                                phase = data.get("phase")
+
+                                                # 记录每个阶段（只在阶段变化时记录）
+                                                if phase and phase != getattr(stream_response, '_last_phase', None):
+                                                    debug_log(f"📈 SSE 阶段: {phase}")
+                                                    stream_response._last_phase = phase
+
+                                                # 处理工具调用
+                                                if phase == "tool_call" and tool_handler:
+                                                    for output in tool_handler.process_tool_call_phase(data, True):
+                                                        yield output
+
+                                                # 处理其他阶段（工具结束）
+                                                elif phase == "other" and tool_handler:
+                                                    for output in tool_handler.process_other_phase(data, True):
+                                                        yield output
+
+                                                # 处理思考内容
+                                                elif phase == "thinking":
+                                                    if not has_thinking:
+                                                        has_thinking = True
+                                                        # 发送初始角色
+                                                        role_chunk = {
+                                                            "choices": [
+                                                                {
+                                                                    "delta": {"role": "assistant"},
+                                                                    "finish_reason": None,
+                                                                    "index": 0,
+                                                                    "logprobs": None,
+                                                                }
+                                                            ],
+                                                            "created": int(time.time()),
+                                                            "id": transformed["body"]["chat_id"],
+                                                            "model": request.model,
+                                                            "object": "chat.completion.chunk",
+                                                            "system_fingerprint": "fp_zai_001",
+                                                        }
+                                                        yield f"data: {json.dumps(role_chunk)}\n\n"
+
+                                                    delta_content = data.get("delta_content", "")
+                                                    if delta_content:
+                                                        # 处理思考内容格式
+                                                        if delta_content.startswith("<details"):
+                                                            content = (
+                                                                delta_content.split("</summary>\n>")[-1].strip()
+                                                                if "</summary>\n>" in delta_content
+                                                                else delta_content
+                                                            )
+                                                        else:
+                                                            content = delta_content
+
+                                                        thinking_chunk = {
+                                                            "choices": [
+                                                                {
+                                                                    "delta": {
+                                                                        "role": "assistant",
+                                                                        "thinking": {"content": content},
+                                                                    },
+                                                                    "finish_reason": None,
+                                                                    "index": 0,
+                                                                    "logprobs": None,
+                                                                }
+                                                            ],
+                                                            "created": int(time.time()),
+                                                            "id": transformed["body"]["chat_id"],
+                                                            "model": request.model,
+                                                            "object": "chat.completion.chunk",
+                                                            "system_fingerprint": "fp_zai_001",
+                                                        }
+                                                        yield f"data: {json.dumps(thinking_chunk)}\n\n"
+
+                                                # 处理答案内容
+                                                elif phase == "answer":
+                                                    edit_content = data.get("edit_content", "")
+                                                    delta_content = data.get("delta_content", "")
+
+                                                    # 处理思考结束和答案开始
+                                                    if edit_content and "</details>\n" in edit_content:
+                                                        if has_thinking:
+                                                            # 发送思考签名
+                                                            thinking_signature = str(int(time.time() * 1000))
+                                                            sig_chunk = {
+                                                                "choices": [
+                                                                    {
+                                                                        "delta": {
+                                                                            "role": "assistant",
+                                                                            "thinking": {
+                                                                                "content": "",
+                                                                                "signature": thinking_signature,
+                                                                            },
+                                                                        },
+                                                                        "finish_reason": None,
+                                                                        "index": 0,
+                                                                        "logprobs": None,
+                                                                    }
+                                                                ],
+                                                                "created": int(time.time()),
+                                                                "id": transformed["body"]["chat_id"],
+                                                                "model": request.model,
+                                                                "object": "chat.completion.chunk",
+                                                                "system_fingerprint": "fp_zai_001",
+                                                            }
+                                                            yield f"data: {json.dumps(sig_chunk)}\n\n"
+
+                                                        # 提取答案内容
+                                                        content_after = edit_content.split("</details>\n")[-1]
+                                                        if content_after:
+                                                            content_chunk = {
+                                                                "choices": [
+                                                                    {
+                                                                        "delta": {
+                                                                            "role": "assistant",
+                                                                            "content": content_after,
+                                                                        },
+                                                                        "finish_reason": None,
+                                                                        "index": 0,
+                                                                        "logprobs": None,
+                                                                    }
+                                                                ],
+                                                                "created": int(time.time()),
+                                                                "id": transformed["body"]["chat_id"],
+                                                                "model": request.model,
+                                                                "object": "chat.completion.chunk",
+                                                                "system_fingerprint": "fp_zai_001",
+                                                            }
+                                                            yield f"data: {json.dumps(content_chunk)}\n\n"
+
+                                                    # 处理增量内容
+                                                    elif delta_content:
+                                                        # 如果还没有发送角色
+                                                        if not has_thinking:
+                                                            role_chunk = {
+                                                                "choices": [
+                                                                    {
+                                                                        "delta": {"role": "assistant"},
+                                                                        "finish_reason": None,
+                                                                        "index": 0,
+                                                                        "logprobs": None,
+                                                                    }
+                                                                ],
+                                                                "created": int(time.time()),
+                                                                "id": transformed["body"]["chat_id"],
+                                                                "model": request.model,
+                                                                "object": "chat.completion.chunk",
+                                                                "system_fingerprint": "fp_zai_001",
+                                                            }
+                                                            yield f"data: {json.dumps(role_chunk)}\n\n"
+
+                                                        content_chunk = {
+                                                            "choices": [
+                                                                {
+                                                                    "delta": {
+                                                                        "role": "assistant",
+                                                                        "content": delta_content,
+                                                                    },
+                                                                    "finish_reason": None,
+                                                                    "index": 0,
+                                                                    "logprobs": None,
+                                                                }
+                                                            ],
+                                                            "created": int(time.time()),
+                                                            "id": transformed["body"]["chat_id"],
+                                                            "model": request.model,
+                                                            "object": "chat.completion.chunk",
+                                                            "system_fingerprint": "fp_zai_001",
+                                                        }
+                                                        output_data = f"data: {json.dumps(content_chunk)}\n\n"
+                                                        debug_log(f"➡️ 输出内容块到客户端: {output_data[:200]}...")
+                                                        yield output_data
+
+                                                    # 处理完成
+                                                    if data.get("usage"):
+                                                        debug_log(f"📦 完成响应 - 使用统计: {json.dumps(data['usage'])}")
+
+                                                        # 只有在非工具调用模式下才发送普通完成信号
+                                                        if not tool_handler or not tool_handler.has_tool_call:
+                                                            finish_chunk = {
+                                                                "choices": [
+                                                                    {
+                                                                        "delta": {"role": "assistant", "content": ""},
+                                                                        "finish_reason": "stop",
+                                                                        "index": 0,
+                                                                        "logprobs": None,
+                                                                    }
+                                                                ],
+                                                                "usage": data["usage"],
+                                                                "created": int(time.time()),
+                                                                "id": transformed["body"]["chat_id"],
+                                                                "model": request.model,
+                                                                "object": "chat.completion.chunk",
+                                                                "system_fingerprint": "fp_zai_001",
+                                                            }
+                                                            finish_output = f"data: {json.dumps(finish_chunk)}\n\n"
+                                                            debug_log(f"➡️ 发送完成信号: {finish_output[:200]}...")
+                                                            yield finish_output
+                                                            debug_log("➡️ 发送 [DONE]")
+                                                            yield "data: [DONE]\n\n"
+
+                                        except json.JSONDecodeError as e:
+                                            debug_log(f"❌ JSON解析错误: {e}, 内容: {chunk_str[:200]}")
+                                        except Exception as e:
+                                            debug_log(f"❌ 处理chunk错误: {e}")
+
+                            # 确保发送结束信号
+                            if not tool_handler or not tool_handler.has_tool_call:
+                                debug_log("📤 发送最终 [DONE] 信号")
+                                yield "data: [DONE]\n\n"
+
+                            debug_log(f"✅ SSE 流处理完成，共处理 {line_count} 行数据")
+                            # 成功处理完成，退出重试循环
+                            return
+
+                except Exception as e:
+                    debug_log(f"❌ 流处理错误: {e}")
+                    import traceback
+                    debug_log(traceback.format_exc())
+
+                    # 标记token失败
+                    if current_token:
+                        transformer.mark_token_failure(current_token, e)
+
+                    # 检查是否还可以重试
+                    retry_count += 1
+                    last_error = str(e)
+
+                    if retry_count > settings.MAX_RETRIES:
+                        # 达到最大重试次数，返回错误
+                        debug_log(f"❌ 达到最大重试次数 ({settings.MAX_RETRIES})，流处理失败")
+                        error_response = {
+                            "error": {
+                                "message": f"Stream processing failed after {settings.MAX_RETRIES} retries: {last_error}",
+                                "type": "stream_error"
+                            }
+                        }
+                        yield f"data: {json.dumps(error_response)}\n\n"
+                        yield "data: [DONE]\n\n"
+                        return
+
+        # 返回流式响应
+        debug_log("🚀 启动 SSE 流式响应")
         
-        # Determine upstream model ID based on requested model
-        if is_air:
-            upstream_model_id = "0727-106B-API"  # AIR model upstream ID
-            upstream_model_name = "GLM-4.5-Air"
-        else:
-            upstream_model_id = "0727-360B-API"  # Default upstream model ID
-            upstream_model_name = "GLM-4.5"
+        # 创建一个包装的生成器来追踪数据流
+        async def logged_stream():
+            chunk_count = 0
+            try:
+                debug_log("📤 开始向客户端流式传输数据...")
+                async for chunk in stream_response():
+                    chunk_count += 1
+                    debug_log(f"📤 发送块[{chunk_count}]: {chunk[:200]}..." if len(chunk) > 200 else f"  📤 发送块[{chunk_count}]: {chunk}")
+                    yield chunk
+                debug_log(f"✅ 流式传输完成，共发送 {chunk_count} 个数据块")
+            except Exception as e:
+                debug_log(f"❌ 流式传输中断: {e}")
+                raise
         
-        # Build upstream request
-        upstream_req = UpstreamRequest(
-            stream=True,  # Always use streaming from upstream
-            chat_id=chat_id,
-            id=msg_id,
-            model=upstream_model_id,  # Dynamic upstream model ID
-            messages=upstream_messages,
-            params={},
-            features={
-                "enable_thinking": is_thinking,
-                "web_search": is_search,
-                "auto_web_search": is_search,
+        return StreamingResponse(
+            logged_stream(),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
             },
-            background_tasks={
-                "title_generation": False,
-                "tags_generation": False,
-            },
-            mcp_servers=[search_mcp] if search_mcp else [],
-            model_item=ModelItem(
-                id=upstream_model_id,
-                name=upstream_model_name,
-                owned_by="openai"
-            ),
-            tool_servers=[],
-            variables={
-                "{{USER_NAME}}": "User",
-                "{{USER_LOCATION}}": "Unknown",
-                "{{CURRENT_DATETIME}}": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            }
         )
-        
-        # Get authentication token
-        auth_token = get_auth_token()
-        
-        # Check if tools are enabled and present
-        has_tools = (settings.TOOL_SUPPORT and 
-                    request.tools and 
-                    len(request.tools) > 0 and 
-                    request.tool_choice != "none")
-        
-        # Handle response based on stream flag
-        if request.stream:
-            handler = StreamResponseHandler(upstream_req, chat_id, auth_token, has_tools)
-            return StreamingResponse(
-                handler.handle(),
-                media_type="text/event-stream",
-                headers={
-                    "Cache-Control": "no-cache",
-                    "Connection": "keep-alive",
-                }
-            )
-        else:
-            handler = NonStreamResponseHandler(upstream_req, chat_id, auth_token, has_tools)
-            return handler.handle()
-            
+
     except HTTPException:
         raise
     except Exception as e:
-        debug_log(f"处理请求时发生错误: {str(e)}")
+        debug_log(f"❌ 处理请求时发生错误: {str(e)}")
         import traceback
-        debug_log(f"错误堆栈: {traceback.format_exc()}")
+
+        debug_log(f"❌ 错误堆栈: {traceback.format_exc()}")
         raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

app/core/response_handlers.py

@@ -1,468 +1,499 @@
-"""
-Response handlers for streaming and non-streaming responses
-"""
-
-import json
-import time
-from typing import Generator, Optional
-import requests
-from fastapi import HTTPException
-from fastapi.responses import JSONResponse, StreamingResponse
-
-from app.core.config import settings
-from app.models.schemas import (
-    Message, Delta, Choice, Usage, OpenAIResponse, 
-    UpstreamRequest, UpstreamData, UpstreamError, ModelItem
-)
-from app.utils.helpers import debug_log, call_upstream_api, transform_thinking_content
-from app.core.token_manager import token_manager
-from app.utils.sse_parser import SSEParser
+"""
+Response handlers for streaming and non-streaming responses
+"""
+
+import json
+import time
+from typing import Generator, Optional
+import requests
+from fastapi import HTTPException
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from app.core.config import settings
+from app.models.schemas import (
+    Message, Delta, Choice, Usage, OpenAIResponse, 
+    UpstreamRequest, UpstreamData, UpstreamError, ModelItem
+)
+from app.utils.helpers import debug_log, call_upstream_api, transform_thinking_content
+from app.core.token_manager import token_manager
+from app.utils.sse_parser import SSEParser
 from app.utils.tools import extract_tool_invocations, remove_tool_json_content
-
-
-def create_openai_response_chunk(
-    model: str,
-    delta: Optional[Delta] = None,
-    finish_reason: Optional[str] = None
-) -> OpenAIResponse:
-    """Create OpenAI response chunk for streaming"""
-    return OpenAIResponse(
-        id=f"chatcmpl-{int(time.time())}",
-        object="chat.completion.chunk",
-        created=int(time.time()),
-        model=model,
-        choices=[Choice(
-            index=0,
-            delta=delta or Delta(),
-            finish_reason=finish_reason
-        )]
-    )
-
-
-def handle_upstream_error(error: UpstreamError) -> Generator[str, None, None]:
-    """Handle upstream error response"""
-    debug_log(f"上游错误: code={error.code}, detail={error.detail}")
-    
-    # Send end chunk
-    end_chunk = create_openai_response_chunk(
-        model=settings.PRIMARY_MODEL,
-        finish_reason="stop"
-    )
-    yield f"data: {end_chunk.model_dump_json()}\n\n"
-    yield "data: [DONE]\n\n"
-
-
-class ResponseHandler:
-    """Base class for response handling"""
-    
-    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str):
-        self.upstream_req = upstream_req
-        self.chat_id = chat_id
-        self.auth_token = auth_token
-    
-    def _call_upstream(self) -> requests.Response:
-        """Call upstream API with error handling"""
-        max_retries = settings.MAX_RETRIES
-        retry_count = 0
-        
-        while retry_count < max_retries:
-            try:
-                debug_log(f"尝试调用上游API (第 {retry_count + 1}/{max_retries} 次)")
-                response = call_upstream_api(self.upstream_req, self.chat_id, self.auth_token)
-                
-                # Check if response is successful
-                if response.status_code == 200:
-                    # Mark token as successful
-                    token_manager.mark_token_success(self.auth_token)
-                    debug_log("上游API调用成功")
-                    return response
-                elif response.status_code in [401, 403]:
-                    # Authentication/authorization error - mark token as failed
-                    debug_log(f"Token认证失败 (状态码: {response.status_code}): {self.auth_token[:20]}...")
-                    token_manager.mark_token_failed(self.auth_token)
-                    
-                    # Try to get a new token
-                    new_token = token_manager.get_next_token()
-                    if new_token and new_token != self.auth_token:
-                        debug_log(f"尝试使用新token: {new_token[:20]}...")
-                        self.auth_token = new_token
-                        retry_count += 1
-                        continue
-                    else:
-                        debug_log("没有更多可用token")
-                        return response
-                elif response.status_code in [429]:
-                    # Rate limit - don't mark token as failed, just retry
-                    debug_log(f"遇到速率限制 (状态码: {response.status_code})，等待后重试")
-                    if retry_count < max_retries - 1:
-                        import time
-                        time.sleep(2 ** retry_count)  # 指数退避
-                        retry_count += 1
-                        continue
-                    else:
-                        return response
-                elif response.status_code >= 500:
-                    # Server error - retry without marking token as failed
-                    debug_log(f"服务器错误 (状态码: {response.status_code})，稍后重试")
-                    if retry_count < max_retries - 1:
-                        import time
-                        time.sleep(1)
-                        retry_count += 1
-                        continue
-                    else:
-                        return response
-                else:
-                    # Other client errors, return response as-is
-                    debug_log(f"客户端错误 (状态码: {response.status_code})")
-                    return response
-                    
-            except Exception as e:
-                error_msg = str(e)
-                debug_log(f"调用上游失败 (尝试 {retry_count + 1}/{max_retries}): {error_msg}")
-                
-                # 判断是否是连接问题还是token问题
-                is_connection_error = any(keyword in error_msg.lower() for keyword in [
-                    'connection', 'timeout', 'network', 'dns', 'socket', 'ssl'
-                ])
-                
-                if is_connection_error:
-                    debug_log("检测到网络连接问题，不标记token失败")
-                    # 网络问题不标记token失败，直接重试
-                    if retry_count < max_retries - 1:
-                        import time
-                        time.sleep(2)  # 等待2秒后重试
-                        retry_count += 1
-                        continue
-                    else:
-                        raise Exception(f"网络连接问题，重试{max_retries}次后仍失败: {error_msg}")
-                else:
-                    # 其他错误可能是token问题，标记失败并尝试新token
-                    debug_log("检测到可能的token问题，标记token失败")
-                    token_manager.mark_token_failed(self.auth_token)
-                    
-                    # Try to get a new token
-                    new_token = token_manager.get_next_token()
-                    if new_token and new_token != self.auth_token and retry_count < max_retries - 1:
-                        debug_log(f"尝试使用新token: {new_token[:20]}...")
-                        self.auth_token = new_token
-                        retry_count += 1
-                        continue
-                    else:
-                        raise
-        
-        # If we get here, all retries failed
-        raise Exception("所有重试尝试均失败")
-    
-    def _handle_upstream_error(self, response: requests.Response) -> None:
-        """Handle upstream error response"""
-        debug_log(f"上游返回错误状态: {response.status_code}")
-        if settings.DEBUG_LOGGING:
-            debug_log(f"上游错误响应: {response.text}")
-
-
-class StreamResponseHandler(ResponseHandler):
-    """Handler for streaming responses"""
-    
-    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str, has_tools: bool = False):
-        super().__init__(upstream_req, chat_id, auth_token)
-        self.has_tools = has_tools
-        self.buffered_content = ""
+from app.utils.sse_tool_handler import SSEToolHandler
+
+
+def create_openai_response_chunk(
+    model: str,
+    delta: Optional[Delta] = None,
+    finish_reason: Optional[str] = None
+) -> OpenAIResponse:
+    """Create OpenAI response chunk for streaming"""
+    return OpenAIResponse(
+        id=f"chatcmpl-{int(time.time())}",
+        object="chat.completion.chunk",
+        created=int(time.time()),
+        model=model,
+        choices=[Choice(
+            index=0,
+            delta=delta or Delta(),
+            finish_reason=finish_reason
+        )]
+    )
+
+
+def handle_upstream_error(error: UpstreamError) -> Generator[str, None, None]:
+    """Handle upstream error response"""
+    debug_log(f"上游错误: code={error.code}, detail={error.detail}")
+    
+    # Send end chunk
+    end_chunk = create_openai_response_chunk(
+        model=settings.PRIMARY_MODEL,
+        finish_reason="stop"
+    )
+    yield f"data: {end_chunk.model_dump_json()}\n\n"
+    yield "data: [DONE]\n\n"
+
+
+class ResponseHandler:
+    """Base class for response handling"""
+    
+    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str):
+        self.upstream_req = upstream_req
+        self.chat_id = chat_id
+        self.auth_token = auth_token
+    
+    def _call_upstream(self) -> requests.Response:
+        """Call upstream API with error handling"""
+        max_retries = settings.MAX_RETRIES
+        retry_count = 0
+        
+        while retry_count < max_retries:
+            try:
+                debug_log(f"尝试调用上游API (第 {retry_count + 1}/{max_retries} 次)")
+                response = call_upstream_api(self.upstream_req, self.chat_id, self.auth_token)
+                
+                # Check if response is successful
+                if response.status_code == 200:
+                    # Mark token as successful
+                    token_manager.mark_token_success(self.auth_token)
+                    debug_log("上游API调用成功")
+                    return response
+                elif response.status_code in [401, 403]:
+                    # Authentication/authorization error - mark token as failed
+                    debug_log(f"Token认证失败 (状态码: {response.status_code}): {self.auth_token[:20]}...")
+                    token_manager.mark_token_failed(self.auth_token)
+                    
+                    # Try to get a new token
+                    new_token = token_manager.get_next_token()
+                    if new_token and new_token != self.auth_token:
+                        debug_log(f"尝试使用新token: {new_token[:20]}...")
+                        self.auth_token = new_token
+                        retry_count += 1
+                        continue
+                    else:
+                        debug_log("没有更多可用token")
+                        return response
+                elif response.status_code in [429]:
+                    # Rate limit - don't mark token as failed, just retry
+                    debug_log(f"遇到速率限制 (状态码: {response.status_code})，等待后重试")
+                    if retry_count < max_retries - 1:
+                        import time
+                        time.sleep(2 ** retry_count)  # 指数退避
+                        retry_count += 1
+                        continue
+                    else:
+                        return response
+                elif response.status_code >= 500:
+                    # Server error - retry without marking token as failed
+                    debug_log(f"服务器错误 (状态码: {response.status_code})，稍后重试")
+                    if retry_count < max_retries - 1:
+                        import time
+                        time.sleep(1)
+                        retry_count += 1
+                        continue
+                    else:
+                        return response
+                else:
+                    # Other client errors, return response as-is
+                    debug_log(f"客户端错误 (状态码: {response.status_code})")
+                    return response
+                    
+            except Exception as e:
+                error_msg = str(e)
+                debug_log(f"调用上游失败 (尝试 {retry_count + 1}/{max_retries}): {error_msg}")
+                
+                # 判断是否是连接问题还是token问题
+                is_connection_error = any(keyword in error_msg.lower() for keyword in [
+                    'connection', 'timeout', 'network', 'dns', 'socket', 'ssl'
+                ])
+                
+                if is_connection_error:
+                    debug_log("检测到网络连接问题，不标记token失败")
+                    # 网络问题不标记token失败，直接重试
+                    if retry_count < max_retries - 1:
+                        import time
+                        time.sleep(2)  # 等待2秒后重试
+                        retry_count += 1
+                        continue
+                    else:
+                        raise Exception(f"网络连接问题，重试{max_retries}次后仍失败: {error_msg}")
+                else:
+                    # 其他错误可能是token问题，标记失败并尝试新token
+                    debug_log("检测到可能的token问题，标记token失败")
+                    token_manager.mark_token_failed(self.auth_token)
+                    
+                    # Try to get a new token
+                    new_token = token_manager.get_next_token()
+                    if new_token and new_token != self.auth_token and retry_count < max_retries - 1:
+                        debug_log(f"尝试使用新token: {new_token[:20]}...")
+                        self.auth_token = new_token
+                        retry_count += 1
+                        continue
+                    else:
+                        raise
+        
+        # If we get here, all retries failed
+        raise Exception("所有重试尝试均失败")
+    
+    def _handle_upstream_error(self, response: requests.Response) -> None:
+        """Handle upstream error response"""
+        debug_log(f"上游返回错误状态: {response.status_code}")
+        if settings.DEBUG_LOGGING:
+            debug_log(f"上游错误响应: {response.text}")
+
+
+class StreamResponseHandler(ResponseHandler):
+    """Handler for streaming responses"""
+    
+    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str, has_tools: bool = False):
+        super().__init__(upstream_req, chat_id, auth_token)
+        self.has_tools = has_tools
+        self.buffered_content = ""
         self.tool_calls = None
-    
-    def handle(self) -> Generator[str, None, None]:
-        """Handle streaming response"""
-        debug_log(f"开始处理流式响应 (chat_id={self.chat_id})")
-        
-        try:
-            response = self._call_upstream()
-        except Exception:
-            yield "data: {\"error\": \"Failed to call upstream\"}\n\n"
-            return
-        
-        if response.status_code != 200:
-            self._handle_upstream_error(response)
-            yield "data: {\"error\": \"Upstream error\"}\n\n"
-            return
-        
-        # Send initial role chunk
-        first_chunk = create_openai_response_chunk(
-            model=settings.PRIMARY_MODEL,
-            delta=Delta(role="assistant")
-        )
-        yield f"data: {first_chunk.model_dump_json()}\n\n"
-        
-        # Process stream
-        debug_log("开始读取上游SSE流")
-        sent_initial_answer = False
-        stream_ended_normally = False
-        
-        try:
-            with SSEParser(response, debug_mode=settings.DEBUG_LOGGING) as parser:
-                for event in parser.iter_json_data(UpstreamData):
-                    upstream_data = event['data']
-                    
-                    # Check for errors
-                    if self._has_error(upstream_data):
-                        error = self._get_error(upstream_data)
-                        yield from handle_upstream_error(error)
-                        stream_ended_normally = True
-                        break
-                    
-                    debug_log(f"解析成功 - 类型: {upstream_data.type}, 阶段: {upstream_data.data.phase}, "
-                             f"内容长度: {len(upstream_data.data.delta_content or '')}, 完成: {upstream_data.data.done}")
-                    
-                    # Process content
-                    yield from self._process_content(upstream_data, sent_initial_answer)
-                    
-                    # Update sent_initial_answer flag if we sent content
-                    if not sent_initial_answer and (upstream_data.data.delta_content or upstream_data.data.edit_content):
-                        sent_initial_answer = True
-                    
-                    # Check if done
-                    if upstream_data.data.done or upstream_data.data.phase == "done":
-                        debug_log("检测到流结束信号")
-                        yield from self._send_end_chunk()
-                        stream_ended_normally = True
-                        break
-                        
-        except Exception as e:
-            debug_log(f"SSE流处理异常: {e}")
-            # 流异常结束，发送错误响应
-            if not stream_ended_normally:
-                error_chunk = create_openai_response_chunk(
-                    model=settings.PRIMARY_MODEL,
-                    delta=Delta(content=f"\n\n[系统提示: 连接中断，响应可能不完整]")
-                )
-                yield f"data: {error_chunk.model_dump_json()}\n\n"
-        
-        # 确保流正常结束
-        if not stream_ended_normally:
-            debug_log("流未正常结束，发送结束信号")
-            yield from self._send_end_chunk(force_stop=True)
-    
-    def _has_error(self, upstream_data: UpstreamData) -> bool:
-        """Check if upstream data contains error"""
-        return bool(
-            upstream_data.error or 
-            upstream_data.data.error or 
-            (upstream_data.data.inner and upstream_data.data.inner.error)
-        )
-    
-    def _get_error(self, upstream_data: UpstreamData) -> UpstreamError:
-        """Get error from upstream data"""
-        return (
-            upstream_data.error or 
-            upstream_data.data.error or 
-            (upstream_data.data.inner.error if upstream_data.data.inner else None)
-        )
-    
-    def _process_content(
+        # Initialize SSE tool handler for improved tool processing
+        self.tool_handler = SSEToolHandler(chat_id, settings.PRIMARY_MODEL) if has_tools else None
+    
+    def handle(self) -> Generator[str, None, None]:
+        """Handle streaming response"""
+        debug_log(f"开始处理流式响应 (chat_id={self.chat_id})")
+        
+        try:
+            response = self._call_upstream()
+        except Exception:
+            yield "data: {\"error\": \"Failed to call upstream\"}\n\n"
+            return
+        
+        if response.status_code != 200:
+            self._handle_upstream_error(response)
+            yield "data: {\"error\": \"Upstream error\"}\n\n"
+            return
+        
+        # Send initial role chunk
+        first_chunk = create_openai_response_chunk(
+            model=settings.PRIMARY_MODEL,
+            delta=Delta(role="assistant")
+        )
+        yield f"data: {first_chunk.model_dump_json()}\n\n"
+        
+        # Process stream
+        debug_log("开始读取上游SSE流")
+        sent_initial_answer = False
+        stream_ended_normally = False
+        
+        try:
+            with SSEParser(response, debug_mode=settings.DEBUG_LOGGING) as parser:
+                for event in parser.iter_json_data(UpstreamData):
+                    upstream_data = event['data']
+                    
+                    # Check for errors
+                    if self._has_error(upstream_data):
+                        error = self._get_error(upstream_data)
+                        yield from handle_upstream_error(error)
+                        stream_ended_normally = True
+                        break
+                    
+                    debug_log(f"解析成功 - 类型: {upstream_data.type}, 阶段: {upstream_data.data.phase}, "
+                             f"内容长度: {len(upstream_data.data.delta_content or '')}, 完成: {upstream_data.data.done}")
+                    
+                    # Process content
+                    yield from self._process_content_with_tools(upstream_data, sent_initial_answer)
+                    
+                    # Update sent_initial_answer flag if we sent content
+                    if not sent_initial_answer and (upstream_data.data.delta_content or upstream_data.data.edit_content):
+                        sent_initial_answer = True
+                    
+                    # Check if done
+                    if upstream_data.data.done or upstream_data.data.phase == "done":
+                        debug_log("检测到流结束信号")
+                        yield from self._send_end_chunk()
+                        stream_ended_normally = True
+                        break
+                        
+        except Exception as e:
+            debug_log(f"SSE流处理异常: {e}")
+            # 流异常结束，发送错误响应
+            if not stream_ended_normally:
+                error_chunk = create_openai_response_chunk(
+                    model=settings.PRIMARY_MODEL,
+                    delta=Delta(content=f"\n\n[系统提示: 连接中断，响应可能不完整]")
+                )
+                yield f"data: {error_chunk.model_dump_json()}\n\n"
+        
+        # 确保流正常结束
+        if not stream_ended_normally:
+            debug_log("流未正常结束，发送结束信号")
+            yield from self._send_end_chunk(force_stop=True)
+    
+    def _has_error(self, upstream_data: UpstreamData) -> bool:
+        """Check if upstream data contains error"""
+        return bool(
+            upstream_data.error or 
+            upstream_data.data.error or 
+            (upstream_data.data.inner and upstream_data.data.inner.error)
+        )
+    
+    def _get_error(self, upstream_data: UpstreamData) -> UpstreamError:
+        """Get error from upstream data"""
+        return (
+            upstream_data.error or 
+            upstream_data.data.error or 
+            (upstream_data.data.inner.error if upstream_data.data.inner else None)
+        )
+    
+    def _process_content(
+        self, 
+        upstream_data: UpstreamData, 
+        sent_initial_answer: bool
+    ) -> Generator[str, None, None]:
+        """Process content from upstream data"""
+        content = upstream_data.data.delta_content or upstream_data.data.edit_content
+        
+        if not content:
+            return
+        
+        # Transform thinking content
+        if upstream_data.data.phase == "thinking":
+            content = transform_thinking_content(content)
+        
+        # Buffer content if tools are enabled
+        if self.has_tools:
+            self.buffered_content += content
+        else:
+            # Handle initial answer content
+            if (not sent_initial_answer and 
+                upstream_data.data.edit_content and 
+                upstream_data.data.phase == "answer"):
+                
+                content = self._extract_edit_content(upstream_data.data.edit_content)
+                if content:
+                    debug_log(f"发送普通内容: {content}")
+                    chunk = create_openai_response_chunk(
+                        model=settings.PRIMARY_MODEL,
+                        delta=Delta(content=content)
+                    )
+                    yield f"data: {chunk.model_dump_json()}\n\n"
+                    sent_initial_answer = True
+            
+            # Handle delta content
+            if upstream_data.data.delta_content:
+                if content:
+                    if upstream_data.data.phase == "thinking":
+                        debug_log(f"发送思考内容: {content}")
+                        chunk = create_openai_response_chunk(
+                            model=settings.PRIMARY_MODEL,
+                            delta=Delta(reasoning_content=content)
+                        )
+                    else:
+                        debug_log(f"发送普通内容: {content}")
+                        chunk = create_openai_response_chunk(
+                            model=settings.PRIMARY_MODEL,
+                            delta=Delta(content=content)
+                        )
+                    yield f"data: {chunk.model_dump_json()}\n\n"
+    
+    def _extract_edit_content(self, edit_content: str) -> str:
+        """Extract content from edit_content field"""
+        parts = edit_content.split("</details>")
+        return parts[1] if len(parts) > 1 else ""
+    
+    def _send_end_chunk(self, force_stop: bool = False) -> Generator[str, None, None]:
+        """Send end chunk and DONE signal"""
+        finish_reason = "stop"
+        
+        if self.has_tools and not force_stop:
+            # Try to extract tool calls from buffered content
+            self.tool_calls = extract_tool_invocations(self.buffered_content)
+            
+            if self.tool_calls:
+                debug_log(f"检测到工具调用: {len(self.tool_calls)} 个")
+                # Send tool calls with proper format
+                for i, tc in enumerate(self.tool_calls):
+                    tool_call_delta = {
+                        "index": i,
+                        "id": tc.get("id"),
+                        "type": tc.get("type", "function"),
+                        "function": tc.get("function", {}),
+                    }
+                    
+                    out_chunk = create_openai_response_chunk(
+                        model=settings.PRIMARY_MODEL,
+                        delta=Delta(tool_calls=[tool_call_delta])
+                    )
+                    yield f"data: {out_chunk.model_dump_json()}\n\n"
+                
+                finish_reason = "tool_calls"
+            else:
+                # Send regular content
+                trimmed_content = remove_tool_json_content(self.buffered_content)
+                if trimmed_content:
+                    debug_log(f"发送常规内容: {len(trimmed_content)} 字符")
+                    content_chunk = create_openai_response_chunk(
+                        model=settings.PRIMARY_MODEL,
+                        delta=Delta(content=trimmed_content)
+                    )
+                    yield f"data: {content_chunk.model_dump_json()}\n\n"
+        elif force_stop:
+            # 强制结束时，发送缓冲的内容（如果有）
+            if self.buffered_content:
+                debug_log(f"强制结束，发送缓冲内容: {len(self.buffered_content)} 字符")
+                content_chunk = create_openai_response_chunk(
+                    model=settings.PRIMARY_MODEL,
+                    delta=Delta(content=self.buffered_content)
+                )
+                yield f"data: {content_chunk.model_dump_json()}\n\n"
+        
+        # Send final chunk
+        end_chunk = create_openai_response_chunk(
+            model=settings.PRIMARY_MODEL,
+            finish_reason=finish_reason
+        )
+        yield f"data: {end_chunk.model_dump_json()}\n\n"
+        yield "data: [DONE]\n\n"
+        debug_log(f"流式响应完成 (finish_reason: {finish_reason})")
+
+
    
+    def _process_content_with_tools(
         self, 
         upstream_data: UpstreamData, 
         sent_initial_answer: bool
     ) -> Generator[str, None, None]:
-        """Process content from upstream data"""
-        content = upstream_data.data.delta_content or upstream_data.data.edit_content
-        
-        if not content:
-            return
-        
-        # Transform thinking content
-        if upstream_data.data.phase == "thinking":
-            content = transform_thinking_content(content)
-        
-        # Buffer content if tools are enabled
-        if self.has_tools:
-            self.buffered_content += content
-        else:
-            # Handle initial answer content
-            if (not sent_initial_answer and 
-                upstream_data.data.edit_content and 
-                upstream_data.data.phase == "answer"):
-                
-                content = self._extract_edit_content(upstream_data.data.edit_content)
-                if content:
-                    debug_log(f"发送普通内容: {content}")
-                    chunk = create_openai_response_chunk(
-                        model=settings.PRIMARY_MODEL,
-                        delta=Delta(content=content)
-                    )
-                    yield f"data: {chunk.model_dump_json()}\n\n"
-                    sent_initial_answer = True
-            
-            # Handle delta content
-            if upstream_data.data.delta_content:
-                if content:
-                    if upstream_data.data.phase == "thinking":
-                        debug_log(f"发送思考内容: {content}")
-                        chunk = create_openai_response_chunk(
-                            model=settings.PRIMARY_MODEL,
-                            delta=Delta(reasoning_content=content)
-                        )
-                    else:
-                        debug_log(f"发送普通内容: {content}")
-                        chunk = create_openai_response_chunk(
-                            model=settings.PRIMARY_MODEL,
-                            delta=Delta(content=content)
-                        )
-                    yield f"data: {chunk.model_dump_json()}\n\n"
-    
-    def _extract_edit_content(self, edit_content: str) -> str:
-        """Extract content from edit_content field"""
-        parts = edit_content.split("</details>")
-        return parts[1] if len(parts) > 1 else ""
-    
-    def _send_end_chunk(self, force_stop: bool = False) -> Generator[str, None, None]:
-        """Send end chunk and DONE signal"""
-        finish_reason = "stop"
-        
-        if self.has_tools and not force_stop:
-            # Try to extract tool calls from buffered content
-            self.tool_calls = extract_tool_invocations(self.buffered_content)
-            
-            if self.tool_calls:
-                debug_log(f"检测到工具调用: {len(self.tool_calls)} 个")
-                # Send tool calls with proper format
-                for i, tc in enumerate(self.tool_calls):
-                    tool_call_delta = {
-                        "index": i,
-                        "id": tc.get("id"),
-                        "type": tc.get("type", "function"),
-                        "function": tc.get("function", {}),
-                    }
-                    
-                    out_chunk = create_openai_response_chunk(
-                        model=settings.PRIMARY_MODEL,
-                        delta=Delta(tool_calls=[tool_call_delta])
-                    )
-                    yield f"data: {out_chunk.model_dump_json()}\n\n"
-                
-                finish_reason = "tool_calls"
-            else:
-                # Send regular content
-                trimmed_content = remove_tool_json_content(self.buffered_content)
-                if trimmed_content:
-                    debug_log(f"发送常规内容: {len(trimmed_content)} 字符")
-                    content_chunk = create_openai_response_chunk(
-                        model=settings.PRIMARY_MODEL,
-                        delta=Delta(content=trimmed_content)
-                    )
-                    yield f"data: {content_chunk.model_dump_json()}\n\n"
-        elif force_stop:
-            # 强制结束时，发送缓冲的内容（如果有）
-            if self.buffered_content:
-                debug_log(f"强制结束，发送缓冲内容: {len(self.buffered_content)} 字符")
-                content_chunk = create_openai_response_chunk(
-                    model=settings.PRIMARY_MODEL,
-                    delta=Delta(content=self.buffered_content)
+        """Process content with improved tool handling"""
+        # Handle tool calls with improved SSE tool handler
+        if self.has_tools and self.tool_handler:
+            # Check if this is a tool_call phase
+            if upstream_data.data.phase == "tool_call":
+                # Use the improved tool handler for tool call processing
+                yield from self.tool_handler.process_tool_call_phase(
+                    upstream_data.data.model_dump(), 
+                    is_stream=True
+                )
+                return
+            elif upstream_data.data.phase == "other":
+                # Handle other phase which may contain tool completion signals
+                yield from self.tool_handler.process_other_phase(
+                    upstream_data.data.model_dump(), 
+                    is_stream=True
                 )
-                yield f"data: {content_chunk.model_dump_json()}\n\n"
+                return
         
-        # Send final chunk
-        end_chunk = create_openai_response_chunk(
-            model=settings.PRIMARY_MODEL,
-            finish_reason=finish_reason
-        )
-        yield f"data: {end_chunk.model_dump_json()}\n\n"
-        yield "data: [DONE]\n\n"
-        debug_log(f"流式响应完成 (finish_reason: {finish_reason})")
+        # Fall back to original content processing
+        yield from self._process_content(upstream_data, sent_initial_answer)
 
 
-class NonStreamResponseHandler(ResponseHandler):
-    """Handler for non-streaming responses"""
-    
-    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str, has_tools: bool = False):
-        super().__init__(upstream_req, chat_id, auth_token)
-        self.has_tools = has_tools
-    
-    def handle(self) -> JSONResponse:
-        """Handle non-streaming response"""
-        debug_log(f"开始处理非流式响应 (chat_id={self.chat_id})")
-        
-        try:
-            response = self._call_upstream()
-        except Exception as e:
-            debug_log(f"调用上游失败: {e}")
-            raise HTTPException(status_code=502, detail="Failed to call upstream")
-        
-        if response.status_code != 200:
-            self._handle_upstream_error(response)
-            raise HTTPException(status_code=502, detail="Upstream error")
-        
-        # Collect full response
-        full_content = []
-        debug_log("开始收集完整响应内容")
-        response_completed = False
-        
-        try:
-            with SSEParser(response, debug_mode=settings.DEBUG_LOGGING) as parser:
-                for event in parser.iter_json_data(UpstreamData):
-                    upstream_data = event['data']
-                    
-                    if upstream_data.data.delta_content:
-                        content = upstream_data.data.delta_content
-                        
-                        if upstream_data.data.phase == "thinking":
-                            content = transform_thinking_content(content)
-                        
-                        if content:
-                            full_content.append(content)
-                    
-                    if upstream_data.data.done or upstream_data.data.phase == "done":
-                        debug_log("检测到完成信号，停止收集")
-                        response_completed = True
-                        break
-                        
-        except Exception as e:
-            debug_log(f"非流式响应收集异常: {e}")
-            if not full_content:
-                # 如果没有收集到任何内容，抛出异常
-                raise HTTPException(status_code=502, detail=f"Response collection failed: {str(e)}")
-            else:
-                debug_log(f"部分内容收集成功，继续处理 ({len(full_content)} 个片段)")
-        
-        if not response_completed and not full_content:
-            debug_log("响应未完成且无内容，可能是连接问题")
-            raise HTTPException(status_code=502, detail="Incomplete response from upstream")
-        
-        final_content = "".join(full_content)
-        debug_log(f"内容收集完成，最终长度: {len(final_content)}")
-        
-        # Handle tool calls for non-streaming
-        tool_calls = None
-        finish_reason = "stop"
-        message_content = final_content
-        
-        if self.has_tools:
-            tool_calls = extract_tool_invocations(final_content)
-            if tool_calls:
-                # Content must be null when tool_calls are present (OpenAI spec)
-                message_content = None
-                finish_reason = "tool_calls"
-                debug_log(f"提取到工具调用: {json.dumps(tool_calls, ensure_ascii=False)}")
-            else:
-                # Remove tool JSON from content
-                message_content = remove_tool_json_content(final_content)
-                if not message_content:
-                    message_content = final_content  # 保留原内容如果清理后为空
-        
-        # Build response
-        response_data = OpenAIResponse(
-            id=f"chatcmpl-{int(time.time())}",
-            object="chat.completion",
-            created=int(time.time()),
-            model=settings.PRIMARY_MODEL,
-            choices=[Choice(
-                index=0,
-                message=Message(
-                    role="assistant",
-                    content=message_content,
-                    tool_calls=tool_calls
-                ),
-                finish_reason=finish_reason
-            )],
-            usage=Usage()
-        )
-        
-        debug_log("非流式响应发送完成")
+class NonStreamResponseHandler(ResponseHandler):
+    """Handler for non-streaming responses"""
+    
+    def __init__(self, upstream_req: UpstreamRequest, chat_id: str, auth_token: str, has_tools: bool = False):
+        super().__init__(upstream_req, chat_id, auth_token)
+        self.has_tools = has_tools
+    
+    def handle(self) -> JSONResponse:
+        """Handle non-streaming response"""
+        debug_log(f"开始处理非流式响应 (chat_id={self.chat_id})")
+        
+        try:
+            response = self._call_upstream()
+        except Exception as e:
+            debug_log(f"调用上游失败: {e}")
+            raise HTTPException(status_code=502, detail="Failed to call upstream")
+        
+        if response.status_code != 200:
+            self._handle_upstream_error(response)
+            raise HTTPException(status_code=502, detail="Upstream error")
+        
+        # Collect full response
+        full_content = []
+        debug_log("开始收集完整响应内容")
+        response_completed = False
+        
+        try:
+            with SSEParser(response, debug_mode=settings.DEBUG_LOGGING) as parser:
+                for event in parser.iter_json_data(UpstreamData):
+                    upstream_data = event['data']
+                    
+                    if upstream_data.data.delta_content:
+                        content = upstream_data.data.delta_content
+                        
+                        if upstream_data.data.phase == "thinking":
+                            content = transform_thinking_content(content)
+                        
+                        if content:
+                            full_content.append(content)
+                    
+                    if upstream_data.data.done or upstream_data.data.phase == "done":
+                        debug_log("检测到完成信号，停止收集")
+                        response_completed = True
+                        break
+                        
+        except Exception as e:
+            debug_log(f"非流式响应收集异常: {e}")
+            if not full_content:
+                # 如果没有收集到任何内容，抛出异常
+                raise HTTPException(status_code=502, detail=f"Response collection failed: {str(e)}")
+            else:
+                debug_log(f"部分内容收集成功，继续处理 ({len(full_content)} 个片段)")
+        
+        if not response_completed and not full_content:
+            debug_log("响应未完成且无内容，可能是连接问题")
+            raise HTTPException(status_code=502, detail="Incomplete response from upstream")
+        
+        final_content = "".join(full_content)
+        debug_log(f"内容收集完成，最终长度: {len(final_content)}")
+        
+        # Handle tool calls for non-streaming
+        tool_calls = None
+        finish_reason = "stop"
+        message_content = final_content
+        
+        if self.has_tools:
+            tool_calls = extract_tool_invocations(final_content)
+            if tool_calls:
+                # Content must be null when tool_calls are present (OpenAI spec)
+                message_content = None
+                finish_reason = "tool_calls"
+                debug_log(f"提取到工具调用: {json.dumps(tool_calls, ensure_ascii=False)}")
+            else:
+                # Remove tool JSON from content
+                message_content = remove_tool_json_content(final_content)
+                if not message_content:
+                    message_content = final_content  # 保留原内容如果清理后为空
+        
+        # Build response
+        response_data = OpenAIResponse(
+            id=f"chatcmpl-{int(time.time())}",
+            object="chat.completion",
+            created=int(time.time()),
+            model=settings.PRIMARY_MODEL,
+            choices=[Choice(
+                index=0,
+                message=Message(
+                    role="assistant",
+                    content=message_content,
+                    tool_calls=tool_calls
+                ),
+                finish_reason=finish_reason
+            )],
+            usage=Usage()
+        )
+        
+        debug_log("非流式响应发送完成")
         return JSONResponse(content=response_data.model_dump(exclude_none=True))

app/core/zai_transformer.py

@@ -0,0 +1,340 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import json
+import time
+import uuid
+import random
+from datetime import datetime
+from typing import Dict, List, Any, Optional, Generator, AsyncGenerator
+import httpx
+import asyncio
+from fake_useragent import UserAgent
+
+from app.core.config import settings
+from app.utils.helpers import debug_log
+from app.core.token_manager import token_manager
+
+# 全局 UserAgent 实例（单例模式）
+_user_agent_instance = None
+
+
+def get_user_agent_instance() -> UserAgent:
+    """获取或创建 UserAgent 实例（单例模式）"""
+    global _user_agent_instance
+    if _user_agent_instance is None:
+        _user_agent_instance = UserAgent()
+    return _user_agent_instance
+
+
+def get_dynamic_headers(chat_id: str = "") -> Dict[str, str]:
+    """生成动态浏览器headers，包含随机User-Agent"""
+    ua = get_user_agent_instance()
+
+    # 随机选择浏览器类型，偏向Chrome和Edge
+    browser_choices = ["chrome", "chrome", "chrome", "edge", "edge", "firefox", "safari"]
+    browser_type = random.choice(browser_choices)
+
+    try:
+        if browser_type == "chrome":
+            user_agent = ua.chrome
+        elif browser_type == "edge":
+            user_agent = ua.edge
+        elif browser_type == "firefox":
+            user_agent = ua.firefox
+        elif browser_type == "safari":
+            user_agent = ua.safari
+        else:
+            user_agent = ua.random
+    except:
+        user_agent = ua.random
+
+    # 提取版本信息
+    chrome_version = "139"
+    edge_version = "139"
+
+    if "Chrome/" in user_agent:
+        try:
+            chrome_version = user_agent.split("Chrome/")[1].split(".")[0]
+        except:
+            pass
+
+    if "Edg/" in user_agent:
+        try:
+            edge_version = user_agent.split("Edg/")[1].split(".")[0]
+            sec_ch_ua = f'"Microsoft Edge";v="{edge_version}", "Chromium";v="{chrome_version}", "Not_A Brand";v="24"'
+        except:
+            sec_ch_ua = f'"Not_A Brand";v="8", "Chromium";v="{chrome_version}", "Google Chrome";v="{chrome_version}"'
+    elif "Firefox/" in user_agent:
+        sec_ch_ua = None  # Firefox不使用sec-ch-ua
+    else:
+        sec_ch_ua = f'"Not_A Brand";v="8", "Chromium";v="{chrome_version}", "Google Chrome";v="{chrome_version}"'
+
+    headers = {
+        "Content-Type": "application/json",
+        "Accept": "application/json, text/event-stream",
+        "User-Agent": user_agent,
+        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+        "X-FE-Version": "prod-fe-1.0.79",
+        "Origin": "https://chat.z.ai",
+    }
+
+    if sec_ch_ua:
+        headers["sec-ch-ua"] = sec_ch_ua
+        headers["sec-ch-ua-mobile"] = "?0"
+        headers["sec-ch-ua-platform"] = '"Windows"'
+
+    if chat_id:
+        headers["Referer"] = f"https://chat.z.ai/c/{chat_id}"
+    else:
+        headers["Referer"] = "https://chat.z.ai/"
+
+    return headers
+
+
+def generate_uuid() -> str:
+    """生成UUID v4"""
+    return str(uuid.uuid4())
+
+
+def get_auth_token_sync() -> str:
+    """同步获取认证令牌（用于非异步场景）"""
+    if settings.ANONYMOUS_MODE:
+        try:
+            headers = get_dynamic_headers()
+            with httpx.Client() as client:
+                response = client.get("https://chat.z.ai/api/v1/auths/", headers=headers, timeout=10.0)
+                if response.status_code == 200:
+                    data = response.json()
+                    token = data.get("token", "")
+                    if token:
+                        debug_log(f"获取访客令牌成功: {token[:20]}...")
+                        return token
+        except Exception as e:
+            debug_log(f"获取访客令牌失败: {e}")
+
+    # 使用token管理器获取备份令牌
+    token = token_manager.get_next_token()
+    if token:
+        debug_log(f"从token池获取令牌: {token[:20]}...")
+        return token
+
+    # 没有可用的token
+    debug_log("⚠️ 没有可用的备份token")
+    return ""
+
+
+class ZAITransformer:
+    """ZAI转换器类"""
+
+    def __init__(self):
+        """初始化转换器"""
+        self.name = "zai"
+        self.base_url = "https://chat.z.ai"
+        self.api_url = settings.API_ENDPOINT
+        self.auth_url = f"{self.base_url}/api/v1/auths/"
+
+        # 模型映射
+        self.model_mapping = {
+            settings.PRIMARY_MODEL: "0727-360B-API",  # GLM-4.5
+            settings.THINKING_MODEL: "0727-360B-API",  # GLM-4.5-Thinking
+            settings.SEARCH_MODEL: "0727-360B-API",  # GLM-4.5-Search
+            settings.AIR_MODEL: "0727-106B-API",  # GLM-4.5-Air
+        }
+
+    async def get_token(self) -> str:
+        """异步获取认证令牌"""
+        if settings.ANONYMOUS_MODE:
+            try:
+                headers = get_dynamic_headers()
+                async with httpx.AsyncClient() as client:
+                    response = await client.get(self.auth_url, headers=headers, timeout=10.0)
+                    if response.status_code == 200:
+                        data = response.json()
+                        token = data.get("token", "")
+                        if token:
+                            debug_log(f"获取访客令牌成功: {token[:20]}...")
+                            return token
+            except Exception as e:
+                debug_log(f"异步获取访客令牌失败: {e}")
+
+        # 使用token管理器获取备份令牌
+        token = token_manager.get_next_token()
+        if token:
+            debug_log(f"从token池获取令牌: {token[:20]}...")
+            return token
+
+        # 没有可用的token
+        debug_log("⚠️ 没有可用的备份token")
+        return ""
+
+    def mark_token_success(self, token: str):
+        """标记token使用成功"""
+        token_manager.mark_token_success(token)
+
+    def mark_token_failure(self, token: str, error: Exception = None):
+        """标记token使用失败"""
+        token_manager.mark_token_failed(token)
+
+    async def transform_request_in(self, request: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        转换OpenAI请求为z.ai格式
+        整合现有功能：模型映射、MCP服务器等
+        """
+        debug_log(f"🔄 开始转换 OpenAI 请求到 Z.AI 格式: {request.get('model', settings.PRIMARY_MODEL)} -> Z.AI")
+
+        # 获取认证令牌
+        token = await self.get_token()
+        debug_log(f"  使用令牌: {token[:20] if token else 'None'}...")
+
+        # 检查token是否有效
+        if not token:
+            debug_log("❌ 无法获取有效的认证令牌")
+            raise Exception("无法获取有效的认证令牌，请检查匿名模式配置或token池配置")
+
+        # 确定请求的模型特性
+        requested_model = request.get("model", settings.PRIMARY_MODEL)
+        is_thinking = requested_model == settings.THINKING_MODEL or request.get("reasoning", False)
+        is_search = requested_model == settings.SEARCH_MODEL
+        is_air = requested_model == settings.AIR_MODEL
+
+        # 获取上游模型ID（使用模型映射）
+        upstream_model_id = self.model_mapping.get(requested_model, "0727-360B-API")
+        debug_log(f"  模型映射: {requested_model} -> {upstream_model_id}")
+        debug_log(f"  模型特性检测: is_search={is_search}, is_thinking={is_thinking}, is_air={is_air}")
+
+        # 处理消息列表
+        debug_log(f"  开始处理 {len(request.get('messages', []))} 条消息")
+        messages = []
+        for idx, orig_msg in enumerate(request.get("messages", [])):
+            msg = orig_msg.copy()
+
+            # 处理system角色转换
+            if msg.get("role") == "system":
+                msg["role"] = "user"
+                content = msg.get("content")
+
+                if isinstance(content, list):
+                    msg["content"] = [
+                        {"type": "text", "text": "This is a system command, you must enforce compliance."}
+                    ] + content
+                elif isinstance(content, str):
+                    msg["content"] = f"This is a system command, you must enforce compliance.{content}"
+
+            # 处理user角色的图片内容
+            elif msg.get("role") == "user":
+                content = msg.get("content")
+                if isinstance(content, list):
+                    new_content = []
+                    for part_idx, part in enumerate(content):
+                        # 处理图片URL（支持base64和http URL）
+                        if (
+                            part.get("type") == "image_url"
+                            and part.get("image_url", {}).get("url")
+                            and isinstance(part["image_url"]["url"], str)
+                        ):
+                            debug_log(f"    消息[{idx}]内容[{part_idx}]: 检测到图片URL")
+                            # 直接传递图片内容
+                            new_content.append(part)
+                        else:
+                            new_content.append(part)
+                    msg["content"] = new_content
+
+            # 处理assistant消息中的reasoning_content
+            elif msg.get("role") == "assistant" and msg.get("reasoning_content"):
+                # 如果有reasoning_content，保留它
+                pass
+
+            messages.append(msg)
+
+        # 构建MCP服务器列表
+        mcp_servers = []
+        if is_search:
+            mcp_servers.append("deep-web-search")
+            debug_log(f"🔍 检测到搜索模型，添加 deep-web-search MCP 服务器")
+
+        debug_log(f"  MCP服务器列表: {mcp_servers}")
+            
+        # 构建上游请求体
+        chat_id = generate_uuid()
+
+        body = {
+            "stream": True,  # 总是使用流式
+            "model": upstream_model_id,  # 使用映射后的模型ID
+            "messages": messages,
+            "params": {},
+            "features": {
+                "image_generation": False,
+                "web_search": is_search,
+                "auto_web_search": is_search,
+                "preview_mode": False,
+                "flags": [],
+                "features": [],
+                "enable_thinking": is_thinking,
+            },
+            "background_tasks": {
+                "title_generation": False,
+                "tags_generation": False,
+            },
+            "mcp_servers": mcp_servers,  # 保留MCP服务器支持
+            "variables": {
+                "{{USER_NAME}}": "Guest",
+                "{{USER_LOCATION}}": "Unknown",
+                "{{CURRENT_DATETIME}}": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                "{{CURRENT_DATE}}": datetime.now().strftime("%Y-%m-%d"),
+                "{{CURRENT_TIME}}": datetime.now().strftime("%H:%M:%S"),
+                "{{CURRENT_WEEKDAY}}": datetime.now().strftime("%A"),
+                "{{CURRENT_TIMEZONE}}": "Asia/Shanghai",  # 使用更合适的时区
+                "{{USER_LANGUAGE}}": "zh-CN",
+            },
+            "model_item": {
+                "id": upstream_model_id,
+                "name": requested_model,
+                "owned_by": "z.ai"
+            },
+            "chat_id": chat_id,
+            "id": generate_uuid(),
+        }
+
+        # 处理工具支持
+        if settings.TOOL_SUPPORT and not is_thinking and request.get("tools"):
+            body["tools"] = request["tools"]
+            debug_log(f"启用工具支持: {len(request['tools'])} 个工具")
+        else:
+            body["tools"] = None
+
+        # 构建请求配置
+        dynamic_headers = get_dynamic_headers(chat_id)
+
+        config = {
+            "url": self.api_url,  # 使用原始URL
+            "headers": {
+                **dynamic_headers,  # 使用动态生成的headers
+                "Authorization": f"Bearer {token}",
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "Pragma": "no-cache",
+                "Sec-Fetch-Dest": "empty",
+                "Sec-Fetch-Mode": "cors",
+                "Sec-Fetch-Site": "same-origin",
+            },
+        }
+
+        debug_log("✅ 请求转换完成")
+
+        # 记录关键的请求信息用于调试
+        debug_log(f"  📋 发送到Z.AI的关键信息:")
+        debug_log(f"    - 上游模型: {body['model']}")
+        debug_log(f"    - MCP服务器: {body['mcp_servers']}")
+        debug_log(f"    - web_search: {body['features']['web_search']}")
+        debug_log(f"    - auto_web_search: {body['features']['auto_web_search']}")
+        debug_log(f"    - 消息数量: {len(body['messages'])}")
+        tools_count = len(body.get('tools') or [])
+        debug_log(f"    - 工具数量: {tools_count}")
+
+        # 返回转换后的请求数据
+        return {
+            "body": body,
+            "config": config,
+            "token": token
+        }

app/utils/sse_tool_handler.py

@@ -0,0 +1,692 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+SSE Tool Handler - 处理工具调用的SSE流
+基于 Z.AI 原生的 edit_index 和 edit_content 机制，更原生地处理工具调用
+"""
+
+import json
+import re
+import time
+from typing import Dict, Any, Optional, Generator, List
+
+from app.utils.helpers import debug_log
+
+
+class SSEToolHandler:
+
+    def __init__(self, chat_id: str, model: str):
+        self.chat_id = chat_id
+        self.model = model
+
+        # 工具调用状态
+        self.has_tool_call = False
+        self.tool_call_usage = None  # 工具调用的usage信息
+        self.content_index = 0
+        self.has_thinking = False
+
+        self.content_buffer = bytearray()  # 使用字节数组提高性能
+        self.last_edit_index = 0  # 上次编辑的位置
+
+        # 工具调用解析状态
+        self.active_tools = {}  # 活跃的工具调用 {tool_id: tool_info}
+        self.completed_tools = []  # 已完成的工具调用
+        self.tool_blocks_cache = {}  # 缓存解析的工具块
+
+    def process_tool_call_phase(self, data: Dict[str, Any], is_stream: bool = True) -> Generator[str, None, None]:
+        """
+        处理tool_call阶段
+        """
+        if not self.has_tool_call:
+            self.has_tool_call = True
+            debug_log("🔧 进入工具调用阶段")
+
+        edit_content = data.get("edit_content", "")
+        edit_index = data.get("edit_index", 0)
+
+        if not edit_content:
+            return
+
+        # debug_log(f"📦 接收内容片段 [index={edit_index}]: {edit_content[:1000]}...")
+
+        # 更新内容缓冲区
+        self._apply_edit_to_buffer(edit_index, edit_content)
+
+        # 尝试解析和处理工具调用
+        yield from self._process_tool_calls_from_buffer(is_stream)
+
+    def _apply_edit_to_buffer(self, edit_index: int, edit_content: str):
+        """
+        在指定位置替换/插入内容更新内容缓冲区
+        """
+        edit_bytes = edit_content.encode('utf-8')
+        required_length = edit_index + len(edit_bytes)
+
+        # 扩展缓冲区到所需长度（如果需要）
+        if len(self.content_buffer) < edit_index:
+            # 如果edit_index超出当前缓冲区，用空字节填充
+            self.content_buffer.extend(b'\x00' * (edit_index - len(self.content_buffer)))
+
+        # 确保缓冲区足够长以容纳新内容
+        if len(self.content_buffer) < required_length:
+            self.content_buffer.extend(b'\x00' * (required_length - len(self.content_buffer)))
+
+        # 在指定位置替换内容（不是插入，而是覆盖）
+        end_index = edit_index + len(edit_bytes)
+        self.content_buffer[edit_index:end_index] = edit_bytes
+
+        # debug_log(f"📝 缓冲区更新 [index={edit_index}, 长度={len(self.content_buffer)}]")
+
+    def _process_tool_calls_from_buffer(self, is_stream: bool) -> Generator[str, None, None]:
+        """
+        从内容缓冲区中解析和处理工具调用
+        """
+        try:
+            # 解码内容并清理空字节
+            content_str = self.content_buffer.decode('utf-8', errors='ignore').replace('\x00', '')
+            yield from self._extract_and_process_tools(content_str, is_stream)
+        except Exception as e:
+            debug_log(f"📦 内容解析暂时失败，等待更多数据: {e}")
+            # 不抛出异常，继续等待更多数据
+
+    def _extract_and_process_tools(self, content_str: str, is_stream: bool) -> Generator[str, None, None]:
+        """
+        从内容字符串中提取和处理工具调用
+        """
+        # 查找所有 glm_block，包括不完整的
+        pattern = r'<glm_block\s*>(.*?)(?:</glm_block>|$)'
+        matches = re.findall(pattern, content_str, re.DOTALL)
+
+        for block_content in matches:
+            # 尝试解析每个块
+            yield from self._process_single_tool_block(block_content, is_stream)
+
+    def _process_single_tool_block(self, block_content: str, is_stream: bool) -> Generator[str, None, None]:
+        """
+        处理单个工具块，支持增量解析
+        """
+        try:
+            # 尝试修复和解析完整的JSON
+            fixed_content = self._fix_json_structure(block_content)
+            tool_data = json.loads(fixed_content)
+            metadata = tool_data.get("data", {}).get("metadata", {})
+
+            tool_id = metadata.get("id", "")
+            tool_name = metadata.get("name", "")
+            arguments_raw = metadata.get("arguments", "{}")
+
+            if not tool_id or not tool_name:
+                return
+
+            debug_log(f"🎯 解析完整工具块: {tool_name}(id={tool_id}), 参数: {arguments_raw}")
+
+            # 检查是否是新工具或更新的工具
+            yield from self._handle_tool_update(tool_id, tool_name, arguments_raw, is_stream)
+
+        except json.JSONDecodeError as e:
+            debug_log(f"📦 JSON解析失败: {e}, 尝试部分解析")
+            # JSON 不完整，��试部分解析
+            yield from self._handle_partial_tool_block(block_content, is_stream)
+        except Exception as e:
+            debug_log(f"📦 工具块处理失败: {e}")
+
+    def _fix_json_structure(self, content: str) -> str:
+        """
+        修复JSON结构中的常见问题
+        """
+        if not content:
+            return content
+
+        # 计算括号平衡
+        open_braces = content.count('{')
+        close_braces = content.count('}')
+
+        # 如果闭括号多于开括号，移除多余的闭括号
+        if close_braces > open_braces:
+            excess = close_braces - open_braces
+            fixed_content = content
+            for _ in range(excess):
+                # 从右侧移除多余的闭括号
+                last_brace_pos = fixed_content.rfind('}')
+                if last_brace_pos != -1:
+                    fixed_content = fixed_content[:last_brace_pos] + fixed_content[last_brace_pos + 1:]
+            return fixed_content
+
+        return content
+
+    def _handle_tool_update(self, tool_id: str, tool_name: str, arguments_raw: str, is_stream: bool) -> Generator[str, None, None]:
+        """
+        处理工具的创建或更新 - 更可靠的参数完整性检查
+        """
+        # 解析参数
+        try:
+            if isinstance(arguments_raw, str):
+                # 先处理转义和清理
+                cleaned_args = self._clean_arguments_string(arguments_raw)
+                arguments = json.loads(cleaned_args) if cleaned_args.strip() else {}
+            else:
+                arguments = arguments_raw
+        except json.JSONDecodeError:
+            debug_log(f"📦 参数解析失败，暂不处理: {arguments_raw}")
+            # 参数解析失败时，不创建或更新工具，等待更完整的数据
+            return
+
+        # 检查参数是否看起来完整（基本的完整性验证）
+        is_args_complete = self._is_arguments_complete(arguments, arguments_raw)
+
+        # 检查是否是新工具
+        if tool_id not in self.active_tools:
+            debug_log(f"🎯 发现新工具: {tool_name}(id={tool_id}), 参数完整性: {is_args_complete}")
+
+            self.active_tools[tool_id] = {
+                "id": tool_id,
+                "name": tool_name,
+                "arguments": arguments,
+                "arguments_raw": arguments_raw,
+                "status": "active",
+                "sent_start": False,
+                "last_sent_args": {},  # 跟踪上次发送的参数
+                "args_complete": is_args_complete,
+                "pending_send": True  # 标记需要发送
+            }
+
+            # 只有在参数看起来完整时才发送工具开始信号
+            if is_stream and is_args_complete:
+                yield self._create_tool_start_chunk(tool_id, tool_name, arguments)
+                self.active_tools[tool_id]["sent_start"] = True
+                self.active_tools[tool_id]["last_sent_args"] = arguments.copy()
+                self.active_tools[tool_id]["pending_send"] = False
+                debug_log(f"📤 发送完整工具开始: {tool_name}(id={tool_id})")
+
+        else:
+            # 更新现有工具
+            current_tool = self.active_tools[tool_id]
+
+            # 检查是否有实质性改进
+            if self._is_significant_improvement(current_tool["arguments"], arguments,
+                                               current_tool["arguments_raw"], arguments_raw):
+                debug_log(f"🔄 工具参数有实质性改进: {tool_name}(id={tool_id})")
+
+                current_tool["arguments"] = arguments
+                current_tool["arguments_raw"] = arguments_raw
+                current_tool["args_complete"] = is_args_complete
+
+                # 如果之前没有发送过开始信号，且现在参数完整，发送开始信号
+                if is_stream and not current_tool["sent_start"] and is_args_complete:
+                    yield self._create_tool_start_chunk(tool_id, tool_name, arguments)
+                    current_tool["sent_start"] = True
+                    current_tool["last_sent_args"] = arguments.copy()
+                    current_tool["pending_send"] = False
+                    debug_log(f"📤 发送延迟的工具开始: {tool_name}(id={tool_id})")
+
+                # 如果已经发送过开始信号，且参数有显著改进，发送参数更新
+                elif is_stream and current_tool["sent_start"] and is_args_complete:
+                    if self._should_send_argument_update(current_tool["last_sent_args"], arguments):
+                        yield self._create_tool_arguments_chunk(tool_id, arguments)
+                        current_tool["last_sent_args"] = arguments.copy()
+                        debug_log(f"📤 发送参数更新: {tool_name}(id={tool_id})")
+
+    def _is_arguments_complete(self, arguments: Dict[str, Any], arguments_raw: str) -> bool:
+        """
+        检查参数是否看起来完整
+        """
+        if not arguments:
+            return False
+
+        # 检查原始字符串是否看起来完整
+        if not arguments_raw or not arguments_raw.strip():
+            return False
+
+        # 检查是否有明显的截断迹象
+        raw_stripped = arguments_raw.strip()
+
+        # 如果原始字符串不以}结尾，可能是截断的
+        if not raw_stripped.endswith('}') and not raw_stripped.endswith('"'):
+            return False
+
+        # 检查是否有不完整的URL（常见的截断情况）
+        for key, value in arguments.items():
+            if isinstance(value, str):
+                # 检查URL是否看起来完整
+                if 'http' in value.lower():
+                    # 如果URL太短或以不完整的域名结尾，可能是截断的
+                    if len(value) < 10 or value.endswith('.go') or value.endswith('.goo'):
+                        return False
+
+                # 检查其他可能的截断迹象
+                if len(value) > 0 and value[-1] in ['.', '/', ':', '=']:
+                    # 以这些字符结尾可能表示截断
+                    return False
+
+        return True
+
+    def _is_significant_improvement(self, old_args: Dict[str, Any], new_args: Dict[str, Any],
+                                   old_raw: str, new_raw: str) -> bool:
+        """
+        检查新参数是否比旧参数有显著改进
+        """
+        # 如果新参数为空，不是改进
+        if not new_args:
+            return False
+        
+        if len(new_args) > len(old_args):
+            return True
+
+        # 检查值的改进
+        for key, new_value in new_args.items():
+            old_value = old_args.get(key, "")
+
+            if isinstance(new_value, str) and isinstance(old_value, str):
+                # 如果新值明显更长且更完整，是改进
+                if len(new_value) > len(old_value) + 5:  # 至少长5个字符才算显著改进
+                    return True
+
+                # 如果旧值看起来是截断的，新值更完整，是改进
+                if old_value.endswith(('.go', '.goo', '.com/', 'http')) and len(new_value) > len(old_value):
+                    return True
+
+        # 检查原始字符串的改进
+        if len(new_raw) > len(old_raw) + 10:  # 原始字符串显著增长
+            return True
+
+        return False
+
+    def _should_send_argument_update(self, last_sent: Dict[str, Any], new_args: Dict[str, Any]) -> bool:
+        """
+        判断是否应该发送参数更新 - 更严格的标准
+        """
+        # 如果参数完全相同，不发送
+        if last_sent == new_args:
+            return False
+
+        # 如果新参数为空但之前有参数，不发送（避免倒退）
+        if not new_args and last_sent:
+            return False
+
+        # 如果新参数有更多键，发送更新
+        if len(new_args) > len(last_sent):
+            return True
+
+        # 检查是否有值变得显著更完整
+        for key, new_value in new_args.items():
+            last_value = last_sent.get(key, "")
+            if isinstance(new_value, str) and isinstance(last_value, str):
+                # 只有在值显著增长时才发送更新（避免微小变化）
+                if len(new_value) > len(last_value) + 5:
+                    return True
+            elif new_value != last_value and new_value:  # 确保新值不为空
+                return True
+
+        return False
+
+    def _handle_partial_tool_block(self, block_content: str, is_stream: bool) -> Generator[str, None, None]:
+        """
+        处理不完整的工具块，尝试提取可用信息
+        """
+        try:
+            # 尝试提取工具ID和名称
+            id_match = re.search(r'"id":\s*"([^"]+)"', block_content)
+            name_match = re.search(r'"name":\s*"([^"]+)"', block_content)
+
+            if id_match and name_match:
+                tool_id = id_match.group(1)
+                tool_name = name_match.group(1)
+
+                # 尝试提取参数部分
+                args_match = re.search(r'"arguments":\s*"([^"]*)', block_content)
+                partial_args = args_match.group(1) if args_match else ""
+
+                debug_log(f"📦 部分工具块: {tool_name}(id={tool_id}), 部分参数: {partial_args[:50]}")
+
+                # 如果是新工具，先创建记录
+                if tool_id not in self.active_tools:
+                    # 尝试解析部分参数为字典
+                    partial_args_dict = self._parse_partial_arguments(partial_args)
+
+                    self.active_tools[tool_id] = {
+                        "id": tool_id,
+                        "name": tool_name,
+                        "arguments": partial_args_dict,
+                        "status": "partial",
+                        "sent_start": False,
+                        "last_sent_args": {},
+                        "args_complete": False,
+                        "partial_args": partial_args
+                    }
+
+                    if is_stream:
+                        yield self._create_tool_start_chunk(tool_id, tool_name, partial_args_dict)
+                        self.active_tools[tool_id]["sent_start"] = True
+                        self.active_tools[tool_id]["last_sent_args"] = partial_args_dict.copy()
+                else:
+                    # 更新部分参数
+                    self.active_tools[tool_id]["partial_args"] = partial_args
+                    # 尝试更新解析的参数
+                    new_partial_dict = self._parse_partial_arguments(partial_args)
+                    if new_partial_dict != self.active_tools[tool_id]["arguments"]:
+                        self.active_tools[tool_id]["arguments"] = new_partial_dict
+
+        except Exception as e:
+            debug_log(f"📦 部分块解析失败: {e}")
+
+    def _clean_arguments_string(self, arguments_raw: str) -> str:
+        """
+        清理和标准化参数字符串，改进对不完整JSON的处理
+        """
+        if not arguments_raw:
+            return "{}"
+
+        # 移除首尾空白
+        cleaned = arguments_raw.strip()
+
+        # 处理特殊值
+        if cleaned.lower() == "null":
+            return "{}"
+
+        # 处理转义的JSON字符串
+        if cleaned.startswith('{\\"') and cleaned.endswith('\\"}'):
+            # 这是一个转义的JSON字符串，需要反转义
+            cleaned = cleaned.replace('\\"', '"')
+        elif cleaned.startswith('"{\\"') and cleaned.endswith('\\"}'):
+            # 双重转义的情况
+            cleaned = cleaned[1:-1].replace('\\"', '"')
+        elif cleaned.startswith('"') and cleaned.endswith('"'):
+            # 简单的引号包围，去除外层引号
+            cleaned = cleaned[1:-1]
+
+        # 处理不完整的JSON字符串
+        cleaned = self._fix_incomplete_json(cleaned)
+
+        # 标准化空格（移除JSON中的多余空格，但保留字符串值中的空格）
+        try:
+            # 先尝试解析，然后重新序列化以标准化格式
+            parsed = json.loads(cleaned)
+            if parsed is None:
+                return "{}"
+            cleaned = json.dumps(parsed, ensure_ascii=False, separators=(',', ':'))
+        except json.JSONDecodeError:
+            # 如果解析失败，只做基本的空格清理
+            debug_log(f"📦 JSON标准化失败，保持原样: {cleaned[:50]}...")
+
+        return cleaned
+
+    def _fix_incomplete_json(self, json_str: str) -> str:
+        """
+        修复不完整的JSON字符串
+        """
+        if not json_str:
+            return "{}"
+
+        # 确保以{开头
+        if not json_str.startswith('{'):
+            json_str = '{' + json_str
+
+        # 处理不完整的字符串值
+        if json_str.count('"') % 2 != 0:
+            # 奇数个引号，可能有未闭合的字符串
+            json_str += '"'
+
+        # 确保以}结尾
+        if not json_str.endswith('}'):
+            json_str += '}'
+
+        return json_str
+
+    def _parse_partial_arguments(self, arguments_raw: str) -> Dict[str, Any]:
+        """
+        解析不完整的参数字符串，尽可能提取有效信息
+        """
+        if not arguments_raw or arguments_raw.strip() == "" or arguments_raw.strip().lower() == "null":
+            return {}
+
+        try:
+            # 先尝试清理字符串
+            cleaned = self._clean_arguments_string(arguments_raw)
+            result = json.loads(cleaned)
+            # 确保返回字典类型
+            return result if isinstance(result, dict) else {}
+        except json.JSONDecodeError:
+            pass
+
+        try:
+            # 尝试修复常见的JSON问题
+            fixed_args = arguments_raw.strip()
+
+            # 处理转义字符
+            if '\\' in fixed_args:
+                fixed_args = fixed_args.replace('\\"', '"')
+
+            # 如果不是以{开头，添加{
+            if not fixed_args.startswith('{'):
+                fixed_args = '{' + fixed_args
+
+            # 如果不是以}结尾，尝试添加}
+            if not fixed_args.endswith('}'):
+                # 计算未闭合的引号和括号
+                quote_count = fixed_args.count('"') - fixed_args.count('\\"')
+                if quote_count % 2 != 0:
+                    fixed_args += '"'
+                fixed_args += '}'
+
+            return json.loads(fixed_args)
+        except json.JSONDecodeError:
+            # 尝试提取键值对
+            return self._extract_key_value_pairs(arguments_raw)
+        except Exception:
+            # 如果所有方法都失败，返回空字典
+            return {}
+
+    def _extract_key_value_pairs(self, text: str) -> Dict[str, Any]:
+        """
+        从文本中提取键值对，作为最后的解析尝试
+        """
+        result = {}
+        try:
+            # 使用正则表达式提取简单的键值对
+            import re
+
+            # 匹配 "key": "value" 或 "key": value 格式
+            pattern = r'"([^"]+)":\s*"([^"]*)"'
+            matches = re.findall(pattern, text)
+
+            for key, value in matches:
+                result[key] = value
+
+            # 匹配数字值
+            pattern = r'"([^"]+)":\s*(\d+)'
+            matches = re.findall(pattern, text)
+
+            for key, value in matches:
+                try:
+                    result[key] = int(value)
+                except ValueError:
+                    result[key] = value
+
+            # 匹配布尔值
+            pattern = r'"([^"]+)":\s*(true|false)'
+            matches = re.findall(pattern, text)
+
+            for key, value in matches:
+                result[key] = value.lower() == 'true'
+
+        except Exception:
+            pass
+
+        return result
+
+    def _complete_active_tools(self, is_stream: bool) -> Generator[str, None, None]:
+        """
+        完成所有活跃的工具调用 - 处理待发送的工具
+        """
+        tools_to_send = []
+
+        for tool_id, tool in self.active_tools.items():
+            # 如果工具还没有发送过且参数看起来完整，现在发送
+            if is_stream and tool.get("pending_send", False) and not tool.get("sent_start", False):
+                if tool.get("args_complete", False):
+                    debug_log(f"📤 完成时发送待发送工具: {tool['name']}(id={tool_id})")
+                    yield self._create_tool_start_chunk(tool_id, tool["name"], tool["arguments"])
+                    tool["sent_start"] = True
+                    tool["pending_send"] = False
+                    tools_to_send.append(tool)
+                else:
+                    debug_log(f"⚠️ 跳过不完整的工具: {tool['name']}(id={tool_id})")
+
+            tool["status"] = "completed"
+            self.completed_tools.append(tool)
+            debug_log(f"✅ 完成工具调用: {tool['name']}(id={tool_id})")
+
+        self.active_tools.clear()
+
+        if is_stream and (self.completed_tools or tools_to_send):
+            # 发送工具完成信号
+            yield self._create_tool_finish_chunk()
+
+    def process_other_phase(self, data: Dict[str, Any], is_stream: bool = True) -> Generator[str, None, None]:
+        """
+        处理other阶段 - 检测工具调用结束和状态更新
+        """
+        edit_content = data.get("edit_content", "")
+        edit_index = data.get("edit_index", 0)
+        usage = data.get("usage")
+
+        # 保存usage信息
+        if self.has_tool_call and usage:
+            self.tool_call_usage = usage
+            debug_log(f"💾 保存工具调用usage: {usage}")
+
+        # 如果有edit_content，继续更新内容缓冲区
+        if edit_content:
+            self._apply_edit_to_buffer(edit_index, edit_content)
+            # 继续处理可能的工具调用更新
+            yield from self._process_tool_calls_from_buffer(is_stream)
+
+        # 检测工具调用结束的多种标记
+        if self.has_tool_call and self._is_tool_call_finished(edit_content):
+            debug_log("🏁 检测到工具调用结束")
+
+            # 完成所有活跃的工具
+            yield from self._complete_active_tools(is_stream)
+
+            if is_stream:
+                debug_log("🏁 发送工具调用完成信号")
+                yield "data: [DONE]"
+
+            # 重置工具调用状态
+            self.has_tool_call = False
+
+    def _is_tool_call_finished(self, edit_content: str) -> bool:
+        """
+        检测工具调用是否结束的多种标记
+        """
+        if not edit_content:
+            return False
+
+        # 检测各种结束标记
+        end_markers = [
+            "null,",  # 原有的结束标记
+            '"status": "completed"',  # 状态完成标记
+            '"is_error": false',  # 错误状态标记
+        ]
+
+        for marker in end_markers:
+            if marker in edit_content:
+                debug_log(f"🔍 检测到结束标记: {marker}")
+                return True
+
+        # 检查是否所有工具都有完整的结构
+        if self.active_tools and '"status": "completed"' in self.content_buffer:
+            return True
+
+        return False
+
+    def _reset_all_state(self):
+        """重置所有状态"""
+        self.has_tool_call = False
+        self.tool_call_usage = None
+        self.content_index = 0
+        self.content_buffer = bytearray()
+        self.last_edit_index = 0
+        self.active_tools.clear()
+        self.completed_tools.clear()
+        self.tool_blocks_cache.clear()
+
+    def _create_tool_start_chunk(self, tool_id: str, tool_name: str, initial_args: Dict[str, Any] = None) -> str:
+        """创建工具调用开始的chunk，支持初始参数"""
+        # 使用提供的初始参数，如果没有则使用空字典
+        args_dict = initial_args or {}
+        args_str = json.dumps(args_dict, ensure_ascii=False)
+
+        chunk = {
+            "choices": [
+                {
+                    "delta": {
+                        "role": "assistant",
+                        "content": None,
+                        "tool_calls": [
+                            {
+                                "id": tool_id,
+                                "type": "function",
+                                "function": {"name": tool_name, "arguments": args_str},
+                            }
+                        ],
+                    },
+                    "finish_reason": None,
+                    "index": self.content_index,
+                    "logprobs": None,
+                }
+            ],
+            "created": int(time.time()),
+            "id": self.chat_id,
+            "model": self.model,
+            "object": "chat.completion.chunk",
+            "system_fingerprint": "fp_zai_001",
+        }
+        return f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
+
+    def _create_tool_arguments_chunk(self, tool_id: str, arguments: Dict) -> str:
+        """创建工具参数的chunk - 只包含参数更新，不包含函数名"""
+        chunk = {
+            "choices": [
+                {
+                    "delta": {
+                        "tool_calls": [
+                            {
+                                "id": tool_id,
+                                "function": {"arguments": json.dumps(arguments, ensure_ascii=False)},
+                            }
+                        ],
+                    },
+                    "finish_reason": None,
+                    "index": self.content_index,
+                    "logprobs": None,
+                }
+            ],
+            "created": int(time.time()),
+            "id": self.chat_id,
+            "model": self.model,
+            "object": "chat.completion.chunk",
+            "system_fingerprint": "fp_zai_001",
+        }
+        return f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
+
+    def _create_tool_finish_chunk(self) -> str:
+        """创建工具调用完成的chunk"""
+        chunk = {
+            "choices": [
+                {
+                    "delta": {"role": "assistant", "content": None, "tool_calls": []},
+                    "finish_reason": "tool_calls",
+                    "index": 0,
+                    "logprobs": None,
+                }
+            ],
+            "created": int(time.time()),
+            "id": self.chat_id,
+            "usage": self.tool_call_usage or None,
+            "model": self.model,
+            "object": "chat.completion.chunk",
+            "system_fingerprint": "fp_zai_001",
+        }
+        return f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"

pyproject.toml

@@ -1,64 +0,0 @@
-[build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
-
-[project]
-name = "z-ai2api-python"
-version = "0.1.0"
-description = "一个为 Z.ai 提供 OpenAI 兼容接口的 Python 代理服务"
-readme = "README.md"
-requires-python = ">=3.9,<=3.12"
-license = { text = "MIT" }
-authors = [{ name = "Contributors" }]
-classifiers = [
-    "Development Status :: 4 - Beta",
-    "Intended Audience :: Developers",
-    "License :: OSI Approved :: MIT License",
-    "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
-    "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.12",
-    "Topic :: Internet :: WWW/HTTP :: HTTP Servers",
-    "Topic :: Software Development :: Libraries :: Python Modules",
-]
-dependencies = [
-    "fastapi==0.104.1",
-    "granian[reload]==2.5.2",
-    "requests==2.32.5",
-    "pydantic==2.11.7",
-    "pydantic-settings==2.10.1",
-    "pydantic-core==2.33.2",
-    "typing-inspection==0.4.1",
-    "fake-useragent==2.2.0",
-]
-
-[project.scripts]
-z-ai2api = "main:app"
-
-[tool.hatch.build.targets.wheel]
-packages = ["."]
-
-[tool.uv]
-dev-dependencies = [
-    "pytest>=7.0.0",
-    "pytest-asyncio>=0.21.0",
-    "requests>=2.30.0",
-    "ruff>=0.1.0",
-]
-
-[tool.ruff]
-line-length = 88
-target-version = "py38"
-select = ["E", "F", "I", "B"]
-ignore = []
-
-[tool.ruff.isort]
-known-first-party = []
-
-[tool.pytest.ini_options]
-asyncio_mode = "auto"
-testpaths = ["tests"]
-python_files = ["test_*.py"]
-python_functions = ["test_*"]