diff --git a/frontend-web/src/roundtable-planning/api/flow-extract.ts b/frontend-web/src/roundtable-planning/api/flow-extract.ts index ec83e616..731361a5 100644 --- a/frontend-web/src/roundtable-planning/api/flow-extract.ts +++ b/frontend-web/src/roundtable-planning/api/flow-extract.ts @@ -101,6 +101,11 @@ export interface FlowExtractRequest { onValidating?: () => void; /** 校验/修复流式增量(同样含思考 + 正文,前端实时显示校验过程)。 */ onValidateDelta?: (text: string) => void; + /** + * 模型自动容错:后端某模型抽取失败,已切到下一个模型重抽。上层据此**清空已累计的抽取增量** + * (`onExtractDelta` 之前吐出的半截内容)并提示用户。`failedModel`/`nextModel`/`reason` 描述切换。 + */ + onModelSwitch?: (info: { failedModel: string; nextModel: string; reason: string }) => void; } /** @@ -161,6 +166,12 @@ export async function streamFlowExtract(req: FlowExtractRequest): Promise, ) => void; + /** + * 模型自动容错:后端检测到当前模型出问题(兜底错误文案 / 交付为空等),已自动切到下一个 + * 模型并即将重跑本轮时触发。上层据此把该角色「正在流式的失败气泡」重置(清空已累计的增量 + * 文本)并给用户一条「『X』模型异常,已切换『Y』继续」的提示。此回调可能多次触发(连续多个 + * 模型都失败时)。触发后,本次 streamMultiAgent 仍会继续用新模型流式,最终照常 resolve。 + */ + onModelSwitch?: (info: ModelSwitchInfo) => void; +} + +/** 模型自动容错通知(`streamMultiAgent` 的 `onModelSwitch` 回调入参)。 */ +export interface ModelSwitchInfo { + /** 角色:leader / seat / seat_parallel / report / summary / dashboard / ingest。 */ + role: string; + /** 出问题的智能体 agent_name。 */ + agentName: string; + /** 失败的模型名。 */ + failedModel: string; + /** 已切换到的下一个模型名。 */ + nextModel: string; + /** 失败原因(已翻译成简短中文,如「服务繁忙」)。 */ + reason: string; } export interface ClarificationPayload { @@ -330,6 +351,8 @@ export interface StreamMultiAgentResult { content: string; /** Echoed agent_name of the agent that produced this run. */ agentName: string | null; + /** 实际产出本轮结果的模型(发生模型自动容错时即最终切换到的模型);后端未下发时为 null。 */ + modelUsed?: string | null; /** * Populated when the leader called ask_clarification — the run ended without * dispatching, and the model is asking the user a question instead. When @@ -555,6 +578,7 @@ export async function streamMultiAgent( content: "", agentName: null, clarification: null, + modelUsed: null, }; // Inferred SSE phase + a tiny helper so we only fire onPhaseChange when the @@ -680,6 +704,28 @@ export async function streamMultiAgent( }); } + // 模型自动容错帧:后端检测到当前模型出问题,已切到下一个模型并即将重跑本轮。 + // 不是终态帧(无 content)→ 不结束流。清空已累计的(失败尝试的)tool_call 增量, + // 把 phase 退回 connected,并通知上层重置该角色气泡 + 给用户提示。随后新模型的 + // 流式增量会照常通过 onTextDelta 继续累计到(已被上层清空的)气泡。 + if ( + parsed && + typeof parsed === "object" && + (parsed as Record).status === "model_switch" + ) { + const obj = parsed as Record; + toolCallAcc.clear(); + currentPhase = "connected"; + req.onModelSwitch?.({ + role: typeof obj.role === "string" ? obj.role : "", + agentName: typeof obj.agent_name === "string" ? obj.agent_name : req.agentName, + failedModel: typeof obj.failed_model === "string" ? obj.failed_model : "", + nextModel: typeof obj.next_model === "string" ? obj.next_model : "", + reason: typeof obj.reason === "string" ? obj.reason : "", + }); + continue; + } + // Phase inference from LangGraph messages-tuple frames. We only flip // to tool_calling/tool_result on chunks that clearly indicate them; // everything else with non-empty AI text means we're streaming. @@ -780,6 +826,10 @@ export async function streamMultiAgent( if (typeof parsed.agent_name === "string") { result.agentName = parsed.agent_name; } + { + const mu = (parsed as Record).model_used; + if (typeof mu === "string" && mu) result.modelUsed = mu; + } // SSE error 帧(后端 stream_generator 在席位/Step3 run 失败时下发 // `{status:"error", content, agent_name}`):上报,便于收集用户侧报错。 if (result.message === "error") { diff --git a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts index 37fc60cb..642f317c 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts @@ -20,6 +20,7 @@ import { useArtifacts } from "@/components/workspace/artifacts"; import { initMultiAgent, streamMultiAgent, + type ModelSwitchInfo, type ClarificationPayload, type LeaderDispatch, type StreamPhase, @@ -1204,6 +1205,24 @@ export function useStep2Orchestration( [], ); + /** + * 模型自动容错通知(前台):后端切到下一个模型重跑本轮时调用。弹 toast + 在时间线插一条 + * 「系统提示」气泡,让用户看到「『X』模型异常 → 已切换『Y』继续」。 + */ + const notifyModelSwitch = useCallback( + (info: ModelSwitchInfo) => { + const who = info.agentName || "智能体"; + triggerToast(`⚠️「${info.failedModel}」模型${info.reason},已自动切换「${info.nextModel}」继续`); + appendDialogue({ + sender: "系统提示", + avatarType: "amber", + content: `⚠️ 模型容错:「${who}」所用模型「${info.failedModel}」${info.reason},已自动切换到「${info.nextModel}」继续。`, + confidence: "模型容错", + }); + }, + [appendDialogue, triggerToast], + ); + const phaseLabel = useCallback( ( phase: StreamPhase, @@ -1319,6 +1338,20 @@ export function useStep2Orchestration( ); }, []); + /** + * 生成某气泡的 onModelSwitch 处理器:弹提示 + 把该气泡里「失败尝试已累计的半截内容」清空 + * (随后新模型的流式增量会重新填进同一气泡),并把状态标为「模型切换中…」。 + */ + const makeModelSwitchHandler = useCallback( + (bubbleId: number, isParallel: boolean) => (info: ModelSwitchInfo) => { + notifyModelSwitch(info); + if (isParallel) parallelStreamRef.current.delete(bubbleId); + setDialogueContent(bubbleId, ""); + setDialogueConfidence(bubbleId, "模型切换中…"); + }, + [notifyModelSwitch, setDialogueContent, setDialogueConfidence], + ); + // ─── §6 并行流式:全接收 / 惰性渲染 / 节流 ─────────────────────────────── /** * 节流 flush(~80ms 一拍,约 12fps):**仅被选中**的席位把 ref 累积的全文挂到 @@ -1650,6 +1683,7 @@ export function useStep2Orchestration( // agent 自身开关取 OR。 seatSkillDirective: seatSkillDirectivesRef.current[a.agentId] ?? false, signal: subAborter.signal, + onModelSwitch: makeModelSwitchHandler(a.bubbleId, a.isParallel), onTextDelta: (text) => a.isParallel ? ingestParallelDelta(a.bubbleId, text) : appendToDialogue(a.bubbleId, text), onPhaseChange: (phase, toolName, args) => { @@ -1739,6 +1773,7 @@ export function useStep2Orchestration( files: takeOnce(pendingInjectFilesRef), model: selectedModelRef.current || undefined, signal: leaderAborter.signal, + onModelSwitch: makeModelSwitchHandler(leaderBubbleId, false), onTextDelta: (text) => appendToDialogue(leaderBubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "leader", toolName); @@ -2014,6 +2049,7 @@ export function useStep2Orchestration( files: takeOnce(pendingInjectFilesRef), model: selectedModelRef.current || undefined, signal: leaderAborter.signal, + onModelSwitch: makeModelSwitchHandler(leaderBubbleId, false), onTextDelta: (text) => appendToDialogue(leaderBubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "leader", toolName); @@ -2095,6 +2131,7 @@ export function useStep2Orchestration( // agent 自身开关取 OR。 seatSkillDirective: seatSkillDirectivesRef.current[seat.agent_id] ?? false, signal: subAborter.signal, + onModelSwitch: makeModelSwitchHandler(subBubbleId, false), onTextDelta: (text) => appendToDialogue(subBubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "special", toolName); @@ -2158,6 +2195,7 @@ export function useStep2Orchestration( suppressPreBroadcast: true, model: selectedModelRef.current || undefined, signal: closeoutAborter.signal, + onModelSwitch: makeModelSwitchHandler(closeoutBubbleId, false), onTextDelta: (text) => appendToDialogue(closeoutBubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "leader", toolName); @@ -2319,6 +2357,7 @@ export function useStep2Orchestration( files: takeOnce(pendingInjectFilesRef), model: selectedModelRef.current || undefined, signal: leaderAborter.signal, + onModelSwitch: makeModelSwitchHandler(leaderBubbleId, false), onTextDelta: (text) => appendToDialogue(leaderBubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "leader", toolName); @@ -2444,6 +2483,7 @@ export function useStep2Orchestration( signal: subAborter.signal, // §6:并行席位走 ref 全量累积 + 节流 flush(只渲染选中的);单节点 // stage 实际串行,1 路流直接 setState 不卡,沿用 appendToDialogue。 + onModelSwitch: makeModelSwitchHandler(bubbleId, isParallel), onTextDelta: (text) => isParallel ? ingestParallelDelta(bubbleId, text) : appendToDialogue(bubbleId, text), onPhaseChange: (phase, toolName, args) => { @@ -2624,6 +2664,7 @@ export function useStep2Orchestration( ...seatModeToRequestFields(seatModesRef.current[agentId] ?? selectedModeRef.current), seatSkillDirective: seatSkillDirectivesRef.current[agentId] ?? false, signal: subAborter.signal, + onModelSwitch: makeModelSwitchHandler(bubbleId, true), onTextDelta: (text) => ingestParallelDelta(bubbleId, text), onPhaseChange: (phase, toolName, args) => { const label = phaseLabel(phase, "special", toolName); diff --git a/frontend-web/src/roundtable-planning/hooks/useStep3Dashboard.ts b/frontend-web/src/roundtable-planning/hooks/useStep3Dashboard.ts index c39d9773..fd154cc1 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep3Dashboard.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep3Dashboard.ts @@ -325,6 +325,12 @@ export function useStep3Dashboard(options: UseStep3DashboardOptions): UseStep3Da newMessage: message, model: selectedModel || undefined, signal: aborter.signal, + // 模型自动容错:后端切到下一个模型重跑 → 清空失败尝试的半截累计(acc + 气泡)。 + onModelSwitch: (info) => { + acc = ""; + setDialogues((prev) => prev.map((d) => (d.id === bubbleId ? { ...d, content: "" } : d))); + setDialogueConfidence(bubbleId, `模型切换:${info.failedModel}→${info.nextModel}(${info.reason})`); + }, onTextDelta: (text) => { appendToDialogue(bubbleId, text); acc += text; diff --git a/frontend-web/src/roundtable-planning/hooks/useStep3Report.ts b/frontend-web/src/roundtable-planning/hooks/useStep3Report.ts index 41bbc4f2..5eb4d4e2 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep3Report.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep3Report.ts @@ -770,6 +770,16 @@ export function useStep3Report( newMessage: prompt, model: selectedModel || undefined, signal: aborter.signal, + // 模型自动容错:后端切到下一个模型重跑 → 清空失败尝试半截正文 + 步骤卡留可见提示。 + onModelSwitch: (info) => { + setDialogues((prev) => prev.map((d) => (d.id === bubbleId ? { ...d, content: "" } : d))); + setDialogueConfidence(bubbleId, `模型切换:${info.failedModel}→${info.nextModel}`); + appendStepIfNew( + bubbleId, + `⚠️ 模型容错:「${info.failedModel}」${info.reason},改用「${info.nextModel}」`, + `model_switch:${info.nextModel}`, + ); + }, onTextDelta: (text) => appendToDialogue(bubbleId, text), onPhaseChange: (p, toolName, toolArgs) => { const label = reportPhaseLabel(p, toolName); diff --git a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts index 48a0b6c7..804dcbc4 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts @@ -1059,6 +1059,16 @@ export function useStep3Summary( model: selectedModel || undefined, files: files && files.length > 0 ? files : undefined, signal: aborter.signal, + // 模型自动容错:后端切到下一个模型重跑 → 清空失败尝试的半截正文 + 在步骤卡留一条可见提示。 + onModelSwitch: (info) => { + setDialogues((prev) => prev.map((d) => (d.id === bubbleId ? { ...d, content: "" } : d))); + setDialogueConfidence(bubbleId, `模型切换:${info.failedModel}→${info.nextModel}`); + appendStepIfNew( + bubbleId, + `⚠️ 模型容错:「${info.failedModel}」${info.reason},改用「${info.nextModel}」`, + `model_switch:${info.nextModel}`, + ); + }, onTextDelta: (text) => appendToDialogue(bubbleId, text), onPhaseChange: (p, toolName, pArgs) => { const label = summaryPhaseLabel(p, toolName); @@ -1160,6 +1170,16 @@ export function useStep3Summary( currentGraph: opts?.currentGraph ?? null, instruction: opts?.instruction, signal: aborter.signal, + // 模型自动容错:后端切到下一个模型重抽 → 清空抽取气泡里失败尝试的半截内容 + 留可见提示。 + onModelSwitch: (info) => { + setDialogues((prev) => prev.map((d) => (d.id === extractBubbleId ? { ...d, content: "" } : d))); + setDialogueConfidence(extractBubbleId, `模型切换:${info.failedModel}→${info.nextModel}`); + appendStepIfNew( + extractBubbleId, + `⚠️ 模型容错:「${info.failedModel}」${info.reason},改用「${info.nextModel}」`, + `model_switch:${info.nextModel}`, + ); + }, onExtractDelta: (t) => appendToDialogue(extractBubbleId, t), onValidating: () => { setFlowPhase("validating"); diff --git a/frontend-web/src/roundtable-planning/lib/job-live.ts b/frontend-web/src/roundtable-planning/lib/job-live.ts index c5f107ec..6560fda0 100644 --- a/frontend-web/src/roundtable-planning/lib/job-live.ts +++ b/frontend-web/src/roundtable-planning/lib/job-live.ts @@ -66,6 +66,10 @@ function toStep2Dialogue(d: JobDialogue, idxByAgent: Map): Step2 if (d.role === "leader") { return { ...base, sender: "总控协调", avatarType: "blue", confidence: "派活说明" }; } + // 模型自动容错的系统提示气泡(后端 progress.model_switch 以 role="system" 落库)。 + if (d.role === "system") { + return { ...base, sender: d.name || "系统提示", avatarType: "amber", confidence: "模型容错" }; + } const aid = d.agentId ?? ""; return { ...base, diff --git a/frontend-web/src/roundtable-planning/lib/job-types.ts b/frontend-web/src/roundtable-planning/lib/job-types.ts index 922edd66..30daffa3 100644 --- a/frontend-web/src/roundtable-planning/lib/job-types.ts +++ b/frontend-web/src/roundtable-planning/lib/job-types.ts @@ -52,7 +52,8 @@ export interface DispatchChainNode { /** 一条研讨对话记录(leader 说明 / 某席位交付)。 */ export interface JobDialogue { id: number; - role: "leader" | "seat"; + // "system" = 模型自动容错等系统提示气泡(后端 progress.model_switch 落库)。 + role: "leader" | "seat" | "system"; agentId: string | null; name: string; content: string; diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 01c702c6..22eca306 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -620,6 +620,19 @@ scheduler runs once per deployment. Per run: - Config values starting with `$` resolved as environment variables - Missing provider modules surface actionable install hints from reflection resolvers (for example `uv add langchain-google-genai`) +### Roundtable Model Fault Tolerance (会商模型自动容错 / 换下一个模型) + +会商第二步(leader/seat)+ 第三步(summary/report/dashboard + 结构化抽取)单轮执行时,**某模型出问题就自动切到下一个模型继续**,并告知用户「『X』模型异常 → 已切换『Y』」。前台实时(`multi_agent.py`)与后台挂起(`engine.py` + `InProcessRoundtableGateway`)两条路径都覆盖。 + +- **可检测的模型失败**:`LLMErrorHandlingMiddleware` 在同一模型重试 3 次后**不抛异常**,而是返回兜底 `AIMessage` —— 现在给它打 `additional_kwargs[LLM_ERROR_MARKER="deerflow_llm_error"]=reason`(model_dump 保留,经 checkpointer 序列化仍在)。`is_llm_error_message(msg)` 据标记(回退按已知兜底文案正文)识别,返回 reason。 +- **候选模型链**:`app/gateway/roundtable_model_fallback.py::fallback_model_chain(current)` 按 `config.yaml models[]` 顺序给出 `[当前模型, …其余]`(去重保序)。`reason_text(reason)` 翻译成简短中文。总开关 env `ROUNDTABLE_MODEL_FALLBACK`(默认开,`0/false/off` 关 → 单模型旧行为);`ROUNDTABLE_MODEL_FALLBACK_MAX`(0=试完全部)。 +- **触发换模型的三种情形**:① 单轮 run 抛异常(start_run 失败 / idle 卡死 → `timeout`,其余 `exception`);② 兜底错误文案(`is_llm_error_message`);③ 交付为空(仅 seat/leader;leader 还含「既无正文又无派活」)。Step3 单例(report/summary/dashboard/ingest/结构化)的产物是落盘文件,**不**把「正文为空」当失败,只对真模型错误换模型。 +- **后台**:`InProcessRoundtableGateway._run_turn_with_fallback(...)` 包住 `_run_agent_turn`,逐候选模型试;换模型落 `model_switch` 诊断、全失败落 `model_fallback_exhausted`。`run_leader/run_seat/run_report/run_summary/run_dashboard_data` 全走它(report/summary/dashboard 的产物自愈重试保留,并记住已切到的可用模型用于后续续写轮)。各 Turn/Result 带回 `switches: list[ModelSwitch]`,引擎 `_emit_switches` → `JobProgressWriter.model_switch(...)` 往研讨时间线插一条 `role="system"` 系统提示气泡。 +- **前台**:`_special_run`/`_leader_run` stream 收尾后检测失败 → yield `{status:"model_switch", role, agent_name, failed_model, next_model, reason}` 帧并用下一个模型重新流式;终态帧带 `model_used`。结构化抽取 `flow_extract.py` 的 round1 `model.astream` 也包了同款换模型循环(emit `event:"model_switch"` 帧)。 +- **Step2 单席位隔离**:`engine._run_recommend/_run_chain/_run_dag` 的每个席位执行包 try/except —— 一个席位彻底失败(落 `seat_failed` 诊断 + 置空交付)不拖垮整批/整链/整个作业。 +- **前端**:`multi-agent.ts` 解析 `model_switch` 帧 → `onModelSwitch` 回调(清空失败尝试的半截气泡 + 累计的 tool_call delta,phase 退回 connected);`useStep2Orchestration` 弹 toast + 时间线插「系统提示」气泡;`useStep3Summary/Report/Dashboard` + `flow-extract.ts` 重置对应气泡并在步骤卡留可见提示;后台路径经 `job-live.ts::toStep2Dialogue` 把 `role="system"` 渲染成 amber「系统提示」气泡。 +- Tests: `tests/test_roundtable_model_fallback.py`。 + ### vLLM Provider (`packages/harness/deerflow/models/vllm_provider.py`) - `VllmChatModel` subclasses `langchain_openai:ChatOpenAI` for vLLM 0.19.0 OpenAI-compatible endpoints diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py index b589bcee..fadc3093 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py @@ -18,12 +18,18 @@ import asyncio import logging import os import uuid +from collections.abc import Callable from datetime import UTC, datetime from types import SimpleNamespace from typing import Any from langgraph.checkpoint.base import empty_checkpoint +from app.gateway.roundtable_model_fallback import ( + fallback_model_chain, + is_llm_error_message, + reason_text, +) from app.gateway.roundtable_run_policy import roundtable_run_policy from app.gateway.roundtable_seat_skills import append_seat_skill_directive from app.gateway.routers._roundtable_seed import ( @@ -36,6 +42,7 @@ from app.gateway.routers._roundtable_seed import ( from deerflow.agents.roundtable_orchestrator.report_json import extract_report_json from deerflow.agents.roundtable_orchestrator.types import ( LeaderTurn, + ModelSwitch, ReportResult, SeatRef, SeatTurn, @@ -521,6 +528,31 @@ _SEAT_MODE_PARAMS: dict[str, tuple[bool | None, str | None, bool]] = { } +def _exception_reason(exc: BaseException) -> str: + """把单轮 run 抛出的异常归类成模型容错原因码(用于换模型提示)。 + + idle 守护判定的卡死(``_await_run_with_idle_guard`` 抛 ``... stalled ...``)→ ``timeout``; + 其余(start_run 失败 / DB 抖动 / 其它)→ ``exception``。 + """ + text = str(exc).lower() + if "stall" in text or "idle" in text or "timeout" in text or "timed out" in text: + return "timeout" + return "exception" + + +def _seat_empty_reason(messages: list[dict[str, Any]]) -> str | None: + """席位/取数轮校验:最后一条有正文的 AI 消息为空 → ``empty``(触发换模型)。""" + return "empty" if not _latest_ai_text(messages).strip() else None + + +def _leader_failed_reason(messages: list[dict[str, Any]]) -> str | None: + """总控轮校验:既无正文、也无任何工具调用(派活/澄清)→ ``empty``(视为模型没干活)。""" + ai = _latest_ai_message(messages) + has_text = bool(_latest_ai_text(messages).strip()) + has_tools = bool(_normalize_tool_calls((ai or {}).get("tool_calls"))) + return None if (has_text or has_tools) else "empty" + + # ── 网关实现 ───────────────────────────────────────────────────────────────── @@ -585,25 +617,116 @@ class InProcessRoundtableGateway: self._seat_intro = _build_seat_intro(self._agents) return thread_ids, COORDINATOR_AGENT_ID + async def _run_turn_with_fallback( + self, + *, + thread_id: str, + agent_name: str, + message: str, + requested_model: str | None, + excluded_tools: list[str], + skill_stop_names: list[str], + thinking_enabled: bool, + reasoning_effort: str, + subagent_enabled: bool = False, + stage: str, + cycle: int | None = None, + validate: Callable[[list[dict[str, Any]]], str | None] | None = None, + ) -> tuple[list[dict[str, Any]], list[ModelSwitch], str | None]: + """跑一轮 agent,并在「模型出问题」时自动切到下一个模型重试。 + + 触发换模型的三种情形(与用户选择一致): + ① ``_run_agent_turn`` 抛异常(start_run 失败 / idle 卡死超时); + ② 返回了中间件兜底错误文案(``is_llm_error_message`` 命中); + ③ 调用方 ``validate`` 判定无效(如席位交付为空 / 总控既无正文又无派活)。 + + 候选模型链按 ``config.yaml models[]`` 顺序(``fallback_model_chain``),首个为 + ``requested_model``(空则作业级 ``self._model``,再空则配置默认)。返回 + ``(最终 messages, 发生的换模型记录, 实际所用模型)``。全部模型都失败时返回最后一次的 + messages(让既有的「空交付 / 产物缺失」逻辑兜底,绝不抛异常拖垮整个作业)。 + """ + base = requested_model if (requested_model not in (None, "")) else self._model + chain = fallback_model_chain(base) or [base] + switches: list[ModelSwitch] = [] + messages: list[dict[str, Any]] = [] + prev_model: str | None = None + prev_reason: str | None = None + used_model: str | None = chain[0] + for idx, candidate in enumerate(chain): + if idx > 0: + label = reason_text(prev_reason) + switches.append(ModelSwitch(failed_model=prev_model or "", next_model=candidate or "", reason_label=label)) + logger.warning( + "roundtable %s '%s' switching model %s -> %s (%s)", + stage, agent_name, prev_model, candidate, prev_reason, + ) + await self._diag.record( + stage=stage, level="warning", event="model_switch", + message=f"「{agent_name}」模型「{prev_model}」{label},已自动切换到「{candidate}」重试", + detail={"failed_model": prev_model, "next_model": candidate, "reason": prev_reason, "attempt": idx + 1}, + agent_name=agent_name, cycle=cycle, + ) + used_model = candidate + try: + messages = await _run_agent_turn( + self._app, + user_id=self._user_id, + thread_id=thread_id, + agent_name=agent_name, + message=message, + model=candidate, + excluded_tools=excluded_tools, + skill_stop_names=skill_stop_names, + thinking_enabled=thinking_enabled, + reasoning_effort=reasoning_effort, + subagent_enabled=subagent_enabled, + diag=self._diag, + stage=stage, + cycle=cycle, + ) + except Exception as exc: # noqa: BLE001 — 模型故障:归类后尝试下一个模型 + prev_model, prev_reason, messages = candidate, _exception_reason(exc), [] + if idx + 1 < len(chain): + continue + await self._diag.record( + stage=stage, level="error", event="model_fallback_exhausted", + message=f"「{agent_name}」所有备选模型均失败(最后 {candidate}:{prev_reason})", + detail={"reason": prev_reason, "tried": chain}, agent_name=agent_name, cycle=cycle, + ) + return messages, switches, used_model + # 跑通了:再看是否「兜底错误文案」或 validate 判定无效。 + reason = is_llm_error_message(_latest_ai_message(messages)) + if reason is None and validate is not None: + reason = validate(messages) + if reason is None: + return messages, switches, used_model + prev_model, prev_reason = candidate, reason + if idx + 1 >= len(chain): + await self._diag.record( + stage=stage, level="error", event="model_fallback_exhausted", + message=f"「{agent_name}」所有备选模型均未产出有效结果(最后 {candidate}:{reason})", + detail={"reason": reason, "tried": chain}, agent_name=agent_name, cycle=cycle, + ) + return messages, switches, used_model + return messages, switches, used_model + async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn: msg = message if self._first_leader and self._seat_intro: msg = self._seat_intro + "\n\n" + message self._first_leader = False policy = roundtable_run_policy("leader") - messages = await _run_agent_turn( - self._app, - user_id=self._user_id, + messages, switches, _used = await self._run_turn_with_fallback( thread_id=thread_ids[coordinator_name], agent_name=coordinator_name, message=msg, - model=model or self._model, + requested_model=model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, - diag=self._diag, stage="step2_leader", + validate=_leader_failed_reason, ) ai = _latest_ai_message(messages) # tool_calls 取最后一条 AI 消息(派活就在这条);正文取最后一条有文字的 AI 消息。 @@ -617,7 +740,7 @@ class InProcessRoundtableGateway: if clar is not None: cargs = clar.get("args") or {} question = str(cargs.get("question") or "").strip() or "总控需要补充信息,请补充后继续。" - return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls) + return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls, switches=switches) # 派活解析(复用 multi_agent 纯函数) from app.gateway.routers.multi_agent import _extract_dispatch @@ -630,7 +753,7 @@ class InProcessRoundtableGateway: dispatched.append((name, task)) except Exception: # noqa: BLE001 logger.warning("dispatch parse failed for tool_call", exc_info=True) - return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls) + return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls, switches=switches) async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn: policy = roundtable_run_policy("seat") @@ -654,13 +777,11 @@ class InProcessRoundtableGateway: override = _SEAT_MODE_PARAMS.get(self._seat_modes.get(agent_id, "")) if override is not None: seat_thinking, seat_reasoning, seat_subagent = override - messages = await _run_agent_turn( - self._app, - user_id=self._user_id, + messages, switches, _used = await self._run_turn_with_fallback( thread_id=thread_ids[agent_id], agent_name=agent_id, message=seat_task, - model=model or self._model, + requested_model=model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=( @@ -670,10 +791,14 @@ class InProcessRoundtableGateway: policy.reasoning_effort if seat_reasoning is None else seat_reasoning ), subagent_enabled=seat_subagent, - diag=self._diag, stage="step2_seat", + validate=_seat_empty_reason, + ) + return SeatTurn( + final_text=_latest_ai_text(messages), + tool_calls=_tool_calls_since_last_human(messages), + switches=switches, ) - return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages)) async def create_report_thread(self) -> str: ensure_roundtable_functional_agents() @@ -686,21 +811,22 @@ class InProcessRoundtableGateway: summary = "" html: str | None = None message = prompt + switches: list[ModelSwitch] = [] + current_model = model # 跨产物自愈轮记住实际所用模型:一旦换到可用模型,后续续写轮就从它起。 for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): - messages = await _run_agent_turn( - self._app, - user_id=self._user_id, + messages, sw, used = await self._run_turn_with_fallback( thread_id=report_thread_id, agent_name=REPORT_AGENT_ID, message=message, - model=model or self._model, + requested_model=current_model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, - diag=self._diag, stage="step3_report", ) + switches.extend(sw) + current_model = used or current_model summary = _latest_ai_text(messages) or summary html = _read_report_html(self._app, report_thread_id, self._user_id) if _html_complete(html): @@ -727,7 +853,7 @@ class InProcessRoundtableGateway: ) # 截断的 HTML 也比纯文字强(浏览器容忍未闭合标签),仅在完全没有文件时才回退正文。 final_html = html if (html and html.strip()) else summary - return ReportResult(html=final_html, summary=summary, model=model or self._model) + return ReportResult(html=final_html, summary=summary, model=current_model or self._model, switches=switches) async def create_summary_thread(self) -> str: ensure_roundtable_functional_agents() @@ -743,21 +869,22 @@ class InProcessRoundtableGateway: closing = "" md: str | None = None message = prompt + switches: list[ModelSwitch] = [] + current_model = model for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): - messages = await _run_agent_turn( - self._app, - user_id=self._user_id, + messages, sw, used = await self._run_turn_with_fallback( thread_id=summary_thread_id, agent_name=SUMMARY_AGENT_ID, message=message, - model=model or self._model, + requested_model=current_model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, - diag=self._diag, stage="step3_summary", ) + switches.extend(sw) + current_model = used or current_model closing = _latest_ai_text(messages) or closing md = _read_summary_md(self._app, summary_thread_id, self._user_id) if md and md.strip(): @@ -783,7 +910,7 @@ class InProcessRoundtableGateway: agent_name=SUMMARY_AGENT_ID, ) final_md = md if (md and md.strip()) else closing - return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id) + return SummaryResult(md=final_md, closing=closing, model=current_model or self._model, thread_id=summary_thread_id, switches=switches) async def run_dashboard_data(self, *, prompt, model) -> str | None: """大屏多页的结构化数据:进程内驱动**专职**「大屏多页数据智能体」 @@ -799,21 +926,20 @@ class InProcessRoundtableGateway: thread_id = await _create_thread(self._app) policy = roundtable_run_policy("dashboard") message = prompt + current_model = model for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): - messages = await _run_agent_turn( - self._app, - user_id=self._user_id, + messages, _sw, used = await self._run_turn_with_fallback( thread_id=thread_id, agent_name=DASHBOARD_AGENT_ID, message=message, - model=model or self._model, + requested_model=current_model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, - diag=self._diag, stage="step3_dashboard", ) + current_model = used or current_model body = _extract_report_json(_latest_ai_text(messages)) if body: return body diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_model_fallback.py b/offline-backend-20260512/backend/app/gateway/roundtable_model_fallback.py new file mode 100644 index 00000000..755e93c5 --- /dev/null +++ b/offline-backend-20260512/backend/app/gateway/roundtable_model_fallback.py @@ -0,0 +1,104 @@ +"""圆桌会商「模型自动容错(换下一个模型)」共享助手。 + +会商的第二步(leader / seat)与第三步(summary / report / dashboard / structure)单轮执行时, +若所用模型出现问题(抛异常 / idle 卡死 / 中间件兜底错误文案 / 交付为空),应自动切到 +**下一个可用模型**继续,并把「『X』模型异常 → 已切换『Y』模型」告知用户。 + +本模块只放**与执行路径无关**的纯逻辑,供前台(``multi_agent.py``)与后台 +(``roundtable_inprocess_gateway.py``)两条路径共用: + + - ``model_fallback_enabled()``:总开关(env ``ROUNDTABLE_MODEL_FALLBACK``,默认开); + - ``fallback_model_chain(current)``:按 ``config.yaml`` ``models[]`` 顺序给出候选模型链 + ``[当前模型, …其余]``(去重保序); + - ``reason_text(reason)``:把失败原因码翻译成简短中文(用于用户提示); + - 复用 harness 的 ``is_llm_error_message`` 识别中间件兜底文案。 + +放在 app 层(依赖 ``deerflow.config`` + harness 的中间件常量,皆为 app→harness 合法方向)。 +""" + +from __future__ import annotations + +import logging +import os + +from deerflow.agents.middlewares.llm_error_handling_middleware import ( + is_llm_error_message as is_llm_error_message, # re-export for callers +) +from deerflow.config import get_app_config + +logger = logging.getLogger(__name__) + +# 失败原因码 → 用户可见简短中文(用于「『X』模型<原因>,已切换…」提示)。 +_REASON_TEXT: dict[str, str] = { + "quota": "额度不足/账单异常", + "auth": "鉴权失败/无访问权限", + "busy": "服务繁忙", + "transient": "网络/连接异常", + "upstream": "上游响应异常", + "circuit": "连续失败被熔断", + "generic": "调用失败", + "exception": "调用异常", + "timeout": "长时间无响应(卡死)", + "empty": "未产出有效内容", +} + + +def reason_text(reason: str | None) -> str: + """把失败原因码翻译成简短中文(未知码 → 「调用失败」)。""" + if not reason: + return "调用失败" + return _REASON_TEXT.get(reason, "调用失败") + + +def model_fallback_enabled() -> bool: + """模型自动容错总开关(默认开;设 ``ROUNDTABLE_MODEL_FALLBACK=0/false/off`` 关闭)。 + + 关闭时 ``fallback_model_chain`` 只返回单个候选(原模型),即恢复旧行为:模型失败就失败, + 不再尝试其它模型。 + """ + raw = (os.getenv("ROUNDTABLE_MODEL_FALLBACK", "1") or "1").strip().lower() + return raw not in {"0", "false", "no", "off"} + + +def _max_attempts() -> int: + """最多尝试的模型个数(含首个)。``ROUNDTABLE_MODEL_FALLBACK_MAX``,0/未设 = 不限(试完全部)。""" + try: + return max(0, int(os.getenv("ROUNDTABLE_MODEL_FALLBACK_MAX", "0") or "0")) + except (TypeError, ValueError): + return 0 + + +def _config_model_names() -> list[str]: + try: + cfg = get_app_config() + return [m.name for m in (cfg.models or []) if getattr(m, "name", None)] + except Exception: # noqa: BLE001 — 拿不到配置(内存后端/测试桩)时退化为空 + logger.debug("fallback_model_chain: get_app_config().models unavailable", exc_info=True) + return [] + + +def fallback_model_chain(current: str | None) -> list[str]: + """按 ``config.yaml`` ``models[]`` 顺序构造候选模型链:``[当前模型, …其余]``,去重保序。 + + - ``current`` 为空/None → 从 ``models[0]`` 起(即后端默认模型),后接其余; + - ``current`` 在配置内 → 它排第一,其余按配置顺序补齐; + - ``current`` 不在配置内(自定义/未知名)→ 仍排第一,再补齐全部配置模型; + - 总开关关闭 → 只返回单个候选(``current`` 或默认模型),等同旧行为; + - 拿不到任何配置模型 → 返回 ``[current]``(current 为空则空列表,调用方自行退化为不换模型)。 + """ + all_names = _config_model_names() + cur = (current or "").strip() + ordered: list[str] = [] + if cur: + ordered.append(cur) + for n in all_names: + if n and n not in ordered: + ordered.append(n) + if not ordered: + return [cur] if cur else [] + if not model_fallback_enabled(): + return ordered[:1] + cap = _max_attempts() + if cap > 0: + ordered = ordered[:cap] + return ordered diff --git a/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py b/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py index c6fb9aea..cfd07c0f 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py +++ b/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py @@ -31,12 +31,12 @@ from fastapi.responses import StreamingResponse from langchain_core.messages import HumanMessage, SystemMessage from pydantic import BaseModel, Field -from deerflow.agents.roundtable_orchestrator.extraction_spec import build_structure_directive - +from app.gateway.roundtable_model_fallback import fallback_model_chain, reason_text from app.gateway.routers._roundtable_seed import ( STRUCTURE_AGENT_ID, ensure_roundtable_functional_agents, ) +from deerflow.agents.roundtable_orchestrator.extraction_spec import build_structure_directive from deerflow.config import get_app_config from deerflow.config.agents_config import load_agent_config, load_agent_soul from deerflow.models.factory import create_chat_model @@ -596,28 +596,60 @@ async def extract_flow_stream(request: Request, payload: FlowExtractRequest = Bo return # ① 抽取 / 编辑(正文 + 思考(reasoning) 拼成 `…正文` 增量逐帧吐,前端实时显示) + # 模型自动容错:结构化抽取非常重要——某模型 astream 抛错 / 产出空时,按 config.yaml + # models[] 顺序切到下一个模型重抽,并发 model_switch 帧让前端把抽取面板已累计的半截内容重置。 round1_messages = _edit_messages(payload) if is_edit else _extract_messages(payload) - model = create_chat_model(name=resolved_model, thinking_enabled=True) + chain = fallback_model_chain(resolved_model) or [resolved_model] extracted = "" - think_open = False - async for chunk in model.astream(round1_messages): - c, r = _chunk_parts(chunk) - out = "" - if r: - if not think_open: - out += "" - think_open = True - out += r - if c: + used_model = chain[0] + prev_model: str | None = None + prev_reason: str | None = None + for idx, candidate in enumerate(chain): + if idx > 0: + yield _frame( + { + "event": "model_switch", + "agent_name": "结构化抽取", + "failed_model": prev_model or "", + "next_model": candidate or "", + "reason": reason_text(prev_reason), + } + ) + used_model = candidate + extracted = "" + think_open = False + try: + model = create_chat_model(name=candidate, thinking_enabled=True) + async for chunk in model.astream(round1_messages): + c, r = _chunk_parts(chunk) + out = "" + if r: + if not think_open: + out += "" + think_open = True + out += r + if c: + if think_open: + out += "" + think_open = False + out += c + extracted += c + if out: + yield _frame({"event": "extract_delta", "content": out}) if think_open: - out += "" - think_open = False - out += c - extracted += c - if out: - yield _frame({"event": "extract_delta", "content": out}) - if think_open: - yield _frame({"event": "extract_delta", "content": ""}) + yield _frame({"event": "extract_delta", "content": ""}) + except Exception as exc: # noqa: BLE001 — 该模型抽取失败:换下一个模型重抽 + prev_model, prev_reason = candidate, "exception" + logger.warning("flow-extract round1 model %s failed: %s", candidate, exc) + if idx + 1 < len(chain): + continue + raise + if extracted.strip(): + break + # 产出为空:换下一个模型(已是最后一个则继续走后续解析兜底)。 + prev_model, prev_reason = candidate, "empty" + if idx + 1 >= len(chain): + break yield _frame({"event": "extract_done"}) # ② 校验 / 修复(同样流式 + 思考实时显示) @@ -625,7 +657,7 @@ async def extract_flow_stream(request: Request, payload: FlowExtractRequest = Bo validated = "" graph: dict[str, Any] | None = None try: - vmodel = create_chat_model(name=resolved_model, thinking_enabled=True) + vmodel = create_chat_model(name=used_model, thinking_enabled=True) think_open = False async for chunk in vmodel.astream(_validate_messages(payload, extracted)): c, r = _chunk_parts(chunk) diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py index 7457d9af..83046e24 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py +++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py @@ -63,6 +63,7 @@ from fastapi.responses import StreamingResponse from langgraph.checkpoint.base import empty_checkpoint from app.gateway.roundtable_diag import record_foreground_diag +from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy from deerflow.agents.roundtable_orchestrator.extraction_spec import build_summary_directive @@ -1547,21 +1548,31 @@ async def _leader_run( if files: _read_tools = {"read_file", "ls", "grep", "view_image"} leader_excluded = [t for t in leader_excluded if t not in _read_tools] - body = _run_payload( - agent_name=agent_name, - model_name=model_name, - thread_id=thread_id, - # 注入了席位清单的输入(leader_input);seat_roster 为空时等同 new_message。 - new_message=leader_input, - excluded_tools=leader_excluded, - skill_stop_names=effective_stop_names, - thinking_enabled=policy.thinking_enabled, - reasoning_effort=policy.reasoning_effort, - # 人工干预/恢复会重发 leader run 到同一总控 thread;抢占掉没结束的旧 run,避免 409。 - multitask_strategy="rollback", - # 用户随干预上传的参考文件 → 总控可用 read_file 等工具读取后再派活。 - files=files, - ) + + # 模型自动容错:按 config.yaml models[] 顺序的候选链。当前模型出问题(兜底错误文案 / + # 既无正文又无派活)时切到下一个模型重跑(见循环内检测)。``_mk_body`` 用候选模型重建 body。 + model_chain = fallback_model_chain(model_name) or [model_name] + model_idx = 0 + current_model = model_chain[0] + + def _mk_body(m: str) -> Any: + return _run_payload( + agent_name=agent_name, + model_name=m, + thread_id=thread_id, + # 注入了席位清单的输入(leader_input);seat_roster 为空时等同 new_message。 + new_message=leader_input, + excluded_tools=leader_excluded, + skill_stop_names=effective_stop_names, + thinking_enabled=policy.thinking_enabled, + reasoning_effort=policy.reasoning_effort, + # 人工干预/恢复会重发 leader run 到同一总控 thread;抢占掉没结束的旧 run,避免 409。 + multitask_strategy="rollback", + # 用户随干预上传的参考文件 → 总控可用 read_file 等工具读取后再派活。 + files=files, + ) + + body = _mk_body(current_model) t_payload_built = time.perf_counter() logger.info( "[leader-timing] %s payload_built=%.3fs (broadcast running in background)", @@ -1645,6 +1656,28 @@ async def _leader_run( leader_text = _flatten_content(dispatch_msg.get("content", "")) tool_calls = dispatch_msg.get("tool_calls") or [] + # 模型自动容错:本轮总控产出是「中间件兜底错误文案」或「既无正文又无派活」(模型没干活) + # → 若还有备选模型,切到下一个重跑(不消耗派活纠正预算)。先发 model_switch 帧让前端把 + # 这条失败气泡重置,再用新模型重新流式。最后一个模型仍失败则照常往下走(错误文案当共识展示)。 + fail_reason = is_llm_error_message(dispatch_msg) + if fail_reason is None and not (leader_text or "").strip() and not tool_calls: + fail_reason = "empty" + if fail_reason and model_idx + 1 < len(model_chain): + model_idx += 1 + next_model = model_chain[model_idx] + logger.warning("[leader-fallback] %s 模型 %s 失败(%s),切换到 %s", agent_name, current_model, fail_reason, next_model) + yield { + "status": "model_switch", + "role": "leader", + "agent_name": agent_name, + "failed_model": current_model, + "next_model": next_model, + "reason": reason_text(fail_reason), + } + current_model = next_model + body = _mk_body(current_model) + continue + # 澄清分支:与 intent.py 相同字段;status 为字符串 "clarification"(非数组), # 前端据此暂停 runOrchestration 循环,展示 Step 2 澄清条。 clarification_call = next( @@ -1665,12 +1698,13 @@ async def _leader_run( "allow_custom": bool(cargs.get("allow_custom", True)), "allow_multiple": resolve_allow_multiple(cargs), "agent_name": agent_name, + "model_used": current_model, } return if not tool_calls: # 空 status 数组 = 总控不再派活,前端视为共识(hasConsensus)。 - yield {"status": [], "content": leader_text, "agent_name": agent_name} + yield {"status": [], "content": leader_text, "agent_name": agent_name, "model_used": current_model} return # 派活前按真实 roster 校验:合法保留、非法丢弃(只丢非法,不殃及同轮合法席位)。 @@ -1706,7 +1740,7 @@ async def _leader_run( # 关键路径优先:**先 yield 派活帧**,让前端立刻启动席位 special run;派活广播(告知 # 其它席位"总控给 X 派了什么活")改成 detached 后台任务,不再 await 阻塞 —— 否则广播 # 读写各席位巨大 thread 状态会卡几十秒,白白拖慢席位启动(见 _spawn_detached_broadcast)。 - yield {"status": dispatched, "content": leader_text, "agent_name": agent_name} + yield {"status": dispatched, "content": leader_text, "agent_name": agent_name, "model_used": current_model} # DAG 派活轮(suppress_pre_broadcast)同样**不**把「总控给 X 派活」广播给兄弟席位 —— # 否则弱模型从历史里看到派活语句又会模仿派活(席位任务由前端 pickDagSeatTask 干净下发, # 跨席位承上靠各 special run 的"完成交付"后广播,不需要这条派活广播)。recommend/chain 照旧。 @@ -1805,95 +1839,128 @@ async def _special_run( directive = build_summary_directive(business_code) if directive: seat_message = f"{new_message}\n\n---\n\n{directive}" - body = _run_payload( - agent_name=agent_name, - model_name=model_name, - thread_id=thread_id, - new_message=seat_message, - excluded_tools=seat_excluded, - skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names), - # 席位执行模式覆盖 policy 的 thinking/reasoning(None → 回退 policy 默认); - # subagent 仅 ultra 档置真。工具裁剪(excluded_tools)仍完全由 policy 决定。 - thinking_enabled=(policy.thinking_enabled if thinking_enabled is None else thinking_enabled), - reasoning_effort=(policy.reasoning_effort if reasoning_effort is None else reasoning_effort), - subagent_enabled=subagent_enabled, - # ultra 席位才会派子智能体 → 才需要 custom 流回传子任务实时进度(task_running 等)。 - stream_custom=subagent_enabled, - # 干预/重新派活会把同一席位 thread 再跑一次;抢占掉没结束的旧 run,避免 409。 - multitask_strategy="rollback", - # 前序席位完整交付 → run context,供 read_peer_delivery 工具按需取全文(§12.9)。 - extra_context=( - {"roundtable_peer_deliveries": peer_deliveries} if peer_deliveries else None - ), - # 用户随干预上传的参考文件 → 席位可用 read_file 等工具读取。 - files=files, - ) + # 模型自动容错:按 config.yaml models[] 顺序的候选链。某模型出问题(兜底错误文案; + # 真正的研讨席位还含「交付为空」)→ 发 model_switch 帧让前端重置该席位气泡,再用下一个 + # 模型重跑。Step3 单例(report/summary/dashboard/ingest)的产物是落盘文件,正文可能很短, + # 故**不**把「正文为空」当失败(与后台 run_report/summary 的 validate=None 一致),只在真正 + # 的模型错误时换模型。 + model_chain = fallback_model_chain(model_name) or [model_name] + empty_is_failure = role in ("seat", "seat_parallel") + + def _mk_body(m: str) -> Any: + return _run_payload( + agent_name=agent_name, + model_name=m, + thread_id=thread_id, + new_message=seat_message, + excluded_tools=seat_excluded, + skill_stop_names=merge_skill_stop_names(policy.skill_stop_names, skill_stop_names), + # 席位执行模式覆盖 policy 的 thinking/reasoning(None → 回退 policy 默认); + # subagent 仅 ultra 档置真。工具裁剪(excluded_tools)仍完全由 policy 决定。 + thinking_enabled=(policy.thinking_enabled if thinking_enabled is None else thinking_enabled), + reasoning_effort=(policy.reasoning_effort if reasoning_effort is None else reasoning_effort), + subagent_enabled=subagent_enabled, + # ultra 席位才会派子智能体 → 才需要 custom 流回传子任务实时进度(task_running 等)。 + stream_custom=subagent_enabled, + # 干预/重新派活会把同一席位 thread 再跑一次;抢占掉没结束的旧 run,避免 409。 + multitask_strategy="rollback", + # 前序席位完整交付 → run context,供 read_peer_delivery 工具按需取全文(§12.9)。 + extra_context=( + {"roundtable_peer_deliveries": peer_deliveries} if peer_deliveries else None + ), + # 用户随干预上传的参考文件 → 席位可用 read_file 等工具读取。 + files=files, + ) + t_payload_built = time.perf_counter() logger.info( "[special-timing] %s payload_built=%.3fs", agent_name, t_payload_built - t0, ) - - captured: list[str] = [] - first_line_ts: float | None = None - line_count = 0 - async for line, done in _stream_upstream(client, base, headers, thread_id, body): - if line is not None: - if first_line_ts is None: - first_line_ts = time.perf_counter() - logger.info( - "[special-timing] %s first_upstream_line=%.3fs since_entry=%.3fs", - agent_name, - first_line_ts - t_payload_built, - first_line_ts - t0, - ) - line_count += 1 - yield line - elif done is not None: - captured = done - t_stream_done = time.perf_counter() - logger.info( - "[special-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs", - agent_name, - t_stream_done - (first_line_ts or t_payload_built), - line_count, - t_stream_done - t0, - ) - - messages = _parse_last_messages(captured) - if messages is None: - # 与 _leader_run 同理:席位 run 被抢占/断开导致流截断,不抛 502,静默收尾。 - # WARNING 级:正常是 rollback 抢占;否则查上游 run 的服务端报错(如 DB 连接断)。 - logger.warning( - "[special-timing] %s upstream truncated (no values frame) — ending quietly " - "(expected on rollback preemption; otherwise check the upstream run's server-side error)", - agent_name, - ) - return - # Sub-agent's final deliverable is the latest AIMessage's text content. - last_msg: dict[str, Any] = {} - for msg in reversed(messages): - if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"): - last_msg = msg - break - content = _flatten_content(last_msg.get("content", "")) # 优先用 display_name(中文友好,例如"情报收集")提示 leader,id 放括号备查; # display_name 缺失时退化为仅 agent_name —— 与旧行为一致。 speaker_label = f"{display_name}({agent_name})" if display_name else agent_name - _prefix = f"子智能体{speaker_label}完成{new_message}工作,交付内容为:" - # 广播给**所有**席位(含总控)的都只是**截断摘要** —— 总控 thread 不再囤积各席位全文, - # 每轮 leader 的上下文因此保持小、响应快。总控做最终综合需要全文时,由 _leader_run 在 - # 「综合轮」按需现读各席位 thread 注入(见 _build_full_deliverables_block),不靠历史囤积。 - brief_msg = _truncate_broadcast_content(_prefix, content) - # 先 yield 交付结果让前端立即收口本席位;后广播改 detached,不再 await 阻塞 —— - # 否则它读写各席位大状态会让 SSE 多挂十几秒,把下一轮 leader 的启动也一起拖慢。 - yield { - "status": f"子智能体{speaker_label}完成{new_message}工作", - "content": content, - "agent_name": agent_name, - } - _spawn_detached_broadcast(base, headers, brief_msg, agent_threads, [agent_name]) + + for idx, candidate in enumerate(model_chain): + body = _mk_body(candidate) + captured: list[str] = [] + first_line_ts: float | None = None + line_count = 0 + async for line, done in _stream_upstream(client, base, headers, thread_id, body): + if line is not None: + if first_line_ts is None: + first_line_ts = time.perf_counter() + logger.info( + "[special-timing] %s first_upstream_line=%.3fs since_entry=%.3fs", + agent_name, + first_line_ts - t_payload_built, + first_line_ts - t0, + ) + line_count += 1 + yield line + elif done is not None: + captured = done + t_stream_done = time.perf_counter() + logger.info( + "[special-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs", + agent_name, + t_stream_done - (first_line_ts or t_payload_built), + line_count, + t_stream_done - t0, + ) + + messages = _parse_last_messages(captured) + if messages is None: + # 与 _leader_run 同理:席位 run 被抢占/断开导致流截断,不抛 502,静默收尾。 + # 这不是模型错误(多半 rollback 抢占),不触发换模型。 + logger.warning( + "[special-timing] %s upstream truncated (no values frame) — ending quietly " + "(expected on rollback preemption; otherwise check the upstream run's server-side error)", + agent_name, + ) + return + # Sub-agent's final deliverable is the latest AIMessage's text content. + last_msg: dict[str, Any] = {} + for msg in reversed(messages): + if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"): + last_msg = msg + break + content = _flatten_content(last_msg.get("content", "")) + + # 模型失败检测:兜底错误文案;研讨席位再加「交付为空」。还有备选模型则换。 + fail_reason = is_llm_error_message(last_msg) + if fail_reason is None and empty_is_failure and not content.strip(): + fail_reason = "empty" + if fail_reason and idx + 1 < len(model_chain): + next_model = model_chain[idx + 1] + logger.warning("[special-fallback] %s 模型 %s 失败(%s),切换到 %s", agent_name, candidate, fail_reason, next_model) + yield { + "status": "model_switch", + "role": role, + "agent_name": agent_name, + "failed_model": candidate, + "next_model": next_model, + "reason": reason_text(fail_reason), + } + continue + + # 成功(或已用尽备选):发交付终态帧 + detached 后广播。 + _prefix = f"子智能体{speaker_label}完成{new_message}工作,交付内容为:" + # 广播给**所有**席位(含总控)的都只是**截断摘要** —— 总控 thread 不再囤积各席位全文, + # 每轮 leader 的上下文因此保持小、响应快。总控做最终综合需要全文时,由 _leader_run 在 + # 「综合轮」按需现读各席位 thread 注入(见 _build_full_deliverables_block),不靠历史囤积。 + brief_msg = _truncate_broadcast_content(_prefix, content) + # 先 yield 交付结果让前端立即收口本席位;后广播改 detached,不再 await 阻塞 —— + # 否则它读写各席位大状态会让 SSE 多挂十几秒,把下一轮 leader 的启动也一起拖慢。 + yield { + "status": f"子智能体{speaker_label}完成{new_message}工作", + "content": content, + "agent_name": agent_name, + "model_used": candidate, + } + _spawn_detached_broadcast(base, headers, brief_msg, agent_threads, [agent_name]) + return # --------------------------------------------------------------------------- diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py index 7d641553..2dcc8200 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py @@ -24,6 +24,63 @@ from deerflow.config.app_config import AppConfig logger = logging.getLogger(__name__) +# 兜底/熔断的 AIMessage 会带上 ``additional_kwargs[LLM_ERROR_MARKER] = reason``, +# 让下游(圆桌会商的单轮执行器等)能**可靠识别**「这条 AI 回复其实是模型调用失败的 +# 兜底文案」而不是真实交付——据此触发自动换模型。``reason`` ∈ {quota, auth, busy, +# transient, upstream, generic, circuit}。注意:model_dump() 会保留 additional_kwargs, +# 故经 checkpointer 序列化后仍在;万一被剥离,``is_llm_error_message`` 还会回退按已知 +# 兜底文案精确匹配。 +LLM_ERROR_MARKER = "deerflow_llm_error" + +# 各 reason 对应的用户可见兜底文案(单一数据源:构造与识别都用它)。 +_USER_MESSAGES: dict[str, str] = { + "quota": "大模型因账户额度不足、账单异常或使用受限而拒绝请求,请检查模型账户后再试。", + "auth": "大模型因鉴权失败或无访问权限而拒绝请求,请检查模型的 API Key 与权限配置后再试。", + "busy": "大模型因服务繁忙或网络异常,多次重试后仍未恢复,请稍候再继续。", + "transient": "大模型因服务繁忙或网络异常,多次重试后仍未恢复,请稍候再继续。", + "upstream": "大模型当前响应异常(上游模型或网关临时故障),多次重试后仍未恢复,请稍后再试或更换模型。", + "generic": "大模型本次调用未能成功完成,请稍后再试或更换模型。", +} +_CIRCUIT_MESSAGE = "大模型因连续多次调用失败已被熔断保护,请稍候再试。" +# 全部兜底文案集合,供 ``is_llm_error_message`` 在 additional_kwargs 被剥离时按正文回退识别。 +_ALL_ERROR_MESSAGES: frozenset[str] = frozenset({*_USER_MESSAGES.values(), _CIRCUIT_MESSAGE}) + + +def is_llm_error_message(message: Any) -> str | None: + """判断一条 AI 消息是否为「模型调用失败的兜底文案」,是则返回 reason,否则 None。 + + 入参可为 LangChain ``BaseMessage``,或 checkpointer 序列化后的 dict(两种都支持)。 + 优先看 ``additional_kwargs[LLM_ERROR_MARKER]``(最可靠);万一被序列化剥离,再回退 + 按正文是否精确等于已知兜底文案识别(容忍前后空白)。 + """ + if message is None: + return None + # additional_kwargs(dict 或对象属性两种形态)。 + extra: Any = None + if isinstance(message, dict): + extra = message.get("additional_kwargs") + else: + extra = getattr(message, "additional_kwargs", None) + if isinstance(extra, dict): + reason = extra.get(LLM_ERROR_MARKER) + if reason: + return str(reason) + # 正文回退匹配。 + if isinstance(message, dict): + content = message.get("content") + else: + content = getattr(message, "content", None) + text = "" + if isinstance(content, str): + text = content + elif isinstance(content, list): + text = "\n".join(p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text") + text = text.strip() + if text and text in _ALL_ERROR_MESSAGES: + return "generic" + return None + + _RETRIABLE_STATUS_CODES = {408, 409, 425, 429, 500, 502, 503, 504} _BUSY_PATTERNS = ( "server busy", @@ -201,17 +258,9 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): return f"大模型因{reason_text}调用失败,正在第 {attempt}/{self.retry_max_attempts} 次重试,{seconds} 秒后继续…" def _build_circuit_breaker_message(self) -> str: - return "大模型因连续多次调用失败已被熔断保护,请稍候再试。" + return _CIRCUIT_MESSAGE def _build_user_message(self, reason: str) -> str: - if reason == "quota": - return "大模型因账户额度不足、账单异常或使用受限而拒绝请求,请检查模型账户后再试。" - if reason == "auth": - return "大模型因鉴权失败或无访问权限而拒绝请求,请检查模型的 API Key 与权限配置后再试。" - if reason in {"busy", "transient"}: - return "大模型因服务繁忙或网络异常,多次重试后仍未恢复,请稍候再继续。" - if reason == "upstream": - return "大模型当前响应异常(上游模型或网关临时故障),多次重试后仍未恢复,请稍后再试或更换模型。" # NOTE: We intentionally do NOT embed the raw exception detail into the # user-facing message. Raw `Error code: 400 - {...}` strings — especially # Python repr-style dict literals — get persisted as AI messages and then @@ -219,7 +268,11 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): # braces can corrupt the upstream JSON payload and cause deterministic # JSONDecodeError loops. Keep the user message clean; full detail is # already in the backend log via `logger.warning(..., exc_info=exc)`. - return "大模型本次调用未能成功完成,请稍后再试或更换模型。" + return _USER_MESSAGES.get(reason, _USER_MESSAGES["generic"]) + + def _error_message(self, content: str, reason: str) -> AIMessage: + """构造带 ``LLM_ERROR_MARKER`` 标记的兜底 AIMessage(供下游识别+换模型)。""" + return AIMessage(content=content, additional_kwargs={LLM_ERROR_MARKER: reason}) def _emit_retry_event(self, attempt: int, wait_ms: int, reason: str) -> None: try: @@ -246,7 +299,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): handler: Callable[[ModelRequest], ModelResponse], ) -> ModelCallResult: if self._check_circuit(): - return AIMessage(content=self._build_circuit_breaker_message()) + return self._error_message(self._build_circuit_breaker_message(), "circuit") attempt = 1 while True: @@ -283,7 +336,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): ) if retriable: self._record_failure() - return AIMessage(content=self._build_user_message(reason)) + return self._error_message(self._build_user_message(reason), reason) @override async def awrap_model_call( @@ -292,7 +345,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): handler: Callable[[ModelRequest], Awaitable[ModelResponse]], ) -> ModelCallResult: if self._check_circuit(): - return AIMessage(content=self._build_circuit_breaker_message()) + return self._error_message(self._build_circuit_breaker_message(), "circuit") attempt = 1 while True: @@ -329,7 +382,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): ) if retriable: self._record_failure() - return AIMessage(content=self._build_user_message(reason)) + return self._error_message(self._build_user_message(reason), reason) def _matches_any(detail: str, patterns: tuple[str, ...]) -> bool: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py index fe36c3b0..81b3f6a4 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py @@ -202,6 +202,26 @@ _SEAT_NO_ASK = ( ) +async def _emit_switches( + progress: JobProgressWriter, + switches: list, + *, + agent_name: str, + cycle: int = 0, + agent_id: str | None = None, +) -> None: + """把单轮执行器带回的模型容错记录逐条插进研讨时间线(系统提示气泡)。""" + for sw in switches or []: + await progress.model_switch( + agent_name=agent_name, + failed_model=getattr(sw, "failed_model", ""), + next_model=getattr(sw, "next_model", ""), + reason_label=getattr(sw, "reason_label", "调用失败"), + cycle=cycle, + agent_id=agent_id, + ) + + def _deliveries_block(deliveries: dict[str, str], name_by_id: dict[str, str]) -> str: parts: list[str] = [] for aid, text in deliveries.items(): @@ -292,6 +312,7 @@ async def _run_gather_phase( logger.warning("gather phase seat %s failed", seat.agent_id, exc_info=True) await progress.seat_done(seat.agent_id, 0) return + await _emit_switches(progress, seat_turn.switches, agent_name=name_by_id.get(seat.agent_id, seat.agent_id), agent_id=seat.agent_id) text = (seat_turn.final_text or "").strip() if text: gathered[seat.agent_id] = text @@ -335,6 +356,7 @@ async def _run_recommend( message=message, model=model, ) + await _emit_switches(progress, turn.switches, agent_name="总控", cycle=cycle) if turn.final_text: last_leader = turn.final_text await progress.add_leader_dialogue( @@ -357,18 +379,32 @@ async def _run_recommend( batch: dict[str, str] = {} async def _run_one(agent_id: str, task: str, cyc: int = cycle) -> None: + name = name_by_id.get(agent_id, agent_id) await progress.seat_active(agent_id) seat_msg = _seat_message(intent_text, prior_snapshot, name_by_id, task, gathered=gathered) - seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model) + try: + seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model) + except Exception: # noqa: BLE001 — 单席位彻底失败不拖垮整批;置空交付 + 记诊断 + logger.warning("recommend seat %s failed", agent_id, exc_info=True) + await progress.diag( + stage="step2_seat", level="error", event="seat_failed", + message=f"席位「{name}」本轮执行异常(已跳过,不影响其它席位)", + agent_id=agent_id, agent_name=name, cycle=cyc, + ) + batch[agent_id] = "" + await progress.add_seat_dialogue(agent_id, name, "", cyc, thread_id=thread_ids.get(agent_id)) + await progress.seat_done(agent_id, cyc) + return + await _emit_switches(progress, seat.switches, agent_name=name, cycle=cyc, agent_id=agent_id) batch[agent_id] = seat.final_text if not (seat.final_text or "").strip(): await progress.diag( stage="step2_seat", level="warning", event="seat_empty_delivery", - message=f"席位「{name_by_id.get(agent_id, agent_id)}」本轮交付为空(可能模型未产出或被截断)", - agent_id=agent_id, agent_name=name_by_id.get(agent_id, agent_id), cycle=cyc, + message=f"席位「{name}」本轮交付为空(可能模型未产出或被截断)", + agent_id=agent_id, agent_name=name, cycle=cyc, ) await progress.add_seat_dialogue( - agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cyc, + agent_id, name, seat.final_text, cyc, tool_calls=seat.tool_calls, thread_id=thread_ids.get(agent_id), ) await progress.seat_done(agent_id, cyc) @@ -426,6 +462,7 @@ async def _run_chain( message=leader_task, model=model, ) + await _emit_switches(progress, turn.switches, agent_name="总控", cycle=i + 1) if turn.final_text: last_leader = turn.final_text await progress.add_leader_dialogue( @@ -440,7 +477,22 @@ async def _run_chain( # chain 严格串行(seat B 看 seat A 交付),但有 gathered 时各席位已带共享资料,不必各自从零检索 → 提速。 await progress.seat_active(seat.agent_id) seat_msg = _seat_message(intent_text, deliveries, name_by_id, task, gathered=gathered) - seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model) + try: + seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model) + except Exception: # noqa: BLE001 — 单席位失败不中断整链,置空交付后继续后续席位 + logger.warning("chain seat %s failed", seat.agent_id, exc_info=True) + await progress.diag( + stage="step2_seat", level="error", event="seat_failed", + message=f"席位「{seat.name}」本轮执行异常(已跳过,链条继续)", + agent_id=seat.agent_id, agent_name=seat.name, cycle=i + 1, + ) + deliveries[seat.agent_id] = "" + await progress.add_seat_dialogue(seat.agent_id, seat.name, "", i + 1, thread_id=thread_ids.get(seat.agent_id)) + await progress.seat_done(seat.agent_id, i + 1) + if total > 0: + await progress.set_consensus_pct(round((i + 1) / total * 90)) + continue + await _emit_switches(progress, seat_turn.switches, agent_name=seat.name, cycle=i + 1, agent_id=seat.agent_id) deliveries[seat.agent_id] = seat_turn.final_text if not (seat_turn.final_text or "").strip(): await progress.diag( @@ -468,6 +520,7 @@ async def _run_chain( message=closeout_msg, model=model, ) + await _emit_switches(progress, closeout.switches, agent_name="总控", cycle=total + 1) if closeout.final_text: last_leader = closeout.final_text await progress.add_leader_dialogue( @@ -532,6 +585,7 @@ async def _run_dag( message=leader_task, model=model, ) + await _emit_switches(progress, turn.switches, agent_name="总控", cycle=i + 1) if turn.final_text: last_leader = turn.final_text await progress.add_leader_dialogue( @@ -545,22 +599,35 @@ async def _run_dag( for seat in stage_seats: await progress.seat_active(seat.agent_id) - async def _run_one(seat: SeatRef, _gathered: dict[str, str] | None = gathered) -> None: + async def _run_one(seat: SeatRef, _gathered: dict[str, str] | None = gathered, _cyc: int = i + 1) -> None: task = pick_dag_seat_task(turn.dispatched, seat.agent_id, seat.name, goal=stage.goal) seat_msg = _seat_message(intent_text, deliveries, name_by_id, task, gathered=_gathered) - seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model) + try: + seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model) + except Exception: # noqa: BLE001 — 单席位失败不拖垮整个 stage 并行批 + logger.warning("dag seat %s failed", seat.agent_id, exc_info=True) + await progress.diag( + stage="step2_seat", level="error", event="seat_failed", + message=f"席位「{seat.name}」本轮执行异常(已跳过,不影响同批其它席位)", + agent_id=seat.agent_id, agent_name=seat.name, cycle=_cyc, + ) + deliveries[seat.agent_id] = "" + await progress.add_seat_dialogue(seat.agent_id, seat.name, "", _cyc, thread_id=thread_ids.get(seat.agent_id)) + await progress.seat_done(seat.agent_id, _cyc) + return + await _emit_switches(progress, seat_turn.switches, agent_name=seat.name, cycle=_cyc, agent_id=seat.agent_id) deliveries[seat.agent_id] = seat_turn.final_text if not (seat_turn.final_text or "").strip(): await progress.diag( stage="step2_seat", level="warning", event="seat_empty_delivery", message=f"席位「{seat.name}」本轮交付为空(可能模型未产出或被截断)", - agent_id=seat.agent_id, agent_name=seat.name, cycle=i + 1, + agent_id=seat.agent_id, agent_name=seat.name, cycle=_cyc, ) await progress.add_seat_dialogue( - seat.agent_id, seat.name, seat_turn.final_text, i + 1, + seat.agent_id, seat.name, seat_turn.final_text, _cyc, tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id), ) - await progress.seat_done(seat.agent_id, i + 1) + await progress.seat_done(seat.agent_id, _cyc) await asyncio.gather(*(_run_one(seat) for seat in stage_seats)) @@ -578,6 +645,7 @@ async def _run_dag( message=closeout_msg, model=model, ) + await _emit_switches(progress, closeout.switches, agent_name="总控", cycle=total + 1) if closeout.final_text: last_leader = closeout.final_text await progress.add_leader_dialogue( @@ -627,6 +695,7 @@ async def _run_step3( summary: SummaryResult = await gateway.run_summary( summary_thread_id=summary_thread_id, prompt=summary_prompt, model=model ) + await _emit_switches(progress, getattr(summary, "switches", None), agent_name="方案总结报告") if (summary.md or "").strip() or (summary.closing or "").strip(): step3["summaryReport"] = { "md": summary.md or "", @@ -710,6 +779,7 @@ async def _run_report( report_thread_id = await gateway.create_report_thread() prompt = build_report_prompt(intent_text, agents, deliveries, consensus_text) report: ReportResult = await gateway.run_report(report_thread_id=report_thread_id, prompt=prompt, model=model) + await _emit_switches(progress, getattr(report, "switches", None), agent_name="结果绘制") return { "html": report.html, "generatedAt": report.generated_at or datetime.now(UTC).isoformat(), diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py index 023c41df..b2977fde 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/progress.py @@ -270,3 +270,27 @@ class JobProgressWriter: self._add("seat", agent_id, name, content or "", cycle, tool_calls=tool_calls, thread_id=thread_id) await self._persist() await self._flush_dialogue() + + async def model_switch( + self, + *, + agent_name: str, + failed_model: str, + next_model: str, + reason_label: str, + cycle: int = 0, + agent_id: str | None = None, + ) -> None: + """模型自动容错通知:当前模型出问题、已自动切到下一个模型时调用。 + + 往研讨时间线插一条 ``role="system"`` 系统提示气泡(前端 ``toStep2Dialogue`` 据此渲染 + 为「系统提示」)。对应的 ``model_switch`` **诊断**由网关 + (``InProcessRoundtableGateway._run_turn_with_fallback``)在换模型当下落库(含完整 + detail),此处只负责时间线展示,避免重复落诊断。``reason_label`` 由调用方(app 层) + 预先翻译好(harness 不依赖 app 的文案表)。""" + fm = (failed_model or "").strip() or "当前模型" + nm = (next_model or "").strip() or "下一个模型" + content = f"⚠️ 模型容错:「{agent_name}」所用模型「{fm}」{reason_label},已自动切换到「{nm}」继续。" + self._add("system", agent_id, "系统提示", content, cycle) + await self._persist() + await self._flush_dialogue() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py index 47ac8cf9..95c67566 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py @@ -43,6 +43,20 @@ class SeatRef: name: str +@dataclass +class ModelSwitch: + """一次「模型自动容错」记录:当前模型出问题、已切到下一个模型继续。 + + 由网关(单轮执行器)在换模型时产出,挂在各 Turn/Result 上带回引擎;引擎据此调 + ``JobProgressWriter.model_switch(...)`` 往研讨时间线插系统提示并落诊断。``reason_label`` + 已由 app 层翻译成简短中文(harness 不依赖 app 文案表)。 + """ + + failed_model: str + next_model: str + reason_label: str + + @dataclass class LeaderTurn: """总控一轮执行的归一化结果(由 gateway.run_leader 产出)。""" @@ -55,6 +69,8 @@ class LeaderTurn: # 本轮总控调用的工具(按出现顺序,每项 {name, args, id})——通常是 agent_orchestration # 派活 / ask_clarification。用于在草稿 run 里重建「工具步骤卡」,与前端手动跑一致。 tool_calls: list[dict] = field(default_factory=list) + # 本轮发生的模型自动容错记录(按发生顺序);引擎据此向用户提示「已切换模型」。 + switches: list[ModelSwitch] = field(default_factory=list) @property def is_clarification(self) -> bool: @@ -73,6 +89,8 @@ class SeatTurn: # 本轮席位调用的工具(按出现顺序,每项 {name, args, id})——用于在草稿 run 里 # 重建「工具步骤卡」+ write_file stub(点开产物),与前端手动跑一致。 tool_calls: list[dict] = field(default_factory=list) + # 本轮发生的模型自动容错记录(按发生顺序)。 + switches: list[ModelSwitch] = field(default_factory=list) @dataclass @@ -83,6 +101,7 @@ class ReportResult: summary: str | None = None model: str | None = None generated_at: str | None = None + switches: list[ModelSwitch] = field(default_factory=list) @dataclass @@ -99,6 +118,7 @@ class SummaryResult: model: str | None = None thread_id: str | None = None generated_at: str | None = None + switches: list[ModelSwitch] = field(default_factory=list) # ── 网关协议 ──────────────────────────────────────────────────────────────── diff --git a/offline-backend-20260512/backend/tests/test_roundtable_model_fallback.py b/offline-backend-20260512/backend/tests/test_roundtable_model_fallback.py new file mode 100644 index 00000000..d4207fbc --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_roundtable_model_fallback.py @@ -0,0 +1,207 @@ +"""会商「模型自动容错(换下一个模型)」测试。 + +覆盖: + - 中间件兜底/熔断 AIMessage 带可识别标记 + ``is_llm_error_message`` 检测(标记 + 正文回退); + - ``fallback_model_chain`` 按 config.yaml models[] 顺序排候选(当前优先、去重、开关关闭=单个); + - ``reason_text`` 文案; + - 引擎 Step2 单席位异常隔离(一个席位抛错不拖垮整链/整批,仅置空交付); + - 网关带回的模型容错记录(switches)→ 引擎在时间线插「系统提示」气泡。 + +纯单元/纯逻辑:引擎用 JobProgressWriter(store=None) 干跑 + 假网关,不需要 DB / 真模型。 +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import app.gateway.roundtable_model_fallback as mf +from app.gateway.roundtable_model_fallback import ( + fallback_model_chain, + is_llm_error_message, + reason_text, +) +from deerflow.agents.middlewares.llm_error_handling_middleware import ( + LLM_ERROR_MARKER, + LLMErrorHandlingMiddleware, +) +from deerflow.agents.middlewares.llm_error_handling_middleware import ( + is_llm_error_message as harness_is_llm_error_message, +) +from deerflow.agents.roundtable_orchestrator import ( + JobProgressWriter, + SeatRef, + run_orchestration, +) +from deerflow.agents.roundtable_orchestrator.types import LeaderTurn, ModelSwitch, SeatTurn + +# ── is_llm_error_message ───────────────────────────────────────────────────── + + +def test_is_llm_error_message_marker_dict() -> None: + msg = {"type": "ai", "content": "随便什么", "additional_kwargs": {LLM_ERROR_MARKER: "busy"}} + assert is_llm_error_message(msg) == "busy" + assert harness_is_llm_error_message(msg) == "busy" + + +def test_is_llm_error_message_content_fallback() -> None: + # additional_kwargs 被剥离时,按已知兜底文案正文回退识别。 + msg = {"type": "ai", "content": "大模型本次调用未能成功完成,请稍后再试或更换模型。"} + assert is_llm_error_message(msg) == "generic" + + +def test_is_llm_error_message_normal_is_none() -> None: + assert is_llm_error_message({"type": "ai", "content": "这是正常的席位交付内容"}) is None + assert is_llm_error_message(None) is None + assert is_llm_error_message({"type": "ai", "content": ""}) is None + + +# ── 中间件给兜底 AIMessage 打标记 ──────────────────────────────────────────── + + +def _middleware() -> LLMErrorHandlingMiddleware: + app_config = SimpleNamespace(circuit_breaker=SimpleNamespace(failure_threshold=5, recovery_timeout_sec=30)) + return LLMErrorHandlingMiddleware(app_config=app_config) + + +def test_middleware_marks_error_message() -> None: + mw = _middleware() + + def _boom(_req): + raise RuntimeError("insufficient_quota: 余额不足") # quota → 非重试 → 立即兜底 + + out = mw.wrap_model_call(None, _boom) + assert harness_is_llm_error_message(out) == "quota" + assert out.additional_kwargs.get(LLM_ERROR_MARKER) == "quota" + + +# ── fallback_model_chain ───────────────────────────────────────────────────── + + +def _patch_models(monkeypatch, names: list[str]) -> None: + cfg = SimpleNamespace(models=[SimpleNamespace(name=n) for n in names]) + monkeypatch.setattr(mf, "get_app_config", lambda: cfg) + + +def test_chain_current_first_then_config_order(monkeypatch) -> None: + _patch_models(monkeypatch, ["m1", "m2", "m3"]) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK_MAX", raising=False) + assert fallback_model_chain("m2") == ["m2", "m1", "m3"] + + +def test_chain_empty_current_uses_config_order(monkeypatch) -> None: + _patch_models(monkeypatch, ["m1", "m2", "m3"]) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) + assert fallback_model_chain("") == ["m1", "m2", "m3"] + assert fallback_model_chain(None) == ["m1", "m2", "m3"] + + +def test_chain_unknown_current_kept_first(monkeypatch) -> None: + _patch_models(monkeypatch, ["m1", "m2"]) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) + assert fallback_model_chain("custom") == ["custom", "m1", "m2"] + + +def test_chain_disabled_single(monkeypatch) -> None: + _patch_models(monkeypatch, ["m1", "m2", "m3"]) + monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK", "0") + assert fallback_model_chain("m2") == ["m2"] + + +def test_chain_cap(monkeypatch) -> None: + _patch_models(monkeypatch, ["m1", "m2", "m3", "m4"]) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) + monkeypatch.setenv("ROUNDTABLE_MODEL_FALLBACK_MAX", "2") + assert fallback_model_chain("m1") == ["m1", "m2"] + + +def test_chain_no_config_returns_current(monkeypatch) -> None: + _patch_models(monkeypatch, []) + monkeypatch.delenv("ROUNDTABLE_MODEL_FALLBACK", raising=False) + assert fallback_model_chain("m9") == ["m9"] + assert fallback_model_chain("") == [] + + +def test_reason_text() -> None: + assert reason_text("busy") == "服务繁忙" + assert reason_text("empty") == "未产出有效内容" + assert reason_text("timeout") == "长时间无响应(卡死)" + assert reason_text(None) == "调用失败" + assert reason_text("不存在的码") == "调用失败" + + +# ── 引擎:单席位异常隔离 + 模型容错记录 → 系统提示气泡 ────────────────────────── + +_AGENTS = [SeatRef("seat-a", "情报"), SeatRef("seat-b", "环境"), SeatRef("seat-c", "方案")] + + +def _writer() -> JobProgressWriter: + return JobProgressWriter(None, job_id="job-x", agents=_AGENTS) + + +def _thread_ids() -> dict[str, str]: + return {a.agent_id: f"tid-{a.agent_id}" for a in _AGENTS} | {"roundtable-coordinator": "tid-co"} + + +class _ChainGateway: + """chain 模式假网关:可指定某席位抛异常 / 带回模型容错记录。""" + + def __init__(self, *, fail_agent: str | None = None, switch_agent: str | None = None) -> None: + self.fail_agent = fail_agent + self.switch_agent = switch_agent + self.seen: list[str] = [] + + async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn: + return LeaderTurn(final_text="总控说明", dispatched=[]) + + async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn: + self.seen.append(agent_id) + if agent_id == self.fail_agent: + raise RuntimeError("boom: 模型连接被打满") + switches = [] + if agent_id == self.switch_agent: + switches = [ModelSwitch(failed_model="m1", next_model="m2", reason_label="服务繁忙")] + return SeatTurn(final_text=f"{agent_id} 的交付", switches=switches) + + async def create_report_thread(self) -> str: # pragma: no cover + raise AssertionError("enable_report=False 不应绘制报告") + + +async def _run_chain(gw: _ChainGateway) -> tuple[object, JobProgressWriter]: + writer = _writer() + result = await run_orchestration( + gateway=gw, + progress=writer, + agents=_AGENTS, + thread_ids=_thread_ids(), + coordinator_name="roundtable-coordinator", + model="m1", + seed_message="种子任务", + mode="chain", + enable_report=False, + ) + return result, writer + + +def test_chain_seat_failure_isolated() -> None: + gw = _ChainGateway(fail_agent="seat-b") + result, writer = asyncio.run(_run_chain(gw)) + # 单席位失败不拖垮整链:仍走完三个席位并收口为 done。 + assert result.status == "done" + assert gw.seen == ["seat-a", "seat-b", "seat-c"] + # 失败席位的交付气泡为空,其余正常。 + seat_dlgs = {d["agentId"]: d["content"] for d in writer.dialogues() if d["role"] == "seat"} + assert seat_dlgs["seat-b"] == "" + assert seat_dlgs["seat-a"] == "seat-a 的交付" + assert seat_dlgs["seat-c"] == "seat-c 的交付" + + +def test_model_switch_emits_system_notice() -> None: + gw = _ChainGateway(switch_agent="seat-b") + _result, writer = asyncio.run(_run_chain(gw)) + systems = [d for d in writer.dialogues() if d["role"] == "system"] + assert len(systems) == 1 + assert "模型容错" in systems[0]["content"] + assert "服务繁忙" in systems[0]["content"] + assert "m2" in systems[0]["content"]