From cb94fb7bfe01768155e33e23114f8b90bcefdf27 Mon Sep 17 00:00:00 2001 From: zk Date: Tue, 7 Jul 2026 21:09:23 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0AI=E9=97=A8=E9=97=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.prod | 12 +++++-- app/services/ai_tool.py | 74 ++++++++++++++++++++++++++++++++++++----- 2 files changed, 75 insertions(+), 11 deletions(-) diff --git a/.env.prod b/.env.prod index b82ac67..a00ba3b 100644 --- a/.env.prod +++ b/.env.prod @@ -14,14 +14,20 @@ DB_USER=root DB_PASSWORD=^CgDatabase2020 DB_NAME=offerpie + # AI 供应商 VOLCENGINE_API_KEY=fd065993-bee2-4f31-8bf2-56d5d3012c02 VOLCENGINE_BASE_URL=https://ark.cn-beijing.volces.com/api/v3 +# Claude（Anthropic 风格） +ANTHROPIC_API_KEY=sk-43ccdb29caa7e9ebe0db8ac0958c63f6d3a2d62e59064d3d26d94332055a9bc9 +ANTHROPIC_BASE_URL=https://code.warpdevloper.cloud + # 岗位清洗参数 -CLEAN_BATCH_SIZE=100 -CLEAN_CONCURRENCY=80 -CLEAN_INTERVAL_SECONDS=200 +CLEAN_BATCH_SIZE=20 +CLEAN_CONCURRENCY=20 +CLEAN_INTERVAL_SECONDS=100 +CLEAN_TOTAL_LIMIT=0 # 公司补充参数 COMPANY_BATCH_SIZE=20 diff --git a/app/services/ai_tool.py b/app/services/ai_tool.py index 43fa852..658b4d0 100644 --- a/app/services/ai_tool.py +++ b/app/services/ai_tool.py @@ -1,5 +1,11 @@ -"""AI 调用工具封装""" +"""AI 调用工具封装 +核心机制：健康门闸（Health Gate） +- ainvoke 抛异常 → 标记故障，全部协程阻塞，单探针退避重试直到成功 +- ainvoke 正常返回 → 放行，后续逻辑不变 +""" + +import asyncio import re from typing import Any @@ -15,6 +21,11 @@ _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 代码块，容错处理""" @@ -29,13 +40,60 @@ def parse_llm_json(text: str) -> Any: async def ai_chat(llm: BaseChatModel, system_prompt: str, user_message: str) -> str: - """异步调用 LLM，返回原始文本""" - messages = [ - SystemMessage(content=system_prompt), - HumanMessage(content=user_message), - ] - response = await llm.ainvoke(messages) - return response.content + """异步调用 LLM，返回原始文本。接口异常时触发门闸阻塞重试直到成功。""" + global _gate_probing + + while True: + # 门闸检查：如果当前处于故障状态，阻塞等待直到探针恢复门闸 + await _gate_event.wait() + + try: + messages = [ + SystemMessage(content=system_prompt), + HumanMessage(content=user_message), + ] + response = await llm.ainvoke(messages) + return response.content + except Exception as e: + # 接口失败 → 关闭门闸 + if _gate_event.is_set(): + _gate_event.clear() + log.error("AI 接口异常，门闸关闭，全部协程阻塞等待恢复: {}", e) + + if not _gate_probing: + # 当前协程成为探针，负责退避重试直到成功 + _gate_probing = True + result = await _probe_until_recover(llm, system_prompt, user_message) + # 探针成功，直接返回结果（不用再调一次） + return result + # 已有探针在工作，回到 while 顶部 await _gate_event.wait() 等待唤醒 + + +async def _probe_until_recover(llm: BaseChatModel, system_prompt: str, user_message: str) -> str: + """探针：退避重试直到接口恢复，返回成功的响应内容，并打开门闸唤醒所有等待协程""" + global _gate_probing + + delay = 1.0 # 初始退避 1 秒 + max_delay = 30.0 # 最大退避 30 秒 + attempt = 0 + + while True: + await asyncio.sleep(delay) + attempt += 1 + try: + messages = [ + SystemMessage(content=system_prompt), + HumanMessage(content=user_message), + ] + response = await llm.ainvoke(messages) + # 成功 → 打开门闸，唤醒所有等待协程 + _gate_probing = False + _gate_event.set() + log.info("AI 接口恢复，门闸打开，第{}次探测成功", attempt) + return response.content + except Exception as e: + log.warning("AI 接口仍不可用，第{}次探测失败(退避{}s): {}", attempt, delay, e) + delay = min(delay * 2, max_delay) async def ai_chat_json(llm: BaseChatModel, system_prompt: str, user_message: str) -> Any: