From 95db459eba16a4867b55d91b6ee64f12c538575c Mon Sep 17 00:00:00 2001 From: rsesot Date: Thu, 17 Jul 2025 18:27:28 +0800 Subject: [PATCH] =?UTF-8?q?ver=201.0.3=20=E4=B8=80=E7=82=B9=E5=B0=8F?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=8A=A0=E6=9B=B4=E6=96=B0=E4=B8=80=E7=82=B9?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/glm.py | 16 ++----- api/mthreads.py | 117 +++++++++++++++++++++++++++++++----------------- api/qwen.py | 6 +++ main.py | 4 +- 4 files changed, 89 insertions(+), 54 deletions(-) diff --git a/api/glm.py b/api/glm.py index 32dbfeb..1bed2e3 100644 --- a/api/glm.py +++ b/api/glm.py @@ -51,18 +51,10 @@ async def proxy_glm_chat(request: ChatRequest): 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] + # 如果客户端请求 web_search=True，则为 GLM 启用网页搜索工具 + if request.web_search is True: + payload["tools"] = [{"type": "web_search", "web_search": {"enable": True}}] + logger.info(f"[{request_id}] Enabled web search for GLM.") logger.debug( f"[{request_id}] Upstream Payload:\n" diff --git a/api/mthreads.py b/api/mthreads.py index ed1a6b8..e62125f 100644 --- a/api/mthreads.py +++ b/api/mthreads.py @@ -2,6 +2,7 @@ import json import logging import uuid import httpx +import asyncio from fastapi import APIRouter from fastapi.responses import StreamingResponse @@ -74,49 +75,85 @@ async def proxy_mthreads_chat(request: ChatRequest): async def stream_generator(): """ - 第一次请求获取 event_id 后，轮询流式数据 + 第一次请求获取 event_id 后,轮询流式数据;总共尝试3次,每次间隔15秒 """ - # --- 第二步: 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 + for attempt in range(1, 4): + logger.info(f"[{request_id}] Attempt {attempt}/3 to stream from MThreads.") + try: + async with httpx.AsyncClient(timeout=150.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: error"): + try: + data_line = await line_iterator.__anext__() + if 'data: "404: Session not found."' in data_line: + logger.warning( + f"[{request_id}] MThreads session not found on attempt {attempt}. " + f"{'Will retry.' if attempt < 3 else 'This was the last attempt.'}" + ) + break + except StopAsyncIteration: + + logger.warning(f"[{request_id}] Stream ended after 'event: error' on attempt {attempt}.") + break + + + 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}'") + + 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 + + return + + except httpx.RequestError as e: + logger.error(f"[{request_id}] Network error on attempt {attempt}: {e}") + if attempt == 3: + async for item in generate_network_error_sse(e, request_id): + yield item + return + except (UpstreamAPIError, StopAsyncIteration, json.JSONDecodeError) as e: + logger.warning(f"[{request_id}] Handled error on attempt {attempt}: {type(e).__name__} - {e}") + if isinstance(e, StopAsyncIteration): + logger.warning(f"[{request_id}] MThreads stream was interrupted on attempt {attempt}. Retrying...") + + if attempt == 3: + 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 unexpectedly on final attempt.") + else: + err = e + async for item in generate_error_sse(err, request_id): + yield item + return + + if attempt < 3: + logger.info(f"[{request_id}] Waiting 15 seconds before next attempt.") + await asyncio.sleep(15) + + logger.error(f"[{request_id}] Failed to get response from MThreads after 3 attempts. Returning 'No content' message.") + error_message = "No content returned from model" + error_payload = json.dumps({"error": {"message": error_message, "type": "proxy_error", "code": 504}}, ensure_ascii=False) + yield f"event: error\n" + yield f"data: {error_payload}\n\n" return StreamingResponse(stream_generator(), media_type="text/event-stream") diff --git a/api/qwen.py b/api/qwen.py index c720ecf..2f10a0e 100644 --- a/api/qwen.py +++ b/api/qwen.py @@ -50,6 +50,12 @@ async def proxy_qwen_chat(request: ChatRequest): "top_p": request.top_p, "max_tokens": request.max_tokens, } + + # 如果客户端请求 web_search=True，则为 Qwen 启用网页搜索 + if request.web_search is True: + payload["enable_search"] = True + logger.info(f"[{request_id}] Enabled web search for Qwen.") + logger.debug( f"[{request_id}] Upstream Payload:\n" f"{json.dumps(payload, indent=2, ensure_ascii=False)}" diff --git a/main.py b/main.py index eb2a920..2f2e925 100644 --- a/main.py +++ b/main.py @@ -71,7 +71,7 @@ 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", + version="1.0.3", ) # 包含来自各个 API 模块的路由 @@ -82,7 +82,7 @@ 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}" + f"Starting LLM API Proxy v1.0.3 on http://{DEFAULT_LISTEN_HOST}:{listen_port}" ) if config.DEBUG: logger.info("Debug mode is enabled. Logging to debug.log")