Hayhooks

Hayhooks automates several data handling tasks for deployed pipelines:

• Validation: Uses Pydantic models to validate incoming requests.
• Serialization: Handles JSON serialization for all responses.
• File Uploads: Supports multipart/form-data requests for handling file uploads.
• Type Conversion: Automatically converts between JSON and Python types.

Hayhooks offers two primary ways to deploy Haystack pipelines and agents depending on your requirements for flexibility and complexity:

1. PipelineWrapper Deployment (Recommended): Use this for complex logic. It involves creating a class that encapsulates your pipeline. This method supports:

   • Custom initialization and execution logic.
   • OpenAI-compatible chat endpoints.
   • Streaming capabilities.
   • File upload handling.

2. YAML Pipeline Deployment: Use this for straightforward pipelines. You deploy directly from YAML definitions. This method provides:

   • Automatic request/response schema generation.
   • No wrapper code required.
   • Limitations: Does NOT support OpenAI-compatible chat endpoints or streaming. The YAML must include inputs and outputs sections with properly defined pipeline components.

YAML-deployed pipelines are simplified and do not support all Hayhooks features. If you require the following, you must use a PipelineWrapper instead:

• No OpenAI Compatibility: Does not support OpenAI-compatible chat endpoints.
• No Streaming: Streaming responses are not supported.
• No File Uploads: File upload handling is unavailable.
• Async Only: Pipelines are executed using Pipeline.run_async().

By default, Hayhooks wraps responses in a Pydantic JSON object. However, if your run_api method returns a FastAPI Response object (such as FileResponse, StreamingResponse, or a plain Response), Hayhooks will bypass the JSON wrapper and pass the object directly to the client. This allows you to serve binary content like images, PDFs, or audio files directly from your pipeline.

# If your run_api returns a FastAPI Response, Hayhooks passes it directly
from fastapi.responses import FileResponse

# Inside your pipeline wrapper's run_api method:
return FileResponse("path/to/image.png")

Hayhooks allows agents to trigger custom UI components in the Chainlit interface by emitting custom_element events.

1. Emission: The pipeline wrapper calls create_custom_element_event("ComponentName", props) to send an event over the SSE stream.
2. Rendering: The Chainlit app receives the event and renders a cl.CustomElement(name="ComponentName", props=...).
3. Resolution: Hayhooks resolves these components by copying .jsx files from the directory specified by --chainlit-custom-elements-dir (or the HAYHOOKS_CHAINLIT_CUSTOM_ELEMENTS_DIR environment variable) into the Chainlit public/elements/ directory at startup.

Note: custom_element events are ignored by Open WebUI, but status and notification events will still render normally.

Both streaming_generator and async_streaming_generator accept an external_event_queue parameter. This allows you to inject custom events (such as dict, OpenWebUIEvent, str, or StreamingChunk) into the stream alongside standard pipeline chunks.

Typical Use Case: Passing the queue to pipeline components to implement Human-in-the-Loop (HITL) patterns, where a component can push an approval request event to the queue while waiting for user input.

The Chainlit UI is mounted as a FastAPI sub-application. It communicates with your pipelines through Hayhooks' OpenAI-compatible endpoints (/v1/chat/completions).

For the integration to work:

1. Your pipelines must implement run_chat_completion or run_chat_completion_async.
2. The UI automatically discovers available pipelines via the /v1/models endpoint.
3. Streaming is supported out-of-the-box if your pipeline returns a Generator or AsyncGenerator (ideally using streaming_generator or async_streaming_generator).

I/O-bound pipelines (waiting on LLM APIs, databases, or HTTP requests) should use asynchronous methods to maximize throughput. Implement run_api_async() or run_chat_completion_async() in your BasePipelineWrapper. Scaling horizontally (adding more replicas/containers) is more effective than adding more workers to a single instance for I/O-bound workloads.

from hayhooks import BasePipelineWrapper, Pipeline

class PipelineWrapper(BasePipelineWrapper):
    def setup(self) -> None:
        self.pipeline = Pipeline()
        # ... build pipeline ...

    async def run_api_async(self, query: str) -> str:
        result = await self.pipeline.run_async({"prompt": {"query": query}})
        return result["llm"]["replies"][0]