企业官网建设流程全解析

Haystack OpenAPI 集成指南：用 OpenAPIConnector 与 OpenAPIServiceConnector 让 Pipeline 直接调用任意 REST API

【免费下载链接】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

OpenAPI（前身 Swagger）规范是描述 REST API 的事实标准。Haystack 的 OpenAPI 集成组件充当 Haystack Pipeline 与任意遵循该规范的 REST 服务之间的桥梁：既可以在没有 LLM 参与的情况下按operation_id直接调用端点，也可以通过 LLM 工具调用（Tool Calling）自动把自然语言查询转换为对远端服务的 API 请求。读完本文，你将掌握OpenAPIConnector、OpenAPIServiceConnector与OpenAPIServiceToFunctions三个组件的完整用法、参数语义、认证处理方式，以及如何在真实 Pipeline 中搭建"LLM 生成工具调用 → 调用 OpenAPI 服务 → 返回结果"的端到端链路。

背景：OpenAPI 集成在 Haystack 中的定位

Haystack 是一个用于构建 LLM 应用的开源编排框架，其核心抽象是组件（Component）与 Pipeline。OpenAPI 相关组件属于"集成"（integration）范畴，它们在官方参考文档 integrations-api/openapi.md（对应 2.22 版本）中得到了完整定义，包含以下三个核心构件：

一句话概括分工：OpenAPIServiceToFunctions负责"翻译规范"，OpenAPIServiceConnector负责"执行调用"，OpenAPIConnector则提供一条不需要 LLM 的直连路径。

安装与导入：openapi-haystack 包

在 Haystack 2.22 及后续版本中，这三个组件已经从 Haystack 主库迁移到了独立的openapi-haystack集成包中。对应发布说明见 remove-openapi-components-2f1de8f6c1b2787f.yaml 与 deprecate-openapi-components-e5f0f7470218fcc4.yaml：

pip install openapi-haystack

导入路径统一以haystack_integrations开头，与 2.22 参考文档中的签名保持一致：

from haystack_integrations.components.connectors.openapi import OpenAPIConnector, OpenAPIServiceConnector from haystack_integrations.components.converters.openapi import OpenAPIServiceToFunctions

迁移说明：若你曾在旧版本中写过from haystack.components.connectors import OpenAPIConnector，升级后需同步改为上述haystack_integrations路径。详细对照见 remove-openapi-components-2f1de8f6c1b2787f.yaml。

OpenAPIConnector：不经过 LLM 直接调用 REST 端点

OpenAPIConnector是 Haystack 与任何遵循 OpenAPI 规范的 REST API 之间的直接桥梁。它动态解析 API 规范，为执行 API 操作提供统一接口，适合那些调用参数由业务逻辑明确决定、无需 LLM 生成载荷的场景。官方介绍详见 connectors/openapiconnector.mdx。

初始化参数

__init__( openapi_spec: str, credentials: Secret | None = None, service_kwargs: dict[str, Any] | None = None, ) -> None

credentials推荐使用Secret.from_env_var()从环境变量读取，避免在代码中硬编码密钥，例如Secret.from_env_var("SERPERDEV_API_KEY")。

run：按 operation_id 调用端点

run( operation_id: str, arguments: dict[str, Any] | None = None ) -> dict[str, Any]

• operation_id：OpenAPI 规范中要调用的operationId。每个操作（如GET /search）在规范里都有唯一的operationId，这是本次调用的"方法名"。
• arguments：可选参数，用于填充端点的 query、path 或 body 参数。
• 返回：包含服务响应的字典，输出键为response，即 REST 端点返回的 JSON 内容。

warm_up 与序列化

• warm_up() -> None：初始化 OpenAPI 客户端，通常在 Pipeline 预热阶段被调用。
• to_dict() -> dict[str, Any]：把组件序列化为字典，便于保存/恢复 Pipeline 配置。
• from_dict(data: dict[str, Any]) -> OpenAPIConnector：从字典反序列化还原组件实例。

独立使用示例

参考文档（integrations-api/openapi.md）给出的完整示例：

from haystack.utils import Secret from haystack_integrations.components.connectors.openapi import OpenAPIConnector serper_dev_token = Secret.from_env_var("SERPERDEV_API_KEY") def my_custom_config_factory(): # Create and return a custom configuration for the OpenAPIClient pass connector = OpenAPIConnector( openapi_spec="https://bit.ly/serperdev_openapi", credentials=serper_dev_token, service_kwargs={"config_factory": my_custom_config_factory()} ) response = connector.run( operation_id="search", arguments={"q": "Who was Nikola Tesla?"} )

注意：service_kwargs是可选的，仅当需要向OpenAPIClient传递额外选项（例如自定义config_factory）时才使用。

在 Pipeline 中集成

OpenAPIConnector也可以作为一个普通组件挂进Pipeline。官方文档 connectors/openapiconnector.mdx 中的示例展示了如何把用户问题作为参数传入：

from haystack import Pipeline from haystack_integrations.components.connectors.openapi import OpenAPIConnector from haystack.dataclasses.chat_message import ChatMessage from haystack.utils import Secret connector = OpenAPIConnector( openapi_spec="https://bit.ly/serperdev_openapi", credentials=Secret.from_env_var("SERPERDEV_API_KEY"), ) user_message = ChatMessage.from_user(text="Who was Nikola Tesla?") pipeline = Pipeline() pipeline.add_component("openapi_connector", connector) response = pipeline.run( data={ "openapi_connector": { "operation_id": "search", "arguments": {"q": user_message.text}, }, }, ) answer = response.get("openapi_connector", {}).get("response", {}) print(answer)

管道运行时会输出形如{"response": { ...REST 端点返回的 JSON... }}的结构，取openapi_connector.response即可拿到服务结果。

patch_request：底层 HTTP 请求补丁

patch_request位于haystack_integrations.components.connectors.openapi.openapi_service模块，是对 OpenAPI 客户端中Operation对象方法的底层封装，负责真正发送 HTTP 请求：

patch_request( self: Operation, base_url: str, *, data: Any | None = None, parameters: dict[str, Any] | None = None, raw_response: bool = False, security: dict[str, str] | None = None, session: Any | None = None, verify: bool | str = True ) -> Any | None

返回值为响应数据，具体是原始响应还是经过处理的响应，取决于raw_response标志。日常使用中开发者一般不需要直接触碰该函数，它主要由OpenAPIServiceConnector在内部驱动。

OpenAPIServiceConnector：基于 ChatMessage 的 LLM 工具调用

如果说OpenAPIConnector是"手工拨号"，那么OpenAPIServiceConnector就是"语音助手代为拨号"：它从ChatMessage的ToolCall条目中解析出要调用的方法名与参数，然后调用 OpenAPI 服务，并把服务响应封装成ChatMessage返回。参考文档见 connectors/openapiserviceconnector.mdx。

初始化与 ssl_verify

__init__(ssl_verify: bool | str | None = None) -> None

该组件没有必填的初始化参数。ssl_verify用于决定是否对请求做 SSL 校验，传入字符串时作为 CA 证书使用；为None时使用默认行为。

run：解析工具调用并执行

run( messages: list[ChatMessage], service_openapi_spec: dict[str, Any], service_credentials: dict | str | None = None, ) -> dict[str, list[ChatMessage]]

• messages：ChatMessage列表。组件会解析列表中的最后一条消息，期望它包含工具调用（ToolCall）。
• service_openapi_spec：服务的 OpenAPI JSON 规范对象，所有$ref引用必须已解析。这正是OpenAPIServiceToFunctions输出openapi_specs的原因——两者天然衔接。
• service_credentials：与服务认证用的凭据。目前仅支持 OpenAPI 规范 v3 中的两类安全方案：http（Basic、Bearer 等 HTTP 认证方案）与apiKey（API Key 与 Cookie 认证）。
• 返回：字典，键为service_response，值为ChatMessage列表；每条消息对应一次工具调用，内容为 JSON 格式的字符串。若一条消息包含多个工具调用，则会有多条响应。
• 异常：若最后一条消息不是来自 assistant，或不包含工具调用，将抛出ValueError。

从发布说明 openapi-connector-auth-enhancement-a78e0666d3cf6353.yaml 可知，service_credentials是每次run调用时动态提供的，无需预先配置一组固定的服务认证，因此可以随时接入带不同认证的新服务。此外 update-openapi-service-connector-e49f665968013425.yaml 记录了对新版ChatMessage格式的兼容支持，complex-types-openapi-support-84d3daf8927ad915.yaml 则补充了对请求/响应复杂类型的处理能力。

使用示例

参考文档中的示例（以 https://serper.dev/ 服务为例，需先定义serper_token，可通过SERPERDEV_API_KEY环境变量或直接赋值）：

import json import httpx from haystack.dataclasses import ChatMessage, ToolCall from haystack.utils import Secret from haystack_integrations.components.connectors.openapi import OpenAPIServiceConnector tool_call = ToolCall( tool_name="search", arguments={"q": "Why was Sam Altman ousted from OpenAI?"}, ) message = ChatMessage.from_assistant(tool_calls=[tool_call]) serper_token = Secret.from_env_var("SERPERDEV_API_KEY").resolve_value() serperdev_openapi_spec = json.loads(httpx.get("https://bit.ly/serper_dev_spec", follow_redirects=True).text) service_connector = OpenAPIServiceConnector() result = service_connector.run( messages=[message], service_openapi_spec=serperdev_openapi_spec, service_credentials=serper_token, ) print(result)

输出形如：

{'service_response': ChatMessage(_role=<ChatRole.USER: 'user'>, _content=[TextContent(text= '{"searchParameters": {"q": "Why was Sam Altman ousted from OpenAI?", "type": "search", "engine": "google"}, "answerBox": {"snippet": "Concerns over AI safety and OpenAI's role in protecting were at the center of Altman's brief ouster from the company."...})])}

可以看到，服务响应被封装为一条ChatRole.USER的ChatMessage，content中是 JSON 字符串。

重要提示：OpenAPIServiceConnector通常不单独使用，而是作为 Pipeline 的一部分，配合OpenAPIServiceToFunctions与具备工具调用能力的 Chat Generator（LLM）协同工作。上面示例中的工具调用载荷是手工构造的，真实场景中一般由 Chat Generator 生成。

OpenAPIServiceToFunctions：把 OpenAPI 规范翻译成函数调用格式

OpenAPIServiceToFunctions把 OpenAPI 服务定义转换为适合 LLM 函数调用（OpenAI Function Calling 风格）的格式，是"规范 → 可调用工具"的翻译器。详细介绍见 converters/openapiservicetofunctions.mdx。

对规范的要求

规范必须遵循OpenAPI 3.0.0 或更高版本，可用 JSON 或 YAML 格式书写。规范中的每个函数必须具备：

• 唯一的operationId；
• description（描述，供 LLM 理解该函数的用途）；
• requestBody和/或parameters；
• 为requestBody和/或parameters提供的schema。

run 与返回字段

run(sources: list[str | Path | ByteStream]) -> dict[str, Any]

• sources：OpenAPI 定义的文件路径或ByteStream对象列表（JSON 或 YAML 格式）。
• 返回：包含两个键的字典：
  • functions：JSON 对象格式的函数定义（每个 path 定义对应一个函数定义）；
  • openapi_specs：引用已解析的 OpenAPI 规范对象（JSON/YAML），可直接作为OpenAPIServiceConnector的service_openapi_spec输入。
• 异常：RuntimeError（无法下载或处理 OpenAPI 定义时）、ValueError（源类型无法识别，或定义中找不到任何函数时）。

使用示例

参考文档给出的最小示例：

from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.converters.openapi import OpenAPIServiceToFunctions converter = OpenAPIServiceToFunctions() spec = ByteStream.from_string( '{"openapi":"3.0.0","info":{"title":"API","version":"1.0.0"},"paths":{"/search":{"get":{"operationId":"search","summary":"Search","parameters":[{"name":"q","in":"query","required":true,"schema":{"type":"string"}}]}}}}' ) result = converter.run(sources=[spec]) assert result["functions"]

示例中的内联规范定义了一个GET /search操作，operationId为search，含必填 query 参数q。转换后result["functions"]中即为对应的函数定义对象。

端到端实战：把 Serper 搜索引擎接入 RAG 式问答管道

将上述组件串成完整 Pipeline 的经典场景是把 serper.dev 搜索服务桥接进 Haystack：OpenAPIServiceToFunctions先获取并转换 Serper 的 OpenAPI 规范为 LLM 可理解的函数定义，OpenAPIServiceConnector再依据该规范实际激活并调用 Serper 服务。以下完整代码来自 connectors/openapiserviceconnector.mdx（运行前需要你自己的 Serper 与 OpenAI API Key）：

import json import requests from typing import Any from haystack import Pipeline from haystack.components.converters import OutputAdapter from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.dataclasses.byte_stream import ByteStream from haystack_integrations.components.connectors.openapi import OpenAPIServiceConnector from haystack_integrations.components.converters.openapi import ( OpenAPIServiceToFunctions, ) def prepare_fc_params(openai_functions_schema: dict[str, Any]) -> dict[str, Any]: return { "tools": [{"type": "function", "function": openai_functions_schema}], "tool_choice": { "type": "function", "function": {"name": openai_functions_schema["name"]}, }, } serperdev_spec = requests.get("https://bit.ly/serper_dev_spec").json() system_prompt = requests.get("https://bit.ly/serper_dev_system").text user_prompt = "Why was Sam Altman ousted from OpenAI?" pipe = Pipeline() pipe.add_component("spec_to_functions", OpenAPIServiceToFunctions()) pipe.add_component( "prepare_fc_adapter", OutputAdapter( "{{functions[0] | prepare_fc}}", dict[str, Any], {"prepare_fc": prepare_fc_params}, ), ) pipe.add_component("functions_llm", OpenAIChatGenerator()) pipe.add_component("openapi_connector", OpenAPIServiceConnector()) pipe.add_component( "message_adapter", OutputAdapter( "{{system_message + service_response}}", list[ChatMessage], unsafe=True, ), ) pipe.add_component("llm", OpenAIChatGenerator()) pipe.connect("spec_to_functions.functions", "prepare_fc_adapter.functions") pipe.connect( "spec_to_functions.openapi_specs", "openapi_connector.service_openapi_spec", ) pipe.connect("prepare_fc_adapter", "functions_llm.generation_kwargs") pipe.connect("functions_llm.replies", "openapi_connector.messages") pipe.connect("openapi_connector.service_response", "message_adapter.service_response") pipe.connect("message_adapter", "llm.messages") result = pipe.run( data={ "functions_llm": { "messages": [ ChatMessage.from_system("Only do tool/function calling"), ChatMessage.from_user(user_prompt), ], }, "openapi_connector": { "service_credentials": serper_dev_key, }, "spec_to_functions": { "sources": [ByteStream.from_string(json.dumps(serperdev_spec))], }, "message_adapter": { "system_message": [ChatMessage.from_system(system_prompt)], }, }, ) print(result["llm"]["replies"][0].text)

这段管道的执行链路可以拆解为五步：

1. spec_to_functions把 Serper 的 OpenAPI 规范转换为函数定义（functions）与引用已解析的规范（openapi_specs）；
2. prepare_fc_adapter用OutputAdapter把函数定义包装成tools/tool_choice形式的generation_kwargs；
3. functions_llm（OpenAIChatGenerator）依据这些函数定义生成工具调用（返回的replies是包含ToolCall的ChatMessage）；
4. openapi_connector（OpenAPIServiceConnector）读取消息中的工具调用，结合openapi_specs与动态传入的service_credentials调用 Serper 的 REST 端点；
5. message_adapter把系统提示与service_response合并回消息列表，交给llm生成最终答案。

运行结果类似："Sam Altman was ousted from OpenAI on November 17, 2023, following a 'deliberative review process' by the board of directors…"，即 LLM 借助实时搜索工具返回的事实性回答。注意 Serper 只是示例，任何符合 OpenAPI 规范的服务都可照此接入。

序列化：to_dict 与 from_dict

OpenAPIConnector与OpenAPIServiceConnector都实现了标准 Haystack 组件的序列化协议：

• to_dict() -> dict[str, Any]：把组件（含初始化参数）序列化为字典，便于保存为 YAML/JSON 形式的 Pipeline 定义。
• from_dict(data: dict[str, Any]) -> 对应组件：从字典重建组件实例。

这保证了 Pipeline 定义可以版本化、跨环境复用，与 Haystack 整体的序列化机制（见 serialization.py）保持一致。

演进与迁移：legacy 状态与 MCPTool 建议

需要特别说明的是，发布说明 deprecate-openapi-components-e5f0f7470218fcc4.yaml 与 remove-openapi-components-2f1de8f6c1b2787f.yaml 明确指出：这三个组件是 Haystack 连接外部 API 的legacy（遗留）方式，已从 Haystack 核心库移除并迁移到openapi-haystack包。官方建议对于大多数用例，优先考虑MCPTool——它是给 Pipeline 与 Agent 提供外部工具和服务的现代、标准化方式。

因此在使用决策上可以这样判断：

• 已有基于这三个组件的存量代码：安装openapi-haystack并更新导入路径即可继续使用（2.22 参考文档所记录的 API 签名与行为均以此为准）；
• 新建项目：优先评估MCPTool等现代工具接入方案，再决定是否沿用 OpenAPI 直连组件。

小结

本文基于 2.22 版 OpenAPI 集成参考文档，结合仓库内对应的组件指南（openapiconnector.mdx、openapiserviceconnector.mdx、openapiservicetofunctions.mdx）与多份发布说明，梳理了 Haystack OpenAPI 集成的完整面貌：

• OpenAPIConnector提供无需 LLM 的直接 REST 调用通道；
• OpenAPIServiceConnector从ChatMessage工具调用中解析并执行服务方法，支持http与apiKey两类认证方案，且凭据可动态传入；
• OpenAPIServiceToFunctions把 OpenAPI 3.0+ 规范转换为函数调用格式，与前者形成"转换 + 执行"的组合；
• 三者配合OutputAdapter、Chat Generator 即可搭建"自然语言 → 工具调用 → REST 调用 → 最终回答"的完整 Agent 链路。

这些组件的实现思路——用标准规范驱动动态调用、用工具调用连接 LLM 与外部世界——至今仍是构建可扩展 Agent 与 RAG 应用的重要设计范式。

【免费下载链接】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