LangGraph多Agent编排教程：构建协作式AI Agent系统

日期：2026-08-14 | 分类：AI工具教程

项目简介

LangGraph是LangChain团队推出的Agent编排框架，基于StateGraph状态图构建多Agent协作系统。支持条件路由、Human-in-loop、检查点持久化、流式输出。

安装

pip install langgraph langchain-openai

export OPENAI_API_KEY="sk-your-key"

定义State

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next: str

创建Agent节点

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-5.6")

# 研究Agent
research_agent = create_react_agent(
    llm,
    tools=[search_tool, web_fetch_tool],
    state_modifier="你是研究专家，负责收集和分析信息"
)

# 写作Agent
writer_agent = create_react_agent(
    llm,
    tools=[write_file_tool],
    state_modifier="你是写作专家，基于研究结果撰写文章"
)

def research_node(state):
    result = research_agent.invoke(state)
    return {"messages": result["messages"], "next": "writer"}

def writer_node(state):
    result = writer_agent.invoke(state)
    return {"messages": result["messages"], "next": END}

构建工作流

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("researcher", research_node)
workflow.add_node("writer", writer_node)

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

# 条件路由
workflow.add_conditional_edges(
    "researcher",
    lambda state: state["next"],
    {"writer": "writer", END: END}
)

workflow.add_edge("writer", END)

# 编译
app = workflow.compile()

Human-in-Loop人工审批

from langgraph.checkpoint import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["writer"]  # 写作前暂停等人工确认
)

# 运行
config = {"configurable": {"thread_id": "1"}}
result = app.invoke({"messages": [("user", "写一篇关于AI的深度文章")]}, config)

# 人工审核后继续
result = app.invoke(None, config)  # 从断点继续

流式输出

async for event in app.astream(
    {"messages": [("user", "分析AI行业趋势")]},
    config={"configurable": {"thread_id": "2"}}
):
    print(event, end="", flush=True)

检查点持久化

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL持久化
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/db"
)
app = workflow.compile(checkpointer=checkpointer)
# 重启后可从断点恢复

相关阅读

• 2026 Agent框架选型指南
• 多Agent协作设计模式

📚 常见问题

LangGraph多Agent编排是什么？

LangGraph是LangChain团队推出的Agent编排框架，基于StateGraph状态图构建多Agent协作系统。支持条件路由、Human-in-loop、检查点持久化、流式输出。

如何上手LangGraph多Agent编排？

2026 Agent框架选型指南