30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度在实际 AI 应用开发中构建一个能自主理解任务、调用工具、并完成复杂流程的智能体Agent是迈向高级 AI 系统的关键一步。然而从零开始搭建一个稳定、可扩展的智能体框架开发者常常会陷入工具链选择、状态管理、流程编排和错误处理的复杂细节中。LangChain 提供了构建链和代理的基础组件而 LangGraph 则在此基础上引入了基于图的状态机模型用于描述具有循环、分支和并行执行能力的多步骤工作流。OpenClaw 作为火山引擎开源的智能体框架则提供了更贴近生产环境的任务编排、工具管理和服务化能力。本文将带你从核心概念出发手把手整合 LangChain、LangGraph 和 OpenClaw搭建一个具备实际感知、决策和执行能力的 AI 智能体并深入探讨其内部工作机制与工程实践。1. 理解智能体Agent的核心从 LangChain 到 LangGraph在开始编码之前必须厘清几个核心概念及其关系。智能体不是一个大语言模型LLM的简单包装而是一个具备感知、规划、执行和反思能力的自治系统。1.1 LangChain智能体的“工具箱”与“粘合剂”LangChain 的核心价值在于标准化了与 LLM 的交互并提供了大量可复用的“链”Chains和“工具”Tools。你可以把它理解为智能体的基础设施层。工具Tools智能体与外部世界交互的接口。一个工具可以是一个函数用于搜索网络、查询数据库、执行计算或调用 API。LangChain 内置了大量工具也支持自定义。链Chains将 LLM 调用、工具调用、数据预处理等步骤顺序组合起来的工作流。它适用于确定性的、线性的任务。代理AgentsLangChain 中的“代理”是一个高级抽象它使用 LLM 作为“大脑”根据当前状态如用户输入、历史对话、工具输出来决定下一步是调用某个工具还是直接给出最终答案。这是早期智能体的常见形态。然而传统的 LangChain Agent 在处理复杂、多步骤、有状态且可能循环的工作流时其控制流显得不够直观和强大。1.2 LangGraph为智能体引入“状态机”与“工作流引擎”LangGraph 构建在 LangChain 之上它引入了“图”Graph的概念来定义智能体的工作流。图中的节点代表一个执行步骤如调用 LLM、运行工具边代表步骤之间的流转条件。状态State一个贯穿整个工作流的共享字典。它包含了所有节点需要读取和写入的信息如用户输入、中间结果、执行历史等。节点Nodes执行具体操作的函数。例如一个“规划”节点调用 LLM 分解任务一个“执行”节点调用工具。边Edges决定工作流走向的条件。通常是基于当前状态或上一个节点输出的值来决定下一个执行哪个节点。这支持了循环如“反思-改进”循环和条件分支。通过 LangGraph你可以清晰地定义如“先规划再执行检查结果若不满意则重新规划”这样的复杂逻辑这是构建强大智能体的基石。1.3 OpenClaw面向生产的智能体“操作系统”OpenClaw 是火山引擎开源的智能体框架。它更侧重于智能体在生产环境中的生命周期管理、任务调度、技能Skill类似于 Tools管理和服务化部署。它提供了 Web UI 和 SDK方便管理和监控智能体。技能SkillOpenClaw 对“工具”的封装通常具备更完善的定义、描述和参数校验。工作流Workflow类似于 LangGraph 的图用于编排技能和 LLM 的调用顺序。服务化OpenClaw 可以将你定义的智能体轻松部署为 HTTP 服务并提供 API 供其他系统调用。在本教程中我们将使用 LangChain 和 LangGraph 来构建智能体的核心推理与执行引擎并探讨如何将其与 OpenClaw 的技能管理、服务化能力相结合形成一个完整的解决方案。2. 环境准备与依赖配置我们将创建一个 Python 项目来搭建智能体。请确保你的 Python 版本在 3.8 以上。2.1 创建项目与虚拟环境首先创建一个新的项目目录并进入。mkdir ai-agent-tutorial cd ai-agent-tutorial强烈建议使用虚拟环境来管理依赖。# 使用 venv python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate2.2 安装核心依赖我们将安装 LangChain、LangGraph 以及 OpenAI 的 SDK作为 LLM 供应商示例。同时为了后续与 OpenClaw 集成我们也安装其 SDK。pip install langchain langgraph langchain-openai # 安装 openclaw 客户端 SDK (请以官方仓库最新版本为准) # pip install openclaw-sdk由于 OpenClaw 的 Python SDK 可能处于快速迭代中如果直接安装失败可以尝试从官方 GitHub 仓库安装。# 示例从GitHub安装 # pip install githttps://github.com/volcengine/openclaw-sdk-python.git此外我们还需要安装一些辅助库。pip install python-dotenv # 用于管理环境变量2.3 配置 API 密钥智能体需要调用大模型。我们以 OpenAI 的 GPT 系列为例。在项目根目录创建一个.env文件用于安全存储密钥。# .env 文件内容 OPENAI_API_KEY你的-openai-api-key # 如果需要其他模型如 Anthropic, 可在此添加 # ANTHROPIC_API_KEY你的-key然后在代码中通过dotenv加载。# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. 构建第一个基础智能体基于 LangChain 的 ReAct 代理让我们从最简单的开始使用 LangChain 构建一个遵循 ReActReasoning Acting范式的智能体。这个智能体会尝试通过“思考”和“行动”来回答问题。3.1 定义工具智能体需要工具来获取信息。我们定义两个简单的工具一个计算器和一个模拟的网络搜索工具。# tools.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **, sqrt() 等。 try: # 警告在生产环境中直接 eval 是危险的此处仅用于演示。 # 应使用安全的表达式解析库如 asteval。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。在实际应用中这里应调用真实的搜索API如 SerperAPI、Google Search API 等。 # 这是一个模拟函数返回固定结果。 knowledge_base { 今天的天气: 北京晴15-25°C上海多云18-28°C。, LangGraph 是什么: LangGraph 是 LangChain 的一个库用于构建具有状态和循环的多智能体工作流。, OpenClaw: OpenClaw 是火山引擎开源的智能体框架提供技能管理和服务化部署能力。 } return knowledge_base.get(query, f未找到关于 {query} 的模拟信息。)3.2 创建代理并运行现在我们使用 LangChain 的create_react_agent来创建一个代理。# simple_agent.py from langchain import hub from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from tools import calculator, search_web from config import OPENAI_API_KEY # 1. 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyOPENAI_API_KEY) # 2. 准备工具列表 tools [calculator, search_web] # 3. 从 LangChain Hub 获取 ReAct 提示词模板这是一个经过优化的模板 prompt hub.pull(hwchase17/react) # 4. 创建 ReAct 代理 agent create_react_agent(llm, tools, prompt) # 5. 创建代理执行器它负责处理代理的循环调用 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 运行代理 if __name__ __main__: questions [ “北京今天的天气怎么样” “计算 3 的 5 次方加上 10 除以 2 的结果。” ] for question in questions: print(f\n用户: {question}) try: result agent_executor.invoke({input: question}) print(f智能体: {result[output]}) except Exception as e: print(f执行出错: {e})运行这个脚本你会看到类似以下的输出展示了代理的“思考-行动-观察”过程 进入新的 AgentExecutor 链... 我需要回答关于北京天气的问题。我没有实时数据但我有一个搜索工具可以获取模拟信息。 行动: search_web 行动输入: 北京今天的天气 观察: 北京晴15-25°C上海多云18-28°C。 思考: 我得到了北京的天气信息。 最终答案: 北京今天的天气是晴气温在15到25摄氏度之间。 链结束。这个基础代理已经能处理简单任务。但它有几个局限1) 状态管理不直观2) 难以定义复杂的多步骤或循环流程3) 错误处理和日志记录需要额外工作。接下来我们用 LangGraph 来解决这些问题。4. 使用 LangGraph 构建可控工作流智能体我们将用 LangGraph 重构上面的代理使其具备更清晰的状态流转和潜在的循环能力例如允许它在结果不理想时重新规划。4.1 定义状态 Schema首先定义智能体工作流中需要共享的状态。我们使用TypedDict来获得更好的类型提示。# graph_state.py from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): 智能体工作流的状态定义。 input: str # 用户的原始输入 plan: str # 当前的执行计划 steps: List[str] # 已执行的步骤记录 tool_outputs: List[str] # 工具调用的输出结果 final_answer: str # 最终给用户的答案 iteration: int # 循环迭代次数用于防止无限循环为了在图中方便地更新列表类型的字段LangGraph 推荐使用add_messages这种注解但对于自定义列表我们可以用operator.add。# 实际上对于自定义字段我们会在节点函数中直接操作 state 字典。 # 下面是一个更实用的状态定义包含合成字段 from typing import Optional from langgraph.graph.message import add_messages class GraphState(TypedDict): 图的状态。 messages: Annotated[list, add_messages] # 对话历史 user_input: str # 当前轮次的用户输入 tool_calls: Optional[list] # LLM 决定的工具调用 tool_outputs: Optional[list] # 工具执行的结果 # 我们可以根据需求添加更多字段如 current_step, is_finished 等为了简化我们采用第一个AgentState定义并在节点函数中手动更新状态。4.2 构建图节点我们将工作流分解为几个节点规划节点、执行节点、判断节点。# graph_nodes.py from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from tools import calculator, search_web from config import OPENAI_API_KEY llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyOPENAI_API_KEY) tools [calculator, search_web] # 为LLM准备工具描述 tool_descriptions \n.join([f- {tool.name}: {tool.description} for tool in tools]) def plan_node(state: AgentState) - AgentState: 规划节点分析用户输入制定初步计划。 user_input state[input] prompt f 你是一个任务规划师。请根据用户的问题和可用工具制定一个简明的步骤计划。 用户问题{user_input} 可用工具 {tool_descriptions} 请输出你的计划格式为1. ... 2. ... 如果问题很简单可能只需要一步。直接回答问题也算作一步。 计划 message [HumanMessage(contentprompt)] response llm.invoke(message) plan response.content.strip() print(f[规划节点] 计划: {plan}) state[plan] plan state[steps].append(f规划: {plan}) return state def execute_node(state: AgentState) - AgentState: 执行节点根据当前计划和历史决定调用哪个工具如果需要。 # 这里我们简化让LLM直接根据当前上下文决定行动。 # 更复杂的实现可以参考 LangChain 的 tool_calling 能力。 history \n.join(state[steps][-3:]) # 最近几步历史 user_input state[input] prompt f 你是一个执行者。根据以下计划、历史记录和用户问题决定下一步做什么。 如果问题已解决直接给出最终答案。否则决定调用哪个工具并给出确切的输入。 用户问题{user_input} 当前计划{state.get(plan, 暂无)} 近期历史 {history} 可用工具 {tool_descriptions} 请严格按以下格式回答 最终答案: [你的答案] 或 调用工具: [工具名称] 工具输入: [输入字符串] message [HumanMessage(contentprompt)] response llm.invoke(message) decision response.content.strip() print(f[执行节点] 决策: {decision}) if decision.startswith(最终答案:): answer decision.replace(最终答案:, ).strip() state[final_answer] answer state[steps].append(f给出最终答案: {answer}) elif decision.startswith(调用工具:): lines decision.split(\n) tool_name lines[0].replace(调用工具:, ).strip() tool_input lines[1].replace(工具输入:, ).strip() if len(lines) 1 else # 查找并调用工具 tool_to_use None for tool in tools: if tool.name tool_name: tool_to_use tool break if tool_to_use: try: output tool_to_use.invoke(tool_input) state[tool_outputs].append(output) state[steps].append(f调用工具 {tool_name} 输入 {tool_input}输出: {output}) print(f[工具调用] {tool_name}: {output}) except Exception as e: error_msg f工具 {tool_name} 调用失败: {e} state[tool_outputs].append(error_msg) state[steps].append(error_msg) else: error_msg f未找到工具 {tool_name} state[tool_outputs].append(error_msg) state[steps].append(error_msg) else: state[steps].append(f无法解析决策: {decision}) return state def judge_node(state: AgentState) - AgentState: 判断节点检查是否满足结束条件。 # 简单的结束条件有最终答案或者迭代次数太多或者工具输出表明任务完成。 if state.get(final_answer): print([判断节点] 任务完成有最终答案。) return state if state.get(iteration, 0) 5: # 防止无限循环 state[final_answer] 经过多次尝试未能完成任务。 print([判断节点] 达到最大迭代次数终止。) return state # 可以添加更复杂的判断逻辑例如检查最近的工具输出是否已回答了问题。 print([判断节点] 任务未完成继续执行。) return state4.3 组装工作流图现在我们将节点和边组合起来形成一个有向图。# build_graph.py from langgraph.graph import StateGraph, END from graph_state import AgentState from graph_nodes import plan_node, execute_node, judge_node def build_agent_graph(): # 初始化图指定状态类型 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(execute, execute_node) workflow.add_node(judge, judge_node) # 设置入口点 workflow.set_entry_point(plan) # 添加边规划后总是去执行 workflow.add_edge(plan, execute) # 执行后去判断 workflow.add_edge(execute, judge) # 条件边从判断节点出发根据状态决定下一步 def decide_next_step(state: AgentState): if state.get(final_answer): return END # 结束 else: # 增加迭代计数 state[iteration] state.get(iteration, 0) 1 return execute # 返回执行节点继续循环 workflow.add_conditional_edges( judge, decide_next_step, { execute: execute, END: END } ) # 编译图 graph workflow.compile() return graph if __name__ __main__: graph build_agent_graph() # 可视化图需要安装 graphviz try: from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) except: print(无法显示图形但图已构建成功。)4.4 运行 LangGraph 智能体最后我们初始化状态并运行这个图。# run_graph_agent.py from build_graph import build_agent_graph def run_agent(query: str): graph build_agent_graph() # 初始化状态 initial_state: AgentState { input: query, plan: , steps: [], tool_outputs: [], final_answer: , iteration: 0 } print(f开始处理查询: {query}) print(- * 50) # 运行图 final_state graph.invoke(initial_state) print(- * 50) print(f处理完成。最终答案: {final_state.get(final_answer, 无)}) print(\n执行步骤回顾:) for i, step in enumerate(final_state.get(steps, [])): print(f {i1}. {step}) return final_state if __name__ __main__: queries [ “上海和北京哪里今天更暖和” “计算 (15 7) * 3 的值。” ] for q in queries: run_agent(q) print(\n *60 \n)运行此脚本你将看到一个清晰的、分步骤的执行过程。LangGraph 的威力在于你可以通过修改图的结构例如在“判断”后增加一个“反思”节点来优化计划来轻松改变智能体的行为逻辑而无需重写核心代理代码。5. 集成 OpenClaw技能管理与服务化到目前为止我们构建的是一个本地运行的脚本智能体。要将其转化为一个可管理、可部署的服务可以借助 OpenClaw。5.1 将工具封装为 OpenClaw SkillOpenClaw 通过“技能”来管理工具。我们需要将之前的calculator和search_web函数包装成符合 OpenClaw 规范的技能。假设 OpenClaw SDK 提供了装饰器skill具体 API 请参考官方文档。# openclaw_skills.py # 注意以下代码为示意需根据 OpenClaw 官方 SDK 实际接口调整 from typing import Dict, Any # from openclaw.skill import skill, SkillContext # 假设的导入 # skill(namecalculator, description计算数学表达式) def calculator_skill(expression: str, context: Dict[str, Any]) - Dict[str, Any]: OpenClaw 技能版本的计算器。 import math try: result eval(expression, {__builtins__: None}, {math: math}) return {success: True, data: result, message: f计算结果: {result}} except Exception as e: return {success: False, data: None, message: f计算错误: {e}} # skill(nameweb_search, description模拟网络搜索) def web_search_skill(query: str, context: Dict[str, Any]) - Dict[str, Any]: OpenClaw 技能版本的网络搜索。 knowledge_base { 今天的天气: 北京晴15-25°C上海多云18-28°C。, LangGraph 是什么: LangGraph 是 LangChain 的一个库用于构建具有状态和循环的多智能体工作流。, } result knowledge_base.get(query, f未找到关于 {query} 的模拟信息。) return {success: True, data: result, message: result}在实际项目中你需要按照 OpenClaw 的指南将这些技能函数注册到 OpenClaw 服务器。5.2 在 LangGraph 智能体中调用 OpenClaw 技能一种集成方式是修改execute_node使其通过 OpenClaw 的客户端 SDK 来远程调用技能而不是直接调用本地函数。# 假设的 OpenClaw 客户端调用 # from openclaw.client import OpenClawClient # client OpenClawClient(base_urlhttp://localhost:8080) def execute_node_with_openclaw(state: AgentState) - AgentState: # ... 前面的决策逻辑与之前类似 ... # 当决定调用工具时 # tool_name, tool_input ... # 调用 OpenClaw 技能 # response client.invoke_skill(tool_name, {input: tool_input}) # if response[success]: # output response[message] # else: # output f技能调用失败: {response[message]} # state[tool_outputs].append(output) # ... return state这种方式将工具的执行与管理职责交给了 OpenClaw我们的智能体核心只负责规划和决策。5.3 将智能体部署为 OpenClaw 工作流更进一步你可以将整个 LangGraph 智能体定义为一个 OpenClaw 的“工作流”。OpenClaw 的工作流可能支持通过代码或 UI 定义一系列节点对应我们的plan_node,execute_node等。这样你可以通过 OpenClaw 的 Web UI 来监控这个智能体的运行状态、查看日志、管理版本。具体部署步骤通常包括将智能体代码打包。通过 OpenClaw 的 API 或 UI 创建工作流定义。配置触发器如 HTTP API。部署并启动工作流。由于 OpenClaw 的具体部署流程可能更新请务必参考其官方文档。6. 常见问题排查与最佳实践在构建和运行 AI 智能体时你会遇到各种问题。以下是一些常见问题的排查路径和工程建议。6.1 常见问题排查表问题现象可能原因检查方式处理建议智能体无限循环不输出答案。1. 结束条件判断逻辑有误。2. LLM 的决策输出格式不符合解析预期。3. 工具调用失败导致状态未更新。1. 检查judge_node的逻辑。2. 打印decision变量查看 LLM 的实际输出。3. 检查工具调用是否抛出异常并被捕获。1. 在judge_node中添加更严格的结束判断例如分析最近工具输出是否包含答案。2. 优化提示词强制 LLM 使用指定格式输出。3. 完善工具调用的错误处理确保失败信息能反馈给状态。工具调用结果未被有效利用。1. 工具输出未正确放入state中供后续节点读取。2. 提示词中没有将历史工具输出作为上下文。1. 检查execute_node中是否将output添加到了state[“tool_outputs”]或state[“steps”]。2. 检查plan_node或后续execute_node的提示词是否包含了state[“tool_outputs”]的内容。1. 确保状态更新逻辑正确。2. 重构提示词明确要求 LLM 参考“之前的工具输出”。例如“以下是已获取的信息[工具输出历史]”。LangGraph 图编译或运行时报错。1. 状态 Schema 定义与节点函数返回值不匹配。2. 图中节点或边引用名称错误。3. 使用了不支持的 Python 类型。1. 仔细核对AgentState的键和节点函数返回字典中的键。2. 检查add_node和add_edge使用的名称是否一致。3. 确保状态中只使用 LangGraph 支持的基本类型和Annotated列表。1. 保持状态 Schema 的简洁性初期只放必要字段。2. 使用graph.get_graph().print_ascii()打印图结构检查节点和边。3. 参考 LangGraph 官方文档关于状态类型的说明。OpenClaw 技能调用失败。1. OpenClaw 服务未启动或网络不通。2. 技能名称或参数格式不正确。3. 身份认证失败。1. 使用curl或 Postman 测试 OpenClaw 基础 API。2. 对照 OpenClaw 技能定义检查传入的参数名和类型。3. 检查 SDK 客户端的 API Key 或 Token 配置。1. 确保 OpenClaw 服务健康运行。2. 在 OpenClaw UI 中测试技能是否能独立运行。3. 封装一个健壮的技能调用客户端包含重试和降级逻辑。LLM 响应慢或超时。1. 网络问题或 API 限流。2. 提示词过长导致 Token 消耗大。3. 图循环次数过多调用 LLM 次数激增。1. 检查网络延迟和 API 状态。2. 计算提示词的 Token 数量可使用tiktoken库。3. 在judge_node中限制最大迭代次数。1. 为 LLM 调用设置合理的超时时间。2. 优化提示词移除冗余信息。对长上下文进行摘要。3. 实施缓存机制对相同或相似的中间问题缓存 LLM 回答。6.2 生产环境最佳实践提示词工程智能体的表现极度依赖提示词。将提示词模板化、外部化如存储在 JSON 或数据库中便于迭代优化和 A/B 测试。可观测性在关键节点如状态转换、工具调用、LLM 请求记录详细的日志和指标。这有助于调试和性能分析。考虑集成像 LangSmith 这样的 LLM 应用监控平台。错误处理与韧性智能体工作流中任何一环都可能失败。要为 LLM 调用、工具调用设置重试、超时和降级策略。确保状态机在异常情况下也能优雅终止或进入错误处理分支。成本控制LLM 调用是主要成本来源。监控 Token 使用量对于内部工具能解决的问题尽量避免询问 LLM。对用户输入和工具输出做长度限制。安全与权限工具调用是高风险操作。严格校验用户输入防止注入攻击。为不同的工具设置权限等级确保智能体不会执行危险操作如删除文件、调用敏感 API。测试为你的智能体工作流编写单元测试和集成测试。模拟 LLM 响应和工具行为测试不同的执行路径和边界条件。7. 扩展方向与下一步你已经搭建了一个融合 LangChain、LangGraph 和 OpenClaw 概念的智能体基础框架。要使其更强大可以考虑以下方向多智能体协作使用 LangGraph 创建多个具有不同角色的智能体如规划者、执行者、评审者让它们通过共享状态或消息进行协作处理更复杂的任务。长期记忆集成向量数据库如 Chroma, Pinecone让智能体能够记住之前的对话和任务结果实现上下文感知。更复杂的工具接入真实的 API如数据库查询、邮件发送、代码执行在沙盒中、企业内部系统等。动态工作流根据任务类型让智能体动态生成或选择不同的子图工作流来执行实现更高层次的抽象。与 OpenClaw 深度集成探索使用 OpenClaw 的工作流编辑器可视化你的 LangGraph利用其提供的技能市场、版本管理和部署流水线。构建 AI 智能体是一个迭代过程。从一个小而专的智能体开始验证其价值然后逐步扩展其能力和可靠性是更稳妥的工程路径。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度