Haystack ToolInvoker 组件完全指南从 Tool Calling 到 Agent 工具执行【免费下载链接】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.22 版本的ToolInvoker组件 API 参考文档系统讲解 LLM 应用中模型决策、组件执行这一 Tool Calling 闭环的核心执行环节ToolInvoker接收包含 Tool Call 的ChatMessage解析并调用对应的工具再把结果包装成带tool角色的消息返回给对话循环。读者将掌握ToolInvoker的初始化参数、同步/异步运行机制、异常体系、状态读写State与序列化方式并了解它在 Haystack 3.x 中如何演进为Agent的内置能力。版本说明文中 API 签名与示例出自 docs-website/reference_versioned_docs/version-2.22/haystack-api/tool_components_api.md对应的运行时代码为 Haystack 2.22.x本文同时结合当前仓库3.2.0-rc0源码与 MIGRATION.md 说明其后续演进请读者按自身安装的版本对照使用。一、ToolInvoker 是什么Tool Calling 流程中的执行器在基于 Haystack 构建的 Agent / RAG 应用中大语言模型本身无法直接访问外部系统它只能准备调用prepare a call真正的执行由组件完成。ToolInvoker就是这个执行环节的标准组件输入一组包含tool_calls的ChatMessage通常由 Chat Generator 依据工具描述生成处理根据tool_name在初始化时注册的工具列表中查找对应Tool将arguments作为参数调用其函数输出一组ChatMessage每条消息的 role 为tool内部包裹一个ToolCallResult记录执行结果与发起调用的原始ToolCall。从源码看这种请求-响应的数据结构定义在 haystack/dataclasses/chat_message.pyToolCall第 76 行起记录tool_name、arguments、可选的id与extra字段to_dict/from_dict支持序列化ToolCallResult第 118 行起由result、origin产生该结果的ToolCall与error是否失败三个字段组成result可以是字符串也可以是TextContent/ImageContent/FileContent序列。1.1 在 Pipeline 中的典型位置ToolInvoker通常与 Chat Generator 串联Generator 把用户消息与工具 schema 一起交给模型模型返回 assistant 消息内含 tool calls随后ToolInvoker消费这批消息并产出工具结果消息这些结果再拼回对话历史供模型生成最终回复。即用户消息 → ChatGenerator(带 tools) → assistant 消息(含 tool_calls) → ToolInvoker → tool 消息 → 下一轮对话这一编排在 Haystack 3.x 中已整体内化为Agent详见文末迁移章节但在 2.22 中它仍是可独立使用的 Pipeline 组件。二、快速上手最小可用示例2.1 手动构造 Tool 与 ToolCall不依赖 LLM参考文档给出的第一个示例不依赖任何模型直接手工构造工具与调用便于单独验证ToolInvoker行为from haystack.dataclasses import ChatMessage, ToolCall from haystack.tools import Tool from haystack.components.tools import ToolInvoker # Tool 定义普通 Python 函数 JSON Schema 参数声明 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) # 通常带 tool_calls 的 ChatMessage 由语言模型生成 # 这里为了演示而手动创建 tool_call ToolCall( tool_nameweather_tool, arguments{city: Berlin} ) message ChatMessage.from_assistant(tool_calls[tool_call]) # ToolInvoker 初始化与运行 invoker ToolInvoker(tools[tool]) result invoker.run(messages[message]) print(result)输出结果中tool_messages列表内每条ChatMessage的 role 为tool内容为一个ToolCallResult { 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的function参数接收同步函数从 haystack/tools/tool.py 的Tool数据类定义第 20 行起可见协程函数应传入async_function且name/description/parameters等文本属性必须准确因为模型正是依据它们来构造调用。parameters使用 JSON Schema 描述工具入参示例中required指定city为必填。2.2 使用 Toolset 批量组织工具当工具数量较多时可以用Toolset统一管理再把整个Toolset交给ToolInvokerfrom haystack.dataclasses import ChatMessage, ToolCall from haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker # Tool 定义同上 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) # 创建 Toolset toolset Toolset([tool]) # 手动构造带 tool_calls 的消息 tool_call ToolCall( tool_nameweather_tool, arguments{city: Berlin} ) message ChatMessage.from_assistant(tool_calls[tool_call]) # 用 Toolset 初始化 ToolInvoker invoker ToolInvoker(toolstoolset) result invoker.run(messages[message]) print(result)从ToolsType类型别名haystack/tools/tool_types.py可以看出tools参数接受Sequence[Tool | Toolset] | Toolset即Tool 与 Toolset 的混合列表或单个 Toolset两种形态都可以被解析。三、构造参数详解控制错误、结果格式与并发ToolInvoker.__init__的完整签名如下def __init__(tools: ToolsType, raise_on_failure: bool True, convert_result_to_json_string: bool False, streaming_callback: StreamingCallbackT | None None, *, enable_streaming_callback_passthrough: bool False, max_workers: int 4)参数类型默认值作用toolsToolsType必填Tool / Toolset 列表或单个 Toolset运行期按此解析工具raise_on_failureboolTrue为True时遇错抛异常为False时返回errorTrue的ChatMessage错误描述写入resultconvert_result_to_json_stringboolFalse为True时用json.dumps把工具结果转成字符串为False时用strstreaming_callbackStreamingCallbackT | NoneNone用于向外发射工具结果的回调结果就绪后一次性发出并非实时增量流式enable_streaming_callback_passthroughboolFalse为True时把streaming_callback透传给支持它的工具要求工具的invoke方法签名含streaming_callback参数使工具能把结果流回客户端max_workersint4线程池执行器的最大工作线程数同时决定最大并发工具调用数注意以上参数中streaming_callback、enable_streaming_callback_passthrough、max_workers之后使用了*表示其为关键字专用参数keyword-only调用时必须以关键字形式传入。初始化时若tools为空、或出现重复的工具名会抛出ValueError。3.1 三个关键参数的行为细节raise_on_failureFalse的容错模式适合在 Pipeline 中不希望一次工具失败就中断整体流程的场景。错误不会抛出而是以ToolCallResult(errorTrue)的形式随tool_messages返回由上层逻辑自行判断。结果字符串化策略工具函数返回的非字符串对象默认经str转成字符串再放回对话开启convert_result_to_json_stringTrue后改用json.dumps更适合保留结构化数据如 dict/list的原始形态方便模型理解。顺带一提这一选项在 3.x 中被移除非字符串结果统一改用json.dumps序列化见 MIGRATION.md 第 289 行附近的说明。max_workers与并发ToolInvoker内部基于线程池执行器并发调用工具。若一次消息中包含多条 tool calls它们会按max_workers限制并行执行在run_async路径下多个工具调用同样并发执行。四、run 与 run_async同步/异步两种执行路径4.1run同步component.output_types(tool_messageslist[ChatMessage], stateState) def run(messages: list[ChatMessage], state: State | None None, streaming_callback: StreamingCallbackT | None None, *, enable_streaming_callback_passthrough: bool | None None, tools: ToolsType | None None) - dict[str, Any]参数说明messages包含 tool calls 的ChatMessage列表逐条处理其中的ToolCallstate运行时状态State工具可以通过它与共享状态读写交互streaming_callback运行时覆盖构造函数中的同名回调行为一致结果就绪后一次性发射enable_streaming_callback_passthrough运行时覆盖项传None时回退到构造函数中设置的值tools运行时覆盖初始化时提供的工具集合适合动态工具场景如每次运行临时增删工具。返回值为字典键tool_messages对应一个ChatMessage列表每个元素是 role 为tool的消息包裹一次工具调用的结果。4.2run_async异步component.output_types(tool_messageslist[ChatMessage], stateState) async def run_async(messages: list[ChatMessage], state: State | None None, streaming_callback: StreamingCallbackT | None None, *, enable_streaming_callback_passthrough: bool | None None, tools: ToolsType | None None) - dict[str, Any]与run唯一的签名差异是streaming_callback必须是异步回调函数。文档明确说明异步模式下多个工具调用会被并发执行Multiple tool calls are performed concurrently。对于 I/O 密集的工具网络请求、文件读取等run_async配合并发能显著缩短整体耗时。4.3 异常体系谁在什么时候抛出异常类触发条件说明ToolInvokerError所有 ToolInvoker 错误的基类见 tool_components_api.md 第 14 行ToolNotFoundException在可用工具列表中找不到被调用的工具tool_name与注册名不匹配时触发StringConversionError工具结果转字符串失败与convert_result_to_json_string配置相关ToolOutputMergeError把工具输出合并进共享状态State失败提供from_exception(tool_name, error)类方法可从任意异常构造run/run_async在raise_on_failureTrue时分别抛出以上异常在raise_on_failureFalse时不抛异常而是返回errorTrue的ChatMessage。五、生命周期与序列化5.1warm_updef warm_up()在正式运行前预热工具执行器会逐个调用已注册工具的warm_up。文档强调该方法幂等——多次调用只会预热一次。这正对应Tool数据类注释中的设计haystack/tools/tool.py对于建立远程连接、加载模型等资源密集型操作应重写warm_up()该方法在 Pipeline / Agent 装配阶段可能被多次调用必须保证可重复执行。5.2to_dict与from_dictto_dict()把组件序列化为字典便于保存到 YAML/JSON 或通过Pipeline.dumps持久化from_dict(data)类方法从字典反序列化重建组件。这两者共同保证了包含ToolInvoker的 Pipeline 可以完整地序列化与还原。六、实战模式与 Chat Generator 组成 Agent 闭环把前面各部分串起来一个典型的 Tool Calling Pipeline 如下from haystack import Pipeline from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.tools import ToolInvoker from haystack.dataclasses import ChatMessage from haystack.tools import Tool def get_weather(city: str) - str: return fThe weather in {city} is 20 degrees. weather_tool Tool( nameget_weather, descriptionGet the current weather for a city, functionget_weather, parameters{ type: object, properties: {city: {type: string}}, required: [city], }, ) chat_generator OpenAIChatGenerator(modelgpt-4o-mini, tools[weather_tool]) tool_invoker ToolInvoker(tools[weather_tool], convert_result_to_json_stringTrue) pipeline Pipeline() pipeline.add_component(llm, chat_generator) pipeline.add_component(tools, tool_invoker) pipeline.connect(llm.replies, tools.messages) result pipeline.run({ llm: {messages: [ChatMessage.from_user(Whats the weather in Berlin?)]}, })流程llm依据工具 schema 生成 assistant 消息含 tool calls→ 消息经tools.messages输入ToolInvoker→ 返回tool_messages给后续节点拼回对话历史。若模型认为无需调用工具assistant 消息不含 tool callsToolInvoker直接透传即可。七、演进路线3.x 中 ToolInvoker 与 Agent 的关系当前仓库主干版本为 3.2.0-rc0ToolInvoker组件已在 3.0 中被移除见 MIGRATION.md 第 209 行起的 ToolInvoker component removed 章节为什么移除工具执行职责收归Agent工具调用循环、状态处理、流式回调透传、预热与同步/异步执行统一由Agent管理避免分散在独立组件中。如何迁移不再把 Chat Generator 接到ToolInvoker而是把工具直接传给Agentfrom haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.tools import tool tool def weather(city: str) - str: Get the weather for a city. return fThe weather in {city} is sunny. agent Agent(chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[weather]) result agent.run(messages[ChatMessage.from_user(What is the weather in Berlin?)])参数平移原ToolInvoker的max_workers变为Agent顶层的tool_concurrency_limit默认 4要求 ≥ 1enable_streaming_callback_passthrough变为tool_streaming_callback_passthrough默认False。这两项参数在 haystack/components/agents/agent.py 第 393-394 行可见且Agent在构造时会校验tool_concurrency_limit 1。结果序列化策略调整convert_result_to_json_string选项被移除非字符串工具结果统一使用json.dumps序列化。因此如果你正在使用 Haystack 2.22.xToolInvoker是编排工具执行的标准组件本文第二节到第五节的内容可直接套用如果你已升级到 3.x则应使用Agent内置的工具执行能力工具定义Tool/Toolset与ToolCall/ToolCallResult数据结构的核心概念仍然一致。参考路径速查API 参考原文docs-website/reference_versioned_docs/version-2.22/haystack-api/tool_components_api.md工具与数据结构源码haystack/tools/tool.py、haystack/tools/toolset.py、haystack/tools/tool_types.py、haystack/dataclasses/chat_message.py3.x 迁移说明MIGRATION.mdToolInvoker component removed 章节3.x Agent 内建工具执行haystack/components/agents/agent.py【免费下载链接】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），仅供参考