Hayhooks

run_chat_completion_async を実装して、OpenAI 互換の Chat Completions API を通じてエージェントを公開します。

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 を使ってエージェントとチャットします。

macOS または Linux では、サーバーと同じ OS アカウントで、2 つ目のターミナルからダッシュボードを開きます。

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

2 つ目のターミナルで、実行を送信します。

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

サーバーは実行 ID とリンクを含む 202 Accepted を返します。以下の 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