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

发布时间:2026/9/23 22:55:40

实战:用 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 我们拆一个「协调者 子团队」的完整案例。
延伸阅读

更多相关文章

2026/9/23 1:51:50

RAG的Top K取3条还是10条?我用20个问题跑了一遍

RAG 调参数时,Top K 往往是最先被改的那个数字。 答案漏了,就从 3 调到 10;还不放心,再改成 20。看起来很合理:多拿几段资料,正确答案总该在里面。 可检索结果真的越多越好吗? 我做了一个 20…

2026/9/21 14:17:36

Git Tag核心用法与团队协作规范详解

1. Git Tag的核心价值解析在团队协作开发中,我们经常遇到这样的场景:某个版本已经通过测试即将发布,但后续提交仍在持续进行。这时候就需要一个可靠的标记机制来锁定特定代码状态——这就是Git Tag的核心作用。不同于普通commit的线性记录&am…

2026/9/23 22:55:15

内容创作失败的三大根源:选题、转化与交付

1. 这个标题不是玩笑,是内容创作者的真实生存切片“1024,鸽了1024篇博文的我……”——看到这个标题,我下意识点开,不是因为好奇,而是心头一紧:这数字太熟了。不是程序员节那个1024,而是我电脑里…

2026/9/23 22:55:15

软考高项257个记忆点:分类刷法、易混考点与避坑指南

简介:面向信息系统项目管理师(软考高项)考生整理的高频考点浓缩笔记,将散布在多本教材中的重点知识汇总为257个要点,覆盖项目管理、系统开发、计算机网络、数据安全、多媒体技术、企业信息化等常考方向。每个知识点以短…

2026/9/23 22:55:15

DeepSeek-R1技术报告解读:GRPO强化学习训练与复现指南

简介:DeepSeek-R1技术报告论文面向大模型研究者、算法工程师及对强化学习推理方向感兴趣的进阶学习者,系统呈现了如何通过大规模强化学习激发大语言模型推理能力。报告围绕DeepSeek-R1-Zero与DeepSeek-R1两代模型展开,前者在无监督微调前提下…

2026/9/23 22:55:15

交换机路由器Console初始配置避坑指南

简介:本资源是一份面向网络工程初学者与高职院校实训学生的交换机与路由器基础配置实验指导文档,聚焦带外管理(Console线连接超级终端配置)与带内远程管理(Telnet/Web/TFTP/SNMP)两大核心能力培养。文档系统…

2026/9/23 12:07:00

GAMP 5 基于风险的计算机化系统验证:软件分类与审计追踪实践

简介:《A Risk-Based Approach to Compliant GxP Computerized Systems》即业内熟知的GAMP 5指南,面向制药企业质量与IT合规人员、验证工程师及计算机化系统管理者,用于解决GxP法规环境下系统合规性难以科学落地的问题。文档以风险管理为主线…

2026/9/23 12:06:55

安全托管MSSP实战:从静态防御到人机协同的攻防运营与应急响应

简介:这份PPT围绕互联网业务安全托管服务展开,面向企业安全负责人、IT运维人员及关注MSSP/MSS选型的读者,重点回应传统安全过度依赖人工、碎片化静态防御难以对抗产业化攻击等痛点。资源共1个pptx文件,包体约30.63MB,以…

2026/9/23 0:01:54

3个实战技巧搞定形式英语:从看教程到跑通性能优化

3个实战技巧搞定形式英语:从看教程到跑通性能优化 看了一堆教程还是不会写项目?别慌,这种“眼高手低”的困境在开发者圈子里太常见了。很多人以为卡点在语法,其实真正拦路虎是缺乏将知识点串联成完整链路的能力。今天咱们不聊虚的,直接拿【形式英语】这…

2026/9/22 16:34:32

USB Type-C PCB布局分区设计:电源、高速信号与PD协议全攻略

做硬件这行,Type-C接口算是典型的“看着简单,做起来全坑”的东西。光引脚就24个,高低速信号、电源、控制线全部塞在一个小小的连接器里,如果PCB布局不做规划,打样回来基本就是“插上没反应”、“高速掉线”、“静电一打…

2026/9/22 20:01:30

系统编程学习原型如何补齐稳定性边界

系统编程学习原型如何补齐稳定性边界预算有限时&#xff0c;我先优化明显多余的复制&#xff0c;而不是猜测性地换容器。用借用传递只读数据通常就能减少分配&#xff1a; fn parse(line: &str) -> Result<Item, Error> { /* ... */ }用基准确认热点确实在分配&am…

2026/9/22 13:25:41

雨花区哪家财务公司代理记账比较好?

在雨花区&#xff0c;企业处理财税事务常常面临诸多挑战&#xff0c;选择一家靠谱的财务公司至关重要。湖南巨勤财务管理咨询有限公司就是本地正规实体财税服务机构&#xff0c;深耕本地工商财税行业多年&#xff0c;熟悉当地工商局、税务局最新政策与申报流程。主营公司注册、…

还想了解更多?直接咨询顾问

免费诊断 + 免费方案 + 透明报价。

全国咨询热线400-8866-253
免费获取方案
咨询二维码