用于集成外部服务的各种连接器。
模块 openapi_service
OpenAPIServiceConnector
一个将 Haystack 框架连接到 OpenAPI 服务的组件。
该OpenAPIServiceConnector 组件将 Haystack 框架连接到 OpenAPI 服务，使其能够根据服务 OpenAPI 规范中定义的调用操作。
它与ChatMessage 数据类集成，其中消息中的 payload 用于确定要调用的方法和要传递的参数。消息 payload 应该是 OpenAI JSON 格式的函数调用字符串，包含方法名和要传递给方法的参数。然后，方法名和参数用于调用 OpenAPI 服务上的方法。来自服务的响应被返回为ChatMessage.
在使用此组件之前，用户通常会在以下组件的帮助下解析服务端点参数：OpenAPIServiceToFunctions 组件。
以下示例演示了如何使用OpenAPIServiceConnector 调用通过 OpenAPI 规范指定的 https://serper.dev/ 服务的某个方法。
但请注意，OpenAPIServiceConnector 通常不直接使用，而是作为包含OpenAPIServiceToFunctions 组件和一个OpenAIChatGenerator 组件的管道的一部分，该组件使用支持函数调用的 LLM。在下面的示例中，我们直接使用函数调用 payload，但在实际场景中，函数调用 payload 通常由OpenAIChatGenerator 组件生成。
使用示例
import json
import requests
from haystack.components.connectors import OpenAPIServiceConnector
from haystack.dataclasses import ChatMessage
fc_payload = [{'function': {'arguments': '{"q": "Why was Sam Altman ousted from OpenAI?"}', 'name': 'search'},
'id': 'call_PmEBYvZ7mGrQP5PUASA5m9wO', 'type': 'function'}]
serper_token = <your_serper_dev_token>
serperdev_openapi_spec = json.loads(requests.get("https://bit.ly/serper_dev_spec").text)
service_connector = OpenAPIServiceConnector()
result = service_connector.run(messages=[ChatMessage.from_assistant(json.dumps(fc_payload))],
service_openapi_spec=serperdev_openapi_spec, service_credentials=serper_token)
print(result)
>> {'service_response': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _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."...
OpenAPIServiceConnector.__init__
def __init__(ssl_verify: Optional[Union[bool, str]] = None)
初始化 OpenAPIServiceConnector 实例
参数:
ssl_verify：决定是否对请求使用 SSL 验证，如果传递了一个字符串，则将用作 CA。
OpenAPIServiceConnector.run
@component.output_types(service_response=dict[str, Any])
def run(
messages: list[ChatMessage],
service_openapi_spec: dict[str, Any],
service_credentials: Optional[Union[dict, str]] = None
) -> dict[str, list[ChatMessage]]
处理一个聊天消息列表以调用 OpenAPI 服务上的某个方法。
它解析列表中的最后一条消息，期望它包含工具调用。
参数:
messages：一组ChatMessage对象，包含要处理的消息。最后一条消息应包含工具调用。service_openapi_spec：要调用的服务的 OpenAPI JSON 规范对象。所有引用都应已解析。service_credentials：用于向服务进行身份验证的凭据。目前仅支持 http 和 apiKey OpenAPI 安全方案。
引发:
ValueError：如果最后一条消息不是来自助手，或者它不包含工具调用。
返回值:
包含以下键的字典
service_response：一个包含服务响应的ChatMessage对象列表。响应为 JSON 格式，content属性包含 JSON 字符串。ChatMessage包含 JSON 字符串。
OpenAPIServiceConnector.to_dict
def to_dict() -> dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
OpenAPIServiceConnector.from_dict
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "OpenAPIServiceConnector"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
模块 openapi
OpenAPIConnector
OpenAPIConnector 支持直接调用 OpenAPI 规范中定义的 REST 端点。
OpenAPIConnector 作为 Haystack 管道和任何遵循 OpenAPI（以前称为 Swagger）规范的 REST API 之间的桥梁。它动态解释 API 规范，并提供执行 API 操作的接口。它通常通过从 Haystack 管道的 run 方法传递输入参数给它，或者由管道中的其他组件将输入参数传递给此组件来调用。
示例:
from haystack.utils import Secret
from haystack.components.connectors.openapi import OpenAPIConnector
connector = OpenAPIConnector(
openapi_spec="https://bit.ly/serperdev_openapi",
credentials=Secret.from_env_var("SERPERDEV_API_KEY"),
service_kwargs={"config_factory": my_custom_config_factory}
)
response = connector.run(
operation_id="search",
arguments={"q": "Who was Nikola Tesla?"}
)
注意:
- 该此组件需要
parameters参数。 - 该
service_kwargs参数是可选的，可用于将其他选项传递给 OpenAPIClient。
OpenAPIConnector.__init__
def __init__(openapi_spec: str,
credentials: Optional[Secret] = None,
service_kwargs: Optional[dict[str, Any]] = None)
使用规范和可选凭据初始化 OpenAPIConnector。
参数:
openapi_spec：OpenAPI 规范的 URL、文件路径或原始字符串credentials：可选的 API 密钥或服务凭据，用 Secret 包装service_kwargs：传递给 OpenAPIClient.from_spec() 的其他关键字参数。例如，您可以传递自定义 config_factory 或其他配置选项。
OpenAPIConnector.to_dict
def to_dict() -> dict[str, Any]
将此组件序列化为字典。
OpenAPIConnector.from_dict
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "OpenAPIConnector"
从字典反序列化此组件。
OpenAPIConnector.run
@component.output_types(response=dict[str, Any])
def run(operation_id: str,
arguments: Optional[dict[str, Any]] = None) -> dict[str, Any]
调用 OpenAPI 规范中指定的 REST 端点。
参数:
operation_id：要调用的 OpenAPI 规范中的 operationIdarguments：端点的可选参数（查询、路径或正文参数）
返回值:
包含服务响应的字典