如何用 LangGraph 构建多 Agent 协作系统？完整实战指南

本文将手把手教你使用 LangGraph 构建一个多 Agent 协作系统，从基础概念到完整代码实战，涵盖状态管理、路由决策、工具调用等核心环节，助你快速掌握 AI Agent 开发。

什么是 LangGraph？

LangGraph 是 LangChain 团队于 2024 年推出的 Agent 编排框架，基于图（Graph）结构来定义 Agent 工作流。与线性链（Chain）不同，LangGraph 允许你构建循环、条件分支、并行执行等复杂流程，非常适合多 Agent 协作场景。

到 2026 年，LangGraph 已经成为构建生产级 Agent 系统的首选框架之一，被广泛应用于智能客服、数据分析、代码审查等场景。

核心概念

LangGraph 的核心概念非常简单：

State（状态）：贯穿整个图的全局状态对象
Node（节点）：执行具体任务的函数，可以是 LLM、工具或自定义逻辑
Edge（边）：连接节点，控制流程走向
Conditional Edge（条件边）：根据状态决定下一个节点

实战：构建智能研究助手

我们将构建一个「智能研究助手」，包含三个 Agent：

Researcher（研究员）：负责搜索和收集信息
Analyst（分析师）：负责分析和总结信息
Writer（写手）：负责撰写最终报告

环境准备

pip install langgraph langchain-openai langchain-community tavily-python --break-system-packages

设置环境变量：

export OPENAI_API_KEY="your-openai-key"
export TAVILY_API_KEY="your-tavily-key"

完整代码

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate

# ==================== 1. 定义全局状态 ====================
class ResearchState(TypedDict):
    """研究任务的全局状态"""
    topic: str                           # 研究主题
    search_results: Annotated[List[str], operator.add]  # 搜索结果（累加）
    analysis: str                        # 分析结果
    report: str                          # 最终报告
    current_step: str                    # 当前步骤
    messages: Annotated[List[dict], operator.add]  # 消息历史

# ==================== 2. 初始化模型和工具 ====================
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
tavily_tool = TavilySearchResults(max_results=5)

# ==================== 3. 定义节点函数 ====================

def researcher_node(state: ResearchState) -> dict:
    """研究员 Agent：执行搜索并收集信息"""
    topic = state["topic"]
    print(f"🔍 Researcher 正在搜索: {topic}")

    # 使用 Tavily 搜索
    search_results = tavily_tool.invoke(f"{topic} 最新发展 2026")

    # 提取搜索摘要
    summaries = []
    for result in search_results:
        summaries.append(f"[{result['title']}] {result['content'][:300]}...")

    return {
        "search_results": summaries,
        "current_step": "research_done",
        "messages": [{"role": "assistant", "content": f"已完成搜索，找到 {len(summaries)} 条结果"}]
    }

def analyst_node(state: ResearchState) -> dict:
    """分析师 Agent：分析搜索结果"""
    print(f"📊 Analyst 正在分析数据...")

    search_text = "\n\n".join(state["search_results"])

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位资深行业分析师。请根据提供的搜索结果，提炼关键观点、趋势和洞察。"),
        ("human", "主题：{topic}\n\n搜索结果：\n{search_results}\n\n请提供结构化分析：")
    ])

    chain = prompt | llm
    analysis = chain.invoke({
        "topic": state["topic"],
        "search_results": search_text
    }).content

    return {
        "analysis": analysis,
        "current_step": "analysis_done",
        "messages": [{"role": "assistant", "content": "分析完成"}]
    }

def writer_node(state: ResearchState) -> dict:
    """写手 Agent：撰写最终报告"""
    print(f"✍️ Writer 正在撰写报告...")

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位技术报告撰写专家。请根据分析结果撰写一份专业、结构清晰的报告。"),
        ("human", "主题：{topic}\n\n分析结果：\n{analysis}\n\n请撰写完整报告（包含摘要、正文、结论）：")
    ])

    chain = prompt | llm
    report = chain.invoke({
        "topic": state["topic"],
        "analysis": state["analysis"]
    }).content

    return {
        "report": report,
        "current_step": "report_done",
        "messages": [{"role": "assistant", "content": "报告撰写完成"}]
    }

# ==================== 4. 条件路由函数 ====================

def route_by_step(state: ResearchState) -> str:
    """根据当前步骤决定下一个节点"""
    step = state.get("current_step", "")
    if step == "research_done":
        return "analyst"
    elif step == "analysis_done":
        return "writer"
    elif step == "report_done":
        return END
    return "researcher"

# ==================== 5. 构建图 ====================
workflow = StateGraph(ResearchState)

# 添加节点
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)

# 设置入口点
workflow.set_entry_point("researcher")

# 添加条件边
workflow.add_conditional_edges(
    "researcher",
    route_by_step,
    {"analyst": "analyst", END: END}
)

workflow.add_conditional_edges(
    "analyst",
    route_by_step,
    {"writer": "writer", END: END}
)

workflow.add_conditional_edges(
    "writer",
    route_by_step,
    {END: END}
)

# 编译图
app = workflow.compile()

# ==================== 6. 运行 ====================
if __name__ == "__main__":
    topic = "AI Agent 在 2026 年的发展趋势"

    # 执行工作流
    result = app.invoke({
        "topic": topic,
        "search_results": [],
        "analysis": "",
        "report": "",
        "current_step": "",
        "messages": []
    })

    print("\n" + "="*60)
    print("最终报告：")
    print("="*60)
    print(result["report"])

运行结果

🔍 Researcher 正在搜索: AI Agent 在 2026 年的发展趋势
📊 Analyst 正在分析数据...
✍️ Writer 正在撰写报告...

============================================================
最终报告：
============================================================
# AI Agent 2026 年发展趋势研究报告

## 摘要

2026 年，AI Agent 技术正经历从单任务工具向多 Agent 协作系统的范式转变...

进阶：带人工审核的循环流程

实际生产环境中，往往需要人工审核环节。下面的示例展示了如何在 Analyst 和 Writer 之间加入人工审核节点：

def human_review_node(state: ResearchState) -> dict:
    """人工审核节点"""
    print("\n" + "="*60)
    print("📋 请审核以下分析结果（输入 yes 继续，或输入修改意见）：")
    print("="*60)
    print(state["analysis"][:500] + "...\n")

    feedback = input("审核意见: ")

    if feedback.lower() == "yes":
        return {"current_step": "analysis_done", "messages": [{"role": "human", "content": "审核通过"}]}
    else:
        # 需要重新分析
        return {
            "current_step": "research_done",
            "messages": [{"role": "human", "content": f"审核意见: {feedback}"}]
        }

# 在图中添加人工审核节点
workflow.add_node("human_review", human_review_node)
workflow.add_conditional_edges(
    "analyst",
    route_by_step,
    {"writer": "human_review", END: END}
)
workflow.add_conditional_edges(
    "human_review",
    lambda s: "writer" if s["messages"][-1]["content"] == "审核通过" else "analyst",
    {"writer": "writer", "analyst": "analyst"}
)

可视化工作流

LangGraph 支持将工作流可视化：

from IPython.display import Image, display

# 生成 Mermaid 图
display(Image(app.get_graph().draw_mermaid_png()))

Mermaid 语法描述：

graph TD
    A[Researcher<br/>搜索收集] --> B[Analyst<br/>分析总结]
    B --> C{Human Review<br/>人工审核}
    C -->|通过| D[Writer<br/>撰写报告]
    C -->|不通过| B
    D --> E[END]

性能优化技巧

并行节点：使用 Send 机制并行执行多个 Researcher
状态精简：避免在 State 中存储大对象，使用引用或外部存储
检查点：启用 checkpointer 实现断点续跑

from langgraph.checkpoint.memory import MemorySaver

# 添加内存检查点
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 支持中断恢复
config = {"configurable": {"thread_id": "research-001"}}
result = app.invoke(initial_state, config=config)

常见问题 FAQ

Q: LangGraph 和 LangChain 的 Chain 有什么区别？ A: Chain 是线性流程，适合简单任务；LangGraph 基于图结构，支持循环、条件分支和并行，适合复杂 Agent 协作。

Q: 如何在 LangGraph 中集成自定义工具？ A: 任何 Python 函数都可以作为节点。只需确保函数接收 State 并返回更新字典即可。

Q: LangGraph 支持流式输出吗？ A: 支持。使用 app.stream() 代替 app.invoke() 即可获取实时流式输出。

Q: 生产环境如何持久化状态？ A: 可以使用 PostgresSaver 或 RedisSaver 替代 MemorySaver，实现分布式状态持久化。

Q: 一个图中可以有多个 LLM 模型吗？ A: 可以。不同节点可以使用不同的 LLM 实例，例如 Researcher 用轻量级模型，Writer 用 GPT-4o。

总结

LangGraph 通过图结构为 Agent 开发带来了极大的灵活性。本文从 State、Node、Edge 三个核心概念出发，完整演示了一个多 Agent 协作系统的构建过程。掌握这些基础后，你可以进一步探索并行执行、子图嵌套、持久化检查点等高级特性，构建更强大的 AI Agent 应用。