feat(roundtable): 第二/三步模型自动容错(出问题自动换下一个模型)

模型出问题时按 config.yaml models[] 顺序自动切到下一个模型继续,并告知用户
「『X』模型异常 → 已切换『Y』继续」。前台实时 + 后台挂起两条路径都覆盖。

- 中间件兜底/熔断 AIMessage 打 LLM_ERROR_MARKER 标记,is_llm_error_message 可识别
- roundtable_model_fallback: fallback_model_chain 候选链 + reason_text + 总开关
  env ROUNDTABLE_MODEL_FALLBACK
- 触发: 异常(start_run失败/idle卡死) / 兜底错误文案 / 交付为空(仅seat/leader)
- 后台 InProcessRoundtableGateway._run_turn_with_fallback 包住每轮 run
- 前台 _special_run/_leader_run 收尾检测失败→model_switch 帧+换模型重流
- 结构化抽取 flow_extract round1 astream 同款换模型循环
- Step2 单席位异常隔离: 单席位失败不拖垮整批/整链/整个作业
- 时间线插「系统提示」气泡 + toast + 诊断 model_switch 事件
- Tests: test_roundtable_model_fallback.py

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
董宏凯andClaude Opus 4.8 committed 2026-06-20 21:37:45 +08:00
1 parent ae2706dad3
commit d383b0e9dd
18 files changed
+1033 -174

No files matched your search

@@ -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<FlowGr
} else if (event === "validate_delta") {
const c = typeof frame.content === "string" ? frame.content : "";
if (c) req.onValidateDelta?.(c);
} else if (event === "model_switch") {
req.onModelSwitch?.({
failedModel: typeof frame.failed_model === "string" ? frame.failed_model : "",
nextModel: typeof frame.next_model === "string" ? frame.next_model : "",
reason: typeof frame.reason === "string" ? frame.reason : "",
});
} else if (event === "final") {
graph = (frame.graph as FlowGraphData) ?? null;
} else if (event === "error") {
@@ -297,6 +297,27 @@ export interface StreamMultiAgentRequest {
name?: string,
args?: Record<string, unknown>,
) => 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<string, unknown>).status === "model_switch"
) {
const obj = parsed as Record<string, unknown>;
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<string, unknown>).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") {
@@ -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);
@@ -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;
@@ -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);
@@ -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");
@@ -66,6 +66,10 @@ function toStep2Dialogue(d: JobDialogue, idxByAgent: Map<string, number>): 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,
@@ -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;
@@ -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
@@ -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
@@ -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
@@ -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) 拼成 `<think>…</think>正文` 增量逐帧吐,前端实时显示)
# 模型自动容错:结构化抽取非常重要——某模型 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>"
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>"
think_open = True
out += r
if c:
if think_open:
out += "</think>"
think_open = False
out += c
extracted += c
if out:
yield _frame({"event": "extract_delta", "content": out})
if think_open:
out += "</think>"
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": "</think>"})
yield _frame({"event": "extract_delta", "content": "</think>"})
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)
@@ -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
# ---------------------------------------------------------------------------
@@ -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:
@@ -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(),
@@ -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()
@@ -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)
# ── 网关协议 ────────────────────────────────────────────────────────────────
@@ -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"]