Haystack ToolInvoker 组件全解析从工具调用到函数执行的桥梁【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本篇技术指南聚焦 Haystack 2.18 版本中负责执行工具调用的核心组件ToolInvokerAPI 参考见 tool_components_api.md。它接收语言模型LLM产出的包含工具调用的ChatMessage解析后执行对应的Tool函数并将执行结果封装为 tool 角色的ChatMessage返回是 Agent、Function Calling 类应用中连接模型决策与真实动作的关键桥梁。读完本文你将掌握ToolInvoker的完整 API 签名、异常体系、同步/异步执行方式以及如何在 Pipeline 中将它与其他组件组合成可用的工具调用闭环。ToolInvoker 是什么ToolInvoker是 Haystack 中一个专门负责执行工具调用的组件。其核心职责可以概括为三个步骤接收处理一组包含工具调用的ChatMessage对象通常由 Chat Generator 产生即 LLM 输出中的 function/tool call调度执行根据ToolCall中的tool_name从初始化时注册的工具列表中查找对应Tool并调用其函数封装返回把每个工具的执行结果包装成带tool角色的ChatMessage内部为ToolCallResult返回给调用方继续注入对话上下文。同时它还负责与共享的State进行读写协作——这使工具不仅能接收用户与模型的输入还能读取/写入 Agent 运行时的共享状态。从源码结构看ToolInvoker位于组件层而它操作的两个核心数据类型分别定义在 chat_message.pyToolCall、ToolCallResult和 tool.pyTool中三者共同构成 Haystack 工具调用体系的最小闭环。快速上手单独使用 ToolInvokerAPI 参考文档给出了最简可用示例手动构造一个带工具调用的ChatMessage交给ToolInvoker执行。from haystack.dataclasses import ChatMessage, ToolCall from haystack.tools import Tool from haystack.components.tools import ToolInvoker # Tool definition def dummy_weather_function(city: str): return fThe weather in {city} is 20 degrees. parameters {type: object, properties: {city: {type: string}}, required: [city]} tool Tool(nameweather_tool, descriptionA tool to get the weather, functiondummy_weather_function, parametersparameters) # Usually, the ChatMessage with tool_calls is generated by a Language Model # Here, we create it manually for demonstration purposes tool_call ToolCall( tool_nameweather_tool, arguments{city: Berlin} ) message ChatMessage.from_assistant(tool_calls[tool_call]) # ToolInvoker initialization and run invoker ToolInvoker(tools[tool]) result invoker.run(messages[message]) print(result)运行输出如下可以看到tool_messages中每个ChatMessage携带一个ToolCallResult其中result是工具函数返回的字符串origin保留触发它的原始ToolCall { tool_messages: [ ChatMessage( _roleChatRole.TOOL: tool, _content[ ToolCallResult( resultThe weather in Berlin is 20 degrees., originToolCall( tool_nameweather_tool, arguments{city: Berlin}, idNone ) ) ], _meta{} ) ] }关于 Tool 数据类示例中的Tool是 Haystack 工具体系的基础数据类定义见 tool.py其核心字段包括name工具名LLM 据此发起调用ToolInvoker据此查找工具名称必须唯一description工具用途描述对 LLM 选择正确工具至关重要parametersJSON Schema 格式的参数定义LLM 依据它生成argumentsfunction/async_function实际执行体至少提供一个同步函数走function协程函数必须传入async_functionoutputs_to_string定义工具输出如何转成字符串支持source提取、handler转换函数、raw_result原样返回等配置inputs_from_state把 State 中的键映射为工具参数如{repository: repo}表示将 State 的repository传给工具的repo参数outputs_to_state把工具输出可选经过handler写回 State 的指定键。从 tool.py 的__post_init__校验逻辑可以看到几个硬性约束function不能是协程函数应放async_functionparameters必须是合法 JSON Schema用Draft202012Validator校验工具重名或outputs_to_state/inputs_from_state引用了不存在的输出/参数都会在构造时直接抛ValueError这有助于把配置错误提前暴露在初始化阶段。使用 Toolset 批量管理工具除了传入list[Tool]ToolInvoker还接受一个Toolset实例。Toolset见 toolset.py是相关工具的集合既可以把多个工具当作一个整体管理也可以作为动态工具加载如从 OpenAPI、MCP 服务器拉取工具的基类from haystack.dataclasses import ChatMessage, ToolCall from haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker # Tool definition def dummy_weather_function(city: str): return fThe weather in {city} is 20 degrees. parameters {type: object, properties: {city: {type: string}}, required: [city]} tool Tool(nameweather_tool, descriptionA tool to get the weather, functiondummy_weather_function, parametersparameters) # Create a Toolset toolset Toolset([tool]) # Usually, the ChatMessage with tool_calls is generated by a Language Model # Here, we create it manually for demonstration purposes tool_call ToolCall( tool_nameweather_tool, arguments{city: Berlin} ) message ChatMessage.from_assistant(tool_calls[tool_call]) # ToolInvoker initialization and run with Toolset invoker ToolInvoker(toolstoolset) result invoker.run(messages[message]) print(result)构造参数详解__init__ToolInvoker的完整构造签名如下def __init__(tools: Union[list[Tool], Toolset], raise_on_failure: bool True, convert_result_to_json_string: bool False, streaming_callback: Optional[StreamingCallbackT] None, *, enable_streaming_callback_passthrough: bool False, max_workers: int 4)各参数含义参数默认值说明tools必填可调用的工具列表或一个能解析工具的Toolset实例raise_on_failureTrue为True时工具未找到、调用失败、结果转换失败、State 合并失败都会抛出对应异常为False时返回errorTrue且result中携带错误描述的ChatMessage便于让 LLM 在循环中自我纠错convert_result_to_json_stringFalse为True时工具结果用json.dumps转字符串为False时用strstreaming_callbackNone用于发射工具结果的回调函数注意结果就绪后一次性发出并非实时增量流式输出enable_streaming_callback_passthroughFalse为True时把streaming_callback透传给支持它的工具要求工具的invoke方法签名中带streaming_callback参数使工具能把结果流式回传客户端max_workers4线程池执行器的最大工作线程数也即最大并发工具调用数异常如果未提供任何工具或存在重复工具名构造时抛出ValueError。run 与 run_async同步与异步执行同步runcomponent.output_types(tool_messageslist[ChatMessage], stateState) def run(messages: list[ChatMessage], state: Optional[State] None, streaming_callback: Optional[StreamingCallbackT] None, *, enable_streaming_callback_passthrough: Optional[bool] None, tools: Optional[Union[list[Tool], Toolset]] None) - dict[str, Any]参数要点messages包含工具调用的ChatMessage列表state工具要使用的运行时状态State类定义见 state.py用于在 Agent 与工具之间共享文档、上下文和中间结果streaming_callback/enable_streaming_callback_passthrough与构造参数同名语义一致如果传入None则沿用构造时的值tools本次运行临时使用的工具设置后覆盖构造时传入的工具适合动态切换工具集的场景该能力由 release note tool-invoker-tools-in-run 引入。返回字典键tool_messages对应一组带 tool 角色的ChatMessage每个对象包裹一次工具调用的结果同时输出state类型标注为State。异步run_asynccomponent.output_types(tool_messageslist[ChatMessage], stateState) async def run_async( messages: list[ChatMessage], state: Optional[State] None, streaming_callback: Optional[StreamingCallbackT] None, *, enable_streaming_callback_passthrough: Optional[bool] None, tools: Optional[Union[list[Tool], Toolset]] None) - dict[str, Any]与run的差异在于多个工具调用会被并发执行配合max_workers控制并发度适合工具数量多、单个工具耗时的场景。对于只提供同步function的工具异步路径会通过asyncio.to_thread将其派发到工作线程执行见 tool.py 的invoke_async实现如果工具提供了async_function则直接 await。异常体系run/run_async在raise_on_failureTrue时会抛出以下异常全部定义于tool_invoker模块ToolNotFoundException在可用工具列表中找不到目标工具ToolInvocationError工具调用本身失败StringConversionError工具结果转字符串失败ToolOutputMergeError把工具输出合并进 State 失败。其中ToolOutputMergeError还提供一个类方法from_exception(cls, tool_name: str, error: Exception)用于从任意异常构造带工具名的合并错误。整个异常家族的基类是ToolInvokerError。在 Pipeline 中集成完整工具调用闭环工具调用的完整流程通常不止ToolInvoker一个组件。官方组件文档 toolinvoker.mdx 给出了标准接线方式Chat Generator 产出回复 →ConditionalRouter判断回复中是否含工具调用 → 含则交给ToolInvoker执行 → 结果回填对话继续循环不含则作为最终回复输出。from haystack.dataclasses import ChatMessage from haystack.components.tools import ToolInvoker from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.routers import ConditionalRouter from haystack.tools import Tool from haystack import Pipeline from typing import List # Ensure List is imported ## Define a dummy weather tool import random def dummy_weather(location: str): return { temp: f{random.randint(-10, 40)} °C, humidity: f{random.randint(0, 100)}%, } weather_tool Tool( nameweather, descriptionA tool to get the weather, functiondummy_weather, parameters{ type: object, properties: {location: {type: string}}, required: [location], }, ) ## Initialize the ToolInvoker with the weather tool tool_invoker ToolInvoker(tools[weather_tool]) ## Initialize the ChatGenerator chat_generator OpenAIChatGenerator(modelgpt-4o-mini, tools[weather_tool]) ## Define routing conditions routes [ { condition: {{replies[0].tool_calls | length 0}}, output: {{replies}}, output_name: there_are_tool_calls, output_type: List[ChatMessage], # Use direct type }, { condition: {{replies[0].tool_calls | length 0}}, output: {{replies}}, output_name: final_replies, output_type: List[ChatMessage], # Use direct type }, ] ## Initialize the ConditionalRouter router ConditionalRouter(routes, unsafeTrue) ## Create the pipeline pipeline Pipeline() pipeline.add_component(generator, chat_generator) pipeline.add_component(router, router) pipeline.add_component(tool_invoker, tool_invoker) ## Connect components pipeline.connect(generator.replies, router) pipeline.connect( router.there_are_tool_calls, tool_invoker.messages, ) # Correct connection ## Example user message user_message ChatMessage.from_user(What is the weather in Berlin?) ## Run the pipeline result pipeline.run({messages: [user_message]}) ## Print the result print(result)典型输出温度/湿度为随机值{ tool_invoker:{ tool_messages:[ ChatMessage(_roleChatRole.TOOL:tool, _content[ ToolCallResult(result{temp: 33 °C, humidity: 79%}, originToolCall(tool_nameweather, arguments{ location:Berlin }, idcall_pUVl8Cycssk1dtgMWNT1T9eT), errorFalse) ], _nameNone, _meta{ }) ] } }需要说明的是循环把 tool 消息回灌给 Generator 继续推理在 Pipeline 场景中通常由外层 Agent 或用户自建循环驱动——这正是工具调用闭环的最后一环。该示例中ConditionalRouter的unsafeTrue用于允许 Jinja 模板求值属于该组件的既定用法。序列化to_dict 与 from_dict作为标准的 Haystack 组件ToolInvoker支持序列化to_dict() - dict[str, Any]将组件含工具列表序列化为字典便于保存到 YAML/JSON 或传输from_dict(cls, data: dict[str, Any]) - ToolInvoker类方法从字典反序列化重建组件实例。与之配套Tool本身也实现了to_dict/from_dict见 tool.py序列化时会通过serialize_callable把function、async_function以及outputs_to_state/outputs_to_string中的 handler 等可调用对象转换为可传输的字符串形式反序列化时再还原。版本适用性与演进提示本文 API 依据的是 2.18 版本参考文档 tool_components_api.md。需要特别注意从仓库 release note remove-tool-invoker-component 可以看到在后续大版本中ToolInvoker组件已被移除工具执行职责并入haystack.components.agents.Agent——Agent 现在直接持有工具/工具集的 warm-up、State 注入、流式回调透传与同步/异步调用原tool_invoker_kwargs参数也改由tool_concurrency_limit与tool_streaming_callback_passthrough承担工具结果统一用json.dumpsensure_asciiFalse仅在结果不可 JSON 序列化时回退str序列化取代了旧的convert_result_to_json_string开关。因此如果你正在使用 2.18 及相近版本并构建自定义 Pipeline本文的 API 直接可用如果面向新版本请把工具直接传给Agent本文讲解的参数语义并发控制、流式透传、State 读写、失败策略在 Agent 中以新参数名延续。延伸阅读组件使用文档toolinvoker.mdxTool数据类源码tool.pyToolset源码toolset.pyToolCall/ToolCallResult定义chat_message.pyState定义state.py相关演进记录add-tool-invoker、add-run-async-tool-invoker、enable-parallel-tool-calling、add-enable-streaming-passthrough【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考