AI Agent是2026年最热门的技术方向之一。LangGraph作为LangChain团队推出的图状态机框架，已经成为构建复杂Agent的首选工具。本文带你从零搭建一个生产级Agent。
为什么选LangGraph？
LangGraph的核心优势在于它把Agent的工作流建模为一个有向图——每个节点是一个处理步骤，边定义了状态转移逻辑。这种设计带来三个关键好处：
- 可视化调试 — 工作流清晰可追踪
- 状态持久化 — 支持断点恢复和人工介入
- 循环与分支 — 原生支持复杂控制流
环境准备
pip install langgraph langchain-openai langchain-community
设置环境变量：
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
Step 1: 定义状态
状态是Agent在执行过程中维护的数据。我们定义一个包含消息历史和工具调用结果的状态：
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
tool_results: list
iteration: int
max_iterations: int
Step 2: 定义工具节点
工具节点负责执行AI选择的工具调用。这里我们用一个模拟的搜索工具作为示例：
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索互联网获取最新信息"""
# 实际项目中替换为真实搜索API
return f"搜索结果: {query}的相关信息..."
@tool
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression) # 生产环境请用ast.literal_eval
return str(result)
except Exception as e:
return f"计算错误: {e}"
tools = [search_web, calculate]
Step 3: 构建图
这是核心部分——定义Agent的决策和执行流程：
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: AgentState) -> AgentState:
"""AI决策节点：决定下一步行动"""
response = llm_with_tools.invoke(state["messages"])
return {
"messages": [response],
"iteration": state["iteration"] + 1
}
def tool_node(state: AgentState) -> AgentState:
"""工具执行节点：执行AI选择的工具"""
last_message = state["messages"][-1]
results = []
for tool_call in last_message.tool_calls:
# 动态调用工具
for t in tools:
if t.name == tool_call["name"]:
result = t.invoke(tool_call["args"])
results.append({"tool_call_id": tool_call["id"], "result": result})
return {"tool_results": results, "messages": []}
def should_continue(state: AgentState) -> str:
"""路由函数：决定下一步走向"""
last_message = state["messages"][-1]
if state["iteration"] >= state["max_iterations"]:
return "end"
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")
app = graph.compile()
Step 4: 运行Agent
from langchain_core.messages import HumanMessage
result = app.invoke({
"messages": [HumanMessage(content="帮我查一下2026年最新的AI编程工具，然后告诉我如果月预算50美元，应该选哪个")],
"tool_results": [],
"iteration": 0,
"max_iterations": 5
})
print(result["messages"][-1].content)
生产环境注意事项
1. 添加重试机制。 工具调用可能失败，用tenacity库添加指数退避重试。
2. 实现人工介入节点。 对敏感操作（如发送邮件、修改数据）添加human-in-the-loop检查点。
3. 集成监控。 接入LangSmith进行链路追踪，监控每次工具调用的延迟和成功率。
4. 限制迭代次数。 始终设置max_iterations防止Agent陷入死循环。
完整代码仓库
本文完整代码已上传GitHub，包含生产级的错误处理、日志记录和Docker部署配置。在公众号后台回复"langgraph"获取链接。
下一步学习建议：
- 尝试接入真实的搜索API（Tavily、SerpAPI）
- 实现多Agent协作（Supervisor模式）
- 添加持久化状态存储（SQLite/Redis）
掌握了LangGraph，你就掌握了构建复杂AI应用的核心能力。动手试试吧！