当大模型不再“只动嘴皮子”而是真正学会使用工具、记住上下文AI Agent 的时代才算真正到来。引言如果你还停留在“调 API 做对话补全”的阶段那么你可能正在错过当下最火热的技术浪潮——AI Agent。从 AutoGPT 到 Manus从 Devin 到各种“自主执行”的智能体业界已经形成了一个共识LLM 的下一个范式就是 Agentic Workflow。然而市面上很多教程要么止步于 OpenAI 的function calling简单示例要么直接封装高级框架让读者看不清底层原理。今天我将带你用LangGraph—— 一个专为构建有状态、多步骤 LLM 应用而生的框架从零搭建一个真正能用的 AI Agent。这个 Agent 不仅能调用外部工具搜索引擎、计算器还能在多次对话中“记住”历史并且支持人工打断和状态回溯。技术栈概览LangGraph (≥0.1.0)OpenAI GPT-4o (或兼容 API)Tavily Search API实时搜索工具Redis持久化记忆可选全文代码均可在 Python 3.10 环境运行完整项目已附在文末。第一步理解 LangGraph 的核心 —— 图、状态与节点传统 ReAct Agent 依赖于 while 循环和手动维护消息列表扩展性和可调试性极差。LangGraph 将 Agent 逻辑建模为一个有向图State一个共享的字典或 TypedDict在所有节点间传递。Node一个 Python 函数接收 State修改后返回新的 State。Edge连接节点决定执行流程。我们设计的 Agent 将包含以下节点call_model调用 LLM根据当前状态决定是回复还是调用工具。call_tool执行具体的工具如搜索、计算。should_continue条件边判断是否需要继续调用工具。最终流程图如下text[START] -- call_model -- (条件边) |-- 若需要工具 -- call_tool -- (回到 call_model) |-- 若完成 -- END第二步环境配置与基础状态定义创建项目并安装依赖bashpip install langgraph langchain-openai tavily-python redis设置环境变量建议使用.env文件textOPENAI_API_KEYsk-xxx TAVILY_API_KEYtvly-xxx REDIS_URLredis://localhost:6379现在编写agent_state.pypythonfrom typing import TypedDict, Annotated, List, Union from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage import operator class AgentState(TypedDict): Agent 状态定义 # 消息历史使用 operator.add 实现自动追加 messages: Annotated[List[BaseMessage], operator.add] # 可选记录当前步骤数 step_count: int这里使用Annotated类型配合operator.add可以让每次节点返回的messages自动合并而非覆盖。第三步定义工具Tool我们实现两个最常用工具搜索引擎实时信息和计算器精确数学。创建tools.pypythonfrom langchain_core.tools import tool from tavily import TavilyClient import os tavily TavilyClient(api_keyos.getenv(TAVILY_API_KEY)) tool def web_search(query: str) - str: 在互联网上搜索最新信息。输入应为搜索关键词或问题。 try: response tavily.search(queryquery, max_results3) # 提取摘要 results [f标题: {r[title]}\n内容: {r[content]} for r in response[results]] return \n\n.join(results) if results else 未找到相关信息。 except Exception as e: return f搜索失败: {str(e)} tool def calculator(expression: str) - str: 计算数学表达式例如 2 3 * 4。 try: # 安全求值注意生产环境建议使用 ast.literal_eval 或 numexpr result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)}为了让 LLM 知道如何调用我们需要将工具转换为 OpenAI 兼容的 JSON Schema。第四步核心图构建逻辑现在编写agent_graph.py这是整篇文章的灵魂。pythonfrom langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from agent_state import AgentState from tools import web_search, calculator import json # 初始化 LLM推荐 gpt-4ogpt-3.5-turbo 也能工作 llm ChatOpenAI(modelgpt-4o, temperature0) # 绑定工具 tools [web_search, calculator] tools_by_name {tool.name: tool for tool in tools} llm_with_tools llm.bind_tools(tools) def call_model(state: AgentState): 节点1调用 LLM 并返回消息 messages state[messages] response llm_with_tools.invoke(messages) # 返回新增的消息LangGraph 会自动添加到 messages 列表 return {messages: [response]} def call_tool(state: AgentState): 节点2执行 LLM 请求的工具 last_message state[messages][-1] tool_calls last_message.tool_calls results [] for tc in tool_calls: tool_func tools_by_name.get(tc[name]) if tool_func: # 调用真正的工具函数 observation tool_func.invoke(tc[args]) tool_message ToolMessage( contentobservation, tool_call_idtc[id] ) results.append(tool_message) else: results.append(ToolMessage( contentf错误未找到工具 {tc[name]}, tool_call_idtc[id] )) return {messages: results} def should_continue(state: AgentState): 条件边判断是否还需要继续调用工具 last_message state[messages][-1] # 如果 LLM 返回了 tool_calls则进入工具节点 if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools else: return end # 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) workflow.add_node(tools, call_tool) # 设置入口 workflow.set_entry_point(agent) # 添加条件边 workflow.add_conditional_edges( agent, should_continue, { tools: tools, end: END } ) # 工具节点执行后回到 agent 节点形成循环 workflow.add_edge(tools, agent) # 编译图 app workflow.compile()就这么简单是的。现在我们已经拥有一个能够自动迭代工具调用的 Agent。第五步加入持久化记忆Checkpointer上面 Agent 每次运行都是独立的无法延续对话。LangGraph 内置了Checkpointer机制可以将状态快照保存到内存或 Redis。我们使用 Redis 实现跨请求的记忆。pythonfrom langgraph.checkpoint.redis import RedisSaver # 创建 Redis 检查点保存器 checkpointer RedisSaver.from_url(os.getenv(REDIS_URL, redis://localhost:6379)) # 必须先设置连接 checkpointer.setup() # 编译时传入 checkpointer app workflow.compile(checkpointercheckpointer)现在每次执行都需要提供一个config其中包含thread_id用于区分不同对话会话pythonconfig {configurable: {thread_id: user_123_session_1}} result app.invoke({messages: [HumanMessage(content今天纽约天气怎么样)]}, configconfig)之后同样的thread_id再次调用Agent 会自动加载历史消息。第六步运行演示 —— 让 Agent 干活编写run.pypythonfrom agent_graph import app from langchain_core.messages import HumanMessage import uuid def ask_agent(user_input: str, session_id: str None): if not session_id: session_id str(uuid.uuid4()) config {configurable: {thread_id: session_id}} final_state app.invoke( {messages: [HumanMessage(contentuser_input)]}, configconfig ) # 获取最后一条 AI 消息 last_msg final_state[messages][-1] return last_msg.content, session_id if __name__ __main__: # 测试1需要搜索 resp, sid ask_agent(2025年奥斯卡最佳影片是哪部) print(fAI: {resp}\n) # 测试2需要计算 resp, _ ask_agent(计算 (12345 * 6789) / 2, sid) print(fAI: {resp}\n) # 测试3延续记忆询问“刚才那个数字是多少” resp, _ ask_agent(刚才计算的结果是什么, sid) print(fAI: {resp})预期输出示例textAI: 2025年奥斯卡最佳影片是《阿诺拉》(Anora)... AI: 计算结果: 41926802.5 AI: 刚才计算的结果是 41926802.5。第七步进阶 —— 人机协同与断点调试LangGraph 最强大的功能之一是可以在节点执行前“中断”等待人工输入。例如在执行call_tool之前暂停让用户确认搜索关键词。修改编译参数pythonapp workflow.compile( checkpointercheckpointer, interrupt_before[tools] # 在执行 tools 节点前中断 )运行时我们可以获取当前状态并决定是否继续pythonconfig {configurable: {thread_id: test_interrupt}} for event in app.stream({messages: [HumanMessage(搜索量子计算机最新进展)]}, config): if event.get(__interrupt__): print(Agent 想要调用工具是否继续(y/n)) user_input input() if user_input.lower() y: # 恢复执行 app.invoke(None, configconfig) else: break这个特性使得构建“半自主”Agent 变得极其简单非常适合企业内部审核流程。性能与成本优化建议减少不必要的工具调用在call_model节点中可以添加一个前置分类器例如一个小型 LLM判断当前问题是否需要工具。缓存工具结果对于相同的搜索查询使用 Redis 缓存web_search的结果。流式输出app.stream()返回生成器可以逐 token 输出提升用户体验。工具超时使用asyncio.timeout或者concurrent.futures为外部 API 调用设置超时。总结与展望通过不到 200 行代码我们已经构建了一个具备以下特性的 AI Agent✅ 自动调用搜索引擎和计算器✅ 多轮对话记忆基于 Redis✅ 可中断、可回溯Checkpointer✅ 完全可控的图结构易于插入新节点例如 RAG 检索节点当前这个 Agent 仍然比较“原子化”你可以轻松扩展加入代码解释器、数据库查询、文件读写等工具。配合 LangSmith 还可以进行完整的追踪和调试。AI Agent 的未来不在于大模型本身有多强而在于我们如何编排它们与外部世界的交互。LangGraph是目前最优雅的答案之一。希望你能够通过这篇文章真正迈入 Agentic AI 的大门。完整的代码仓库欢迎在评论区留言获取 GitHub 链接或者自行将所有代码块组合。互动问题你在实际项目中遇到了哪些 Agent 编排的痛点欢迎留言讨论我会挑选典型问题在下篇博客中分析。
