AI 可観測性: LLM、トークン、エージェントを監視
の大量導入により、 大規模言語モデル (LLM) 生産中に出現します 可観測性の新しい領域: AI ベースのアプリケーション監視。電話 LLM には、従来の API と比較して、変動費ベースという独特の特徴があります。 トークンの使用状況、予測不可能なレイテンシー、非決定的な出力、およびリスクについて 幻覚。ザ」AIの可観測性 古典的な可観測性の原則を適用します この新しいパラダイムに。
この記事では、言語モデル呼び出しをインストルメント化する方法を検討します。 AI エージェントの動作を追跡し、リアルタイムのコストを監視し、検出します OpenTelemetry と新たなセマンティック規約を使用したモデル応答の異常 AIの場合。
この記事で学べること
- AI アプリケーションには特殊な可観測性が必要なため
- 特定のスパンと属性を使用して LLM 呼び出しをトレースする
- トークンの使用状況とコストをリアルタイムで監視する
- AI エージェントの動作を観察する (ツール呼び出し、推論)
- 幻覚と品質劣化を検出
- AI 可観測性のためのフレームワークとツール
AI アプリケーションには特殊な可観測性が必要なため
LLM ベースのアプリケーションには、根本的に異なる特徴があります。 可観測性の観点から従来のアプリケーションと比較すると、次のようになります。
AI 可観測性の特有の課題
| チャレンジ | 従来のアプリケーション | AI/LLM アプリケーション |
|---|---|---|
| コスト | 固定 (コンピューティング、ストレージ) | 変数 (トークンごと、リクエストごと) |
| レイテンシ | 予測可能 (ミリ秒範囲) | 高速かつ可変 (ストリームごとに 1 ～ 60 秒) |
| 出力 | 決定論的 | 非決定的（温度、サンプリング） |
| エラー | ステータスコードをクリアする | 幻覚、一貫性のない反応 |
| テスト | 決定論的な単体テスト | 定性的評価（eval） |
LLM コールのインストルメント化
言語モデル呼び出しのインストルメンテーションは、OTel と同じパターンに従います。 外部 API 用ですが、AI メタデータをキャプチャするための特定の属性が含まれています: テンプレート 使用済み、トークン数、温度、推定コスト。
from opentelemetry import trace, metrics
import time
import tiktoken
tracer = trace.get_tracer("ai-service", "1.0.0")
meter = metrics.get_meter("ai-service", "1.0.0")
# Metriche specifiche per LLM
llm_token_usage = meter.create_counter(
name="llm.token.usage",
description="Token utilizzati nelle chiamate LLM",
unit="token"
)
llm_request_duration = meter.create_histogram(
name="llm.request.duration",
description="Durata delle chiamate LLM",
unit="ms"
)
llm_cost = meter.create_counter(
name="llm.cost.total",
description="Costo stimato delle chiamate LLM",
unit="USD"
)
def call_llm(messages, model="gpt-4", temperature=0.7, max_tokens=1000):
with tracer.start_as_current_span(
"llm.chat.completion",
attributes={
# Semantic conventions per AI (draft)
"gen_ai.system": "openai",
"gen_ai.request.model": model,
"gen_ai.request.temperature": temperature,
"gen_ai.request.max_tokens": max_tokens,
"gen_ai.request.top_p": 1.0,
# Contesto applicativo
"llm.prompt.messages_count": len(messages),
"llm.prompt.system_prompt_length": len(messages[0]["content"])
if messages[0]["role"] == "system" else 0,
}
) as span:
start_time = time.monotonic()
try:
# Conteggio token input (pre-chiamata)
encoding = tiktoken.encoding_for_model(model)
input_tokens = sum(
len(encoding.encode(m["content"])) for m in messages
)
span.set_attribute("gen_ai.usage.prompt_tokens", input_tokens)
# Chiamata al modello
response = openai_client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
# Attributi dalla risposta
output_tokens = response.usage.completion_tokens
total_tokens = response.usage.total_tokens
span.set_attribute("gen_ai.usage.completion_tokens", output_tokens)
span.set_attribute("gen_ai.usage.total_tokens", total_tokens)
span.set_attribute("gen_ai.response.model", response.model)
span.set_attribute("gen_ai.response.finish_reason",
response.choices[0].finish_reason)
# Calcolo costo stimato
cost = estimate_cost(model, input_tokens, output_tokens)
span.set_attribute("llm.cost.estimated_usd", cost)
# Registrare metriche
duration_ms = (time.monotonic() - start_time) * 1000
common_attrs = {
"gen_ai.system": "openai",
"gen_ai.request.model": model
}
llm_token_usage.add(input_tokens,
{**common_attrs, "gen_ai.token.type": "input"})
llm_token_usage.add(output_tokens,
{**common_attrs, "gen_ai.token.type": "output"})
llm_request_duration.record(duration_ms, common_attrs)
llm_cost.add(cost, common_attrs)
span.set_status(StatusCode.OK)
return response
except RateLimitError as e:
span.record_exception(e)
span.set_status(StatusCode.ERROR, "Rate limit exceeded")
span.set_attribute("llm.error.type", "rate_limit")
raise
except Exception as e:
span.record_exception(e)
span.set_status(StatusCode.ERROR, str(e))
raise
AIエージェントの追跡
Gli AIエージェント LLM を使用して推論、計画、および実行するシステムです。 ツールを呼び出して (ツール呼び出し)、複雑なタスクを完了します。彼らの行動と 複数の推論サイクルが含まれるため、観察するのが特に困難です。 ツール選択の決定と結果の構成。
def trace_agent_execution(task, agent):
with tracer.start_as_current_span(
"agent.execute",
attributes={
"agent.name": agent.name,
"agent.model": agent.model,
"agent.task": task.description,
"agent.max_iterations": agent.max_iterations,
"agent.tools_available": ",".join(agent.tool_names)
}
) as agent_span:
iteration = 0
total_tokens = 0
total_cost = 0.0
tools_called = []
while not agent.is_done() and iteration < agent.max_iterations:
iteration += 1
# Span per ogni iterazione del reasoning loop
with tracer.start_as_current_span(
f"agent.iteration.{iteration}",
attributes={
"agent.iteration": iteration,
"agent.state": agent.current_state
}
) as iter_span:
# Span per la chiamata LLM di reasoning
with tracer.start_as_current_span("agent.reasoning") as reason_span:
decision = agent.reason(task)
reason_span.set_attribute("agent.decision.type",
decision.type) # "tool_call" | "final_answer"
reason_span.set_attribute("agent.decision.confidence",
decision.confidence)
total_tokens += decision.tokens_used
# Se l'agente decide di chiamare un tool
if decision.type == "tool_call":
with tracer.start_as_current_span(
"agent.tool_call",
attributes={
"agent.tool.name": decision.tool_name,
"agent.tool.input_summary": decision.tool_input[:200]
}
) as tool_span:
result = agent.execute_tool(decision)
tool_span.set_attribute("agent.tool.success",
result.success)
tool_span.set_attribute("agent.tool.output_length",
len(str(result.output)))
tools_called.append(decision.tool_name)
# Attributi finali dell'agente
agent_span.set_attribute("agent.iterations_total", iteration)
agent_span.set_attribute("agent.tokens_total", total_tokens)
agent_span.set_attribute("agent.cost_total_usd", total_cost)
agent_span.set_attribute("agent.tools_called",
",".join(tools_called))
agent_span.set_attribute("agent.success", agent.is_done())
リアルタイムのコスト監視
LLM 呼び出しのコストは、特に GPT-4 のようなモデルの場合、急速に拡大する可能性があります。 またはクロード。リアルタイムのコスト監視により、異常（ループ）を検出できます。 無限に消費するトークン)、予算アラートを設定し、テンプレートの使用を最適化します。
# Alert rules Prometheus per AI cost monitoring
groups:
- name: ai-cost-alerts
rules:
# Alert: spesa oraria sopra il budget
- alert: HighAICostPerHour
expr: |
sum(rate(llm_cost_total[1h])) * 3600 > 50
for: 5m
labels:
severity: warning
annotations:
summary: "AI cost exceeding $50/hour"
# Alert: tasso di errori LLM alto
- alert: HighLLMErrorRate
expr: |
rate(llm_request_duration_count{status="error"}[5m])
/ rate(llm_request_duration_count[5m]) > 0.1
for: 3m
labels:
severity: critical
annotations:
summary: "LLM error rate above 10%"
# Alert: latenza LLM degradata
- alert: HighLLMLatency
expr: |
histogram_quantile(0.95,
rate(llm_request_duration_bucket[5m])
) > 30000
for: 5m
labels:
severity: warning
annotations:
summary: "LLM P95 latency above 30 seconds"
# Alert: anomalia token usage
- alert: AnomalousTokenUsage
expr: |
rate(llm_token_usage[5m]) > 3 * avg_over_time(rate(llm_token_usage[1h])[24h:1h])
for: 10m
labels:
severity: warning
annotations:
summary: "Token usage 3x above 24h average"
AI 可観測性の主要な指標
- トークンの使用法: モデルごと、エンドポイントごと、ユーザーごとの入出力トークン
- リクエストあたりのコスト: 使用されるモデルとトークンに基づくコストの見積もり
- TTFT レイテンシ: ストリーミング アプリケーションにとって重要な最初のトークンまでの時間
- エラー率: レート制限、タイムアウト、モデルエラー
- 終了理由: 配布停止、max_tokens、tool_call、content_filter
- エージェントの反復: 完了したタスクごとの推論サイクル数
- ツール呼び出しの成功率: 成功したツール呼び出しと失敗したツール呼び出しの割合
幻覚の検出と品質
のモニタリング 回答の質 AIのユニークな側面 観察性。技術的なエラー (タイムアウト、レート制限) は簡単に検出できますが、 幻覚と低品質の応答には代理指標と評価が必要 自動。
応答の品質に関するプロキシシグナル
応答の長さ: 回答が短すぎるか長すぎます 敬意を表します
平均値が低い場合は、品質に問題があることを示している可能性があります。
異常終了理由：高い割合 max_tokens を示します
切り捨てられた応答。高い割合 content_filter ブロックされたコンテンツを示します。
ユーザーからのフィードバック: 応答後のユーザーのアクションを追跡します (再試行、
サムダウン、放棄）を品質の間接的なシグナルとして使用します。
類似性スコア: 回答を参考回答と比較します
エンベディングを使用して品質のドリフトを検出します。
AI 可観測性のフレームワーク
AI の可観測性を簡素化するために、いくつかのフレームワークが登場しています。 主要な AI ライブラリと事前構成されたダッシュボードの自動インストルメンテーション。 最も関連性の高いものとしては、次のようなものがあります。 OpenLLMetry (LLM の OTel に基づく)、 ラング・スミス (ラングチェーンの場合)、 ヘリコン (OpenAIのプロキシ) そして GenAI の OTel セマンティック規約 (新たな標準)。
結論と次のステップ
AI 可観測性は、可観測性の原則を拡張する急速に進化している分野です クラシックから LLM ベースのアプリケーションまで。特有の課題（変動費、非生産性） 決定論的、幻覚など）には、特殊な測定基準と監視パターンが必要です。
AI 可観測性の 3 つの柱は次のとおりです。 コスト監視 (トークンの使用法、 リクエストあたりのコスト)、 パフォーマンス監視 (レイテンシ、TTFT、エラー率) e 品質監視 (終了理由、ユーザーからのフィードバック、類似性スコア)。
シリーズの次の最終記事では、 完全なケーススタディ: マイクロサービス アーキテクチャ向けの可観測性システムのエンドツーエンドの実装、 OpenTelemetry の導入前と導入後のメトリクスを示します。