尧图网页设计 
Haystack Tools 统一抽象实战用 Tool / ComponentTool / PipelineTool / Toolset 为 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 模块提供了一套统一抽象让 LLM 可以准备一次函数调用再由框架侧真正执行无论是普通 Python 函数、Haystack 组件、完整流水线还是从 MCP/OpenAPI 动态加载的外部能力都能以一致的Tool形态接入 Agent 与 Chat Generator。读完本文你将掌握Tool、ComponentTool、PipelineTool、Toolset四大核心 API 的用法、参数语义与序列化机制并能把现有组件与流水线直接封装为可供 LLM 调用的工具。Tools 在 Haystack 中的作用与核心接口在 Haystack 中一个Tool是Language Models can prepare a call for的载体LLM 根据工具的name、description与parametersJSON Schema生成一次调用请求框架随后通过invoke()真正执行并返回结果。因此文本属性的准确程度直接决定 LLM 能否正确发起调用。Tool的核心接口由 haystack/tools/tool.py 定义tool_spec返回{name, description, parameters}字典即提交给 LLM 的完整工具规格warm_up()用于建立远程连接、加载模型等资源密集型初始化。该方法必须幂等因为在流水线/Agent 组装阶段可能被多次调用invoke(**kwargs)用关键字参数同步调用工具底层函数to_dict()/from_dict()序列化与反序列化保证工具可以随流水线一起持久化。从源码结构看Tool是一个dataclass其__post_init__haystack/tools/tool.py#L111-L212会执行一系列防御性校验function与async_function至少提供一个、同步/协程类型不能放错位置、parameters必须是合法 JSON Schema、outputs_to_state/outputs_to_string/inputs_from_state的结构与引用必须正确。这意味着大部分配置错误会在工具构造阶段就被拦截而不是等到运行时才暴露。除同步调用外Tool还提供invoke_async()优先等待async_function否则通过asyncio.to_thread在 worker 线程中执行同步function。对应测试见 test/tools/test_tool.py如test_invoke_async_falls_back_to_sync_function、test_invoke_on_async_only_tool_raises。Tool 数据类字段与参数语义Tool的字段定义haystack/tools/tool.py#L102-L109如下字段类型说明namestr工具名称LLM 用它选择工具descriptionstr工具描述LLM 据此判断何时使用parametersdict期望参数的 JSON SchemafunctionCallable \| None同步调用函数协程函数必须放入async_functionasync_functionCallable \| None可选的协程函数供invoke_async()使用outputs_to_stringdict \| None工具输出到字符串的转换配置inputs_from_statedict \| None从 Agent State 注入的输入映射outputs_to_statedict \| None工具输出写回 Agent State 的映射输出转换outputs_to_string 的两种格式outputs_to_string支持两种配置格式单输出格式——在根级使用source、handler、raw_result{source: docs, handler: format_documents, raw_result: False}source指定只把某个输出键交给handler不提供则把整个工具结果交给handlerhandler接收工具输出或提取出的source值并返回最终结果的可调用对象raw_result为True时结果不做字符串转换直接返回仍会应用handler。该模式专为返回图片等富内容的工具设计此时工具函数或handler必须返回TextContent/ImageContent对象列表才能与 Chat Generator 兼容。多输出格式——把键映射到各自独立的配置{ formatted_docs: {source: docs, handler: format_documents}, summary: {source: summary_text, handler: str.upper} }每个键对应一个可含source与/或handler的字典。注意raw_result不支持多输出格式此限制在Tool.__post_init__中会被强制校验haystack/tools/tool.py#L181-L188。与 Agent State 联动inputs_from_state 与 outputs_to_stateinputs_from_state: {repository: repo}表示把 State 中键repository的值注入工具参数repo。该映射在构造时会通过_get_valid_inputs()校验参数名真实存在防止拼写错误对应测试test_inputs_from_state_validation_with_invalid_parameteroutputs_to_state定义工具输出如何写回 State提供source时只把指定输出键交给handler再入 State省略source时整个工具结果交给handler。# 指定输出键 处理器 {documents: {source: docs, handler: custom_handler}} # 省略 source整个结果交给 handler {documents: {handler: custom_handler}}对于函数型工具_get_valid_outputs()默认返回None跳过输出校验而ComponentTool会重写该方法返回组件输出 socket 名从而对outputs_to_state的source做严格校验haystack/tools/component_tool.py#L265-L274。从函数创建 Toolcreate_tool_from_function 与 tool 装饰器create_tool_from_functionhaystack/tools/from_function.py#L18-L193把任意带类型标注的函数转换为Tool是无组件、纯函数场景的入口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)关键行为与约束类型提示是硬性要求所有参数必须带类型标注否则抛出ValueError。推荐使用typing.Annotated提供参数描述Annotated[str, 描述]其元数据会成为 JSON Schema 中的description支持的基础类型str、int、float、bool、list、dict、tuple等基本 Python 类型其他类型可能可用但不保证schema 生成机制内部通过 Pydanticcreate_model构建模型再调用model_json_schema()无默认值的参数以...标记为必填haystack/tools/from_function.py#L141-L170随后_remove_title_from_schema会剔除 Pydantic 自动添加的冗余title关键字名称与描述回退未传name时用函数名未传description时用函数 docstring想刻意留空则传空字符串协程支持async def函数会自动放入结果的async_function字段异常类型参数缺类型标注抛ValueErrorschema 生成失败抛SchemaGenerationError。更简洁的 tool 装饰器对于简单场景推荐直接使用tool装饰器haystack/tools/from_function.py#L220-L336。它可以带参数或不带参数使用tool # 不带参数 def my_function(): ... tool(namecustom_name) # 带参数 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) # 输出结果与 create_tool_from_function 完全相同tool与create_tool_from_function支持相同的name、description、inputs_from_state、outputs_to_state、outputs_to_string参数两者本质是同一套转换逻辑的两种调用方式。ComponentTool把 Haystack 组件封装成工具ComponentToolhaystack/tools/component_tool.py让任意 Haystack 组件可以被 LLM 直接调用。它的核心能力是从组件输入 socket 自动生成 LLM 兼容的 tool schema而 socket 本身源自组件run方法的签名与类型标注。关键特性从组件输入 socket 自动生成 LLM 工具调用 schema对组件输入做类型转换与校验支持的类型dataclass、dataclass 列表、基础类型str、int、float、bool、dict及其列表未指定名称时由组件类名自动生成PascalCase 转 snake_case描述默认取自组件 docstring。用法示例包装搜索组件from 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 # 创建 SerperDev 搜索组件 search SerperDevWebSearch(api_keySecret.from_env_var(SERPERDEV_API_KEY), top_k3) # 从组件创建工具 tool ComponentTool( componentsearch, nameweb_search, # 可选默认 serper_dev_web_search descriptionSearch the web for current information on any topic # 可选默认取组件 docstring ) # 构建 PipelineLLM 决定调用工具ToolInvoker 真正执行 pipeline Pipeline() pipeline.add_component(llm, OpenAIChatGenerator(tools[tool])) pipeline.add_component(tool_invoker, ToolInvoker(tools[tool])) pipeline.connect(llm.replies, tool_invoker.messages) message ChatMessage.from_user(Use the web search tool to find information about Nikola Tesla) result pipeline.run({llm: {messages: [message]}}) print(result)构造参数与底层原理ComponentTool.__init__完整签名haystack/tools/component_tool.py#L93-L103def __init__( component: Component, name: str | None None, description: str | None None, parameters: dict[str, Any] | None None, *, outputs_to_string: dict[str, str | Callable[[Any], str]] | None None, inputs_from_state: dict[str, str] | None None, outputs_to_state: dict[str, dict[str, str | Callable]] | None None ) - Noneparameters手动提供 JSON Schema不提供则回退到组件run方法签名生成的 schemaoutputs_to_string/inputs_from_state/outputs_to_state语义与Tool完全一致格式相同异常非组件实例抛TypeError组件已被加入 Pipeline、或 schema 生成失败时抛ValueError。底层调用链ComponentTool内部构造了一个component_invoker以及组件支持异步时的async_component_invoker执行时会把 LLM 给出的 kwargs 按输入 socket 类型做转换_convert_param再调用component.run(**converted_kwargs)haystack/tools/component_tool.py#L182-L220。类型转换逻辑支持 dataclass 及其列表的from_dict调用以及通过TypeAdapter做 Pydantic 校验。Schema 生成遍历component.__haystack_input__._sockets_dict跳过 Callable 类型与 State 类型参数并使用组件参数描述haystack/tools/component_tool.py#L327-L377。相关测试覆盖于 test/tools/test_component_tool.py。PipelineTool把整条流水线封装成工具PipelineToolhaystack/tools/pipeline_tool.py继承自ComponentTool把一条 Haystack Pipeline 包装为工具schema 从流水线输入 socket 自动生成输入描述提取自底层组件 docstring。它适合把检索 生成等完整流程作为一个原子能力暴露给 Agent。用法示例检索流水线作为 Agent 工具from haystack import Document, Pipeline from haystack.dataclasses import ChatMessage from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.embedders.sentence_transformers_text_embedder import SentenceTransformersTextEmbedder from haystack.components.embedders.sentence_transformers_document_embedder import ( SentenceTransformersDocumentEmbedder ) from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.retrievers import InMemoryEmbeddingRetriever from haystack.components.agents import Agent from haystack.tools import PipelineTool # 初始化文档存储并写入文档 document_store InMemoryDocumentStore() document_embedder SentenceTransformersDocumentEmbedder(modelsentence-transformers/all-MiniLM-L6-v2) documents [ Document(contentNikola Tesla was a Serbian-American inventor and electrical engineer.), Document( contentHe is best known for his contributions to the design of the modern alternating current (AC) electricity supply system. ), ] document_embedder.warm_up() docs_with_embeddings document_embedder.run(documentsdocuments)[documents] document_store.write_documents(docs_with_embeddings) # 构建简单检索流水线 retrieval_pipeline Pipeline() retrieval_pipeline.add_component( embedder, SentenceTransformersTextEmbedder(modelsentence-transformers/all-MiniLM-L6-v2) ) retrieval_pipeline.add_component(retriever, InMemoryEmbeddingRetriever(document_storedocument_store)) retrieval_pipeline.connect(embedder.embedding, retriever.query_embedding) # 将流水线包装为工具 retriever_tool PipelineTool( pipelineretrieval_pipeline, input_mapping{query: [embedder.text]}, output_mapping{retriever.documents: documents}, namedocument_retriever, descriptionFor any questions about Nikola Tesla, always use this tool, ) # 创建携带该工具的 Agent agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4.1-mini), tools[retriever_tool] ) result agent.run([ChatMessage.from_user(Who was Nikola Tesla?)]) print(Tool Call Result:) print(result[messages][2].tool_call_result.result) print() print(Answer:) print(result[messages][-1].text)构造参数input_mapping 与 output_mappingPipelineTool.__init__的完整签名haystack/tools/pipeline_tool.py#L96-L108def __init__( pipeline: Pipeline | AsyncPipeline, *, name: str, description: str, input_mapping: dict[str, list[str]] | None None, output_mapping: dict[str, str] | None None, parameters: dict[str, Any] | None None, outputs_to_string: dict[str, str | Callable[[Any], str]] | None None, inputs_from_state: dict[str, str] | None None, outputs_to_state: dict[str, dict[str, str | Callable]] | None None ) - Nonename与description为必填input_mapping把工具参数名映射到流水线输入 socket 路径列表。不提供时基于全部流水线输入自动生成默认映射。示例{query: [retriever.query, prompt_builder.query]}——一个query参数同时喂给多个组件输入output_mapping把流水线输出 socket 路径映射为工具输出名。不提供时自动生成。示例{retriever.documents: documents, generator.replies: replies}parameters手动 JSON Schema回退到组件run方法签名生成outputs_to_string/inputs_from_state/outputs_to_state与Tool/ComponentTool语义一致异常pipeline不是合法 Pipeline 实例时抛ValueError。从源码实现看PipelineTool内部先把流水线包装进SuperComponent(pipeline..., input_mapping..., output_mapping...)再交给ComponentTool.__init__完成 schema 生成与调用逻辑haystack/tools/pipeline_tool.py#L185-L200因此它天然继承了组件级类型转换与校验能力。相关测试见 test/tools/test_pipeline_tool.py。Toolset工具集合与动态加载Toolsethaystack/tools/toolset.py是相关工具的集合可作为一个整体被管理与使用服务两大目的分组管理相关工具把多个工具组织进单一集合统一交给ToolInvoker、Agent或 Chat Generator 使用动态工具加载的基类子类化Toolset可从 OpenAPI URL、MCP 服务器等外部来源动态加载工具。Toolset实现了完整集合接口__iter__、__contains__、__len__、__getitem__行为类似工具列表__contains__同时支持按工具实例与工具名称字符串判断成员关系。分组示例数学工具集from haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker def add_numbers(a: int, b: int) - int: return a b def subtract_numbers(a: int, b: int) - int: return a - b 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 ) math_toolset Toolset([add_tool, subtract_tool]) # 直接把 Toolset 交给 ToolInvoker invoker ToolInvoker(toolsmath_toolset)动态加载示例子类化 Toolsetfrom 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): # 真实场景中应在此从外部来源动态加载工具 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: {}, # 工具动态定义无需序列化数据 } classmethod def from_dict(cls, data): return cls() # 反序列化时重新动态创建工具 calculator_toolset CalculatorToolset() invoker ToolInvoker(toolscalculator_toolset)实现自定义动态 Toolset 的指导原则在__init__中执行动态加载或按 haystack/tools/toolset.py 文档所述在warm_up()中加载并赋值给self.tools同时用自身状态保证幂等若工具是动态定义的必须重写to_dict()/from_dict()序列化端点描述符而非工具实例当工具从外部来源加载时应序列化 URL、服务器信息等描述符而不是动态生成的Tool实例。这样既能保持 Toolset 的动态性、避免序列化大量工具对象的开销也能确保反序列化时能准确重建工具——即使自上次序列化以来工具已被修改或删除。若反序列化工具配置可能加载过期或错误的配置导致运行错误覆盖warm_up()可做共享资源初始化如数据库连接、HTTP 会话而非逐个 warm up 单个工具class MCPToolset(Toolset): def warm_up(self) - None: # 只初始化共享的 MCP 连接不逐个初始化工具 self.mcp_connection establish_connection(self.server_url)集合操作方法add(tool)添加单个Tool或合并另一个Toolset。重名抛ValueError类型不对抛TypeErrorwarm_up()默认遍历并 warm up 所有工具子类可重写__add__(other)支持与Tool、Toolset或list[Tool]拼接返回新的Toolset_ToolsetWrapper内部类当组合不同类型 toolset 时提供统一接口保留各自配置的同时可与ToolInvoker兼容。序列化与反序列化让工具可持久化所有工具抽象都实现了to_dict()/from_dict()返回结构统一为{type: 完全限定类名, data: {...}}其中可调用对象function、async_function、handler通过 haystack/utils/callable_serialization.py 的serialize_callable/deserialize_callable序列化haystack/tools/tool.py#L324-L365。ComponentTool.to_dict()序列化内部组件component_to_dict与全部配置项from_dict()通过deserialize_component_inplace恢复组件haystack/tools/component_tool.py#L285-L325PipelineTool.to_dict()序列化self._pipeline.to_dict()from_dict()中会兼容移除旧版is_pipeline_async键haystack/tools/pipeline_tool.py#L202-L246工具级序列化辅助函数位于 haystack/tools/serde_utils.pyserialize_tools_or_toolset保持 Tool / Toolset 边界deserialize_tools_or_toolset_inplace支持从字典中原位还原单个 Toolset 或 Tool/Toolset 混合列表tools参数指定存储键默认tools。在流水线与 Agent 上下文中工具通常通过deserialize_tools_or_toolset_inplace还原相关测试见 test/tools/test_serde_utils.py。与 Agent、ToolInvoker 的集成方式Tool、ComponentTool、PipelineTool、Toolset可以统一出现在tools参数中ToolsType类型见 haystack/tools/tool_types.py传给OpenAIChatGenerator(tools[tool])、Agent(chat_generator..., tools...)或ToolInvoker(tools...)Toolset因实现集合接口可直接传给期望可迭代工具的组件组合使用时Tool与Toolset可混放在同一个列表里如tools[tool1, math_toolset]由ToolInvoker统一消费haystack/tools/utils.py 提供warm_up_tools统一预热各类工具形态与flatten_tools_or_toolsets把混合列表拍平为list[Tool]。实战建议与注意事项描述质量决定调用准确率name与description是 LLM 决定何时使用工具的关键依据务必准确、无歧义参数描述用Annotated为函数参数添加Annotated[T, 描述]元数据会让生成的 JSON Schema 包含description显著提升 LLM 传参质量默认值即可选参数带默认值的参数在 schema 中标记为可选并携带default无默认值参数标记为必填异步工具async def函数放入async_function同步函数放入function二者放错位置会在构造时直接抛ValueError输出为图片等富内容使用outputs_to_string的raw_result: True模式并确保工具函数或handler返回TextContent/ImageContent列表warm_up 必须幂等流水线/Agent 可能在每次运行前调用warm_up()连接建立、模型加载等操作要用状态标志或判空守卫保护动态 Toolset 序列化描述符从 MCP / OpenAPI 动态加载工具时to_dict()应序列化端点描述符而非工具实例保证反序列化可重建且不过期。综上Haystack Tools 抽象把函数、组件、流水线、动态外部能力统一为一套 LLM 可理解的契约schema 自动生成降低接入成本warm_up/invoke/ 序列化机制保障了与 Agent、Pipeline 深度集成的工程可靠性。从 haystack/tools/ 目录源码与 test/tools/ 测试中可以进一步看到每种工具的边界行为与校验细节。【免费下载链接】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），仅供参考 Haystack Tools 统一抽象实战：用 Tool / ComponentTool / PipelineTool / Toolset 为 LLM 应用构建可调用工具层
Haystack Tools 统一抽象实战用 Tool / ComponentTool / PipelineTool / Toolset 为 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 模块提供了一套统一抽象让 LLM 可以准备一次函数调用再由框架侧真正执行无论是普通 Python 函数、Haystack 组件、完整流水线还是从 MCP/OpenAPI 动态加载的外部能力都能以一致的Tool形态接入 Agent 与 Chat Generator。读完本文你将掌握Tool、ComponentTool、PipelineTool、Toolset四大核心 API 的用法、参数语义与序列化机制并能把现有组件与流水线直接封装为可供 LLM 调用的工具。Tools 在 Haystack 中的作用与核心接口在 Haystack 中一个Tool是Language Models can prepare a call for的载体LLM 根据工具的name、description与parametersJSON Schema生成一次调用请求框架随后通过invoke()真正执行并返回结果。因此文本属性的准确程度直接决定 LLM 能否正确发起调用。Tool的核心接口由 haystack/tools/tool.py 定义tool_spec返回{name, description, parameters}字典即提交给 LLM 的完整工具规格warm_up()用于建立远程连接、加载模型等资源密集型初始化。该方法必须幂等因为在流水线/Agent 组装阶段可能被多次调用invoke(**kwargs)用关键字参数同步调用工具底层函数to_dict()/from_dict()序列化与反序列化保证工具可以随流水线一起持久化。从源码结构看Tool是一个dataclass其__post_init__haystack/tools/tool.py#L111-L212会执行一系列防御性校验function与async_function至少提供一个、同步/协程类型不能放错位置、parameters必须是合法 JSON Schema、outputs_to_state/outputs_to_string/inputs_from_state的结构与引用必须正确。这意味着大部分配置错误会在工具构造阶段就被拦截而不是等到运行时才暴露。除同步调用外Tool还提供invoke_async()优先等待async_function否则通过asyncio.to_thread在 worker 线程中执行同步function。对应测试见 test/tools/test_tool.py如test_invoke_async_falls_back_to_sync_function、test_invoke_on_async_only_tool_raises。Tool 数据类字段与参数语义Tool的字段定义haystack/tools/tool.py#L102-L109如下字段类型说明namestr工具名称LLM 用它选择工具descriptionstr工具描述LLM 据此判断何时使用parametersdict期望参数的 JSON SchemafunctionCallable \| None同步调用函数协程函数必须放入async_functionasync_functionCallable \| None可选的协程函数供invoke_async()使用outputs_to_stringdict \| None工具输出到字符串的转换配置inputs_from_statedict \| None从 Agent State 注入的输入映射outputs_to_statedict \| None工具输出写回 Agent State 的映射输出转换outputs_to_string 的两种格式outputs_to_string支持两种配置格式单输出格式——在根级使用source、handler、raw_result{source: docs, handler: format_documents, raw_result: False}source指定只把某个输出键交给handler不提供则把整个工具结果交给handlerhandler接收工具输出或提取出的source值并返回最终结果的可调用对象raw_result为True时结果不做字符串转换直接返回仍会应用handler。该模式专为返回图片等富内容的工具设计此时工具函数或handler必须返回TextContent/ImageContent对象列表才能与 Chat Generator 兼容。多输出格式——把键映射到各自独立的配置{ formatted_docs: {source: docs, handler: format_documents}, summary: {source: summary_text, handler: str.upper} }每个键对应一个可含source与/或handler的字典。注意raw_result不支持多输出格式此限制在Tool.__post_init__中会被强制校验haystack/tools/tool.py#L181-L188。与 Agent State 联动inputs_from_state 与 outputs_to_stateinputs_from_state: {repository: repo}表示把 State 中键repository的值注入工具参数repo。该映射在构造时会通过_get_valid_inputs()校验参数名真实存在防止拼写错误对应测试test_inputs_from_state_validation_with_invalid_parameteroutputs_to_state定义工具输出如何写回 State提供source时只把指定输出键交给handler再入 State省略source时整个工具结果交给handler。# 指定输出键 处理器 {documents: {source: docs, handler: custom_handler}} # 省略 source整个结果交给 handler {documents: {handler: custom_handler}}对于函数型工具_get_valid_outputs()默认返回None跳过输出校验而ComponentTool会重写该方法返回组件输出 socket 名从而对outputs_to_state的source做严格校验haystack/tools/component_tool.py#L265-L274。从函数创建 Toolcreate_tool_from_function 与 tool 装饰器create_tool_from_functionhaystack/tools/from_function.py#L18-L193把任意带类型标注的函数转换为Tool是无组件、纯函数场景的入口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)关键行为与约束类型提示是硬性要求所有参数必须带类型标注否则抛出ValueError。推荐使用typing.Annotated提供参数描述Annotated[str, 描述]其元数据会成为 JSON Schema 中的description支持的基础类型str、int、float、bool、list、dict、tuple等基本 Python 类型其他类型可能可用但不保证schema 生成机制内部通过 Pydanticcreate_model构建模型再调用model_json_schema()无默认值的参数以...标记为必填haystack/tools/from_function.py#L141-L170随后_remove_title_from_schema会剔除 Pydantic 自动添加的冗余title关键字名称与描述回退未传name时用函数名未传description时用函数 docstring想刻意留空则传空字符串协程支持async def函数会自动放入结果的async_function字段异常类型参数缺类型标注抛ValueErrorschema 生成失败抛SchemaGenerationError。更简洁的 tool 装饰器对于简单场景推荐直接使用tool装饰器haystack/tools/from_function.py#L220-L336。它可以带参数或不带参数使用tool # 不带参数 def my_function(): ... tool(namecustom_name) # 带参数 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) # 输出结果与 create_tool_from_function 完全相同tool与create_tool_from_function支持相同的name、description、inputs_from_state、outputs_to_state、outputs_to_string参数两者本质是同一套转换逻辑的两种调用方式。ComponentTool把 Haystack 组件封装成工具ComponentToolhaystack/tools/component_tool.py让任意 Haystack 组件可以被 LLM 直接调用。它的核心能力是从组件输入 socket 自动生成 LLM 兼容的 tool schema而 socket 本身源自组件run方法的签名与类型标注。关键特性从组件输入 socket 自动生成 LLM 工具调用 schema对组件输入做类型转换与校验支持的类型dataclass、dataclass 列表、基础类型str、int、float、bool、dict及其列表未指定名称时由组件类名自动生成PascalCase 转 snake_case描述默认取自组件 docstring。用法示例包装搜索组件from 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 # 创建 SerperDev 搜索组件 search SerperDevWebSearch(api_keySecret.from_env_var(SERPERDEV_API_KEY), top_k3) # 从组件创建工具 tool ComponentTool( componentsearch, nameweb_search, # 可选默认 serper_dev_web_search descriptionSearch the web for current information on any topic # 可选默认取组件 docstring ) # 构建 PipelineLLM 决定调用工具ToolInvoker 真正执行 pipeline Pipeline() pipeline.add_component(llm, OpenAIChatGenerator(tools[tool])) pipeline.add_component(tool_invoker, ToolInvoker(tools[tool])) pipeline.connect(llm.replies, tool_invoker.messages) message ChatMessage.from_user(Use the web search tool to find information about Nikola Tesla) result pipeline.run({llm: {messages: [message]}}) print(result)构造参数与底层原理ComponentTool.__init__完整签名haystack/tools/component_tool.py#L93-L103def __init__( component: Component, name: str | None None, description: str | None None, parameters: dict[str, Any] | None None, *, outputs_to_string: dict[str, str | Callable[[Any], str]] | None None, inputs_from_state: dict[str, str] | None None, outputs_to_state: dict[str, dict[str, str | Callable]] | None None ) - Noneparameters手动提供 JSON Schema不提供则回退到组件run方法签名生成的 schemaoutputs_to_string/inputs_from_state/outputs_to_state语义与Tool完全一致格式相同异常非组件实例抛TypeError组件已被加入 Pipeline、或 schema 生成失败时抛ValueError。底层调用链ComponentTool内部构造了一个component_invoker以及组件支持异步时的async_component_invoker执行时会把 LLM 给出的 kwargs 按输入 socket 类型做转换_convert_param再调用component.run(**converted_kwargs)haystack/tools/component_tool.py#L182-L220。类型转换逻辑支持 dataclass 及其列表的from_dict调用以及通过TypeAdapter做 Pydantic 校验。Schema 生成遍历component.__haystack_input__._sockets_dict跳过 Callable 类型与 State 类型参数并使用组件参数描述haystack/tools/component_tool.py#L327-L377。相关测试覆盖于 test/tools/test_component_tool.py。PipelineTool把整条流水线封装成工具PipelineToolhaystack/tools/pipeline_tool.py继承自ComponentTool把一条 Haystack Pipeline 包装为工具schema 从流水线输入 socket 自动生成输入描述提取自底层组件 docstring。它适合把检索 生成等完整流程作为一个原子能力暴露给 Agent。用法示例检索流水线作为 Agent 工具from haystack import Document, Pipeline from haystack.dataclasses import ChatMessage from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.embedders.sentence_transformers_text_embedder import SentenceTransformersTextEmbedder from haystack.components.embedders.sentence_transformers_document_embedder import ( SentenceTransformersDocumentEmbedder ) from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.retrievers import InMemoryEmbeddingRetriever from haystack.components.agents import Agent from haystack.tools import PipelineTool # 初始化文档存储并写入文档 document_store InMemoryDocumentStore() document_embedder SentenceTransformersDocumentEmbedder(modelsentence-transformers/all-MiniLM-L6-v2) documents [ Document(contentNikola Tesla was a Serbian-American inventor and electrical engineer.), Document( contentHe is best known for his contributions to the design of the modern alternating current (AC) electricity supply system. ), ] document_embedder.warm_up() docs_with_embeddings document_embedder.run(documentsdocuments)[documents] document_store.write_documents(docs_with_embeddings) # 构建简单检索流水线 retrieval_pipeline Pipeline() retrieval_pipeline.add_component( embedder, SentenceTransformersTextEmbedder(modelsentence-transformers/all-MiniLM-L6-v2) ) retrieval_pipeline.add_component(retriever, InMemoryEmbeddingRetriever(document_storedocument_store)) retrieval_pipeline.connect(embedder.embedding, retriever.query_embedding) # 将流水线包装为工具 retriever_tool PipelineTool( pipelineretrieval_pipeline, input_mapping{query: [embedder.text]}, output_mapping{retriever.documents: documents}, namedocument_retriever, descriptionFor any questions about Nikola Tesla, always use this tool, ) # 创建携带该工具的 Agent agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4.1-mini), tools[retriever_tool] ) result agent.run([ChatMessage.from_user(Who was Nikola Tesla?)]) print(Tool Call Result:) print(result[messages][2].tool_call_result.result) print() print(Answer:) print(result[messages][-1].text)构造参数input_mapping 与 output_mappingPipelineTool.__init__的完整签名haystack/tools/pipeline_tool.py#L96-L108def __init__( pipeline: Pipeline | AsyncPipeline, *, name: str, description: str, input_mapping: dict[str, list[str]] | None None, output_mapping: dict[str, str] | None None, parameters: dict[str, Any] | None None, outputs_to_string: dict[str, str | Callable[[Any], str]] | None None, inputs_from_state: dict[str, str] | None None, outputs_to_state: dict[str, dict[str, str | Callable]] | None None ) - Nonename与description为必填input_mapping把工具参数名映射到流水线输入 socket 路径列表。不提供时基于全部流水线输入自动生成默认映射。示例{query: [retriever.query, prompt_builder.query]}——一个query参数同时喂给多个组件输入output_mapping把流水线输出 socket 路径映射为工具输出名。不提供时自动生成。示例{retriever.documents: documents, generator.replies: replies}parameters手动 JSON Schema回退到组件run方法签名生成outputs_to_string/inputs_from_state/outputs_to_state与Tool/ComponentTool语义一致异常pipeline不是合法 Pipeline 实例时抛ValueError。从源码实现看PipelineTool内部先把流水线包装进SuperComponent(pipeline..., input_mapping..., output_mapping...)再交给ComponentTool.__init__完成 schema 生成与调用逻辑haystack/tools/pipeline_tool.py#L185-L200因此它天然继承了组件级类型转换与校验能力。相关测试见 test/tools/test_pipeline_tool.py。Toolset工具集合与动态加载Toolsethaystack/tools/toolset.py是相关工具的集合可作为一个整体被管理与使用服务两大目的分组管理相关工具把多个工具组织进单一集合统一交给ToolInvoker、Agent或 Chat Generator 使用动态工具加载的基类子类化Toolset可从 OpenAPI URL、MCP 服务器等外部来源动态加载工具。Toolset实现了完整集合接口__iter__、__contains__、__len__、__getitem__行为类似工具列表__contains__同时支持按工具实例与工具名称字符串判断成员关系。分组示例数学工具集from haystack.tools import Tool, Toolset from haystack.components.tools import ToolInvoker def add_numbers(a: int, b: int) - int: return a b def subtract_numbers(a: int, b: int) - int: return a - b 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 ) math_toolset Toolset([add_tool, subtract_tool]) # 直接把 Toolset 交给 ToolInvoker invoker ToolInvoker(toolsmath_toolset)动态加载示例子类化 Toolsetfrom 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): # 真实场景中应在此从外部来源动态加载工具 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: {}, # 工具动态定义无需序列化数据 } classmethod def from_dict(cls, data): return cls() # 反序列化时重新动态创建工具 calculator_toolset CalculatorToolset() invoker ToolInvoker(toolscalculator_toolset)实现自定义动态 Toolset 的指导原则在__init__中执行动态加载或按 haystack/tools/toolset.py 文档所述在warm_up()中加载并赋值给self.tools同时用自身状态保证幂等若工具是动态定义的必须重写to_dict()/from_dict()序列化端点描述符而非工具实例当工具从外部来源加载时应序列化 URL、服务器信息等描述符而不是动态生成的Tool实例。这样既能保持 Toolset 的动态性、避免序列化大量工具对象的开销也能确保反序列化时能准确重建工具——即使自上次序列化以来工具已被修改或删除。若反序列化工具配置可能加载过期或错误的配置导致运行错误覆盖warm_up()可做共享资源初始化如数据库连接、HTTP 会话而非逐个 warm up 单个工具class MCPToolset(Toolset): def warm_up(self) - None: # 只初始化共享的 MCP 连接不逐个初始化工具 self.mcp_connection establish_connection(self.server_url)集合操作方法add(tool)添加单个Tool或合并另一个Toolset。重名抛ValueError类型不对抛TypeErrorwarm_up()默认遍历并 warm up 所有工具子类可重写__add__(other)支持与Tool、Toolset或list[Tool]拼接返回新的Toolset_ToolsetWrapper内部类当组合不同类型 toolset 时提供统一接口保留各自配置的同时可与ToolInvoker兼容。序列化与反序列化让工具可持久化所有工具抽象都实现了to_dict()/from_dict()返回结构统一为{type: 完全限定类名, data: {...}}其中可调用对象function、async_function、handler通过 haystack/utils/callable_serialization.py 的serialize_callable/deserialize_callable序列化haystack/tools/tool.py#L324-L365。ComponentTool.to_dict()序列化内部组件component_to_dict与全部配置项from_dict()通过deserialize_component_inplace恢复组件haystack/tools/component_tool.py#L285-L325PipelineTool.to_dict()序列化self._pipeline.to_dict()from_dict()中会兼容移除旧版is_pipeline_async键haystack/tools/pipeline_tool.py#L202-L246工具级序列化辅助函数位于 haystack/tools/serde_utils.pyserialize_tools_or_toolset保持 Tool / Toolset 边界deserialize_tools_or_toolset_inplace支持从字典中原位还原单个 Toolset 或 Tool/Toolset 混合列表tools参数指定存储键默认tools。在流水线与 Agent 上下文中工具通常通过deserialize_tools_or_toolset_inplace还原相关测试见 test/tools/test_serde_utils.py。与 Agent、ToolInvoker 的集成方式Tool、ComponentTool、PipelineTool、Toolset可以统一出现在tools参数中ToolsType类型见 haystack/tools/tool_types.py传给OpenAIChatGenerator(tools[tool])、Agent(chat_generator..., tools...)或ToolInvoker(tools...)Toolset因实现集合接口可直接传给期望可迭代工具的组件组合使用时Tool与Toolset可混放在同一个列表里如tools[tool1, math_toolset]由ToolInvoker统一消费haystack/tools/utils.py 提供warm_up_tools统一预热各类工具形态与flatten_tools_or_toolsets把混合列表拍平为list[Tool]。实战建议与注意事项描述质量决定调用准确率name与description是 LLM 决定何时使用工具的关键依据务必准确、无歧义参数描述用Annotated为函数参数添加Annotated[T, 描述]元数据会让生成的 JSON Schema 包含description显著提升 LLM 传参质量默认值即可选参数带默认值的参数在 schema 中标记为可选并携带default无默认值参数标记为必填异步工具async def函数放入async_function同步函数放入function二者放错位置会在构造时直接抛ValueError输出为图片等富内容使用outputs_to_string的raw_result: True模式并确保工具函数或handler返回TextContent/ImageContent列表warm_up 必须幂等流水线/Agent 可能在每次运行前调用warm_up()连接建立、模型加载等操作要用状态标志或判空守卫保护动态 Toolset 序列化描述符从 MCP / OpenAPI 动态加载工具时to_dict()应序列化端点描述符而非工具实例保证反序列化可重建且不过期。综上Haystack Tools 抽象把函数、组件、流水线、动态外部能力统一为一套 LLM 可理解的契约schema 自动生成降低接入成本warm_up/invoke/ 序列化机制保障了与 Agent、Pipeline 深度集成的工程可靠性。从 haystack/tools/ 目录源码与 test/tools/ 测试中可以进一步看到每种工具的边界行为与校验细节。【免费下载链接】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），仅供参考