From b59cbd51d9087fcdf11f8c593e54cfe3f5dceb89 Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Mon, 22 Jun 2026 16:36:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(ai-writing):=20=E5=90=84=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E8=B0=83=E7=94=A8=E8=87=AA=E5=8A=A8=E5=AE=B9?= =?UTF-8?q?=E9=94=99=EF=BC=8C=E5=A4=B1=E8=B4=A5=E6=97=B6=E6=8D=A2=E4=B8=8B?= =?UTF-8?q?=E4=B8=80=E4=B8=AA=E6=A8=A1=E5=9E=8B=E7=BB=A7=E7=BB=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 参考多智能体会商的容错思路：意图解析/大纲/正文/审核四个 LLM 环节失败时，按 config.yaml models 顺序换下一个模型重跑该环节， 并通过 model_switch 事件在前端时间线提示「『X』模型<原因>， 已自动切换到『Y』继续」。正文逐章节、审核逐维度独立切换， 已生成的章节/维度保留；意图解析成功的模型还会写回 state 供下游阶段沿用，避免每阶段重复经历同一次失败。 - nodes/_model_fallback.py: 候选模型链 + 异常分类 + model_switch 事件构造（沿用 ROUNDTABLE_MODEL_FALLBACK[_MAX] env 开关） - intent_parser/writer_outline/writer_draft/editor: 失败循环换模型 - 前端 types/reducer/timeline: 新增 model_switch 事件类型，按 stage 复位对应流式缓冲并以 amber 系统提示渲染 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ai-writing/AIWritingTimeline.tsx | 1 + .../components/ai-writing/TimelineStep.tsx | 8 +- .../open-canvas/hooks/useAIWritingStream.ts | 68 ++++++ .../src/open-canvas/types/ai-writing.ts | 11 + .../ai_writing/nodes/_model_fallback.py | 197 ++++++++++++++++++ .../agents/ai_writing/nodes/editor.py | 72 +++++-- .../agents/ai_writing/nodes/intent_parser.py | 112 ++++++---- .../agents/ai_writing/nodes/writer_draft.py | 149 ++++++++----- .../agents/ai_writing/nodes/writer_outline.py | 124 +++++++---- 9 files changed, 598 insertions(+), 144 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_model_fallback.py diff --git a/frontend-web/src/open-canvas/components/ai-writing/AIWritingTimeline.tsx b/frontend-web/src/open-canvas/components/ai-writing/AIWritingTimeline.tsx index eedf8a69..c86a652d 100644 --- a/frontend-web/src/open-canvas/components/ai-writing/AIWritingTimeline.tsx +++ b/frontend-web/src/open-canvas/components/ai-writing/AIWritingTimeline.tsx @@ -19,6 +19,7 @@ const DISPLAYED_TYPES = new Set([ 'await_user', 'resumed', 'notice', + 'model_switch', 'done', 'error', ]); diff --git a/frontend-web/src/open-canvas/components/ai-writing/TimelineStep.tsx b/frontend-web/src/open-canvas/components/ai-writing/TimelineStep.tsx index b1289ca6..f5ed8327 100644 --- a/frontend-web/src/open-canvas/components/ai-writing/TimelineStep.tsx +++ b/frontend-web/src/open-canvas/components/ai-writing/TimelineStep.tsx @@ -13,6 +13,7 @@ interface TimelineStepProps { function statusOf(event: ProgressEvent, isLast: boolean): StepStatus { if (event.type === 'error') return 'error'; if (event.type === 'notice') return 'warning'; + if (event.type === 'model_switch') return 'warning'; if (event.type === 'await_user') return 'paused'; // step_start 只有在列表末尾（后续事件还未到达）时才显示转圈，否则已有后续事件说明该步骤已完成 if (event.type === 'step_start') return isLast ? 'running' : 'done'; @@ -30,8 +31,11 @@ const AGENT_COLOR: Record = { export function TimelineStep({ event, isLast, children }: TimelineStepProps) { const status = statusOf(event, isLast ?? true); - // notice=非致命告警（如 ES 波动、模型兜底），error=致命错误，两者都展示「标题 + 真实原因」 - const isAlert = event.type === 'error' || event.type === 'notice'; + // notice=非致命告警（如 ES 波动、模型兜底），error=致命错误，model_switch=模型自动容错切换（amber 同 notice） + const isAlert = + event.type === 'error' || + event.type === 'notice' || + event.type === 'model_switch'; const isError = event.type === 'error'; return ( diff --git a/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts b/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts index 991a6941..04607a21 100644 --- a/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts +++ b/frontend-web/src/open-canvas/hooks/useAIWritingStream.ts @@ -388,6 +388,37 @@ function handleEvent(prev: AIWritingState, event: ProgressEvent): AIWritingState case 'notice': return base; + case 'model_switch': { + // 模型自动容错：丢弃当前阶段已累积的流式缓冲，避免上一个失败模型的半截输出 + // 与下一个模型的输出粘在一起；时间线展示侧由 TimelineStep 按 amber 渲染（同 notice）。 + const stage = event.modelSwitch?.stage; + const sectionIndex = event.modelSwitch?.sectionIndex; + const resetByStage: Partial = (() => { + switch (stage) { + case 'intent': + return { intentStreamingText: '', intentThinkingText: '' }; + case 'outline': + return { outlineStreamingText: '', outlineThinkingText: '' }; + case 'draft': { + // 只清掉失败那一节的流式片段，已写好的章节保持不动 + if (typeof sectionIndex === 'number') { + return { + streamingDraftSections: prev.streamingDraftSections.filter( + (s) => s.sectionIndex !== sectionIndex, + ), + }; + } + return { streamingDraftSections: [], streamingDraftTitle: '' }; + } + case 'review': + return { streamingReviewItems: [] }; + default: + return {}; + } + })(); + return { ...base, ...resetByStage }; + } + default: return base; } @@ -506,6 +537,22 @@ export function normalizeProgressDict(parsed: Record, sessionId out.reviewItemIssues = (out.issues as Record[]).map(normalizeReviewIssue); } + // model_switch（updates 通道里的 progress_events 也可能带）：折成 modelSwitch 子对象 + if (out.type === 'model_switch' && !out.modelSwitch) { + out.modelSwitch = { + stage: (out.stage as 'intent' | 'outline' | 'draft' | 'review') ?? 'intent', + failedModel: String(out.failed_model ?? out.failedModel ?? ''), + nextModel: String(out.next_model ?? out.nextModel ?? ''), + reason: String(out.reason ?? ''), + sectionIndex: typeof out.section_index === 'number' + ? (out.section_index as number) + : (typeof out.sectionIndex === 'number' ? (out.sectionIndex as number) : undefined), + sectionTitle: typeof out.section_title === 'string' + ? (out.section_title as string) + : (typeof out.sectionTitle === 'string' ? (out.sectionTitle as string) : undefined), + }; + } + if (!out.message) out.message = ''; return out as unknown as ProgressEvent; } @@ -522,6 +569,7 @@ export function customToProgressEvent(raw: unknown, sessionId: string): Progress 'research_progress', 'outline_chunk', 'draft_chunk', 'review_chunk', 'review_item_done', 'skill_search_step', 'skill_select_chunk', + 'model_switch', ]; if (!allowed.includes(type as ProgressEventType)) return null; @@ -536,10 +584,30 @@ export function customToProgressEvent(raw: unknown, sessionId: string): Progress if (Array.isArray(merged.issues)) { merged.reviewItemIssues = (merged.issues as Record[]).map(normalizeReviewIssue); } + if ('agent_name' in merged && !merged.agentName) merged.agentName = merged.agent_name; // skill_search_step：把字段折成 skillStep 子对象，方便 reducer / UI 取用 if (type === 'skill_search_step') { merged.skillStep = normalizeSkillStep(o); } + // model_switch：把 snake_case 字段折成 modelSwitch 子对象（reducer + TimelineStep 取用） + if (type === 'model_switch') { + merged.modelSwitch = { + stage: (o.stage as 'intent' | 'outline' | 'draft' | 'review') ?? 'intent', + failedModel: String(o.failed_model ?? (o as Record).failedModel ?? ''), + nextModel: String(o.next_model ?? (o as Record).nextModel ?? ''), + reason: String(o.reason ?? ''), + sectionIndex: typeof o.section_index === 'number' + ? (o.section_index as number) + : (typeof (o as Record).sectionIndex === 'number' + ? ((o as Record).sectionIndex as number) + : undefined), + sectionTitle: typeof o.section_title === 'string' + ? (o.section_title as string) + : (typeof (o as Record).sectionTitle === 'string' + ? ((o as Record).sectionTitle as string) + : undefined), + }; + } return merged as unknown as ProgressEvent; } diff --git a/frontend-web/src/open-canvas/types/ai-writing.ts b/frontend-web/src/open-canvas/types/ai-writing.ts index 361cb090..e3f2f58d 100644 --- a/frontend-web/src/open-canvas/types/ai-writing.ts +++ b/frontend-web/src/open-canvas/types/ai-writing.ts @@ -172,6 +172,7 @@ export type ProgressEventType = | "await_user" | "resumed" | "notice" + | "model_switch" | "skill_search_step" | "skill_select_chunk" | "done" @@ -215,6 +216,16 @@ export interface ProgressEvent { blockedSections?: BlockedSection[]; // skill_search_step：技能检索步骤事件（list_skills / select_skills / invoke_skill） skillStep?: SkillSearchStep; + // model_switch：模型自动容错切换（某模型挂掉，换下一个模型继续） + // stage ∈ intent / outline / draft / review，前端 reducer 据此清掉对应阶段的流式缓冲。 + modelSwitch?: { + stage: "intent" | "outline" | "draft" | "review"; + failedModel: string; + nextModel: string; + reason: string; + sectionIndex?: number; + sectionTitle?: string; + }; } // ── 用户干预提交体 ──────────────────────────────────────────────────────────── diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_model_fallback.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_model_fallback.py new file mode 100644 index 00000000..17fb5a8e --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_model_fallback.py @@ -0,0 +1,197 @@ +"""AI 写作模型自动容错（按 ``config.yaml models[]`` 顺序换下一个模型继续）。 + +与多智能体会商 ``app/gateway/roundtable_model_fallback.py`` 同思路，但**只放在 harness 层** +避免 ``app→harness`` 单向依赖被反向打破： + + - ``fallback_chain(current)``：候选模型链 ``[当前, …其余]``（按 config.yaml 顺序，去重保序）； + - ``classify_exception(exc)``：把异常归到一个用户可读的失败原因码； + - ``reason_text(reason)``：原因码 → 简短中文（用于「『X』模型 <原因>，已切换『Y』继续」）； + - ``emit_model_switch_event(writer, ...)``：把 ``model_switch`` 推到 LangGraph custom 通道 + 供前端时间线展示； + - ``build_switch_progress_event(...)``：节点 return 的 ``progress_events`` 里同样塞一份 + （走 updates 通道，前端 reducer 据此刷新缓冲并落到时间线）。 + +总开关 / 上限沿用会商的 env： + - ``ROUNDTABLE_MODEL_FALLBACK``：默认开；``0/false/off/no`` 关闭（链只含当前模型）； + - ``ROUNDTABLE_MODEL_FALLBACK_MAX``：候选数上限（含当前），0/未设 = 不限。 + +放在 ai_writing 节点目录下而非 harness/agents/middlewares：本模块不是中间件， +且仅服务 ai_writing 的几个流式节点；如果未来圆桌也想下沉合并，再统一拎出去。 +""" +from __future__ import annotations + +import logging +import os +from typing import Any, Callable + +logger = logging.getLogger(__name__) + + +_REASON_TEXT: dict[str, str] = { + "quota": "额度不足/账单异常", + "auth": "鉴权失败/无访问权限", + "busy": "服务繁忙", + "transient": "网络/连接异常", + "upstream": "上游响应异常", + "config": "模型未配置或配置异常", + "exception": "调用异常", + "generic": "调用失败", +} + + +def reason_text(reason: str | None) -> str: + """原因码 → 简短中文（未知码 → 「调用失败」）。""" + if not reason: + return "调用失败" + return _REASON_TEXT.get(reason, "调用失败") + + +_QUOTA_PATTERNS = ("insufficient_quota", "quota", "billing", "credit", "payment", + "余额不足", "超出限额", "额度不足", "欠费") +_AUTH_PATTERNS = ("authentication", "unauthorized", "invalid api key", "invalid_api_key", + "permission", "forbidden", "access denied", "无权", "未授权") +_BUSY_PATTERNS = ("server busy", "temporarily unavailable", "try again later", "please retry", + "please try again", "overloaded", "high demand", "rate limit", + "负载较高", "服务繁忙", "稍后重试", "请稍后重试") +_UPSTREAM_PATTERNS = ("bad_response_status_code", "bad response status", + "expecting ',' delimiter", "expecting value", "expecting property name", + "unexpected character", "unterminated string", "jsondecodeerror") +_TRANSIENT_EXC_NAMES = {"APITimeoutError", "APIConnectionError", "InternalServerError", + "ReadError", "RemoteProtocolError", "TimeoutError"} + + +def classify_exception(exc: BaseException) -> str: + """异常 → 失败原因码（quota/auth/busy/transient/upstream/config/generic/exception）。 + + 与 ``LLMErrorHandlingMiddleware._classify_error`` 同口径，简化版：不区分是否可重试 + （ai_writing 的容错只换模型，不在同模型上重试）。``create_chat_model`` 抛 + ``ValueError("Model X not found in config")`` 这类配置缺失另归 ``config``。 + """ + name = exc.__class__.__name__ + detail = str(exc).lower() + + if isinstance(exc, ValueError) and "not found in config" in detail: + return "config" + if any(p in detail for p in _QUOTA_PATTERNS): + return "quota" + if any(p in detail for p in _AUTH_PATTERNS): + return "auth" + if name in _TRANSIENT_EXC_NAMES: + return "transient" + if any(p in detail for p in _BUSY_PATTERNS): + return "busy" + if any(p in detail for p in _UPSTREAM_PATTERNS): + return "upstream" + return "exception" + + +def _fallback_enabled() -> bool: + raw = (os.getenv("ROUNDTABLE_MODEL_FALLBACK", "1") or "1").strip().lower() + return raw not in {"0", "false", "no", "off"} + + +def _max_attempts() -> int: + 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: + from deerflow.config import get_app_config + cfg = get_app_config() + return [m.name for m in (cfg.models or []) if getattr(m, "name", None)] + except Exception: # noqa: BLE001 — 拿不到配置（测试桩）时退化为空 + logger.debug("ai_writing fallback_chain: get_app_config().models unavailable", exc_info=True) + return [] + + +def fallback_chain(current: str | None) -> list[str]: + """``[当前模型, …config.yaml 其余模型]``（去重保序）。 + + - ``current`` 为空 → 链头落到 ``models[0]``； + - ``current`` 在配置内 → 排第一，余者按配置顺序补齐； + - ``current`` 不在配置内 → 仍排第一（让 ``create_chat_model`` 在第一轮按原行为抛， + 之后再换模型），再补齐全部配置模型； + - 总开关关闭 → 只返回头一个； + - 拿不到配置 → ``[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 _fallback_enabled(): + return ordered[:1] + cap = _max_attempts() + if cap > 0: + ordered = ordered[:cap] + return ordered + + +def build_switch_progress_event( + *, + agent_name: str, + stage: str, + failed_model: str, + next_model: str, + reason: str, + section_index: int | None = None, + section_title: str | None = None, +) -> dict: + """构造一条 ``model_switch`` ``progress_event`` 字典（供节点 return 时塞进 ``progress_events`` 数组）。 + + ``stage`` ∈ {intent, outline, draft, review}：前端 reducer 据此清掉对应阶段的流式缓冲， + 避免半段输出被下一个模型的输出粘在一起。 + """ + fm = (failed_model or "").strip() or "当前模型" + nm = (next_model or "").strip() or "下一个模型" + rt = reason_text(reason) + detail = f"「{agent_name}」所用模型「{fm}」{rt}，已自动切换到「{nm}」继续。" + payload: dict[str, Any] = { + "type": "model_switch", + "agent_name": agent_name, + "title": "模型容错", + "detail": detail, + "message": detail, + "stage": stage, + "failed_model": fm, + "next_model": nm, + "reason": reason, + } + if section_index is not None: + payload["section_index"] = section_index + if section_title: + payload["section_title"] = section_title + return payload + + +def emit_model_switch_event( + writer: Callable[[dict], None] | None, + *, + agent_name: str, + stage: str, + failed_model: str, + next_model: str, + reason: str, + section_index: int | None = None, + section_title: str | None = None, +) -> None: + """把 ``model_switch`` 事件推到 LangGraph custom 通道（前端 useStream onCustomEvent 收）。""" + if writer is None: + return + payload = build_switch_progress_event( + agent_name=agent_name, stage=stage, + failed_model=failed_model, next_model=next_model, reason=reason, + section_index=section_index, section_title=section_title, + ) + try: + writer(payload) + except Exception: # noqa: BLE001 — writer 在无 runtime 时是 no-op，理论不抛 + logger.debug("emit_model_switch_event: writer raised", exc_info=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py index abafe383..191cc602 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py @@ -21,7 +21,13 @@ from ._agent_runner import ( run_writing_agent, use_builtin_agent, ) -from ._utils import get_ai_writing_config, get_writing_model, parse_json, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer +from ._model_fallback import ( + build_switch_progress_event, + classify_exception, + emit_model_switch_event, + fallback_chain, +) +from ._utils import get_ai_writing_config, parse_json, resolve_writing_model_name, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -154,22 +160,17 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: "title": "模型错误", "detail": str(e), "message": "审核失败", }], } + # 旧路径下模型由 fallback_chain 在每维度审查前按候选顺序解析；模型挂掉自动换下一个继续。 + editor_chain: list[str] = [] + current_model_name: str | None = None if not use_agent: - # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError， - # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。 - try: - model = get_writing_model(config, state) - except Exception as e: - logger.error("editor model creation failed: %s", e, exc_info=True) - return { - "status": "error", - "progress_events": [{ - "type": "error", "agent_name": "编辑", - "title": "模型错误", "detail": str(e), "message": "审核失败", - }], - } + initial_name, _ud = resolve_writing_model_name(config, state) + editor_chain = fallback_chain(initial_name) + current_model_name = editor_chain[0] if editor_chain else None + from deerflow.models import create_chat_model results: dict[str, dict] = {} review_errors: list[str] = [] # 分项审查的模型异常，供完成后非致命提示 + switch_notices: list[dict] = [] # 模型容错切换记录，最终塞进 progress_events 走时间线 _CATEGORY_LABEL = {"fact": "事实性", "logic": "逻辑性", "language": "语言文字"} @@ -192,7 +193,47 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: if res.get("_llm_error"): review_errors.append(f"「{_CATEGORY_LABEL.get(category, category)}审查」{res['_llm_error']}") else: - full_text = await _stream_one_review(model, user_prompt, category, writer) + # 旧路径：按 fallback_chain 试候选模型，失败就换下一个继续本维度。 + # 与 writer_draft 同思路：以「上一维度成功使用的模型」为头部。 + dim_chain: list[str] = [] + if current_model_name: + dim_chain.append(current_model_name) + for n in editor_chain: + if n and n not in dim_chain: + dim_chain.append(n) + if not dim_chain: + dim_chain = [None] # type: ignore[list-item] + full_text = "" + dim_succeeded = False + last_dim_exc: BaseException | None = None + for d_idx, candidate in enumerate(dim_chain): + try: + m = create_chat_model(name=candidate) if candidate else create_chat_model() + full_text = await _stream_one_review(m, user_prompt, category, writer) + dim_succeeded = True + current_model_name = candidate or current_model_name + break + except Exception as e: + last_dim_exc = e + reason = classify_exception(e) + logger.error( + "editor review [%s] failed on model=%s (reason=%s): %s", + category, candidate, reason, e, exc_info=True, + ) + next_candidate = dim_chain[d_idx + 1] if d_idx + 1 < len(dim_chain) else None + if next_candidate: + emit_model_switch_event( + writer, agent_name=agent_label, stage="review", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + ) + switch_notices.append(build_switch_progress_event( + agent_name=agent_label, stage="review", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + )) + continue + if not dim_succeeded: + assert last_dim_exc is not None + raise last_dim_exc res = parse_json(full_text, fallback) except Exception as e: logger.warning("editor review [%s] failed: %s", category, e) @@ -252,6 +293,7 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: "review_result": review, "progress_events": [ {"type": "step_start", "agent_name": agent_label, "message": "开始审核"}, + *switch_notices, *notices, { "type": "review_ready", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py index ac94bc91..3e66f7b0 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py @@ -9,10 +9,15 @@ from langchain_core.runnables import RunnableConfig from langchain_core.messages import HumanMessage, SystemMessage from ..state import AIWritingState +from ._model_fallback import ( + build_switch_progress_event, + classify_exception, + emit_model_switch_event, + fallback_chain, +) from ._utils import ( ThinkTagParser, _content_to_text, - get_model, parse_json, reasoning_delta, safe_get_stream_writer, @@ -207,47 +212,74 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d # 流式调用：与 writer_outline / writer_draft 同套机制 —— LLM 逐字吐出 JSON， # 通过 ``intent_chunk`` 推给前端打字机；完成后 ``parse_json`` 解析结构化结果， # 再通过 ``intent_ready`` 推一份结构化字段给前端结果卡片。 + # + # 模型自动容错：失败时按 fallback_chain 顺序换下一个模型重试本环节。每次切换 + # 都通过 ``model_switch`` custom 事件 + ``progress_events`` 同步告诉前端，前端 + # 清掉本环节已累积的 intent_chunk 缓冲，避免半段输出和下个模型的输出拼接错乱。 writer = safe_get_stream_writer() - llm_error: str | None = None + initial_model = (configurable.get("model_name") or "").strip() or None + chain = fallback_chain(initial_model) + progress_events: list[dict] = [] + from deerflow.models import create_chat_model + full_text = "" - parser = ThinkTagParser() - try: - model = get_model(config) - async for chunk in model.astream( - [SystemMessage(content=system), HumanMessage(content=user)] - ): - # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送 - reasoning = reasoning_delta(chunk) - if reasoning: - writer({"type": "intent_chunk", "text": reasoning, "is_thinking": True}) - delta = _content_to_text(chunk.content) - if not delta: + success_model: str | None = None + last_error: BaseException | None = None + for idx, candidate in enumerate(chain or [None]): + full_text = "" # 每次重试都从头累积，避免不同模型的输出粘在一起 + parser = ThinkTagParser() + try: + model = create_chat_model(name=candidate) if candidate else create_chat_model() + async for chunk in model.astream( + [SystemMessage(content=system), HumanMessage(content=user)] + ): + # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送 + reasoning = reasoning_delta(chunk) + if reasoning: + writer({"type": "intent_chunk", "text": reasoning, "is_thinking": True}) + delta = _content_to_text(chunk.content) + if not delta: + continue + for is_thinking, text in parser.feed(delta): + if not is_thinking: + full_text += text + writer({ + "type": "intent_chunk", + "text": text, + "is_thinking": is_thinking, + }) + success_model = candidate or "" + break + except Exception as exc: + last_error = exc + reason = classify_exception(exc) + logger.error( + "intent_parser streaming failed on model=%s (reason=%s): %s", + candidate, reason, exc, exc_info=True, + ) + next_candidate = chain[idx + 1] if idx + 1 < len(chain) else None + if next_candidate: + emit_model_switch_event( + writer, agent_name="系统", stage="intent", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + ) + progress_events.append(build_switch_progress_event( + agent_name="系统", stage="intent", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + )) continue - for is_thinking, text in parser.feed(delta): - if not is_thinking: - full_text += text - writer({ - "type": "intent_chunk", - "text": text, - "is_thinking": is_thinking, - }) - except Exception as exc: - # 模型 / 网络挂掉 —— 不阻断流程，用 fallback 继续。 - logger.error("intent_parser streaming failed: %s", exc, exc_info=True) - llm_error = f"{type(exc).__name__}: {exc}" + # 链尾仍失败：保留 fallback 默认值，向用户报一条非致命 notice，流程继续。 + progress_events.append({ + "type": "notice", + "agent_name": "系统", + "title": "模型错误", + "detail": ( + f"意图解析失败，已按默认配置兜底。原因：{type(exc).__name__}: {exc}" + ), + }) result = parse_json(full_text, fallback_result) if full_text else fallback_result - # 意图解析模型调用失败时以非致命 notice 提示用户（fallback 已兜底，流程仍可继续） - progress_events: list[dict] = [] - if llm_error: - progress_events.append({ - "type": "notice", - "agent_name": "系统", - "title": "模型错误", - "detail": f"意图解析失败，已按默认配置兜底。原因：{llm_error}", - }) - final_article_type, final_target_audience, final_word_count, final_refined_intent = _resolve_intent_fields( state, result, fallback_result, valid_keys, default_key ) @@ -267,9 +299,11 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d # 捕获用户表单选定的主模型进 state（仅本节点能可靠拿到 configurable.model_name — # 它运行在首次 invocation；后续大纲/写作/审核在 resume 后跑，config 已不带它）。 # 大纲/写作/审核经 get_writing_model 严格读 state['model_name']，不回退默认。 - selected_model = (configurable.get("model_name") or "").strip() - if selected_model: - updates["model_name"] = selected_model + # 容错切换后用实际成功的模型作为后续阶段起点，避免每个阶段都重复经历同一次失败。 + persisted_model = (success_model or initial_model or "").strip() + if persisted_model: + updates["model_name"] = persisted_model + _ = last_error # 链尾失败已生成 notice，这里保留变量供调试时断点查看 updates["article_type"] = final_article_type updates["target_audience"] = final_target_audience updates["word_count_target"] = final_word_count diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py index 7b7a082d..5f96f458 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py @@ -21,6 +21,12 @@ from ._agent_runner import ( stream_writing_agent, use_builtin_agent, ) +from ._model_fallback import ( + build_switch_progress_event, + classify_exception, + emit_model_switch_event, + fallback_chain, +) from ._utils import resolve_writing_model_name, _content_to_text, count_words, reasoning_delta, ThinkTagParser, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -234,31 +240,25 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di "title": "模型错误", "detail": str(e), "message": "草稿生成失败", }], } + # 旧路径的模型由 fallback_chain 在每章流式调用前按候选顺序解析（per-section）， + # 不再在循环外预创建：模型名配错 / 模型挂掉时不再硬崩溃，按链顺序换下一个模型重写本章节。 + draft_chain: list[str] = [] if not use_agent: - # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError， - # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。 - try: - model_name, used_default = resolve_writing_model_name(config, state) - if used_default: - model_default_notice = { - "type": "notice", "agent_name": "作家", - "title": "模型提示", "detail": "未检测到所选写作模型，已使用默认模型继续写作。", - } - from deerflow.models import create_chat_model - model = create_chat_model(name=model_name) - except Exception as e: - logger.error("draft model creation failed: %s", e, exc_info=True) - return { - "status": "error", - "progress_events": [{ - "type": "error", "agent_name": "作家", - "title": "模型错误", "detail": str(e), "message": "草稿生成失败", - }], + initial_name, used_default = resolve_writing_model_name(config, state) + if used_default: + model_default_notice = { + "type": "notice", "agent_name": "作家", + "title": "模型提示", "detail": "未检测到所选写作模型，已使用默认模型继续写作。", } + draft_chain = fallback_chain(initial_name) + from deerflow.models import create_chat_model section_titles: list[str] = [] section_bufs: list[str] = [] # 各章节正文（不含 ## 标题），供扩写兜底逐节替换 section_errors: list[str] = [] # 分节写作的模型异常，供完成后非致命提示 blocked_sections: list[dict] = [] # 严格模式下素材不足、无法写作的章节 + switch_notices: list[dict] = [] # 模型容错切换记录，最终塞进 progress_events 走时间线 + # 上一节成功使用的模型 —— 优先尝试，命中则零开销；与会商「当前模型先试」一致。 + current_model_name: str | None = (draft_chain[0] if draft_chain else None) for i, section in enumerate(sections): sec_title = _section_title(i, section) @@ -292,9 +292,12 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di ) section_buf = "" # 只累积正文（非思考）用于拼接 Markdown - parser = ThinkTagParser() + section_succeeded = False try: if use_agent: + parser = ThinkTagParser() # 内置 agent 走自有路径，本变量保留以避免类型告警 + _ = parser + def _on_text(text: str, is_thinking: bool, _i: int = i, _t: str = sec_title) -> None: writer({ "type": "draft_chunk", @@ -312,36 +315,78 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di config=config, on_text=_on_text, ) + section_succeeded = True else: - async for chunk in model.astream([ - SystemMessage(content=DRAFT_SECTION_SYSTEM), - HumanMessage(content=user_msg), - ]): - # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送 - reasoning = reasoning_delta(chunk) - if reasoning: - writer({ - "type": "draft_chunk", - "section_index": i, - "section_title": sec_title, - "text": reasoning, - "is_thinking": True, - "is_section_done": False, - }) - delta = _content_to_text(chunk.content) - if not delta: - continue - for is_thinking, text in parser.feed(delta): - if not is_thinking: - section_buf += text - writer({ - "type": "draft_chunk", - "section_index": i, - "section_title": sec_title, - "text": text, - "is_thinking": is_thinking, - "is_section_done": False, - }) + # 旧路径：按 fallback_chain 试候选模型，失败就换下一个继续本章节。 + # 以「上一节成功使用的模型」为头部 —— 同 config.yaml 顺序的链尾再补全。 + section_chain: list[str] = [] + if current_model_name: + section_chain.append(current_model_name) + for n in draft_chain: + if n and n not in section_chain: + section_chain.append(n) + if not section_chain: + section_chain = [None] # type: ignore[list-item] + last_section_exc: BaseException | None = None + for s_idx, candidate in enumerate(section_chain): + section_buf = "" + parser = ThinkTagParser() + try: + m = create_chat_model(name=candidate) if candidate else create_chat_model() + async for chunk in m.astream([ + SystemMessage(content=DRAFT_SECTION_SYSTEM), + HumanMessage(content=user_msg), + ]): + reasoning = reasoning_delta(chunk) + if reasoning: + writer({ + "type": "draft_chunk", + "section_index": i, + "section_title": sec_title, + "text": reasoning, + "is_thinking": True, + "is_section_done": False, + }) + delta = _content_to_text(chunk.content) + if not delta: + continue + for is_thinking, text in parser.feed(delta): + if not is_thinking: + section_buf += text + writer({ + "type": "draft_chunk", + "section_index": i, + "section_title": sec_title, + "text": text, + "is_thinking": is_thinking, + "is_section_done": False, + }) + section_succeeded = True + current_model_name = candidate or current_model_name + break + except Exception as e: + last_section_exc = e + reason = classify_exception(e) + logger.error( + "draft section [%s] failed on model=%s (reason=%s): %s", + sec_title, candidate, reason, e, exc_info=True, + ) + next_candidate = section_chain[s_idx + 1] if s_idx + 1 < len(section_chain) else None + if next_candidate: + emit_model_switch_event( + writer, agent_name=agent_label, stage="draft", + failed_model=candidate or "", next_model=next_candidate, + reason=reason, section_index=i, section_title=sec_title, + ) + switch_notices.append(build_switch_progress_event( + agent_name=agent_label, stage="draft", + failed_model=candidate or "", next_model=next_candidate, + reason=reason, section_index=i, section_title=sec_title, + )) + continue + if not section_succeeded: + assert last_section_exc is not None + raise last_section_exc # 章节完成 writer({ @@ -387,6 +432,7 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di "status": "awaiting_section_help", "progress_events": [ {"type": "step_start", "agent_name": agent_label, "message": f"开始写作（第{revision_count+1}稿）"}, + *switch_notices, { "type": "notice", "agent_name": agent_label, "title": "素材不足", "detail": f"有 {len(blocked_sections)} 个章节缺少足够素材支撑写作，需要你决定如何处理。", @@ -432,7 +478,9 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di on_text=None, )).strip() else: - resp = await model.ainvoke([ + # 扩写是兜底优化，不再做模型链兜底：拿当前已成功的模型试一次即可，失败就保留原文。 + expand_model = create_chat_model(name=current_model_name) if current_model_name else create_chat_model() + resp = await expand_model.ainvoke([ SystemMessage(content=DRAFT_SECTION_SYSTEM), HumanMessage(content=expand_msg), ]) @@ -479,6 +527,7 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di "status": "awaiting_draft_confirm", "progress_events": [ {"type": "step_start", "agent_name": agent_label, "message": f"开始写作（第{revision_count+1}稿）"}, + *switch_notices, *notices, { "type": "draft_ready", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py index 8e50a4f0..1a938d25 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py @@ -29,7 +29,13 @@ from ._agent_runner import ( stream_writing_agent, use_builtin_agent, ) -from ._utils import ThinkTagParser, _content_to_text, get_writing_model, llm_json, parse_json, reasoning_delta, safe_get_stream_writer +from ._model_fallback import ( + build_switch_progress_event, + classify_exception, + emit_model_switch_event, + fallback_chain, +) +from ._utils import ThinkTagParser, _content_to_text, llm_json, parse_json, reasoning_delta, resolve_writing_model_name, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -292,10 +298,11 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> writer = safe_get_stream_writer() full_text = "" # 只累积正文（非思考）用于 JSON 解析 - parser = ThinkTagParser() + extra_progress_events: list[dict] = [] - try: - if use_agent: + if use_agent: + # 内置 agent 路径走 stream_writing_agent，不在本模块做模型链兜底（agent 自带模型解析）。 + try: def _on_text(text: str, is_thinking: bool) -> None: writer({"type": "outline_chunk", "text": text, "is_thinking": is_thinking}) @@ -306,41 +313,81 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> config=config, on_text=_on_text, ) - else: - # get_writing_model 放进 try：用户模型缺失 / 模型名配错时抛 ValueError， - # 留在 try 外会硬崩溃；放进来则统一走下方的 status:error 兜底。 - model = get_writing_model(config, state) - async for chunk in model.astream([ - SystemMessage(content=system_prompt), - HumanMessage(content=user_prompt), - ]): - # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送 - reasoning = reasoning_delta(chunk) - if reasoning: - writer({"type": "outline_chunk", "text": reasoning, "is_thinking": True}) - delta = _content_to_text(chunk.content) - if not delta: + except Exception as e: + logger.error("outline streaming failed: %s", e, exc_info=True) + return { + "status": "error", + "progress_events": [{ + "type": "error", + "agent_name": agent_label, + "title": "模型错误", + "detail": str(e), + "message": "大纲生成失败", + }], + } + else: + # 旧路径：按 fallback_chain 顺序尝试，失败即换下一个模型重跑该环节。 + initial_name, _used_default = resolve_writing_model_name(config, state) + chain = fallback_chain(initial_name) + from deerflow.models import create_chat_model + + success = False + last_exc: BaseException | None = None + for idx, candidate in enumerate(chain or [None]): + full_text = "" + parser = ThinkTagParser() + try: + model = create_chat_model(name=candidate) if candidate else create_chat_model() + async for chunk in model.astream([ + SystemMessage(content=system_prompt), + HumanMessage(content=user_prompt), + ]): + reasoning = reasoning_delta(chunk) + if reasoning: + writer({"type": "outline_chunk", "text": reasoning, "is_thinking": True}) + delta = _content_to_text(chunk.content) + if not delta: + continue + for is_thinking, text in parser.feed(delta): + if not is_thinking: + full_text += text + writer({ + "type": "outline_chunk", + "text": text, + "is_thinking": is_thinking, + }) + success = True + break + except Exception as e: + last_exc = e + reason = classify_exception(e) + logger.error( + "outline streaming failed on model=%s (reason=%s): %s", + candidate, reason, e, exc_info=True, + ) + next_candidate = chain[idx + 1] if idx + 1 < len(chain) else None + if next_candidate: + emit_model_switch_event( + writer, agent_name=agent_label, stage="outline", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + ) + extra_progress_events.append(build_switch_progress_event( + agent_name=agent_label, stage="outline", + failed_model=candidate or "", next_model=next_candidate, reason=reason, + )) continue - for is_thinking, text in parser.feed(delta): - if not is_thinking: - full_text += text - writer({ - "type": "outline_chunk", - "text": text, - "is_thinking": is_thinking, - }) - except Exception as e: - logger.error("outline streaming failed: %s", e, exc_info=True) - return { - "status": "error", - "progress_events": [{ - "type": "error", - "agent_name": agent_label, - "title": "模型错误", - "detail": str(e), - "message": "大纲生成失败", - }], - } + if not success: + assert last_exc is not None + return { + "status": "error", + "progress_events": extra_progress_events + [{ + "type": "error", + "agent_name": agent_label, + "title": "模型错误", + "detail": str(last_exc), + "message": "大纲生成失败（所有候选模型均失败）", + }], + } # 编辑模式解析失败时回退为「当前大纲」，避免把用户已有大纲弄丢 if is_revise: @@ -387,6 +434,7 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> "status": "awaiting_outline_confirm", "progress_events": [ {"type": "step_start", "agent_name": agent_label, "message": "开始规划大纲"}, + *extra_progress_events, {"type": "outline_ready", "agent_name": agent_label, "message": f"大纲已生成，共 {section_count} 个章节"}, ], }