Haystack ToolInvoker 组件全解析：从工具调用到函数执行的桥梁-创锋一号

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 版本中负责执行工具调用的核心组件ToolInvoker（API 参考见 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.py（ToolCall、ToolCallResult）和 tool.py（Tool）中，三者共同构成 Haystack 工具调用体系的最小闭环。

快速上手：单独使用 ToolInvoker

API 参考文档给出了最简可用示例：手动构造一个带工具调用的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 f"The weather in {city} is 20 degrees." parameters = {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]} tool = Tool(name="weather_tool", description="A tool to get the weather", function=dummy_weather_function, parameters=parameters) # Usually, the ChatMessage with tool_calls is generated by a Language Model # Here, we create it manually for demonstration purposes tool_call = ToolCall( tool_name="weather_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( >> _role=<ChatRole.TOOL: 'tool'>, >> _content=[ >> ToolCallResult( >> result='"The weather in Berlin is 20 degrees."', >> origin=ToolCall( >> tool_name='weather_tool', >> arguments={'city': 'Berlin'}, >> id=None >> ) >> ) >> ], >> _meta={} >> ) >> ] >> }

关于 Tool 数据类

示例中的Tool是 Haystack 工具体系的基础数据类（定义见 tool.py），其核心字段包括：

name：工具名，LLM 据此发起调用，ToolInvoker据此查找工具，名称必须唯一；
description：工具用途描述，对 LLM 选择正确工具至关重要；
parameters：JSON Schema 格式的参数定义，LLM 依据它生成arguments；
function/async_function：实际执行体，至少提供一个；同步函数走function，协程函数必须传入async_function；
outputs_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_function）；parameters必须是合法 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 f"The weather in {city} is 20 degrees." parameters = {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]} tool = Tool(name="weather_tool", description="A tool to get the weather", function=dummy_weather_function, parameters=parameters) # 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_name="weather_tool", arguments={"city": "Berlin"} ) message = ChatMessage.from_assistant(tool_calls=[tool_call]) # ToolInvoker initialization and run with Toolset invoker = ToolInvoker(tools=toolset) 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_failure`	`True`	为`True`时，工具未找到、调用失败、结果转换失败、State 合并失败都会抛出对应异常；为`False`时返回`error=True`且`result`中携带错误描述的`ChatMessage`，便于让 LLM 在循环中自我纠错
`convert_result_to_json_string`	`False`	为`True`时工具结果用`json.dumps`转字符串，为`False`时用`str`
`streaming_callback`	`None`	用于发射工具结果的回调函数；注意结果就绪后一次性发出，并非实时增量流式输出
`enable_streaming_callback_passthrough`	`False`	为`True`时把`streaming_callback`透传给支持它的工具（要求工具的`invoke`方法签名中带`streaming_callback`参数），使工具能把结果流式回传客户端
`max_workers`	`4`	线程池执行器的最大工作线程数，也即最大并发工具调用数

异常：如果未提供任何工具，或存在重复工具名，构造时抛出ValueError。

run 与 run_async：同步与异步执行

同步`run`

@component.output_types(tool_messages=list[ChatMessage], state=State) 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_async`

@component.output_types(tool_messages=list[ChatMessage], state=State) 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_failure=True时会抛出以下异常（全部定义于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( name="weather", description="A tool to get the weather", function=dummy_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(model="gpt-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, unsafe=True) ## 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(_role=<ChatRole.TOOL":"tool"">", "_content="[ "ToolCallResult(result=""{'temp': '33 °C', 'humidity': '79%'}", "origin=ToolCall(tool_name=""weather", "arguments="{ "location":"Berlin" }, "id=""call_pUVl8Cycssk1dtgMWNT1T9eT"")", "error=False)" ], "_name=None", "_meta="{ }")" ] } }

需要说明的是，循环（把 tool 消息回灌给 Generator 继续推理）在 Pipeline 场景中通常由外层 Agent 或用户自建循环驱动——这正是"工具调用闭环"的最后一环。该示例中ConditionalRouter的unsafe=True用于允许 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.dumps（ensure_ascii=False，仅在结果不可 JSON 序列化时回退str）序列化，取代了旧的convert_result_to_json_string开关。

因此：如果你正在使用 2.18 及相近版本并构建自定义 Pipeline，本文的 API 直接可用；如果面向新版本，请把工具直接传给Agent，本文讲解的参数语义（并发控制、流式透传、State 读写、失败策略）在 Agent 中以新参数名延续。

企业官网建设流程全解析