diff --git a/app/services/ai_tool.py b/app/services/ai_tool.py index 1ce0c97..ab634e0 100644 --- a/app/services/ai_tool.py +++ b/app/services/ai_tool.py @@ -1,8 +1,7 @@ """AI 调用工具封装 -核心机制:健康门闸(Health Gate) -- ainvoke 抛异常 → 标记故障,全部协程阻塞,单探针退避重试直到成功 -- ainvoke 正常返回 → 放行,后续逻辑不变 +每个协程独立重试,失败后 sleep 200ms 再试,直到成功。 +不再有全局门闸,互不影响。 """ import asyncio @@ -26,11 +25,6 @@ _THINK_RE = re.compile(r".*?", re.DOTALL | re.IGNORECASE) # 匹配 ```json ... ``` 代码块,提取中间的 JSON 内容 _CODE_BLOCK_RE = re.compile(r"```(?:json\w*)?\s*\n?(.*?)\n?\s*```", re.DOTALL | re.IGNORECASE) -# ──────────── 健康门闸 ──────────── -_gate_event = asyncio.Event() # clear=故障阻塞中, set=健康放行 -_gate_event.set() # 初始状态:健康 -_gate_probing = False # 是否已有探针在重试 - def parse_llm_json(text: str) -> Any: """解析 AI 输出的 JSON,自动去除思考标签、markdown 代码块,容错处理""" @@ -45,13 +39,11 @@ def parse_llm_json(text: str) -> Any: async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str, scene: Optional[str] = None) -> str: - """异步调用 LLM,返回原始文本。接口异常时触发门闸阻塞重试直到成功。""" - global _gate_probing + """异步调用 LLM,返回原始文本。失败后独立重试直到成功。""" + attempt = 0 while True: - # 门闸检查:如果当前处于故障状态,阻塞等待直到探针恢复门闸 - await _gate_event.wait() - + attempt += 1 try: messages = [ SystemMessage(content=system_prompt), @@ -64,52 +56,8 @@ async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str, sce await _save_call_log(scene, system_prompt, user_message, content) return content except Exception as e: - # 接口失败 → 关闭门闸 - if _gate_event.is_set(): - _gate_event.clear() - log.error("AI 接口异常,门闸关闭,全部协程阻塞等待恢复: {}", e) - - if not _gate_probing: - # 当前协程成为探针,负责退避重试直到成功 - _gate_probing = True - await _probe_until_recover(llm, system_prompt, user_message) - # 探针恢复后,回到 while 循环顶部用业务数据重新调用 - continue - # 已有探针在工作,回到 while 顶部 await _gate_event.wait() 等待唤醒 - - -async def _probe_until_recover(llm: BaseChatModel, system_prompt: str, user_message: str) -> str: - """探针:用轻量 hello 消息探测接口是否恢复,成功后打开门闸,原协程重新走正常流程""" - global _gate_probing - - delay = 0.2 # 探测间隔 200ms - attempt = 0 - - try: - while True: - await asyncio.sleep(delay) - attempt += 1 - try: - log.info("探针第{}次探测开始...", attempt) - messages = [ - SystemMessage(content="You are a helpful assistant."), - HumanMessage(content="hello"), - ] - await llm.ainvoke(messages) - # 成功 → 打开门闸,唤醒所有等待协程 - _gate_probing = False - _gate_event.set() - log.info("探针第{}次探测成功,AI 接口恢复,门闸打开", attempt) - break - except Exception as e: - log.warning("探针第{}次探测失败: {}", attempt, e) - except (asyncio.CancelledError, Exception) as e: - # 探针意外死亡 → 必须恢复门闸状态,否则所有 worker 永远阻塞 - _gate_probing = False - _gate_event.set() - log.error("探针协程意外退出,强制打开门闸: {}", e) - - return "" + log.warning("AI 调用失败(第{}次), 200ms 后重试: {}", attempt, e) + await asyncio.sleep(0.2) async def ai_chat_json(llm: BaseChatModel, system_prompt: str, user_message: str, scene: Optional[str] = None) -> Any: