30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度这次我们来看一个关于 AI Agent智能体的实战教程。如果你对如何从零开始搭建一个能自主思考、执行任务的 AI 智能体感兴趣特别是想了解 LangChain、LangGraph 这些核心框架如何落地那么这篇文章就是为你准备的。AI Agent 的核心是让大语言模型LLM具备“思考-行动-观察”的循环能力而不仅仅是单次问答。本教程将手把手带你使用 LangChain 和 LangGraph 框架结合 OpenClaw 等工具搭建一个功能完整的智能体。重点不是空谈概念而是提供一套可运行、可验证的代码和部署方案让你能快速上手并理解其运作机制。本文将涵盖从环境搭建、核心概念理解、代码实现到功能测试的全流程。无论你是想将智能体集成到现有业务中还是单纯进行技术探索都能从中获得清晰的路径。我们会重点关注智能体的架构设计、工具调用、状态管理以及如何通过 LangGraph 实现复杂的多步骤工作流。1. 核心能力速览在深入代码之前我们先快速了解本次搭建的 AI Agent 智能体具备哪些核心能力以及你需要准备什么。能力项说明项目类型AI Agent智能体开发框架与实践教程核心框架LangChain编排基础、LangGraph工作流与状态管理关键组件大语言模型如 OpenAI GPT、智谱 GLM 等、工具Tools、记忆Memory、智能体Agent硬件门槛无特殊 GPU 要求。智能体的核心是逻辑编排和 API 调用主要依赖 CPU 和网络。大模型推理部分通常调用云端 API如 OpenAI本地如需运行开源模型则需相应算力。启动方式通过 Python 脚本启动提供 Web 界面或命令行交互两种方式根据具体实现而定。接口能力支持标准的 API 调用可集成到其他系统智能体本身也可作为服务对外提供问答或任务执行接口。批量任务通过工作流设计理论上支持异步和批量任务处理但需自行实现队列管理。适合场景自动化客服、数据分析助手、智能工作流编排、研究助理、个性化内容生成等需要多步骤、多工具协作的场景。2. 适用场景与使用边界AI Agent 不是万能的理解其适用边界能帮助你更好地设计应用。它适合谁开发者希望将 LLM 能力转化为可执行、可复用的智能应用。业务人员有明确的、可拆解为多步骤的自动化流程需求如信息搜集、报告生成。技术爱好者希望深入理解 Agent、Chain、Graph 等前沿概念及其实现。能解决什么问题复杂任务分解将“分析上周销售数据并生成总结报告”这样的模糊指令自动分解为查询数据库、计算指标、撰写文本等子任务。多工具协同自动判断并调用搜索引擎、计算器、代码解释器、绘图工具等完成一个综合目标。状态持久化与循环处理需要多轮交互、信息累积的任务如调试代码、深度调研。不适合什么场景对实时性要求极高的简单查询直接调用模型 API 更高效。涉及高度确定性、无模糊空间的逻辑计算传统编程更可靠。缺乏清晰步骤定义或成功标准的完全开放式创作。合规与安全边界工具调用安全智能体调用的外部工具如网络搜索、文件读写需设置权限边界防止越权操作。内容合规需在模型调用层面设置合理的system prompt进行约束过滤不当请求。数据隐私如果处理用户敏感数据需确保传输、存储符合相关法规避免通过工具泄露。授权使用确保所使用的模型 API如 OpenAI和工具服务均获得合法授权。3. 环境准备与前置条件开始搭建前请确保你的开发环境满足以下要求。1. 操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。推荐使用 Linux 或 WSL2 (Windows) 以获得最佳兼容性。2. Python 环境Python 版本: 3.8 至 3.113.12 需注意部分包兼容性。推荐使用 3.10。包管理工具: 使用pip或conda。强烈建议创建独立的虚拟环境。# 使用 venv 创建虚拟环境 python -m venv agent_env # 激活环境 # Windows: agent_env\Scripts\activate # Linux/macOS: source agent_env/bin/activate3. 核心依赖包我们将安装 LangChain、LangGraph 及相关组件。# 基础框架 pip install langchain langgraph langchain-core # 用于连接主流大模型选择你需要的 pip install langchain-openai # 连接 OpenAI # 或 pip install langchain-community # 包含许多社区模型和工具 # 可选用于构建Web界面 pip install streamlit gradio # 可选用于示例中的工具如网络搜索、数学计算 pip install langchain-community duckduckgo-search numexpr4. 模型 API 密钥本教程以 OpenAI GPT 系列模型为例你需要准备有效的OPENAI_API_KEY。访问 OpenAI 平台创建 API Key。安全提示切勿将 API Key 直接硬编码在代码中或提交到版本控制系统如 Git。应使用环境变量管理。# 在终端中设置环境变量临时 export OPENAI_API_KEYyour-api-key-here # Linux/macOS # set OPENAI_API_KEYyour-api-key-here # Windows CMD # $env:OPENAI_API_KEYyour-api-key-here # Windows PowerShell4. 安装部署与启动方式我们将构建一个具备“思考-行动”循环的基础智能体并通过 LangGraph 将其可视化。第一步初始化智能体核心组件创建一个名为basic_agent.py的文件。import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain import hub # 1. 初始化大语言模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) # 2. 定义工具Tools # 示例工具1一个简单的字符串长度计算器 def calculate_length(text: str) - str: 计算输入字符串的长度。 return f字符串的长度是 {len(text)} 个字符。 # 示例工具2一个模拟的天气查询 def get_weather(city: str) - str: 获取指定城市的天气信息模拟。 weather_data { 北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C } return weather_data.get(city, f未找到{city}的天气信息。) # 将函数包装成 LangChain Tool 对象 tools [ Tool( nameString Length Calculator, funccalculate_length, description当需要计算一个字符串的长度时使用此工具。输入应为一个字符串。 ), Tool( nameWeather Checker, funcget_weather, description当需要查询城市的天气时使用此工具。输入应为城市名称如‘北京’。 ), ] # 3. 从 LangChain Hub 拉取一个预设的智能体提示词Prompt prompt hub.pull(hwchase17/react) # 4. 创建 ReAct 智能体 agent create_react_agent(llm, tools, prompt) # 5. 创建执行器AgentExecutor agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 测试运行 if __name__ __main__: result agent_executor.invoke({input: 深圳的天气怎么样然后告诉我‘Hello, Agent!’这句话有多长。}) print(\n--- 最终结果 ---) print(result[output])运行测试python basic_agent.py如果配置正确你将看到智能体的完整思考过程verboseTrue它会依次调用天气查询工具和字符串长度计算工具并给出最终答案。第二步引入 LangGraph 构建可视化工作流LangGraph 的核心是“图”Graph它由节点Nodes和边Edges组成能清晰描述智能体的状态流转。创建一个新文件graph_agent.py。import os from typing import TypedDict, Annotated, Sequence import operator from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain.tools import Tool # 1. 定义状态State结构 class AgentState(TypedDict): messages: Annotated[Sequence, add_messages] # 消息历史 turn_count: int # 记录对话轮次用于控制循环 # 2. 初始化模型和工具复用之前的工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [Tool(nameWeather, funclambda x: f{x}的天气是模拟的25°C晴天。, description查询天气)] llm_with_tools llm.bind_tools(tools) # 3. 定义图节点Nodes def call_model(state: AgentState): 调用大模型决定下一步行动。 messages state[messages] # 将历史消息发给模型 response llm_with_tools.invoke(messages) # 将模型的响应添加到消息历史中 return {messages: [response], turn_count: state[turn_count] 1} def call_tool(state: AgentState): 执行模型选择的工具。 last_message state[messages][-1] tool_calls last_message.tool_calls tool_messages [] for tool_call in tool_calls: tool next((t for t in tools if t.name tool_call[name]), None) if tool: result tool.invoke(tool_call[args]) tool_messages.append(ToolMessage(contentresult, tool_call_idtool_call[id])) return {messages: tool_messages} # 4. 定义条件边Conditional Edges def should_continue(state: AgentState) - str: 根据模型输出和轮次决定下一步是调用工具还是结束。 last_message state[messages][-1] turn_count state[turn_count] # 如果模型没有调用工具或者对话轮次超过3轮则结束 if not last_message.tool_calls or turn_count 3: return end return continue # 5. 构建图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) workflow.add_node(action, call_tool) # 设置入口点 workflow.set_entry_point(agent) # 添加条件边 workflow.add_conditional_edges( agent, should_continue, { continue: action, # 需要调用工具则前往 action 节点 end: END # 结束则前往 END } ) workflow.add_edge(action, agent) # 执行完工具后回到 agent 节点继续思考 # 编译图 app workflow.compile() # 6. 运行图 if __name__ __main__: from langchain_core.messages import HumanMessage # 初始化状态 initial_state {messages: [HumanMessage(content查询一下北京的天气。)], turn_count: 0} # 运行图 for event in app.stream(initial_state): for k, v in event.items(): print(f节点 [{k}] 的输出) print(v) print(---) final_state app.invoke(initial_state) print(\n最终消息历史) for msg in final_state[messages]: print(f{type(msg).__name__}: {msg.content})运行测试python graph_agent.py你将看到智能体在agent和action节点间的流转过程直观地理解“思考-行动”循环。5. 功能测试与效果验证现在让我们对构建的智能体进行系统性测试验证其核心能力。5.1 基础工具调用测试测试目的验证智能体能否正确理解用户意图并选择、调用合适的工具。输入“请先计算‘LangGraph is powerful’的长度再告诉我杭州的天气。”操作步骤使用basic_agent.py或graph_agent.py中的执行器。将上述输入传递给智能体。预期结果智能体应首先识别出需要调用“字符串长度计算器”工具。得到结果后再识别出需要调用“天气查询”工具或模拟天气工具。最终输出应包含字符串长度和杭州的天气信息。判断成功输出结果准确且日志verboseTrue显示工具被依次调用。常见失败工具选择错误检查工具的描述description是否清晰模型是否理解。工具调用参数错误检查工具函数的输入参数格式是否与模型调用匹配。5.2 多轮对话与状态保持测试测试目的验证智能体在连续对话中能否记住上下文。输入序列“我叫小明。”“我的名字是什么”操作步骤需要在智能体配置中启用记忆Memory功能。修改basic_agent.py使用ConversationBufferMemory。from langchain.memory import ConversationBufferMemory memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) # 在创建 AgentExecutor 时传入 memory agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, memorymemory)连续调用agent_executor.invoke两次。预期结果对于第二个问题智能体应能回答“你是小明”。判断成功智能体正确利用了历史对话信息。常见失败记忆未正确连接或未在 prompt 中体现。5.3 复杂任务分解测试测试目的验证智能体处理需要多步骤推理的复杂任务的能力。输入“我想知道特斯拉TSLA当前的股价并用人民币估算一下我持有10股的价值。假设1美元等于7.2人民币。”操作步骤需要为智能体提供更多工具例如StockPriceTool: 模拟或真实调用股票API获取TSLA股价例如返回250.00。CalculatorTool: 进行数学计算股价 * 股数 * 汇率。观察智能体是否规划出“获取股价 - 计算美元总价值 - 换算人民币”的步骤。预期结果输出类似“特斯拉当前股价约为250美元。您持有10股总价值约为2500美元折合人民币约18000元。”判断成功智能体按逻辑顺序调用工具并得出正确计算结果。常见失败步骤顺序错乱、工具调用缺失、计算错误。5.4 LangGraph 工作流可视化测试测试目的直观验证智能体的决策循环。操作步骤运行graph_agent.py。观察控制台输出的节点流转信息。高级使用 LangGraph Studio 进行可视化需额外配置。预期结果在控制台看到清晰的agent - (continue) - action - agent - (end)流程。判断成功状态state在节点间正确传递循环在满足条件时终止。常见失败条件边逻辑错误导致无限循环或提前退出。6. 接口 API 与批量任务将智能体封装成 API 服务是集成到其他系统的关键。这里使用 FastAPI 创建一个简单的 Web 服务。第一步创建 API 服务创建文件agent_api.py。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain import hub from langchain.memory import ConversationBufferMemory import asyncio import uuid from contextlib import asynccontextmanager # 定义请求和响应模型 class AgentRequest(BaseModel): query: str session_id: Optional[str] None # 用于支持多会话 class AgentResponse(BaseModel): session_id: str answer: str tool_calls: List[str] [] # 全局智能体执行器简单示例生产环境需优化 agent_executor None memory_store {} # 简单的内存存储生产环境需用Redis等 asynccontextmanager async def lifespan(app: FastAPI): # 启动时初始化智能体 global agent_executor llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [ Tool(nameDummy Tool, funclambda x: fYou said: {x}, descriptionA dummy tool for test.) ] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) # 注意这里使用全局记忆生产环境应为每个session_id创建独立的memory memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent_executor AgentExecutor(agentagent, toolstools, verboseFalse, memorymemory, handle_parsing_errorsTrue) yield # 关闭时清理资源 agent_executor None app FastAPI(lifespanlifespan) app.post(/chat, response_modelAgentResponse) async def chat_with_agent(request: AgentRequest): if agent_executor is None: raise HTTPException(status_code503, detailAgent not initialized) session_id request.session_id or str(uuid.uuid4()) # 生产环境应根据session_id获取对应的memory # 此处简化处理使用全局agent try: # 注意直接调用invoke可能在异步环境中阻塞生产环境应使用线程池 loop asyncio.get_event_loop() result await loop.run_in_executor( None, agent_executor.invoke, {input: request.query} ) return AgentResponse( session_idsession_id, answerresult.get(output, No output), tool_calls[] # 实际应从result中解析工具调用历史 ) except Exception as e: raise HTTPException(status_code500, detailfAgent execution failed: {str(e)}) app.get(/health) async def health_check(): return {status: healthy, agent_ready: agent_executor is not None} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)第二步启动与测试 API安装依赖pip install fastapi uvicorn启动服务python agent_api.py测试接口# 使用 curl 测试 curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {query: 你好智能体}批量任务思路对于批量查询可以改造/chat接口接受一个查询列表。更健壮的做法是引入任务队列如 Celery RedisAPI 接收任务后返回任务 ID客户端轮询结果。每个任务在独立的 AgentExecutor 实例中运行避免状态污染。7. 资源占用与性能观察AI Agent 服务的资源消耗主要来自大模型 API 调用和本地逻辑执行。1. 主要资源消耗点网络 I/O与 OpenAI 等云端模型 API 的通信延迟是主要性能瓶颈。内存ConversationBufferMemory会存储所有历史消息长对话会占用较多内存。使用ConversationSummaryMemory或ConversationBufferWindowMemory可优化。CPU本地工具函数执行、LangChain/LangGraph 框架本身的逻辑处理消耗 CPU。2. 性能观察方法日志记录启用 AgentExecutor 的verboseTrue观察每个步骤耗时。API 监控使用模型供应商提供的监控面板查看 Token 消耗、响应时间。自定义计时在关键函数前后添加时间戳。import time start time.time() result agent_executor.invoke({input: query}) end time.time() print(f总耗时: {end - start:.2f}秒)3. 优化建议缓存对频繁且结果不变的查询如某些工具调用结果进行缓存。异步调用对于 I/O 密集型操作如多个独立的工具调用使用异步函数。精简上下文合理设置max_token_limit定期清理或总结记忆避免上下文过长导致 API 费用增加和速度变慢。选择合适模型对简单任务使用轻量模型如gpt-3.5-turbo复杂任务再用gpt-4。8. 常见问题与排查方法在开发和运行 AI Agent 过程中你可能会遇到以下问题。问题现象可能原因排查方式解决方案导入 LangChain 或 LangGraph 失败1. 未安装包。2. Python 版本不兼容。3. 虚拟环境未激活。1.pip list | grep langchain。2.python --version。1. 使用pip install正确安装。2. 确保 Python 版本在 3.8-3.11。3. 激活虚拟环境。运行时报错OPENAI_API_KEY未设置环境变量未正确设置或代码未读取。1.echo $OPENAI_API_KEY(Linux/macOS) 或echo %OPENAI_API_KEY%(Windows)。2. 检查代码中读取密钥的方式。1. 在终端或系统设置中正确设置环境变量。2. 或在代码中临时设置os.environ[‘OPENAI_API_KEY’] ‘your-key’仅用于测试。智能体不调用工具直接回答1. 工具描述不清晰。2. 模型温度 (temperature) 设置过高。3. Prompt 未优化。1. 检查工具description是否准确。2. 设置temperature0减少随机性。3. 查看模型接收到的完整 Prompt。1. 优化工具描述明确使用场景和输入格式。2. 使用更结构化的 Prompt明确要求模型使用工具。3. 尝试hub.pull中其他 Agent Prompt。LangGraph 图陷入无限循环条件边 (should_continue) 逻辑有误或状态更新不正确。1. 打印每轮循环的state和should_continue的返回值。2. 检查turn_count等控制变量是否被正确更新。1. 确保结束条件 (END) 能被触发。2. 在状态中增加最大轮次限制并严格执行。API 服务并发请求出错或状态混乱多个请求共享了同一个AgentExecutor和Memory实例。观察不同session_id的请求是否看到了对方的对话历史。为每个会话 (session_id) 创建独立的AgentExecutor和Memory实例。可使用工厂模式管理。工具调用出错参数类型错误等工具函数定义的参数与模型生成的调用参数不匹配。1. 查看verbose日志中模型生成的工具调用 JSON。2. 对比工具函数的参数签名。1. 确保工具函数参数简单通常为单个str。2. 在工具函数内部进行参数解析和类型转换。处理长文本或复杂任务时 Token 超限对话历史或上下文过长超过模型 Token 限制。监控 API 返回的错误信息通常是context_length_exceeded。1. 使用ConversationSummaryMemory。2. 使用ConversationBufferWindowMemory只保留最近 N 轮对话。3. 在调用模型前手动截断历史消息。9. 最佳实践与使用建议基于上述实践总结出以下建议帮助你更稳健地开发和部署 AI Agent。从简单开始逐步复杂化先用 1-2 个简单工具验证智能体基础流程。成功后再逐步增加工具复杂度、引入记忆、构建多智能体协作通过 LangGraph。精心设计工具Tools描述清晰工具的description是模型选择工具的主要依据务必准确描述其功能和输入格式。功能单一一个工具只做一件事保持纯粹性。复杂功能应由智能体通过组合多个工具调用完成。健壮性工具函数内部要做好异常处理返回友好的错误信息避免因单个工具失败导致整个智能体崩溃。有效管理记忆Memory区分会话在生产环境中必须为每个用户或会话维护独立的记忆存储。控制长度根据成本和性能权衡选择合适的 Memory 类型。对于长对话ConversationSummaryMemory是平衡效果和成本的常用选择。关键信息提取考虑将对话中的关键实体如用户名、项目编号提取出来单独存储而非完全依赖非结构化的历史消息。利用 LangGraph 管理复杂流程可视化设计在纸上或白板上画出智能体的状态图明确节点和边再用代码实现。状态设计State应包含所有必要的上下文信息但也要保持简洁。调试多用app.stream()观察状态流转这是理解智能体决策过程的最佳方式。生产环境部署考量API 密钥管理使用环境变量或专业的密钥管理服务如 AWS Secrets Manager。错误处理与重试对模型 API 和工具调用添加重试机制和断路器。监控与日志记录智能体的完整决策链包括中间步骤便于问题追溯和效果分析。成本控制监控 Token 消耗设置预算和用量告警。合规与伦理用户知情明确告知用户正在与 AI 交互并说明其能力边界。工具权限隔离对文件读写、网络访问、数据库操作等敏感工具实施严格的权限控制遵循最小权限原则。输出审核对于生成内容可能直接面向用户的应用建立人工或自动化的审核机制。10. 总结与下一步通过本教程你完成了从零搭建一个具备“思考-行动”循环的 AI Agent 的核心步骤。我们使用了 LangChain 作为智能体编排的基础利用 LangGraph 实现了清晰可控的工作流状态管理。最关键的是你不仅看到了概念还运行了代码测试了功能并了解了如何将其封装为 API 服务。最值得尝试的下一步集成真实工具将示例中的模拟工具替换为真实的 API如 Serper搜索、WolframAlpha计算、Apify网页抓取等。探索多智能体Multi-Agent使用 LangGraph 创建多个具有不同角色的智能体如研究员、写手、校对员让它们通过协作完成更复杂的任务。接入 OpenClaw 等专业库OpenClaw 等库提供了大量预构建的工具和技能。尝试将其集成到你的智能体中快速扩展能力。构建 Web 界面使用 Gradio 或 Streamlit 快速构建一个聊天界面让非技术用户也能与你的智能体交互。最容易踩的坑状态污染在 API 服务中多个请求共享了同一个记忆对象。提示词工程工具描述不清或 Prompt 指令不明确导致模型行为不符合预期。错误处理缺失工具或模型 API 调用失败导致整个流程中断。AI Agent 的开发是一个迭代和实验的过程。建议你克隆本文的示例代码从一个具体的、小范围的任务开始逐步增加其复杂性和可靠性。随着 LangChain 和 LangGraph 生态的不断成熟构建强大智能体的门槛正在迅速降低。现在正是动手探索的最佳时机。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度