企业官网建设流程全解析

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/haystack

Haystack 的 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与parameters（JSON 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）如下：

输出转换：outputs_to_string 的两种格式

outputs_to_string支持两种配置格式：

1. 单输出格式——在根级使用source、handler、raw_result：

{"source": "docs", "handler": format_documents, "raw_result": False}

• source：指定只把某个输出键交给handler；不提供则把整个工具结果交给handler；
• handler：接收工具输出（或提取出的source值）并返回最终结果的可调用对象；
• raw_result：为True时结果不做字符串转换直接返回（仍会应用handler）。该模式专为返回图片等富内容的工具设计，此时工具函数或handler必须返回TextContent/ImageContent对象列表，才能与 Chat Generator 兼容。

2. 多输出格式——把键映射到各自独立的配置：

{ "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_state

• inputs_from_state: {"repository": "repo"}表示把 State 中键repository的值注入工具参数repo。该映射在构造时会通过_get_valid_inputs()校验参数名真实存在，防止拼写错误（对应测试test_inputs_from_state_validation_with_invalid_parameter）；
• outputs_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）。

从函数创建 Tool：create_tool_from_function 与 @tool 装饰器

create_tool_from_function（haystack/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 f"Weather report for {city}: 20 {unit}, sunny" tool = create_tool_from_function(get_weather) print(tool) # Tool(name='get_weather', description='A 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'}}}, # function=<function 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字段；
• 异常类型：参数缺类型标注抛ValueError；schema 生成失败抛SchemaGenerationError。

更简洁的 @tool 装饰器

对于简单场景，推荐直接使用@tool装饰器（haystack/tools/from_function.py#L220-L336）。它可以带参数或不带参数使用：

@tool # 不带参数 def my_function(): ... @tool(name="custom_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 f"Weather 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 组件封装成工具

ComponentTool（haystack/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_key=Secret.from_env_var("SERPERDEV_API_KEY"), top_k=3) # 从组件创建工具 tool = ComponentTool( component=search, name="web_search", # 可选：默认 "serper_dev_web_search" description="Search the web for current information on any topic" # 可选：默认取组件 docstring ) # 构建 Pipeline：LLM 决定调用工具，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-L103）：

def __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 ) -> None

• parameters：手动提供 JSON Schema；不提供则回退到组件run方法签名生成的 schema；
• outputs_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：把整条流水线封装成工具

PipelineTool（haystack/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(model="sentence-transformers/all-MiniLM-L6-v2") documents = [ Document(content="Nikola Tesla was a Serbian-American inventor and electrical engineer."), Document( content="He 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(documents=documents)["documents"] document_store.write_documents(docs_with_embeddings) # 构建简单检索流水线 retrieval_pipeline = Pipeline() retrieval_pipeline.add_component( "embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2") ) retrieval_pipeline.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store)) retrieval_pipeline.connect("embedder.embedding", "retriever.query_embedding") # 将流水线包装为工具 retriever_tool = PipelineTool( pipeline=retrieval_pipeline, input_mapping={"query": ["embedder.text"]}, output_mapping={"retriever.documents": "documents"}, name="document_retriever", description="For any questions about Nikola Tesla, always use this tool", ) # 创建携带该工具的 Agent agent = Agent( chat_generator=OpenAIChatGenerator(model="gpt-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_mapping

PipelineTool.__init__的完整签名（haystack/tools/pipeline_tool.py#L96-L108）：

def __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 ) -> None

• name与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：工具集合与动态加载

Toolset（haystack/tools/toolset.py）是相关工具的集合，可作为一个整体被管理与使用，服务两大目的：

1. 分组管理相关工具：把多个工具组织进单一集合，统一交给ToolInvoker、Agent或 Chat Generator 使用；
2. 动态工具加载的基类：子类化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( name="add", description="Add two numbers", parameters={ "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"} }, "required": ["a", "b"] }, function=add_numbers ) subtract_tool = Tool( name="subtract", description="Subtract b from a", parameters={ "type": "object", "properties": { "a": {"type": "integer"}, "b": {"type": "integer"} }, "required": ["a", "b"] }, function=subtract_numbers ) math_toolset = Toolset([add_tool, subtract_tool]) # 直接把 Toolset 交给 ToolInvoker invoker = ToolInvoker(tools=math_toolset)

动态加载示例：子类化 Toolset

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): # 真实场景中应在此从外部来源动态加载工具 tools = [] add_tool = Tool( name="add", description="Add two numbers", parameters={ "type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"], }, function=lambda a, b: a + b, ) multiply_tool = Tool( name="multiply", description="Multiply two numbers", parameters={ "type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"], }, function=lambda 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(tools=calculator_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，类型不对抛TypeError；
• warm_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-L325）；
• PipelineTool.to_dict()序列化self._pipeline.to_dict()，from_dict()中会兼容移除旧版is_pipeline_async键（haystack/tools/pipeline_tool.py#L202-L246）；
• 工具级序列化辅助函数位于 haystack/tools/serde_utils.py：serialize_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