2026年で最も高くつくバグは、メモリリークではありません。それは推論ループです。
このシナリオを想像してみてください:あなたは「サポートエージェント」を本番環境にデプロイします。その目的はユーザーのパスワードをリセットすることです。エージェントは予期しないAPIエラーに遭遇します。適切に失敗する代わりに、LLMは「異なるパラメータで再試行する」ことを決定します。再試行します。そしてまた再試行します。そしてまた再試行します。
APIが200 OK(本文にエラーメッセージを含む)を返すため、従来のAPMツール(Datadog、New Relic)は緑色のライトを表示します。
その間、あなたのエージェントは1秒あたり50トークンの速度で回転し続けています。10分後にダッシュボードを確認する頃には、その単一セッションでGPT-5のコンピュートクレジット500ドルを消費してしまっています。
エージェント時代において、CPUメトリクスは無関係です。 必要なのは認知的オブザーバビリティです。HTTPリクエストだけでなく、思考プロセスをトレースする必要があります。
このガイドでは、OpenTelemetry (OTel)とArize Phoenixを使用して、ゾンビエージェントがあなたを破産させる前に検出、診断、停止するためのフルスタックAIオブザーバビリティの実装方法を詳しく説明します。
盲点:標準的なAPMが失敗する理由
従来のトレーシング(Jaeger、Zipkin)は決定論的なマイクロサービス向けに構築されました。サービスAがサービスBを呼び出します。
エージェントは非決定論的なループです。
- スパン問題: エージェントの1回の「実行」は単一のリクエストではなく、50以上のLLM呼び出し、ツール実行、ベクトル検索からなるグラフです。
- ペイロード問題: 呼び出しが発生したことを知るだけでは役に立ちません。何をエージェントが考えていたかを知る必要があります。ツールのパラメータを幻覚したのか?観察結果を誤解したのか?
- コスト問題: マイクロサービスでは、コストは固定(サーバー時間)です。AIでは、コストは可変(トークン数)です。1セントの端数も特定のテナントIDまたはユーザーIDに帰属させる必要があります。
アーキテクチャ:OpenInferenceとOTel
車輪の再発明は必要ありません。OpenTelemetryを使用しますが、OpenInferenceのセマンティック規約(2026年のLLMアプリケーションのトレーシング標準)を採用します。
スタック
- インストルメンテーション: Python OpenTelemetry SDK +
openinference-instrumentation-langchain。 - コレクター: OTelコレクター(サイドカーとして実行)。
- バックエンド: Arize Phoenix(トレースの可視化と評価用)またはGrafana Tempo(長期保存用)。
設計図:10のエリート設定とコードスニペット
以下は、LangGraphエージェントをインストルメントするための本番環境対応の設定です。このセットアップは、入力、出力、ツール呼び出し、そして重要なことにステップごとのトークンコストをキャプチャします。
1. OTelトレーサープロバイダーのセットアップ(基礎)
グローバルトレーサーを設定し、データをArize(ローカルまたはクラウド)にエクスポートします。
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from openinference.instrumentation.langchain import LangChainInstrumentor
# 1. トレーサープロバイダーを初期化
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
# 2. エクスポーターを設定(Arize Phoenixローカルインスタンスをターゲット)
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
span_processor = BatchSpanProcessor(otlp_exporter)
tracer_provider.add_span_processor(span_processor)
# 3. LangChain / LangGraphを自動インストルメント
# これにより、すべての「ノード」実行がスパンとしてマジックのようにキャプチャされます
LangChainInstrumentor().instrument(tracer_provider=tracer_provider)
2. 「テナント対応」コンテキスト伝播
コスト帰属のために、特定のユーザーIDまたはテナントIDでトレースにタグを付ける方法。
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
def run_agent_for_user(user_query: str, tenant_id: str, user_id: str):
# インタラクション全体のルートスパンを開始
with tracer.start_as_current_span("agent_execution") as span:
# すべての子スパン(LLM呼び出し、DB検索)にカスケードする属性を設定
span.set_attribute("user.id", user_id)
span.set_attribute("tenant.id", tenant_id)
span.set_attribute("project.environment", "production")
# ここでLangGraphエージェントを呼び出し
# インストルメンターはこれらの属性をLLM呼び出しに自動的にアタッチします
agent_graph.invoke({"messages": [user_query]})
3. 「トークン流出」検出器(コスト追跡)
モデル名に基づいてリアルタイムでコストを計算するカスタムスパンプロセッサ。
# コストプロセッサロジックの疑似コード
MODEL_COSTS = {
"gpt-4o-2024-05-13": {"input": 5.00, "output": 15.00}, # 100万トークンあたりのドル
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00}
}
def calculate_cost(span):
model = span.attributes.get("llm.model_name")
input_tokens = span.attributes.get("llm.token_count.prompt")
output_tokens = span.attributes.get("llm.token_count.completion")
if model in MODEL_COSTS:
cost = (input_tokens / 1e6 * MODEL_COSTS[model]['input']) +
(output_tokens / 1e6 * MODEL_COSTS[model]['output'])
span.set_attribute("llm.cost.usd", cost)
4. 「無限ループ」サーキットブレーカー
ループを観察するだけでなく、停止させます。このロジックはトレースの深さをチェックします。
# LangGraphノード内
def reasoning_node(state):
# 状態から再帰の深さをチェック
current_depth = len(state['messages'])
if current_depth > 20:
# 緊急停止
# OTelに「重大イベント」スパンを記録
with tracer.start_as_current_span("emergency_halt") as span:
span.set_attribute("error", True)
span.set_attribute("reason", "最大再帰深度を超えました")
return {"messages": [SystemMessage(content="エラー:エージェントがループに陥りました。終了します。")]}
# 通常のロジックを続行...
5. トレース内のPIIマスキング(GDPR準拠)
PIIを含む生のプロンプトをログに記録してはいけません。アプリケーションからデータが流出する前に機密データを編集するカスタムフックを使用します。
from openinference.instrumentation import SafeLogger
class PIIRedactor:
def redact(self, text):
# メールアドレス、社会保障番号、クレジットカード用の正規表現
return re.sub(r'b[w.-]+@[w.-]+.w{2,4}b', '[EMAIL_REDACTED]', text)
# インストルメンターを編集機を使用するように設定
LangChainInstrumentor().instrument(
tracer_provider=tracer_provider,
response_hook=lambda span, response: span.set_attribute("output", PIIRedactor().redact(response))
)
6. ツール実行のトレーシング(「アクション」レイヤー)
エージェントがAPIに渡した引数を正確に可視化します。
# トレース可視化に確実に表示されるようにツールをデコレート
from langchain.tools import tool
@tool
def query_sql_database(query: str):
"""SQLクエリを実行します。"""
# 'tool'デコレータとOTel自動インストルメンテーションにより以下をキャプチャ:
# 1. 入力 'query'
# 2. 実行時間
# 3. 結果の行(またはエラー)
return db.run(query)
7. 「幻覚率」メトリクス
Arizeの評価ライブラリを使用して、トレースを非同期に評価します。
from phoenix.evals import HallucinationEvaluator, run_evals
import pandas as pd
# 過去1時間のトレースを取得
dataframe = phoenix_client.get_spans_dataframe(filter="span_kind == 'LLM'")
# 幻覚評価器を実行(小さなLLMを使用して大きなLLMを評価)
eval_results = run_evals(
evaluators=[HallucinationEvaluator(model="gpt-4-turbo")],
dataframe=dataframe,
provide_explanation=True
)
# 幻覚率が10%を超えた場合、PagerDutyをトリガー
if eval_results['hallucination_score'].mean() > 0.1:
alert_devops_team()
8. 分散トレーシングID注入(メタプロンプティング)
トレースIDをシステムプロンプトに注入し、LLMが自身のデバッグIDを「認識」できるようにします。
trace_id = trace.get_current_span().get_span_context().trace_id
system_prompt = f"""
あなたは親切なアシスタントです。
DEBUG_TRACE_ID: {trace_id}
致命的なエラーに遭遇した場合、このトレースIDをユーザーに出力してください。
"""
9. ステップタイプ別レイテンシバケット化
「思考時間」(LLM生成)と「実行時間」(ツールレイテンシ)を区別します。
# Arize / Prometheusクエリ
sum(rate(span_duration_seconds_sum{span_kind="LLM"}[5m]))
vs
sum(rate(span_duration_seconds_sum{span_kind="TOOL"}[5m]))
# 洞察:ツールレイテンシが急増した場合、それはあなたのDBです。LLMレイテンシが急増した場合、それはOpenAIです。
10. トレースサンプリングポリシー
本番環境ですべてをトレースしてはいけません。速度が低下します。
# ヘッドベースサンプリングを設定
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
# リクエストの10%をトレース、ただしエラーがあるリクエストは常にトレース
sampler = ParentBased(root=TraceIdRatioBased(0.1))
tracer_provider = TracerProvider(sampler=sampler)
2026年実装のベストプラクティス
1. AIの「ゴールデンシグナル」
CPU/RAMは忘れましょう。ダッシュボードには以下を表示すべきです:
- トークンスループット(TPS): 1秒あたりのトークン数。
- セッションあたりのコスト: 最も重要なビジネスメトリクス。
- ループカウント: セッションあたりの平均ターン数。これが急増した場合、プロンプトが壊れています。
- フィードバックスコア: 特定のトレースIDにリンクされたユーザーの / 比率。
2. 本番環境の文字列でデバッグしない
print(response)の使用をやめましょう。スケーラブルではなく、安全でもありません。Arizeのトレースウォーターフォールビューを使用します。これにより、単一のチャットセッションを展開し、正確なシーケンスを確認できます:
ユーザークエリ -> ルーター -> RAG検索 -> コンテキストスタッフィング -> LLM生成 -> ツール呼び出し -> 最終回答。
3. 「コスト速度」でアラート
アラートを設定します:「単一のトレースIDが2ドル以上を消費した場合、接続を切断する。」
これは、再帰ループバグが一晩で財布を空にするのを防ぐ保険証券です。
「ブラックボックス」から「ガラスボックス」へ
トレースできないものは、信頼できません。
OTelなしで自律エージェントを構築することは、窓が黒く塗られた飛行機を飛ばすようなものです。速く移動しているかもしれませんが、目的地に向かっているのか山に向かっているのか全くわかりません。
今日すぐにarize-phoenixをローカルにインストールしてください(pip install arize-phoenix)。開発マシンでトレースプロバイダーセットアップ(設定 #1)を実行してください。エージェントの「思考プロセス」がタイムラインとして初めて可視化されるのを見ることは、啓示です。
