目录

一、前言

二、Langchain 工具常用高级特性

2.1 return_direct

2.1.1 案例代码

2.2.2 return_direct 参数说明

2.2 InjectedToolCallId 使用

2.2.1 案例代码

2.3 ToolException

2.3.1 案例代码

2.4 handle_tool_error

2.4.1 示例代码

2.4.2 参数补充说明

2.4.3 带Agent 的错误处理

三、InjectedState 与 InjectedStore使用

3.1 InjectedState 使用

3.1.1 案例代码

3.1.2 Agent 中使用 InjectedState

3.2 Agent 中结合 Store使用

3.3 InjectedState vs InjectedStore 对比

四、写在文末

一、前言

在之前的文章中，我们分享了Langchain 中工具（Tool）的用法，工具在整个Langchain 体系中都占据着非常重要的作用，可以说是连接大模型与应用系统很重要的结合点，本篇将继续深入探讨一些关于Langchain 工具的高级特性。

二、Langchain 工具常用高级特性

接下来将介绍工具的高级特性：return_direct、InjectedToolCallId、ToolException 和错误处理。

2.1 return_direct

顾名思义，直接返回最终结果，默认情况下，工具执行后结果会返回给模型，模型再基于工具结果生成最终回复。但有时工具结果本身就是你想要的最终答案。

设置return_direct=True后，工具执行完就立即结束 Agent 循环，工具返回内容直接作为最终输出。

2.1.1 案例代码

在下面的案例代码中，使用@tool装饰器定义了两个工具方法，其中一个设置return_direct=True

from 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(override=True) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY") OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE") model = ChatOpenRouter( model="deepseek/deepseek-v4-flash", api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_API_BASE, ) # 普通工具：结果返回给模型，模型再做总结 @tool def search_normal(keyword: str) -> str: """ 搜索小米教程的课程（普通模式） """ return f"搜索结果：Python3 基础教程、Python 数据分析、Python 爬虫入门" # return_direct 工具：结果直接作为最终输出 @tool(return_direct=True) def search_direct(keyword: str) -> str: """ 搜索小米的课程（直接返回模式）。 当用户只需要搜索结果，不需要额外分析时使用此工具。 """ return f"搜索结果：Python3 基础教程、Python 数据分析、Python 爬虫入门" # 对比：普通模式 vs 直接返回模式 agent_normal = create_agent( model=model, tools=[search_normal], system_prompt="你是小米教程的学习顾问。", ) agent_direct = create_agent( model=model, 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_direct=True 之后，工具中直接将查到的结果返回给用户，而不走大模型的二次整理

2.2.2return_direct参数说明

下面对return_direct=True 参数设置做补充说明

当设置 return_direct=True 时，Agent 会跳过后续的模型思考步骤，直接返回工具自身的结果。这在节省 Token 和降低时延方面非常有价值，但也意味着模型不会对工具结果做任何二次加工。

注意：

如果一个 Agent 同时挂载了多个工具，其中不仅有 return_direct=True 的工具，也有普通工具，只要模型在这一轮调用中触发了任意一个 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_error=True：出错时不再抛异常， # 而是把 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_error=True 时，错误信息会作为正常结果返回，而不是抛出异常 result = get_weather.invoke({"city": "北境"}) print(f"handle_tool_error=True: {result}") # Agent 复用上面已启用 handle_tool_error 的 get_weather 工具。 # 该配置仅处理此工具抛出的 ToolException，不是整个 Agent 的全局错误兜底。 load_dotenv(override=True) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY") OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE") model = ChatOpenRouter( model="deepseek/deepseek-v4-flash", api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_API_BASE, ) agent = create_agent( model=model, 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 参数补充说明

注意：

不要用tool.with_config(handle_tool_errors=True)这种写法 ——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(override=True) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY") OPENROUTER_API_BASE = os.getenv("OPENROUTER_API_BASE") model = ChatOpenRouter( model="deepseek/deepseek-v4-flash", api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_API_BASE, ) agent = create_agent( model=model, 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_error=True后，ToolException 的内容会以一条 ToolMessage 的形式正常出现在对话历史里（上面输出中的 [tool] 那一行），而不会让程序崩溃；

• 模型看到这条错误信息后，会用自然语言把它转达给用户，并给出可用的替代方案。这正是 return_direct、InjectedToolCallId、ToolException 和 handle_tool_error 组合使用时的典型效果：既保证了工具调用的健壮性，又不牺牲用户体验。

三、InjectedState 与 InjectedStore使用

有时候，工具需要访问更多上下文信息，比如当前对话的状态、用户的持久化数据等。

LangChain 通过依赖注入机制，让工具函数能够自动获取这些信息。

这里涉及到两个重要的组件，InjectedState 与 InjectedStore，下面分别说明。

3.1 InjectedState 使用

默认情况下，工具只能通过参数接收模型传来的数据。但有时工具需要知道当前对话的上下文 —— 比如之前的对话历史、用户已确认的信息等。

InjectedState让工具可以直接读取 Agent 的完整状态。

3.1.1 案例代码

如下完整示例代码，在自定义工具方法参数中，使用InjectedState 修饰入参 state

from 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(store=store) # ============================================================ # 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(override=True) 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( model="deepseek/deepseek-v4-flash", api_key=OPENROUTER_API_KEY, base_url=OPENROUTER_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( model=model, tools=[query_course_price], store=store, # 将 Store 传入 Agent system_prompt="你是小鸟教程的课程顾问。", ) result = agent.invoke({ "messages": [HumanMessage(content="Python3 基础教程和 Python 数据分析分别多少钱？")] }) print(result["messages"][-1].content)

运行效果如下：

3.3 InjectedState vs InjectedStore 对比

为了方便后续的深入学习研究，下面对两者做一个对比

四、写在文末

本文详细介绍了Langchain 中有关Tool 的一些常见的高级用法，希望对看到的同学有用哦，本篇到此结束，感谢观看。