LangGraph 实战：构建可靠的 AI Agent 多步骤工作流

随着大语言模型（LLM）能力的不断提升，AI Agent 已经从概念走向实际生产落地。然而，许多开发者在构建 Agent 时面临同一个难题：如何让 Agent 的行为更加可靠、可控、可调试？LangGraph 正是为了解决这一问题而生的框架，它通过图结构来组织 Agent 的决策流程，让复杂的多步骤自动化工作流变得清晰可维护。本文将带你从零开始，用 LangGraph 构建一个真实可用的 AI Agent 工作流。

为什么选择 LangGraph？

在 LangGraph 出现之前，构建多步骤 Agent 的常见方案是 ReAct 循环或简单的链式调用。这些方案在面对复杂场景时往往捉襟见肘：

• 状态管理混乱：多轮对话和中间结果难以持久化，调试困难
• 流程控制薄弱：条件分支、循环重试等逻辑难以优雅表达
• 并行能力缺失：多个子任务只能串行执行，效率低下
• 可观测性差：Agent 的每一步决策过程对开发者不透明

LangGraph 借鉴了有向图（DAG）和状态机的思想，将 Agent 的执行过程建模为节点（Node）+ 边（Edge）的图结构。每个节点负责一项具体功能，边定义节点之间的跳转条件，整个工作流的状态在节点间流转和累积。

核心概念与基础架构

在动手写代码之前，先理解 LangGraph 的三个核心概念：

• StateGraph：工作流的容器，持有全局状态 Schema，所有节点共享同一个状态对象
• Node（节点）：接收当前状态、执行操作、返回状态更新的函数
• Edge（边）：定义节点之间的跳转关系，支持条件边（Conditional Edge）实现动态路由

下面是一个最简单的 LangGraph 工作流骨架：

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

# 1. 定义状态 Schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str
    result: str

# 2. 定义节点函数
def planner_node(state: AgentState) -> dict:
    """规划器：决定下一步做什么"""
    # 调用 LLM 分析任务
    ...
    return {"next_action": "search"}

def executor_node(state: AgentState) -> dict:
    """执行器：调用工具执行任务"""
    ...
    return {"result": "执行结果"}

# 3. 构建图
builder = StateGraph(AgentState)
builder.add_node("planner", planner_node)
builder.add_node("executor", executor_node)

# 4. 添加边
builder.set_entry_point("planner")
builder.add_edge("planner", "executor")
builder.add_edge("executor", END)

graph = builder.compile()

实战：构建一个信息收集与总结 Agent

下面我们来构建一个真实场景下的 Agent：给定一个研究主题，Agent 自动搜索相关信息、提取关键内容、生成结构化报告。

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults
from typing import TypedDict, Annotated, List
import operator

llm = ChatOpenAI(model="gpt-4o", temperature=0)
search_tool = TavilySearchResults(max_results=5)

class ResearchState(TypedDict):
    topic: str
    search_queries: List[str]
    search_results: Annotated[list, operator.add]
    final_report: str
    iteration: int

def generate_queries(state: ResearchState) -> dict:
    """生成多个搜索关键词"""
    prompt = f"针对主题'{state['topic']}'，生成3个不同角度的搜索关键词，用换行分隔"
    response = llm.invoke(prompt)
    queries = response.content.strip().split("\n")
    return {"search_queries": queries[:3], "iteration": state.get("iteration", 0) + 1}

def execute_search(state: ResearchState) -> dict:
    """并行执行搜索"""
    all_results = []
    for query in state["search_queries"]:
        results = search_tool.invoke(query)
        all_results.extend(results)
    return {"search_results": all_results}

def synthesize_report(state: ResearchState) -> dict:
    """综合搜索结果生成报告"""
    context = "\n\n".join([r.get("content", "") for r in state["search_results"][:10]])
    prompt = f"""基于以下搜索结果，为主题'{state['topic']}'生成一份结构化研究报告：

{context}

要求：包含背景、核心发现、关键数据、结论四个部分"""
    response = llm.invoke(prompt)
    return {"final_report": response.content}

def should_refine(state: ResearchState) -> str:
    """条件边：判断是否需要精化搜索"""
    if state["iteration"] < 2 and len(state["search_results"]) < 5:
        return "refine"
    return "synthesize"

# 构建图
builder = StateGraph(ResearchState)
builder.add_node("generate_queries", generate_queries)
builder.add_node("execute_search", execute_search)
builder.add_node("synthesize_report", synthesize_report)

builder.set_entry_point("generate_queries")
builder.add_edge("generate_queries", "execute_search")
builder.add_conditional_edges("execute_search", should_refine, {
    "refine": "generate_queries",
    "synthesize": "synthesize_report"
})
builder.add_edge("synthesize_report", END)

graph = builder.compile()

# 运行 Agent
result = graph.invoke({"topic": "2025年大模型推理优化技术进展"})
print(result["final_report"])

上面的代码展示了 LangGraph 最强大的特性之一：条件循环。当搜索结果不足时，Agent 会自动返回到查询生成节点进行精化，直到满足条件才进入报告生成阶段，整个过程完全自动化。

生产环境的最佳实践

将 LangGraph Agent 推向生产环境时，以下几点至关重要：

• 持久化 Checkpointer：使用 SqliteSaver 或 PostgresSaver 保存工作流状态，支持断点续跑和人工介入（Human-in-the-loop）
• 流式输出：调用 graph.astream() 实现节点级流式返回，提升用户体验
• 超时与重试：在节点函数内包装 tenacity 重试逻辑，防止 LLM API 抖动导致整体失败
• 可观测性集成：接入 LangSmith 或 Phoenix，实现每个节点的 Trace 追踪和性能分析
• 并行节点：利用 Send() API 动态生成并行分支，大幅提升多任务处理效率

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph

# 添加持久化支持
memory = SqliteSaver.from_conn_string(":memory:")  # 生产环境换成文件路径
graph = builder.compile(checkpointer=memory)

# 带 thread_id 运行，支持多会话隔离
config = {"configurable": {"thread_id": "research-session-001"}}
result = graph.invoke({"topic": "量子计算商业化进展"}, config=config)

# 获取完整执行历史
for step in graph.get_state_history(config):
    print(f"节点: {step.metadata.get('step')}, 状态: {list(step.values.keys())}")

LangGraph 目前已成为构建生产级 AI Agent 的事实标准之一，其活跃的社区和持续迭代的 API 为开发者提供了可靠的保障。如果你正在考虑将 LLM 能力引入复杂的业务流程自动化，LangGraph 值得作为首选框架深入研究。