LangGraph 多智能体编排实战：从单 Agent 到团队协作

2026年7月26日 AI 前沿资讯苏三

为什么需要多 Agent 协作？

单 Agent 在处理简单任务时表现不错，但面对需要多个专业领域知识的复杂问题时，往往会力不从心。就像你不能指望一个全栈工程师独立完成数据库设计、前端开发和运维监控一样，单个 LLM Agent 也应该把不同职责交给专门的子 Agent。

LangGraph 是 LangChain 推出的一个基于图结构的工作流编排框架，专门为多 Agent 系统设计。与传统的线性链式调用不同，它允许 Agent 之间有循环、分支和条件判断，更接近真实世界的业务流程。

架��设计：三位一体的 Agent 团队

在本文中，我们将构建一个技术知识库问答系统，包含三个角色：

Researcher Agent：负责搜索知识库中的相关文档
Writer Agent：根据搜索结果撰写结构化的回答
Reviewer Agent：审查回答质量，决定是否通过或返回修改

这个设计与 Java 后端技术栈中的微服务分工理念有异曲同工之妙——每个组件只做好一件事。

代码实现

from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState
from typing import TypedDict

# 定义工作流状态
class AgentState(TypedDict):
    messages: list
    confidence: float
    review_rounds: int

# Researcher Node
def researcher_node(state: AgentState):
    research_prompt = """You are a technical knowledge base researcher.
Search for relevant documentation about the user's query.
Return all findings as structured notes."""
    messages = state["messages"] + [
        {"role": "system", "content": research_prompt}
    ]
    # 实际项目中接入向量检索
    response = {"role": "assistant", "content": "Found 3 relevant documents..."}
    return {"messages": messages + [response]}

# Writer Node
def writer_node(state: AgentState):
    write_prompt = """Based on the research findings, compose
a clear and well-structured answer to the user's question."""
    messages = state["messages"] + [
        {"role": "system", "content": write_prompt}
    ]
    response = {"role": "assistant", "content": "Here is your answer..."}
    return {"messages": messages + [response]}

# Reviewer Node
def reviewer_node(state: AgentState):
    review_prompt = """Review the generated answer for accuracy,
completeness, and readability. Return approve or revise."""
    messages = state["messages"] + [
        {"role": "system", "content": review_prompt}
    ]
    # 模拟评分
    confidence = 0.85
    return {"confidence": confidence, "review_rounds": state.get("review_rounds", 0) + 1}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)

# 设置入口
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")

# 条件边：根据审查结果决定下一步
def route_review(state: AgentState):
    if state["confidence"] >= 0.8 and state.get("review_rounds", 0) < 3:
        return "final"
    return "revise"

workflow.add_conditional_edges(
    "reviewer", route_review,
    {"final": None, "revise": "writer"}  # None = 结束
)

app = workflow.compile()

运行工作流

result = app.invoke({
    "messages": [{
        "role": "user",
        "content": "如何理解 Spring Boot 中的 Bean 生命周期？"
    }]
})
print(result["messages"][-1]["content"])