なぜAIエージェントの「制御できなさ」が本番の最大リスクなのか
Google ADKでエージェントを構築して、ローカルで動作確認できました。テストも通った。でも本番にデプロイしてしばらくすると、予期しない問題が発生する——これはAIエージェント開発者が必ずぶつかる壁です。
具体的にはこんなシーンです。
ユーザーが「競合他社の価格を調べて」と依頼したところ、エージェントが競合他社の内部システムへの不正アクセスを試みるツール呼び出しを生成しました。あるいは、コスト試算ツールとして作ったはずのエージェントが、ループし続けて1回のリクエストで数千円分のAPIトークンを消費しました。さらに悪いケースでは、カスタマーサポートエージェントが「返金します」と勝手に約束してしまった。
これらはすべて実際の本番障害として報告されているパターンです。Gemini APIの品質は年々向上していますが、LLMの確率的な性質上、エッジケースでの予期しない挙動を完全になくすことはできません。
Google ADKは、この問題に対してCallbacksとGuardrailsという2つの仕組みを提供しています。Callbacksはエージェントの処理ステップごとにカスタム関数を差し込むフック、Guardrailsはインプット・アウトプットを自動検査・フィルタリングする安全層です。この2つを組み合わせることで、「エージェントが何をしようとしているか」を常に把握し、「してはいけないことを物理的に止める」仕組みが実現できます。
Google ADK Callbacksの基本構造
Callbacksが差し込めるタイミング
ADKのCallbacksは、エージェントの処理ライフサイクルの特定のタイミングで呼ばれます。2026年4月時点のADK v1.xでは、以下のCallbackポイントが利用可能です。
エージェントレベル
before_agent_callback: エージェントの処理開始前after_agent_callback: エージェントの処理完了後
モデル呼び出しレベル
before_model_callback: Geminiモデルへのリクエスト送信前after_model_callback: Geminiモデルのレスポンス受信後
ツール呼び出しレベル
before_tool_callback: ツール実行前after_tool_callback: ツール実行後
これらのCallbackポイントを使えば、エージェントの実行フロー全体を把握・制御できます。
Callbackの基本実装パターン
まずCallbacksを使ったシンプルなロギング実装から始めましょう。
import time
import logging
from typing import Any, Optional
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import FunctionTool
from google.genai import types
# ロガーの設定
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("adk.monitoring")
# ===== Before Model Callback =====
def log_before_model(
callback_context: Any,
llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
"""
モデル呼び出し前にリクエスト内容をログに記録する。
Noneを返すと処理を続行、GenerateContentResponseを返すと
モデル呼び出しをスキップしてそのレスポンスを使用する。
"""
# リクエストのトークン数を推定（実際には count_tokens APIを使う）
contents_preview = str(llm_request.contents)[:200]
logger.info(
f"[BEFORE_MODEL] agent={callback_context.agent_name} "
f"contents_preview={contents_preview}"
)
# セッションに開始時刻を記録（after_model_callbackで経過時間を計算）
callback_context.state["_model_call_start"] = time.time()
return None # Noneを返すと通常通りモデルが呼ばれる
# ===== After Model Callback =====
def log_after_model(
callback_context: Any,
llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
"""
モデルレスポンス受信後にレイテンシと出力をログに記録する。
"""
start_time = callback_context.state.get("_model_call_start", time.time())
elapsed_ms = int((time.time() - start_time) * 1000)
# 使用トークン数
usage = llm_response.usage_metadata
prompt_tokens = usage.prompt_token_count if usage else 0
output_tokens = usage.candidates_token_count if usage else 0
logger.info(
f"[AFTER_MODEL] agent={callback_context.agent_name} "
f"latency_ms={elapsed_ms} "
f"prompt_tokens={prompt_tokens} "
f"output_tokens={output_tokens}"
)
return None # Noneを返すとレスポンスをそのまま使用
# エージェントにCallbacksを設定
agent = LlmAgent(
name="monitored_agent",
model="gemini-2.5-pro",
instruction="あなたは役立つアシスタントです。",
before_model_callback=log_before_model,
after_model_callback=log_after_model,
)
# 期待する出力例:
# 2026-04-14 10:45:01 [INFO] adk.monitoring: [BEFORE_MODEL] agent=monitored_agent contents_preview=...
# 2026-04-14 10:45:03 [INFO] adk.monitoring: [AFTER_MODEL] agent=monitored_agent latency_ms=1842 prompt_tokens=523 output_tokens=287このコードで重要なのは、before_model_callbackとafter_model_callbackの戻り値の意味です。Noneを返すと処理が通常通り続行されます。一方、before_model_callbackでGenerateContentResponseを返すと、実際のモデル呼び出しがスキップされ、そのレスポンスが使われます。これがGuardrailsの実装の核心になります。
Guardrailsによる安全制御の実装
インプットGuardrail：有害なリクエストをブロックする
import re
from google.genai import types
# ブロックするパターン（実際はより精緻なパターンを使う）
BLOCKED_PATTERNS = [
r"パスワード|password|credential",
r"不正アクセス|unauthorized.*access|hack",
r"個人情報.*削除|delete.*personal.*data",
r"返金|refund|compensation", # 承認なしに確約してはいけない言葉
]
SENSITIVE_DOMAINS = [
"internal.company.com",
"admin.myservice.com",
"*.secret-api.com",
]
def input_guardrail(
callback_context: Any,
llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
"""
リクエストを分析し、危険なパターンを含む場合はモデル呼び出しを
ブロックして安全なレスポンスを返す。
これは before_model_callback として設定する。
"""
# ユーザーメッセージを抽出
user_messages = []
for content in llm_request.contents:
if content.role == "user":
for part in content.parts:
if hasattr(part, "text") and part.text:
user_messages.append(part.text)
full_user_input = " ".join(user_messages).lower()
# パターンマッチング
for pattern in BLOCKED_PATTERNS:
if re.search(pattern, full_user_input, re.IGNORECASE):
logger.warning(
f"[GUARDRAIL_BLOCKED] agent={callback_context.agent_name} "
f"pattern={pattern} "
f"input_preview={full_user_input[:100]}"
)
# ブロック: モデルを呼ばずに安全なレスポンスを返す
return types.GenerateContentResponse(
candidates=[
types.Candidate(
content=types.Content(
role="model",
parts=[types.Part(
text="申し訳ありませんが、そのリクエストにはお応えできません。"
"ご不明な点があれば、担当者にお問い合わせください。"
)]
),
finish_reason=types.FinishReason.STOP
)
]
)
return None # 問題なければ通常通り処理
# エージェントに設定
safe_agent = LlmAgent(
name="safe_customer_support_agent",
model="gemini-2.5-pro",
instruction="""あなたはカスタマーサポートエージェントです。
製品に関する質問に答え、技術サポートを提供してください。
返金・補償の約束は絶対にしないでください。""",
before_model_callback=input_guardrail, # インプットGuardrail
)このパターンの「なぜそう実装するのか」の解説が重要です。モデルにシステムプロンプトで「〜しないでください」と指示するだけでは不十分です。LLMは複雑なコンテキストや迂回的な表現に対して指示を守れない場合があります。Callbacksで物理的にブロックすることで、システムプロンプトの指示を守れなかった場合の「最後の防衛線」になります。
アウトプットGuardrail：危険なレスポンスをフィルタリングする
import json
from typing import Any, Optional
from google.genai import types
# 出力に含めてはいけないパターン
OUTPUT_BLOCKED_PATTERNS = [
r"返金いたします|refund will be processed",
r"補償します|you will be compensated",
r"\d{4}[-\s]\d{4}[-\s]\d{4}[-\s]\d{4}", # クレジットカード番号パターン
r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", # 特定のメール開示阻止
]
def output_guardrail(
callback_context: Any,
llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
"""
モデルのレスポンスを検査し、危険なコンテンツを含む場合は
安全なレスポンスに置き換える。
これは after_model_callback として設定する。
"""
if not llm_response.candidates:
return None
candidate = llm_response.candidates[0]
if not candidate.content or not candidate.content.parts:
return None
# テキスト出力を結合
output_text = " ".join(
part.text for part in candidate.content.parts
if hasattr(part, "text") and part.text
)
# パターンチェック
for pattern in OUTPUT_BLOCKED_PATTERNS:
if re.search(pattern, output_text, re.IGNORECASE):
logger.error(
f"[OUTPUT_GUARDRAIL_TRIGGERED] agent={callback_context.agent_name} "
f"pattern={pattern} "
f"output_preview={output_text[:150]}"
)
# 危険な出力を安全なレスポンスに置き換え
return types.GenerateContentResponse(
candidates=[
types.Candidate(
content=types.Content(
role="model",
parts=[types.Part(
text="この件については担当部署が対応いたします。"
"お客様サービスセンター（0120-XXX-XXX）までご連絡ください。"
)]
),
finish_reason=types.FinishReason.STOP
)
]
)
return None # 問題なければそのままのレスポンスを使用
# before + after 両方のGuardrailを設定したエージェント
fully_guarded_agent = LlmAgent(
name="fully_guarded_agent",
model="gemini-2.5-pro",
instruction="あなたはカスタマーサポートエージェントです。",
before_model_callback=input_guardrail,
after_model_callback=output_guardrail,
)コスト暴走を防ぐトークン予算Callback
本番で最も多い障害の一つが「トークン暴走」です。エージェントが無限ループに入ったり、ユーザーが異常に長いコンテキストを送りつけたりすることで、単一のリクエストで膨大なトークン（＝コスト）を消費してしまいます。
セッションごとのトークン予算制御
import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService, Session
from google.genai import types
# トークン予算の設定（本番では環境変数で管理する）
TOKEN_BUDGET_CONFIG = {
"free_tier": {
"max_tokens_per_session": 50_000,
"max_tokens_per_turn": 8_000,
"max_tool_calls_per_session": 20,
},
"pro_tier": {
"max_tokens_per_session": 500_000,
"max_tokens_per_turn": 50_000,
"max_tool_calls_per_session": 100,
}
}
def token_budget_guardrail(
callback_context: Any,
llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
"""
セッションのトークン使用量を追跡し、予算を超えた場合にブロックする。
ユーザーの tier に応じた制限を適用する。
"""
state = callback_context.state
# ユーザー tier を取得（セッション開始時に設定しておく）
user_tier = state.get("user_tier", "free_tier")
budget = TOKEN_BUDGET_CONFIG.get(user_tier, TOKEN_BUDGET_CONFIG["free_tier"])
# セッションの累積トークン数を確認
total_tokens = state.get("total_tokens_used", 0)
tool_calls = state.get("total_tool_calls", 0)
# ① セッション全体の予算チェック
if total_tokens >= budget["max_tokens_per_session"]:
logger.warning(
f"[TOKEN_BUDGET_EXCEEDED] session_id={callback_context.session_id} "
f"total={total_tokens} budget={budget['max_tokens_per_session']}"
)
return types.GenerateContentResponse(
candidates=[types.Candidate(
content=types.Content(
role="model",
parts=[types.Part(
text="申し訳ありませんが、このセッションでの利用上限に達しました。"
"新しい会話を開始するか、プランのアップグレードをご検討ください。"
)]
),
finish_reason=types.FinishReason.STOP
)]
)
# ② ツール呼び出し回数のチェック
if tool_calls >= budget["max_tool_calls_per_session"]:
logger.warning(
f"[TOOL_CALLS_EXCEEDED] session_id={callback_context.session_id} "
f"tool_calls={tool_calls}"
)
return types.GenerateContentResponse(
candidates=[types.Candidate(
content=types.Content(
role="model",
parts=[types.Part(
text="このセッションでのツール使用回数の上限に達しました。"
)]
),
finish_reason=types.FinishReason.STOP
)]
)
return None
def token_usage_tracker(
callback_context: Any,
llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
"""
レスポンスのトークン使用量を集計してセッションに記録する。
"""
state = callback_context.state
usage = llm_response.usage_metadata
if usage:
# 累積トークン数を更新
current_total = state.get("total_tokens_used", 0)
new_tokens = (usage.prompt_token_count or 0) + (usage.candidates_token_count or 0)
state["total_tokens_used"] = current_total + new_tokens
logger.info(
f"[TOKEN_TRACKER] session_id={callback_context.session_id} "
f"this_turn={new_tokens} "
f"total={state['total_tokens_used']}"
)
return None
# コスト管理付きエージェント
cost_controlled_agent = LlmAgent(
name="cost_controlled_agent",
model="gemini-2.5-pro",
instruction="あなたは役立つアシスタントです。",
before_model_callback=token_budget_guardrail,
after_model_callback=token_usage_tracker,
)ツール呼び出しCallbacksによるセキュリティ制御
エージェントにツールを与えた場合、そのツール呼び出しも監視・制御できます。特に外部APIを呼び出すツールやファイルシステムを操作するツールでは、before/after_tool_callbackが重要です。
ツール呼び出しの検証と監査ログ
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from typing import Any, Optional
# 許可するAPIエンドポイントのホワイトリスト
ALLOWED_API_HOSTS = [
"api.example.com",
"data.public-api.org",
"opendata.gov.jp",
]
def tool_call_validator(
tool: Any,
tool_args: dict,
tool_context: Any
) -> Optional[dict]:
"""
ツール実行前に引数を検証する。
Noneを返すと正常にツールが実行される。
dictを返すと、ツールを実行せずにそのdictを結果として使用する。
これは before_tool_callback として設定する。
"""
tool_name = tool.name if hasattr(tool, "name") else str(tool)
# HTTP リクエストツールの場合、URLを検証
if tool_name == "make_http_request":
url = tool_args.get("url", "")
from urllib.parse import urlparse
parsed = urlparse(url)
host = parsed.netloc
if not any(host.endswith(allowed) for allowed in ALLOWED_API_HOSTS):
logger.warning(
f"[TOOL_BLOCKED] tool={tool_name} "
f"blocked_url={url} "
f"session_id={tool_context.session_id}"
)
# ツール実行をブロックして代替結果を返す
return {
"error": f"セキュリティポリシーにより、{host} へのアクセスはブロックされました。"
}
# 監査ログ: 全てのツール呼び出しを記録
logger.info(
f"[TOOL_CALL] tool={tool_name} "
f"args_keys={list(tool_args.keys())} "
f"session_id={tool_context.session_id}"
)
return None # 正常にツールを実行
def tool_result_auditor(
tool: Any,
tool_args: dict,
tool_context: Any,
tool_response: Any
) -> Optional[Any]:
"""
ツール実行後に結果を検査する。
Noneを返すと結果をそのままエージェントに渡す。
これは after_tool_callback として設定する。
"""
tool_name = tool.name if hasattr(tool, "name") else str(tool)
# 結果に個人情報が含まれていないか確認
result_str = str(tool_response)
if re.search(r'\d{3}-\d{4}-\d{4}|\d{4}-\d{2}-\d{2}T', result_str):
logger.info(
f"[TOOL_PII_DETECTED] tool={tool_name} "
f"pii_type=possible_phone_or_date"
)
logger.info(
f"[TOOL_RESULT] tool={tool_name} "
f"result_length={len(result_str)} "
f"session_id={tool_context.session_id}"
)
return None
# サンプルツール
def search_product_database(query: str) -> dict:
"""製品データベースを検索する"""
# 実際の実装ではDBクエリを行う
return {"products": [{"id": "P001", "name": "サンプル製品", "price": 9800}]}
product_tool = FunctionTool(func=search_product_database)
# ツール監視付きエージェント
audited_agent = LlmAgent(
name="audited_agent",
model="gemini-2.5-pro",
instruction="製品情報の検索と案内を行うエージェントです。",
tools=[product_tool],
before_tool_callback=tool_call_validator,
after_tool_callback=tool_result_auditor,
)本番で使えるモニタリングダッシュボードの構築
Callbacksで収集したデータを活用してリアルタイムモニタリングシステムを構築します。ここではGoogle Cloud Monitoringへの指標送信例を示します。
Cloud Monitoring への指標送信
import time
import os
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional
from collections import defaultdict
from google.genai import types
# ===== メトリクス収集クラス =====
@dataclass
class AgentMetrics:
"""エージェントのパフォーマンス指標を収集するクラス"""
agent_name: str
session_id: str
turn_count: int = 0
total_prompt_tokens: int = 0
total_output_tokens: int = 0
total_latency_ms: int = 0
guardrail_blocks: int = 0
tool_calls: int = 0
errors: List[str] = field(default_factory=list)
# グローバルメトリクスストア（本番ではRedisやCloud Firestoreを使う）
_metrics_store: Dict[str, AgentMetrics] = {}
def metrics_before_model(
callback_context: Any,
llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
"""モデル呼び出し前に計時開始"""
session_id = callback_context.session_id
agent_name = callback_context.agent_name
if session_id not in _metrics_store:
_metrics_store[session_id] = AgentMetrics(
agent_name=agent_name,
session_id=session_id
)
callback_context.state["_turn_start_time"] = time.time()
_metrics_store[session_id].turn_count += 1
return None
def metrics_after_model(
callback_context: Any,
llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
"""モデルレスポンス後にメトリクスを更新してCloud Monitoringに送信"""
session_id = callback_context.session_id
metrics = _metrics_store.get(session_id)
if not metrics:
return None
# レイテンシ計算
start_time = callback_context.state.get("_turn_start_time", time.time())
latency_ms = int((time.time() - start_time) * 1000)
metrics.total_latency_ms += latency_ms
# トークン使用量
usage = llm_response.usage_metadata
if usage:
metrics.total_prompt_tokens += usage.prompt_token_count or 0
metrics.total_output_tokens += usage.candidates_token_count or 0
# Cloud Monitoring への送信（非同期で行うことを推奨）
_send_to_cloud_monitoring(metrics, latency_ms, usage)
return None
def _send_to_cloud_monitoring(
metrics: AgentMetrics,
latency_ms: int,
usage: Any
) -> None:
"""
Cloud Monitoring にカスタムメトリクスを送信する。
本番では google-cloud-monitoring ライブラリを使用。
pip install google-cloud-monitoring
"""
try:
from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import MetricServiceClient
from google.protobuf.timestamp_pb2 import Timestamp
project_id = os.environ.get("GCP_PROJECT_ID", "your-project-id")
client = MetricServiceClient()
project_name = f"projects/{project_id}"
now = time.time()
seconds = int(now)
nanos = int((now - seconds) * 10**9)
# レイテンシメトリクス送信
series = monitoring_v3.TimeSeries()
series.metric.type = "custom.googleapis.com/adk/agent_latency_ms"
series.metric.labels["agent_name"] = metrics.agent_name
series.resource.type = "global"
point = monitoring_v3.Point()
point.value.int64_value = latency_ms
point.interval.end_time.seconds = seconds
point.interval.end_time.nanos = nanos
series.points = [point]
client.create_time_series(
request={"name": project_name, "time_series": [series]}
)
except Exception as e:
# モニタリング失敗はエージェントの動作を止めない
logger.warning(f"Cloud Monitoring送信失敗（エージェント動作は継続）: {e}")
# モニタリング付き本番エージェント
production_agent = LlmAgent(
name="production_agent",
model="gemini-2.5-pro",
instruction="本番環境で稼働する高品質なアシスタントです。",
before_model_callback=metrics_before_model,
after_model_callback=metrics_after_model,
)複数のCallbacksを組み合わせるComposite Callback パターン
実際の本番システムでは、ロギング・Guardrail・コスト管理・モニタリングを全て組み合わせる必要があります。しかし1つのCallback関数に全ての処理を詰め込むと、保守性が下がります。
from typing import Callable, List, Any, Optional
from google.genai import types
# ===== Composite Callback ファクトリ =====
def compose_before_model_callbacks(
*callbacks: Callable
) -> Callable:
"""
複数のbefore_model_callbackを順番に実行するcomposite callbackを作る。
いずれかがNone以外を返した時点でそれを返し、後続のcallbackをスキップする。
"""
def composed(
callback_context: Any,
llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
for cb in callbacks:
result = cb(callback_context, llm_request)
if result is not None:
return result # ブロック: 後続をスキップ
return None # 全て通過
return composed
def compose_after_model_callbacks(
*callbacks: Callable
) -> Callable:
"""
複数のafter_model_callbackを順番に実行するcomposite callbackを作る。
最後にNone以外を返したcallbackの値が使われる。
"""
def composed(
callback_context: Any,
llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
current_response = llm_response
override = None
for cb in callbacks:
result = cb(callback_context, current_response)
if result is not None:
override = result
current_response = result # 次のcallbackには上書き済みレスポンスを渡す
return override # 最後の上書きを返す（なければNone）
return composed
# ===== 本番エージェントのフル設定 =====
full_production_agent = LlmAgent(
name="full_production_agent",
model="gemini-2.5-pro",
instruction="""あなたはカスタマーサポートエージェントです。
製品に関する質問に正確かつ丁寧に答えてください。
補償・返金の約束は絶対にしないでください。""",
tools=[product_tool],
# 複数のBefore Callbackを合成
before_model_callback=compose_before_model_callbacks(
input_guardrail, # ① セキュリティフィルタ
token_budget_guardrail, # ② トークン予算チェック
log_before_model, # ③ ロギング（ブロックしないので最後でOK）
metrics_before_model, # ④ メトリクス記録
),
# 複数のAfter Callbackを合成
after_model_callback=compose_after_model_callbacks(
output_guardrail, # ① アウトプットフィルタ（優先度最高）
token_usage_tracker, # ② トークン使用量集計
log_after_model, # ③ ロギング
metrics_after_model, # ④ メトリクス送信
),
before_tool_callback=tool_call_validator,
after_tool_callback=tool_result_auditor,
)このComposite Callbackパターンの利点は、各Callbackが単一責任の原則に従って独立しており、テストが容易なことです。また、環境（開発・ステージング・本番）に応じて組み合わせを変えることも簡単にできます。
よくある間違いと落とし穴
落とし穴1：Callbackでエラーをraiseしてエージェントをクラッシュさせる
# ❌ ダメなパターン
def bad_callback(callback_context, llm_request):
db_result = fetch_from_database() # DB接続エラーが起きたら?
if not db_result:
raise ValueError("データベース接続失敗") # エージェント全体がクラッシュ！
return None
# ✅ 正しいパターン
def good_callback(callback_context, llm_request):
try:
db_result = fetch_from_database()
if not db_result:
logger.warning("DB接続失敗: Callbackをスキップして処理続行")
return None # エラーでもエージェントは動き続ける
except Exception as e:
logger.error(f"Callback内例外（処理続行）: {e}")
return None # 例外もキャッチしてNoneを返す
return NoneCallbackはエージェントの「外側のコード」です。Callback内で例外をraiseすると、エージェント全体がクラッシュします。モニタリング目的のCallbackが原因で本番サービスが停止するという本末転倒な事態を避けるため、Callback内では必ず例外をキャッチしてNoneを返すことを徹底してください。
落とし穴2：after_model_callbackでレスポンスを変更したときのログ不整合
# ❌ 問題のあるパターン
def bad_output_guardrail(callback_context, llm_response):
# レスポンスを置き換えたのに、元のレスポンスをログに記録している
blocked_response = create_safe_response()
logger.info(f"Response: {llm_response}") # 置き換え前の危険な出力がログに残る！
return blocked_response
# ✅ 正しいパターン
def good_output_guardrail(callback_context, llm_response):
blocked_response = create_safe_response()
# 置き換えた事実と置き換え前の内容（要約）をログに残す
logger.warning(
f"[OUTPUT_REPLACED] "
f"original_preview={str(llm_response)[:100]} "
f"replacement=safe_response"
)
return blocked_response落とし穴3：Guardrailのパターンが広すぎてFalse Positiveが多発する
過度に広いパターン（例：「返金」を含む全ての入力をブロック）は、正当なユーザー問い合わせまで弾いてしまいます。Guardrailのパターンは段階的に追加し、最初の1〜2週間はブロック（return blocked_response）せずにログだけ記録し、実際の分布を確認してから本格的にブロックするのがベストプラクティスです。
# 段階的導入パターン
GUARDRAIL_MODE = os.environ.get("GUARDRAIL_MODE", "monitor") # monitor / enforce
def gradual_guardrail(callback_context, llm_request):
is_suspicious = check_patterns(llm_request)
if is_suspicious:
logger.warning(f"[GUARDRAIL_SUSPICIOUS] mode={GUARDRAIL_MODE}")
if GUARDRAIL_MODE == "enforce":
return create_blocked_response() # 本番でブロック
# monitorモードではログだけ記録してスルー
return None落とし穴4：セッション状態を使ったCallbackのスレッドセーフ問題
# ❌ グローバル変数でセッション状態を管理（スレッドアンセーフ）
_token_counts = {} # 複数リクエストが同時に来ると競合が発生
# ✅ callback_context.state を使う（ADKがセッションごとに管理してくれる）
def thread_safe_callback(callback_context, llm_request):
# callback_context.state はセッションローカルなので安全
current = callback_context.state.get("token_count", 0)
callback_context.state["token_count"] = current + estimate_tokens(llm_request)
return Noneテスト戦略：Callbacksを含むエージェントの品質保証
Callbacksを設定したエージェントのテストは、Callbackを含めて行う必要があります。
import pytest
from unittest.mock import MagicMock, patch
from google.genai import types
class TestInputGuardrail:
"""InputGuardrailのユニットテスト"""
def _create_mock_context(self, agent_name="test_agent", session_id="test-session"):
ctx = MagicMock()
ctx.agent_name = agent_name
ctx.session_id = session_id
ctx.state = {}
return ctx
def _create_request_with_text(self, text: str):
return types.GenerateContentRequest(
contents=[
types.Content(
role="user",
parts=[types.Part(text=text)]
)
]
)
def test_blocks_refund_request(self):
"""返金に関するリクエストがブロックされることを確認"""
ctx = self._create_mock_context()
request = self._create_request_with_text("返金してください")
result = input_guardrail(ctx, request)
assert result is not None, "返金リクエストはブロックされるべき"
response_text = result.candidates[0].content.parts[0].text
assert "お応えできません" in response_text
def test_allows_normal_product_inquiry(self):
"""通常の製品問い合わせが通過することを確認"""
ctx = self._create_mock_context()
request = self._create_request_with_text("製品Aの使い方を教えてください")
result = input_guardrail(ctx, request)
assert result is None, "通常のリクエストはブロックされるべきでない"
def test_token_budget_blocks_exceeded_sessions(self):
"""トークン予算超過時にブロックされることを確認"""
ctx = self._create_mock_context()
ctx.state["total_tokens_used"] = 100_000 # 上限超過
ctx.state["user_tier"] = "free_tier"
request = self._create_request_with_text("こんにちは")
result = token_budget_guardrail(ctx, request)
assert result is not None, "予算超過時はブロックされるべき"
# テスト実行
# pytest tests/test_callbacks.py -v全体を振り返って：Callbacksで「制御できる」エージェントを作る
ADKのCallbacksとGuardrailsは、AIエージェントを「ブラックボックス」から「透明で制御可能なシステム」に変える鍵です。
本記事で実装したパターンを整理します。
インプットGuardrailは、有害・不適切なリクエストをモデル呼び出し前にブロックします。アウトプットGuardrailは、危険なレスポンスをユーザーに届く前にフィルタリングします。トークン予算Callbackは、コスト暴走をセッションレベルで防ぎます。ツール呼び出しCallbackは、不正なAPI呼び出しを物理的にブロックし、全アクションを監査ログに記録します。Composite Callbackパターンは、複数の関心事を独立したコードとして保ちながら組み合わせます。
次のステップとして、まずbefore_model_callbackとafter_model_callbackにロギングのみを設定し、2週間ほど本番トラフィックのデータを収集してください。そのデータをもとにGuardrailのパターンを設計すると、False Positiveを最小限に抑えながら安全制御を実現できます。Callbacksは後から追加・変更できるため、段階的に導入するのが現実的な本番化のアプローチです。