实战：用 LangGraph 搭建「数据搜集员+撰稿人」协作团队

单 Agent 把查资料、写稿、审校全塞进一个 prompt。多 Agent 把任务拆成有向图每个节点只干一件事产出物在共享状态里接力。这篇把协作落成能跑的代码。实战用 LangGraph 搭建「数据搜集员撰稿人」协作团队同一个选题让一个 Agent 一口气写完和让「搜集员 撰稿人」两个 Agent 接力写完差在哪多智能体协作不是把 prompt 写得更长而是把任务拆成一张有向图。单 Agent 的写法把查资料、写稿、审校全塞进一个 prompt上下文越拖越长错误一路带到底。多 Agent 的写法把任务拆成一张有向图每个节点只干一件事产出物在共享状态里接力。这篇文章把这套协作落成能跑的代码。装好 langgraph 和 langchain-openai 两个包、配好 key就能复现「研究笔记 → 成稿」的最小闭环。1 图编排StateGraph、节点与边的分工LangGraph 用 StateGraph 把协作建模成有向图。State 是节点共享的数据快照Node 是处理逻辑Edge 决定下一步执行谁。官方文档[1]的原话是「nodes do the work, edges tell what to do next」。组合 Node 和 Edge就能构建随时间演进的循环工作流。LangChain 官方博客[2]给协作价值的理由很直白把大任务拆成小任务每个子 agent 只做一件擅长的事整体更可控、更容易扩展。选「搜集员 撰稿人」而不是「一个万能 prompt」就是让 prompt 变短、上下文不臃肿。信息传递方式单 Agent 靠 prompt 内部衔接多 Agent 靠 State 字段显式交接。并行能力单 Agent 天然串行多 Agent 可并行分派多个节点。可调试性单 Agent 出错要翻整段对话多 Agent 能单独调任一节点。失败影响面单 Agent 错误贯穿全程多 Agent 单点可重试。两种模式各有取舍一句话收口。串行单 Agent 适合线性小任务多 Agent 图编排在并行与可控性上胜出代价是图结构和状态设计更复杂。为什么拆成两个节点而不是一个失败影响面不同。研究环节的幻觉会直接污染正文拆开后 research_notes 和 draft 是独立字段可以分别校验、分别重试。整张图的骨架只需六行核心依赖只有两个包。graph StateGraph(AgentState) # 状态协作契约 graph.add_node(researcher, researcher_node) # 节点处理逻辑 graph.add_node(writer, writer_node) graph.add_edge(researcher, writer) # 边决定下一步 app graph.compile() # 编译为可执行图pip install langgraph langchain-openai2 节点与状态AgentState 与双角色节点函数节点是纯函数输入完整 State返回增量更新。增量指只返回自己改动的字段框架自动合并没改的字段保持原值。from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import SystemMessage, HumanMessage from langchain_openai import ChatOpenAI # 读者替换为自己的模型与 keystub 验证时此变量会被覆盖 llm ChatOpenAI(modelgpt-4o-mini) class AgentState(TypedDict): topic: str research_notes: str draft: str RESEARCHER_SYSTEM ( 你是一位专业的技术研究员。 根据主题生成研究笔记包含核心概念、技术要点、常见问题与最佳实践。 ) WRITER_SYSTEM ( 你是资深技术博客作家。 只根据研究笔记撰写文章引言 → 核心概念 → 实战示例 → 总结。 ) def researcher_node(state: AgentState) - dict: response llm.invoke( [ SystemMessage(contentRESEARCHER_SYSTEM), HumanMessage(contentf请研究{state[topic]}), ] ) return {research_notes: response.content} # 只更新自己负责的字段 def writer_node(state: AgentState) - dict: response llm.invoke( [ SystemMessage(contentWRITER_SYSTEM), HumanMessage(contentf研究笔记\n{state[research_notes]}), ] ) return {draft: response.content}返回增量而非完整状态是省 token 的关键。完整状态可能几千字每次全量返回几个节点跑下来上下文就爆了。增量只带 diff框架负责合并。AgentState 用 TypedDict 定义协作契约字段清单就是契约文档。topic输入进图时给定。research_notesResearcher 的产出Writer 的输入。draftWriter 的产出最终交付。想加第三个角色改的是契约而不是图。比如加 Reviewer就在 AgentState 里加 review_score 和 review_feedback 两个字段节点函数照旧只写自己负责的字段。职责边界由字段契约保证。Researcher 不碰 draftWriter 不碰 topic越界改字段在类型定义里就能查出来。3 边与状态共享增量返回驱动协作add_edge 声明「搜集员完成触发撰稿人」。执行顺序由边决定不写死在代码嵌套里所以任一节点都能单独抽出来调试。START 是图的统一入口END 是出口普通节点都挂在两者之间。def build_graph(): graph StateGraph(AgentState) graph.add_node(researcher, researcher_node) graph.add_node(writer, writer_node) graph.add_edge(START, researcher) graph.add_edge(researcher, writer) graph.add_edge(writer, END) return graph.compile() app build_graph() result app.invoke({topic: LangGraph 多智能体系统实战}) print(result[draft])invoke 返回的是最终 Stateresult[“draft”] 就是写稿节点的产出。中途想观察每一步给图挂 checkpointer 或打印节点返回值都可以。增量合并每次节点返回增量框架自动合并进 State没更新的字段原样保留。并行基础多个节点同时只写各自字段互不覆盖。§5 的 Send 扇出正是靠这个语义。图3 增量返回驱动协作每步只新增自己负责的字段框架自动合并未更新字段保留。调试时可以构造假 state只调单个节点函数不必把整条链跑一遍。字段没更新也不会被覆盖合并行为由框架保证。4 条件边用评分驱动修改回环顺序边只能一路走到底质量把关要靠条件边。Reviewer 节点给草稿打分1-10并写反馈add_conditional_edges 按路由函数的结果决定下一步。REVIEWER_SYSTEM ( 你是资深技术审核员。评估文档的准确性、完整性、清晰度和结构。 必须以以下格式回复\n SCORE: 整数 1-10\n FEEDBACK: 详细修改意见 ) def parse_score(content: str) - int: for line in content.splitlines(): if line.strip().upper().startswith(SCORE:): return int(line.split(:, 1)[1].strip()) return 0 def parse_feedback(content: str) - str: for line in content.splitlines(): if line.strip().upper().startswith(FEEDBACK:): return line.split(:, 1)[1].strip() return content def reviewer_node(state: AgentState) - dict: response llm.invoke( [ SystemMessage(contentREVIEWER_SYSTEM), HumanMessage(contentstate[draft]), ] ) return { review_score: parse_score(response.content), review_feedback: parse_feedback(response.content), } def route_after_review(state: AgentState) - str: score state.get(review_score, 0) revisions state.get(revision_count, 0) if score 7: return editor # 合格进入终审 if revisions MAX_REVISIONS: return writer # 不合格回去改次数未用尽 return editor # 修改次数用完强制通过 graph.add_conditional_edges( reviewer, route_after_review, {writer: writer, editor: editor}, )score 低于 7 回写 WriterWriter 进入修改模式读反馈改稿。revision_count 每次递增MAX_REVISIONS3 封顶防止图死循环。同一个 Writer 节点双模工作首次创作看研究笔记修改模式看反馈。上限设 3给质量留余地也给流程留出口不会无限返工。分数制比二元判断多一层灰度6 分和 8 分的处理路径不同反馈更有针对性。阈值 7 和上限 3 都是参数按业务松紧调。图4 审核回环score 分叉路由revision_count 限次兜底。路由返回的是字符串映射表{writer: writer, editor: editor}把它接到具体节点。以后要加第三分支改路由函数和映射表两处就够了。这里有一条官方 warning不要从同一个节点混用普通边和动态路由。重试保护要设计进图结构不是事后补丁。5 生产化人工介入、状态持久化与并行分派闭环跑通后往生产走有三个进阶方向。每个给关键 API 示意非完整实现。模式适用场景关键 API人工介入发布前审批、关键步骤确认interrupt_before / interrupt状态持久化断点续跑、故障恢复SqliteSaver / PostgresSaver并行分派多路独立研究再汇聚Send三个方向解决三类生产问题人机边界、故障恢复、吞吐。逐个加不必一次全上。人工介入interrupt_before 让图在指定节点前暂停等人工确认再继续。确认通过后携带同一 thread_id 重新 invoke或用 Command(resume…) 注入审批结果图从断点接着跑不需要重跑前面的节点。状态持久化检查点存进数据库进程挂了从上次断点接着跑不重头再来。SqliteSaver 适合单机PostgresSaver 适合多实例共享。并行分派Send 把多个子任务同时发给同一个节点多路独立研究完成后在汇聚节点合并结果。互不依赖的子问题可以同时查。# 示意一人工审批暂停非完整实现 app graph.compile(checkpointersaver, interrupt_before[editor]) # 示意二SqliteSaver 断点续跑先安装 pip install langgraph-checkpoint-sqlite from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as saver: app graph.compile(checkpointersaver) # 示意三Send 扇出多路研究 from langgraph.types import Send def fan_out(state): return [Send(researcher, {topic: t}) for t in state[topics]]三个方向可以叠加。先加 checkpointer 保证断点续跑再在关键节点前加 interrupt_before最后用 Send 提速。再往上还有 Supervisor 层级模式一个协调者调度多个子团队本系列 6.3 展开。LangGraph 生态仍在快速演进2026 时效材料不足本文不展开。6 最小实现搜集员撰稿人双节点闭环把前几节的节点、状态、边拼成一个文件就是完整的最小实现。代码与本地实跑验证版本逐字一致复制即跑。Researcher Writer 双节点最小闭环LangGraph。 运行前提: pip install langgraph langchain langchain-openai export OPENAI_API_KEY... 若未配置 API key可用 code/verify_01.py 注入 stub LLM 验证图机制。 from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import SystemMessage, HumanMessage from langchain_openai import ChatOpenAI # 读者替换为自己的模型与 keystub 验证时此变量会被覆盖 llm ChatOpenAI(modelgpt-4o-mini) class AgentState(TypedDict): topic: str research_notes: str draft: str RESEARCHER_SYSTEM ( 你是一位专业的技术研究员。 根据主题生成研究笔记包含核心概念、技术要点、常见问题与最佳实践。 ) WRITER_SYSTEM ( 你是资深技术博客作家。 只根据研究笔记撰写文章引言 → 核心概念 → 实战示例 → 总结。 ) def researcher_node(state: AgentState) - dict: response llm.invoke( [ SystemMessage(contentRESEARCHER_SYSTEM), HumanMessage(contentf请研究{state[topic]}), ] ) return {research_notes: response.content} # 只更新自己负责的字段 def writer_node(state: AgentState) - dict: response llm.invoke( [ SystemMessage(contentWRITER_SYSTEM), HumanMessage(contentf研究笔记\n{state[research_notes]}), ] ) return {draft: response.content} def build_graph(): graph StateGraph(AgentState) graph.add_node(researcher, researcher_node) graph.add_node(writer, writer_node) graph.add_edge(START, researcher) graph.add_edge(researcher, writer) graph.add_edge(writer, END) return graph.compile() def main() - None: app build_graph() result app.invoke({topic: LangGraph 多智能体系统实战}) print(result[draft]) if __name__ __main__: main()预期跑通效果python minimal.py退出码 0终端输出一篇基于 research_notes 生成的 draft。本地用 stub LLM 实跑verify_01.py输出与真实链路同构research_notesRAG检索增强生成先向量化语料检索 top-k再交给 LLM 生成。draft基于研究笔记的报告RAG 的引言、核心概念、实战示例与总结。没有 key 也能验证图机制stub 思路见 code/verify_01.py。换 key、换模型、换 topic图结构都不用动。7 小结从双角色顺序协作到审核回环的能力阶梯这篇覆盖的能力阶梯分三步。顺序协作§1-3State 定义契约Node 各自干活Edge 串起顺序。质量回环§4条件边打分不合格回写修改次数封顶防死循环。生产化§5人工介入、断点续跑、并行分派。多智能体的价值在「拆分 可控」不在堆 Agent 数量。拆得越细单个节点越好测。边画得越清楚流程越可控。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】整套能力阶梯的核心先让协作跑起来再让它可信最后让它可运维。从双角色协作往上是 Supervisor 层级编排。6.3 我们拆一个「协调者 子团队」的完整案例。