AI 写作四阶段内置智能体 Phase 1-4 落地

新增 ai-writing-researcher/outliner/writer/editor 种子 agent、_agent_runner 与 use_builtin_agents 灰度开关；四节点可按 flag 切换 SOUL 模式；时间线补大纲规划师配色并更新改造文档。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-11 22:58:38 +08:00
1 parent 1a41ac97d2
commit a4a9a4d26e
@@ -0,0 +1,269 @@
+"""AI 写作「内置功能型智能体」运行助手。
+
+四个写作阶段(素材收集/大纲/写作/审核)的图节点在 ``config.yaml →
+ai_writing.use_builtin_agents.<stage>`` 打开时,不再使用 ``prompts/*.py`` 的
+硬编码提示词直调 LLM,而是改为:
+
+1. 从 ``.deer-flow/agents/<agent-id>/`` 加载该内置 agent 的 **SOUL.md**
+   (人设 + 工作流 + 结构化输出契约)作为 system prompt;
+2. 模型按 ``agent config.yaml 的 model > 节点显式回退模型 > state.model_name``
+   的优先级解析(admin 可在智能体管理页给某阶段单独配模型);
+3. 流式调用该模型,chunk 通过调用方传入的回调透传到 LangGraph ``custom``
+   通道(前端时间线打字机不回归);
+4. 按 SOUL 规定的 marker(如 ``[OUTLINE_READY]``)+ ```json``` 块解析结构化
+   输出,解析复用 ``_utils.parse_json`` 多级容错。
+
+设计取舍(对应 ai-writing-内置智能体改造.md §3.2 / §7):
+
+- **编排控制权留在节点**:writer 的逐章循环、researcher 的检索编排、editor 的
+  三维度循环都不交给 agent——agent 只承担「单次结构化产出」,这样干预点/逐章
+  流式/严格模式求助逻辑全部原样保留。
+- **不经过 lead agent 运行时**:本模块位于 harness 层,不能 import ``app.*``
+  (intent.py 的 loopback 机制在 app 层),也无需为单轮结构化调用付出
+  thread/checkpointer/中间件链的成本。SOUL + per-agent 模型已覆盖本场景的
+  可配置性诉求;researcher 的「挂载技能」(Phase 3)再考虑接入完整运行时。
+- **agent 目录缺失时抛 ``WritingAgentUnavailable``**:由节点捕获后回退旧的
+  硬编码 prompt 路径,种子器(app 层 ``_ai_writing_seed.py``)失败不阻断写作。
+"""
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import RunnableConfig
+
+from ._utils import ThinkTagParser, _content_to_text, parse_json, reasoning_delta
+
+logger = logging.getLogger(__name__)
+
+# ── 内置写作 agent 的稳定 id(与 app 层种子器 _ai_writing_seed.py 对齐)──────────
+RESEARCHER_AGENT_ID = "ai-writing-researcher"
+OUTLINER_AGENT_ID = "ai-writing-outliner"
+WRITER_AGENT_ID = "ai-writing-writer"
+EDITOR_AGENT_ID = "ai-writing-editor"
+
+# ── SOUL 规定的结构化输出 marker(契约见各 agent 的 SOUL.md)─────────────────────
+KEYWORDS_READY_MARKER = "[KEYWORDS_READY]"
+MATERIALS_RANKED_MARKER = "[MATERIALS_RANKED]"
+OUTLINE_READY_MARKER = "[OUTLINE_READY]"
+REVIEW_READY_MARKER = "[REVIEW_READY]"
+
+# flag 名(config.yaml ai_writing.use_builtin_agents.<stage>)→ agent id
+STAGE_AGENT_IDS = {
+    "researcher": RESEARCHER_AGENT_ID,
+    "outliner": OUTLINER_AGENT_ID,
+    "writer": WRITER_AGENT_ID,
+    "editor": EDITOR_AGENT_ID,
+}
+
+
+class WritingAgentUnavailable(RuntimeError):
+    """内置写作 agent 的 SOUL.md 缺失/为空,无法按 agent 模式运行。
+
+    节点捕获后应回退到旧的硬编码 prompt 路径(等价于 flag 关闭),不让写作
+    流程因 agent 目录被删而中断。
+    """
+
+
+def use_builtin_agent(stage: str) -> bool:
+    """读 ``config.yaml → ai_writing.use_builtin_agents.<stage>`` 开关。
+
+    缺失/非法一律视为 False(默认走旧节点逻辑)。config 带 mtime 热加载,
+    改完 config.yaml 新会话即生效。
+    """
+    from ._utils import get_ai_writing_config
+
+    flags = get_ai_writing_config().get("use_builtin_agents")
+    if not isinstance(flags, dict):
+        return False
+    return bool(flags.get(stage, False))
+
+
+def load_writing_agent(agent_id: str):
+    """加载内置写作 agent 的 (config, soul);SOUL 缺失抛 WritingAgentUnavailable。"""
+    from deerflow.config.agents_config import load_agent_config, load_agent_soul
+
+    soul = load_agent_soul(agent_id)
+    if not soul:
+        raise WritingAgentUnavailable(
+            f"内置写作智能体 '{agent_id}' 的 SOUL.md 缺失或为空"
+            "(目录可能被删除,重启 Gateway 或重新打开写作页可自动补建)"
+        )
+    try:
+        agent_cfg = load_agent_config(agent_id)
+    except ValueError:
+        # config.yaml 解析失败不致命:SOUL 仍可用,模型回退节点/表单选择。
+        logger.warning("load_writing_agent: config.yaml of '%s' is invalid, ignoring", agent_id)
+        agent_cfg = None
+    return agent_cfg, soul
+
+
+def agent_display_name(agent_id: str, default: str) -> str:
+    """取 agent 的展示名(progress_events.agent_name 用),加载失败回退默认值。"""
+    try:
+        from deerflow.config.agents_config import load_agent_config
+
+        cfg = load_agent_config(agent_id)
+        name = (cfg.name or "").strip() if cfg else ""
+        return name or default
+    except Exception:
+        return default
+
+
+def resolve_agent_model_name(
+    agent_cfg,
+    state: dict,
+    config: RunnableConfig,
+    *,
+    explicit_fallback: str | None = None,
+    strict: bool = True,
+) -> str | None:
+    """解析本次 agent 调用使用的模型名。
+
+    优先级:agent config.yaml 的 ``model`` > ``explicit_fallback``(节点显式
+    传入,如 researcher 的 keyword_model/rank_model)> ``state.model_name``
+    (用户写作表单选的主模型)> configurable.model_name。
+
+    strict=True(大纲/写作/审核):全部为空抛 ValueError(与 get_writing_model
+    口径一致——不静默回退默认模型);strict=False(researcher 辅助任务):
+    返回 None,由 create_chat_model 回退 config.yaml 的 models[0]。
+    """
+    candidates = (
+        (getattr(agent_cfg, "model", None) or "") if agent_cfg else "",
+        explicit_fallback or "",
+        (state or {}).get("model_name") or "",
+        (config.get("configurable", {}) or {}).get("model_name") or "",
+    )
+    for name in candidates:
+        if isinstance(name, str) and name.strip():
+            return name.strip()
+    if strict:
+        raise ValueError("未指定写作模型(model_name 为空):请在写作表单选择模型后重试")
+    return None
+
+
+def parse_marker_json(text: str, marker: str, fallback: dict) -> dict:
+    """从 agent 输出中按「marker + ```json``` 块」契约解析 JSON,多级降级。
+
+    1. ``marker\\n```json {...} ``` ``(SOUL 规定的标准形态,正则仿
+       intent.py::_INTENT_READY_PATTERN);
+    2. marker 存在但没包代码块 → 取 marker 之后的全部文本交给 parse_json;
+    3. marker 缺失(弱模型漏输出)→ 整段文本交给 parse_json;
+    4. 全部失败 → fallback。
+    """
+    if not text or not text.strip():
+        return fallback
+
+    pattern = re.compile(re.escape(marker) + r"\s*```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)
+    m = pattern.search(text)
+    if m:
+        result = parse_json(m.group(1), fallback)
+        if result is not fallback:
+            return result
+
+    idx = text.find(marker)
+    if idx != -1:
+        tail = text[idx + len(marker):]
+        result = parse_json(tail, fallback)
+        if result is not fallback:
+            return result
+
+    return parse_json(text, fallback)
+
+
+async def stream_writing_agent(
+    *,
+    agent_id: str,
+    user_payload: str,
+    state: dict,
+    config: RunnableConfig,
+    on_text: Callable[[str, bool], None] | None = None,
+    explicit_model_fallback: str | None = None,
+    strict_model: bool = True,
+) -> str:
+    """在「SOUL 作 system prompt + 单条 Human 输入」契约下流式运行一个写作 agent。
+
+    Args:
+      agent_id: 内置写作 agent id(STAGE_AGENT_IDS 的值之一)。
+      user_payload: 本阶段输入(意图/素材/大纲/草稿等拼成的一段文本)。
+      on_text: 每个流式片段回调 ``(text, is_thinking)``,调用方在回调里把
+        chunk 包装成本阶段的 custom 事件(outline_chunk / draft_chunk /
+        review_chunk ...)推给前端;None = 不透传(如 researcher 的辅助任务)。
+      explicit_model_fallback / strict_model: 见 resolve_agent_model_name。
+
+    Returns:
+      累积的**正文**文本(<think> 思考与 reasoning_content 已剥离),供
+      parse_marker_json / 章节拼装使用。
+
+    Raises:
+      WritingAgentUnavailable: SOUL 缺失(调用方回退旧 prompt 路径)。
+      其余模型异常原样上抛,由调用方按既有 try/except 兜底成 status:error。
+    """
+    agent_cfg, soul = load_writing_agent(agent_id)
+    model_name = resolve_agent_model_name(
+        agent_cfg, state, config,
+        explicit_fallback=explicit_model_fallback,
+        strict=strict_model,
+    )
+
+    from deerflow.models import create_chat_model
+
+    model = create_chat_model(name=model_name)
+
+    full_text = ""
+    parser = ThinkTagParser()
+    async for chunk in model.astream([
+        SystemMessage(content=soul),
+        HumanMessage(content=user_payload),
+    ]):
+        # 推理模型的思考走 reasoning_content(content 为空):作为思考实时推送
+        reasoning = reasoning_delta(chunk)
+        if reasoning and on_text is not None:
+            on_text(reasoning, 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
+            if on_text is not None:
+                on_text(text, is_thinking)
+    return full_text
+
+
+async def run_writing_agent(
+    *,
+    agent_id: str,
+    user_payload: str,
+    marker: str,
+    fallback: dict,
+    state: dict,
+    config: RunnableConfig,
+    on_text: Callable[[str, bool], None] | None = None,
+    explicit_model_fallback: str | None = None,
+    strict_model: bool = True,
+) -> dict:
+    """跑一个写作 agent 并按 marker 契约解析其结构化输出(失败回 fallback)。
+
+    模型调用异常时与 ``_utils.llm_json`` 同口径:返回
+    ``{**fallback, "_llm_error": str(e)}``,不上抛(SOUL 缺失除外——
+    WritingAgentUnavailable 原样上抛,由节点切回旧 prompt 路径)。
+    """
+    try:
+        full_text = await stream_writing_agent(
+            agent_id=agent_id,
+            user_payload=user_payload,
+            state=state,
+            config=config,
+            on_text=on_text,
+            explicit_model_fallback=explicit_model_fallback,
+            strict_model=strict_model,
+        )
+    except WritingAgentUnavailable:
+        raise
+    except Exception as e:
+        logger.error("run_writing_agent(%s) failed: %s", agent_id, e, exc_info=True)
+        return {**fallback, "_llm_error": str(e)}
+    return parse_marker_json(full_text, marker, fallback)
@@ -11,6 +11,16 @@ from ..state import AIWritingState, ReviewResult, ReviewIssue
 from ..prompts.editor_prompts import (
    EDITOR_SYSTEM, FACT_REVIEW_USER, LOGIC_REVIEW_USER, LANGUAGE_REVIEW_USER,
 )
+from ._agent_runner import (
+    EDITOR_AGENT_ID,
+    REVIEW_READY_MARKER,
+    WritingAgentUnavailable,
+    agent_display_name,
+    load_writing_agent,
+    resolve_agent_model_name,
+    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

 logger = logging.getLogger(__name__)
@@ -119,19 +129,45 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
        ("language", LANGUAGE_REVIEW_USER.format(article=article)),
    ]

-    # 模型创建守卫：模型名配错时 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": "审核失败",
-            }],
-        }
+    # 内置 agent 模式（config.yaml ai_writing.use_builtin_agents.editor）：
+    # 三维度循环仍由节点编排（review_chunk / review_item_done 流式形态不变），
+    # 每个维度跑一次 ai-writing-editor agent（SOUL 作 system prompt，输出
+    # [REVIEW_READY] + {"score","issues"}）。agent 目录缺失则静默回退旧路径。
+    use_agent = use_builtin_agent("editor")
+    agent_label = "编辑"
+    model = None
+    if use_agent:
+        try:
+            agent_cfg, _soul = load_writing_agent(EDITOR_AGENT_ID)
+            # 模型守卫提前到循环外：模型名缺失/配错时与旧路径同样转成 status:error
+            resolve_agent_model_name(agent_cfg, state, config)
+            agent_label = agent_display_name(EDITOR_AGENT_ID, "编辑")
+        except WritingAgentUnavailable as e:
+            logger.warning("editor builtin agent unavailable, falling back to legacy prompts: %s", e)
+            use_agent = False
+        except Exception as e:
+            logger.error("editor agent model resolution failed: %s", e, exc_info=True)
+            return {
+                "status": "error",
+                "progress_events": [{
+                    "type": "error", "agent_name": agent_label,
+                    "title": "模型错误", "detail": str(e), "message": "审核失败",
+                }],
+            }
+    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": "审核失败",
+                }],
+            }
    results: dict[str, dict] = {}
    review_errors: list[str] = []  # 分项审查的模型异常，供完成后非致命提示

@@ -140,8 +176,24 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
    for category, user_prompt in REVIEW_TASKS:
        fallback = {"score": 80, "issues": []}
        try:
-            full_text = await _stream_one_review(model, user_prompt, category, writer)
-            res = parse_json(full_text, fallback)
+            if use_agent:
+                def _on_text(text: str, is_thinking: bool, _cat: str = category) -> None:
+                    writer({"type": "review_chunk", "category": _cat, "text": text, "is_thinking": is_thinking})
+
+                res = await run_writing_agent(
+                    agent_id=EDITOR_AGENT_ID,
+                    user_payload=user_prompt,
+                    marker=REVIEW_READY_MARKER,
+                    fallback=fallback,
+                    state=state,
+                    config=config,
+                    on_text=_on_text,
+                )
+                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)
+                res = parse_json(full_text, fallback)
        except Exception as e:
            logger.warning("editor review [%s] failed: %s", category, e)
            res = fallback
@@ -191,7 +243,7 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
    if review_errors:
        notices.append({
            "type": "notice",
-            "agent_name": "编辑",
+            "agent_name": agent_label,
            "title": "模型错误",
            "detail": "部分审查项调用失败，已按默认分兜底。原因：" + "；".join(review_errors[:3]),
        })
@@ -199,11 +251,11 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
    return {
        "review_result": review,
        "progress_events": [
-            {"type": "step_start", "agent_name": "编辑", "message": "开始审核"},
+            {"type": "step_start", "agent_name": agent_label, "message": "开始审核"},
            *notices,
            {
                "type": "review_ready",
-                "agent_name": "编辑",
+                "agent_name": agent_label,
                "message": f"审核完成，综合得分 {overall_score}，结论：{'通过' if verdict == 'pass' else '打回修改'}",
            },
        ],
@@ -18,10 +18,46 @@ from ..prompts.researcher_prompts import (
    EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER,
    RANK_MATERIALS_SYSTEM, RANK_MATERIALS_USER,
 )
+from ._agent_runner import (
+    KEYWORDS_READY_MARKER,
+    MATERIALS_RANKED_MARKER,
+    RESEARCHER_AGENT_ID,
+    WritingAgentUnavailable,
+    agent_display_name,
+    load_writing_agent,
+    run_writing_agent,
+    use_builtin_agent,
+)
 from ._utils import get_ai_writing_config, llm_json, safe_get_stream_writer

 logger = logging.getLogger(__name__)

+# ── 内置 agent 模式的子任务 payload（SOUL 已含人设与输出契约，这里只给事实输入）──
+
+_AGENT_KEYWORDS_USER = """任务：提取检索词
+
+写作意图：{user_intent}
+文章类型：{article_type}
+目标读者：{target_audience}
+
+请提取 {keyword_count} 个搜索关键词，用于搜集相关素材。
+关键词数量必须为 {keyword_count} 个，不多不少；必须依据上述写作意图提炼真实检索词。
+严格按 SOUL 约定输出 [KEYWORDS_READY] + JSON。
+
+/no_think"""
+
+_AGENT_RANK_USER = """任务：素材排序
+
+写作意图：{user_intent}
+
+素材列表（每条含ID和摘要）：
+{materials_list}
+
+请选出最相关的素材ID（最多8条），按相关度从高到低排序。
+严格按 SOUL 约定输出 [MATERIALS_RANKED] + JSON。
+
+/no_think"""
+
 _NOTEBOOK_SEARCH_PATH = "/api/v1/source/mentioned-documents/search"
 _NOTEBOOK_MAX_KEYWORDS = 3   # 每次写作最多对笔记发起的检索轮数
 _NOTEBOOK_TOP_K = 5          # 每次检索取回的 chunk 数
@@ -89,12 +125,12 @@ def _per_keyword_limit(max_materials: int, total_keywords: int) -> int:
    return (max_materials + total_keywords - 1) // total_keywords


-def _notice(title: str, detail: str, raw: str = "") -> dict:
+def _notice(title: str, detail: str, raw: str = "", agent_name: str = "素材收集专家") -> dict:
    """构造一个非致命告警事件，前端在时间线显示「标题 + 真实原因」，不打断写作流程。

    raw 非空时附带接口的原始错误响应体，前端默认折叠展示。
    """
-    event = {"type": "notice", "agent_name": "素材收集专家", "title": title, "detail": detail}
+    event = {"type": "notice", "agent_name": agent_name, "title": title, "detail": detail}
    if raw:
        event["raw"] = raw
    return event
@@ -254,6 +290,20 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
    # 素材总量上限（config.yaml: ai_writing.max_materials），按检索词数均分到每个词
    max_materials = _resolve_max_materials(ai_writing_config.get("max_materials"))

+    # 内置 agent 模式（config.yaml ai_writing.use_builtin_agents.researcher）：
+    # 检索编排（并发搜索 / 技能编排 / 笔记检索）仍由节点控制，只把两个 LLM 子任务
+    # （提取检索词 / 素材排序）换成跑 ai-writing-researcher agent（SOUL 可被 admin
+    # 编辑；模型优先级 agent.model > keyword_model/rank_model > 表单主模型）。
+    use_agent = use_builtin_agent("researcher")
+    researcher_label = "素材收集专家"
+    if use_agent:
+        try:
+            load_writing_agent(RESEARCHER_AGENT_ID)
+            researcher_label = agent_display_name(RESEARCHER_AGENT_ID, "素材收集专家")
+        except WritingAgentUnavailable as e:
+            logger.warning("researcher builtin agent unavailable, falling back to legacy prompts: %s", e)
+            use_agent = False
+
    def _emit(message: str) -> None:
        writer({"type": "research_progress", "message": message})

@@ -272,18 +322,35 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        _emit(f"追加检索 {len(keywords)} 个关键词…")
    else:
        _emit("正在生成检索词…")
-        kw_result = await llm_json(
-            EXTRACT_KEYWORDS_SYSTEM,
-            EXTRACT_KEYWORDS_USER.format(
-                user_intent=intent,
-                article_type=article_type,
-                target_audience=target_audience,
-                keyword_count=keyword_count,
-            ),
-            fallback={"keywords": [intent]},
-            config=config,
-            model_name=keyword_model,
-        )
+        if use_agent:
+            kw_result = await run_writing_agent(
+                agent_id=RESEARCHER_AGENT_ID,
+                user_payload=_AGENT_KEYWORDS_USER.format(
+                    user_intent=intent,
+                    article_type=article_type,
+                    target_audience=target_audience,
+                    keyword_count=keyword_count,
+                ),
+                marker=KEYWORDS_READY_MARKER,
+                fallback={"keywords": [intent]},
+                state=state,
+                config=config,
+                explicit_model_fallback=keyword_model,
+                strict_model=False,
+            )
+        else:
+            kw_result = await llm_json(
+                EXTRACT_KEYWORDS_SYSTEM,
+                EXTRACT_KEYWORDS_USER.format(
+                    user_intent=intent,
+                    article_type=article_type,
+                    target_audience=target_audience,
+                    keyword_count=keyword_count,
+                ),
+                fallback={"keywords": [intent]},
+                config=config,
+                model_name=keyword_model,
+            )
        # 清洗关键词：剔除占位符 / 空值，全空时回退写作意图
        keywords = _clean_keywords(kw_result.get("keywords"), intent)
        if extra_keywords:
@@ -321,7 +388,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        # 与前端 [ai-writing:custom] 日志对齐，钉死是否实时推送。
        logger.info("[skill-timing] t=%.3f emit skill_search_step type=%s skill=%s",
                    time.monotonic(), step.get("type"), step.get("skill_name") or "")
-        writer({"type": "skill_search_step", "agent_name": "素材收集专家", **step})
+        writer({"type": "skill_search_step", "agent_name": researcher_label, **step})

    if is_skill_search:
        # ── 技能编排检索 ──
@@ -509,13 +576,26 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
            f"- ID: {m['id']}  标题: {m['title']}  摘要: {m['content'][:100]}"
            for m in raw_materials
        )
-        rank_result = await llm_json(
-            RANK_MATERIALS_SYSTEM,
-            RANK_MATERIALS_USER.format(user_intent=intent, materials_list=materials_list),
-            fallback={"ranked_ids": [m["id"] for m in raw_materials[:max_materials]], "summary": f"共搜索到{len(raw_materials)}条素材"},
-            config=config,
-            model_name=rank_model,
-        )
+        rank_fallback = {"ranked_ids": [m["id"] for m in raw_materials[:max_materials]], "summary": f"共搜索到{len(raw_materials)}条素材"}
+        if use_agent:
+            rank_result = await run_writing_agent(
+                agent_id=RESEARCHER_AGENT_ID,
+                user_payload=_AGENT_RANK_USER.format(user_intent=intent, materials_list=materials_list),
+                marker=MATERIALS_RANKED_MARKER,
+                fallback=rank_fallback,
+                state=state,
+                config=config,
+                explicit_model_fallback=rank_model,
+                strict_model=False,
+            )
+        else:
+            rank_result = await llm_json(
+                RANK_MATERIALS_SYSTEM,
+                RANK_MATERIALS_USER.format(user_intent=intent, materials_list=materials_list),
+                fallback=rank_fallback,
+                config=config,
+                model_name=rank_model,
+            )
        rank_error = rank_result.get("_llm_error", "")
        ranked_ids: list[str] = rank_result.get("ranked_ids", [m["id"] for m in raw_materials])
        id_to_mat = {m["id"]: m for m in raw_materials}
@@ -554,9 +634,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
    # 容错告警：检索接口 / 模型调用出错时以非致命 notice 提示用户，不打断写作流程
    notices: list[dict] = []
    if kw_result.get("_llm_error"):
-        notices.append(_notice("模型错误", f"检索词生成失败，已用默认关键词兜底。原因：{kw_result['_llm_error']}"))
+        notices.append(_notice("模型错误", f"检索词生成失败，已用默认关键词兜底。原因：{kw_result['_llm_error']}", agent_name=researcher_label))
    if rank_error:
-        notices.append(_notice("模型错误", f"素材排序失败，已按相关度兜底排序。原因：{rank_error}"))
+        notices.append(_notice("模型错误", f"素材排序失败，已按相关度兜底排序。原因：{rank_error}", agent_name=researcher_label))
    if search_errors:
        uniq = list(dict.fromkeys(search_errors))
        # 解析失败时收集到的接口原始响应体，前端折叠展示便于定位
@@ -567,6 +647,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
            "ES 波动导致检索异常",
            "；".join(uniq[:3]),
            raw="\n\n----------\n\n".join(raw_dumps[:3]),
+            agent_name=researcher_label,
        ))

    # 进度事件文案：用户输入 / 追加模式 / 技能模式下让用户看到差异化提示
@@ -583,8 +664,8 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        # 补充检索：直接回写作（沿用原大纲）；常规检索：进入素材确认暂停点
        "status": "writing_draft" if is_supplement else "awaiting_material_confirm",
        "progress_events": [
-            {"type": "step_start", "agent_name": "素材收集专家", "message": step_start_msg},
+            {"type": "step_start", "agent_name": researcher_label, "message": step_start_msg},
            *notices,
-            {"type": "materials_ready", "agent_name": "素材收集专家", "message": summary},
+            {"type": "materials_ready", "agent_name": researcher_label, "message": summary},
        ],
    }
@@ -12,6 +12,15 @@ from ..prompts.writer_prompts import (
    DRAFT_SECTION_SYSTEM, DRAFT_SECTION_USER, DRAFT_EXPAND_USER,
    WRITER_MODE_LOOSE, WRITER_MODE_STRICT, SECTION_BLOCKED_MARKER,
 )
+from ._agent_runner import (
+    WRITER_AGENT_ID,
+    WritingAgentUnavailable,
+    agent_display_name,
+    load_writing_agent,
+    resolve_agent_model_name,
+    stream_writing_agent,
+    use_builtin_agent,
+)
 from ._utils import get_writing_model, _content_to_text, count_words, reasoning_delta, ThinkTagParser, safe_get_stream_writer

 logger = logging.getLogger(__name__)
@@ -194,19 +203,44 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
    def _section_title(idx: int, sec: dict) -> str:
        return sec.get("section_title") or sec.get("sectionTitle") or f"章节{idx + 1}"

-    # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError，
-    # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。
-    try:
-        model = get_writing_model(config, state)
-    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": "草稿生成失败",
-            }],
-        }
+    # 内置 agent 模式（config.yaml ai_writing.use_builtin_agents.writer）：
+    # 逐章循环控制权仍在节点（严格模式求助 / 逐章流式不动），每章跑一次
+    # ai-writing-writer agent（SOUL 作 system prompt，直接产出章节 Markdown）。
+    use_agent = use_builtin_agent("writer")
+    agent_label = "作家"
+    model = None
+    if use_agent:
+        try:
+            agent_cfg, _soul = load_writing_agent(WRITER_AGENT_ID)
+            # 模型守卫提前到循环外：模型名缺失/配错与旧路径同样转成 status:error
+            resolve_agent_model_name(agent_cfg, state, config)
+            agent_label = agent_display_name(WRITER_AGENT_ID, "作家")
+        except WritingAgentUnavailable as e:
+            logger.warning("writer builtin agent unavailable, falling back to legacy prompts: %s", e)
+            use_agent = False
+        except Exception as e:
+            logger.error("draft agent model resolution failed: %s", e, exc_info=True)
+            return {
+                "status": "error",
+                "progress_events": [{
+                    "type": "error", "agent_name": agent_label,
+                    "title": "模型错误", "detail": str(e), "message": "草稿生成失败",
+                }],
+            }
+    if not use_agent:
+        # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError，
+        # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。
+        try:
+            model = get_writing_model(config, state)
+        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] = []  # 分节写作的模型异常，供完成后非致命提示
@@ -246,36 +280,55 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
        section_buf = ""   # 只累积正文（非思考）用于拼接 Markdown
        parser = ThinkTagParser()
        try:
-            async for chunk in model.astream([
-                SystemMessage(content=DRAFT_SECTION_SYSTEM),
-                HumanMessage(content=user_msg),
-            ]):
-                # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
-                reasoning = reasoning_delta(chunk)
-                if reasoning:
+            if use_agent:
+                def _on_text(text: str, is_thinking: bool, _i: int = i, _t: str = sec_title) -> None:
                    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,
+                        "section_index": _i,
+                        "section_title": _t,
                        "text": text,
                        "is_thinking": is_thinking,
                        "is_section_done": False,
                    })

+                section_buf = await stream_writing_agent(
+                    agent_id=WRITER_AGENT_ID,
+                    user_payload=user_msg,
+                    state=state,
+                    config=config,
+                    on_text=_on_text,
+                )
+            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,
+                        })
+
            # 章节完成
            writer({
                "type": "draft_chunk",
@@ -319,9 +372,9 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
            "blocked_sections": blocked_sections,
            "status": "awaiting_section_help",
            "progress_events": [
-                {"type": "step_start", "agent_name": "作家", "message": f"开始写作（第{revision_count+1}稿）"},
+                {"type": "step_start", "agent_name": agent_label, "message": f"开始写作（第{revision_count+1}稿）"},
                {
-                    "type": "notice", "agent_name": "作家", "title": "素材不足",
+                    "type": "notice", "agent_name": agent_label, "title": "素材不足",
                    "detail": f"有 {len(blocked_sections)} 个章节缺少足够素材支撑写作，需要你决定如何处理。",
                },
            ],
@@ -356,14 +409,23 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
                existing_text=section_bufs[idx],
                target_words=expand_target,
            )
-            resp = await model.ainvoke([
-                SystemMessage(content=DRAFT_SECTION_SYSTEM),
-                HumanMessage(content=expand_msg),
-            ])
-            parser = ThinkTagParser()
-            expanded = "".join(
-                t for is_thinking, t in parser.feed(_content_to_text(resp.content)) if not is_thinking
-            ).strip()
+            if use_agent:
+                expanded = (await stream_writing_agent(
+                    agent_id=WRITER_AGENT_ID,
+                    user_payload=expand_msg,
+                    state=state,
+                    config=config,
+                    on_text=None,
+                )).strip()
+            else:
+                resp = await model.ainvoke([
+                    SystemMessage(content=DRAFT_SECTION_SYSTEM),
+                    HumanMessage(content=expand_msg),
+                ])
+                parser = ThinkTagParser()
+                expanded = "".join(
+                    t for is_thinking, t in parser.feed(_content_to_text(resp.content)) if not is_thinking
+                ).strip()
            # 仅在扩写确实更长时才采用，避免越改越短
            if count_words(expanded) > count_words(section_bufs[idx]):
                section_bufs[idx] = expanded
@@ -390,7 +452,7 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
    if section_errors:
        notices.append({
            "type": "notice",
-            "agent_name": "作家",
+            "agent_name": agent_label,
            "title": "模型错误",
            "detail": "部分章节生成失败，对应段落已占位。原因：" + "；".join(section_errors[:3]),
        })
@@ -400,11 +462,11 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
        "drafts": [draft],
        "status": "awaiting_draft_confirm",
        "progress_events": [
-            {"type": "step_start", "agent_name": "作家", "message": f"开始写作（第{revision_count+1}稿）"},
+            {"type": "step_start", "agent_name": agent_label, "message": f"开始写作（第{revision_count+1}稿）"},
            *notices,
            {
                "type": "draft_ready",
-                "agent_name": "作家",
+                "agent_name": agent_label,
                "message": f"草稿完成，共 {word_count} 字",
                "draft_title": outline.get("title", "文章"),
                "draft_markdown": full_markdown,
@@ -19,10 +19,51 @@ from ..prompts.writer_prompts import (
    OUTLINE_USER,
 )
 from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
+from ._agent_runner import (
+    OUTLINE_READY_MARKER,
+    OUTLINER_AGENT_ID,
+    WritingAgentUnavailable,
+    agent_display_name,
+    load_writing_agent,
+    parse_marker_json,
+    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

 logger = logging.getLogger(__name__)

+# ── 内置 agent 模式的输入 payload（SOUL 已含人设/层级语义/输出契约，这里只给本次任务的事实输入）──
+
+_AGENT_OUTLINE_USER = """任务模式：新规划
+
+写作意图：{user_intent}
+文章类型：{article_type}
+目标读者：{target_audience}
+目标字数：{word_count_target}字{feedback_section}
+
+可用素材摘要：
+{materials_summary}
+
+请规划文章大纲（3-5 个章节），严格按 SOUL 约定输出 [OUTLINE_READY] + JSON。"""
+
+_AGENT_OUTLINE_REVISE_USER = """任务模式：修改模式
+
+写作意图：{user_intent}
+文章类型：{article_type}
+目标读者：{target_audience}
+目标字数：{word_count_target}字
+
+当前大纲（JSON）：
+{current_outline}
+
+用户修改要求：{feedback}
+
+可用素材摘要：
+{materials_summary}
+
+请在「当前大纲」基础上落实「用户修改要求」，未提到的章节保持原样，严格按 SOUL 约定输出 [OUTLINE_READY] + 调整后的完整大纲 JSON。"""
+
 # 上下文压缩：传给大纲规划的素材摘要做条数和单条长度双重截断，防止 token 溢出。
 _MAX_OUTLINE_MATERIALS = 12
 _MATERIAL_SNIPPET_CHARS = 150
@@ -212,28 +253,40 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
    if user_outline_raw and not current_outline and not is_revise:
        return await _build_outline_from_user(state, config, pkg, user_outline_raw)

+    # 内置 agent 模式（config.yaml ai_writing.use_builtin_agents.outliner）：
+    # SOUL 作 system prompt + [OUTLINE_READY] 契约解析；目录缺失静默回退旧路径。
+    use_agent = use_builtin_agent("outliner")
+    agent_label = "作家"
+    if use_agent:
+        try:
+            load_writing_agent(OUTLINER_AGENT_ID)
+            agent_label = agent_display_name(OUTLINER_AGENT_ID, "大纲规划师")
+        except WritingAgentUnavailable as e:
+            logger.warning("outliner builtin agent unavailable, falling back to legacy prompts: %s", e)
+            use_agent = False
+
+    fmt_args = dict(
+        user_intent=state["user_intent"],
+        article_type=state.get("article_type", "analysis"),
+        target_audience=state.get("target_audience", "普通读者"),
+        word_count_target=state.get("word_count_target", 800),
+        materials_summary=_materials_summary(pkg),
+    )
    if is_revise:
        system_prompt = OUTLINE_REVISE_SYSTEM
-        user_prompt = OUTLINE_REVISE_USER.format(
-            user_intent=state["user_intent"],
-            article_type=state.get("article_type", "analysis"),
-            target_audience=state.get("target_audience", "普通读者"),
-            word_count_target=state.get("word_count_target", 800),
+        revise_args = dict(
+            fmt_args,
            current_outline=json.dumps(current_outline, ensure_ascii=False, indent=2),
            feedback=feedback,
-            materials_summary=_materials_summary(pkg),
        )
+        user_prompt = OUTLINE_REVISE_USER.format(**revise_args)
+        agent_payload = _AGENT_OUTLINE_REVISE_USER.format(**revise_args)
    else:
        system_prompt = OUTLINE_SYSTEM
        feedback_section = f"\n重新规划要求：{feedback}" if feedback else ""
-        user_prompt = OUTLINE_USER.format(
-            user_intent=state["user_intent"],
-            article_type=state.get("article_type", "analysis"),
-            target_audience=state.get("target_audience", "普通读者"),
-            word_count_target=state.get("word_count_target", 800),
-            materials_summary=_materials_summary(pkg),
-            feedback_section=feedback_section,
-        )
+        gen_args = dict(fmt_args, feedback_section=feedback_section)
+        user_prompt = OUTLINE_USER.format(**gen_args)
+        agent_payload = _AGENT_OUTLINE_USER.format(**gen_args)

    # 流式推送：阶段 2 起统一走 LangGraph 原生 custom 通道（get_stream_writer）。
    writer = safe_get_stream_writer()
@@ -242,35 +295,47 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
    parser = ThinkTagParser()

    try:
-        # 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:
-                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,
-                })
+        if use_agent:
+            def _on_text(text: str, is_thinking: bool) -> None:
+                writer({"type": "outline_chunk", "text": text, "is_thinking": is_thinking})
+
+            full_text = await stream_writing_agent(
+                agent_id=OUTLINER_AGENT_ID,
+                user_payload=agent_payload,
+                state=state,
+                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:
+                    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_name": agent_label,
                "title": "模型错误",
                "detail": str(e),
                "message": "大纲生成失败",
@@ -309,7 +374,10 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
                },
            ],
        }
-    result = parse_json(full_text, fallback)
+    if use_agent:
+        result = parse_marker_json(full_text, OUTLINE_READY_MARKER, fallback)
+    else:
+        result = parse_json(full_text, fallback)

    outline = _parse_outline(result)
    section_count = len(outline["sections"])
@@ -318,7 +386,7 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
        "current_outline": outline,
        "status": "awaiting_outline_confirm",
        "progress_events": [
-            {"type": "step_start", "agent_name": "作家", "message": "开始规划大纲"},
-            {"type": "outline_ready", "agent_name": "作家", "message": f"大纲已生成，共 {section_count} 个章节"},
+            {"type": "step_start", "agent_name": agent_label, "message": "开始规划大纲"},
+            {"type": "outline_ready", "agent_name": agent_label, "message": f"大纲已生成，共 {section_count} 个章节"},
        ],
    }