LangGraph实战用智普AI大模型打造你的第一个ToolAgent附完整代码最近在AI应用开发领域一个明显的趋势是开发者不再满足于让大语言模型仅仅进行对话或文本生成。我们更希望它能“动手做事”——调用外部工具、处理复杂流程、根据状态做出决策。这就像给一个聪明的“大脑”配上了“手脚”让它能真正融入我们的工作流。如果你也尝试过用传统的链式Chain或代理Agent模式来构建这类应用可能会遇到逻辑缠绕、状态管理混乱的麻烦。这时LangGraph的出现提供了一种更优雅的解决方案。LangGraph并非要取代LangChain而是作为其上的一个强大扩展。它引入了有向图Graph的编程范式让你能够清晰地定义AI应用的工作流。每个节点可以是一个LLM调用、一个工具执行或一个条件判断节点之间的边则定义了状态流转的路径。这种模式特别适合构建需要多步骤、有条件分支、甚至循环执行的复杂智能体Agent。今天我们就抛开理论直接上手用智普AI的大模型作为“大脑”结合LangGraph从零构建一个能调用真实工具的ToolAgent。整个过程你会看到如何定义工具、管理状态、设计节点并最终得到一个可复用的、结构清晰的智能体应用。1. 环境搭建与核心概念初探在开始敲代码之前我们需要先理清几个核心概念并准备好开发环境。LangGraph的核心思想是将一个任务流程建模为一张状态图State Graph。图中的每个节点都是一个可执行单元它们共享并操作一个全局的状态State。状态通常是一个字典Dict包含了任务执行过程中所有需要传递的信息比如用户输入、对话历史、中间结果等。这种设计带来的最大好处是显式的状态管理和清晰的控制流。相比于传统链式调用中隐式的状态传递在LangGraph中状态是明确定义和可见的。控制流则由图的边Edge来定义可以是无条件跳转到下一个节点也可以根据当前状态的某些属性进行条件分支Conditional Edge。这使得调试和理解整个Agent的决策过程变得直观许多。我们的实战将基于智普AI的GLM系列模型。选择智普AI的原因在于其API设计与OpenAI高度兼容特别是对tools函数调用功能的支持非常完善这让我们在定义和调用工具时几乎可以无缝迁移OpenAI生态的代码和经验极大地降低了开发门槛。首先我们来搭建环境。确保你已经安装了Python建议3.8以上版本然后通过pip安装必要的库。pip install langgraph langchain langchain-community langchainhub pydantic这里需要特别注意langgraph是一个独立的库它不会随着langchain的安装而自动安装必须手动添加。langchainhub用于拉取预置的智能体提示词模板pydantic则用于定义工具的参数模式。接下来我们需要设置智普AI的API密钥。你可以从智普AI的官方平台获取。为了安全起见不建议将密钥硬编码在代码中通常使用环境变量来管理。import os from langchain_community.chat_models import ChatZhipuAI # 推荐通过环境变量设置API Key # 在终端中执行export ZHIPUAI_API_KEYyour_api_key_here zhipuai_api_key os.getenv(ZHIPUAI_API_KEY) # 初始化智普AI的聊天模型 # 这里我们选择GLM-3-Turbo它在响应速度和成本间有很好的平衡 chat_model ChatZhipuAI( modelglm-3-turbo, api_keyzhipuai_api_key, temperature0.1 # 对于工具调用任务较低的temperature有助于输出更稳定、结构化的结果 )环境准备就绪后我们进入最有趣的部分为我们的Agent定义它能使用的“工具”。2. 定义Agent的“武器库”多功能工具一个ToolAgent的强大与否很大程度上取决于它的工具集。工具本质上是一个函数它封装了特定的能力比如查询天气、搜索数据库、调用某个API或者进行数学计算。LangChain和智普AI的tools接口让我们能够以一种标准化的方式将这些函数“描述”给大模型模型就能学会在何时、以何种参数去调用它们。在LangChain中定义工具有多种方式最便捷的是使用tool装饰器。这个装饰器会自动从你的函数签名和文档字符串中提取工具的名称、描述和参数模式并将其包装成模型能理解的格式。为了让模型更准确地理解参数我们通常会结合Pydantic的BaseModel来定义严格的参数模式Schema。让我们来定义三个具有代表性的工具覆盖信息查询、文本分析和数据获取三种常见场景。第一个工具天气查询工具。这是一个调用外部API的典型例子。它接收经纬度坐标从公开的天气服务API中获取当前温度。import requests import datetime from langchain.tools import tool from pydantic import BaseModel, Field class WeatherInput(BaseModel): 查询天气所需的经纬度坐标 latitude: float Field(description纬度例如40.7128) longitude: float Field(description经度例如-74.0060) tool(args_schemaWeatherInput) def get_current_weather(latitude: float, longitude: float) - str: 根据提供的经纬度坐标获取该地点的当前温度。 坐标应使用十进制度数表示。 # 使用一个免费的天气API BASE_URL https://api.open-meteo.com/v1/forecast params { latitude: latitude, longitude: longitude, hourly: temperature_2m, forecast_days: 1, } try: response requests.get(BASE_URL, paramsparams, timeout10) response.raise_for_status() data response.json() except requests.RequestException as e: return f请求天气数据失败{e} # 从返回的每小时数据中找到最接近当前UTC时间的数据点 current_utc_time datetime.datetime.utcnow() times [datetime.datetime.fromisoformat(t.replace(Z, 00:00)) for t in data[hourly][time]] temperatures data[hourly][temperature_2m] closest_idx min(range(len(times)), keylambda i: abs(times[i] - current_utc_time)) current_temp temperatures[closest_idx] return f当前位置纬度{latitude}, 经度{longitude}的当前温度约为 {current_temp}°C。注意在实际生产环境中调用外部API需要考虑错误处理、重试机制、速率限制和API密钥管理。这里为了示例清晰做了简化。第二个工具文本情感与语言分析工具。这个工具展示了如何处理和分析用户输入的文本内容。它接收一段文本返回其情感倾向正面/负面/中性和语言类型。from pydantic import BaseModel, Field class TextAnalysisInput(BaseModel): 需要分析的文本内容 text: str Field(description需要进行分析的文本字符串) tool(args_schemaTextAnalysisInput) def analyze_text_sentiment_and_language(text: str) - str: 分析一段文本的情感倾向和所使用的语言。 情感倾向分为正面(pos)、负面(neg)、中性(neutral)。 语言以ISO 639-1双字母代码表示如zh代表中文en代表英文。 # 注意在实际应用中这里应该调用一个情感分析模型或API。 # 为了示例我们返回一个模拟结果。你可以轻松地替换为调用智普AI的API进行分析。 # 模拟逻辑简单判断 positive_words [好, 棒, 优秀, 高兴, 喜欢] negative_words [差, 糟, 讨厌, 伤心, 不好] sentiment neutral language en # 默认英文 # 非常简单的语言检测仅作演示 if any(\u4e00 char \u9fff for char in text): language zh if any(word in text for word in positive_words): sentiment pos elif any(word in text for word in negative_words): sentiment neg else: if any(word.lower() in text.lower() for word in [good, great, excellent, happy]): sentiment pos elif any(word.lower() in text.lower() for word in [bad, terrible, hate, sad]): sentiment neg return f分析结果文本语言为 {language}情感倾向为 {sentiment}。第三个工具文本摘要与关键词提取工具。这个工具用于处理长文本生成摘要、识别语言和提取关键词。class TextOverviewInput(BaseModel): 需要生成概述的文本内容 content: str Field(description需要生成摘要和提取关键词的长文本) tool(args_schemaTextOverviewInput) def generate_text_overview(content: str) - str: 对输入的文本内容生成一个简洁的摘要识别文本语言并提取关键关键词。 适用于快速理解长文档的核心内容。 # 同样这里应该是调用LLM或专用NLP服务的地方。 # 我们模拟一个简单的处理流程。 words content.split() # 模拟摘要取前50个词 summary .join(words[:min(50, len(words))]) (... if len(words) 50 else ) # 模拟语言检测同上 language zh if any(\u4e00 char \u9fff for char in content) else en # 模拟关键词提取取频率较高的几个词实际应用应用TF-IDF等算法 from collections import Counter # 简单过滤掉常见短词 common_words {the, a, an, in, on, at, to, for, 是, 的, 了, 在} filtered_words [w.lower() for w in words if w.lower() not in common_words and len(w) 2] top_keywords [word for word, _ in Counter(filtered_words).most_common(3)] keywords_str , .join(top_keywords) if top_keywords else 未提取到显著关键词 return f**文本概述**\n- **摘要**: {summary}\n- **语言**: {language}\n- **关键词**: {keywords_str}将这三个工具放入一个列表就构成了我们Agent的初始工具集。tools [get_current_weather, analyze_text_sentiment_and_language, generate_text_overview]定义好工具后模型在推理时就能“看到”这些工具的描述并决定是否需要调用、调用哪一个以及传入什么参数。接下来我们需要设计一个“状态”容器来承载整个工作流中需要共享和传递的信息。3. 构建工作流的“记忆体”状态设计在LangGraph中状态State是贯穿整个图执行过程的共享数据存储。它定义了Agent在完成任务过程中需要记住什么、传递什么。一个好的状态设计能让图的逻辑清晰节点间的协作顺畅。状态通常被定义为一个TypedDict这为字典的键提供了类型提示增强了代码的可读性和安全性。对于我们的ToolAgent我们需要考虑哪些信息是必要的。回顾一下典型的多轮工具调用场景用户提出一个初始请求。AgentLLM思考决定是直接回答还是调用工具。如果调用工具则执行工具并得到结果。将工具结果作为上下文Agent再次思考决定下一步继续调用工具还是给出最终答案。重复步骤3-4直到任务完成。在这个过程中我们需要跟踪用户的原始输入这是任务的起点。对话历史在多轮交互中记录之前的对话内容保持上下文连贯。Agent的决策输出LLM每次“思考”的结果可能是一个直接回答AgentFinish也可能是一个工具调用指令AgentAction。中间步骤记录所有已经执行过的工具调用及其结果避免重复工作也为后续决策提供历史依据。基于此我们定义如下的AgentStatefrom typing import TypedDict, Union, Annotated, List import operator from langchain_core.agents import AgentAction, AgentFinish from langchain_core.messages import BaseMessage class AgentState(TypedDict): LangGraph Agent的状态定义。 它描述了在执行图中流动和更新的数据。 # 用户的最新输入 input: str # 对话历史由BaseMessage对象组成 chat_history: List[BaseMessage] # Agent的决策结果可能是要执行的动作(AgentAction)也可能是最终答案(AgentFinish) agent_outcome: Union[AgentAction, AgentFinish, None] # 已执行的动作观察结果对列表。 # 使用Annotated和operator.add注解意味着对这个字段的更新是“追加”操作 # 而不是覆盖。这是LangGraph处理状态更新的一个关键机制。 intermediate_steps: Annotated[List[tuple[AgentAction, str]], operator.add]对intermediate_steps字段的Annotated注解是LangGraph状态管理的精髓之一。operator.add告诉框架当多个节点尝试更新这个字段时应该将新的值追加append到现有列表之后而不是覆盖它。这完美契合了“记录步骤历史”的需求。其他字段如input和agent_outcome通常采用默认的覆盖式更新。状态定义好后它就成为了连接图中各个节点的“血管”。每个节点都可以读取状态的当前值并返回一个字典来指定要更新状态的哪些部分。接下来我们就来创建处理这些状态的“器官”——图的节点。4. 绘制智能体的“思维导图”节点与图构建有了状态和工具现在我们可以用LangGraph来组装Agent了。图的构建分为三步定义节点Node、定义边Edge最后编译Compile成可执行的应用。第一步定义节点函数。我们至少需要两个核心节点Agent节点负责“思考”。它接收当前状态调用大模型让模型根据当前输入、对话历史和已执行的步骤决定下一步是调用工具还是结束任务。工具执行节点负责“动手”。它接收Agent节点发出的工具调用指令执行对应的工具函数并将结果返回。首先我们需要创建一个基于智普AI模型和工具的“可运行智能体Runnable Agent”。LangChain的create_openai_tools_agent函数可以帮我们快速生成一个符合OpenAI工具调用格式的Agent。from langchain.agents import create_openai_tools_agent from langchain import hub # 从LangChain Hub拉取一个为工具调用优化过的提示词模板 # 这个模板会指导模型如何思考和使用工具 prompt hub.pull(hwchase17/openai-tools-agent) # 创建Agent Runnable # 它将我们的聊天模型、工具集和提示词模板组合在一起 agent_runnable create_openai_tools_agent(chat_model, tools, prompt)现在定义Agent节点函数。这个函数很简单调用agent_runnable传入当前状态得到模型的决策agent_outcome然后返回一个只包含agent_outcome的更新字典。def run_agent_node(state: AgentState) - dict: Agent节点调用LLM进行决策。 输入当前状态。 输出一个字典更新状态中的agent_outcome字段。 # 将状态传递给agent_runnable agent_outcome agent_runnable.invoke(state) # 返回需要更新的状态部分 return {agent_outcome: agent_outcome}接着定义工具执行节点。它需要从状态中取出最新的agent_outcome应该是一个AgentAction包含了要调用的工具名和参数然后执行它。LangGraph提供了一个方便的ToolExecutor来简化这个过程。from langgraph.prebuilt.tool_executor import ToolExecutor # 初始化工具执行器 tool_executor ToolExecutor(tools) def execute_tools_node(state: AgentState) - dict: 工具执行节点执行Agent选择的工具。 输入当前状态其中agent_outcome应为AgentAction。 输出更新intermediate_steps字段记录本次执行的动作和观察结果。 # 从状态中获取Agent的决策 agent_action state[agent_outcome] # 确保这是一个工具调用动作 if not isinstance(agent_action, AgentAction): # 理论上不应该进入这个分支因为只有AgentAction才会路由到此节点 raise ValueError(fExpected AgentAction, got {type(agent_action)}) # 使用ToolExecutor执行工具 observation tool_executor.invoke(agent_action) # 将本次执行的动作结果对追加到历史中 # 返回的字典中intermediate_steps键对应的值会被operator.add追加到原列表 return {intermediate_steps: [(agent_action, observation)]}第二步定义边与条件逻辑。节点定义好了我们需要规定它们如何连接。这里有一个关键的逻辑Agent节点执行后我们需要判断它的输出是AgentFinish任务完成还是AgentAction需要调用工具。这需要一个条件边Conditional Edge。def should_continue(state: AgentState) - str: 条件判断函数决定图的下一个节点。 输入当前状态。 输出字符串决定下一个节点的路由键。 outcome state[agent_outcome] if isinstance(outcome, AgentFinish): # 如果是最终答案则结束流程 return end else: # 如果是工具调用动作则继续到工具节点 return continue第三步组装图。现在我们可以使用StateGraph来构建完整的工作流了。from langgraph.graph import StateGraph, END # 1. 初始化一个状态图并指定状态的结构类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(agent, run_agent_node) workflow.add_node(action, execute_tools_node) # 3. 设置入口点从agent节点开始 workflow.set_entry_point(agent) # 4. 添加条件边从agent节点出发根据should_continue的结果决定去向 workflow.add_conditional_edges( agent, # 源节点 should_continue, # 条件判断函数 { continue: action, # 如果返回continue则前往action节点 end: END # 如果返回end则结束图执行 } ) # 5. 添加普通边从action节点执行完后无条件返回agent节点进行下一轮思考 workflow.add_edge(action, agent) # 6. 编译图生成一个可执行的Runnable对象 app workflow.compile()至此一个具备完整“思考-行动-再思考”循环的ToolAgent就构建完成了。图的结构非常清晰从agent开始判断是否需要行动需要则执行action执行完再回到agent思考直到agent认为任务完成。这种显式的图结构让复杂的多步推理流程一目了然。5. 运行、调试与进阶优化现在让我们来实际运行这个Agent看看它如何工作。我们用一个需要结合多个工具才能回答的复杂问题来测试它。# 准备输入状态 inputs { input: 先分析一下这句话的情感‘今天天气真糟糕但你的帮助很暖心’。然后如果这句话是中文请为它生成一个概述。, chat_history: [] # 初始对话历史为空 } # 调用编译好的图应用 print(开始执行Agent...) final_state app.invoke(inputs) print(\n--- 执行结束 ---\n) # 输出最终结果 if isinstance(final_state[agent_outcome], AgentFinish): print(最终回答) print(final_state[agent_outcome].return_values[output]) else: print(Agent未返回最终答案。) # 查看整个执行过程中的中间步骤 print(\n--- 执行的中间步骤 ---) for i, (action, observation) in enumerate(final_state.get(intermediate_steps, [])): print(f\n步骤 {i1}:) print(f 动作: {action.tool} - {action.tool_input}) print(f 结果: {observation[:200]}...) # 截取前200字符显示运行这段代码你会看到Agent的完整思考轨迹。它可能会先调用analyze_text_sentiment_and_language工具分析情感得到“中文”和“情感复杂可能先负后正”的结果后再调用generate_text_overview工具生成概述。所有的决策、调用和结果都被记录在intermediate_steps中方便我们复盘和调试。提示在实际开发中你可能会遇到模型不调用工具、工具参数解析错误等问题。调试时可以打印出agent_outcome的原始内容检查模型的思考过程。也可以调整提示词prompt或模型的temperature参数来影响其行为。我们的基础Agent已经能工作但一个健壮的生产级应用还需要更多考虑。以下是一些进阶优化方向1. 引入人工确认节点对于某些关键操作如发送邮件、修改数据库可以在工具调用前插入一个“人工确认”节点。这个节点可以暂停图执行等待用户输入如“确认”或“取消”再决定是否继续。2. 超时与错误处理在execute_tools_node中用try...except包裹工具调用将错误信息作为observation返回给Agent让它有机会尝试其他方案或向用户报错。3. 状态持久化与多轮对话当前的chat_history字段就是为多轮对话准备的。每次交互后将用户输入和Agent输出作为HumanMessage和AIMessage存入chat_history并在下一次调用时传入Agent就能记住之前的对话上下文。4. 并行与分支LangGraph支持更复杂的图结构比如并行执行多个工具节点add_node多个然后从条件边同时指向它们或者根据工具执行结果进行更复杂的分支判断。这可以用来构建工作流审批、多路径信息检索等高级应用。构建这个Agent的过程让我体会到LangGraph将Agent的“状态”和“流程”从隐式的代码逻辑中抽离出来变成了显式的、可视化的结构。这对于开发、调试和团队协作来说价值巨大。当你需要修改流程时不再是去追踪复杂的函数调用链而是直接修改图的结构。这种范式转变或许是开发复杂AI应用的下一个常态。