目录一、前言二、Langchain 工具常用高级特性2.1 return_direct2.1.1 案例代码2.2.2 return_direct 参数说明2.2 InjectedToolCallId 使用2.2.1 案例代码2.3 ToolException2.3.1 案例代码2.4 handle_tool_error2.4.1 示例代码2.4.2 参数补充说明2.4.3 带Agent 的错误处理三、InjectedState 与 InjectedStore使用3.1 InjectedState 使用3.1.1 案例代码3.1.2 Agent 中使用 InjectedState3.2 Agent 中结合 Store使用3.3 InjectedState vs InjectedStore 对比四、写在文末一、前言在之前的文章中我们分享了Langchain 中工具Tool的用法工具在整个Langchain 体系中都占据着非常重要的作用可以说是连接大模型与应用系统很重要的结合点本篇将继续深入探讨一些关于Langchain 工具的高级特性。二、Langchain 工具常用高级特性接下来将介绍工具的高级特性return_direct、InjectedToolCallId、ToolException 和错误处理。2.1 return_direct顾名思义直接返回最终结果默认情况下工具执行后结果会返回给模型模型再基于工具结果生成最终回复。但有时工具结果本身就是你想要的最终答案。设置return_directTrue后工具执行完就立即结束 Agent 循环工具返回内容直接作为最终输出。2.1.1 案例代码在下面的案例代码中使用tool装饰器定义了两个工具方法其中一个设置return_directTruefrom dotenv import load_dotenv import os from langchain.tools import tool from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage from langchain_openrouter import ChatOpenRouter load_dotenv(overrideTrue) OPENROUTER_API_KEY os.getenv(OPENROUTER_API_KEY) OPENROUTER_API_BASE os.getenv(OPENROUTER_API_BASE) model ChatOpenRouter( modeldeepseek/deepseek-v4-flash, api_keyOPENROUTER_API_KEY, base_urlOPENROUTER_API_BASE, ) # 普通工具结果返回给模型模型再做总结 tool def search_normal(keyword: str) - str: 搜索小米教程的课程普通模式 return f搜索结果Python3 基础教程、Python 数据分析、Python 爬虫入门 # return_direct 工具结果直接作为最终输出 tool(return_directTrue) def search_direct(keyword: str) - str: 搜索小米的课程直接返回模式。 当用户只需要搜索结果不需要额外分析时使用此工具。 return f搜索结果Python3 基础教程、Python 数据分析、Python 爬虫入门 # 对比普通模式 vs 直接返回模式 agent_normal create_agent( modelmodel, tools[search_normal], system_prompt你是小米教程的学习顾问。, ) agent_direct create_agent( modelmodel, tools[search_direct], system_prompt你是小米教程的学习顾问。, ) # 普通模式模型会基于搜索结果再生成一段总结 result agent_normal.invoke({ messages: [HumanMessage(content搜索 Python 课程)] }) print( 普通模式模型会再加工) print(result[messages][-1].content[:150]) # 直接返回模式工具结果就是最终答案 result agent_direct.invoke({ messages: [HumanMessage(content搜索 Python 课程)] }) print(\n 直接返回模式工具结果即最终答案) print(result[messages][-1].content[:150])运行上面的代码通过输出效果可以看到设置return_directTrue 之后工具中直接将查到的结果返回给用户而不走大模型的二次整理2.2.2return_direct参数说明下面对return_directTrue 参数设置做补充说明模式工具执行后适用场景return_directFalse默认模型收到工具结果 → 模型继续思考 → 生成最终回复需要分析/总结/进一步决策return_directTrue工具执行后立即结束 → 工具结果就是最终输出查询类、数据获取类、已格式化好的结果当设置 return_directTrue 时Agent 会跳过后续的模型思考步骤直接返回工具自身的结果。这在节省 Token 和降低时延方面非常有价值但也意味着模型不会对工具结果做任何二次加工。注意如果一个 Agent 同时挂载了多个工具其中不仅有 return_directTrue 的工具也有普通工具只要模型在这一轮调用中触发了任意一个 return_direct 工具Agent 循环就会立即结束——即使同一轮还并行调用了其他普通工具它们的结果也不会再被模型加工总结。设计包含多个工具的 Agent 时要留意这一点避免该总结的内容被跳过。2.2 InjectedToolCallId 使用有时工具需要知道是谁调用了它方便在后续追溯问题业务审计时使用这种情况下可以考虑使用InjectedToolCallId可以在工具函数中注入当前的 tool_call_id。2.2.1 案例代码参考下面的完整示例代码在下面的代码中定义了一个模拟登录的方法传入一个InjectedToolCallId 类型的参数在执行过程中作为一个类似记录日志ID的参数from typing import Annotated from langchain.messages import ToolCall from langchain.tools import tool, InjectedToolCallId tool def log_user_action( action: str, tool_call_id: Annotated[str, InjectedToolCallId], ) - str: 记录用户操作到日志系统。 Args: action: 用户操作描述 tool_call_id: 系统自动注入的工具调用 ID # 实际项目中这里会写入数据库或发送到日志服务 return f操作已记录 (调用ID: {tool_call_id}): {action} # 手动模拟模型产生的 ToolCall tool_call: ToolCall { name: log_user_action, args: { action: 用户查询了 Python 课程 }, id: call_log_001, type: tool_call, } # InjectedToolCallId 会从 tool_call.id 中自动注入 result log_user_action.invoke(tool_call) print(result)运行效果如下带有InjectedToolArg标记的参数不需要由 Agent模型提供 这些参数由 LangChain 运行时在执行工具时自动注入。 由于这些参数不会出现在工具的 schema 中模型无法看到它们 因此不应该把它们作为用户需要填写的工具参数进行描述。例如InjectedToolCallId是一种特殊的注入参数 LangChain 会在工具执行阶段自动传入当前 ToolCall 的唯一 ID。 它不会出现在模型生成的参数中但会和当前这一次工具调用自动绑定。直接调用.invoke()并手动构造ToolCall 只是为了演示 LangChain 如何完成参数注入机制。实际项目中通常不会手动传递这个 ID。在 Agent 工作流中InjectedToolCallId更常用于需要关联当前调用上下文的场景 例如配合Command对象在工具内部更新 Agent 状态 向messages列表追加关联当前调用的ToolMessage 或者实现工具调用追踪、审计、日志关联等功能。这也是它被设计为注入参数而不是普通参数的原因 它代表的是 LangChain Runtime 当前正在执行的这一次 ToolCall 上下文 而不是用户输入的一部分。2.3 ToolException实际上工具在执行过程中可能会出错此时可以使用ToolException抛出明确的工具异常让 Agent 知道出了问题方便做后续的业务处理。2.3.1 案例代码如下完整的示例代码在自定义的工具方法中在可能出现代码异常的位置使用 raiseToolException将异常抛出然后再在invoke调用的时候使用 try - except 进行捕获from langchain.tools import tool, ToolException tool def get_user_info(user_id: int) - str: 根据用户 ID 查询用户信息。 Args: user_id: 用户 ID必须是正整数 # 数据校验 if user_id 0: # 抛出 ToolException而不是普通 Exception # ToolException 会被 Agent 捕获并告知模型 raise ToolException(f用户 ID 必须为正整数收到了: {user_id}) # 模拟数据库查询 users { 1: 张三VIP 会员注册于 2024-01-15, 2: 李四普通用户注册于 2024-03-20, } if user_id not in users: raise ToolException(f未找到 ID 为 {user_id} 的用户) return users[user_id] # 正常调用 print(get_user_info.invoke({user_id: 1})) # 异常调用 1无效 ID try: get_user_info.invoke({user_id: -1}) except ToolException as e: print(f工具异常: {e}) # 异常调用 2用户不存在 try: get_user_info.invoke({user_id: 999}) except ToolException as e: print(f工具异常: {e})运行上面的代码效果如下注意这里之所以能在 .invoke() 外层用 try/except 捕获到 ToolException是因为工具默认的handle_tool_error为 False —— 异常不会被工具自己吞掉而是照常向上抛出。如果希望工具自己扛住错误、把错误信息转成字符串返回给模型而不是抛异常就是接下来要说的 handle_tool_error。2.4 handle_tool_error如果希望某个工具执行错误的时候程序不会中断而是把错误信息转成一段文本、当作正常返回值交给模型自己去理解和修正时可以在定义工具时设置handle_tool_error注意是单数没有 s。这是 BaseTool 上的一个属性最简单的设置方式是直接写在 tool 装饰器里2.4.1 示例代码参考下面的完整示例代码自定义一个get_weather的工具方法然后设置get_weather.handle_tool_error True 即可达到效果from langchain.tools import tool, ToolException from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage from langchain_openrouter import ChatOpenRouter from dotenv import load_dotenv import os # handle_tool_errorTrue出错时不再抛异常 # 而是把 ToolException 的内容转成字符串正常返回给模型 tool def get_weather(city: str) - str: 查询指定城市的天气。 Args: city: 城市名称必须是中文全称如 杭州、北京 weather_data { 杭州: 晴25°C, 北京: 多云18°C, 上海: 小雨22°C, } if city not in weather_data: # 城市不在数据中时抛出 ToolException raise ToolException( f未收录城市 {city}。 f可使用城市{, .join(weather_data.keys())}。 f请使用中文城市全称。 ) return f{city}天气{weather_data[city]} # handle_tool_error 是工具实例的属性不是 tool 装饰器的参数 get_weather.handle_tool_error True # 测试用错误城市名调用 # handle_tool_errorTrue 时错误信息会作为正常结果返回而不是抛出异常 result get_weather.invoke({city: 北境}) print(fhandle_tool_errorTrue: {result}) # Agent 复用上面已启用 handle_tool_error 的 get_weather 工具。 # 该配置仅处理此工具抛出的 ToolException不是整个 Agent 的全局错误兜底。 load_dotenv(overrideTrue) OPENROUTER_API_KEY os.getenv(OPENROUTER_API_KEY) OPENROUTER_API_BASE os.getenv(OPENROUTER_API_BASE) model ChatOpenRouter( modeldeepseek/deepseek-v4-flash, api_keyOPENROUTER_API_KEY, base_urlOPENROUTER_API_BASE, ) agent create_agent( modelmodel, tools[get_weather], system_prompt你是一个天气查询助手。, ) result agent.invoke({ messages: [HumanMessage(content查询北境的天气)] }) print(\n Agent 收到 handle_tool_error 转换后的错误信息并据此向用户解释 ) print(result[messages][-1].content[:200])运行上面的代码看到下面的效果2.4.2 参数补充说明handle_tool_error行为适用场景False默认ToolException 照常向上抛出调用方需自行 try/except不可恢复的错误需要中断流程True捕获 ToolException将其内容作为工具的正常返回值交给模型希望 Agent 自行阅读错误信息并修正/重试str捕获异常后用指定的固定字符串替换错误信息不想暴露具体报错细节只给统一提示Callable[[ToolException], str]捕获异常后用自定义函数处理异常并生成返回内容需要按异常类型/内容定制不同的提示注意不要用tool.with_config(handle_tool_errorsTrue)这种写法 ——with_config()是 Runnable 通用的运行时配置方法只用于设置 callbacks、tags、metadata 等并不会真正修改工具的错误处理行为。控制单个工具的错误处理要用tool(handle_tool_error...)单数如果想在 Agent/图Graph层面统一为所有工具配置错误处理策略则是在底层的 ToolNode 上设置 handle_tool_errors复数而不是作用在单个工具对象上。两者字段名相似但作用层级不同使用时注意区分。2.4.3 带Agent 的错误处理下面的案例代码中演示了如何结合Agent 一起使用from langchain.tools import tool, ToolException from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage from langchain_openrouter import ChatOpenRouter from dotenv import load_dotenv import os tool def book_course(user_name: str, course_name: str) - str: 为用户预订小米教程的课程。 Args: user_name: 用户姓名 course_name: 课程名称 # 校验用户是否存在 valid_users {张三, 李四, 王五} if user_name not in valid_users: raise ToolException( f用户 {user_name} 不存在。 f有效用户{, .join(sorted(valid_users))} ) # 校验课程是否存在 valid_courses {Python3 基础教程, HTML 基础教程, Java 面向对象} if course_name not in valid_courses: raise ToolException( f课程 {course_name} 不存在。 f有效课程{, .join(sorted(valid_courses))} ) return f已为 {user_name} 成功预订《{course_name}》 # handle_tool_error 是工具实例的属性不是 tool 装饰器的参数 book_course.handle_tool_error True load_dotenv(overrideTrue) OPENROUTER_API_KEY os.getenv(OPENROUTER_API_KEY) OPENROUTER_API_BASE os.getenv(OPENROUTER_API_BASE) model ChatOpenRouter( modeldeepseek/deepseek-v4-flash, api_keyOPENROUTER_API_KEY, base_urlOPENROUTER_API_BASE, ) agent create_agent( modelmodel, tools[book_course], system_prompt你是小米教程的课程顾问。, ) # 正常调用 result agent.invoke({ messages: [HumanMessage(content帮张三预订 Python3 基础教程)] }) print(f成功: {result[messages][-1].content[:100]}) # 错误调用用户不存在 result agent.invoke({ messages: [HumanMessage(content帮赵六预订 Python3 基础教程)] }) print(f\n错误-用户不存在:) for msg in result[messages]: if msg.type tool: print(f [{msg.type}] {msg.content[:80]}) elif msg.type ai and msg.content: print(f [{msg.type}] {msg.content[:100]})运行效果如下可以看到工具设置了handle_tool_errorTrue后ToolException 的内容会以一条 ToolMessage 的形式正常出现在对话历史里上面输出中的 [tool] 那一行而不会让程序崩溃模型看到这条错误信息后会用自然语言把它转达给用户并给出可用的替代方案。这正是 return_direct、InjectedToolCallId、ToolException 和 handle_tool_error 组合使用时的典型效果既保证了工具调用的健壮性又不牺牲用户体验。三、InjectedState 与 InjectedStore使用有时候工具需要访问更多上下文信息比如当前对话的状态、用户的持久化数据等。LangChain 通过依赖注入机制让工具函数能够自动获取这些信息。这里涉及到两个重要的组件InjectedState 与 InjectedStore下面分别说明。3.1 InjectedState 使用默认情况下工具只能通过参数接收模型传来的数据。但有时工具需要知道当前对话的上下文 —— 比如之前的对话历史、用户已确认的信息等。InjectedState让工具可以直接读取 Agent 的完整状态。3.1.1 案例代码如下完整示例代码在自定义工具方法参数中使用InjectedState 修饰入参 statefrom typing import Annotated, Any from dotenv import load_dotenv load_dotenv() from langchain.tools import tool, InjectedState from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage tool def remember_preference( preference: str, state: Annotated[dict[str, Any], InjectedState], ) - str: 记住用户的偏好设置。 Args: preference: 用户的偏好内容 state: 系统自动注入的当前 Agent 状态 # 从状态中获取之前的消息历史 messages state.get(messages, []) message_count len(messages) # 可以读取状态中的任何字段 previous_prefs state.get(user_preferences, 无) return ( f已记住偏好: {preference}。 f(当前对话共 {message_count} 条消息 f之前偏好: {previous_prefs}) ) # InjectedState 会被自动注入Agent 不需要传这个参数 result remember_preference.invoke({ preference: 暗色主题, state: {messages: []} # 手动补上 state }) print(result)运行结果如下说明InjectedState 让你可以访问 AgentState 中的所有字段。如果你扩展了 state_schema增加了自定义字段这些字段也可以被 InjectedState 读取到。3.1.2 Agent 中使用 InjectedState有时候需要在工具中访问持久化存储因为Agent 状态state是对话级别的对话结束就没了。而Store是跨会话的持久化存储可以用来保存用户偏好、学习进度等长期信息。InjectedStore让工具可以直接读写 Store。from typing import Annotated, TypedDict from langgraph.store.base import BaseStore from langgraph.store.memory import InMemoryStore from langgraph.prebuilt import ToolNode, InjectedStore from langgraph.graph import StateGraph, START, END from langchain.tools import tool from langchain_core.messages import AIMessage, AnyMessage # # 1. 存储与预置数据 # store InMemoryStore() store.put((users, user_001), profile, { data: { name: 小明, level: 入门, completed_courses: [HTML 基础教程], } }) # # 2. 工具定义与之前一致 # tool def get_user_profile( store: Annotated[BaseStore, InjectedStore()], ) - str: 获取当前用户的学习档案信息。 item store.get((users, user_001), profile) if item is None: return 未找到用户档案 profile item.value[data] return ( f用户档案姓名{profile[name]} f水平{profile[level]} f已完成课程{, .join(profile[completed_courses])} ) tool def save_course_progress( course_name: str, store: Annotated[BaseStore, InjectedStore()], ) - str: 保存用户的学习进度到持久化存储。 item store.get((users, user_001), profile) profile item.value[data] if item else { name: 小明, level: 入门, completed_courses: [], } if course_name not in profile[completed_courses]: profile[completed_courses].append(course_name) store.put((users, user_001), profile, {data: profile}) return ( f学习进度已更新已完成 {len(profile[completed_courses])} 门课程 f{, .join(profile[completed_courses])} ) # # 3. 构建最简图START - tools - END # class State(TypedDict): messages: list[AnyMessage] builder StateGraph(State) builder.add_node(tools, ToolNode([get_user_profile, save_course_progress])) builder.add_edge(START, tools) builder.add_edge(tools, END) # 关键store 在这里传入InjectedStore 才能生效 graph builder.compile(storestore) # # 4. 测试函数模拟一次工具调用 # def run_tool(tool_name: str, tool_args: dict) - str: 通过图执行一次工具调用让 InjectedStore 自动注入。 message AIMessage( content, tool_calls[{ name: tool_name, args: tool_args, id: call_1, }], ) result graph.invoke({messages: [message]}) # 返回工具输出的文本 return result[messages][0].content if __name__ __main__: print(run_tool(get_user_profile, {})) print(run_tool(save_course_progress, {course_name: Python3 基础教程})) print(run_tool(get_user_profile, {}))运行效果如下3.2 Agent 中结合 Store使用将 Store 传给 create_agent()Agent 中的所有工具都能通过 InjectedStore 访问它如下完整示例代码import os from typing import Annotated from dotenv import load_dotenv from langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.tools import tool, InjectedStore from langgraph.store.base import BaseStore from langgraph.store.memory import InMemoryStore # 从.env文件中加载环境变量 load_dotenv(overrideTrue) OPENROUTER_API_KEY os.getenv(OPENROUTER_API_KEY) OPENROUTER_API_BASE os.getenv(OPENROUTER_API_BASE) # 以init_chat_model为例 from langchain_openrouter import ChatOpenRouter model ChatOpenRouter( modeldeepseek/deepseek-v4-flash, api_keyOPENROUTER_API_KEY, base_urlOPENROUTER_API_BASE, ) # 创建 Store 并预置数据 store InMemoryStore() store.put((runoob, courses), catalog, { data: { Python3 基础教程: {price: 免费, duration: 20小时}, Python 数据分析: {price: 会员, duration: 30小时}, HTML 基础教程: {price: 免费, duration: 15小时}, } }) tool def query_course_price( course_name: str, store: Annotated[BaseStore, InjectedStore()], ) - str: 查询菜鸟教程 RUNOOB 中指定课程的价格信息。 Args: course_name: 课程名称 item store.get((runoob, courses), catalog) catalog item.value[data] if item else {} if course_name in catalog: info catalog[course_name] return f《{course_name}》- 价格{info[price]}学习时长{info[duration]} return f未找到课程《{course_name}》 agent create_agent( modelmodel, tools[query_course_price], storestore, # 将 Store 传入 Agent system_prompt你是小鸟教程的课程顾问。, ) result agent.invoke({ messages: [HumanMessage(contentPython3 基础教程和 Python 数据分析分别多少钱)] }) print(result[messages][-1].content)运行效果如下3.3 InjectedState vs InjectedStore 对比为了方便后续的深入学习研究下面对两者做一个对比维度InjectedStateInjectedStore生命周期对话结束即消失持久化存储典型用途读取消息历史、当前对话的中间结果用户偏好、学习进度、配置信息传入方式InjectedState自动注入InjectedStore()需要括号数据组织扁平字典命名空间 键的层级结构四、写在文末本文详细介绍了Langchain 中有关Tool 的一些常见的高级用法希望对看到的同学有用哦本篇到此结束感谢观看。