您现在的位置是:首页 > 文章详情

从“单兵”到“团队”:用LangGraph构建你的第一个多智能体(Multi-Agent)系统

日期:2025-08-15点击:7

所有完整内容,首发于公众号『技术老金』及个人博客,欢迎关注交流

老金导读:

你是否发现,为了让AI完成一个复杂任务,你的提示词(Prompt)正变得越来越长、越来越臃肿?你试图将一个研究员、一个程序员、一个测试员的职责,全部塞给同一个AI,期望它成为一个"全能超人"。

但结果往往是:AI"精神错乱",输出质量忽高忽低,成本飙升,且一旦出错,调试如同噩梦。

这条路走不通。

专业的玩家,会像一个真正的架构师一样思考:我们需要的不是一个"超级个体",而是一支"精英团队"

这,就是"多智能体(Multi-Agent)系统"的魅力。本文将手把手带你,用当今最强大的流程编排工具LangGraph,构建一个由"研究员-写手-评审员"组成的内容创作AI团队。你将掌握让AI互相协作、审查、甚至循环修改工作的核心技术。

这篇文章是独立的、端到端的实战教程。 如果你对LangGraph还完全陌生,可以先阅读我们的《LangGraph入门与避坑指南》作为预习,那会让你更容易上手。


一、为什么需要"AI团队",而不是一个"全能AI"?

我们很容易陷入一个误区:期望用一个"超级提示词"让一个AI模型解决所有问题。但在真实、复杂的业务场景中,这几乎行不通。原因有三:

  1. 职责不清(Responsibility Confusion): 一个冗长的、包含多种任务指令的Prompt,很容易让LLM"精神错乱",顾此失彼,导致输出质量严重下降。
  2. 成本高昂(High Cost): 复杂的任务链往往需要最强大的模型(如GPT-4o)来处理,如果每一步都用它,成本会非常高。而实际上,某些环节(如格式整理、意图识别)用更便宜的模型就足够了。
  3. 难以维护(Hard to Maintain): "超级提示词"是一个巨大的黑盒,一旦某个环节出了问题,调试和迭代的难度极高,牵一发而动全身。

而多智能体系统,就像一个分工明确的人类团队,每个Agent都有单一、清晰的职责。这让我们的系统更清晰、更健壮、也更经济。

二、实战:构建一个"内容创作AI团队"

我们将构建一个由三位AI专家组成的团队:

  • 研究员(Researcher): 负责根据主题,上网搜索、整理资料。
  • 写手(Writer): 负责根据研究员的资料,撰写初稿。
  • 评审员(Reviewer): 负责审查初稿,提出修改意见,并决定是通过还是打回重写。

这个流程中,最关键的是"评审员"可以决定工作流的走向,如果他不满意,可以将任务"打回"给写手,形成一个循环(Loop)。这正是LangGraph的威力所在。

下面,我们来看代码如何实现。

第一步:定义团队的"共享工作台" (TeamState)

State是LangGraph的灵魂,它是所有智能体共享信息的地方,就像团队的中央白板。

from typing import TypedDict, Annotated class TeamState(TypedDict): topic: str # 任务主题 research_material: Annotated[str, "研究员收集的资料"] draft: Annotated[str, "写手撰写的初稿"] review_feedback: Annotated[str, "评审员的反馈意见"] revision_count: int # 修订次数,防止无限循环 final_report: Annotated[str, "最终报告"] 

我们定义了一个TeamState,包含了任务流转中所有需要的数据。Annotated可以让我们为每个字段添加描述,增加可读性。

第二步:创建"专家"节点 (Agent Nodes)

每个Agent都是一个Python函数,它接收当前的State作为输入,并返回一个包含更新后数据的字典。

1. 研究员 (Researcher): 它需要一个工具来上网搜索。这里我们用DuckDuckGoSearchRun

from langchain_community.tools import DuckDuckGoSearchRun search_tool = DuckDuckGoSearchRun() def researcher_node(state: TeamState): print("--- 节点: 研究员 ---") topic = state["topic"] research_material = search_tool.run(f"深入研究主题: {topic}") return {"research_material": research_material} 

2. 写手 (Writer): 它接收research_material,并调用LLM撰写初稿。如果review_feedback存在,它会一并纳入考虑进行修改。

def writer_node(state: TeamState): print("--- 节点: 写手 ---") # ... (构建prompt) ... llm = ChatOpenAI(model="gpt-4o", temperature=0.7) draft = llm.invoke(prompt).content # 更新修订次数 revision_count = state["revision_count"] + 1 return {"draft": draft, "revision_count": revision_count} 

3. 评审员 (Reviewer): 这是最关键的决策节点。为了让LLM给出非黑即白的判断,我们使用了Function Calling 。我们定义一个ReviewDecision模型,强制LLM必须决定"通过"或"需要修改"。

from langchain_core.pydantic_v1 import BaseModel, Field class ReviewDecision(BaseModel): """评审决定""" decision: str = Field(description="只能是 '通过' 或 '需要修改'") feedback: str = Field(description="如果'需要修改', 请提供修改意见。") def reviewer_node(state: TeamState): print("--- 节点: 评审员 ---") # ... (构建prompt) ... llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools([ReviewDecision]) response = llm_with_tools.invoke(prompt) tool_call = response.tool_calls[0] decision_data = tool_call['args'] if decision_data['decision'] == '需要修改': return {"review_feedback": decision_data['feedback']} else: return {"review_feedback": "", "final_report": state["draft"]} 

第三步:定义"工作流" (Conditional Edges)

这是LangGraph的精髓所在,它决定了团队的协作规则。

def should_continue(state: TeamState): print("--- 条件判断 ---") # 如果达到最大修订次数,或评审员没有给出反馈,则结束 if state["revision_count"] > 3 or not state.get("review_feedback"): return END else: # 否则,打回给写手 return "writer" # ... 在Graph构建中 ... graph.add_conditional_edges( "reviewer", # 在这个节点之后进行判断 should_continue, # 使用这个函数来决策 { "writer": "writer", # 如果返回"writer",则下一个节点是writer END: END # 如果返回END,则流程结束 } ) 

add_conditional_edges方法让我们的Graph具备了"思考"能力。它在reviewer节点执行完毕后,调用should_continue函数,根据函数的返回值("writer"END)来决定下一个节点的走向,从而完美地实现了循环修改的业务逻辑。

第四步:组装与运行

最后,我们将所有节点和边组装起来,编译成一个可执行的app

from langgraph.graph import StateGraph, END graph = StateGraph(TeamState) # 添加节点 graph.add_node("researcher", researcher_node) # ... (添加writer, reviewer) # 设置入口和常规边 graph.set_entry_point("researcher") graph.add_edge("researcher", "writer") graph.add_edge("writer", "reviewer") # 添加条件边 (如上所示) # 编译 app = graph.compile() # 运行 topic = "LangGraph相比于CrewAI的优势是什么?" initial_state = {"topic": topic, "revision_count": 0} for s in app.stream(initial_state): print(s) 

当你运行这段代码时,你会清晰地看到三个智能体如何依次被调用,评审员如何做出判断,以及在报告被打回后,流程如何自动地回到写手节点,开始新一轮的修改。

老金总结

今天,我们从一个"线性流程"迈向了一个"协作团队"。通过LangGraph,我们不仅能定义任务的顺序,更能注入判断、循环、协作等复杂的业务逻辑。

这套"研究员-写手-评审员"的模式,几乎可以平移到任何需要"生成-审核"机制的场景中,例如:

  • 代码生成与测试: 程序员Agent写代码,测试员Agent跑测试,不通过就打回。
  • 营销文案与合规: 文案Agent生成广告词,法务Agent进行合规审查。
  • 数据分析与洞察: 分析师Agent跑数据,策略师Agent从报告中提炼商业洞察。

掌握了多智能体系统的构建,你就真正拥有了用AI解决复杂、真实世界问题的能力。这,才是AI Agent开发的真正魅力所在。

觉得有用,别忘了给老金点个赞,关注一下!


[版权声明]

本文由 "技术老金" 原创首发于个人博客及微信公众号『技术老金』,转载请注明出处。

原文链接:https://my.oschina.net/xxjin/blog/18688347
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章