読者が抱える課題
AIエージェントを本番環境へ導入する際、従来のWebアプリケーションとは異なる以下のような課題が発生します。
- 非同期処理やTool呼び出し（Function Calling）が多段化し、どこで遅延やエラーが発生したか追跡できない
- プロンプトの変更やモデルの挙動変化による精度低下の原因特定が困難
- LLMのAPIトークン利用量やコストの急増をリアルタイムに検知・分析できない
- ユーザーのプライバシー（PII）や機密情報が外部サービスログに流出するリスクがある
この記事で分かること
- AIエージェントに特化した3つの観点（トレース・評価/品質・コスト）におけるObservability設計方針
- OpenTelemetryおよびLangChain/LangSmithを用いた具体的なトレース実装例
- マスク処理（PII対策）を含めた実務向けログ出力パターン
- 本番運用開始時に確認すべきObservabilityチェックリスト
対象読者・前提条件
- LLMフレームワーク（LangChain, LlamaIndex, OpenAI SDK等）を用いてAIエージェントを構築・運用するエンジニア
- OpenTelemetryやDatadog, LangSmith等を用いた分散トレースの基礎知識があること
AIエージェントにおけるObservabilityの3大要素
AIエージェントの監視は、通常のシステム監視（CPU/メモリ/HTTPレスポンスタイム）に加え、以下の要素を収集・視覚化する必要があります。
| 監視カテゴリ | 収集すべきデータ項目 | 活用目的 | 主なツール例 |
|---|---|---|---|
| 分散トレース | Agent Step, Tool Execution, LLM Callの親子関係と所要時間 | 遅延のボトルネック特定、無限ループの検知 | OpenTelemetry, Datadog, LangSmith |
| トークン・コスト | Prompt Tokens, Completion Tokens, Model ID | コスト予測、異常消費の早期アラート | LangSmith, Helicone, CloudWatch |
| 入出力・品質ログ | Input/Output Prompt, Toolパラメータ, PIIマスク済データ | 誤回答（ハルシネーション）のデバッグ、回帰テストのデータ収集 | Langfuse, Phoenix, Elasticsearch |
実装例：OpenTelemetryとLangChainを用いたトレース・ログ収集
以下は、PythonのLangChainおよびOpenTelemetryを使用して、エージェントの実行コンテキストやトークン情報をログおよびトレースとして出力する実装例です。
※このコードは概念を示す独自の実装例です。実行前に使用するライブラリのバージョン（LangChain, OpenTelemetry SDK）に合わせて適宜調整してください。
import os
import logging
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType, Tool
from langchain.callbacks.base import BaseCallbackHandler
# 1. OpenTelemetryの基本設定
provider = TracerProvider()
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("ai-agent-tracer")
# 2. 独自CallbackHandlerによるトークン数・ツール呼び出しのログ収集例
class AgentObservabilityCallbackHandler(BaseCallbackHandler):
def __init__(self):
self.logger = logging.getLogger("AgentLogger")
self.logger.setLevel(logging.INFO)
def on_llm_start(self, serialized, prompts, **kwargs):
span = trace.get_current_span()
if span:
span.set_attribute("llm.model_name", kwargs.get("invocation_params", {}).get("model_name", "unknown"))
def on_llm_end(self, response, **kwargs):
llm_output = response.llm_output or {}
token_usage = llm_output.get("token_usage", {})
# OpenTelemetry Spanへの属性設定（個人情報を含まないメタデータ）
span = trace.get_current_span()
if span and token_usage:
span.set_attribute("llm.prompt_tokens", token_usage.get("prompt_tokens", 0))
span.set_attribute("llm.completion_tokens", token_usage.get("completion_tokens", 0))
span.set_attribute("llm.total_tokens", token_usage.get("total_tokens", 0))
self.logger.info(f"[LLM End] Token Usage: {token_usage}")
def on_tool_start(self, serialized, input_str, **kwargs):
self.logger.info(f"[Tool Start] Tool: {serialized.get('name')} | Input: {input_str}")
# 3. エージェントの実行例
def run_agent(user_query: str):
with tracer.start_as_current_span("agent_execution") as span:
span.set_attribute("user.query_length", len(user_query))
# 疑似ツールの定義
def dummy_search(query: str):
return "検索結果データ"
tools = [
Tool(
name="SearchTool",
func=dummy_search,
description="情報を検索するためのツール"
)
]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
handler = AgentObservabilityCallbackHandler()
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=False
)
# Callbackを渡して実行
result = agent.run(user_query, callbacks=[handler])
return result
if __name__ == "__main__":
# テスト実行
run_agent("最新のニュースを調べてください")
運用上の注意点とセキュリティ設計
1. PII（個人識別情報）・機密情報のマスク処理
プロンプトやツールの引数には、ユーザーの入力した個人情報や機密データが含まれる可能性があります。ログストレージやサードパーティ製分析ツールへデータを送信する前に、必ずアプリケーション層でマスキング処理（サニタイズ）を行ってください。
- 悪い例: 生のユーザー入力（氏名、メールアドレス、クレジットカード番号など）をそのまま外部SaaSのトレース属性に格納する。
- 良い例: 正規表現や専用ライブラリ（Presidio等）を用いてハッシュ化・置換を行った状態でログ送信する。
2. トークン上限・ステップ上限による無限ループ防止
エージェントが誤った判断を繰り返し、Tool呼び出しを無限ループさせるリスクがあります。Observabilityツールでのアラート設定に加え、アプリ側でのハードリミットを必ず導入してください。
- 最大再帰ステップ数（Max Iterations）の明示的指定
- 1リクエストあたりの最大許容トークン数およびタイムアウトの設定
3. APIレスポンス変更および非推奨機能への対応
LLMベンダーのAPI仕様変更や標準ライブラリのバージョンアップに伴い、レスポンス構造（token_usageの格納位置など）が変更される場合があります。本番導入前に、最新の公式ドキュメントおよびリリースノートを確認してください。
本番運用開始時のObservabilityチェックリスト
本番環境にAIエージェントをデプロイする前に、以下の項目が考慮されているか確認してください。
- トレースIDの伝播: リクエストヘッダーからエージェント、内部API、LLM呼び出しまで一貫したトレースIDが引き継がれているか
- トークンコストの計測: モデル別・機能別にトークン使用量が集計され、ダッシュボードで確認できるか
- ログのサニタイズ: 外部SaaSやログ出力先に個人情報やAPIキーがプレーンテキストで出力されていないか
- エラーハンドリング: ツール呼び出し失敗やタイムアウト時に、適切なエラーログとステータスコードが記録されるか
- 異常検知アラート: 短時間でのトークン消費量急増や、エージェントのループ発生率に対するアラート閾値が設定されているか
まとめ
AIエージェントの非決定的な挙動を安全に制御するためには、従来のシステム監視に加え、Step単位のトレース、トークン使用量の監視、および徹底したプライバシー保護設計が不可欠です。本記事の実装パターンやチェックリストを参考に、継続的な改善が可能なObservability基盤を構築してください。