AIエージェント入門

AIエージェントのオブザーバビリティ設計

AIエージェントのオブザーバビリティ設計

この記事の結論

AIエージェントの監視をOpenTelemetryで実装。構造化ログ・トレーシング・メトリクス設計。

AIエージェントが本番で動き始めると、ほぼ確実に「なぜ失敗したのか分からない」という壁にぶつかる。

通常のWebアプリなら、エラーログを見れば原因はだいたい特定できる。だがエージェントは違う。LLMの判断が非決定論的で、ツール呼び出しが連鎖し、同じ入力でも毎回異なる実行パスを通る。「あの時のエージェントが何を考えて、どのツールを叩いて、なぜその判断をしたのか」——これを後から追えない状態で本番運用するのは、目を閉じて飛行機を操縦するようなものだ。

この記事では、AIエージェントのオブザーバビリティ(可観測性)を、OpenTelemetryを中心に設計する方法をコード付きで解説する。

そもそもAIエージェントのオブザーバビリティは何が難しいのか

通常のAPM(Application Performance Monitoring)の前提は「決定論的なコードパス」だ。関数Aが呼ばれ、関数Bが呼ばれ、結果が返る——このシーケンスは基本的に再現できる。

AIエージェントはこの前提を壊す:

  • 非決定性: 同じプロンプトでも、LLMが毎回違うツールを選ぶ可能性がある
  • トークンコスト: 遅いだけでなく「高い」という失敗が起きる(コスト監視がなければ気づかない)
  • マルチステップチェーン: 1回のユーザーリクエストで10回のLLM呼び出し、5回のツール実行が起きることもある
  • データ感度: プロンプトにPIIが含まれる場合、ログをどこまで残すかに慎重な設計が必要

OpenTelemetryとGenAI Semantic Conventionsの基本

OpenTelemetryは、トレース・メトリクス・ログを標準化する観測フレームワーク。2025年からGenAI(生成AI)専用のセマンティック規約の整備が進み、2026年時点ではLangChain、CrewAI、AutoGen、LangGraphなど主要フレームワークが対応している。

GenAI Semantic Conventionsが定義するスパン属性の主なもの:

属性名 内容
gen_ai.system AIプロバイダー openai, anthropic, gemini
gen_ai.request.model 使用モデル gpt-4o, claude-3-5-sonnet
gen_ai.usage.input_tokens 入力トークン数 1234
gen_ai.usage.output_tokens 出力トークン数 567
gen_ai.response.finish_reasons 生成終了理由 stop, max_tokens, tool_calls
gen_ai.request.temperature Temperature設定 0.7

ツール呼び出し、LLM呼び出し、検索ステップがそれぞれ子スパンになり、エージェントの推論チェーン全体がひとつのトレースとして可視化される。

即効セットアップ:OpenLIT SDKでワンライン計装

最も手軽なのはOpenLIT SDKを使う方法だ。OpenAI、Anthropic、LangChain、LlamaIndexに対して、コード変更1行で自動計装できる。


# 動作環境: Python 3.10+, openlit>=1.28.0, opentelemetry-sdk>=1.24
# インストール: pip install openlit opentelemetry-sdk opentelemetry-exporter-otlp
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import openlit
from openai import OpenAI

# OpenLITで自動計装(1行)
openlit.init(
    otlp_endpoint="http://localhost:4318",  # OTLPコレクターのエンドポイント
    application_name="my-ai-agent",
    environment="production",
    # プロンプト内容のキャプチャ(PII注意)
    capture_message_content=False,  # 本番はFalse推奨
)

client = OpenAI()

# 以降のOpenAI API呼び出しは自動でトレースされる
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "AIエージェントの設計について教えて"}],
)

# gen_ai.system, gen_ai.usage.input_tokens 等が自動で記録される

これだけで、LLM呼び出しのトレースとトークン使用量がOpenTelemetryコレクター経由でGrafana、Jaeger、Langfuseなどに送られる。

カスタムスパン:ツール呼び出しとエージェントステップを手動計装

フレームワークを使わずに自前でエージェントを実装している場合や、特定のビジネスロジックをトレースしたい場合は手動計装が必要だ。


# カスタムスパンによるエージェントステップの計装
# インストール: pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.trace import Status, StatusCode
import time

# トレーサーの初期化
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("ai-agent-tracer")

def run_agent(user_query: str) -> str:
    """エージェントの実行全体をトレース"""
    with tracer.start_as_current_span("agent.run") as agent_span:
        agent_span.set_attribute("agent.query", user_query[:200])  # 長すぎる場合は切り詰め
        agent_span.set_attribute("gen_ai.system", "openai")

        try:
            # Step 1: ツール選択
            with tracer.start_as_current_span("agent.select_tool") as tool_span:
                tool_name = select_tool(user_query)  # LLMがツールを選択
                tool_span.set_attribute("agent.tool.name", tool_name)

            # Step 2: ツール実行
            with tracer.start_as_current_span(f"tool.{tool_name}") as exec_span:
                start = time.time()
                result = execute_tool(tool_name, user_query)
                exec_span.set_attribute("tool.duration_ms", int((time.time() - start) * 1000))
                exec_span.set_attribute("tool.result_length", len(str(result)))

            # Step 3: 最終回答生成
            with tracer.start_as_current_span("agent.generate_response") as gen_span:
                response, token_usage = generate_final_response(user_query, result)
                gen_span.set_attribute("gen_ai.usage.input_tokens", token_usage["input"])
                gen_span.set_attribute("gen_ai.usage.output_tokens", token_usage["output"])

            agent_span.set_status(Status(StatusCode.OK))
            return response

        except Exception as e:
            agent_span.set_status(Status(StatusCode.ERROR, str(e)))
            agent_span.record_exception(e)
            raise

構造化ログ:プロンプト内容のPII安全な記録方法

スパン属性にプロンプト全文を入れるのはアンチパターンだ。属性は常にインデックス化され、サイズ制限があり、バックエンドにPIIが蓄積するリスクがある。プロンプト内容はスパンイベントとして記録し、コレクターレベルでフィルタリングする。


# PII安全なプロンプトログの実装
import re
from opentelemetry import trace

def mask_pii(text: str) -> str:
    """基本的なPIIマスキング"""
    # メールアドレス
    text = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}', '[EMAIL]', text)
    # 電話番号(日本形式)
    text = re.sub(r'0d{1,4}-d{2,4}-d{4}', '[PHONE]', text)
    # クレジットカード番号(基本パターン)
    text = re.sub(r'bd{4}[s-]?d{4}[s-]?d{4}[s-]?d{4}b', '[CARD]', text)
    return text

def log_llm_interaction(span, prompt: str, response: str, tokens: dict):
    """プロンプトとレスポンスをイベントとして記録(PII安全)"""
    tracer = trace.get_current_span()

    # スパンイベントとして記録(属性ではなくイベント)
    span.add_event(
        name="llm.prompt",
        attributes={
            "prompt.masked": mask_pii(prompt)[:500],  # マスキング済み、500文字に制限
            "prompt.length": len(prompt),
        }
    )
    span.add_event(
        name="llm.response",
        attributes={
            "response.length": len(response),
            "gen_ai.usage.input_tokens": tokens.get("input", 0),
            "gen_ai.usage.output_tokens": tokens.get("output", 0),
        }
    )

メトリクス:コスト・レイテンシ・エラー率の監視

トレース(何が起きたか)に加えて、メトリクス(傾向・集計値)も必要だ。エージェントの運用で特に重要なメトリクスを定義しよう。


# AIエージェント向けメトリクスの定義
# インストール: pip install opentelemetry-sdk
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter

# メータープロバイダーの初期化
exporter = OTLPMetricExporter(endpoint="http://localhost:4318/v1/metrics")
reader = PeriodicExportingMetricReader(exporter, export_interval_millis=60000)
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)

meter = metrics.get_meter("ai-agent-metrics")

# 主要メトリクスの定義
llm_request_duration = meter.create_histogram(
    name="gen_ai.request.duration",
    description="LLM APIリクエストのレイテンシ(ミリ秒)",
    unit="ms",
)

llm_token_usage = meter.create_counter(
    name="gen_ai.usage.tokens",
    description="トークン使用量の累計",
    unit="tokens",
)

agent_error_count = meter.create_counter(
    name="agent.errors",
    description="エージェントのエラー件数",
)

# 使用例
def record_llm_call(model: str, duration_ms: float, input_tokens: int, output_tokens: int):
    """LLM呼び出しのメトリクスを記録"""
    labels = {"gen_ai.system": "openai", "gen_ai.request.model": model}

    llm_request_duration.record(duration_ms, attributes=labels)
    llm_token_usage.add(input_tokens, attributes={**labels, "token_type": "input"})
    llm_token_usage.add(output_tokens, attributes={**labels, "token_type": "output"})

【要注意】よくある失敗パターンと回避策

失敗1:プロンプト全文をスパン属性に保存する

span.set_attribute("prompt", full_prompt_text)
⭕ スパンイベントを使い、PII部分はマスキングして記録する

なぜ重要か: スパン属性はインデックス化されてすべてのバックエンドに転送される。PIIを含むプロンプトが監視ツールに保存されると、データプライバシー規制(GDPR等)への対応が複雑になる。

失敗2:同期エクスポーターで本番レイテンシを増やす

SimpleSpanProcessor(テスト向け)を本番で使う
BatchSpanProcessorでバックグラウンド送信する

なぜ重要か: SimpleSpanProcessorはスパン終了のたびに同期的にエクスポートを行う。本番では必ずBatchSpanProcessorを使い、エージェントのレスポンスタイムへの影響を最小化する。

失敗3:トレースIDをログに含めない

❌ 構造化ログにtrace_idを含めず、ログとトレースを別々に見る
⭕ すべてのログにtrace_idとspan_idを付与して相関分析できるようにする


# ログにトレースIDを含める実装
import logging
from opentelemetry import trace

class TraceIDFilter(logging.Filter):
    """ログレコードにOpenTelemetryのトレースIDを付与"""
    def filter(self, record):
        span = trace.get_current_span()
        if span and span.is_recording():
            ctx = span.get_span_context()
            record.trace_id = format(ctx.trace_id, '032x')
            record.span_id = format(ctx.span_id, '016x')
        else:
            record.trace_id = "no-trace"
            record.span_id = "no-span"
        return True

# ロガーの設定
logger = logging.getLogger("ai-agent")
logger.addFilter(TraceIDFilter())

まとめ:今日から始める3つのアクション

  1. 今日やること: OpenLITをインストールして既存のLLM呼び出しに openlit.init() を追加する。Jaegerかlocal OTLPコレクターを起動してトレースが届くことを確認する(所要30分)
  2. 今週中: カスタムスパンでエージェントのステップを手動計装し、ツール呼び出しごとのレイテンシとトークン使用量を可視化する
  3. 今月中: メトリクスをGrafanaに連携し、コスト・エラー率・P95レイテンシのアラートを設定する。SLO(サービスレベル目標)を定義して運用体制を整える

あわせて読みたい:


参考・出典

AIエージェントの監視設計・運用支援のご相談は株式会社Uravation(お問い合わせ)からお気軽にどうぞ。

この記事はAIgent Lab編集部がお届けしました。

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事