🐝 用 LangGraph 构建生产级 AI Agent：从概念到实战

2026年，AI Agent 已经从实验室走向生产环境。从客服自动化到复杂的数据分析流水线，Agent 正在重塑我们构建智能应用的方式。然而，许多开发者面临的共同问题是：如何从”调用一次大模型”进化到”编排一个能自主决策的多步骤 Agent”？

LangGraph 正是为解决这个问题而生。作为 LangChain 生态的编排层，它用图结构（Graph）来定义 Agent 的状态流转，让复杂的多步骤推理变得可控、可观测、可调试。

为什么需要 LangGraph？

先看看痛点。用原生 LangChain 构建 Agent 时，你很快就会遇到这些问题：

• 循环控制困难：Agent 需要反复调用工具直到完成目标，但循环逻辑容易失控
• 状态管理混乱：多轮对话、工具调用结果、中间推理状态散落在各处
• 中断与审批：生产环境中需要人工审批某些操作，原生方案难以优雅实现
• 可观测性差：出了问题不知道 Agent 在哪一步走错了

LangGraph 的核心思想是：把 Agent 建模为状态机。节点（Node）是执行步骤，边（Edge）是流转条件，全局状态（State）贯穿整个执行过程。这种范式让一切变得显式可控。

核心概念速览

理解 LangGraph 只需要掌握四个核心概念：

1. State（状态）：一个 TypedDict，存储 Agent 运行时的所有数据
2. Node（节点）：一个函数，接收 State 并返回更新后的 State
3. Edge（边）：定义节点之间的流转关系
4. Graph（图）：将节点和边组装成可编译的工作流

实战：构建一个智能代码审查 Agent

让我们通过一个实际案例来学习。我们要构建一个能自动审查 GitHub PR 的 Agent，它包含以下步骤：获取代码变更 → 静态分析 → LLM 深度审查 → 生成报告。

第一步：定义状态

from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import add_messages

class CodeReviewState(TypedDict):
    """Agent 的全局状态"""
    repo_url: str
    pr_number: int
    diff_content: str              # PR 的代码变更
    static_issues: List[str]       # 静态分析发现的问题
    review_comments: Annotated[list, add_messages]  # 审查意见（自动合并）
    final_report: Optional[str]    # 最终审查报告
    current_step: str              # 当前执行步骤
    retry_count: int               # 重试次数

注意 Annotated[list, add_messages] 的用法——这是 LangGraph 的 Reducer 机制。多个节点可以往同一个字段追加数据，而不是覆盖。

第二步：实现节点函数

import subprocess
import httpx

def fetch_pr_diff(state: CodeReviewState) -> CodeReviewState:
    """节点1：获取 PR 的代码变更"""
    # 实际项目中调用 GitHub API
    # 这里用模拟数据演示
    state["diff_content"] = """
    def process_payment(amount):
        # 潜在问题：没有参数校验
        result = amount * 1.1
        return result
    """
    state["current_step"] = "fetch_diff"
    return state

def static_analysis(state: CodeReviewState) -> CodeReviewState:
    """节点2：运行静态代码分析"""
    issues = []
    diff = state["diff_content"]
    
    # 简化的静态检查逻辑
    if "eval(" in diff:
        issues.append("⚠️ 检测到 eval() 使用，存在代码注入风险")
    if "password" in diff.lower() or "secret" in diff.lower():
        issues.append("🔒 疑似硬编码敏感信息")
    if "except:" in diff and "except Exception" not in diff:
        issues.append("⚠️ 捕获了过于宽泛的异常")
    
    state["static_issues"] = issues
    state["current_step"] = "static_analysis"
    return state

def llm_review(state: CodeReviewState) -> CodeReviewState:
    """节点3：使用 LLM 进行深度代码审查"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    
    prompt = f"""你是一个资深代码审查员。请审查以下代码变更：

代码变更：
{state['diff_content']}

静态分析发现的问题：
{chr(10).join(state['static_issues']) if state['static_issues'] else '无'}

请给出具体的改进建议，格式为 Markdown。"""
    
    response = llm.invoke(prompt)
    state["review_messages"].append({
        "role": "assistant",
        "content": response.content
    })
    state["current_step"] = "llm_review"
    return state

def generate_report(state: CodeReviewState) -> CodeReviewState:
    """节点4：汇总生成最终审查报告"""
    issues_summary = "\n".join(
        f"- {issue}" for issue in state["static_issues"]
    ) or "未发现静态问题"
    
    llm_comments = "\n".join(
        msg["content"] for msg in state["review_messages"]
        if msg["role"] == "assistant"
    )
    
    report = f"""# 🔍 代码审查报告

**仓库：** {state['repo_url']}
**PR：** #{state['pr_number']}

## 静态分析结果
{issues_summary}

## LLM 深度审查
{llm_comments}

## 审查结论
审查完成，请根据以上建议进行修改。
"""
    state["final_report"] = report
    state["current_step"] = "report_generated"
    return state

第三步：组装图并添加条件分支

from langgraph.graph import StateGraph, END

def should_retry(state: CodeReviewState) -> str:
    """条件边：判断是否需要重试"""
    if state["retry_count"] >= 3:
        return "max_retries_reached"
    if not state["diff_content"]:
        return "fetch_diff"  # 重新获取
    return "static_analysis"

def has_critical_issues(state: CodeReviewState) -> str:
    """条件边：有关键问题时需要人工介入"""
    critical_keywords = ["注入", "硬编码", "泄露"]
    for issue in state["static_issues"]:
        if any(kw in issue for kw in critical_keywords):
            return "human_review"
    return "llm_review"

# 构建图
workflow = StateGraph(CodeReviewState)

# 添加节点
workflow.add_node("fetch_diff", fetch_pr_diff)
workflow.add_node("static_analysis", static_analysis)
workflow.add_node("llm_review", llm_review)
workflow.add_node("generate_report", generate_report)
workflow.add_node("human_review", lambda s: {**s, "current_step": "awaiting_human"})

# 添加边
workflow.set_entry_point("fetch_diff")
workflow.add_conditional_edges("fetch_diff", should_retry)
workflow.add_conditional_edges("static_analysis", has_critical_issues)
workflow.add_edge("llm_review", "generate_report")
workflow.add_edge("human_review", "static_analysis")  # 人工修改后重新分析
workflow.add_edge("generate_report", END)

# 编译
app = workflow.compile()

这个图的关键亮点是 条件边（Conditional Edges）。should_retry 和 has_critical_issues 让 Agent 能根据运行时状态动态决定下一步，这是传统线性 Pipeline 无法做到的。

第四步：执行与流式输出

import asyncio

async def run_review():
    initial_state = {
        "repo_url": "https://github.com/example/project",
        "pr_number": 42,
        "diff_content": "",
        "static_issues": [],
        "review_messages": [],
        "final_report": None,
        "current_step": "start",
        "retry_count": 0,
    }
    
    # 流式执行，实时看到每一步的输出
    async for event in app.astream(initial_state):
        for node_name, node_output in event.items():
            print(f"✅ 完成节点: {node_name}")
            print(f"   当前步骤: {node_output.get('current_step')}")
            if node_output.get("final_report"):
                print("\n📋 最终报告:")
                print(node_output["final_report"])

asyncio.run(run_review())

生产环境最佳实践

from langgraph.checkpoint.postgres import PostgresSaver

# 使用 PostgreSQL 作为状态存储
with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db") as checkpointer:
    app = workflow.compile(checkpointer=checkpointer)
    
    # 通过 thread_id 追踪会话
    config = {"configurable": {"thread_id": "review-pr-42"}}
    result = app.invoke(initial_state, config)

from langgraph.graph import interrupt

def execute_deployment(state):
    """部署节点：需要人工审批"""
    approval = interrupt(
        f"确认部署 {state['service_name']} 到生产环境？"
        f"变更内容：{state['change_summary']}"
    )
    if approval != "approved":
        raise Exception("部署被拒绝")
    # 执行部署...
    return state

性能优化技巧

• 并行节点：当多个节点之间没有依赖关系时，LangGraph 会自动并行执行。合理拆分节点可以显著降低端到端延迟
• 子图复用：将通用的 Agent 逻辑（如”搜索+总结”）封装为子图，在不同工作流中复用
• 模型分级：简单判断用便宜的小模型（如 GPT-4o-mini），复杂推理用强模型（如 Claude Sonnet），在节点级别指定模型
• 缓存层：对确定性节点（如静态分析）的结果做缓存，避免重复计算

总结

LangGraph 代表了 AI Agent 编排的一个重要范式转变：从”黑盒式”的 ReAct Agent 到”白盒式”的图编排。它让开发者能够精确控制 Agent 的每一步决策，同时保留了必要的灵活性。

当然，没有银弹。对于简单的单轮问答，直接调用大模型可能更高效；但对于涉及多步骤推理、工具调用、人工审批和状态管理的复杂场景，LangGraph 提供了目前最成熟的解决方案之一。

如果你正在构建下一个 AI 应用，建议从一个小而完整的 Agent 开始——定义好状态、画好图、跑通流程——然后逐步迭代。工具只是工具，真正有价值的是你对业务逻辑的深刻理解。

📚 参考资源：

• LangGraph 官方文档
• GitHub 仓库
• LangGraph 多 Agent 协作模式深度解析（后续文章预告）