LangGraph 安装与使用指南：构建弹性有状态AI代理 | 极客的赛博空间

LangGraph 是一个低级别编排框架，用于构建、管理和部署长期运行、有状态的AI代理。它提供了持久化执行、人机协作、综合记忆和调试等核心能力，被Klarna、Replit、Elastic等公司用于构建生产级AI应用。

1. 安装

LangGraph 是一个Python库，通过 pip 安装非常简单。

1.1 基本安装

1	pip install -U langgraph

1.2 可选依赖（按需安装）

根据您需要的功能，可以安装额外的依赖组：

# 如果需要与LangSmith集成进行调试
pip install -U langgraph langsmith

# 如果需要使用内置的持久化/检查点功能（推荐生产环境）
pip install -U langgraph-checkpoint

1.3 验证安装

在Python环境中运行以下命令，确认安装成功：

1 2	import langgraph print(langgraph.__version__)

2. 核心概念与基本使用

LangGraph 的核心是构建一个状态图 (StateGraph)，其中节点（Nodes）代表执行步骤，边（Edges）定义执行流程，状态（State）在整个图执行过程中传递和演变。

2.1 定义状态结构

首先，定义您的代理状态。通常使用 TypedDict 或 Pydantic BaseModel。

from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[dict]  # 对话消息列表
    next_step: str        # 下一步要执行的动作
    # 可以添加任何其他需要的字段

2.2 创建节点函数

节点是执行实际工作的函数。它们接收当前状态，处理后返回状态的部分更新。

def research_node(state: AgentState) -> dict:
    # 模拟研究步骤
    print("执行研究...")
    # 这里可以调用LLM、API或工具
    return {"messages": state["messages"] + [{"role": "assistant", "content": "研究完成"}]}

def generate_node(state: AgentState) -> dict:
    print("生成最终答案...")
    return {"messages": state["messages"] + [{"role": "assistant", "content": "这是生成的答案"}]}

2.3 构建图并添加节点和边

from langgraph.graph import StateGraph, END

# 创建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("research", research_node)
workflow.add_node("generate", generate_node)

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

# 添加边，定义执行流程
workflow.add_edge("research", "generate")  # research -> generate
workflow.add_edge("generate", END)         # generate -> 结束

# 编译图
app = workflow.compile()

2.4 执行图（代理）

使用 invoke 方法（同步）或 ainvoke（异步）来运行图。

# 初始化状态
initial_state = {
    "messages": [{"role": "user", "content": "请帮我研究一个主题"}],
    "next_step": ""
}

# 运行代理
result = app.invoke(initial_state)
print(result["messages"])

3. 核心高级功能

3.1 持久化与检查点 (Persistence & Checkpointing)

LangGraph 支持将图的状态保存到检查点，以实现持久化执行和从失败中恢复。这对于长时间运行或需要人工介入的代理至关重要。

使用 MemorySaver（内存存储，适合测试）：

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

使用 SqliteSaver（持久化到SQLite，适合生产）：

1	pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

# 连接到SQLite数据库
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    app = workflow.compile(checkpointer=saver)

3.2 人机协作 (Human-in-the-loop)

您可以配置图在特定节点中断执行，等待人工输入或审批后再继续。

from langgraph.graph import StateGraph

# ... 创建图和节点 ...

# 在 "human_review" 节点前中断
workflow.add_node("human_review", ...)
workflow.add_edge("research", "human_review")
workflow.add_edge("human_review", "generate")

# 编译时指定中断节点
app = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

# 运行到中断点...
# 获取当前状态...
# 人工检查并更新状态...
# 继续执行

3.3 流式输出 (Streaming)

支持实时流式输出节点的执行结果，提升用户体验。

# 流式执行
for event in app.stream(initial_state):
    for key, value in event.items():
        print(f"节点 '{key}' 输出了: {value}")

4. 与 LangChain 和其他工具集成

LangGraph 可以与 LangChain 生态无缝集成，利用其丰富的组件。

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.prebuilt import ToolExecutor

# 定义工具
@tool
def search(query: str) -> str:
    """模拟搜索工具"""
    return f"搜索 '{query}' 的结果"

# 创建LLM
llm = ChatOpenAI(model="gpt-4")

# 使用预构建的ToolExecutor和Agent节点
from langgraph.prebuilt import ToolExecutor, create_agent

tool_executor = ToolExecutor([search])

# 更简单：使用create_agent快速创建
agent = create_agent(llm, tools=[search])
result = agent.invoke({"messages": [("user", "帮我搜索LangGraph")]})

5. 常见问题排查

问题	可能原因	解决方案
导入错误 `ModuleNotFoundError`	依赖未完整安装	检查是否安装了 `langgraph` 及所有可选依赖（如 `langgraph-checkpoint`）。
状态更新未生效	节点函数未正确返回状态更新字典	确保节点函数返回一个字典，包含需要更新的状态键值对。
持久化状态丢失	检查点存储器（saver）未正确配置或使用	确保编译时传入了 `checkpointer` 参数，并且后续执行时使用了相同的 `thread_id` 配置。
执行卡住或超时	图循环或死锁，或步骤过长	检查图逻辑，确保有明确的结束条件。考虑增加超时设置或拆分长步骤。