From 755b1155dd4ad8a6b1f699023463dae69d3f9d76 Mon Sep 17 00:00:00 2001 From: rsesot Date: Wed, 16 Jul 2025 15:40:02 +0800 Subject: [PATCH] ver 1.0.2 --- README.md | 30 +++++- api/glm.py | 76 ++++++++++++++ api/mthreads.py | 122 +++++++++++++++++++++++ api/qwen.py | 68 +++++++++++++ config.py | 20 ++++ main.py | 96 ++++++++++++++++++ models.py | 25 +++++ requirements.txt | 3 + utils.py | 252 +++++++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 691 insertions(+), 1 deletion(-) create mode 100644 api/glm.py create mode 100644 api/mthreads.py create mode 100644 api/qwen.py create mode 100644 config.py create mode 100644 main.py create mode 100644 models.py create mode 100644 requirements.txt create mode 100644 utils.py diff --git a/README.md b/README.md index f1a8eeb..611f2b0 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,30 @@ -# nekochatapi +# > x < +乱写的,用于统一简化API交互,主要用于neko.ci的misskey翻译用 + +示例 +``` +rsesot@babydoll:~$ curl -N -X POST http://localhost:3060/api/mthreads/chat -H "Content-Type: application/json" -d '{ + "prompt": [ + "你说一句日语冷笑话" + ], + "temperature": 0.7, + "top_p": 0.9, + "max_tokens": 1024 +}' +event: complete +data: ["ちょっと変わった日本の冷えたジョークです:\n\nなぜか冬に氷が暖かいと言いますか？\n\n氷は冬にしか作れないからです！ 😄❄️\n\n（訳注: 冬にしか作れないから「暖かい」と冗談を言っています。）"] +``` +``` +rsesot@babydoll:~$ curl -N -X POST http://localhost:3060/api/glm/chat -H "Content-Type: application/json" -d '{ + "prompt": [ + "你说一句日语冷笑话" + ], + "temperature": 0.5, + "top_p": 0.9, + "max_tokens": 1024, + "web_search": true +}' +event: complete +data: ["日本語の冷笑话です：\n「猿が電車に乗ると、猿の毛が落ちて電車の座席に溜まります。すると、次の乗客が「猿の毛が落ちてる！」って言うと、猿が「いや、私の毛じゃないんだよ！」って言うけど、誰も信じてくれないんです。」（猿が電車に乗ると、猿の毛が落ちて電車の座席に溜まります。すると、次の乗客が「猿の毛が落ちてる！」って言うと、猿が「いや、私の毛じゃないんだよ！」って言うけど、誰も信じてくれないんです。）"] +``` diff --git a/api/glm.py b/api/glm.py new file mode 100644 index 0000000..32dbfeb --- /dev/null +++ b/api/glm.py @@ -0,0 +1,76 @@ +import logging +import uuid +import json + +from typing import Any, Dict +from fastapi import APIRouter, HTTPException, status +from fastapi.responses import StreamingResponse + +from models import ChatRequest +from utils import convert_prompt_to_messages, standard_sse_generator +from config import GLM_API_KEY, GLM_UPSTREAM_URL + +router = APIRouter() +logger = logging.getLogger("proxy") + + +@router.post("/glm/chat") +async def proxy_glm_chat(request: ChatRequest): + + request_id = f"glm-{uuid.uuid4()}" + + if not GLM_API_KEY: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="GLM API key is not configured.", + ) + + client_request_body = request.model_dump() + logger.debug( + f"[{request_id}] === GLM New Request ===\n" + f"[{request_id}] Client Request Body:\n" + f"{json.dumps(client_request_body, indent=2, ensure_ascii=False)}" + ) + + messages = convert_prompt_to_messages(request.prompt) + logger.debug( + f"[{request_id}] Agent Transformed Messages:\n" + f"{json.dumps(messages, indent=2, ensure_ascii=False)}" + ) + + # 基础的请求体 + payload: Dict[str, Any] = { + "model": "glm-4-flash", + "messages": messages, + "stream": True, + "temperature": request.temperature, + "top_p": request.top_p, + } + + # 如果请求中指定了 max_tokens ,则添加到 payload + if request.max_tokens: + payload["max_tokens"] = request.max_tokens + + # 处理 GLM web_search 功能 + if request.web_search: + tool = {"type": "web_search", "web_search": {}} + if request.web_search is True: + # 如果是简单的 true,则启用默认的网络搜索 + tool["web_search"] = {"enable": True} + elif isinstance(request.web_search, dict): + # 如果是字典,则使用用户提供的配置 + tool["web_search"] = request.web_search + # 确保 enable 标志为 True + tool["web_search"]["enable"] = True + payload["tools"] = [tool] + + logger.debug( + f"[{request_id}] Upstream Payload:\n" + f"{json.dumps(payload, indent=2, ensure_ascii=False)}" + ) + + headers = {"Authorization": f"Bearer {GLM_API_KEY}"} + return StreamingResponse( + standard_sse_generator(GLM_UPSTREAM_URL, payload, headers, "GLM", request_id), + media_type="text/event-stream", + ) diff --git a/api/mthreads.py b/api/mthreads.py new file mode 100644 index 0000000..ed1a6b8 --- /dev/null +++ b/api/mthreads.py @@ -0,0 +1,122 @@ +import json +import logging +import uuid +import httpx + +from fastapi import APIRouter +from fastapi.responses import StreamingResponse + +from config import MTHREADS_UPSTREAM_URL +from models import ChatRequest +from utils import ( + UpstreamAPIError, + sanitize_mthreads_response, + handle_upstream_response, + generate_error_sse, + generate_network_error_sse, +) + +router = APIRouter() +logger = logging.getLogger("proxy") + + +@router.post("/mthreads/chat") +async def proxy_mthreads_chat(request: ChatRequest): + + request_id = f"mthreads-{uuid.uuid4()}" + client_request_body = request.model_dump() + + logger.debug( + f"[{request_id}] === MThreads New Request ===\n" + f"[{request_id}] Client Request Body:\n" + f"{json.dumps(client_request_body, indent=2, ensure_ascii=False)}" + ) + + prompt_str = "\n".join(request.prompt) + mthreads_payload = { + "data": [prompt_str, request.temperature, request.top_p, request.max_tokens] + } + logger.debug( + f"[{request_id}] Agent Transformed Payload (to be sent to upstream):\n" + f"{json.dumps(mthreads_payload, indent=2, ensure_ascii=False)}" + ) + + try: + # --- 第一步: POST 请求获取 event_id --- + async with httpx.AsyncClient(timeout=60.0) as client: + response = await client.post( + MTHREADS_UPSTREAM_URL, + json=mthreads_payload, + headers={"Content-Type": "application/json"}, + ) + await handle_upstream_response(response) + response_json = response.json() + logger.debug( + f"[{request_id}] Upstream Initial Response:\n" + f"{json.dumps(response_json, indent=2, ensure_ascii=False)}" + ) + event_id = response_json.get("event_id") + if not event_id: + raise UpstreamAPIError( + status_code=502, + message="Failed to get event_id from MThreads response.", + details=response_json, + ) + except httpx.RequestError as e: + return StreamingResponse( + generate_network_error_sse(e, request_id), + media_type="text/event-stream", + ) + except (UpstreamAPIError, json.JSONDecodeError) as e: + if isinstance(e, json.JSONDecodeError): + e = UpstreamAPIError(status_code=502, message="MThreads returned a non-JSON response.") + return StreamingResponse(generate_error_sse(e, request_id), media_type="text/event-stream") + + async def stream_generator(): + """ + 第一次请求获取 event_id 后，轮询流式数据 + """ + # --- 第二步: GET 请求流式数据 --- + stream_url = f"{MTHREADS_UPSTREAM_URL}/{event_id}" + logger.debug(f"[{request_id}] Streaming from URL: {stream_url}") + try: + async with httpx.AsyncClient(timeout=300.0) as client: + async with client.stream("GET", stream_url) as response: + await handle_upstream_response(response) + line_iterator = response.aiter_lines() + async for line in line_iterator: + + if line.startswith("event: complete"): + data_line = await line_iterator.__anext__() + logger.debug(f"[{request_id}] Upstream Raw Data Line: {data_line}") + raw_text = json.loads(data_line.split(":", 1)[1].strip())[0] + logger.debug(f"[{request_id}] Upstream Parsed Raw Text: '{raw_text}'") + + # 清理 MThreads 返回的非标准文本 + final_text = sanitize_mthreads_response(raw_text) + logger.debug(f"[{request_id}] Agent Cleaned Text: '{final_text}'") + + formatted_data = json.dumps([final_text], ensure_ascii=False) + chunk_for_client_event = f"event: complete\n" + chunk_for_client_data = f"data: {formatted_data}\n\n" + logger.debug( + f"[{request_id}] Yielding to Client:\n{chunk_for_client_event}{chunk_for_client_data}" + ) + yield chunk_for_client_event + yield chunk_for_client_data + # 收到 complete 事件后,认为流已结束 + break + except httpx.RequestError as e: + async for item in generate_network_error_sse(e, request_id): + yield item + except (UpstreamAPIError, StopAsyncIteration, json.JSONDecodeError) as e: + if isinstance(e, json.JSONDecodeError): + err = UpstreamAPIError(status_code=502, message="Failed to parse MThreads stream response.") + elif isinstance(e, StopAsyncIteration): + err = UpstreamAPIError(status_code=502, message="MThreads stream was interrupted.") + else: + err = e + async for item in generate_error_sse(err, request_id): + yield item + + return StreamingResponse(stream_generator(), media_type="text/event-stream") diff --git a/api/qwen.py b/api/qwen.py new file mode 100644 index 0000000..c720ecf --- /dev/null +++ b/api/qwen.py @@ -0,0 +1,68 @@ +import logging +import uuid +import json + +from fastapi import APIRouter, HTTPException, status +from fastapi.responses import StreamingResponse + +from config import QWEN_API_KEY, QWEN_UPSTREAM_URL +from models import ChatRequest +from utils import convert_prompt_to_messages, standard_sse_generator + +router = APIRouter() +logger = logging.getLogger("proxy") + + +@router.post("/qwen/chat") +async def proxy_qwen_chat(request: ChatRequest): + + # 为每个请求生成一个唯一的 ID,方便在日志中追踪 + request_id = f"qwen-{uuid.uuid4()}" + + # 检查 API Key 是否已在 config.py 中配置 + if not QWEN_API_KEY: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Qwen API key is not configured.", + ) + + # 记录从客户端收到的原始请求体 + client_request_body = request.model_dump() + logger.debug( + f"[{request_id}] === Qwen New Request ===\n" + f"[{request_id}] Client Request Body:\n" + f"{json.dumps(client_request_body, indent=2, ensure_ascii=False)}" + ) + + # 将 prompt 格式转换为 API 所需的 messages 格式 + messages = convert_prompt_to_messages(request.prompt) + logger.debug( + f"[{request_id}] Agent Transformed Messages:\n" + f"{json.dumps(messages, indent=2, ensure_ascii=False)}" + ) + + # 构建发送到上游 API 的 payload + payload = { + "model": "qwen-plus", # 指定使用的模型 + "messages": messages, + "stream": True, + "temperature": request.temperature, + "top_p": request.top_p, + "max_tokens": request.max_tokens, + } + logger.debug( + f"[{request_id}] Upstream Payload:\n" + f"{json.dumps(payload, indent=2, ensure_ascii=False)}" + ) + + # 请求头,包含认证信息 + headers = {"Authorization": f"Bearer {QWEN_API_KEY}"} + + # 返回一个流式响应对象 + # standard_sse_generator 会处理与上游的通信和流式数据的转发 + return StreamingResponse( + standard_sse_generator( + QWEN_UPSTREAM_URL, payload, headers, "Qwen", request_id + ), + media_type="text/event-stream", # 指定 SSE 的 MIME 类型 + ) diff --git a/config.py b/config.py new file mode 100644 index 0000000..c97a9b5 --- /dev/null +++ b/config.py @@ -0,0 +1,20 @@ +# --- API Key Configuration --- +# GLM API Key +GLM_API_KEY = "xxxxxxx" +# Qwen API Key +QWEN_API_KEY = "xxxxxxx" + +# --- Upstream Service URLs --- +# MThreads Playground API +MTHREADS_UPSTREAM_URL = "https://qwen2dot5-14b-instruct-1m.playground.mthreads.com/call/generate_response" +# Qwen DashScope API +QWEN_UPSTREAM_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" +# GLM BigModel API +GLM_UPSTREAM_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions" + +# --- Default Listener Configuration --- +DEFAULT_LISTEN_HOST = "0.0.0.0" +DEFAULT_LISTEN_PORT = 3060 + +# --- Global Variables --- +DEBUG = False diff --git a/main.py b/main.py new file mode 100644 index 0000000..eb2a920 --- /dev/null +++ b/main.py @@ -0,0 +1,96 @@ +import argparse +import logging +import sys +import uuid + +import uvicorn +from fastapi import FastAPI + +import config +# 导入 API 路由模块 +from api import glm, mthreads, qwen +from config import DEFAULT_LISTEN_HOST, DEFAULT_LISTEN_PORT + + +def setup_logging(debug: bool): + """ + 配置日志系统，设置 proxy 专用日志记录器,避免与 uvicorn 等第三方库的日志冲突 + """ + # 将记录器设置为最低级别(DEBUG),以便将所有消息传递给其处理器 + app_logger = logging.getLogger("proxy") + app_logger.setLevel(logging.DEBUG) + app_logger.propagate = False # 防止日志消息传播到根记录器,避免重复记录 + + # 清除现有处理器,确保配置唯一性 + app_logger.handlers.clear() + + # 定义所有日志共享的格式 + formatter = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + ) + + # 在控制台显示 INFO 及以上级别的日志,保证控制台不会被 DEBUG 信息淹没 + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(logging.INFO) + console_handler.setFormatter(formatter) + app_logger.addHandler(console_handler) + + if debug: + # 在调试模式下,将 DEBUG 及以上级别的日志写入文件 + file_handler = logging.FileHandler("debug.log", mode="w", encoding="utf-8") + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(formatter) + app_logger.addHandler(file_handler) + + +# 命令行参数解析 +parser = argparse.ArgumentParser(description="LLM API Proxy Server") +parser.add_argument( + "-p", + "--port", + type=int, + default=DEFAULT_LISTEN_PORT, + help=f"Port to run the server on (default: {DEFAULT_LISTEN_PORT})", +) +parser.add_argument( + "-d", + "--debug", + action="store_true", + help="Enable debug mode, writing detailed logs to debug.log", +) +args = parser.parse_args() +listen_port = args.port +config.DEBUG = args.debug + +# 配置日志系统 +setup_logging(config.DEBUG) + +# 获取配置好的专用日志记录器 +logger = logging.getLogger("proxy") + +app = FastAPI( + title="LLM Chat API Proxy", + description="A proxy for LLM APIs, supporting GLM, Qwen, and MThreads Playground.", + version="1.0.2", +) + +# 包含来自各个 API 模块的路由 +app.include_router(mthreads.router, prefix="/api") +app.include_router(qwen.router, prefix="/api") +app.include_router(glm.router, prefix="/api") + + +if __name__ == "__main__": + logger.info( + f"Starting LLM API Proxy v1.0.2 on http://{DEFAULT_LISTEN_HOST}:{listen_port}" + ) + if config.DEBUG: + logger.info("Debug mode is enabled. Logging to debug.log") + + uvicorn.run( + "main:app", + host=DEFAULT_LISTEN_HOST, + port=listen_port, + reload=True, + reload_delay=1.0, + ) diff --git a/models.py b/models.py new file mode 100644 index 0000000..2d8b635 --- /dev/null +++ b/models.py @@ -0,0 +1,25 @@ +from typing import Any, Dict, List, Optional, Union + +from pydantic import BaseModel, Field + + +class ChatRequest(BaseModel): + + prompt: List[str] = Field( + ..., + min_length=1, + description="The user's prompt, where the first element is the system prompt.", + ) + + temperature: float = 0.7 + + top_p: float = 0.9 + + max_tokens: int = 2048 + + # 是否启用网络搜索功能,可选项 + # 可以是一个布尔值或一个包含详细配置的字典 + # 例如:{"enable": True, "max_results": 5} + web_search: Optional[Union[bool, Dict[str, Any]]] = Field( + None, description="Enable or configure web search capabilities." + ) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3f46dd9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastapi +uvicorn[standard] +httpx \ No newline at end of file diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..ca3c704 --- /dev/null +++ b/utils.py @@ -0,0 +1,252 @@ +import json +import logging +import re +from typing import Any, Dict, List + +import httpx + +from config import DEBUG + +logger = logging.getLogger("proxy") + + +class UpstreamAPIError(Exception): + """ + 自定义异常类,表示与上游 API 通信时发生的错误;继承自 Exception ,增加 status_code 和 details 属性,方便格式统一 + """ + def __init__(self, status_code: int, message: str, details: Any = None): + self.status_code = status_code + self.message = message + self.details = details + super().__init__(f"[{self.status_code}] {self.message}: {self.details}") + + +async def handle_upstream_response(response: httpx.Response): + """ + 辅助函数,检查上游 API 的 HTTP 响应;如果响应状态码表示失败(非 2xx),则会解析错误信息并抛出 UpstreamAPIError 异常 + """ + + if response.is_success: + return + + status_code = response.status_code + try: + # 尝试将响应解析为 JSON 格式 + details = response.json() + except json.JSONDecodeError: + # 解析失败,直接截取文本作为错误详情 + details = response.text[:200] + + # 预设常见 HTTP 错误状态码 + error_messages = { + 400: "Invalid request", + 401: "Authentication failed", + 403: "Forbidden", + 404: "Upstream resource not found", + 429: "Too many requests", + 500: "Upstream server internal error", + 502: "Upstream server gateway error", + 503: "Upstream service unavailable", + 504: "Upstream server gateway timeout", + } + # 不在预设则使用通用错误信息 + message = error_messages.get(status_code, "Unknown upstream API error") + + raise UpstreamAPIError(status_code=status_code, message=message, details=details) + + +async def generate_error_sse(error: UpstreamAPIError, request_id: str): + """ + 捕获到 UpstreamAPIError 时,生成符合 SSE 错误响应 + """ + error_content = { + "error": { + "message": f"Upstream API Error: {error.message}", + "code": error.status_code, + "details": error.details + } + } + # 使用 json.dumps 将错误详情字典转换为 JSON 字符串 + formatted_data = json.dumps(error_content, ensure_ascii=False) + logger.debug( + f"[{request_id}] Yielding Error to Client:\n" + f"event: error\n" + f"data: {formatted_data}\n\n" + ) + yield f"event: error\n" + yield f"data: {formatted_data}\n\n" + + +async def generate_network_error_sse(exc: httpx.RequestError, request_id: str): + """ + 捕获到 httpx 网络请求错误时,生成符合 SSE 错误响应 + """ + error_content = { + "error": { + "message": f"Network request error: {exc}", + "code": "network_error" + } + } + formatted_data = json.dumps(error_content, ensure_ascii=False) + logger.debug( + f"[{request_id}] Yielding Network Error to Client:\n" + f"event: error\n" + f"data: {formatted_data}\n\n" + ) + yield f"event: error\n" + yield f"data: {formatted_data}\n\n" + + +def convert_prompt_to_messages(prompt: List[str]) -> List[Dict[str, str]]: + """ + 将自定义的 prompt 列表格式转换为符合 OpenAI API 标准的 messages 格式 + - prompt 列表的第一个元素被视为 system 角色的内容 + - prompt 列表的其余所有元素被合并,并视为 user 角色的内容 + - 如果 prompt 只有一个元素,则直接视为 user 消息 + """ + if not prompt: + return [] + if len(prompt) == 1: + return [{"role": "user", "content": prompt[0]}] + + # 第一个元素作为 system message + system_message = {"role": "system", "content": prompt[0]} + # 后续所有元素合并成一个 user message + user_content = "\n".join(prompt[1:]) + user_message = {"role": "user", "content": user_content} + return [system_message, user_message] + +def sanitize_mthreads_response(raw_text: str) -> str: + """ + 清理 MThreads Qwen API 返回的无效内容 + - 拼接 <|FunctionCallBegin|>...<|FunctionCallEnd|> 中的内容 + - 移除各种形式的 <|...|> + - 移除末尾可能残留的特殊字符 `]>` + - 处理 Unicode 转义字符 + """ + final_text_parts = [] + + def _extract_from_call(match): + """re.sub 的回调函数,用于处理函数调用标记""" + call_content = match.group(1) + if not call_content or call_content == "[]": + return "" + try: + call_json = json.loads(call_content) + if call_json.get("name") == "send_message": + args_str = call_json.get("arguments", "{}") + args_json = json.loads(args_str) + message_text = args_json.get("text", "") + if message_text: + final_text_parts.append(message_text) + except (json.JSONDecodeError, TypeError): + # 解析失败，直接忽略这个函数调用块 + pass + return "" + + func_call_pattern = re.compile( + r"<\|FunctionCallBegin\|>(.*?)<\|FunctionCallEnd\|>", re.DOTALL + ) + + main_text = func_call_pattern.sub(_extract_from_call, raw_text) + + main_text = re.sub(r"<\|.*?\|>", "", main_text) + + main_text = main_text.strip().rstrip(']>') + + if main_text.strip(): + final_text_parts.insert(0, main_text.strip()) + + processed_text = "\n".join(final_text_parts).strip() + + # 处理双重转义字符;用 latin-1 编码后再用 unicode-escape 解码修复 + return processed_text.encode("latin-1", "backslashreplace").decode("unicode-escape") + + +async def standard_sse_generator( + upstream_url: str, + payload: dict, + headers: dict, + service_name: str, + request_id: str, +): + """ + 用于处理与 OpenAI-like API 流式通信的异步生成器;向上游服务发送请求,并以 Server-Sent Events (SSE) 的格式流式返回响应 + + 参数: + - upstream_url:上游 API 的地址 + - payload:发送给上游 API 的请求体 (JSON) + - headers:请求头，通常包含认证信息 + - service_name:服务名称,日志追踪 + - request_id:当前请求的唯一 ID,日志追踪 + """ + try: + # 使用 httpx.AsyncClient 以支持异步和连接池 + async with httpx.AsyncClient(timeout=httpx.Timeout(600.0)) as client: + # 使用 client.stream 发起流式请求 + async with client.stream( + "POST", upstream_url, json=payload, headers=headers + ) as response: + # 检查初始响应头,如果状态码非 2xx 则会抛出异常 + await handle_upstream_response(response) + + accumulated_content = [] + # 异步迭代来自服务器的每一行数据 + async for line in response.aiter_lines(): + + # SSE 事件通常以 "data:" 开头 + if not line.startswith("data:"): + continue + + # 提取 data: 后面的内容 + data_str = line.split(":", 1)[1].strip() + + # SSE 流以 [DONE] 标记结束 + if data_str == "[DONE]": + break + try: + data_json = json.loads(data_str) + # 遵循 OpenAI 的格式,从 choices[0].delta.content 中提取文本块 + delta = data_json.get("choices", [{}])[0].get("delta", {}) + content_chunk = delta.get("content") + if content_chunk: + accumulated_content.append(content_chunk) + except (json.JSONDecodeError, IndexError): + # 如果某一行不是有效的 JSON 或结构不符合预期,则忽略 + pass + + # 将所有接收到的文本块拼接成最终的完整文本 + final_text = "".join(accumulated_content) + if not final_text: + final_text = "[No content returned from model]" + logger.debug(f"[{request_id}] Final Combined Text: '{final_text}'") + + # 将最终文本包装成我们统一的 SSE 'complete' 事件格式 + formatted_data = json.dumps([final_text], ensure_ascii=False) + + chunk_for_client_event = f"event: complete\n" + chunk_for_client_data = f"data: {formatted_data}\n\n" + + logger.debug( + f"[{request_id}] Yielding to Client:\n{chunk_for_client_event}{chunk_for_client_data}" + ) + + # 产出最终的 SSE 事件 + yield chunk_for_client_event + yield chunk_for_client_data + + except httpx.RequestError as exc: + # 处理网络层面的错误 + async for item in generate_network_error_sse(exc, request_id): + yield item + except UpstreamAPIError as exc: + # 处理应用层面的上游 API 错误 + async for item in generate_error_sse(exc, request_id): + yield item + except Exception as exc: + # 处理其他所有未预料到的异常 + error = UpstreamAPIError( + status_code=500, message="An unknown error occurred in the proxy server", details=str(exc) + ) + async for item in generate_error_sse(error, request_id): + yield item