当前位置：首页 > news >正文

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

news 2026/8/9 0:36:18

单 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-openai

2 节点与状态：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 # 读者替换为自己的模型与 key；stub 验证时此变量会被覆盖 llm = ChatOpenAI(model="gpt-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(content=RESEARCHER_SYSTEM), HumanMessage(content=f"请研究：{state['topic']}"), ] ) return {"research_notes": response.content} # 只更新自己负责的字段 def writer_node(state: AgentState) -> dict: response = llm.invoke( [ SystemMessage(content=WRITER_SYSTEM), HumanMessage(content=f"研究笔记：\n{state['research_notes']}"), ] ) return {"draft": response.content}

返回增量而非完整状态，是省 token 的关键。完整状态可能几千字，每次全量返回，几个节点跑下来上下文就爆了。增量只带 diff，框架负责合并。

AgentState 用 TypedDict 定义协作契约，字段清单就是契约文档。

topic：输入，进图时给定。
research_notes：Researcher 的产出，Writer 的输入。
draft：Writer 的产出，最终交付。

想加第三个角色，改的是契约而不是图。比如加 Reviewer，就在 AgentState 里加 review_score 和 review_feedback 两个字段，节点函数照旧只写自己负责的字段。

职责边界由字段契约保证。Researcher 不碰 draft，Writer 不碰 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 返回的是最终 State，result[“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(content=REVIEWER_SYSTEM), HumanMessage(content=state["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 回写 Writer，Writer 进入修改模式，读反馈改稿。revision_count 每次递增，MAX_REVISIONS=3 封顶，防止图死循环。同一个 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(checkpointer=saver, 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(checkpointer=saver) # 示意三：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 # 读者替换为自己的模型与 key；stub 验证时此变量会被覆盖 llm = ChatOpenAI(model="gpt-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(content=RESEARCHER_SYSTEM), HumanMessage(content=f"请研究：{state['topic']}"), ] ) return {"research_notes": response.content} # 只更新自己负责的字段 def writer_node(state: AgentState) -> dict: response = llm.invoke( [ SystemMessage(content=WRITER_SYSTEM), HumanMessage(content=f"研究笔记：\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_notes：RAG（检索增强生成）：先向量化语料，检索 top-k，再交给 LLM 生成。
draft：基于研究笔记的报告：RAG 的引言、核心概念、实战示例与总结。

没有 key 也能验证图机制，stub 思路见 code/verify_01.py。换 key、换模型、换 topic，图结构都不用动。

7 小结：从双角色顺序协作到审核回环的能力阶梯

这篇覆盖的能力阶梯分三步。

顺序协作（§1-3）：State 定义契约，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时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～