langgraph - 基于图结构的智能体编排框架，用于构建稳定可控的AI工作流

langgraph - 基于图结构的智能体编排框架，用于构建稳定可控的AI工作流

当你想用大语言模型构建一个复杂的AI应用时，是否遇到过这样的困境：简单的链式调用无法处理分支逻辑，基础智能体难以保证执行的可控性，复杂的业务场景需要精细化的流程控制？langgraph正是为了解决这些问题而生的。它是langchain生态中的明星项目，将智能体的执行过程建模为有向图，通过节点和边的组合，让你可以用代码的方式精确控制AI应用的状态流转、分支判断和循环执行，构建出既灵活又稳定的复杂工作流。

项目基本信息

一、项目介绍

langgraph是langchain生态系统中专门用于构建复杂AI工作流的框架。它的核心理念是将智能体的执行过程抽象为图结构——节点代表执行步骤，边代表状态流转。这种基于图的建模方式，让开发者可以精确控制AI应用的所有执行路径，包括顺序执行、条件分支、循环迭代、并行处理等复杂逻辑。

与传统的链式框架相比，langgraph最大的优势在于其灵活性和可控性。在langchain中，链是线性的执行序列，一旦定义就无法在运行时动态改变。而langgraph中的图是状态机驱动的，每个节点执行后可以根据当前状态决定下一步走向，支持循环、重试、人机交互等高级模式。这种设计让构建能够自主决策、自我纠错、与用户交互的复杂智能体成为可能。

langgraph的核心组件包括状态定义、节点函数和边规则。状态是一个可变的字典对象，在整个图执行过程中传递和更新。节点是图的基本执行单元，可以是大语言模型调用、工具执行、人工输入或任意自定义函数。边定义了节点之间的转移规则，可以是固定的顺序边，也可以是根据状态动态判断的条件边。此外，langgraph还支持检查点机制，可以自动保存和恢复执行状态，让工作流具备容错和断点续传能力。

二、核心优势

开源免费
langgraph采用MIT许可证，代码完全开放。作为langchain官方项目，它与langchain生态系统深度集成，可以无缝使用langchain的所有组件和工具。

社区支持
作为langchain生态的核心项目，langgraph拥有强大的社区支持。官方文档完善，示例丰富，GitHub Issues响应及时。Discord社区中有专门的频道讨论langgraph的使用问题。

持续更新
从2026年3月31日的最后更新可以看出，langgraph保持着活跃的开发状态。新功能如并行节点、子图、持久化存储等持续加入，框架的稳定性和性能不断提升。

功能丰富
langgraph提供了构建复杂工作流的完整能力：

• 状态管理：支持自定义状态模式，自动管理状态流转
• 节点系统：支持多种节点类型，包括模型节点、工具节点、函数节点、人工节点
• 边规则：支持固定边、条件边、动态边，精确控制执行路径
• 循环支持：原生支持循环执行，适合构建自我迭代的智能体
• 并行执行：支持节点级别的并行执行，提升效率
• 检查点机制：自动保存执行状态，支持恢复和回溯
• 子图支持：支持嵌套图，实现模块化设计

三、适用场景

开发者学习和参考
对于希望深入理解智能体编排机制的开发者，langgraph是极佳的学习材料。通过研究其图执行引擎的实现，可以学习状态机、工作流编排、错误处理等系统设计知识。

个人项目使用和集成
如果你正在开发一个需要多步骤、分支逻辑的AI应用，langgraph是理想的选择。比如构建一个多轮对话的客服系统，需要根据用户意图分支到不同的处理流程；或者构建一个数据分析助手，需要循环迭代直到获得满意的分析结果。

企业级应用开发
对于需要构建生产级AI工作流的企业，langgraph提供了可靠的编排能力。它的状态持久化支持让长时间运行的任务可以安全中断和恢复；它的人工节点支持让人机协同成为可能；它的并行执行能力可以显著提升复杂任务的吞吐量。

日常工作和效率提升
对于AI工程师，langgraph是处理复杂业务逻辑的利器。通过图形化的方式组织代码，可以让复杂的流程变得清晰易懂。配合检查点功能，调试和测试变得更加便捷。

四、安装教程

系统要求

安装步骤

步骤一：创建虚拟环境（推荐）

python -m venv langgraph_env
source langgraph_env/bin/activate  # Linux/macOS
# 或 langgraph_env\Scripts\activate  # Windows

步骤二：安装langgraph

# 安装核心包
pip install langgraph

# 如果需要与langchain集成
pip install langchain langchain-openai

步骤三：验证安装

python -c "from langgraph.graph import StateGraph; print('安装成功')"

五、使用示例

示例一：基础图构建

创建一个最简单的顺序执行图：

from typing import TypedDict
from langgraph.graph import StateGraph, END

# 定义状态结构
class State(TypedDict):
    messages: list
    current_step: str

# 定义节点函数
def step_1(state: State) -> State:
    print("执行步骤1")
    return {"messages": state["messages"] + ["步骤1完成"], "current_step": "step1"}

def step_2(state: State) -> State:
    print("执行步骤2")
    return {"messages": state["messages"] + ["步骤2完成"], "current_step": "step2"}

# 构建图
builder = StateGraph(State)
builder.add_node("step1", step_1)
builder.add_node("step2", step_2)

# 添加边
builder.set_entry_point("step1")
builder.add_edge("step1", "step2")
builder.add_edge("step2", END)

# 编译并执行
graph = builder.compile()
result = graph.invoke({"messages": [], "current_step": ""})
print(result)

示例二：条件分支图

根据状态动态决定执行路径：

from typing import Literal
from langgraph.graph import StateGraph, END

class State(TypedDict):
    query: str
    intent: str
    result: str

def classify_intent(state: State) -> State:
    # 模拟意图分类
    if "天气" in state["query"]:
        intent = "weather"
    elif "新闻" in state["query"]:
        intent = "news"
    else:
        intent = "unknown"
    return {"intent": intent}

def handle_weather(state: State) -> State:
    return {"result": f"查询{state['query']}的天气结果"}

def handle_news(state: State) -> State:
    return {"result": f"查询{state['query']}的新闻结果"}

def handle_unknown(state: State) -> State:
    return {"result": "抱歉，我不理解您的问题"}

# 定义路由函数
def route_by_intent(state: State) -> Literal["weather", "news", "unknown"]:
    return state["intent"]

# 构建图
builder = StateGraph(State)
builder.add_node("classify", classify_intent)
builder.add_node("weather", handle_weather)
builder.add_node("news", handle_news)
builder.add_node("unknown", handle_unknown)

builder.set_entry_point("classify")
builder.add_conditional_edges("classify", route_by_intent)
builder.add_edge("weather", END)
builder.add_edge("news", END)
builder.add_edge("unknown", END)

graph = builder.compile()
result = graph.invoke({"query": "今天天气怎么样", "intent": "", "result": ""})
print(result["result"])

示例三：循环迭代图

构建一个能够自我迭代优化的智能体：

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

class State(TypedDict):
    input: str
    current_output: str
    iterations: int
    is_satisfied: bool

def generate_response(state: State) -> State:
    """生成回答"""
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    response = llm.invoke([HumanMessage(content=state["input"])])
    return {
        "current_output": response.content,
        "iterations": state["iterations"] + 1
    }

def evaluate_quality(state: State) -> State:
    """评估回答质量"""
    # 模拟质量评估逻辑
    quality = len(state["current_output"]) > 50
    return {"is_satisfied": quality}

def should_continue(state: State) -> Literal["generate", "end"]:
    """决定是否继续迭代"""
    if state["is_satisfied"] or state["iterations"] >= 3:
        return "end"
    return "generate"

def finalize(state: State) -> State:
    """输出最终结果"""
    print(f"最终回答（经过{state['iterations']}轮迭代）")
    return state

# 构建图
builder = StateGraph(State)
builder.add_node("generate", generate_response)
builder.add_node("evaluate", evaluate_quality)
builder.add_node("finalize", finalize)

builder.set_entry_point("generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges("evaluate", should_continue, {
    "generate": "generate",
    "end": "finalize"
})
builder.add_edge("finalize", END)

graph = builder.compile()
result = graph.invoke({
    "input": "写一篇关于人工智能的短文",
    "current_output": "",
    "iterations": 0,
    "is_satisfied": False
})

示例四：带人工节点的图

让人工介入决策过程：

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver

class State(TypedDict):
    document: str
    approved: bool
    feedback: str

def generate_draft(state: State) -> State:
    """生成文档草稿"""
    return {"document": f"草稿：{state['document']}"}

def human_review(state: State) -> State:
    """人工审核节点"""
    # 在实际应用中，这里会等待人工输入
    # 这里模拟人工决策
    user_input = input("是否批准文档？(y/n): ")
    return {"approved": user_input.lower() == "y"}

def revise(state: State) -> State:
    """根据反馈修改"""
    feedback = input("请输入修改意见: ")
    return {"document": f"修改后：{state['document']}\n反馈：{feedback}"}

def publish(state: State) -> State:
    """发布文档"""
    print("文档已发布")
    return state

def route_after_review(state: State) -> Literal["revise", "publish"]:
    return "publish" if state["approved"] else "revise"

# 构建图
builder = StateGraph(State)
builder.add_node("generate", generate_draft)
builder.add_node("human_review", human_review)
builder.add_node("revise", revise)
builder.add_node("publish", publish)

builder.set_entry_point("generate")
builder.add_edge("generate", "human_review")
builder.add_conditional_edges("human_review", route_after_review)
builder.add_edge("revise", "human_review")
builder.add_edge("publish", END)

# 启用检查点，支持状态持久化
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 执行
config = {"configurable": {"thread_id": "doc_review_1"}}
result = graph.invoke({"document": "初始内容"}, config)

示例五：并行执行图

同时执行多个独立任务：

from langgraph.graph import StateGraph, END
from langgraph.constants import Send

class State(TypedDict):
    items: list
    results: dict

def prepare_items(state: State) -> list[Send]:
    """准备并行任务"""
    # 返回多个Send对象，每个代表一个并行任务
    return [
        Send("process_item", {"item": item})
        for item in state["items"]
    ]

def process_item(state: dict) -> dict:
    """处理单个项目"""
    item = state["item"]
    # 模拟处理逻辑
    result = f"处理完成: {item}"
    return {"item": item, "result": result}

def aggregate_results(state: State, results: list) -> State:
    """聚合并行结果"""
    aggregated = {}
    for r in results:
        aggregated[r["item"]] = r["result"]
    return {"results": aggregated}

# 构建图
builder = StateGraph(State)
builder.add_node("prepare", prepare_items)
builder.add_node("process_item", process_item)
builder.add_node("aggregate", aggregate_results)

builder.set_entry_point("prepare")
builder.add_conditional_edges("prepare", lambda s: s)  # Send分发
builder.add_edge("process_item", "aggregate")
builder.add_edge("aggregate", END)

graph = builder.compile()
result = graph.invoke({"items": ["任务A", "任务B", "任务C"], "results": {}})
print(result["results"])

示例六：使用LangGraph与LangChain工具集成

结合langchain的工具系统构建智能体：

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.messages import HumanMessage

@tool
def search_web(query: str) -> str:
    """搜索网页"""
    return f"搜索结果：{query}的相关信息"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except:
        return "计算错误"

class State(TypedDict):
    messages: list
    tools_used: list

def agent_node(state: State) -> State:
    """智能体节点"""
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    tools = [search_web, calculate]
    
    agent = create_tool_calling_agent(llm, tools)
    executor = AgentExecutor(agent=agent, tools=tools)
    
    result = executor.invoke({"input": state["messages"][-1].content})
    return {
        "messages": state["messages"] + [AIMessage(content=result["output"])],
        "tools_used": state["tools_used"] + [result.get("intermediate_steps", [])]
    }

# 构建图
builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.set_entry_point("agent")
builder.add_edge("agent", END)

graph = builder.compile()
result = graph.invoke({
    "messages": [HumanMessage(content="15加27等于多少？然后再搜一下天气")],
    "tools_used": []
})
print(result["messages"][-1].content)

六、常见问题

问题一：图执行卡住或无限循环

原因：条件边缺少终止条件，或循环逻辑设计不当。

解决方案：

• 确保条件边函数在所有情况下都能返回有效值
• 添加最大迭代次数限制
• 使用检查点功能调试执行过程

问题二：状态更新不正确

原因：节点函数返回值覆盖了不应修改的状态字段。

解决方案：

• 节点函数返回字典只包含要更新的字段
• 使用TypedDict明确定义状态结构
• 考虑使用add_node的update参数控制更新行为

问题三：并行执行不生效

原因：未正确使用Send API。

解决方案：

• 在分发节点返回Send对象列表
• 确保每个Send指定正确的目标节点
• 使用聚合节点收集所有结果

问题四：检查点恢复失败

原因：检查点配置不正确或状态不可序列化。

解决方案：

• 使用MemorySaver或SqliteSaver作为检查点存储器
• 确保状态中的所有对象都可序列化（JSON兼容）
• 为每个执行线程提供唯一的thread_id

问题五：子图执行报错

原因：状态在父子图之间传递不正确。

解决方案：

• 使用add_subgraph方法添加子图
• 确保子图的状态是父图状态的子集
• 检查子图的输入输出映射配置

七、总结

langgraph是langchain生态中处理复杂工作流的利器。它将智能体的执行过程建模为图结构，通过节点和边的组合，让开发者可以精确控制应用的执行逻辑。与传统的链式框架相比，langgraph提供了无与伦比的灵活性和可控性，特别适合构建需要分支判断、循环迭代、人工介入的复杂AI应用。

与其他工作流框架相比，langgraph最大的特色是与langchain生态的深度集成。你可以无缝使用langchain的所有模型、工具和组件，构建功能强大的智能体。同时，langgraph的检查点机制为生产级应用提供了可靠的状态管理能力。

如果你正在开发复杂的AI应用，需要处理多步骤、分支逻辑、循环迭代等场景，langgraph是值得深入学习的工具。它不仅能帮你构建更强大的应用，更重要的是，它提供了一种系统化的方法来思考和设计AI工作流，让复杂的逻辑变得清晰可控。