ローカルで動くAIエージェントを作るところまでは、もう多くの開発者がたどり着いています。ところが「これを本番で安定稼働させてくれ」と言われた瞬間、話は一気に難しくなります。エンドポイント化、状態管理、スケール、シークレット、監視——どれも開発時には見えていなかった壁です。
実際にクライアントのAIエージェントを本番デプロイした際、最初に詰まったのは「同期で返すか、ストリーミングで返すか、ジョブにするか」という出口設計でした。ここを最初に決めずにコードを書き始めると、後から全面的な作り直しになります。
この記事では、手元のエージェントを本番でサービングするための実装パターンを、FastAPIの最小コードから監視・コスト管理まで、コピペ可能なコードつきで解説します。記載した数値・仕様はすべて2026年6月時点の公式ドキュメントで確認したものです。
まず試したい「最小デプロイ」セットアップ3選
本番デプロイの全体像は後述しますが、まず手を動かして「エージェントが外からAPIで叩ける」状態を作りましょう。以下の3つはどれもコピペで動きます。
セットアップ1：FastAPIでエージェントをエンドポイント化する（同期）
もっとも単純な出口は、リクエストを受けてエージェントの最終回答を1回で返す同期エンドポイントです。クライアントから1回叩いて1回返す、いわゆるリクエスト・レスポンス型です。
# 動作環境: Python 3.11+, fastapi>=0.110, uvicorn[standard]>=0.29
# 必要パッケージ: pip install "fastapi" "uvicorn[standard]"
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class AgentRequest(BaseModel):
message: str
session_id: str | None = None
class AgentResponse(BaseModel):
answer: str
# run_agent は手元のエージェント実装に差し替える
def run_agent(message: str, session_id: str | None) -> str:
# ここで LLM 呼び出し・ツール実行などを行う
return f"echo: {message}"
@app.post("/agent/invoke", response_model=AgentResponse)
def invoke(req: AgentRequest) -> AgentResponse:
answer = run_agent(req.message, req.session_id)
return AgentResponse(answer=answer)
@app.get("/healthz")
def healthz() -> dict[str, str]:
return {"status": "ok"}
ローカルでの起動はこれだけです。
uvicorn main:app --host 0.0.0.0 --port 8000
ポイント
/healthzのようなヘルスチェック用エンドポイントを最初から用意する。コンテナやロードバランサが「このプロセスは生きているか」を判定するために使います。- リクエスト・レスポンスを
BaseModelで定義しておくと、入力バリデーションが自動で効き、不正な入力を早期に弾けます。 - 同期型はLLMの応答に数秒かかると、その間クライアントを待たせます。体感を上げたい場合は次のストリーミングを使います。
セットアップ2：トークンを逐次返すストリーミング
LLMの出力を1文字ずつ流すと、ユーザーは「考えている途中」を見られるので体感速度が大きく改善します。FastAPIでは StreamingResponse を使います。
# 動作環境: Python 3.11+, fastapi>=0.110
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
app = FastAPI()
async def token_stream(message: str):
# 実際には LLM の streaming API から chunk を受け取る
for token in ["こ", "ん", "に", "ち", "は"]:
yield token
await asyncio.sleep(0.05)
@app.post("/agent/stream")
async def stream(message: str):
return StreamingResponse(
token_stream(message),
media_type="text/plain; charset=utf-8",
)
ポイント
- ジェネレータ関数で
yieldした内容が順次クライアントに流れます。LLMのstreaming APIから受け取ったchunkをそのままyieldすれば、最終回答を待たずに表示できます。 - ブラウザ向けにイベント形式で流したい場合はServer-Sent Events（SSE）を使います。SSE/リアルタイム配信の具体的な実装はLLMエージェントのストリーミング配信ガイドで詳しく扱っています。
セットアップ3：時間のかかる処理を非同期ジョブにする
複数ツールを連鎖させる複雑なエージェントは、応答に数十秒〜数分かかることがあります。同期で待たせるとタイムアウトするため、「受付IDだけ即返して、結果は後から取りに来てもらう」非同期ジョブ型にします。
# 動作環境: Python 3.11+, fastapi>=0.110
# 注意: これはインメモリ実装。本番では Redis 等の外部ストアを使うこと。
from fastapi import FastAPI, BackgroundTasks
from uuid import uuid4
app = FastAPI()
JOBS: dict[str, dict] = {} # 本番では外部ストアに置き換える
def run_long_agent(job_id: str, message: str) -> None:
result = run_agent(message, None) # 重い処理
JOBS[job_id] = {"status": "done", "result": result}
@app.post("/agent/jobs")
def create_job(message: str, bg: BackgroundTasks) -> dict[str, str]:
job_id = str(uuid4())
JOBS[job_id] = {"status": "running", "result": None}
bg.add_task(run_long_agent, job_id, message)
return {"job_id": job_id, "status": "running"}
@app.get("/agent/jobs/{job_id}")
def get_job(job_id: str) -> dict:
return JOBS.get(job_id, {"status": "not_found"})
ポイント
- FastAPIの公式ドキュメントによれば、
BackgroundTasksはレスポンス送信後に同一プロセスで実行されます（FastAPI公式・2026年6月時点）。手軽ですが、プロセスが落ちるとジョブも消えます。 - 永続化・リトライ・水平スケールが必要になったら、CeleryやRQのような外部ジョブキューに移行します。判断軸は「失敗したときに作り直せないと困るか」です。
- 上の
JOBS辞書はあくまで動作確認用です。プロセスを複数に増やすと共有されないため、本番ではRedisなどの外部ストアに置き換えます（理由は次章で解説します）。
本番サービングは「3つの出口」で設計する
エージェントをどう外に出すかは、用途によって最適解が変わります。最初にこの3つのどれにするかを決めると、後の実装がぶれません。
| 出口 | 向いている用途 | 応答時間の目安 | 実装 |
|---|---|---|---|
| 同期（invoke） | 短い回答・分類・抽出 | 数秒以内 | 通常のPOSTエンドポイント |
| ストリーミング（stream） | チャット・長文生成 | 体感を改善したい時 | StreamingResponse / SSE |
| 非同期ジョブ（jobs） | 多段ツール実行・バッチ | 数十秒〜数分 | ジョブ投入＋結果ポーリング |
1つのエージェントが全部を兼ねる必要はありません。たとえば「短い質問は同期、チャットUIはストリーミング、レポート生成は非同期ジョブ」と用途別に出口を分けるのが現実的です。設計の全体像をもっと体系的に押さえたい場合はAIエージェント実装の完全ロードマップを併せて参照してください。
状態・セッション・会話履歴の持ち方
本番デプロイで最も事故が起きやすいのが状態管理です。理由は、本番ではプロセスが複数に増えるからです。
FastAPI公式の本番デプロイ概念ページでは、複数のワーカープロセスを並べて負荷を分散する構成が推奨されています。そして重要なのは、各プロセスはメモリを共有しないという点です。「もしモデルが1GBなら、4ワーカーで合計4GBのRAMが必要になる」と明記されています（FastAPI公式・2026年6月時点）。
これは会話履歴にも同じことが言えます。プロセス内の変数に会話履歴を貯めると、リクエストごとに別のワーカーに振り分けられた瞬間、履歴が消えたように見えます。そこで本番では「アプリ自体はステートレスにして、状態は外部ストアに置く」のが原則です。
- セッションIDを発行する：クライアントから毎回
session_idを受け取る設計にする。 - 会話履歴は外部ストアに保存する：Redis（高速・揮発可）やPostgreSQL（永続・検索可）に
session_idをキーとして保存する。 - リクエストごとに履歴を読み出す：エージェント実行前にストアから履歴を読み、実行後に追記する。
- 有効期限（TTL）を設定する：会話履歴に期限を付けてストアが無限に肥大化しないようにする。
# 動作環境: Python 3.11+, redis>=5.0
# 必要パッケージ: pip install redis
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
import json
import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
def load_history(session_id: str) -> list[dict]:
raw = r.get(f"history:{session_id}")
return json.loads(raw) if raw else []
def save_history(session_id: str, history: list[dict]) -> None:
# 24時間で自動失効（TTL=86400秒）
r.set(f"history:{session_id}", json.dumps(history), ex=86400)
ポイント
- アプリをステートレスにしておくと、ワーカーを何個に増やしても破綻しません。スケールの前提条件です。
- 会話履歴が長くなるとトークン数とコストが増えます。古い履歴の要約・圧縮の手法はコンテキストエンジニアリングのガイドで詳しく解説しています。
スケールと並行処理
状態を外に出したら、次はプロセスを増やしてさばける量を上げます。FastAPIはUvicornのワーカー機能で複数プロセスを並べられます。
# 4ワーカーで起動（マネージャプロセスが1つのポートを受けて分配する）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
FastAPI公式の概念ページによれば、1つのマネージャプロセスがポートを監視し、複数のUvicornワーカープロセスにリクエストを分配します。CPUとRAMの使用率は50〜90%を目安にし、低すぎれば無駄、高すぎればクラッシュやスローダウンのリスクがあると説明されています（FastAPI公式・2026年6月時点）。
並行処理で気をつけるべき設計上の論点を整理します。
- タイムアウト：LLM呼び出しや外部ツールには必ずタイムアウトを設定する。設定しないと1リクエストがワーカーを長時間占有し、他のリクエストがさばけなくなります。
- リトライと冪等性：ネットワークの一時障害でリトライする際、同じ処理を二重実行しないよう冪等性を持たせる。具体的な実装はエージェントのリトライ・冪等性・タイムアウト設計ガイドを参照してください。
- キュー：突発的なアクセス集中に備え、重い処理はジョブキューに逃がしてピークを平準化する。
- ワーカー数とメモリ：前述のとおりワーカーごとにメモリが乗算されるため、ワーカー数はサーバーのRAMから逆算する。
デプロイ先の選び方（コンテナ／サーバーレス／マネージド）
どこに置くかは、エージェントの応答時間とGPU要件で決まります。AIエージェントは長時間実行やGPU推論が絡むため、一般的なWeb APIとは選び方が違います。
| 選択肢 | 向いている条件 | 注意点 |
|---|---|---|
| コンテナ（自前運用・k8s等） | 常時稼働・カスタマイズ自由度が必要 | 運用負荷が高い |
| マネージドコンテナ（Cloud Run等） | スケールゼロしたい・GPUも使いたい | 実行時間に上限がある |
| サーバーレス関数（Lambda等） | 短時間・軽量・イベント駆動 | 実行時間とGPUに制約あり |
| マネージドAgentサービス（LangGraph Platform等） | 永続化・長時間タスクを自前で作りたくない | プラットフォーム依存 |
具体的な制約を公式情報で確認しておきましょう。AWS Lambdaは関数の最大実行時間が15分という固定上限があります（AWS公式ドキュメント・2026年6月時点）。一方、Google Cloud Runはスケールゼロに対応し、GPU（NVIDIA L4）も利用できます（Google Cloud公式・2026年6月時点）。多段ツールを呼ぶ重いエージェントで15分を超える可能性があるなら、実行時間に余裕のあるコンテナ系を選ぶのが無難です。
「サービング基盤そのものを作りたくない」場合は、マネージドなAgentサービスという選択肢もあります。LangChainは2026年時点で、新規プロジェクトに対し従来のLangServeではなくLangGraph Platformを推奨しており、永続化・メモリ・長時間タスク・ストリーミングなどを標準で備えるとしています（LangChain公式・2026年6月時点）。自前でステートストアやジョブ管理を組むコストと、プラットフォーム依存のトレードオフを天秤にかけて選びます。
シークレット・APIキー管理と環境分離
本番デプロイでAPIキーをコードに直書きするのは、最も避けるべきパターンです。OWASP Top 10 for LLM Applicationsでも、機密情報の漏えいや過剰な権限付与はリスクとして整理されています（OWASP公式・2026年6月時点）。
- キーは環境変数または専用のシークレットマネージャから読む：コードにもGitにも書かない。The Twelve-Factor Appが説く「設定は環境に持たせる」原則です（12factor.net・2026年6月時点）。
- 開発・ステージング・本番でキーを分ける：環境ごとに別のAPIキーを使い、本番キーが開発環境に混入しないようにする。
- マネージドなシークレットストアを使う：AWS Secrets Manager、Google Secret Managerなどを使うと、ローテーションや権限管理が楽になる。
- 最小権限の原則：エージェントが使うキー・トークンには必要最小限の権限だけを与える。
# 動作環境: Python 3.11+
# 注意: APIキーはコードにもGitにも書かない。環境変数から読む。
import os
def get_api_key() -> str:
key = os.environ.get("LLM_API_KEY")
if not key:
# 起動時に欠落を検知して落とす（本番で気付かず動くより安全）
raise RuntimeError("LLM_API_KEY が設定されていません")
return key
ポイント
- キーが無いまま起動して途中で500を返すより、起動時に明示的に落とした方が事故に早く気付けます。
- 入力検証も忘れずに。ユーザー入力をそのままプロンプトに渡すとプロンプトインジェクションの余地が生まれます。入力の長さ制限・バリデーションは最低限入れましょう。
監視・ログ・コスト管理の最低限
デプロイは「公開して終わり」ではありません。本番では「いま壊れていないか」「いくらかかっているか」が見えていないと運用できません。最低限そろえるべきものを挙げます。
- 構造化ログ：リクエストID・session_id・処理時間・使用トークン数をJSON形式で出す。あとから問題を追跡できるようにします。
- ヘルスチェック：前述の
/healthzを監視対象にし、落ちたら自動で再起動・通知する。 - レイテンシとエラー率：応答時間の分布（中央値・95パーセンタイル）とエラー率を継続的に記録する。
- トークンとコスト：1リクエストあたりの入力・出力トークンを記録し、日次でコストを集計する。会話履歴が長い設計だとここが想定外に膨らみます。
- レート制限：濫用や暴走ループでコストが青天井にならないよう、利用者ごとのレート制限を設ける。
# 動作環境: Python 3.11+
# リクエストごとに構造化ログを出すミドルウェアの最小例
import json
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
elapsed_ms = (time.perf_counter() - start) * 1000
log = {
"path": request.url.path,
"status": response.status_code,
"elapsed_ms": round(elapsed_ms, 1),
}
print(json.dumps(log, ensure_ascii=False))
return response
【要注意】よくある失敗パターンと回避策
失敗1：アプリ内変数に状態を持ったまま複数ワーカーで動かす
❌ 会話履歴やジョブ情報をプロセス内の辞書に貯める
⭕ 状態は外部ストア（Redis/DB）に出し、アプリはステートレスにする
なぜ重要か：本番ではワーカーが複数に増え、リクエストごとに別プロセスに振り分けられます。プロセス内に貯めた状態は共有されないため、「たまに履歴が消える」という再現性の低いバグになります。FastAPI公式も各プロセスがメモリを共有しないことを明記しています。
失敗2：タイムアウトを設定しない
❌ LLM呼び出しや外部APIをタイムアウトなしで呼ぶ
⭕ すべての外部呼び出しにタイムアウトを設定し、超過時はエラーを返す
なぜ重要か：応答が返ってこない1リクエストがワーカーを占有し続けると、そのワーカーが処理できる他のリクエストも巻き添えで詰まります。少数の遅いリクエストがサービス全体を止める典型的なパターンです。
失敗3：実行時間の制約を確認せずデプロイ先を選ぶ
❌ 多段ツールを呼ぶ重いエージェントをサーバーレス関数に置く
⭕ 想定実行時間とデプロイ先の上限（Lambdaは15分など）を照合してから選ぶ
なぜ重要か：開発時は数秒で終わっていた処理も、本番で外部ツールが増えると数分かかることがあります。実行時間の上限を超えると、本番で初めてタイムアウトが多発します。
まとめ：今日から始める3つのアクション
- 今日：手元のエージェントを「セットアップ1」のFastAPI同期エンドポイントでラップし、
/healthzをつけて外から叩けるようにする。 - 今週中：会話履歴をアプリ内変数からRedis等の外部ストアに移し、アプリをステートレスにする。
- 今月中：想定実行時間とGPU要件からデプロイ先を選び、構造化ログとレート制限を入れて本番に出す。
この記事を読んで導入イメージが固まってきた方へ
UravationではAIエージェント導入の研修・コンサルを行っています。
参考・出典
- FastAPI 公式ドキュメント — Deployment Concepts（ワーカープロセス・メモリ・再起動）（2026年6月時点）
- FastAPI 公式ドキュメント — Background Tasks（2026年6月時点）
- AWS Lambda 公式ドキュメント — Quotas（実行時間15分上限）（2026年6月時点）
- Google Cloud Run 公式ドキュメント — GPU configuration（2026年6月時点）
- LangChain 公式 — LangGraph / LangGraph Platform（2026年6月時点）
- The Twelve-Factor App — Config（設定を環境に持たせる）（2026年6月時点）
- OWASP Top 10 for LLM Applications（2026年6月時点）
この記事はAIgent Lab編集部がお届けしました。
