diff --git a/CLAUDE.md b/CLAUDE.md index 22435448..f474ac6c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -81,6 +81,7 @@ backend/ - 更新：`backend/app/api/llm_engines.py`（新增 `PUT /ai-steward/llm-engines/settings`，维护本地知识库问答开关） - 更新：`backend/app/api/ai_steward.py`（移除 `/ai-steward/command/dispatch` OpenClaw 指挥规划入口） - 更新：`backend/app/services/ai_steward.py`（零配置项目创建任一步 LLM 失败直接报错，不再回退默认草稿） +- 更新：`backend/app/services/signal_source_orchestration_service.py`（信号源 onboarding 规划与 worker 执行失败均 fail-fast，不再回退 heuristic/fallback 摘要） - 更新：`backend/app/services/project_knowledge_graph.py`（新增 `ai_model -> governance_policy -> interface` 治理图谱同步能力，增量任务新增 `model_guardrails` 阶段） - 移除：`backend/.env.example`（配置模板统一到根目录 `.env.example`） - 新增：`backend/app/services/project_command_utils.py`（项目创建标题/描述提取与归一化辅助） @@ -247,6 +248,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 所有 LLM 调用禁止模型级兜底：禁止高阶升级、Safety 引擎、应急直连、stub 假成功；模型异常、超时或结构化输出不合法时必须显式报错 - 本地 `ollama` 仅允许知识库问答链路使用，且必须通过 `llm_engine.settings.enable_local_kb_qa` 显式开启；未开启时运行期与连通测试都不可用 - AI Steward 零配置创建任一步 LLM 失败、超时或 JSON 非法时必须直接报错，禁止回退默认 analysis/signal/charter/tasks 草稿 +- Signal Source Onboarding 的 planner/worker 任一步失败必须显式报错或标记 failed，禁止 heuristic/fallback 兜底计划与“最小闭环”摘要 - AI Steward 不再暴露 OpenClaw 指挥规划入口，`/ai-steward/command/dispatch` 视为下线路径 ## 9. 新增代码的参考模式 @@ -308,5 +310,4 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - `operation_logs.created_at_beijing` 为应用层计算字段（非 DB 列），审计查询接口返回时统一补齐 - 历史脏信号源修复优先复用同项目已存在 `signal_id`，仅无法复用时才触发 `capture-feishu` 补录 - 兼容 SQL 查询入口：`ai_pmo_work_items` 为视图别名（底层映射 `work_items`），避免历史脚本直接查旧表名失败 -- `project.create` 在 AI 管家建项链路连续失败时，允许 TypedExecutor 启用本地兜底建项（最小项目 + 初始信号），并在返回 payload 标记 `fallback.enabled=true` - `signal_source.onboard` 已加入工作流门禁旁路：进入执行阶段后按闭环结果返回 `succeeded/pending_manual/failed_safe`，不再被前置仿真门禁直接拦截为 `waiting_manual` diff --git a/backend/app/services/ai_pmo_intent_routing.py b/backend/app/services/ai_pmo_intent_routing.py index 17536960..bc6fced9 100644 --- a/backend/app/services/ai_pmo_intent_routing.py +++ b/backend/app/services/ai_pmo_intent_routing.py @@ -666,11 +666,5 @@ class IntentResolver: @staticmethod def _build_signal_title(text: str, *, default: str) -> str: - cleaned = re.sub(r"\s+", " ", str(text or "")).strip(" ，,:：") - if not cleaned: - return default - for token in ("。", "；", ";", "!", "！", "?", "？"): - if token in cleaned: - cleaned = cleaned.split(token, 1)[0].strip() - break - return cleaned[:120] or default + raw = str(text or "").strip() + return raw or default diff --git a/backend/app/services/ai_pmo_kernel_mixins_closure_payload.py b/backend/app/services/ai_pmo_kernel_mixins_closure_payload.py index 796e6804..0bd926f5 100644 --- a/backend/app/services/ai_pmo_kernel_mixins_closure_payload.py +++ b/backend/app/services/ai_pmo_kernel_mixins_closure_payload.py @@ -12,7 +12,7 @@ for _name, _value in vars(_common).items(): class KernelClosurePayloadMixin: @staticmethod def _is_generic_signal_source_message(text: str) -> bool: - compact = re.sub(r"\s+", "", str(text or "")).strip().lower() + compact = "".join(str(text or "").split()).strip().lower() if not compact: return False if bool(GENERIC_SIGNAL_SOURCE_MESSAGE_RE.match(compact)): @@ -181,11 +181,6 @@ class KernelClosurePayloadMixin: intent.signal_summary = closure_summary elif not str(intent.signal_summary or "").strip(): intent.signal_summary = cls._normalize_dialogue_text(message_text, max_len=300) - if cls._is_generic_signal_source_message(str(intent.signal_title or "")): - if modules: - intent.signal_title = f"信号源接入 · {str(modules[0])[:60]}"[:120] - else: - intent.signal_title = "信号源接入请求" metadata = dict(intent.metadata or {}) metadata["closure_summary"] = closure_summary or None metadata["closure_dialogue"] = closure_dialogue or None diff --git a/backend/app/services/ai_pmo_typed_executor_mixins_discovery.py b/backend/app/services/ai_pmo_typed_executor_mixins_discovery.py index f28ea7bc..26ae352b 100644 --- a/backend/app/services/ai_pmo_typed_executor_mixins_discovery.py +++ b/backend/app/services/ai_pmo_typed_executor_mixins_discovery.py @@ -57,29 +57,19 @@ class TypedExecutorDiscoveryMixin: requested_count=requested_count, raw_preview=str(raw_text or "")[:260], ) - if strict_analysis_required: - error_message = "你要求先分析再入库，但当前信号分析结果不可解析，本次未自动入库。" - logger.warning( - "ai_pmo_signal_batch_analysis_strict_blocked", - tenant_id=self.tenant_id, - project_id=intent.project_id, - requested_count=requested_count, - message_preview=str(message_text or "")[:200], - ) - return { - "items": [], - "used_fallback": False, - "analysis_error": error_message, - "strict_analysis_required": True, - } + error_message = "你要求先分析再入库，但当前信号分析结果不可解析，本次未自动入库。" + logger.warning( + "ai_pmo_signal_batch_analysis_strict_blocked", + tenant_id=self.tenant_id, + project_id=intent.project_id, + requested_count=requested_count, + message_preview=str(message_text or "")[:200], + ) return { - "items": self._build_fallback_signal_batch_items( - message_text=message_text, - requested_count=requested_count, - ), - "used_fallback": True, - "analysis_error": "signal_batch_analysis_invalid_payload", - "strict_analysis_required": False, + "items": [], + "used_fallback": False, + "analysis_error": error_message, + "strict_analysis_required": strict_analysis_required, } return { "items": self._normalize_signal_batch_items( @@ -335,13 +325,34 @@ class TypedExecutorDiscoveryMixin: for index, row in enumerate(analyzed_signals[:requested_count]): if not isinstance(row, dict): continue + try: + refined_title = await self._refine_signal_title_with_ai( + default_title=f"候选信号 {index + 1}", + title_seed=str(row.get("title") or intent.signal_title or ""), + message_text=message_text, + summary=str(row.get("summary") or ""), + content=str(row.get("content") or ""), + intent_name=ROUTE_PROJECT_SIGNAL_DISCOVER, + ) + except Exception as exc: + return { + "status": "failed_safe", + "workflow_action": ROUTE_ACTION_DISCOVER, + "workflow_route": ROUTE_PROJECT_SIGNAL_DISCOVER, + "project_id": int(project_id), + "requested_count": requested_count, + "error_message": self._build_signal_title_refine_error_message(exc), + "candidate_signals": [], + } evidence_excerpt = "" if index < len(retrieval_items) and isinstance(retrieval_items[index], dict): evidence_excerpt = str(retrieval_items[index].get("content") or "")[:240] - dedupe_key = self._build_candidate_dedupe_key(project_id=int(project_id), candidate=row) + candidate_for_dedupe = dict(row) + candidate_for_dedupe["title"] = refined_title + dedupe_key = self._build_candidate_dedupe_key(project_id=int(project_id), candidate=candidate_for_dedupe) candidate_signals.append( { - "title": str(row.get("title") or "")[:120], + "title": refined_title, "summary": str(row.get("summary") or "")[:300], "content": str(row.get("content") or "")[:1800], "signal_type": str(row.get("signal_type") or "新需求"), @@ -389,4 +400,3 @@ class TypedExecutorDiscoveryMixin: "请回复“确认”提交，回复“换一批”刷新，或回复“取消”结束本次候选。" ), } - diff --git a/backend/app/services/ai_pmo_typed_executor_mixins_execute.py b/backend/app/services/ai_pmo_typed_executor_mixins_execute.py index 5fcc48b2..3532ce84 100644 --- a/backend/app/services/ai_pmo_typed_executor_mixins_execute.py +++ b/backend/app/services/ai_pmo_typed_executor_mixins_execute.py @@ -54,23 +54,11 @@ class TypedExecutorExecuteMixin: "error_message": "未找到可用项目创建人，当前无法自动创建项目。", } steward = AIStewardService(self.db) - try: - payload = await steward.create_project_from_natural_language( - input_text=message_text, - creator_id=int(operator_user_id), - tenant_id=self.tenant_id, - ) - except Exception as exc: - logger.warning( - "ai_pmo_project_create_local_fallback", - tenant_id=self.tenant_id, - reason=str(exc), - ) - payload = await self._create_project_with_local_fallback( - message_text=message_text, - operator_user_id=int(operator_user_id), - fallback_reason=str(exc), - ) + payload = await steward.create_project_from_natural_language( + input_text=message_text, + creator_id=int(operator_user_id), + tenant_id=self.tenant_id, + ) project = payload.get("project") if isinstance(payload.get("project"), dict) else {} signal = payload.get("signal") if isinstance(payload.get("signal"), dict) else {} return { @@ -88,6 +76,12 @@ class TypedExecutorExecuteMixin: intent=intent, message_text=message_text, ) + signal_summary = str(semantic_payload.get("summary") or intent.signal_summary or message_text) + signal_content = str(semantic_payload.get("content") or message_text) + initial_title_seed = self._normalize_title_candidate( + str(semantic_payload.get("title") or intent.signal_title or "") + ) + initial_signal_title = initial_title_seed if self._is_acceptable_title(initial_title_seed) else "信号源接入请求" strategic_payload = ( semantic_payload.get("strategic_analysis") if isinstance(semantic_payload.get("strategic_analysis"), dict) @@ -102,9 +96,9 @@ class TypedExecutorExecuteMixin: source_context["channel_ref"] = channel_ref strategic_payload["source_context"] = source_context signal = await self._create_signal( - title=str(semantic_payload.get("title") or intent.signal_title or "新增信号源接入请求"), - summary=str(semantic_payload.get("summary") or intent.signal_summary or message_text), - content=str(semantic_payload.get("content") or message_text), + title=initial_signal_title, + summary=signal_summary, + content=signal_content, project_id=intent.project_id, source="PMFlow", signal_type="新需求", @@ -129,6 +123,11 @@ class TypedExecutorExecuteMixin: operator_user_id=operator_user_id, ) orchestration = domain_workflow.get("raw_result") if isinstance(domain_workflow.get("raw_result"), dict) else {} + signal_source_payload = orchestration.get("signal_source") if isinstance(orchestration.get("signal_source"), dict) else {} + ai_source_title = self._normalize_title_candidate(signal_source_payload.get("source_name")) + if self._is_acceptable_title(ai_source_title): + signal.title = ai_source_title + await self.db.flush() return { "status": "succeeded" if str(domain_workflow.get("status") or "").lower() == "completed" else ( "pending_manual" if str(domain_workflow.get("status") or "").lower() == "pending_manual" else "failed_safe" @@ -172,6 +171,26 @@ class TypedExecutorExecuteMixin: signal_summaries: list[str] = [] for signal_row in analyzed_signals: summary_text = str(signal_row.get("summary") or "")[:300] + signal_content = str(signal_row.get("content") or message_text) + try: + refined_signal_title = await self._refine_signal_title_with_ai( + default_title="新信号", + title_seed=str(signal_row.get("title") or intent.signal_title or ""), + message_text=message_text, + summary=summary_text or str(intent.signal_summary or message_text), + content=signal_content, + intent_name=intent.name, + ) + except Exception as exc: + return { + "status": "failed_safe", + "workflow_action": ROUTE_ACTION_CREATE, + "workflow_route": ROUTE_PROJECT_SIGNAL_CREATE, + "error_message": self._build_signal_title_refine_error_message(exc), + "project_id": intent.project_id, + "requested_signal_count": requested_count, + "analyzed_signal_count": len(signal_ids), + } strategic_analysis = { "source_context": { "source_message_id": source_message_id, @@ -191,9 +210,9 @@ class TypedExecutorExecuteMixin: }, } signal = await self._create_signal( - title=str(signal_row.get("title") or intent.signal_title or "新信号"), + title=refined_signal_title, summary=summary_text or str(intent.signal_summary or message_text), - content=str(signal_row.get("content") or message_text), + content=signal_content, project_id=intent.project_id, source="PMFlow", signal_type=str(signal_row.get("signal_type") or "新需求"), @@ -263,77 +282,114 @@ class TypedExecutorExecuteMixin: message = f"{marker}{reason}" return message if "请稍后重试" in message else f"{message}。请稍后重试。" - async def _create_project_with_local_fallback( + @staticmethod + def _build_signal_title_refine_error_message(error: Exception) -> str: + reason = " ".join(str(error or "").split()).strip() + if not reason: + reason = "unknown_error" + return f"信号标题 AI 提炼失败：{reason}。" + + @staticmethod + def _normalize_title_candidate(value: Any) -> str: + text = str(value or "").strip() + if not text: + return "" + blocked_chars = {"\n", "\r", "\t", "{", "}", "[", "]"} + if any(ch in text for ch in blocked_chars): + return "" + return text + + @staticmethod + def _is_acceptable_title(text: str) -> bool: + if not text: + return False + if len(text) < 2: + return False + if len(text) > 200: + return False + return True + + @staticmethod + def _extract_title_from_ai_response(raw_text: str) -> str: + raw = str(raw_text or "").strip() + if not raw: + return "" + if raw.startswith("```"): + chunks = [chunk.strip() for chunk in raw.split("```") if chunk.strip()] + if chunks: + first = chunks[0] + if first.lower().startswith("json"): + first = first[4:].strip() + raw = first.strip() + if raw.startswith("{") and raw.endswith("}"): + try: + payload = json.loads(raw) + except Exception: + payload = {} + if isinstance(payload, dict): + title = payload.get("title") + if isinstance(title, str): + return title.strip() + if "{" in raw and "}" in raw: + start = raw.find("{") + end = raw.rfind("}") + if start >= 0 and end > start: + try: + payload = json.loads(raw[start : end + 1]) + except Exception: + payload = {} + if isinstance(payload, dict): + title = payload.get("title") + if isinstance(title, str): + return title.strip() + first_line = raw.splitlines()[0].strip() + return first_line.strip("'\" ") + + async def _refine_signal_title_with_ai( self, *, + default_title: str, + title_seed: str, message_text: str, - operator_user_id: int, - fallback_reason: str, - ) -> dict[str, Any]: - cleaned = re.sub(r"]*>", " ", str(message_text or ""), flags=re.IGNORECASE) - cleaned = re.sub(r"@_user_\\d+", " ", cleaned) - cleaned = re.sub(r"\\s+", " ", cleaned).strip() - title = "" - patterns = ( - r"(?:请|帮我)?(?:新建|新增|创建|建立)(?:一个|个)?项目[：:，,\\s《\"]*([^》\"，。,；;]{2,80})", - r"(?:项目)[：:，,\\s《\"]*([^》\"，。,；;]{2,80})", - ) - for pattern in patterns: - match = re.search(pattern, cleaned, re.IGNORECASE) - if not match: - continue - candidate = re.sub(r"\\s+", " ", str(match.group(1) or "")).strip(" 《》\"'，,。:：;；") - if candidate: - title = candidate - break - if not title: - title = f"AI-PMO项目-{_utc_now_naive().strftime('%m%d-%H%M%S')}" - - project = Project( - tenant_id=self.tenant_id, - name=title[:255], - title=title[:255], - description=f"本地兜底创建：{cleaned[:500]}", - owner_id=int(operator_user_id), - status="active", - domain_id=None, - assigned_role=None, - signal_count=0, - ) - self.db.add(project) - await self.db.flush() - - signal = await self._create_signal( - title=f"{title} 初始信号", - summary=f"由 AI PMO 本地兜底创建，原始指令：{cleaned[:120]}", - content=f"project_create_fallback_reason={fallback_reason[:240]}", - project_id=int(project.id), - source="PMFlow", - signal_type="新需求", - recommended_action="enter_only", - ) - project.signal_count = max(int(project.signal_count or 0), 1) - await self.db.flush() - - return { - "project": { - "id": str(project.id), - "name": project.name, - "title": project.title, - "status": project.status, - }, - "signal": { - "id": str(signal.id), - "content": signal.content, - "impact_score": signal.impact_score, - "priority": signal.priority, - }, - "tasks": [], - "fallback": { - "enabled": True, - "reason": fallback_reason[:240], - }, + summary: str, + content: str, + intent_name: str, + ) -> str: + payload = { + "intent": str(intent_name or ""), + "default_title": str(default_title or ""), + "title_seed": str(title_seed or ""), + "message_text": str(message_text or ""), + "summary": str(summary or ""), + "content": str(content or ""), } + prompt = ( + "你是 PMFlow 的信号标题提炼器。请根据输入语义提炼一个可入库标题。\n" + "要求：\n" + "1) 必须是语义主题短语，不要命令语气，不要'请/帮我/新增/增加'等动作词。\n" + "2) 不要项目ID、@提及、括号说明。\n" + "3) 使用中文，8-24字。\n" + "4) 仅输出标题文本，不要前后缀说明。\n\n" + f"输入数据：{json.dumps(payload, ensure_ascii=False)}" + ) + try: + raw_response = await llm_service.generate( + prompt, + db=self.db, + tenant_id=self.tenant_id, + function_name="project_draft_analysis_glm47_fallback", + use_prompt_profile=False, + temperature=0.0, + max_tokens=240, + response_mode="complete", + ) + ai_title = self._normalize_title_candidate(self._extract_title_from_ai_response(raw_response)) + if self._is_acceptable_title(ai_title): + return ai_title + raise ValueError("ai_pmo_signal_title_refine_invalid_payload") + except Exception as exc: + logger.warning("ai_pmo_signal_title_refine_failed error=%s", exc) + raise RuntimeError(f"ai_pmo_signal_title_refine_failed:{type(exc).__name__}") from exc async def _create_signal( self, @@ -352,10 +408,13 @@ class TypedExecutorExecuteMixin: normalized_priority = str(priority or "medium").strip().lower() if normalized_priority not in {"high", "medium", "low"}: normalized_priority = "medium" + normalized_title = self._normalize_title_candidate(title) + if not self._is_acceptable_title(normalized_title): + raise ValueError("signal_title_invalid") signal = Signal( tenant_id=self.tenant_id, source=source, - title=(title or "新信号")[:120], + title=normalized_title, summary=(summary or content or "")[:300], content=(content or summary or "")[:2000], signal_type=signal_type, diff --git a/backend/app/services/ai_pmo_typed_executor_mixins_parse.py b/backend/app/services/ai_pmo_typed_executor_mixins_parse.py index 05052b98..12fb3258 100644 --- a/backend/app/services/ai_pmo_typed_executor_mixins_parse.py +++ b/backend/app/services/ai_pmo_typed_executor_mixins_parse.py @@ -12,20 +12,13 @@ for _name, _value in vars(_common).items(): class TypedExecutorParseMixin: @staticmethod def _normalize_short_text(value: Any, *, max_len: int) -> str: - text = re.sub(r"\s+", " ", str(value or "")).strip() + text = " ".join(str(value or "").split()).strip() if not text: return "" + if max_len <= 0: + return text return text[:max_len] - @staticmethod - def _is_generic_signal_source_title(value: str) -> bool: - compact = re.sub(r"\s+", "", str(value or "")).strip().lower() - if not compact: - return True - if bool(GENERIC_SIGNAL_SOURCE_MESSAGE_RE.match(compact)): - return True - return compact in {"信号源接入请求", "新增信号源接入请求", "新增信号源"} - @classmethod def _build_signal_source_semantic_payload( cls, @@ -44,12 +37,7 @@ class TypedExecutorParseMixin: metadata.get("closure_dialogue") or message_text, max_len=1800, ) - title_seed = cls._normalize_short_text(intent.signal_title or "信号源接入请求", max_len=120) - if cls._is_generic_signal_source_title(title_seed): - if normalized_modules: - title_seed = f"信号源接入 · {normalized_modules[0]}"[:120] - else: - title_seed = "信号源接入请求" + title_seed = str(intent.signal_title or "").strip() or str(message_text or "").strip() or "信号源接入请求" reasoning_bullets = ["由 AI PMO kernel 统一入库。"] if normalized_modules: reasoning_bullets.append(f"闭环提炼模块：{'、'.join(normalized_modules[:4])}") @@ -77,7 +65,7 @@ class TypedExecutorParseMixin: @staticmethod def _compact_text(value: Any) -> str: - return re.sub(r"\s+", "", str(value or "")).strip().lower() + return "".join(str(value or "").split()).strip().lower() @classmethod def _looks_like_raw_instruction(cls, *, candidate: str, source: str) -> bool: @@ -233,7 +221,7 @@ class TypedExecutorParseMixin: for index in range(max(requested_count, 1)): row = raw_items[index] if index < len(raw_items) and isinstance(raw_items[index], dict) else {} - title = cls._normalize_short_text(row.get("title"), max_len=120) + title = cls._normalize_short_text(row.get("title"), max_len=0) summary = cls._normalize_short_text(row.get("summary"), max_len=300) content = cls._normalize_short_text(row.get("content"), max_len=1800) @@ -247,7 +235,7 @@ class TypedExecutorParseMixin: summary = cls._normalize_short_text(summary_seed, max_len=300) if not title: title_seed = summary or content or f"信号拆分序号 {index + 1}" - title = IntentResolver._build_signal_title(title_seed, default=f"信号拆分 {index + 1}") + title = cls._normalize_short_text(title_seed, max_len=0) or f"信号拆分 {index + 1}" if not content: content = f"{summary} 建议继续补充验收标准、里程碑与负责人。" @@ -282,48 +270,3 @@ class TypedExecutorParseMixin: } ) return normalized_items - - @classmethod - def _build_fallback_signal_batch_items( - cls, - *, - message_text: str, - requested_count: int, - ) -> list[dict[str, Any]]: - cleaned = cls._normalize_short_text(message_text, max_len=1200) - segments = [ - cls._normalize_short_text(chunk, max_len=220) - for chunk in re.split(r"[，,。；;\n]+", cleaned) - if cls._normalize_short_text(chunk, max_len=220) - ] - fallback_rows: list[dict[str, Any]] = [] - for index in range(max(requested_count, 1)): - seed = segments[index] if index < len(segments) else cleaned - if index == 0: - suffix = "目标与范围" - elif index == 1: - suffix = "执行与风险" - else: - suffix = f"落地步骤{index + 1}" - base_title = IntentResolver._build_signal_title(seed or f"信号拆分 {index + 1}", default=f"信号拆分 {index + 1}") - fallback_rows.append( - { - "title": cls._normalize_short_text(f"{base_title} · {suffix}", max_len=120), - "summary": cls._normalize_short_text( - f"围绕用户目标拆分出的第{index + 1}条执行信号，需继续补充可交付结果与验收标准。", - max_len=300, - ), - "content": cls._normalize_short_text( - f"原始指令要点：{seed or cleaned}。建议按负责人、时间窗口、依赖项进一步细化后推进执行。", - max_len=1800, - ), - "signal_type": "新需求", - "priority": "medium", - "reasoning_bullets": [ - "高级模型输出不可解析，已按规则拆分。", - f"信号拆分序号：{index + 1}。", - ], - } - ) - return fallback_rows - diff --git a/backend/app/services/ai_steward.py b/backend/app/services/ai_steward.py index ecbddb1d..24f59fba 100644 --- a/backend/app/services/ai_steward.py +++ b/backend/app/services/ai_steward.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import ast import json import os import re @@ -51,17 +52,32 @@ def _env_int(name: str, default: int) -> int: PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS = max( 8.0, - _env_float("AI_STEWARD_PROJECT_CREATION_STEP_TIMEOUT_SECONDS", 18.0), + _env_float("AI_STEWARD_PROJECT_CREATION_STEP_TIMEOUT_SECONDS", 30.0), ) PROJECT_CREATION_STEP_RETRY_COUNT = max( 0, _env_int("AI_STEWARD_PROJECT_CREATION_STEP_RETRY_COUNT", 1), ) +PROJECT_CREATION_JSON_PARSE_RETRY_COUNT = max( + 0, + _env_int("AI_STEWARD_PROJECT_CREATION_JSON_PARSE_RETRY_COUNT", 1), +) +PROJECT_CREATION_TASK_CHAIN_TIMEOUT_SECONDS = max( + PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS, + _env_float("AI_STEWARD_PROJECT_CREATION_TASK_CHAIN_TIMEOUT_SECONDS", 60.0), +) PROJECT_CREATION_STUB_MARKERS = ( "我暂时无法生成可靠回答", "模型服务异常", "请稍后重试", ) +PROJECT_CREATION_LOW_TO_HIGH_FUNCTION_MAP = { + "project_draft_analysis_glm47_fallback": "project_from_text_analysis", + "project_draft_signal_glm47_fallback": "project_from_text_signal", + "project_draft_charter_glm47_fallback": "project_from_text_charter", + "project_draft_task_chain_glm47_fallback": "project_from_text_task_chain", +} +PROJECT_CREATION_SERIAL_DRAFT_STEPS = _env_int("AI_STEWARD_PROJECT_CREATION_SERIAL_DRAFT_STEPS", 1) == 1 class AIStewardService: @@ -217,11 +233,16 @@ class AIStewardService: analysis: Dict[str, Any], tenant_id: int, ) -> dict[str, Any]: - signal_draft, charter_draft, task_drafts = await asyncio.gather( - self._generate_competitive_signal_draft(analysis=analysis, tenant_id=tenant_id), - self._generate_charter_draft(analysis=analysis, tenant_id=tenant_id), - self._generate_task_chain_draft(analysis=analysis, tenant_id=tenant_id), - ) + if PROJECT_CREATION_SERIAL_DRAFT_STEPS: + signal_draft = await self._generate_competitive_signal_draft(analysis=analysis, tenant_id=tenant_id) + charter_draft = await self._generate_charter_draft(analysis=analysis, tenant_id=tenant_id) + task_drafts = await self._generate_task_chain_draft(analysis=analysis, tenant_id=tenant_id) + else: + signal_draft, charter_draft, task_drafts = await asyncio.gather( + self._generate_competitive_signal_draft(analysis=analysis, tenant_id=tenant_id), + self._generate_charter_draft(analysis=analysis, tenant_id=tenant_id), + self._generate_task_chain_draft(analysis=analysis, tenant_id=tenant_id), + ) return { "analysis": analysis, "project": { @@ -244,6 +265,147 @@ class AIStewardService: return True return any(marker in normalized for marker in PROJECT_CREATION_STUB_MARKERS) + @staticmethod + def _project_creation_high_tier_function(function_name: str) -> str: + return str(PROJECT_CREATION_LOW_TO_HIGH_FUNCTION_MAP.get(str(function_name or "").strip(), "") or "").strip() + + @staticmethod + def _strip_code_fence(text: str) -> str: + raw = str(text or "").strip() + if not raw.startswith("```"): + return raw + chunks = raw.split("```") + if len(chunks) < 2: + return raw + candidate = chunks[1].strip() + if candidate.lower().startswith("json"): + candidate = candidate[4:].strip() + return candidate + + @staticmethod + def _extract_balanced_json_segment(text: str, open_char: str, close_char: str) -> str: + raw = str(text or "") + start = raw.find(open_char) + if start < 0: + return "" + depth = 0 + in_str = False + escaped = False + for idx in range(start, len(raw)): + ch = raw[idx] + if in_str: + if escaped: + escaped = False + elif ch == "\\": + escaped = True + elif ch == '"': + in_str = False + continue + if ch == '"': + in_str = True + continue + if ch == open_char: + depth += 1 + elif ch == close_char: + depth -= 1 + if depth == 0: + return raw[start : idx + 1].strip() + return "" + + def _extract_json_candidate(self, text: str) -> str: + raw = self._strip_code_fence(text) + if raw.startswith("{") or raw.startswith("["): + return raw + object_candidate = self._extract_balanced_json_segment(raw, "{", "}") + array_candidate = self._extract_balanced_json_segment(raw, "[", "]") + if object_candidate and array_candidate: + return object_candidate if raw.find("{") <= raw.find("[") else array_candidate + return object_candidate or array_candidate or raw + + @staticmethod + def _repair_json_text(raw: str) -> str: + text = str(raw or "") + if not text: + return text + buf: list[str] = [] + in_str = False + escaped = False + for ch in text: + if in_str: + if escaped: + escaped = False + elif ch == "\\": + escaped = True + elif ch == '"': + in_str = False + elif ch == "\n": + buf.append("\\n") + continue + elif ch == "\r": + buf.append("\\r") + continue + elif ch == "\t": + buf.append("\\t") + continue + else: + if ch == '"': + in_str = True + buf.append(ch) + repaired = "".join(buf) + repaired = re.sub(r",\s*([}\]])", r"\1", repaired) + return repaired + + def _parse_json_object_response(self, response_text: str, *, stage_label: str) -> Dict[str, Any]: + normalized = self._extract_json_candidate(response_text) + try: + payload = json.loads(normalized) + except Exception as exc: + repaired = self._repair_json_text(normalized) + try: + payload = json.loads(repaired) + except Exception: + try: + literal_payload = ast.literal_eval(normalized) + except Exception: + raise RuntimeError(f"{stage_label}：{exc}") from exc + payload = literal_payload + if not isinstance(payload, dict): + raise RuntimeError(f"{stage_label}：返回内容不是 JSON 对象") + return payload + + async def _generate_project_creation_object( + self, + *, + prompt: str, + tenant_id: int, + function_name: str, + stage_label: str, + timeout_seconds: float | None = None, + ) -> Dict[str, Any]: + max_rounds = PROJECT_CREATION_JSON_PARSE_RETRY_COUNT + 1 + last_error: Exception | None = None + for round_index in range(max_rounds): + response = await self._generate_project_creation_text( + prompt=prompt, + tenant_id=tenant_id, + function_name=function_name, + timeout_seconds=float(timeout_seconds) if timeout_seconds is not None else PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS, + ) + try: + return self._parse_json_object_response(response, stage_label=stage_label) + except Exception as exc: + last_error = exc if isinstance(exc, Exception) else RuntimeError(str(exc)) + logger.warning( + "ai_steward_project_creation_json_retry", + tenant_id=int(tenant_id), + function_name=function_name, + stage_label=stage_label, + round=round_index + 1, + error_type=type(last_error).__name__, + error=str(last_error), + ) + raise RuntimeError(str(last_error or f"{stage_label}：unknown_error")) from last_error + async def _generate_project_creation_text( self, *, @@ -260,10 +422,12 @@ class AIStewardService: response = await asyncio.wait_for( self.llm.generate( prompt, - db=self.db, tenant_id=tenant_id, function_name=function_name, response_mode="fast", + use_prompt_profile=False, + temperature=0.0, + max_tokens=2200, ), timeout=timeout_seconds, ) @@ -281,12 +445,54 @@ class AIStewardService: return cleaned_response except Exception as exc: last_error = exc if isinstance(exc, Exception) else RuntimeError(str(exc)) + high_tier_function = self._project_creation_high_tier_function(function_name) + if high_tier_function: + high_started_at = time.perf_counter() + try: + fallback_response = await asyncio.wait_for( + self.llm.generate( + prompt, + tenant_id=tenant_id, + function_name=high_tier_function, + response_mode="fast", + use_prompt_profile=False, + temperature=0.0, + max_tokens=2200, + ), + timeout=timeout_seconds, + ) + cleaned_fallback = str(fallback_response or "").strip() + high_elapsed_seconds = round(time.perf_counter() - high_started_at, 2) + if self._looks_like_project_creation_stub_response(cleaned_fallback): + raise RuntimeError("project_creation_stub_response") + logger.info( + "ai_steward_project_creation_step_escalated_ok", + tenant_id=int(tenant_id), + function_name=function_name, + high_tier_function=high_tier_function, + attempt=attempt_index + 1, + elapsed_seconds=high_elapsed_seconds, + ) + return cleaned_fallback + except Exception as high_exc: + last_error = high_exc if isinstance(high_exc, Exception) else RuntimeError(str(high_exc)) + logger.warning( + "ai_steward_project_creation_step_escalated_retry", + tenant_id=int(tenant_id), + function_name=function_name, + high_tier_function=high_tier_function, + attempt=attempt_index + 1, + elapsed_seconds=round(time.perf_counter() - high_started_at, 2), + error_type=type(high_exc).__name__, + error=str(high_exc), + ) logger.warning( "ai_steward_project_creation_step_retry", tenant_id=int(tenant_id), function_name=function_name, attempt=attempt_index + 1, elapsed_seconds=round(time.perf_counter() - started_at, 2), + error_type=type(exc).__name__, error=str(exc), ) @@ -582,24 +788,14 @@ class AIStewardService: "success_metrics": ["成功指标"] }} -只返回 JSON，不要其他内容。""" +只返回 JSON，不要其他内容。必须返回可被 json.loads 直接解析的单个对象，不要 markdown。""" - response = await self._generate_project_creation_text( + analysis = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, - function_name="project_from_text_analysis", + function_name="project_draft_analysis_glm47_fallback", + stage_label="AI 管家项目需求分析失败", ) - try: - response = response.strip() - if response.startswith("```"): - response = response.split("```")[1] - if response.startswith("json"): - response = response[4:] - analysis = json.loads(response.strip()) - except Exception as exc: - raise RuntimeError(f"AI 管家项目需求分析失败：{exc}") from exc - if not isinstance(analysis, dict): - raise RuntimeError("AI 管家项目需求分析失败：返回内容不是 JSON 对象") return analysis async def _generate_competitive_signal_draft( @@ -629,24 +825,16 @@ class AIStewardService: "impact_score": 0.9, "priority": "high", "strategic_analysis": {{"is_strategic": true, "threat_level": "MEDIUM"}} -}}""" +}} - response = await self._generate_project_creation_text( +只返回 JSON，不要 markdown，不要额外解释。""" + + payload = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, - function_name="project_from_text_signal", + function_name="project_draft_signal_glm47_fallback", + stage_label="AI 管家竞品信号生成失败", ) - try: - response = response.strip() - if response.startswith("```"): - response = response.split("```")[1] - if response.startswith("json"): - response = response[4:] - payload = json.loads(response.strip()) - except Exception as exc: - raise RuntimeError(f"AI 管家竞品信号生成失败：{exc}") from exc - if not isinstance(payload, dict): - raise RuntimeError("AI 管家竞品信号生成失败：返回内容不是 JSON 对象") return payload async def _generate_charter_draft( @@ -672,27 +860,24 @@ class AIStewardService: 以 JSON 格式返回： {{ "title": "宪章标题", - "content": "宪章完整内容（包含上述4个方面）", + "content": {{ + "objectives": ["目标1", "目标2"], + "success_criteria": ["标准1", "标准2"], + "constraints": ["约束1", "约束2"], + "milestones": ["里程碑1", "里程碑2"] + }}, "success_criteria": ["标准1", "标准2"], "constraints": ["约束1", "约束2"] -}}""" +}} - response = await self._generate_project_creation_text( +只返回 JSON，不要 markdown，不要额外解释。""" + + charter_data = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, - function_name="project_from_text_charter", + function_name="project_draft_charter_glm47_fallback", + stage_label="AI 管家项目宪章生成失败", ) - try: - response = response.strip() - if response.startswith("```"): - response = response.split("```")[1] - if response.startswith("json"): - response = response[4:] - charter_data = json.loads(response.strip()) - except Exception as exc: - raise RuntimeError(f"AI 管家项目宪章生成失败：{exc}") from exc - if not isinstance(charter_data, dict): - raise RuntimeError("AI 管家项目宪章生成失败：返回内容不是 JSON 对象") raw_content = charter_data.get("content", "") if isinstance(raw_content, dict): @@ -722,8 +907,9 @@ class AIStewardService: 时间周期：{analysis.get('timeline', '3周')} 请生成： -1. 任务列表（按优先级排序） +1. 任务列表（按优先级排序，最多 4 条） 2. 每个任务包含：名称、类型、预估工时、依赖关系、技能要求 +3. 每个 description 必须控制在 60 字以内，不要换行 以 JSON 格式返回： {{ @@ -737,25 +923,20 @@ class AIStewardService: "required_skills": ["技能1", "技能2"] }} ] -}}""" +}} - response = await self._generate_project_creation_text( +只返回 JSON，不要 markdown，不要额外解释。""" + + tasks_data = await self._generate_project_creation_object( prompt=prompt, tenant_id=tenant_id, - function_name="project_from_text_task_chain", + function_name="project_draft_task_chain_glm47_fallback", + stage_label="AI 管家任务链生成失败", + timeout_seconds=PROJECT_CREATION_TASK_CHAIN_TIMEOUT_SECONDS, ) - try: - response = response.strip() - if response.startswith("```"): - response = response.split("```")[1] - if response.startswith("json"): - response = response[4:] - tasks_data = json.loads(response.strip()) - tasks_list = tasks_data.get("tasks", []) - if isinstance(tasks_list, list) and tasks_list: - return tasks_list - except Exception as exc: - raise RuntimeError(f"AI 管家任务链生成失败：{exc}") from exc + tasks_list = tasks_data.get("tasks", []) + if isinstance(tasks_list, list) and tasks_list: + return tasks_list raise RuntimeError("AI 管家任务链生成失败：返回 tasks 为空") @staticmethod diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index f0185da2..0a52a81e 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -172,8 +172,8 @@ FAST_LOCAL_PLANNER_FUNCTION_HINTS = ( "signal_source_onboarding_plan", ) FAST_LOCAL_WORKER_FUNCTION_HINTS = ("signal_source_ai_worker_execution",) -FAST_LOCAL_FALLBACK_FAST_TIMEOUT_SECONDS = 3.0 -FAST_LOCAL_FALLBACK_COMPLETE_TIMEOUT_SECONDS = 6.0 +FAST_LOCAL_FALLBACK_FAST_TIMEOUT_SECONDS = 20.0 +FAST_LOCAL_FALLBACK_COMPLETE_TIMEOUT_SECONDS = 35.0 STRUCTURED_CACHE_TTL_SECONDS = 300 LLM_ENGINE_SETTINGS_DEFAULTS = { "enable_local_kb_qa": False, @@ -202,6 +202,10 @@ FAIL_FAST_ERROR_MARKERS = ( ) CIRCUIT_BREAKER_FAILURE_THRESHOLD = 2 CIRCUIT_BREAKER_COOLDOWN_SECONDS = 300.0 +CIRCUIT_BREAKER_EXEMPT_FUNCTION_PREFIXES = ( + "project_from_text_", + "project_draft_", +) DDL_API_GUARDRAIL_FUNCTION_EXACT = { "classify_intent_ai_pmo_kernel", "classify_intent_complex", @@ -214,15 +218,21 @@ DDL_API_GUARDRAIL_SYSTEM_PROMPT = """ - `classify_intent_ai_pmo_kernel` / `classify_intent_complex` 必须返回 `query.read_only`，并在 reasoning 包含 `ddl_api_required`。 """.strip() STRUCTURED_OBJECT_SCHEMAS: dict[str, dict[str, Any]] = { + "ai_pmo_signal_title_refine": { + "required_keys": ("title",), + "type_checks": { + "title": "str", + }, + }, "signal_source_onboarding_plan": { - "required_keys": ("should_onboard", "summary", "worker_roles", "execution_instruction"), + "required_keys": ("should_onboard", "summary"), "type_checks": { "should_onboard": "bool", "summary": "str", - "worker_roles": "list", - "execution_instruction": "str", }, "optional_type_checks": { + "worker_roles": ("list",), + "execution_instruction": ("str_or_none",), "source": ("dict_or_none",), }, }, @@ -481,6 +491,10 @@ class LLMService: text = str(exc or "").strip().lower() return any(marker in text for marker in FAIL_FAST_ERROR_MARKERS) + @classmethod + def _is_provider_circuit_open_exception(cls, exc: Exception) -> bool: + return "provider_circuit_open" in str(exc or "").strip().lower() + @classmethod def _is_circuit_breaker_worthy_exception(cls, exc: Exception) -> bool: if cls._is_fail_fast_exception(exc): @@ -512,6 +526,8 @@ class LLMService: @classmethod def _record_provider_failure(cls, circuit_key: str, exc: Exception) -> None: + if cls._is_provider_circuit_open_exception(exc): + return if not cls._is_circuit_breaker_worthy_exception(exc): return now = time.monotonic() @@ -903,6 +919,24 @@ class LLMService: raise ValueError("invalid_structured_output") return response_text except Exception as exc: + failover_response, failover_engine = await self._retry_with_high_tier_after_low_model_failure( + db=db, + tenant_id=tenant_id, + function_name=function_name, + failed_engine=engine_meta, + prompt=prompt_with_context, + system_prompt=str(prompt_profile.get("system_prompt") or ""), + temperature=temperature, + max_tokens=max_tokens, + response_mode=response_mode, + error=exc, + ) + if failover_response and isinstance(failover_engine, dict): + routing_decision = "glm47_error_escalate_to_high" + status = "success_failover" + response_text = failover_response + engine_meta = failover_engine + return failover_response status = "failed" reason_text = self._build_model_error_message( function_name=function_name, @@ -962,6 +996,7 @@ class LLMService: merged_prompt = prompt_with_context provider = "" model = "" + engine_meta: dict[str, Any] = {} status = "success" response_text = "" try: @@ -977,6 +1012,7 @@ class LLMService: tenant_id=tenant_id, function_name=function_name, ) + engine_meta = engine provider = str(engine.get("provider") or "") model = str(engine.get("model") or "") cache_key = "" @@ -1038,6 +1074,24 @@ class LLMService: status = extraction.error_code or "success" return extraction except Exception as exc: + failover_result, failover_engine = await self._retry_structured_with_high_tier_after_low_model_failure( + db=db, + tenant_id=tenant_id, + function_name=function_name, + failed_engine=engine_meta, + prompt=prompt_with_context, + system_prompt=str(prompt_profile.get("system_prompt") or ""), + temperature=temperature, + max_tokens=max_tokens, + response_mode=response_mode, + error=exc, + ) + if isinstance(failover_result, StructuredExtractionResult) and isinstance(failover_engine, dict): + provider = str(failover_engine.get("provider") or "") + model = str(failover_engine.get("model") or "") + response_text = failover_result.raw_text or json.dumps(failover_result.payload or {}, ensure_ascii=False) + status = failover_result.error_code or "success_failover" + return failover_result error_code = self._classify_structured_exception(exc) status = error_code reason_text = self._build_model_error_message( @@ -2003,6 +2057,190 @@ class LLMService: return False return self._looks_like_unanswerable_response(response_text) + @staticmethod + def _is_glm_47_model(model_name: str) -> bool: + normalized = re.sub(r"[\s_]+", "", str(model_name or "")).lower() + if not normalized: + return False + return "glm" in normalized and "4.7" in normalized + + def _should_retry_glm47_with_high_tier( + self, + *, + function_name: str, + failed_engine: dict[str, Any], + ) -> bool: + if not self._low_to_high_escalation_enabled(): + return False + failed_model = str(failed_engine.get("model") or "").strip() + if self._is_glm_47_model(failed_model): + return True + # 兜底兼容：低阶函数遇到 4.x GLM 也允许升级高阶重试。 + if self._is_low_tier_function(function_name): + lowered = failed_model.lower() + return "glm" in lowered and "4." in lowered + return False + + async def _retry_with_high_tier_after_low_model_failure( + self, + *, + db: AsyncSession, + tenant_id: int, + function_name: str, + failed_engine: dict[str, Any], + prompt: str, + system_prompt: str, + temperature: float, + max_tokens: int, + response_mode: str, + error: Exception, + ) -> tuple[Optional[str], Optional[dict[str, Any]]]: + if not self._should_retry_glm47_with_high_tier(function_name=function_name, failed_engine=failed_engine): + return None, None + try: + _, bundle = await self._get_or_init_bundle(db, tenant_id=tenant_id) + engines = [item for item in bundle.get("engines", []) if isinstance(item, dict)] + active_engine = self._resolve_active_engine_from_bundle(bundle) + high_engine = self._resolve_tier_engine( + tier="high", + engines=engines, + active_engine=active_engine, + ) + if not isinstance(high_engine, dict): + return None, None + high_engine = self._ensure_engine_allowed_for_function( + engine=high_engine, + function_name=function_name, + settings=self._normalize_engine_settings(bundle.get("settings")), + engine_label="high_tier_failover", + ) + failed_provider = str(failed_engine.get("provider") or "").strip().lower() + failed_model = str(failed_engine.get("model") or "").strip() + high_provider = str(high_engine.get("provider") or "").strip().lower() + high_model = str(high_engine.get("model") or "").strip() + if failed_provider == high_provider and failed_model == high_model: + return None, None + _runtime_log( + "LLM路由", + ( + "低阶模型失败自动升级：" + f"function={function_name}, from={failed_provider}:{failed_model}, " + f"to={high_provider}:{high_model}, reason={str(error)[:120]}" + ), + ) + text = await self._chat_completion( + provider=high_provider, + prompt=prompt, + system_prompt=system_prompt, + api_key=str(high_engine.get("api_key") or ""), + base_url=str(high_engine.get("base_url") or ""), + model=high_model, + temperature=temperature, + max_tokens=max_tokens, + function_name=function_name, + response_mode=response_mode, + ) + if not text.strip(): + raise ValueError("high_tier_empty_response") + if self._is_invalid_structured_output(function_name=function_name, response_text=text): + raise ValueError("high_tier_invalid_structured_output") + return text, high_engine + except Exception as retry_exc: + _runtime_log( + "LLM路由", + ( + "低阶模型失败升级高阶也失败：" + f"function={function_name}, error={type(retry_exc).__name__}:{str(retry_exc)[:180]}" + ), + ) + return None, None + + async def _retry_structured_with_high_tier_after_low_model_failure( + self, + *, + db: AsyncSession, + tenant_id: int, + function_name: str, + failed_engine: dict[str, Any], + prompt: str, + system_prompt: str, + temperature: float, + max_tokens: int, + response_mode: str, + error: Exception, + ) -> tuple[Optional[StructuredExtractionResult], Optional[dict[str, Any]]]: + if not self._should_retry_glm47_with_high_tier(function_name=function_name, failed_engine=failed_engine): + return None, None + try: + _, bundle = await self._get_or_init_bundle(db, tenant_id=tenant_id) + engines = [item for item in bundle.get("engines", []) if isinstance(item, dict)] + active_engine = self._resolve_active_engine_from_bundle(bundle) + high_engine = self._resolve_tier_engine( + tier="high", + engines=engines, + active_engine=active_engine, + ) + if not isinstance(high_engine, dict): + return None, None + high_engine = self._ensure_engine_allowed_for_function( + engine=high_engine, + function_name=function_name, + settings=self._normalize_engine_settings(bundle.get("settings")), + engine_label="high_tier_structured_failover", + ) + failed_provider = str(failed_engine.get("provider") or "").strip().lower() + failed_model = str(failed_engine.get("model") or "").strip() + high_provider = str(high_engine.get("provider") or "").strip().lower() + high_model = str(high_engine.get("model") or "").strip() + if failed_provider == high_provider and failed_model == high_model: + return None, None + _runtime_log( + "LLM路由", + ( + "低阶结构化失败自动升级：" + f"function={function_name}, from={failed_provider}:{failed_model}, " + f"to={high_provider}:{high_model}, reason={str(error)[:120]}" + ), + ) + completion = await self._chat_completion_raw( + provider=high_provider, + prompt=prompt, + system_prompt=system_prompt, + api_key=str(high_engine.get("api_key") or ""), + base_url=str(high_engine.get("base_url") or ""), + model=high_model, + temperature=temperature, + max_tokens=max_tokens, + function_name=function_name, + response_mode=response_mode, + ) + extraction = self._extract_structured_object_from_response( + function_name=function_name, + response=completion.raw_response, + fallback_text=completion.text, + provider=high_provider, + model=high_model, + ) + if extraction.payload is None: + raise ValueError(str(extraction.error_code or "structured_schema_invalid")) + _runtime_log( + "LLM结构化", + ( + f"function={function_name}, structured_source={extraction.source_field or 'none'}, " + f"validation={extraction.validation}, fallback=true" + ), + ) + return extraction, high_engine + except Exception as retry_exc: + _runtime_log( + "LLM路由", + ( + "低阶结构化失败升级高阶也失败：" + f"function={function_name}, error={type(retry_exc).__name__}:{str(retry_exc)[:180]}" + ), + ) + return None, None + @staticmethod def _requires_ddl_api_guardrail(function_name: str) -> bool: fn = str(function_name or "").strip().lower() @@ -2166,6 +2404,9 @@ class LLMService: return min(grown, 3000) def _use_provider_circuit_breaker(self, function_name: str) -> bool: + fn = str(function_name or "").strip().lower() + if any(fn.startswith(prefix) for prefix in CIRCUIT_BREAKER_EXEMPT_FUNCTION_PREFIXES): + return False return not self._prefer_fast_local_fallback(function_name) def _structured_output_required(self, function_name: str) -> bool: @@ -2332,10 +2573,12 @@ class LLMService: @staticmethod def _reasoning_content_fallback_enabled(*, function_name: str) -> bool: + fn = str(function_name or "").strip().lower() + if fn and ("title_refine" in fn or "project_draft_analysis_glm47_fallback" in fn): + return True raw = str(os.getenv("LLM_REASONING_CONTENT_FALLBACK", "0")).strip().lower() if raw not in {"1", "true", "yes", "on"}: return False - fn = str(function_name or "").strip().lower() if not fn: return True if fn.startswith("generate_structured") or fn.startswith("llm_generate_structured"): @@ -2723,7 +2966,7 @@ class LLMService: upper = tier.upper() default_provider_by_tier = { "low": "glm", - "high": "glm", + "high": "openai", "safety": "glm", } provider_env = str( diff --git a/backend/app/services/signal_source_orchestration_service.py b/backend/app/services/signal_source_orchestration_service.py index 9feb4385..40978f7c 100644 --- a/backend/app/services/signal_source_orchestration_service.py +++ b/backend/app/services/signal_source_orchestration_service.py @@ -503,20 +503,12 @@ class SignalSourceOrchestrationService: tenant_id=tenant_id, function_name="signal_source_onboarding_plan", use_prompt_profile=False, - max_tokens=420, - response_mode=response_mode, + max_tokens=1800, + response_mode="complete" if response_mode == "fast" else response_mode, ) diagnostics = self._build_planner_diagnostics(extraction) if extraction.payload is None: - return ( - self._heuristic_onboarding_plan( - initiating_text=initiating_text, - requested_via=requested_via, - error=str(extraction.error_code or "structured_unknown"), - ), - "fallback", - diagnostics, - ) + raise ValueError(f"signal_source_onboarding_plan_failed:{str(extraction.error_code or 'structured_empty')}") plan = extraction.payload normalized = self._normalize_onboarding_plan( plan=plan, @@ -525,21 +517,7 @@ class SignalSourceOrchestrationService: ) return normalized, "planned", diagnostics except Exception as exc: - return ( - self._heuristic_onboarding_plan( - initiating_text=initiating_text, - requested_via=requested_via, - error=str(exc), - ), - "fallback", - { - "used_fallback": True, - "error_code": "structured_empty", - "source_field": None, - "provider": "", - "model": "", - }, - ) + raise RuntimeError(f"signal_source_onboarding_plan_failed:{str(exc)}") from exc def _normalize_onboarding_plan( self, @@ -576,26 +554,6 @@ class SignalSourceOrchestrationService: )[:600], } - def _heuristic_onboarding_plan( - self, - *, - initiating_text: str, - requested_via: str, - error: str, - ) -> dict[str, Any]: - source = self._normalize_source_payload( - source_payload={}, - fallback_text=initiating_text, - requested_via=requested_via, - ) - return { - "should_onboard": True, - "summary": "系统已按规则兜底为信号源接入请求，并生成最小接入闭环。", - "source": source, - "worker_roles": list(DEFAULT_WORKER_ROLE_KEYS), - "execution_instruction": self._build_execution_instruction(source, initiating_text), - } - def _normalize_source_payload( self, *, @@ -1079,24 +1037,23 @@ class SignalSourceOrchestrationService: "used_fallback": False, } except Exception as exc: - fallback = self._fallback_worker_output( - role_name=role_name, - source_name=str(source_payload.get("source_name") or "信号源"), - ) if execution_row is not None: - execution_row.status = "completed" - execution_row.summary = str(fallback.get("summary") or "").strip()[:1000] - execution_row.output_payload = fallback + execution_row.status = "failed" + execution_row.summary = str(exc)[:1000] + execution_row.output_payload = { + "error": str(exc)[:1000], + "used_fallback": False, + } execution_row.error_message = str(exc)[:1000] execution_row.completed_at = _utc_now_naive() await self.db.commit() return { "worker_role_key": role_key, "worker_role_name": role_name, - "status": "completed", + "status": "failed", "reason": str(exc), - "summary": str(fallback.get("summary") or "").strip(), - "used_fallback": True, + "summary": str(exc)[:200], + "used_fallback": False, "execution_id": int(execution_row.id) if execution_row is not None else None, } @@ -1326,7 +1283,7 @@ class SignalSourceOrchestrationService: @staticmethod def _build_planner_diagnostics(extraction: Any) -> dict[str, Any]: return { - "used_fallback": extraction.error_code is not None, + "used_fallback": False, "error_code": extraction.error_code, "source_field": extraction.source_field or None, "provider": str(extraction.provider or ""), @@ -1694,16 +1651,6 @@ class SignalSourceOrchestrationService: "used_fallback": False, } - @staticmethod - def _fallback_worker_output(*, role_name: str, source_name: str) -> dict[str, Any]: - return { - "summary": f"{role_name} 已为 {source_name} 生成兜底执行摘要。", - "deliverables": [f"梳理 {source_name} 的接入目标与边界"], - "risks": [f"{source_name} 的标识与接入细节仍需补充确认"], - "next_actions": ["等待显式服务完成落库并继续推进执行"], - "used_fallback": True, - } - async def _persist_signal_orchestration_result( self, *,