文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
文档

Hayhooks

Hayhooks 是一款您可用于通过 HTTP 端点服务 Haystack 管道的 Web 应用程序。本页概述了 Hayhooks 的主要功能。

建议编辑

📘

Hayhooks GitHub

您可以在 Hayhooks GitHub 仓库 中找到代码和对功能的深入解释。

概述

Hayhooks 简化了将 Haystack 管道作为 REST API 部署的过程。它允许您：

  • 将 Haystack 管道公开为 HTTP 端点，包括兼容 OpenAI 的聊天端点，
  • 在保持最少样板代码的同时自定义逻辑，
  • 快速高效地部署管道。

安装

使用 pip 安装 Hayhooks

pip install hayhooks

该 hayhooks 包同时提供了服务器和客户端组件，客户端能够启动服务器。在 shell 中，使用以下命令启动服务器：

$ hayhooks run
INFO:     Started server process [44782]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on <https://:1416> (Press CTRL+C to quit)

检查状态

在另一个 shell 中，您可以使用以下命令查询服务器的状态：

$ hayhooks status
Hayhooks server is up and running.

配置

Hayhooks 可以通过三种方式配置：

  1. 使用项目根目录中的.env 文件。
  2. 运行命令时传递环境变量。
  3. 使用命令行参数，通过hayhooks run.

环境变量

变量描述
HAYHOOKS_HOST服务器的主机地址
HAYHOOKS_PORT服务器的端口
HAYHOOKS_PIPELINES_DIR包含管道的目录
HAYHOOKS_ROOT_PATH服务器的根路径
HAYHOOKS_ADDITIONAL_PYTHON_PATH要包含的附加 Python 路径
HAYHOOKS_DISABLE_SSL禁用 SSL 验证（布尔值）
HAYHOOKS_SHOW_TRACEBACKS显示错误堆栈跟踪（布尔值）

CORS 设置

变量描述
HAYHOOKS_CORS_ALLOW_ORIGINS允许的源列表（默认[*])
HAYHOOKS_CORS_ALLOW_METHODS允许的 HTTP 方法列表（默认[*])
HAYHOOKS_CORS_ALLOW_HEADERS允许的头部列表（默认[*])
HAYHOOKS_CORS_ALLOW_CREDENTIALS允许凭证（默认false)
HAYHOOKS_CORS_ALLOW_ORIGIN_REGEX允许源的正则表达式模式（默认null)
HAYHOOKS_CORS_EXPOSE_HEADERS在响应中公开的头部（默认[])
HAYHOOKS_CORS_MAX_AGE预检响应的最大年龄（默认600)

运行 Hayhooks

要启动服务器，请执行

hayhooks run

这将会在以下地址启动 Hayhooks：HAYHOOKS_HOST:HAYHOOKS_PORT.

部署管道

步骤

  1. 准备管道定义（.yml 文件）和pipeline_wrapper.py 文件。

  2. 部署管道

    hayhooks pipeline deploy-files -n my_pipeline my_pipeline_dir

  3. 通过以下端点访问管道：{pipeline_name}/run。

管道包装器

一个需要 PipelineWrapper 类来包装管道。

from pathlib import Path
from haystack import Pipeline
from hayhooks import BasePipelineWrapper

class PipelineWrapper(BasePipelineWrapper):
    def setup(self) -> None:
        pipeline_yaml = (Path(__file__).parent / "pipeline.yml").read_text()
        self.pipeline = Pipeline.loads(pipeline_yaml)

    def run_api(self, input_text: str) -> str:
        result = self.pipeline.run({"input": {"text": input_text}})
        return result["output"]["text"]

文件上传

Hayhooks 使您能够在管道包装器的run_api 方法中通过包含files: Optional[List[UploadFile]] = None 作为参数来处理文件上传。

def run_api(self, files: Optional[List[UploadFile]] = None) -> str:
    if files and len(files) > 0:
        filenames = [f.filename for f in files if f.filename is not None]
        file_contents = [f.file.read() for f in files]
        return f"Received files: {', '.join(filenames)}"
    return "No files received"

Hayhooks 会自动处理上传的文件，并在存在时将其传递给run_api 方法。HTTP 请求必须是multipart/form-data 请求。

组合文件和参数

Hayhooks 还支持在同一请求中处理文件和其他参数，方法是将它们作为参数包含在run_api:

def run_api(self, files: Optional[List[UploadFile]] = None, additional_param: str = "default") -> str:
    ...

从 CLI 运行管道

使用 JSON 兼容的参数

您可以使用hayhooks pipeline run 命令通过命令行执行管道。内部而言，这将触发管道包装器的run_api 方法，并将参数作为 JSON payload 传递。

此方法非常适合在不编写额外代码的情况下从 CLI 测试已部署的管道。

hayhooks pipeline run <pipeline_name> --param 'question="Is this recipe vegan?"'

使用文件上传

要执行需要文件输入的管道，请使用multipart/form-data 请求。您可以在同一请求中同时提交文件和参数。

确保已部署的管道支持文件处理。

# Upload a directory
hayhooks pipeline run <pipeline_name> --dir files_to_index

# Upload a single file
hayhooks pipeline run <pipeline_name> --file file.pdf

# Upload multiple files
hayhooks pipeline run <pipeline_name> --dir files_to_index --file file1.pdf --file file2.pdf

# Upload a file with an additional parameter
hayhooks pipeline run <pipeline_name> --file file.pdf --param 'question="Is this recipe vegan?"'

MCP 支持

MCP 服务器

Hayhooks 支持模型上下文协议 (MCP)，并可以充当 MCP 服务器。它会自动将您已部署的管道列为 MCP 工具，并使用 Server-Sent Events (SSE) 作为传输方法。

要启动 Hayhooks MCP 服务器，请运行

hayhooks mcp run

这将会在以下地址启动服务器：HAYHOOKS_MCP_HOST:HAYHOOKS_MCP_PORT.

创建 PipelineWrapper

要将 Haystack 管道公开为 MCP 工具，您需要一个具有以下属性的PipelineWrapper：

  • name：工具的名称
  • description：工具的描述
  • inputSchema：工具输入参数的 JSON Schema 对象

对于每个已部署的管道，Hayhooks 将：

  1. 使用管道包装器名称作为 MCP 工具名称，
  2. 使用run_api 方法的 docstring 作为 MCP 工具描述（如果存在），
  3. 从run_api 方法参数生成 Pydantic 模型。

PipelineWrapper 示例

from pathlib import Path
from typing import List
from haystack import Pipeline
from hayhooks import BasePipelineWrapper

class PipelineWrapper(BasePipelineWrapper):
    def setup(self) -> None:
        pipeline_yaml = (Path(__file__).parent / "chat_with_website.yml").read_text()
        self.pipeline = Pipeline.loads(pipeline_yaml)

    def run_api(self, urls: List[str], question: str) -> str:
        """
        Ask a question about one or more websites using a Haystack pipeline.
        """
        result = self.pipeline.run({"fetcher": {"urls": urls}, "prompt": {"query": question}})
        return result["llm"]["replies"][0]

跳过 MCP 工具列出

要部署一个管道而不将其列为 MCP 工具，请在您的类中设置skip_mcp = True。

class PipelineWrapper(BasePipelineWrapper):
    # This will skip the MCP Tool listing
    skip_mcp = True
    
    def setup(self) -> None:
        ...
    
    def run_api(self, urls: List[str], question: str) -> str:
        ...

OpenAI 兼容性

Hayhooks 通过run_chat_completion 方法支持兼容 OpenAI 的端点。

from hayhooks import BasePipelineWrapper, get_last_user_message

class PipelineWrapper(BasePipelineWrapper):
    def run_chat_completion(self, model: str, messages: list, body: dict):
        question = get_last_user_message(messages)
        return self.pipeline.run({"query": question})

流式响应

Hayhooks 提供了一个streaming_generator 工具，用于将管道输出流式传输到客户端。

from hayhooks import streaming_generator

def run_chat_completion(self, model: str, messages: list, body: dict):
    question = get_last_user_message(messages)
    return streaming_generator(pipeline=self.pipeline, pipeline_run_args={"query": question})

以编程方式运行

Hayhooks 可以嵌入到 FastAPI 应用程序中。

import uvicorn
from hayhooks.settings import settings
from fastapi import Request
from hayhooks import create_app

# Create the Hayhooks app
hayhooks = create_app()

# Add a custom route
@hayhooks.get("/custom")
async def custom_route():
    return {"message": "Hi, this is a custom route!"}

# Add a custom middleware
@hayhooks.middleware("http")
async def custom_middleware(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Custom-Header"] = "custom-header-value"
    return response

if __name__ == "__main__":
    uvicorn.run("app:hayhooks", host=settings.host, port=settings.port)

更新于 4 个月前