Hayhooks

实现 run_chat_completion_async，通过兼容 OpenAI 的 Chat Completions API公开 Agent。

在 pipeline_wrapper.py 顶部添加以下导入：

from collections.abc import AsyncGenerator
from hayhooks import async_streaming_generator

在 PipelineWrapper 中添加此方法：

    async def run_chat_completion_async(
        self, model: str, messages: list[dict], body: dict
    ) -> AsyncGenerator[str, None]:
        return async_streaming_generator(
            pipeline=self.pipeline,
            pipeline_run_args={
                "messages": [ChatMessage.from_openai_dict_format(m) for m in messages],
            },
        )

重新部署包装器，然后请求流式响应：

hayhooks pipeline deploy-files -n my_agent ./my_agent_dir --overwrite
curl -N http://localhost:1416/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "my_agent",
    "messages": [{"role": "user", "content": "What can you do?"}],
    "stream": true
  }'

响应将以服务器发送事件的形式到达。连接 Open WebUI，或使用嵌入式 Chainlit UI与 Agent 聊天。

在 macOS 或 Linux 上，使用与服务端相同的操作系统账户，在第二个终端中打开仪表板：

pip install "hayhooks[tui]"
hayhooks dashboard

按 e 查看执行记录，或按 v 查看追踪，然后按 Enter 查看详情。默认启用本地捕获。最近的普通运行记录会一直保留，直到服务端重启或缓冲区将其淘汰。

如需在 Windows 上操作或远程访问，请参阅仪表板连接指南。如需将跨度导出到外部 OpenTelemetry 后端，请参阅追踪与 OpenTelemetry。

此示例会在调用演示搜索工具前等待审批。将包装器保存为 pipelines/research_agent/pipeline_wrapper.py。

from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIResponsesChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import tool
from pydantic import BaseModel

from hayhooks import BasePipelineWrapper, DurableContext


@tool
def search(query: str) -> str:
    """Search the knowledge base."""
    return f"Hayhooks serves Haystack pipelines and agents; results for {query!r}."


class Task(BaseModel):
    question: str


class Approval(BaseModel):
    approved: bool


class Report(BaseModel):
    answer: str


class PipelineWrapper(BasePipelineWrapper):
    durable_revision = "research-agent-v1"  # compatibility key for persisted inputs and checkpoints
    durable_resume_model = Approval  # typed input for human-in-the-loop waits

    def setup(self) -> None:
        generator = OpenAIResponsesChatGenerator(model="gpt-4o-mini")
        self.pipeline = Agent(chat_generator=generator, tools=[search])

    async def run_durable_async(self, context: DurableContext, task: Task) -> Report:
        # Persist the wait in Redis so a worker serving this revision can resume it.
        if not context.state.get("approved"):
            decision = context.resume_input  # consumed on first read
            if decision is None:
                await context.suspend({"kind": "approval", "message": f"Research {task.question!r}?"})
            if not Approval.model_validate(decision).approved:
                raise ValueError("research was rejected")
            context.state["approved"] = True

        # Agent state is checkpointed after tool calls: recovery continues the loop.
        result = await context.run_agent_async(messages=[ChatMessage.from_user(task.question)])
        return Report(answer=result["last_message"].text)

如果快速入门服务仍在运行，请将其停止。安装额外依赖，启动 Redis，并以持久化模式提供流水线目录：

pip install "hayhooks[durable]"
docker run -d -p 6379:6379 redis
export OPENAI_API_KEY="your-api-key"
HAYHOOKS_DURABLE_MODE=true hayhooks run --pipelines-dir ./pipelines

在第二个终端中提交运行任务：

curl -i http://localhost:1416/research_agent/run-durable \
  -H 'Content-Type: application/json' -d '{"question": "What does Hayhooks do?"}'

服务端会返回 202 Accepted、执行 ID 和链接。将下方的 EXECUTION_ID 替换为该 ID，然后检查运行状态：

curl http://localhost:1416/research_agent/executions/EXECUTION_ID

状态变为 waiting 后，你可以测试恢复功能：停止服务端，然后使用相同命令重新启动。等待状态会在重启后保留。

恢复需要有工作进程提供相同的修订版本。部署不兼容的更改时，请让旧修订版本继续运行，直到其任务完成。请参阅修订版本兼容性。

批准运行任务并跟踪其输出：

curl -X POST http://localhost:1416/research_agent/executions/EXECUTION_ID/resume \
  -H 'Content-Type: application/json' -d '{"approved": true}'

curl -N http://localhost:1416/research_agent/executions/EXECUTION_ID/stream