1. 一次生产需求：为什么“单个模型一把梭”总在翻车
1.1 一个典型任务：从素材到成稿的全过程
我之前接到过一个任务：基于一堆分散的资料，输出一篇有观点、有结构、能落地的行业分析稿。任务看起来不复杂，但拆开之后其实是四件差异很大的事——先要把资料里的核心事实捞出来，这是"研究"；再从材料中提炼出可论证的论点，这是"分析"；接着把论点扩展成完整文章，这是"写作"；最后从头到尾检查一遍，看数据是否准确、观点是否自洽，这是"审核"。
起初我的做法和大多数人的第一反应一样：写一段超长的系统提示词，把资料全部塞进去，让大模型一口气生成全文。这样的确省事，但产出质量非常不稳定。后来我意识到，这不是某一个大模型能力不够，而是任务结构本身有问题。单一调用要同时扮演研究员、分析师、写作者、审核者四个角色，上下文越长，模型在后面的行为就越不可控。
1.2 三个死穴：幻觉、丢上下文、无人复核
先说我踩到的三个具体问题。
第一个是幻觉。资料里没有的数据，模型会基于"最可能的数值"脑补出来。如果只有一步调用，你很难在生成后逐条核对引用来源，因为这些"来源"可能压根不存在。第二个是上下文丢失。把二十份材料拼成一份几万字的上下文，模型在写作时最常引用的往往是开头和结尾那几段，中间的关键信息大量被忽略。第三个是没有复核闭环。单次调用结束就输出成品，没有任何一道关卡检查它是否跑偏，错误被闷声带到终稿里。
这三个问题其实指向同一个解法：把任务拆开，让每个环节只做一件事，并且在环节之间传递结构化产物，最后加一个"质量门"来决定是否返工。这正是多智能体协作的价值所在。
1.3 多智能体不是炫技，而是把流程固化进代码
我见过不少团队一提"多智能体"就觉得是给模型套不同的人设，再让它们互相聊天。但真正可落地的多智能体，强调的是流程编排，不是角色扮演。你需要把"研究-分析-写作-审核"当作一条流水线，每个节点有明确的输入、输出和退出条件，节点之间通过状态对象传递数据。
LangGraph 恰好是干这件事的工具。它是 LangChain 生态里的图编排框架，允许你定义节点、连线、条件分支，甚至支持从后往前回环重跑。这正好解决了上面提到的质量问题：审核不通过，就自动把文章退回给写作节点重写，而不是人工复制粘贴再催一遍。下面我先把这套框架的核心机制讲清楚，再逐节点拆解实现。
2. LangGraph的编排心智：StateGraph就是一张可以回头的流程图
2.1 状态（State）：所有节点共同维护一块黑板
LangGraph 建立在你对"状态"的理解之上。所谓状态，就是一个可共享的数据结构，通常用TypedDict定义，它像一块黑板：每个节点负责往黑板上写一部分内容，下一个节点从黑板上读自己需要的内容，写完之后再放回去。
我习惯把状态字段分成两类：一类是"单值覆盖型"，比如最终草稿draft，每次写作节点一执行就直接覆盖；另一类是"累积型"，比如审核意见review_history，每次审核完都要追加一条，不能覆盖丢历史。这个区别在后面的实现里特别容易出错，我会在第 5 节专门展开。
黑板的粒度决定了系统的复杂度。如果你把所有中间产物都塞进状态里，模型会越来越迷茫；如果字段设计得太细碎，节点之间又会被迫做大量数据拼装。我的经验是：状态的字段数量控制在 6 到 8 个，每个字段承载一个明确的阶段产物或控制变量。
2.2 节点与边：函数即节点，条件函数决定流向
LangGraph 里，一个节点就是一个普通 Python 函数。它的输入是当前状态字典，返回值是一个部分状态字典，LangGraph 会自动做合并。你不需要手动维护全局变量，框架帮你完成状态传递。
节点之间的连接方式有两种：
add_edge：无条件连线，从 A 执行完必然到 B。add_conditional_edge：条件连线，执行完当前节点后，根据一个返回字符串的函数跳到不同分支。
条件边是实现"审核不通过就重写"的关键。审核节点执行完，程序读取审核结果，如果passed为 True 就走结束分支，否则就回到写作节点。这一步让整个图变成了真正的流程闭环，而不只是从左到右的单向管道。
2.3 为什么Graph形态比Chain更适合生产内容
如果你用过 LangChain 的传统链式调用，你会发现链也是按顺序执行的，那为什么不直接用链？核心区别就是"能不能回头"。传统链是一个有向无环图，写完就结束了；而 LangGraph 允许条件边把执行流指回之前的节点，形成循环。
在这个团队场景里，循环就是质量保证。没有循环，审核者只是装饰品；有了循环，审核意见才能真正驱动写作者改进。而且 LangGraph 本身就是用图数据结构来管理这种复杂流程的，中间态可以随时检查，这对排查"是哪一步产生了坏输出"非常重要。
下面这段代码是一个最小可运行的状态图框架，我用它来展示整体结构：
from typing import TypedDict, Annotated, operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): task: str # 用户原始任务 research_results: list # 研究员输出：材料条目 analysis: dict # 分析师输出：论点骨架 draft: str # 写作者输出：文章草稿 review_result: dict # 审核者输出：判决结果 rewrite_times: int # 已重写次数 review_history: Annotated[list, operator.add] # 审核意见累积追加 graph = StateGraph(AgentState) graph.add_node("researcher", researcher_node) graph.add_node("analyst", analyst_node) graph.add_node("writer", writer_node) graph.add_node("reviewer", reviewer_node) graph.add_edge(START, "researcher") graph.add_edge("researcher", "analyst") graph.add_edge("analyst", "writer") graph.add_edge("writer", "reviewer") graph.add_conditional_edge("reviewer", decide_next, {"rewrite": "writer", "end": END}) app = graph.compile()decide_next就是条件函数，它读取当前状态里的review_result和rewrite_times，返回"rewrite"或"end"。四个节点函数的具体实现，接下来逐个拆解。
3. 研究员与分析师：从“搜集资料”到“知识骨架”的两级蒸馏
3.1 研究节点的Prompt与输出约束
研究节点是整个团队的信息入口。它的任务是：理解用户任务，执行检索/资料查询，然后把外部信息整理成结构化条目。
我在设计研究节点时，最看重的是"可追溯性"。每个输出条目必须包含来源说明，否则下一个节点就可能拿着来路不明的信息去写文章。下面是我常用的输出结构：
{ "title": "条目标题", "url": "来源地址或路径", "summary": "一到两句话的核心内容", "key_points": ["关键要点1", "关键要点2"], "source_label": "已验证 / 待核实" }source_label是我后来加上的字段。让模型自己承认某条信息"待核实"，比让它假装一切都可靠要实用得多。分析节点看到"待核实"标签，会自动降低这条材料的优先级，写作节点也不会轻易引用它。
研究节点的系统提示词里，我会反复强调一句话："只输出你实际获得的内容，不要推断、不要补全、不要为了让报告好看而编造来源。" 模型在单次调用中一般不会主动撒谎，但如果提示词里没有这条约束，它就会把"看起来合理"当成"肯定是这样"。
3.2 分析节点的“二级蒸馏”逻辑
如果说研究节点做的是"一级蒸馏"——把原始资料提炼成条目，那么分析节点就是"二级蒸馏"——把条目提炼成可以用来写作的论点骨架。
分析节点的输入有三个部分：研究条目列表、用户的原始任务、以及一个隐含的"目标读者"字段。它的输出我用一个analysis字典承接，字段设计如下：
{ "thesis": "文章的核心论点", "target_audience": "目标读者画像", "arguments": [ {"topic": "论点一", "stance": "对该论点的态度"}, {"topic": "论点二", "stance": "对该论点的态度"} ], "evidence_mapping": {"论点一": [0, 2], "论点二": [1, 3]} }evidence_mapping是我觉得最值得保留的设计。它把论点与研究条目的索引号对应起来，比如"论点一"用第 0 条和第 2 条材料支撑。这相当于给写作节点画了一张引用地图：写某个段落时，只需要看对应的那几条材料，而不是重新翻一遍全部资料。
这个设计也解决了第 1 节说到的"上下文丢失"问题。写作节点不再需要把二十份材料全部塞进上下文，它只需要拿到论点骨架和对应的少量支撑材料。上下文变短，模型的注意力自然更集中。
3.3 两级中间产物的字段设计
研究节点和分析节点的字段设计，直接影响后续写作和审核的质量。我总结出三个原则：
第一，所有中间产物都要有明确的消费方。研究节点的条目是给分析节点看的，分析节点的骨架是给写作节点看的。如果某个字段没有任何节点会用到，就不要让它出现在输出里。第二，尽量用列表和映射结构，而不是大段长文本。长文本对模型来说仍然是一团需要重新理解的内容，而列表和键值对可以精准定位。第三，预留"置信度"或"待核实"之类的元信息字段，这能让后面的质量门有据可依。
我早期犯过一个错：让分析节点输出一篇完整的"半成品文章"作为中间产物。结果写作节点拿到之后，在它的基础上做了大篇幅的改写，两个版本的观点发生了冲突。后来我把中间产物严格限定为论点骨架，不让它成文，才把"写什么"和"怎么写"这两件事彻底分开。
4. 写作者与审核者：生成结束只是开始，质量门才是闭环
4.1 写作节点如何避免“再创作”
写作节点的职责非常单纯：拿到分析骨架和对应材料，逐段扩展成完整文章。它的提示词里我通常会写三条规定：
- 只能基于分析骨架中的论点展开，不能新增论点。
- 只能引用
analysis["evidence_mapping"]指定的材料，不能引用其他内容。 - 不允许出现骨架中没有的数字和结论，如果材料缺失就明确写"此处需补充资料"。
之所以要这样约束，是因为写作节点的核心价值是"表达"，而不是"创作"。一旦让它自由发挥，模型很可能会在行文中悄悄塞入新的推测，这就是幻觉重回系统的入口。
写作者的温度参数我一般设置在 0.7 左右，让它有足够的语言组织自由度，但又不至于发散到脱离骨架。如果一个模型在写作时表现出明显的内容重复，我会把温度降到 0.5，或者检查是不是上下文里重复出现了同一篇资料。
另外，写作节点有一个容易被忽略的细节：它的系统提示词里要包含"基于审核意见修改"的通道。当图在审核不通过后回流到写作节点时，写作者必须能看到上一条审核意见，并针对性地修改，而不是重新生成一篇全新的文章。
4.2 审核节点的评分标准与结构化判决
审核节点是这套流水线里我最花心思的地方。它不是简单地说一句"写得不错"或"需要修改"，而是输出一份结构化的审核判决书。
我设计的review_result长这样：
{ "passed": False, "score": 68, "issues": [ {"type": "factual_consistency", "detail": "第三段引用的数据与材料2不一致"}, {"type": "argument_coherence", "detail": "论点二与论点一存在立场冲突"} ], "suggestion": "重点修正论点二的数据引用，并统一两个论点的立场" }审核维度我通常固定为四个：事实一致性、观点一致性、结构完整性、可读性。事实一致性是最重要的，只要有一条factual_consistency级别的问题，passed就一定是 False。观点一致性问题次之，结构问题再次。
让审核节点输出结构化判决，能带来两个好处：一是给条件边一个明确的判断依据；二是审核意见可以被当作结构化数据直接注入写作节点。写作者不需要读一大段"修改建议汇总"，它只需要看到issues列表就能知道具体要改哪几处。这也是多智能体之间高效协作的关键——最小化自然语言废话。
4.3 条件边把“重写”变成可控循环
审核完成后的流向判断，是整套流程闭环的开关。我用一个decide_next函数实现：
MAX_REWRITE_TIMES = 3 def decide_next(state: AgentState) -> str: if state["review_result"]["passed"]: return "end" if state["rewrite_times"] >= MAX_REWRITE_TIMES: return "end" return "rewrite"这里的逻辑是：审核通过直接结束；审核不通过但重写次数还没到上限，就回写作节点；重写次数达到上限，无论如何都要结束，避免死循环烧钱。
你可能想问：为什么不一直重写直到通过为止？因为在实践中，如果模型两三轮重写后审核分数仍然很低，说明它在当前配置下无法胜任这个任务，继续循环只会浪费 tokens 和时间。对系统来说，最稳的策略是保留一个"历史最高分版本"，如果最后一次仍未通过，就把最高分版本作为兜底输出，至少不会比完全没审核更差。
5. 最容易踩的四个状态管理的坑：Reducer、上下文膨胀、结构化输出、成本失控
5.1 字段覆盖问题：LangGraph的Reducer机制
LangGraph 状态更新的默认行为，是"后者覆盖前者"。每个节点返回的字典只会对相同 key 做替换，不会自动合并。这在大多数场景下没问题，但遇到"累积追加"型字段就必须显式声明 Reducer。
比如我定义了review_history: Annotated[list, operator.add]，意思是每次有节点返回这个字段时，LangGraph 会把新值追加到原列表后面，而不是覆盖旧列表。这是非常容易踩的坑：如果你没有声明 Reducer，审核意见永远只保留最新一条，写作者就看不到第一次审核时指出的旧问题，改了新的漏了旧的。
字段覆盖问题还体现在rewrite_times上。审核节点每次执行时，需要在原状态基础上加 1 后返回，而不是直接返回一个固定值，否则重写次数永远停留在 1。
5.2 别把整个历史塞给每个节点
多智能体系统失败的第一原因，就是上下文盲目膨胀。一个很自然的做法是：既然状态里存了所有历史，那我就把它们全部组装成提示词，让节点自己挑重点。这个做法会让系统在跑了几轮之后，上下文越来越像一个堆满碎纸片的房间，模型的判断力急剧下降。
正确的做法是：每个节点只从状态里取自己需要的那几个字段。研究员只需要task，分析师只需要task和research_results，写作节点只需要analysis和对应的少量材料，审核节点只需要draft和task。上下文短了，不仅质量稳定，tokens 成本也肉眼可见地降下来了。
5.3 结构化输出不是万能的：解析兜底
多智能体协作特别依赖结构化中间产物，所以很多框架都提供了"让我直接输出 JSON"的能力。但实践中你会发现，不同模型服务商对结构化输出的支持程度并不一致，有的模型会偶尔多写一个注释，或者把 JSON 包裹在 Markdown 代码块里，导致解析失败。
我的做法是双保险：先用框架自带的结构化输出方法，如果返回结果解析失败，就退回到"让模型输出纯文本 JSON，再用正则提取最外层大括号，最后用json.loads解析"的兜底路径。审核节点这种关键节点的解析，必须加异常处理，宁可解析失败让整次任务重来，也不能让一条解析错乱的判决混进条件边。
5.4 成本与延时：给智能体分级
四个节点如果全部使用同一个最强模型，单次任务的成本会很高。我在实践中会把模型按环节分级：研究员和分析师需要较强的理解能力，审核者需要严格的判断能力，写作节点相对没那么吃模型上限，可以用更轻快的配置。这样一轮完整的任务下来，成本能减少不少。
延迟方面也要有个预期。四节点串行执行，加上一次或两次重写循环，总耗时通常是单次调用的三到五倍。如果业务场景对实时性要求高，可以考虑把研究和分析节点合并为一步预计算，让用户看到结果后，再异步触发写作和审核。实时反馈和批量质量处理可以分开做。
6. 完整代码串联与一次“审核驳回两轮”的实测复盘
6.1 完整代码：一条带反馈回路的四审流水线
把前面所有设计拼起来，就是一个可以跑的完整流程。下面的代码我尽量精简，突出流程骨架：
from typing import TypedDict, Annotated, operator from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): task: str research_results: list analysis: dict draft: str review_result: dict rewrite_times: int review_history: Annotated[list, operator.add] def researcher_node(state: AgentState) -> dict: prompt = f""" 你是研究员，任务是：{state['task']} 请检索相关资料，只输出以下JSON数组（不要额外解释）： [{{"title": str, "url": str, "summary": str, "key_points": [str], "source_label": "已验证|待核实"}}] 无法确认来源的信息，source_label 必须为"待核实"。 """ raw = call_model(prompt) # 这里替换为你自己的模型调用方式 results = parse_json_robust(raw) return {"research_results": results} def analyst_node(state: AgentState) -> dict: prompt = f""" 任务：{state['task']} 基于以下材料，输出文章分析骨架： {json.dumps(state['research_results'], ensure_ascii=False)} 输出JSON：{{"thesis": str, "target_audience": str, "arguments": [{{"topic": str, "stance": str}}], "evidence_mapping": {{"论点": [材料索引]}}}} """ raw = call_model(prompt) analysis = parse_json_robust(raw) return {"analysis": analysis} def writer_node(state: AgentState) -> dict: advice = "" if state["review_history"]: advice = f"\n以下是最近一次审核意见，请逐条修改：\n{json.dumps(state['review_history'][-1], ensure_ascii=False)}" prompt = f""" 你是写作者。任务：{state['task']} 基于分析骨架写作： {json.dumps(state['analysis'], ensure_ascii=False)} 只允许使用骨架中的论点，引用对应编号的材料，禁止新增事实。 {advice} 输出完整文章正文。 """ draft = call_model(prompt, temperature=0.7) return {"draft": draft} def reviewer_node(state: AgentState) -> dict: prompt = f""" 你是审核者。任务：{state['task']} 审核以下文章： {state['draft']} 输出JSON：{{"passed": bool, "score": int(0-100), "issues": [{{"type": "factual_consistency|argument_coherence|structure|readability", "detail": str}}], "suggestion": str}} 只要存在 factual_consistency 问题，passed 就必须为 false。 """ raw = call_model(prompt, temperature=0) review = parse_json_robust(raw) return { "review_result": review, "rewrite_times": state["rewrite_times"] + 1, "review_history": [review], } MAX_REWRITE_TIMES = 3 def decide_next(state: AgentState) -> str: if state["review_result"]["passed"]: return "end" if state["rewrite_times"] >= MAX_REWRITE_TIMES: return "end" return "rewrite" graph = StateGraph(AgentState) graph.add_node("researcher", researcher_node) graph.add_node("analyst", analyst_node) graph.add_node("writer", writer_node) graph.add_node("reviewer", reviewer_node) graph.add_edge(START, "researcher") graph.add_edge("researcher", "analyst") graph.add_edge("analyst", "writer") graph.add_edge("writer", "reviewer") graph.add_conditional_edge("reviewer", decide_next, {"rewrite": "writer", "end": END}) app = graph.compile() initial_state = { "task": "写一篇关于本地知识库检索方案选型的分析稿", "research_results": [], "analysis": {}, "draft": "", "review_result": {"passed": False}, "rewrite_times": 0, "review_history": [], } result = app.invoke(initial_state) print(result["draft"]) print(result["review_result"])我对这份代码的期望是：你可以直接把call_model替换成你正在用的模型服务商接入代码，然后填充自己的业务字段，就能跑通整条流水线。重点看四个节点的输入输出格式和条件边的流转逻辑。
6.2 一次实测：两轮驳回后的修订日志
我拿"本地知识库检索方案选型"这个任务实际跑了一次，把关键中间态贴出来供参考。
研究阶段输出了四条材料，其中有两条标了"待核实"，原因是没有拿到一手来源。分析阶段生成的骨架包含两个论点：一个偏向"向量检索为主、关键词检索兜底"，另一个是"召回质量需要人工标注反馈闭环"。这两个论点在骨架层面是互补的。
第一稿完成很快，但审核节点给出了 62 分，passed为 False。审核意见列了两条：第一条是"论文中引用的召回率数据与材料索引 [1] 不一致，材料中是 78%，文章写的是 87%"，第二条是"论点一和论点二有微妙冲突，前者强调向量检索为主，后者暗示关键词兜底更重要，建议明确主次关系"。
写作节点收到审核意见后，在第二次输出里修正了数据，但审核节点仍然打了 78 分，因为结构上出现了新的问题：第二段内容冗长且与第三段存在重复论述。第二轮回流后，写作节点删掉了重复段落，重组了结构，最终审核 91 分通过。整个流程从启动到结束，一共执行了五次节点调用，其中写作和审核各执行两次。
这次实测让我最直观的感受是：审核节点的价值不在于否定，而在于给出"可执行"的修改方向。没有结构化 issues 列表，写作者只会收到一句"写得不好，再改改"，然后陷入随机重试的循环；有了具体的"数据不一致、重复论述"这类描述，重写才是有方向的重写。
6.3 调试与上线前的建议
LangGraph 自带的流式输出对排查这类流程问题非常有用。你可以遍历每一步生成的中间态，看到每个节点返回的完整数据：
for step in app.stream(initial_state): for node_name, node_output in step.items(): print(f"--- {node_name} ---") print(node_output)这样你就能清晰地看到：研究节点输出了什么、分析骨架长什么样、哪一轮写作出问题了、审核判决的关键字段是什么。
上线前建议再补两层保护：一层是对输入和输出做长度限制，防止某个模型返回超长文本导致状态异常；另一层是给每个节点加失败重试和异常捕获，因为任何一次 JSON 解析失败都可能让整条流水线中断。
最后想多提一句：这套"研究-分析-写作-审核"的架构并不只适用于文章生产。我后来把它迁移到过两份不同的内容整理任务上，只需要替换研究节点的工具来源和分析节点的字段定义，整体结构都不需要改动。核心思路都是一样的——把复杂的知识工作拆成职责单一的节点，用状态对象把中间产物标准化，再用条件回环让质量持续得到反馈。搭好这次流水线之后，我最大的改变是不再把"写一篇文章"当成一次性的生成请求，而是当成一个需要流程管理、质量门控的小型项目来对待。
