はじめに
ノートブックで実行されているエージェント ≠ 実稼働環境で実行されているエージェント。この記事では、ラップ エージェントから API → コンテナ化 → デプロイ → スケールまでのパイプライン全体について説明します。
1. FastAPI ラッパー
from fastapi import FastAPI, WebSocket
from pydantic import BaseModel
app = FastAPI()
class AgentRequest(BaseModel):
message: str
session_id: str = None
@app.post("/chat")
async def chat(request: AgentRequest):
agent = get_or_create_agent(request.session_id)
response = await agent.run(request.message)
return {"response": response, "session_id": agent.session_id}
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
await websocket.accept()
agent = create_agent()
while True:
data = await websocket.receive_text()
async for chunk in agent.stream(data):
await websocket.send_text(chunk)
2. ドッカー
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
3. スケーリングに関する考慮事項
- ステートフル エージェントのセッション アフィニティ
- 共有状態の Redis
- 長時間実行タスクのキューベースの処理
- ユーザー/セッションあたりのコスト予算
概要
- FastAPI + WebSocket によるリアルタイム エージェント API
- 再現可能なデプロイメントのための Docker
- クラウド展開: AWS ECS、GCP Cloud Run、鉄道
- セッション管理とスケーリングが重要な課題
演習
- SimpleAgent を FastAPI アプリにラップする
- Docker化してローカルでテストする
- WebSocketストリーミングの実装
- Railway または Cloud Run へのデプロイ