第 28 章 实战项目二：长文写作流水线

28.1 需求与场景

长文写作流水线面向深度研报、技术白皮书、营销长文、行业分析等结构化长文档的自动生产。输入为写作主题、目标读者、字数要求与风格要求，输出为 Markdown 格式的完整文档。流程跨越数分钟到数小时，包含大纲生成、用户确认、多章节并行检索、多章节并行写作、交叉审阅、反思修订、统稿七个阶段，必须具备断点续跑、并行扇出、反思循环与异步任务能力。

典型工作流：运营或研究员提交"2026 年中国 AIGC 基础设施行业研究报告"主题；系统在 30 秒内生成 6-8 章大纲并弹出确认页；用户可直接批准、逐章修改要点或拒绝重新生成；确认后系统为每一章并行启动研究任务（联网搜索 + 本地知识库检索 + 摘要压缩）；研究完成后并行启动写作任务生成各章草稿；批评节点对整篇草稿审阅，列出问题清单；若不通过则针对问题章节再次并行修订（最多 2 轮）；通过后统稿生成目录、摘要、结语与 Markdown 文件，通过 webhook 回调通知用户下载。

功能需求包括：大纲自动生成与 HITL 确认，用户可 approve/edit/reject；多章节并行资料检索（Map-Reduce Send API，扇出扇入原语）；多章节并行写作，章节间互不阻塞；Critic Agent（批评 Agent）交叉审阅并给出可执行、指向具体段落的反馈；根据反馈反思修订，最多两轮避免无限循环；全文统稿润色，统一术语、风格与引用格式；最终输出 Markdown 文件并持久化到对象存储。

非功能需求：单文档目标字数 1-3 万字；章节级并行，研究阶段 P95 耗时 5 分钟内（以 6 章计）；支持用户中断大纲并修改，断点可持续数小时甚至数天；单个章节失败重试不丢失其他章节已完成结果；LLM 成本可预测，研究阶段用小模型，写作与批评用强模型，研究结果缓存可跨任务复用；Markdown 输出风格一致、引用来源清晰、目录自动生成；支持 webhook 回调通知。

用户交互流程：前端提供输入表单（主题、读者、字数、风格），提交后显示任务进度条；大纲生成后弹出可编辑树状控件，用户可直接修改章节标题与要点、删除章节或调整顺序；修改后点击"开始写作"触发 resume；写作过程中实时推送各章节进度（研究中/写作中/审阅中/已完成）；审阅阶段可展示 critic 返回的问题清单给用户查看；完成后提供 Markdown 下载链接与预览页。

28.2 架构设计

流水线采用 Supervisor + Send Map-Reduce + HITL + 反思循环组合模式。plan_outline 生成大纲后 await_outline_approval 以 interrupt 暂停等待用户确认；确认后 route_to_research 通过 Send 原语扇出到 N 个 research_section 节点并行执行；研究结果以 operator.add 归约器自动聚合；route_to_write 再次 Send 扇出到 write_section 并行写作；所有章节完成后 join 到 critic 节点对合并草稿审阅；should_revise 判断是否进入 revise_sections，修订同样 Send 并行；达到轮次上限或审阅通过后进入 assemble 统稿并写入 final_document。

┌────────┐  submit   ┌────────────┐                                 ┌──────────────────────┐
│ Client │ ────────▶ │ Async API  │ ── enqueue ──▶ Worker ────────▶ │ Writing Pipeline     │
│        │ ◀──────── │ /webhook   │ ◀─── notify on done / await     │                      │
└────────┘           └────────────┘                                 │  START               │
                                                                   │    │                 │
                                                                   │    ▼                 │
                                                                   │ plan_outline        │
                                                                   │    │                 │
                                                                   │    ▼                 │
                                                                   │ await_outline ◀──── interrupt (HITL 用户确认大纲)
                                                                   │  _approval          │
                                                                   │    │                 │
                                                                   │    ▼                 │
                                                                   │ route_to_research ─ Send ─▶ research_section × N
                                                                   │    │ (join by operator.add)
                                                                   │    ▼                 │
                                                                   │ route_to_write ───── Send ─▶ write_section × N
                                                                   │    │ (join by operator.add)
                                                                   │    ▼                 │
                                                                   │  critic             │
                                                                   │    │                 │
                                                                   │    ▼                 │
                                                                   │ should_revise       │
                                                                   │    │pass            │
                                                                   │    ├───▶ assemble ─▶ END
                                                                   │    │revise(rev<2)    │
                                                                   │    ▼                 │
                                                                   │ revise_sections ──── Send ─▶ write_section × N (re-entry)
                                                                   │    │                 │
                                                                   │    └──▶ critic (loop)│
                                                                   └──────────────────────┘
                                                                           │
                                                         ┌────────┬────────┼────────┬────────┐
                                                         ▼        ▼        ▼        ▼        ▼
                                                    ┌────────┐┌───────┐┌────────┐┌────────┐┌───────┐
                                                    │Postgres││Tavily││Vector  ││Object  ││Lang   │
                                                    │Saver   ││Search││DB      ││Store   ││Smith  │
                                                    └────────┘└───────┘└────────┘└────────┘└───────┘

技术选型：LangGraph 1.2.x Send 原语实现动态扇出扇入；PostgresSaver 支持小时级断点与 worker 重启恢复；Tavily 作为联网搜索工具；pgvector 向量库缓存章节参考资料并做本地 RAG；研究阶段与路由判断使用 gpt-4o-mini 以控制成本；写作与批评使用 gpt-4o 保证质量；Semaphore（信号量）限制并发 LLM 调用避免触发模型供应商 rate limit；异步任务队列使用 arq（基于 Redis）或 Celery。

Send 原语关键细节：每个 Send("node_name", args_dict) 中 args_dict 是该节点接收到的 state "覆盖"——它会与父 state 合并传入节点，节点返回的更新按归约器（operator.add）合并回父 state。对于需要为每个并行分支传递局部参数（如 section）的场景，将局部参数放在 args_dict 中即可；对于需要读取聚合字段（如 research_results）的场景，节点在接收时可直接读取父 state 中已有的聚合值，LangGraph 在所有并行分支完成后才会推进下一个普通节点，所以聚合字段在 join 完成时一定包含所有分支结果。这一特性使得 N 路并行的 Map-Reduce 无需手工协调 Future 或锁。

并行度控制实践：不能仅依赖 Send 数量控制并发，因为 Send 会立即派生出 N 个任务，当 N=20 时会瞬间向 LLM 发起 20 个并发请求。正确做法是在节点内部使用 asyncio.Semaphore 作为门闩：

# utils/rate_limit.py
import asyncio
# 按供应商配额与部署 worker 数综合设置，单 worker 建议 4-8
LLM_SEMAPHORE = asyncio.Semaphore(6)
SEARCH_SEMAPHORE = asyncio.Semaphore(10)

每个调用 LLM 的位置都在 async with LLM_SEMAPHORE 内执行，超出并发的协程自动等待。多 worker 部署时，总并发 = worker 数 × Semaphore 值，需与模型供应商配额匹配；若供应商按 API Key 维度做全局限流，可引入 Redis 分布式信号量做跨 worker 限流。

项目目录结构：

writing_pipeline/
├── app/
│   ├── main.py              # FastAPI 异步入口、webhook 与审批
│   ├── worker.py            # arq worker 启动脚本
│   ├── graph.py             # 流水线图构建
│   ├── state.py             # 状态定义
│   ├── nodes/
│   │   ├── plan.py
│   │   ├── research.py
│   │   ├── writer.py
│   │   ├── critic.py
│   │   └── assemble.py
│   ├── tools/
│   │   ├── search.py        # Tavily 搜索封装
│   │   ├── vectordb.py      # 参考资料存取（pgvector）
│   │   └── objstore.py      # 对象存储读写
│   └── utils/
│       ├── llm.py           # 按阶段选择模型
│       └── rate_limit.py    # 并发信号量
├── prompts/
│   ├── plan.yaml
│   ├── research.yaml
│   ├── write.yaml
│   └── critic.yaml
├── tests/
│   ├── test_plan.py
│   ├── test_send_fanout.py
│   ├── test_critique_loop.py
│   └── test_hitl_resume.py
├── docker-compose.yml
├── requirements.txt
└── .env

依赖安装命令：

pip install "langgraph>=1.2,<1.3" langchain langchain-openai tavily-python \
    psycopg[binary,pool] pgvector fastapi uvicorn httpx pydantic pyyaml \
    arq redis python-multipart

Send 扇出扇入机制说明：当条件边返回的不是字符串而是 Send 对象列表时，LangGraph 会将每个 Send 作为独立任务并行派发到目标节点，每个任务携带独立的 state 片段（由 Send 的 args 指定）；并行分支全部完成后，框架按归约器（此处为 operator.add）将各分支对同一字段的更新合并回父 state，自动完成 join。这一模式是长文并行写作的关键，无需手工写 asyncio.gather 或线程池，所有并行分支的 checkpoint、错误重试、流式事件都由框架统一管理。并行度通过 Semaphore 而非 Send 数量控制，因为 Send 默认会立即派发所有分支。

模型分层策略：plan_outline 阶段决定整体结构，使用强模型 gpt-4o；research_section 阶段主要做检索结果压缩，使用 gpt-4o-mini；write_section 阶段是质量核心，使用 gpt-4o；critic 阶段需要精确发现问题，使用 gpt-4o；assemble 阶段做合并润色，使用 gpt-4o；路由节点（route_to_research/route_to_write/should_revise）是纯 Python 逻辑不调用 LLM，开销为零。

28.3 State 与节点设计

# state.py
from typing import Annotated, List, Optional
import operator
from typing_extensions import TypedDict
from pydantic import BaseModel, Field

class Section(BaseModel):
    """章节结构。"""
    id: int
    title: str
    key_points: List[str]
    word_target: int = Field(default=1500, description="目标字数")

class WritingState(TypedDict):
    topic: str
    audience: str                  # 目标读者
    style: str                     # 风格要求
    word_count_target: int
    outline: List[dict]            # 用户确认后的 Section 列表
    research_results: Annotated[list, operator.add]   # 各章节资料
    section_drafts: Annotated[list, operator.add]     # 各章节草稿
    critique_feedback: str
    revision_count: int
    final_document: str
    user_id: str
    task_id: str
    _revised_sections: Optional[List[int]]  # critic 标记需修改的章节 id

节点职责如下：

• plan_outline：使用大模型根据 topic、audience、style、word_count_target 生成 Section 列表（标题 + 要点 + 每章字数目标）。
• await_outline_approval：调用 interrupt 将 outline 抛给前端；用户可 approve、edit（修改后确认）或 reject（回到 plan_outline 重新生成）。
• route_to_research：读取 outline 为每个 section 生成 Send("research_section", {"section": s})，返回 Send 列表即触发并行扇出。
• research_section：对单个章节做联网检索（Tavily）+ 本地向量库 RAG，将检索结果用小模型压缩为结构化资料，返回 {"research_results": [...]}。
• route_to_write：按 outline 顺序为每个 section 生成 Send("write_section", {"section": s, "revision":False})，携带已聚合的研究资料。
• write_section：读取 research_results 中与 section.id 匹配的资料撰写 Markdown 草稿；若为修订轮次（revision=True）则附加 critique_feedback 针对性修改。
• critic：将所有 section_drafts 按 id 排序合并后交给批评模型，输出结构化问题清单（精确到 section_id 与问题类型）与整体评分。
• should_revise：revision_count < 2 且存在问题章节时路由至 revise_sections；否则到 assemble。
• revise_sections：根据 critique_feedback 生成 Send 列表，将需要修改的章节再次发送到 write_section（revision=True），revision_count 自增。
• assemble：按 section.id 排序合并草稿，生成标题、目录、摘要、结语；统一 Markdown 风格、术语与引用标号；写入 final_document 并上传至对象存储。

核心图构建：

# graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from .state import WritingState
from .nodes.plan import plan_outline, await_outline_approval
from .nodes.research import route_to_research, research_section
from .nodes.writer import route_to_write, write_section, revise_sections
from .nodes.critic import critic, should_revise
from .nodes.assemble import assemble

def _route_revise(state: WritingState):
    revised = state.get("_revised_sections") or []
    if not revised:
        return "critic"
    return [Send("write_section",
                 {"section": next(s for s in state["outline"] if s["id"]==sid),
                  "revision": True})
            for sid in revised]

builder = StateGraph(WritingState)
builder.add_node("plan_outline", plan_outline)
builder.add_node("await_outline_approval", await_outline_approval)
builder.add_node("research_section", research_section)
builder.add_node("write_section", write_section)
builder.add_node("critic", critic)
builder.add_node("revise_sections", revise_sections)
builder.add_node("assemble", assemble)

builder.add_edge(START, "plan_outline")
builder.add_edge("plan_outline", "await_outline_approval")
builder.add_conditional_edges("await_outline_approval", route_to_research,
                              path_map=["research_section"])
# research_section 所有并行分支完成后，LangGraph 自动 join，
# 触发下一个条件边 route_to_write
builder.add_conditional_edges("research_section", route_to_write,
                              path_map=["write_section"])
builder.add_edge("write_section", "critic")
builder.add_conditional_edges("critic", should_revise,
                              path_map={"revise": "revise_sections",
                                        "assemble": "assemble"})
builder.add_conditional_edges("revise_sections", _route_revise,
                              path_map={"write_section": "write_section",
                                        "critic": "critic"})
builder.add_edge("assemble", END)

plan_outline 与 await_outline_approval 节点示例：

# nodes/plan.py
from langchain.chat_models import init_chat_model
from langgraph.types import interrupt, Command
from ..state import Section

async def plan_outline(state: WritingState):
    llm = init_chat_model("openai:gpt-4o").with_structured_output(list[Section])
    sections = await llm.ainvoke([
        {"role":"system","content":"该角色为资深报告策划，根据主题、目标读者、字数目标生成 5-8 章的报告大纲，每章给出标题、3-5 个关键要点与字数目标。"},
        {"role":"user","content":f"主题：{state['topic']}\n目标读者：{state['audience']}\n风格：{state['style']}\n总字数目标：{state['word_count_target']}"}
    ])
    return {"outline": [s.model_dump() for s in sections],
            "revision_count": 0,
            "research_results": [],
            "section_drafts": [],
            "critique_feedback": "",
            "_revised_sections": None}

async def await_outline_approval(state: WritingState):
    decision = interrupt({"type":"outline_approval",
                          "outline": state["outline"],
                          "topic": state["topic"],
                          "awaiting": "user_outline_decision"})
    # decision: {"action":"approve"|"edit"|"reject", "outline":[...]}
    if decision["action"] == "reject":
        return Command(goto="plan_outline", update={})
    new_outline = decision.get("outline") or state["outline"]
    return {"outline": new_outline}

28.4 并行、成本与错误处理

research_section 节点接入信号量控制并发，避免瞬间发起过多 LLM 调用触发 429：

# utils/rate_limit.py
import asyncio
llm_semaphore = asyncio.Semaphore(4)   # 全局并发 LLM 调用数
search_semaphore = asyncio.Semaphore(8)

# nodes/research.py
import json
from langchain.chat_models import init_chat_model
from ..tools.search import tavily_search
from ..utils.rate_limit import llm_semaphore, search_semaphore
from ..utils.llm import get_model  # 按阶段返回 mini 或 4o

async def research_section(state: WritingState, section: dict):
    writer = get_stream_writer()
    writer({"event":"research_start","section_id":section["id"],"title":section["title"]})
    async with search_semaphore:
        query = f"{state['topic']} {section['title']} {' '.join(section['key_points'])}"
        results = await tavily_search.ainvoke(query, max_results=8)
    async with llm_semaphore:
        mini = get_model("mini")
        summary = await mini.ainvoke(
            f"请基于以下检索结果整理为章节《{section['title']}》的事实材料摘要，保留关键数据、来源 URL、案例与对比信息，用要点形式输出：\n{json.dumps(results, ensure_ascii=False)[:12000]}")
    writer({"event":"research_end","section_id":section["id"]})
    return {"research_results": [{
        "section_id": section["id"],
        "materials": summary.content,
        "source_count": len(results) if isinstance(results, list) else 0}]}

def route_to_research(state: WritingState):
    return [Send("research_section", {"section": s}) for s in state["outline"]]

write_section 节点根据是否为修订轮次附加批评反馈：

# nodes/writer.py
from langchain.chat_models import init_chat_model
from ..utils.llm import get_model
from ..utils.rate_limit import llm_semaphore

async def write_section(state: WritingState, section: dict, revision: bool = False):
    writer = get_stream_writer()
    writer({"event":"write_start","section_id":section["id"],"title":section["title"]})
    # 从 research_results 中取对应章节的资料
    materials = "\n\n".join(
        r["materials"] for r in state["research_results"]
        if r["section_id"] == section["id"]) or "（无检索资料）"
    system = f"该角色为专业报告写手，请撰写章节《{section['title']}》，目标字数约 {section.get('word_target',1500)} 字，输出 Markdown。\n要点：{ '; '.join(section['key_points']) }"
    user_blocks = [f"参考资料：\n{materials}"]
    if revision and state.get("critique_feedback"):
        user_blocks.append(f"上一轮批评反馈（请针对性修改本章节）：\n{state['critique_feedback']}")
    async with llm_semaphore:
        llm = get_model("strong")
        draft = await llm.ainvoke([
            {"role":"system","content":system},
            {"role":"user","content":"\n\n".join(user_blocks)}])
    writer({"event":"write_end","section_id":section["id"]})
    return {"section_drafts": [{
        "section_id": section["id"],
        "section_title": section["title"],
        "content": draft.content}]}

def route_to_write(state: WritingState):
    return [Send("write_section", {"section": s, "revision": False})
            for s in state["outline"]]

critic 节点返回结构化反馈，避免空泛评价：

# nodes/critic.py
from pydantic import BaseModel
from langchain.chat_models import init_chat_model
from ..utils.llm import get_model

class CritiqueIssue(BaseModel):
    section_id: int
    issue_type: str   # fact/structure/argument/expression/redundancy
    issue: str
    suggestion: str

class Critique(BaseModel):
    needs_revision: bool
    issues: list[CritiqueIssue]
    overall_score: float = Field(ge=0, le=10)

async def critic(state: WritingState):
    ordered = sorted(state["section_drafts"], key=lambda x: x["section_id"])
    joined = "\n\n".join(
        f"## {d['section_title']}\n{d['content']}" for d in ordered)
    llm = get_model("strong").with_structured_output(Critique)
    c: Critique = await llm.ainvoke([
        {"role":"system","content":"该角色为严格的技术写作审阅者，指出事实错误、结构问题、论证缺失、表达冗余等具体问题，每条问题必须给出修改建议，避免空泛评价。"},
        {"role":"user","content":f"请审阅《{state['topic']}》草稿：\n{joined[:30000]}"}])
    revised_ids = [i.section_id for i in c.issues] if c.needs_revision else []
    return {"critique_feedback": c.model_dump_json(indent=2),
            "_revised_sections": revised_ids}

def should_revise(state):
    if state["revision_count"] >= 2:
        return "assemble"
    return "revise" if state.get("_revised_sections") else "assemble"

async def revise_sections(state):
    return {"revision_count": state["revision_count"] + 1}

assemble 节点负责合并、润色、持久化：

# nodes/assemble.py
from langchain.chat_models import init_chat_model
from ..tools.objstore import upload_text
from ..utils.llm import get_model

async def assemble(state: WritingState):
    ordered = sorted(state["section_drafts"], key=lambda x: x["section_id"])
    body = "\n\n".join(
        f"## {d['section_title']}\n\n{d['content']}" for d in ordered)
    toc = "\n".join(
        f"- {i+1}. {d['section_title']}" for i,d in enumerate(ordered))
    # 生成摘要
    llm = get_model("mini")
    abstract = await llm.ainvoke(
        f"为以下报告撰写 200 字以内中文摘要：\n{body[:8000]}")
    doc = (f"# {state['topic']}\n\n"
           f"> 摘要：{abstract.content}\n\n"
           f"## 目录\n{toc}\n\n{body}\n")
    obj_key = f"docs/{state['task_id']}.md"
    await upload_text(obj_key, doc)
    return {"final_document": doc}

持久化、成本控制与错误处理：长流程必须使用 PostgresSaver，防止 worker 重启或用户长时间不确认导致状态丢失。thread_id 按任务维度设为 write:{task_id}；Store 以 ("writing_cache", topic_hash, section_id) 为 key 缓存章节研究结果（TTL 7 天），相同主题与章节的重复任务可直接复用缓存以节省成本；向量库存储历史参考资料 embedding，检索时优先本地库再决定是否调用 Tavily。

成本控制三层：模型分层（研究/摘要/路由用 gpt-4o-mini，写作/批评/统稿用 gpt-4o）；缓存复用（research_results 写入 Store 与向量库，命中时跳过 research_section）；max_tokens 限制（研究摘要 800 token、章节草稿 2500 token、批评 1200 token、统稿 2000 token）。提供 cost_estimate 节点（可选）在大纲确认前估算 token 与美元成本，给用户确认后再启动写作。

错误处理：对 research_section 与 write_section 节点在 add_node 时配置 retry=2，单个章节失败自动重试；Tavily 检索为空时 materials 置空字符串并标记 no_source:True，write_section 降级为无参考写作并在段首以 Markdown 斜体标注"本章无检索资料，内容由模型基于常识生成"；critic 输出结构不符合 Pydantic 时自动重试一次，仍失败则跳过批评直接统稿；整体任务失败时通过 webhook 回调推送失败状态与失败节点名，checkpoint 保留以便人工介入 resume；长时间无 HITL 响应（如 24 小时未审批大纲）通过定时任务标记任务为 timeout，前端提醒用户。

反思循环的有效性依赖 critic 反馈的可执行性。critic 提示词中必须要求模型"给出 section_id、问题类型、具体问题、修改建议"四元组，避免"论证不够充分""语言不够流畅"等空泛评价；writer 节点在修订轮次时将 critique_feedback 按 section_id 过滤后注入，只给当前章节需要的反馈，避免 prompt 被无关反馈淹没；revision_count 熔断是硬性保护，避免 critic 与 writer 反复拉锯导致任务永不结束。

vectordb 工具实现示例：

# tools/vectordb.py
from pgvector.psycopg import register_vector
import psycopg
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

async def index_materials(task_id: str, section_id: int, text: str):
    """将章节资料写入 pgvector 供未来相似主题复用。"""
    vec = await embeddings.aembed_query(text[:8000])
    with psycopg.connect("postgresql://user:pwd@pg:5432/vectors") as conn:
        register_vector(conn)
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO writing_materials (task_id, section_id, embedding, content) VALUES (%s,%s,%s,%s)",
                (task_id, section_id, vec, text[:50000]))
            conn.commit()

async def recall_similar(query: str, k: int = 3):
    """召回与当前章节相似的历史资料。"""
    vec = await embeddings.aembed_query(query)
    with psycopg.connect("postgresql://user:pwd@pg:5432/vectors") as conn:
        register_vector(conn)
        with conn.cursor() as cur:
            cur.execute(
                "SELECT content FROM writing_materials ORDER BY embedding <=> %s LIMIT %s",
                (vec, k))
            return [r[0] for r in cur.fetchall()]

对象存储工具示例：

# tools/objstore.py
import boto3
from botocore.config import Config
s3 = boto3.client("s3", endpoint_url="http://minio:9000",
                  aws_access_key_id="minio", aws_secret_access_key="minio123",
                  config=Config(signature_version="s3v4"))
BUCKET = "writing-docs"

async def upload_text(key: str, text: str):
    s3.put_object(Bucket=BUCKET, Key=key, Body=text.encode("utf-8"),
                  ContentType="text/markdown")
    return f"s3://{BUCKET}/{key}"

Tavily 搜索封装：

# tools/search.py
from langchain_community.tools.tavily_search import TavilySearchResults
tavily_search = TavilySearchResults(max_results=8, search_depth="advanced")

llm 选择工具：

# utils/llm.py
from langchain.chat_models import init_chat_model
def get_model(tier: Literal["mini","strong"]):
    name = "openai:gpt-4o-mini" if tier == "mini" else "openai:gpt-4o"
    return init_chat_model(name, temperature=0.3 if tier=="strong" else 0.1)

arq worker 配置示例：

# worker.py
from arq import create_pool
from arq.connections import RedisSettings
from .graph import graph
import asyncio, json

async def run_pipeline_task(ctx, task_id: str, payload: dict):
    config = {"configurable": {"thread_id": f"write:{task_id}"},
              "recursion_limit": 200}
    try:
        async for ev in graph.astream(
            {"task_id": task_id, **payload, "revision_count":0},
            config=config, stream_mode=["custom","updates"]):
            # 将进度事件推到 Redis Pub/Sub 供前端订阅
            redis = ctx["redis"]
            await redis.publish(f"task:{task_id}", json.dumps({"mode":ev[0],"preview":str(ev[1])[:500]}, ensure_ascii=False))
        final = await graph.aget_state(config)
        await ctx["redis"].set(f"task_result:{task_id}",
                               final.values.get("final_document","")[:200000], ex=7*86400)
    except Exception as e:
        await ctx["redis"].set(f"task_error:{task_id}", str(e), ex=7*86400)

class WorkerSettings:
    redis_settings = RedisSettings(host="redis", port=6379)
    functions = [run_pipeline_task]
    max_jobs = 20  # 单 worker 并发任务数
    timeout = 3600  # 单任务超时 1 小时

写作质量控制要点：write_section 提示词中应明确要求"基于参考资料撰写，不确定的信息标注来源或声明存疑""使用 Markdown 标题层级，数据用表格呈现，对比用列表""章节内包含引言、主体、小结三部分"；assemble 阶段统一术语（例如整篇文档统一使用"大语言模型"而非混用"大模型/LLM/语言模型"），统一数字格式（千分位、百分比保留位数）、统一链接格式与引用上标；assemble 节点可调用一次强模型做"全文润色"，但需注意不要覆盖章节专业性。

HITL 扩展点——草稿审阅：在 critic 与 assemble 之间可增设 await_draft_approval 节点，将草稿（outline + section_drafts 合并的预览）通过 interrupt 抛出给用户，用户可 approve/request_changes；request_changes 时用户反馈作为额外反馈注入下一轮 writer，revision_count 仍受 2 轮上限保护。该扩展适合对质量要求极高的研报场景，但会增加一次人工等待，一般营销长文可省略。

Send 故障隔离：单个 research_section 或 write_section 节点若抛出未捕获异常，LangGraph 默认会中断整个图。为避免一个章节失败导致整篇文档作废，在这些节点外层用 try/except 包裹，失败时返回标记 {"section_drafts": [{"section_id":id, "section_title":title, "content":"> 本章写作失败，已在统稿时标记", "error": str(e)}]}，assemble 阶段检测 error 字段并在该章位置写入红色提示块，其他章节正常输出，整体任务仍然完成。若业务要求必须所有章节成功才算完成，则在 assemble 前增加 check_all_sections_ok 节点，存在 error 字段时路由到 retry_failed_sections 节点，仅对失败章节 Send 重试一次。

主题变体与模板化：对经常生产的文档类型（如季度行业报告、产品白皮书），可预定义 outline_template，plan_outline 检测到主题匹配模板时直接套用模板结构，减少大纲生成的 token 消耗与不稳定性；模板中可固定章节标题、要点、字数目标，用户仅能微调不可删除核心章节，提升批量生产效率。

参考资料引用格式：research_section 摘要材料中保留来源 URL，write_section 在引用数据处以上标 [1][2] 形式标注，assemble 阶段在文末生成"参考资料"章节汇总所有 URL，保证可溯源性；对无法追溯来源的数据统一标注"模型估计"或"据公开资料整理"，避免误导读者。

输入校验与任务拒绝：plan_outline 前应增加 validate_topic 节点，检测主题是否涉及违规内容（医疗诊断建议、金融荐股、违法违规指导等），命中时直接返回拒绝文案并不启动写作，避免生成合规风险内容；校验通过后再进入 plan_outline。对用户输入字数设限（主题 ≤ 100 字，要点描述 ≤ 2000 字），防止滥用长输入消耗大量 token。

进度事件规范：每个节点通过 get_stream_writer() 推送结构化进度事件，字段统一为 {event, section_id?, title?, message?}，前端根据 event 渲染进度条与状态标签；典型事件包括 outline_generated、awaiting_approval、research_start/end、write_start/end、critique_start/end、assemble_start、task_done、task_failed；对于 revise 轮次在事件中带 revision_round 字段，方便前端展示"第 N 轮审阅修订"。

多格式输出扩展：assemble 节点生成 Markdown 后，可扩展 pandoc_converter 节点将 Markdown 转换为 DOCX/PDF/HTML；转换为 PDF 时注意中文字体配置（Noto Sans CJK）；转换后的二进制文件上传对象存储，最终响应包含各格式下载链接。该节点不修改 final_document 字段，只是附加派生字段，避免污染主文档。

写作风格控制：prompts/write.yaml 中按 style 字段（专业严谨/营销活泼/学术严谨/简洁纪要）动态加载不同的写作指令，并在 assemble 阶段统一检查口吻一致性；可通过 configurable 传入 writing_tone 与 reading_level（专家/入门），生成不同深度的版本供不同读者使用。

参考资料去重：多个章节可能检索到同一来源 URL，assemble 生成"参考资料"章节时按 URL 去重并统一编号，避免同一链接重复出现；对检索结果中的内容重复段，在 research_section 摘要阶段通过小模型做去重压缩，减少 token 浪费与冗余信息干扰。

成本监控：每次 LLM 调用后从 response metadata 中读取 token 使用量（prompt_tokens、completion_tokens），按模型单价实时计算成本写入 metrics；对单任务设置预算上限（如 2 美元），超预算可中止流程并通知用户；成本数据同时写入审计库，按租户/任务/模型维度做日报。

28.5 部署与测试

部署采用异步 API + 后台 worker 模式。前端 POST /tasks 提交主题立即返回 task_id 与 run_id；任务入 Redis 队列，arq worker 拉取执行；需要审批（大纲确认、可选的草稿审阅）时 interrupt 暂停，前端通过 SSE /tasks/{id}/stream 订阅进度事件；审批通过 /tasks/approve 提交 Command(resume=...)；完成或失败通过 webhook 回调业务系统。

# main.py 核心片段
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langgraph.types import Command
import uuid, json, asyncio
import httpx
from .graph import graph
from .state import WritingState

app = FastAPI()
jobs = {}   # 生产环境用 Redis Hash 存状态

class TaskIn(BaseModel):
    topic: str
    audience: str = "行业从业者"
    style: str = "专业严谨"
    word_count_target: int = 15000
    user_id: str
    webhook: str | None = None

class ApproveIn(BaseModel):
    task_id: str
    action: str
    outline: list | None = None

@app.post("/tasks")
async def create_task(t: TaskIn):
    task_id = str(uuid.uuid4())
    jobs[task_id] = {"status":"queued"}
    # 异步启动执行（生产环境应推入 arq 队列）
    asyncio.create_task(run_pipeline(task_id, t.model_dump()))
    return {"task_id": task_id}

async def run_pipeline(task_id: str, payload: dict):
    config = {"configurable": {"thread_id": f"write:{task_id}"},
              "recursion_limit": 200}
    jobs[task_id]["status"] = "running"
    try:
        async for ev in graph.astream(
            {"task_id": task_id, **payload,
             "revision_count":0},
            config=config,
            stream_mode=["custom","updates"]):
            mode, data = ev
            if mode == "custom" and data.get("type") == "interrupt":
                jobs[task_id]["status"] = "awaiting_input"
                jobs[task_id]["awaiting"] = data
        final = await graph.aget_state(config)
        jobs[task_id]["status"] = "done"
        jobs[task_id]["doc_preview"] = final.values.get("final_document","")[:500]
        if payload.get("webhook"):
            async with httpx.AsyncClient(timeout=10) as c:
                await c.post(payload["webhook"],
                             json={"task_id":task_id,"status":"done"})
    except Exception as e:
        jobs[task_id]["status"] = "failed"
        jobs[task_id]["error"] = str(e)

@app.post("/tasks/approve")
async def approve(inp: ApproveIn):
    config = {"configurable":{"thread_id":f"write:{inp.task_id}"}}
    # 注入 resume 值，图从 interrupt 断点继续
    asyncio.create_task(graph.astream(
        Command(resume={"action":inp.action,"outline":inp.outline}),
        config=config, stream_mode="updates"))
    return {"status":"resumed"}

@app.get("/tasks/{task_id}/stream")
async def task_stream(task_id: str):
    async def gen():
        config = {"configurable":{"thread_id":f"write:{task_id}"}}
        async for mode, payload in graph.astream(
                None, config=config, stream_mode=["custom","messages"]):
            yield f"data: {json.dumps({'mode':mode,'payload':str(payload)[:500]}, ensure_ascii=False)}\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

运维与容量：arq worker 进程数按 CPU 核数配置，每个 worker 内 Semaphore(4) 控制 LLM 并发；Redis 用于队列与任务进度 Pub/Sub；Postgres 存储 checkpoint 时需调大 work_mem 与 max_wal_size，因为长任务写入的 checkpoint 是 JSONB 大字段；Tavily QPS 按账户配额限制，超出则排队等待。监控指标：wp_tasks_total{status}、wp_section_duration_seconds{stage}、wp_llm_calls_total{model}、wp_token_cost_total{model}、wp_hitl_wait_seconds、wp_revision_rounds、wp_failed_sections_total。关键告警：任务失败率 > 5%、Tavily 错误率 > 10%、单章节写作时长 P99 > 5 分钟、HITL 等待超过 24 小时。

docker-compose 核心配置参考：

version: "3.9"
services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
    environment:
      - DATABASE_URL=postgresql://wp:wp@pg:5432/wp
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - TAVILY_API_KEY=${TAVILY_API_KEY}
    depends_on: [pg, redis]
  worker:
    build: .
    command: arq app.worker.WorkerSettings
    deploy:
      replicas: 4
    environment:
      - DATABASE_URL=postgresql://wp:wp@pg:5432/wp
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - TAVILY_API_KEY=${TAVILY_API_KEY}
    depends_on: [pg, redis]
  pg:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: wp; POSTGRES_PASSWORD: wp; POSTGRES_DB: wp
    volumes: ["wp_pg:/var/lib/postgresql/data"]
  redis:
    image: redis:7
volumes:
  wp_pg:

长任务断点续跑：worker 启动时先从 PostgresSaver 读取所有处于 interrupted 或 running 状态的任务，将 running 任务重新入队（因为这些任务是在 worker 崩溃前未完成的），避免任务永远挂起；用户在大纲确认页停留数天后再点击确认，前端调用 /tasks/approve 即可，框架自动从 checkpoint 恢复后续所有节点状态，包括已完成的章节研究结果，不会重跑。

反思循环有效性的评估指标：每轮 critic 的 issue 数量应递减（例如第 0 轮 12 条、第 1 轮 5 条、第 2 轮 1 条），若两轮 issue 数量无明显下降说明 writer 未正确响应反馈，应检查 prompt 中批评反馈的注入方式或切换更强模型；通过 LangSmith 数据集离线评估 critic 的 issue 检出准确率（人工标注的错误是否被 critic 识别）。

测试方案：单元测试覆盖 plan_outline 输出 Section 列表合法性、critic 返回 Critique Pydantic 对象、assemble 目录顺序与正文章节数一致、Send 返回列表长度等于 outline 长度；集成测试以"LangGraph 1.2 技术白皮书"为标准输入，mocking Tavily 与 LLM 使用固定语料响应，端到端跑通大纲确认→并行研究→并行写作→批评→修订→统稿流程，断言 final_document 字数区间、章节数、HITL 断点类型为 outline_approval、revision_count 在 0-2 之间；并发测试 10 个任务并行提交，验证 Semaphore 生效、Postgres checkpoint 不出现 thread_id 冲突、任务间不串稿；成本审计测试统计单文档 token 消耗并与预算阈值对比；失败注入测试模拟 Tavily 超时、单章节 LLM 失败、critic 格式错误、用户 reject 大纲，断言重试触发、降级标记生效、回到 plan_outline 重生成。