用 LangGraph 构建生产级 AI Agent：从概念到落地

2026 年，AI Agent 已经从”能聊天的机器人”进化为”能干活的智能体”。但很多人一上手就踩坑：状态管理混乱、工具调用出错、循环推理停不下来。这篇文章带你用 LangGraph 框架，从零搭一个靠谱的生产级 Agent。

为什么选 LangGraph？

LangChain 的 Chain 适合线性流程，但 Agent 的本质是图——有条件分支、有循环、有回溯。LangGraph 把 Agent 建模为状态图（StateGraph），每个节点是一个处理步骤，边是状态转移。这种范式天然契合 Agent 的决策循环。

核心优势：

• 显式状态管理：每一步的状态清晰可见，调试不再靠 print
• 原生支持循环：Agent 可以反复调用工具直到完成任务
• 人工审批节点：关键操作可以插入人工确认
• 流式输出：支持实时 token 流推送

架构设计：ReAct Agent 模式

ReAct（Reasoning + Acting）是当前最成熟的 Agent 模式。核心循环：思考 → 调用工具 → 观察结果 → 继续思考。用 LangGraph 表达就是：

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

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    next_step: str
    tool_calls: list
    result: str

# 2. 定义节点
def think_node(state: AgentState) -> AgentState:
    """LLM 思考：决定下一步做什么"""
    response = llm.invoke(state["messages"])
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "next_step": "tool" if response.tool_calls else "end",
        "tool_calls": response.tool_calls or []
    }

def tool_node(state: AgentState) -> AgentState:
    """执行工具调用"""
    results = []
    for call in state["tool_calls"]:
        tool = tools[call["function"]["name"]]
        output = tool.invoke(call["function"]["arguments"])
        results.append({
            "role": "tool",
            "tool_call_id": call["id"],
            "content": str(output)
        })
    return {
        "messages": results,
        "next_step": "think"  # 回到思考节点
    }

# 3. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("think", think_node)
workflow.add_node("tool", tool_node)
workflow.set_entry_point("think")

# 条件边：有工具调用就去 tool，否则结束
workflow.add_conditional_edges(
    "think",
    lambda s: "tool" if s["next_step"] == "tool" else END,
    {"tool": "tool", END: END}
)
workflow.add_edge("tool", "think")  # 工具执行完回到思考

app = workflow.compile()

这个图只有两个节点，但能处理任意深度的工具调用链。关键是 tool → think 这条回边，它构成了 Agent 的推理循环。

实战：带搜索能力的 AI 助手

光看理论不够，我们来搭一个能联网搜索、能读网页的实用 Agent。

第一步：定义工具

from langchain_community.tools import TavilySearchRun
from langchain_core.tools import tool
import requests
from bs4 import BeautifulSoup

@tool
def web_search(query: str, max_results: int = 3) -> str:
    """搜索互联网获取最新信息"""
    search = TavilySearchRun(max_results=max_results)
    results = search.invoke(query)
    return "\n".join([
        f"[{r['title']}]\n{r['url']}\n{r['content'][:200]}"
        for r in results
    ])

@tool
def fetch_page(url: str) -> str:
    """抓取网页内容并提取正文"""
    resp = requests.get(url, timeout=10, headers={
        "User-Agent": "Mozilla/5.0 (compatible; AIAgent/1.0)"
    })
    soup = BeautifulSoup(resp.text, "html.parser")
    # 过滤脚本和样式
    for tag in soup(["script", "style", "nav", "footer"]):
        tag.decompose()
    return soup.get_text(separator="\n")[:3000]

tools = {"web_search": web_search, "fetch_page": fetch_page}

第二步：带错误处理的 Agent 节点

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(
    [web_search, fetch_page]
)

def agent_node(state: AgentState) -> AgentState:
    messages = state["messages"]
    try:
        response = llm.invoke(messages)
    except Exception as e:
        return {
            "messages": [{"role": "assistant", "content": f"抱歉，模型调用出错: {e}"}],
            "next_step": "end"
        }
    
    tool_calls = response.tool_calls or []
    return {
        "messages": [response],
        "next_step": "execute_tools" if tool_calls else "end",
        "tool_calls": tool_calls
    }

def execute_tools_node(state: AgentState) -> AgentState:
    """执行工具调用，带错误恢复"""
    results = []
    for call in state["tool_calls"]:
        tool_name = call["function"]["name"]
        try:
            args = json.loads(call["function"]["arguments"])
            output = tools[tool_name].invoke(args)
        except Exception as e:
            output = f"工具 {tool_name} 执行失败: {e}"
        
        results.append({
            "role": "tool",
            "tool_call_id": call["id"],
            "content": str(output)
        })
    
    return {
        "messages": results,
        "next_step": "agent",
        "tool_calls": []
    }

# 构建完整工作流
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("execute_tools", execute_tools_node)
workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    lambda s: s["next_step"],
    {"execute_tools": "execute_tools", "end": END}
)
workflow.add_edge("execute_tools", "agent")

agent_app = workflow.compile()

第三步：运行与流式输出

# 同步调用
result = agent_app.invoke({
    "messages": [{"role": "user", "content": "2026年AI Agent领域有哪些最新突破？请搜索最新信息。"}]
})
print(result["messages"][-1]["content"])

# 流式输出（适合 Web 应用）
for chunk in agent_app.stream(
    {"messages": [{"role": "user", "content": "帮我查今天的科技新闻"}]},
    stream_mode="values"
):
    last_msg = chunk["messages"][-1]
    if last_msg.get("content"):
        print(last_msg["content"], end="", flush=True)

生产环境关键优化

1. 防止无限循环

Agent 最容易出的bug就是死循环。务必设置最大迭代次数：

def agent_node_with_limit(state: AgentState) -> AgentState:
    # 检查迭代次数
    iterations = len([m for m in state["messages"] if m["role"] == "assistant"])
    if iterations >= 10:
        return {
            "messages": [{"role": "assistant", "content": "已达到最大迭代次数，请简化您的问题。"}],
            "next_step": "end"
        }
    # ... 正常逻辑

2. 人工审批节点（Human-in-the-Loop）

涉及敏感操作时，插入人工确认：

from langgraph.graph.interrupt import interrupt

def approval_node(state: AgentState):
    """等待人工确认后再执行"""
    approved = interrupt({
        "question": f"Agent 即将执行 {state['tool_calls']}，是否允许？",
        "context": state["tool_calls"]
    })
    if approved:
        return {"next_step": "execute_tools"}
    else:
        return {
            "messages": [{"role": "assistant", "content": "操作已被用户取消"}],
            "next_step": "end"
        }

# 在图中插入审批节点
workflow.add_node("approval", approval_node)
workflow.add_edge("agent", "approval")  # agent 后先审批
workflow.add_conditional_edges("approval", ...)

3. 状态持久化与断点续传

长任务可能耗时几分钟，需要持久化：

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 持久化
checkpointer = SqliteSaver.from_conn_string("agent_memory.db")
agent_app = workflow.compile(checkpointer=checkpointer)

# 带线程 ID 的调用（支持多用户并发）
config = {"configurable": {"thread_id": "user_123"}}
result = agent_app.invoke(input_state, config)

性能对比与选型建议

选型建议：单 Agent 场景首选 LangGraph，多 Agent 协作考虑 CrewAI 或 AutoGen。如果是简单问答链，LangChain LLM 就够了，不需要图框架。

部署建议

生产部署推荐用 LangGraph Cloud 或自建 FastAPI 服务：

from fastapi import FastAPI
from langgraph.graph import StateGraph

app = FastAPI()

@app.post("/chat")
async def chat(query: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent_app.invoke(
        {"messages": [{"role": "user", "content": query}]},
        config
    )
    return {"response": result["messages"][-1]["content"]}

# 启动：uvicorn main:app --host 0.0.0.0 --port 8000

总结

LangGraph 把 Agent 开发从”手工艺”变成了”工程化”。核心思路：把 Agent 拆成状态 + 节点 + 边，每个节点职责单一，状态流转显式可控。记住三个关键点：

• 状态设计先行：先想清楚 Agent 需要什么信息，再设计 State
• 节点保持纯净：每个节点只做一件事，工具调用和推理分离
• 永远设安全阀：最大迭代、超时、人工审批，一个都不能少

AI Agent 时代已经到来，LangGraph 是目前最成熟的工程化框架之一。动手试试，从上面的代码开始，搭你自己的第一个 Agent 吧。