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

老金导读：

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

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

这条路走不通。

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

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

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

一、为什么需要“AI团队”，而不是一个“全能AI”？

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

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

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

二、实战：构建一个“内容创作AI团队”

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

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

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

下面，我们来看代码如何实现。（完整代码可见同目录下的T08_LangGraph多智能体实战.py）

第一步：定义团队的“共享工作台” (`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)

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