在开发基于大型语言模型的应用时,工具调用(Tool Calling)和它背后的感知行动循环,是我觉得最值得优先搞明白的组合。LangChain提供了把普通函数转成工具的各种能力,LangGraph又把这些工具调用串成可以多轮决策的循环。很多项目在Demo阶段看起来成果喜人,模型对答如流,可一旦接到真实业务里,要查数据库、调外部接口、更新订单状态,模型就无能为力了。让模型真正“动手干活”的核心机制就是工具调用,而当调用链路变得复杂、需要多轮决策时,LangGraph里的感知行动循环就是承载这一切的骨架。这篇文章,我想从最基础的自定义函数讲起,一直讲到完整感知行动循环的工程化实现,把我实际项目里验证过的方法、踩过的坑和代码结构完整整理一遍。
1. 工具调用的底层逻辑:模型负责选择,代码负责执行
1.1 一个反直觉的事实:模型并不会运行你的函数
第一次真正打通工具调用时,我盯着返回的消息看了很久。模型返回了一个tool_calls字段,里面是一个函数名、一份JSON格式的参数、一个唯一ID。它没有去执行任何代码。真正运行的,是我自己的Python函数。
这个分离是工具调用设计的精髓:模型只做“决策”——在给定工具清单里选哪个、参数怎么填;执行由你的代码完成。这也就带来了第一个工程化启示:模型的不确定性被隔离在了决策层。模型可能把参数猜错、把必填字段漏掉、甚至选了一个完全无关的工具,但只要你在执行层做好校验和转换,系统整体仍然是可控的。所以工具调用的工程化重点,不在“怎么调”,而在“怎么定义”——模型对每一个工具的全部理解,几乎都来自你给它的Schema和描述文本。
1.2 一次完整调用的四个环节
一个完整的工具调用闭环,由四个环节组成:
- 工具注册:把自定义函数转成模型可见的JSON Schema。
- 模型决策:模型根据用户请求,在工具清单中选择一个工具并生成参数。
- 代码执行:你的程序解析
tool_calls,调用真实函数。 - 结果回填:把执行结果作为一条
ToolMessage送还给模型,模型继续判断任务是否完成。
举个具体的例子。你定义了一个get_weather(city, days)函数,用户说“查一下青云市未来三天的天气”。模型不会直接执行Python,它只会在输出里生成一个结构:{"name": "get_weather", "args": {"city": "青云市", "days": 3}}。你的代码拿到这个结构后自己调用函数,再把结果以消息形式加回对话。模型看到结果后,给出最终回答或继续发起下一次调用。
这种设计看起来多了一层“中间人”,但它解决了一个根本问题:LLM不是稳定的函数执行器,而是概率性的意图识别器。它擅长的是判断“当前应该做什么”,而不是保证每行代码正确执行。把执行交给确定性代码,把决策交给模型,两边各干各擅长的。
这里也回应一个常见的疑问:为什么不直接在工具调用里自动执行代码?因为自动执行的前提是参数百分百正确、调度百分百安全,这在真实业务里根本不成立。你需要在中间插入校验、权限判断、限流和日志。工具调用把这一步显式暴露给了开发者,其实是一件好事。
2. 三种把自定义函数变成工具的方法与选型权衡
2.1 最省事的@tool装饰器:绝大多数场景的首选
在LangChain里,把一个普通函数变成工具,最直接的方式是加一个装饰器。
from langchain_core.tools import tool @tool def get_weather(city: str, days: int = 1) -> dict: """查询指定城市未来几天的天气情况。 Args: city: 城市名称,使用中文,例如“青云市”、“临江市”。 days: 查询的天数,默认1天。 Returns: 包含温度、湿度、风速等信息的字典。 """ # 这里写真实的数据获取逻辑,可以是调用天气服务商的API return {"city": city, "days": days, "temperature": 26.5, "humidity": 60}用@tool之后,LangChain会读取函数的签名、类型注解和docstring,自动生成JSON Schema。模型看到的工具定义,本质上就是这段结构化的描述:
{ "name": "get_weather", "description": "查询指定城市未来几天的天气情况。", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,使用中文" }, "days": { "type": "integer", "description": "查询天数" } }, "required": ["city"] } }这里想强调一点:函数的docstring直接决定了模型会不会正确使用这个工具。光写“查询天气”远远不够。我在项目中总结的经验是,docstring里要包含三个要素——什么时候用这个工具、每个参数到底填什么、有什么限制条件。模型在决策时会把docstring当作“使用说明”来读,说明写得越清楚,模型的选择越准确。
2.2 需要精细控制Schema时:StructuredTool.from_function
当你需要特别控制某个参数的描述、默认值、取值集合时,@tool装饰器就不够细了。这时可以用StructuredTool.from_function配合Pydantic模型显式定义输入Schema。
from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str = Field( description="城市中文名称,例如青云市、临江市、海州市", ) days: int = Field( default=1, ge=1, le=7, description="查询天数,范围为1到7天", ) def get_weather(city: str, days: int = 1) -> dict: return {"city": city, "days": days} weather_tool = StructuredTool.from_function( func=get_weather, name="get_weather", description="当用户询问天气时使用。", args_schema=WeatherInput, )Field里的ge、le会被编进Schema的约束,模型在生成参数时会尽量遵守这些约束。实际测试中,给days加上范围约束后,模型几乎不再产生days=0或days=10这类不合理请求。
2.3 完全自定义行为时:继承BaseTool手动实现
第三种方式最灵活但代码量也最大:继承BaseTool。它适合工具本身带有状态、需要特殊序列化逻辑、或者要与某个第三方库深度绑定的情况。这里给出一个骨架:
from langchain_core.tools import BaseTool from typing import Type from pydantic import BaseModel, Field class WeatherInput(BaseModel): city: str = Field(description="城市中文名称") class CustomWeatherTool(BaseTool): name: str = "get_weather" description: str = "当用户询问天气时使用" args_schema: Type[BaseModel] = WeatherInput api_key: str = "" # 工具内部状态 def _run(self, city: str) -> dict: # 这里可以使用 self.api_key 调用真实API return {"city": city, "temperature": 26.5}继承BaseTool时要注意:实际执行逻辑写在_run(同步)和_arun(异步)里,这两者至少要实现一个。如果你的工具有状态,要格外小心实例复用时的状态污染。
三种方式的权衡,我对比如下:
| 实现方式 | 代码量 | 灵活性 | Schema精细度 | 适用场景 |
|---|---|---|---|---|
@tool装饰器 | 少 | 中 | 依赖注解和docstring | 大多数业务函数 |
StructuredTool.from_function | 中 | 较高 | 可完全用Pydantic控制 | 参数约束多的接口 |
继承BaseTool | 多 | 高 | 完全自定义 | 带状态、特殊行为 |
2.4 工具集合的组织方式
工具一多,很容易出现名字撞车、描述相互干扰的问题。我的做法是给工具集做一个统一前缀:用户相关的工具叫user_get_info、user_update_profile,订单相关的叫order_query_status、order_cancel。这样工具名本身自带归属信息,模型在决策时也更不容易混乱。
整个工具列表最好维护在一份常量里,单一来源,避免在多个地方重复定义:
ALL_TOOLS = [ user_get_info, user_update_profile, order_query_status, order_cancel, ]后面LangGraph绑定工具时,直接把ALL_TOOLS打进去就行。
3. 工具定义里的隐藏决定因素:描述、Schema和返回值的工程化约定
3.1 Description和docstring才是“模型眼中的工具文档”
工具调用的底层是让模型去识别“什么情况下该用什么工具”,而模型唯一的信息来源就是工具名、描述和参数Schema。我见过很多初学同学把description写成“天气查询”四个字,然后抱怨模型老是不主动调用。原因很简单:模型不知道这个工具到底解决什么问题、在什么语境下触发。
我实际项目里的经验是,一个合格的description应当包含三部分:触发场景、参数要求、边界条件。比如:
当用户询问城市天气情况时使用。城市参数必须是中文名称, 如果用户只说了“天气”但没有城市,必须先询问用户具体城市。把这句话完整写进description后,模型乱调工具的概率大幅下降。
docstring也是同理。工具函数内部的docstring会被LangChain提取,描述每一个参数的具体含义。我一直建议:docstring里的参数说明要按“用户可能怎么问”来写,而不是按“变量叫什么”来写。变量叫city,用户不会说“city是什么”,用户会说“临江市天气怎样”——所以docstring里应该写明city参数接收“临江市”这样的城市名。
3.2 参数Schema的常见坑位
第一,类型注解尽量收窄。不要写param: Any或者param: Union[str, int],模型在宽松的类型约束下更容易给出模糊或错误的值。能用Literal枚举就尽量用:
from typing import Literal class QueryInput(BaseModel): granularity: Literal["hour", "day", "week"] = Field( description="时间粒度" )第二,必填与可选要慎重。Pydantic里没有默认值的字段默认必填;有默认值的字段模型可以不提供。如果你希望某个场景下模型必须向用户追问信息,就把参数设为必填。反过来,如果某个参数不是每次都需要,不要设置成必填,否则模型会尝试编造一个值填进去。
第三,所有参数都写description。不写description的参数对模型来说就是个“黑盒”,它只能猜。猜对了还好,猜错了就会出现把订单号填成用户昵称这种事故。
3.3 返回值的处理逻辑:模型读取的是文本
工具执行之后,返回结果要转成ToolMessage回填给模型。模型本质上读的是文本,所以返回值要尽量“可读”。我的习惯是:
- 返回精简的扁平结构,不要嵌套一堆无用层级。
- 字段名用自然语言或常见缩写,让它本身就有语义。
- 如果数据量很大,不直接塞全部数据,而是塞摘要。例如查询订单列表有几千条,模型不需要看全部,只需要知道“共97条,最近一条订单号是xxx,状态是已发货”。
这里有一个细节容易被忽略:ToolMessage的tool_call_id必须对应AIMessage里tool_calls数组里的id。LangGraph内部依赖这个ID把工具结果和对应的调用请求配对。如果你手动构造消息时漏掉或写错了这个ID,循环就会出现“工具结果找不到归属”的问题,甚至直接报错。
我一般这样构造ToolMessage:
from langchain_core.messages import ToolMessage tool_call = message.tool_calls[0] result = get_weather(**tool_call["args"]) tool_message = ToolMessage( content=str(result), tool_call_id=tool_call["id"], )AIMessage里每一个tool_call都有唯一ID,执行完相应工具后,用这个ID回填,才能保证消息链完整。
4. 在LangGraph中搭建感知行动循环:代码拆解与运行机制
4.1 什么是感知行动循环,为什么需要它
标题里的“感知行动循环”,本质上就是Agent里那个“思考-行动-观察-再思考”的闭环。放到代码层面看其实很简单:模型根据当前所有可观察到的信息(用户问题、历史对话、前面的工具结果)决定下一步行动;如果行动是调用工具,就执行工具并把结果作为新的观察反馈给模型;模型再根据新增信息继续决策,直到它认为任务完成。
为什么不能用“调用一次模型+执行一次工具”就完事?因为真实任务往往需要多步。用户说“帮我查一下最近一笔订单,如果还没发货就取消它”,这至少需要两步:先查订单,再根据订单状态决定是否取消。每一步都依赖前一步的结果。感知行动循环解决的就是这种“顺序依赖、动态决策”的问题。
4.2 用StateGraph实现循环:核心代码
我直接给出一个可运行的结构。这里用LangGraph的StateGraph管理状态流转。
from typing import Annotated, Literal, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import AIMessage, ToolMessage class AgentState(TypedDict): messages: Annotated[list, add_messages] def call_model(state: AgentState, llm, tools): # 把工具绑定到模型,让模型知道当前有哪些工具可选 model_with_tools = llm.bind_tools(tools) response = model_with_tools.invoke(state["messages"]) return {"messages": [response]} def call_tools(state: AgentState, tools): tool_map = {tool.name: tool for tool in tools} last_message = state["messages"][-1] output_messages = [] for tool_call in last_message.tool_calls: tool = tool_map[tool_call["name"]] try: result = tool.invoke(tool_call["args"]) content = str(result) except Exception as exc: # 注意:错误也要回填给模型,让模型决定如何处理 content = f"工具执行失败: {type(exc).__name__}: {exc}" output_messages.append( ToolMessage(content=content, tool_call_id=tool_call["id"]) ) return {"messages": output_messages} def should_continue(state: AgentState) -> Literal["tools", "end"]: last_message = state["messages"][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return "tools" return "end"然后是构图:
builder = StateGraph(AgentState) builder.add_node( "agent", lambda state: call_model(state, llm=llm, tools=ALL_TOOLS) ) builder.add_node( "tools", lambda state: call_tools(state, tools=ALL_TOOLS) ) builder.set_entry_point("agent") builder.add_conditional_edges( "agent", should_continue, {"tools": "tools", "end": END} ) builder.add_edge("tools", "agent") app = builder.compile()运行:
result = app.invoke({"messages": [{"role": "user", "content": "查一下青云市明天天气"}]})4.3 循环运行机制逐一说透
图里一共两个普通节点:agent节点执行模型推理,tools节点执行工具调用。关键在should_continue这个条件边:它检查agent节点产出的最后一条消息有没有tool_calls。如果有,进入tools节点;如果没有,直接走向END。
tools节点执行完成后,有一条add_edge("tools", "agent")把控制权交还给agent。此时工具结果已经以ToolMessage的形式并入了state里的messages列表。agent节点再次读取messages时,就同时拥有了用户问题、历史对话和行动结果——这就是“感知”部分的完整实现。
这里有一个与工程设计密切相关的细节:messages字段使用了Annotated[list, add_messages]注解。add_messages是一个reducer,它确保每次节点返回的新消息都追加到已有的消息列表尾部,而不是覆盖。没有这个reducer,循环里每走一步,前面的消息就被冲掉了,模型根本没有办法看到工具执行结果。
在call_tools函数里我加了一段异常捕获,这个非常重要。工具执行必然会遇到网络超时、参数不合规、第三方服务返回错误码等情况。如果让异常直接抛到LangGraph内部,整个图就崩了,用户看到的是错误栈。更合理的工程化做法是:把错误信息转成正常的ToolMessage返回给模型,让模型判断怎么处理——它可以换个参数重试,也可以直接告诉用户失败原因。这样等于把一部分容错也交给了模型决策循环。
4.4 多工具结果与并行执行
一个值得注意的点是:模型在一个节点里可能同时产生多个tool_calls,也就是并行工具调用。比如用户问“青云市和临江市的天气分别是多少”,模型可能一次性生成两个get_weather调用,每个参数不同。上面这段代码里的for循环天然支持这种情况:它遍历last_message.tool_calls,逐个执行,把结果统一追加到消息列表。模型在下一轮会同时看到两个ToolMessage,自然能把信息综合成最终回答。
不过要注意,工具是并行返回还是顺序执行,取决于你的代码。在LangGraph中执行工具时,如果想让多个独立工具真正并行跑,可以改用asyncio.gather或用ThreadPoolExecutor包装。不过顺序执行在多数场景已经够用,还省去了一堆并发状态管理的问题。我的建议是:不是高并发场景,先不要上异步,保持执行顺序,日志和排障都会简单很多。
4.5 防止死循环:控制最大步数
感知行动循环理论上会一直转下去:模型每次都能找到理由再调一个工具。为了防止无限循环,LangGraph给了recursion_limit参数:
from langgraph.errors import GraphRecursionError try: result = app.invoke( {"messages": [{"role": "user", "content": "请帮我分析这个订单"}]}, config={"recursion_limit": 15} ) except GraphRecursionError: result = {"messages": [AIMessage(content="抱歉,这个任务需要的步骤太多,暂时无法完成。")]}把recursion_limit设为15,意味着整个图最多执行15个节点,超过就报错。实际项目里我会把默认值设在10到20之间,具体看Agent复杂度。捕获GraphRecursionError后可以做降级回复,至少不能让用户看到一段原始异常信息。
5. 多工具协作、错误兜底与调用成本控制:落地才能用的细节
5.1 工具数量膨胀后的治理策略
当工具数量超过10个,模型的决策质量会明显下降。我曾经在一个项目里挂了30多个工具,系统提示词被工具Schema撑得非常长,模型开始出现“选择困难”,甚至调用完全不相关的工具。后来总结下来,有三件事值得做。
第一,按业务域拆分工具集。不要把所有工具一股脑丢给模型。可以先做一个路由分类,只把与当前用户意图相关的10个以内工具传入bind_tools。这一步对稳定性和成本都有帮助。
第二,精简description。工具数量多的时候,每个description里冗余的话都是模型推理时的噪音。把description控制在两三句话内,只保留触发条件和最重要的参数约束。
第三,工具命名规律化。前面提到过加前缀,order_、user_、product_这种前缀能让工具名本身带有分类信息,模型在判断该用哪个工具时多了一个线索。
5.2 工具内部的三层防护
给工具本身加防护,是工程化落地时绕不开的事:
- 入参校验:无论模型是否按Schema传参,工具内部都要再校验一遍。特别是外部系统接口,不能信任模型生成的任何值。
- 超时与降级:所有调用外部API的工具,必须设置合理超时。外部服务如果挂掉,不能让Agent里等着的工具调用卡住整个循环。
- 权限边界:工具能干的动作要收敛到最小必要范围。一个负责“读”的工具绝不能写;写操作的工具必须要求用户二次确认,这可以在工具描述里写明“删除操作必须先向用户确认”。
我在实际项目中还用了一种简单有效的方案:给高风险的写操作工具加一个confirm必填参数,模型在调用这类工具时必须同时返回用户是否已确认。如果用户明确说“不用了”或者态度模棱两可,模型就无法完成调用,因为它缺一个必填字段。
5.3 成本控制:token都花在了哪里
工具调用对token的消耗,很多人一开始并没意识到。每一次模型推理,工具Schema都会完整地出现在请求里。也就是说,工具越多、参数描述越长,固定消耗就越大。而且每次循环都会重新计一遍。控制成本可以从这几个角度下手:
- 工具返回内容限长:工具结果会作为下一轮模型输入的一部分。如果返回一个几千行的JSON,下一轮模型处理的token会暴涨。工具层应当统一做截断或摘要。
- 按需绑定工具子集:如前面说的,不是所有工具都需要绑定给模型。用户问天气就只绑天气相关工具,每轮推理的固定token会低很多。
- 限制最大步数:
recursion_limit除了防死循环,也是在控成本。每多跑一轮,就多花一轮推理的token。
5.4 可观测性:排查问题的“摄像头”
循环式Agent最让人难受的地方是“黑盒”。模型在内部做了好几轮工具调用,最后才给出答案,中间到底发生了什么,如果不做观测,几乎无法排查。我的做法是给工具调用和模型推理都加上日志与追踪。
给LangGraph应用接入可观测性时,推荐在节点函数里记录关键信息:进入agent节点时的消息数量、每次模型调用返回是否包含tool_calls、每个工具的执行耗时和返回内容长度。工具内部记录入参和出参。排查线上问题时,这些日志往往能直接定位到是“模型选错了工具”还是“工具处理数据出了错”。
我自己习惯把运行链路先打出来:
[Agent] 第1轮,messages=3,tool_calls=1 [Tool] get_weather 入参 {"city": "青云市"} 耗时0.8s 返回长度42 [Agent] 第2轮,messages=4,tool_calls=0 [Agent] 结束,输出最终回答这种逐行日志虽然朴素,但任何Agent部署跑起来后,第一件能让你放心的事,就是能看到每一步的轨迹。
最后再分享一个我个人的经验。最初我实现感知行动循环时,习惯把工具内部的逻辑写得很重——工具里做参数修正、数据清洗、错误重试,想着这样模型拿到的一定是干净结果。结果适得其反,工具越重,模型的决策越飘,因为工具的描述和它实际能做的差异化越来越大。后来我把几乎所有复杂逻辑都从工具里剥出去,让每个工具只做一件事:校验入参、调用外部依赖、返回精简结果。需要多步骤完成的复杂任务,拆成多个工具,让模型一步步决策。工具简单了,模型的调用准确率反而大幅提升,整个循环的可解释性也好了很多。
如果你的项目刚好也卡在“模型乱调工具”或者“Agent跑飞”的问题上,不妨先检查一件事:你的工具是不是把“理解”和“执行”混在一起了。把工具做得足够薄、描述写得足够清楚、循环步数控制得足够严,这套东西其实比想象中稳定。