feat(ai-writing): 各阶段模型调用自动容错,失败时换下一个模型继续
参考多智能体会商的容错思路:意图解析/大纲/正文/审核四个 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) <noreply@anthropic.com>
This commit is contained in:
1 parent
23743ee193
commit
b59cbd51d9
9 files changed
+516
-62
No files matched your search
@@ -19,6 +19,7 @@ const DISPLAYED_TYPES = new Set([
|
||||
'await_user',
|
||||
'resumed',
|
||||
'notice',
|
||||
'model_switch',
|
||||
'done',
|
||||
'error',
|
||||
]);
|
||||
|
||||
@@ -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<string, string> = {
|
||||
|
||||
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 (
|
||||
|
||||
@@ -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<AIWritingState> = (() => {
|
||||
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<string, unknown>, sessionId
|
||||
out.reviewItemIssues = (out.issues as Record<string, unknown>[]).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<string, unknown>[]).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<string, unknown>).failedModel ?? ''),
|
||||
nextModel: String(o.next_model ?? (o as Record<string, unknown>).nextModel ?? ''),
|
||||
reason: String(o.reason ?? ''),
|
||||
sectionIndex: typeof o.section_index === 'number'
|
||||
? (o.section_index as number)
|
||||
: (typeof (o as Record<string, unknown>).sectionIndex === 'number'
|
||||
? ((o as Record<string, unknown>).sectionIndex as number)
|
||||
: undefined),
|
||||
sectionTitle: typeof o.section_title === 'string'
|
||||
? (o.section_title as string)
|
||||
: (typeof (o as Record<string, unknown>).sectionTitle === 'string'
|
||||
? ((o as Record<string, unknown>).sectionTitle as string)
|
||||
: undefined),
|
||||
};
|
||||
}
|
||||
return merged as unknown as ProgressEvent;
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
}
|
||||
|
||||
// ── 用户干预提交体 ────────────────────────────────────────────────────────────
|
||||
|
||||
+197
@@ -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)
|
||||
+57
-15
@@ -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",
|
||||
|
||||
+50
-16
@@ -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,12 +212,24 @@ 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 = ""
|
||||
success_model: str | None = None
|
||||
last_error: BaseException | None = None
|
||||
for idx, candidate in enumerate(chain or [None]):
|
||||
full_text = "" # 每次重试都从头累积,避免不同模型的输出粘在一起
|
||||
parser = ThinkTagParser()
|
||||
try:
|
||||
model = get_model(config)
|
||||
model = create_chat_model(name=candidate) if candidate else create_chat_model()
|
||||
async for chunk in model.astream(
|
||||
[SystemMessage(content=system), HumanMessage(content=user)]
|
||||
):
|
||||
@@ -231,23 +248,38 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d
|
||||
"text": text,
|
||||
"is_thinking": is_thinking,
|
||||
})
|
||||
success_model = candidate or ""
|
||||
break
|
||||
except Exception as exc:
|
||||
# 模型 / 网络挂掉 —— 不阻断流程,用 fallback 继续。
|
||||
logger.error("intent_parser streaming failed: %s", exc, exc_info=True)
|
||||
llm_error = 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:
|
||||
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
|
||||
# 链尾仍失败:保留 fallback 默认值,向用户报一条非致命 notice,流程继续。
|
||||
progress_events.append({
|
||||
"type": "notice",
|
||||
"agent_name": "系统",
|
||||
"title": "模型错误",
|
||||
"detail": f"意图解析失败,已按默认配置兜底。原因:{llm_error}",
|
||||
"detail": (
|
||||
f"意图解析失败,已按默认配置兜底。原因:{type(exc).__name__}: {exc}"
|
||||
),
|
||||
})
|
||||
|
||||
result = parse_json(full_text, fallback_result) if full_text else fallback_result
|
||||
|
||||
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
|
||||
|
||||
+67
-18
@@ -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)
|
||||
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
|
||||
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": "草稿生成失败",
|
||||
}],
|
||||
}
|
||||
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,12 +315,28 @@ 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([
|
||||
# 旧路径:按 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_content(content 为空):作为思考实时推送
|
||||
reasoning = reasoning_delta(chunk)
|
||||
if reasoning:
|
||||
writer({
|
||||
@@ -342,6 +361,32 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
|
||||
"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",
|
||||
|
||||
+59
-11
@@ -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:
|
||||
# 内置 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,15 +313,35 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
|
||||
config=config,
|
||||
on_text=_on_text,
|
||||
)
|
||||
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:
|
||||
# get_writing_model 放进 try:用户模型缺失 / 模型名配错时抛 ValueError,
|
||||
# 留在 try 外会硬崩溃;放进来则统一走下方的 status:error 兜底。
|
||||
model = get_writing_model(config, state)
|
||||
# 旧路径:按 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_content(content 为空):作为思考实时推送
|
||||
reasoning = reasoning_delta(chunk)
|
||||
if reasoning:
|
||||
writer({"type": "outline_chunk", "text": reasoning, "is_thinking": True})
|
||||
@@ -329,16 +356,36 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
|
||||
"text": text,
|
||||
"is_thinking": is_thinking,
|
||||
})
|
||||
success = True
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error("outline streaming failed: %s", e, exc_info=True)
|
||||
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
|
||||
if not success:
|
||||
assert last_exc is not None
|
||||
return {
|
||||
"status": "error",
|
||||
"progress_events": [{
|
||||
"progress_events": extra_progress_events + [{
|
||||
"type": "error",
|
||||
"agent_name": agent_label,
|
||||
"title": "模型错误",
|
||||
"detail": str(e),
|
||||
"message": "大纲生成失败",
|
||||
"detail": str(last_exc),
|
||||
"message": "大纲生成失败(所有候选模型均失败)",
|
||||
}],
|
||||
}
|
||||
|
||||
@@ -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} 个章节"},
|
||||
],
|
||||
}
|
||||
Reference in new issue
Block a user