Haystack Tools API 完全指南：用统一抽象为 LLM 应用构建可调用的工具层

Haystack Tools API 完全指南用统一抽象为 LLM 应用构建可调用的工具层【免费下载链接】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/haystackHaystack 的tools模块为整个框架提供了一套统一的工具抽象无论是普通 Python 函数、Haystack 组件还是外部服务OpenAPI、MCP都可以被包装成 LLM 可以直接发起调用的Tool。本文以 Haystack 2.18 版本的 Tools API 参考文档为主线结合 haystack/tools/ 下的真实源码与 test/tools/ 测试用例系统讲解Tool、create_tool_from_function、tool装饰器、ComponentTool与Toolset的定义、用法、序列化机制与底层实现帮助你在 Pipeline 与 Agent 中正确设计、组织并落地工具调用。整体架构Haystack 中的工具抽象层次在 Haystack 中工具Tool是Language Model 可以准备一次调用的实体name、description与parameters三个文本属性的准确性直接影响 LLM 生成调用参数的质量。整个工具体系由三层构成Tool最底层的数据类统一承载工具是什么、接受什么参数、调用什么函数这三个要素是函数型工具与组件型工具的公共底座haystack/tools/tool.py。工具工厂create_tool_from_function与tool装饰器把普通函数自动转成带 JSON Schema 参数的ToolComponentTool把 Haystack 组件包装成Toolhaystack/tools/from_function.py、haystack/tools/component_tool.py。Toolset工具的集合抽象既用于把相关工具作为一个整体交给Agent、ToolInvoker或 Chat Generator也作为动态加载外部工具的基类haystack/tools/toolset.py。从 haystack/tools/init.py 可以看出haystack.tools对外暴露的核心符号包括Tool、tool、create_tool_from_function、Toolset、ComponentTool、PipelineTool、AgentTool、SearchableToolset、SkillToolset以及ToolsType等其中Tool、tool、create_tool_from_function在包加载时即被急切导入其余符号通过LazyImporter惰性加载以优化导入开销。Tool数据类一切工具的公共契约Tool是一个 dataclass其字段定义于 haystack/tools/tool.py字段类型说明namestr工具名称LLM 用它来选择要调用的工具descriptionstr工具描述LLM 用它判断何时应该调用该工具parametersdict[str, Any]定义工具期望参数的 JSON SchemaLLM 按此生成调用参数functionCallable \| None工具被调用时执行的同步函数async_functionCallable \| None可选的协程函数供invoke_async使用outputs_to_stringdict \| None定义工具输出如何转换为字符串可带自定义 handlerinputs_from_statedict[str, str] \| None把 State 中的键映射到工具参数名如{repository: repo}outputs_to_statedict[str, dict] \| None定义工具输出如何写回 State可带自定义 handler构造期校验Tool的__post_init__haystack/tools/tool.py在实例化时执行一组严格校验从源码看包括function与async_function至少设置一个否则抛出ValueErrorfunction必须是普通同步函数协程函数必须放到async_function反之async_function必须是async def定义的协程函数parameters必须是合法的 JSON Schema使用Draft202012Validator.check_schema校验outputs_to_state中每个配置必须是字典source必须是字符串、handler必须可调用且source引用的输出键必须真实存在outputs_to_string支持单输出配置根级使用source/handler/raw_result与多输出配置每个键映射到各自的source/handler配置两种格式两种格式互斥多输出格式不支持raw_resultinputs_from_state中映射到的参数名必须存在于工具函数的签名或 schema 的属性中防止拼写错误在构造期就被发现。这些校验在test/tools/test_tool.py中有大量对应测试例如传入协程函数到function会抛错、非法 JSON Schema 会抛错等边界场景。核心方法tool_spec只读属性返回{name: ..., description: ..., parameters: ...}三元组这正是要交给 LLM 的工具规格haystack/tools/tool.py。invoke(**kwargs)同步调用工具。若工具只有async_function而function为None会抛出带tool_name的ToolInvocationError底层调用抛出的任何异常也会被包装成ToolInvocationError。invoke_async(**kwargs)异步调用。若设置了async_function则直接await否则通过asyncio.to_thread把同步function派发到工作线程执行。warm_up()预留的资源初始化钩子默认空实现。对于连接远程服务、加载模型等重量级操作子类应重写此方法且必须保持幂等可能被多次调用。to_dict()/from_dict()序列化与反序列化。序列化时使用asdict展开字段并通过serialize_callable/deserialize_callable转换函数与 handler最终包一层{type: ..., data: ...}结构from_dict按type字段反序列化出具体子类实例。从函数创建工具create_tool_from_function与toolcreate_tool_from_functioncreate_tool_from_function(function, nameNone, descriptionNone, inputs_from_stateNone, outputs_to_stateNone, outputs_to_stringNone)是函数型工具的主要工厂haystack/tools/from_function.py其工作流程从源码看分为四步解析签名用inspect.signature遍历函数参数构建 Pydantic 模型把每个带类型标注的参数作为字段通过create_model动态创建模型并生成 JSON Schema。无默认值的参数用...Ellipsis标记为必填清洗 Schema用_remove_title_from_schema递归删除 Pydantic 自动添加的title关键字它们是冗余信息LLM 不需要该函数对properties、$defs等以用户命名键为键的映射做了特殊处理确保名为title的参数不会被误删default、enum等数据关键字中的title也会被保留haystack/tools/from_function.py注入描述使用typing.Annotated的元数据作为参数描述写入 schema 的properties。官方文档给出的完整示例from typing import Annotated, Literal from haystack.tools import create_tool_from_function def get_weather( city: Annotated[str, the city for which to get the weather] Munich, unit: Annotated[Literal[Celsius, Fahrenheit], the unit for the temperature] Celsius): A simple function to get the current weather for a location. return fWeather report for {city}: 20 {unit}, sunny tool create_tool_from_function(get_weather) print(tool) Tool(nameget_weather, descriptionA simple function to get the current weather for a location., parameters{ type: object, properties: { city: {type: string, description: the city for which to get the weather, default: Munich}, unit: { type: string, enum: [Celsius, Fahrenheit], description: the unit for the temperature, default: Celsius, }, } }, functionfunction get_weather at 0x7f7b3a8a9b80)参数行为与限制官方文档 源码双重确认name缺省时取函数名description缺省时取函数 docstring要刻意留空描述传空字符串函数所有参数必须带类型标注否则抛出ValueError推荐使用基本 Python 类型str、int、float、bool、list、dict、tuple其他类型可能可用但不保证被inputs_from_state映射的参数、State类型参数、Callable类型参数会被从 schema 中排除State参数由 Agent 在运行时注入schema 生成失败时抛出SchemaGenerationError传入async def协程函数时函数会被自动放到async_function字段function置为None。以上行为在 test/tools/test_from_function.py 中有系统验证docstring 作为描述、自定义 name/description、空描述、缺失类型标注抛错、Annotated描述注入、协程函数自动归类等。tool装饰器tool装饰器是create_tool_from_function的简化封装haystack/tools/from_function.py支持带参数与不带参数两种用法tool # without parameters def my_function(): ... tool(namecustom_name) # with parameters def my_function(): ...官方文档的完整示例from typing import Annotated, Literal from haystack.tools import tool tool def get_weather( city: Annotated[str, the city for which to get the weather] Munich, unit: Annotated[Literal[Celsius, Fahrenheit], the unit for the temperature] Celsius): A simple function to get the current weather for a location. return fWeather report for {city}: 20 {unit}, sunny print(get_weather) Tool(nameget_weather, descriptionA simple function to get the current weather for a location., parameters{...})从源码实现看当function is None时tool(...)返回一个decorator否则直接调用decorator(function)两种调用方式统一走create_tool_from_function。tool的完整签名支持name、description、inputs_from_state、outputs_to_state、outputs_to_string五个可选参数。在 test/tools/test_toolset.py 中可以看到推荐的最佳实践——用tool装饰器定义带Annotated描述的工具再放入Toolset交给 Agenttool def weather(location: Annotated[str, the location to get the weather for]) - dict: Provides weather information for a given location. ...工具输出处理outputs_to_string、inputs_from_state与outputs_to_state这三个配置是函数工具与 Agent State 协作的关键官方文档给出的完整配置格式如下outputs_to_string工具输出转为字符串/结果两种格式# 单输出格式根级使用 source / handler / raw_result { source: docs, handler: format_documents, raw_result: False } # 多输出格式每个键映射到各自的 source/handler { formatted_docs: {source: docs, handler: format_documents}, summary: {source: summary_text, handler: str.upper} }source提供时只把指定输出键交给 handler省略时把整个工具结果交给 handlerhandler是一个接收工具输出或提取出的 source 值并返回最终结果的函数raw_resultTrue时结果不做字符串转换仍会应用 handler专用于返回图片等非文本内容的工具——此时工具函数或 handler 必须返回TextContent/ImageContent对象列表以兼容 Chat Generator多输出格式不支持raw_result。inputs_from_state把 State 键映射到工具参数名例如{repository: repo}表示把 State 中的repository值注入工具的repo参数。outputs_to_state定义工具输出如何写入 State。source提供时只写指定输出键省略时写整个工具结果# 带 source只取 docs 输出键交给 handler { documents: {source: docs, handler: custom_handler} } # 不带 source整个工具结果交给 handler { documents: {handler: custom_handler} }在序列化层面outputs_to_state与outputs_to_string中的handler会在to_dict/from_dict时通过_serialize_outputs_to_state、_deserialize_outputs_to_state、_serialize_outputs_to_string、_deserialize_outputs_to_string等辅助函数完成可调用对象与字符串表示之间的转换haystack/tools/tool.py确保工具可以被完整地序列化保存与还原。ComponentTool把 Haystack 组件变成 LLM 工具ComponentTool是Tool的子类用于把 Haystack 组件包装成 LLM 可直接调用的工具haystack/tools/component_tool.py。其核心特性官方文档明确列出从组件输入 socket自动生成 LLM 工具调用 schemaschema 来源于组件run方法的签名与类型标注对组件输入做类型转换与校验支持的数据类型dataclass、dataclass 列表、基本类型str、int、float、bool、dict及基本类型列表工具名自动由组件类名生成PascalCase 转 snake_case例如SerperDevWebSearch→serper_dev_web_search描述自动取自组件 docstring。构造与校验ComponentTool(component, nameNone, descriptionNone, parametersNone, *, outputs_to_stringNone, inputs_from_stateNone, outputs_to_stateNone)haystack/tools/component_tool.py在构造期做如下检查component必须是 HaystackComponent实例否则抛TypeError组件不能已经被加入某个 Pipeline__haystack_added_to_pipeline__标记为真时抛ValueErrorparameters缺省时由_create_tool_parameters_schema依据组件输入 socket 自动生成跳过被inputs_from_state映射的参数、Callable类型与State类型参数必填参数用...标记run方法 docstring 中的参数描述会被提取进 schemahaystack/tools/component_tool.py。调用与类型转换组件调用由内部闭包component_invoker及异步版async_component_invoker完成haystack/tools/component_tool.pyLLM 生成的 kwargs 会逐一经_convert_param转换后传给component.run()。_convert_param的类型转换逻辑haystack/tools/component_tool.py支持解开Optional类型对 union 类型提取其中的list[T]分支目标类型或其列表元素类型拥有from_dict方法时把字典/list 转成对应对象从而支持 dataclass 及 dataclass 列表输入其余情况回退到 PydanticTypeAdapter做类型校验。此外ComponentTool重写了_get_valid_inputs/_get_valid_outputs分别返回组件输入/输出 socket 名集合从而让inputs_from_state与outputs_to_state在构造期就能校验引用的参数/输出是否真实存在warm_up()会调用被包装组件的warm_up()带幂等保护。官方使用示例官方文档给出的完整示例将 SerperDev 网络搜索组件包装成工具并放入 Pipelinefrom haystack import component, Pipeline from haystack.tools import ComponentTool from haystack.components.websearch import SerperDevWebSearch from haystack.utils import Secret from haystack.components.tools.tool_invoker import ToolInvoker from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage # Create a SerperDev search component search SerperDevWebSearch(api_keySecret.from_env_var(SERPERDEV_API_KEY), top_k3) # Create a tool from the component tool ComponentTool( componentsearch, nameweb_search, # Optional: defaults to serper_dev_web_search descriptionSearch the web for current information on any topic # Optional: defaults to component docstring ) # Create pipeline with OpenAIChatGenerator and ToolInvoker pipeline Pipeline() pipeline.add_component(llm, OpenAIChatGenerator(modelgpt-4o-mini, tools[tool])) pipeline.add_component(tool_invoker, ToolInvoker(tools[tool])) # Connect components pipeline.connect(llm.replies, tool_invoker.messages) message ChatMessage.from_user(Use the web search tool to find information about Nikola Tesla) # Run pipeline result pipeline.run({llm: {messages: [message]}}) print(result)说明在仓库当前主线版本中ComponentTool的官方示例推荐把工具直接交给Agent(chat_generator..., tools[tool])使用见 haystack/tools/component_tool.py 中的示例两种方式都可行——前者以显式 Pipeline 串联 Chat Generator 与工具执行器后者由 Agent 内部编排工具调用循环。ComponentTool的序列化to_dict/from_dicthaystack/tools/component_tool.py会把被包装组件通过component_to_dict一并序列化parameters字段只保存用户显式传入的未解析 schema自动生成的 schema 在反序列化时由组件重建handler 同样经过可调用对象序列化转换。对应测试见 test/tools/test_component_tool.py其中覆盖了组件 schema 自动生成、dataclass 输入转换、组件校验、序列化往返等场景。Toolset工具集合与动态加载基类Toolset是一个 dataclass内部持有tools: list[Tool]haystack/tools/toolset.py官方文档明确其两大用途用途一把相关工具组织成一个整体官方文档的完整示例手写 Tool 实例 ToolsetToolInvokerfrom haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker # Define math functions def add_numbers(a: int, b: int) - int: return a b def subtract_numbers(a: int, b: int) - int: return a - b # Create tools with proper schemas add_tool Tool( nameadd, descriptionAdd two numbers, parameters{ type: object, properties: { a: {type: integer}, b: {type: integer} }, required: [a, b] }, functionadd_numbers ) subtract_tool Tool( namesubtract, descriptionSubtract b from a, parameters{ type: object, properties: { a: {type: integer}, b: {type: integer} }, required: [a, b] }, functionsubtract_numbers ) # Create a toolset with the math tools math_toolset Toolset([add_tool, subtract_tool]) # Use the toolset with a ToolInvoker or ChatGenerator component invoker ToolInvoker(toolsmath_toolset)用途二动态工具加载的基类通过子类化Toolset可以实现从 OpenAPI URL、MCP 服务器等外部来源动态加载工具。官方文档给出的子类示例from haystack.core.serialization import generate_qualified_class_name from haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker class CalculatorToolset(Toolset): A toolset for calculator operations. def __init__(self): tools self._create_tools() super().__init__(tools) def _create_tools(self): # These Tool instances are obviously defined statically and for illustration purposes only. # In a real-world scenario, you would dynamically load tools from an external source here. tools [] add_tool Tool( nameadd, descriptionAdd two numbers, parameters{ type: object, properties: {a: {type: integer}, b: {type: integer}}, required: [a, b], }, functionlambda a, b: a b, ) multiply_tool Tool( namemultiply, descriptionMultiply two numbers, parameters{ type: object, properties: {a: {type: integer}, b: {type: integer}}, required: [a, b], }, functionlambda a, b: a * b, ) tools.append(add_tool) tools.append(multiply_tool) return tools def to_dict(self): return { type: generate_qualified_class_name(type(self)), data: {}, # no data to serialize as we define the tools dynamically } classmethod def from_dict(cls, data): return cls() # Recreate the tools dynamically during deserialization # Create the dynamic toolset and use it with ToolInvoker calculator_toolset CalculatorToolset() invoker ToolInvoker(toolscalculator_toolset)从 haystack/tools/toolset.py 的当前源码看动态加载子类的推荐模式演进为在warm_up()中建立连接并把工具赋值给self.tools用if self._client is not None: return保证幂等序列化时只保存端点描述符如{endpoint: ...}而非工具实例本身——因为工具是动态解析出来的序列化实例既开销大又可能过期反序列化时通过描述符重建才最可靠。集合接口与操作Toolset实现了完整的集合协议官方文档确认__iter__返回工具迭代器使Toolset可被用在任何期望工具列表的地方Agent、Chat Generator 等__contains__支持按 Tool 实例或工具名字符串判断成员即tool in toolset与tool_name in toolset都可用__len__返回工具数量__getitem__按下标获取工具add(tool)追加工具重复工具名会抛ValueError非Tool对象抛TypeError__add__(other)与另一个Tool、Toolset或list[Tool]拼接产生新Toolset重复名抛ValueError__post_init__初始化时校验——直接传入单个Tool会抛TypeError必须用列表Toolset([tool])并检查初始工具集合是否有重名_check_duplicate_tool_names定义于 haystack/tools/tool.pywarm_up()默认遍历并预热所有工具子类可重写为建立共享连接、动态加载工具等注意保持幂等get_selectable_tools()/spawn()分别用于按名称选择工具如Agent.run(tools[tool_name])与为单次运行创建隔离副本支持带运行态的子类如SearchableToolset在并发运行时不互相污染。序列化to_dict()默认实现输出{type: ..., data: {tools: [每个工具的 to_dict]}}适用于工具静态解析的场景from_dict()按每个工具的type字段用import_class_by_name导入类并调用其from_dict同时校验导入类必须是Tool的子类动态加载子类的序列化策略官方文档强调应序列化端点描述符URL、服务器信息而非工具实例从而保留动态性、降低序列化开销并确保反序列化能基于最新描述符准确重建工具避免加载过期或错误的工具配置。错误处理与边界行为工具体系涉及的异常定义于 haystack/tools/errors.pySchemaGenerationError自动生成 JSON Schema 失败时抛出如函数参数类型无法被 Pydantic 建模ToolInvocationError工具调用失败时抛出携带tool_name属性标识是哪个工具调用失败便于在 Agent/Pipeline 中定位问题。典型边界行为汇总场景行为依据function与async_function均为空构造时抛ValueErrorhaystack/tools/tool.py协程函数传入function构造时抛ValueError提示改用async_functionhaystack/tools/tool.pyparameters不是合法 JSON Schema构造时抛ValueErrorhaystack/tools/tool.py函数参数缺类型标注create_tool_from_function抛ValueErrorhaystack/tools/from_function.pySchema 生成失败抛SchemaGenerationErrorhaystack/tools/from_function.py工具只有async_function却同步调用invoke抛ToolInvocationErrorhaystack/tools/tool.py底层函数调用抛异常包装为ToolInvocationErrorhaystack/tools/tool.pyComponentTool传入非组件对象构造时抛TypeErrorhaystack/tools/component_tool.py组件已被加入 Pipeline构造时抛ValueErrorhaystack/tools/component_tool.pyToolset直接传入单个 Tool构造时抛TypeError提示用列表haystack/tools/toolset.pyToolset 中出现重复工具名抛ValueError构造/add/__add__均校验haystack/tools/toolset.py实践建议如何选择工具构建方式综合官方文档与仓库源码可以按以下原则选择合适的工具构建方式简单函数 →tool装饰器推荐方式代码最简洁所有参数务必带类型标注用Annotated提供参数描述用 docstring 提供工具描述。需要自定义 name/description 或在构造期动态创建 →create_tool_from_function与tool底层等价适合在工厂函数中批量生成工具。复用已有 Haystack 组件 →ComponentTool自动从组件输入 socket 生成 schema天然支持 dataclass 输入与类型转换还自动获得组件级warm_up与序列化能力。注意组件不能已加入 Pipeline。组织多个工具或对接外部服务 →Toolset把相关工具作为一个整体交给Agent/ToolInvoker/ Chat Generator从 MCP、OpenAPI 等外部来源动态加载工具时子类化Toolset并在warm_up()中加载、序列化端点描述符。工具与 Agent State 协作通过inputs_from_state注入 State 值与outputs_to_state把输出写回 State可带 handler打通工具调用与 Agent 记忆/上下文管理。异步场景为工具提供async def协程函数放入async_function即可在Agent.run_async等异步执行路径中被直接await纯同步函数在异步调用时会自动通过asyncio.to_thread派发到工作线程。相关资源本文依据的 API 参考文档docs-website/reference_versioned_docs/version-2.18/haystack-api/tools_api.md核心实现haystack/tools/tool.py、haystack/tools/from_function.py、haystack/tools/component_tool.py、haystack/tools/toolset.py、haystack/tools/errors.py包导出与工具类型haystack/tools/init.py、haystack/tools/tool_types.py测试用例test/tools/test_tool.py、test/tools/test_from_function.py、test/tools/test_component_tool.py、test/tools/test_toolset.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），仅供参考