Langfuse×OTelで多段ツールの遅延と無限ループを検知・可視化する

ツール呼び出しを多段に組んだLLMアプリで「全体が遅い」「なぜかコストが跳ねた」と言われても、原因がLLM推論なのか、外部APIなのか、リトライの連鎖なのかを即答できない。そんな状況は珍しくありません。本記事では、LangfuseとOpenTelemetryを組み合わせ、ツール呼び出しを分散トレーシングとして可視化する構成を紹介します。スパン設計、ループとコンテキスト肥大の検知、本番でのマスキングとサンプリングまで、実装コードを交えて整理します。

多段呼び出しで顕在化する「ブラックボックス化」とレイテンシの正体

多段呼び出しでは、1回のユーザーリクエストが「LLM推論 → ツール実行 → 結果を含めた再推論」を繰り返します。HTTPのアクセスログやAPMの1リクエスト1スパンという粒度では、この内部が見えません。見えないまま起きやすい問題は大きく3つです。

1. 直列化による遅延の積み上げ: 各ステップは数百msから数秒でも、5〜10ステップ連なると体感で数十秒になります。
2. 意図しない連続リトライ: ツールがエラーを返すと、モデルが同じ引数で再呼び出しを繰り返すケースがあります。
3. コンテキスト肥大: ツール結果が履歴に積み上がり、ステップごとに入力トークンが増えます。遅延とコストが同時に悪化します。

遅延の正体は、多くの場合「1つの遅い呼び出し」ではなく「ステップ数 × 入力トークンの増加」です。したがって、ステップ単位の所要時間、入力・出力トークン、同一ツールの呼び出し回数を、1本のトレースの中で並べて見られる必要があります。分散トレーシングの親子スパン構造は、この要件にそのまま合います。

OpenTelemetryとLangfuseを組み合わせるアーキテクチャ設計

役割を分けると設計が単純になります。OpenTelemetryは計装とデータ転送の標準、Langfuseは入出力・トークン・コストをLLM向けに閲覧するUIです。LangfuseはOTLPのエンドポイント(/api/public/otel)を公開しており、OTLP/HTTP(JSONとprotobuf)で受け取れます。gRPCは対象外なので、エクスポーターはHTTP系を選んでください。認証は、公開鍵と秘密鍵を : でつないでBase64化したBasic認証です。

構成は次の流れを推奨します。

1. アプリケーションがOTel SDKでスパンを生成します。
2. OpenTelemetry Collectorが受け取り、マスキングとサンプリングを適用します。
3. LLMトレース用にLangfuseへ、インフラ全体の相関用に既存のバックエンド(Tempo、Jaeger、Datadogなど)へファンアウトします。

Collectorを挟む理由は、機密情報の除去やサンプリングをアプリのデプロイと切り離して変更できる点にあります。小規模なら、アプリから直接Langfuseへ送っても動きます。

export OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel"
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic ${LANGFUSE_AUTH_BASE64}"

セルフホストの場合は、エンドポイントを自環境のホストに差し替えます。リージョンやパスはバージョンで変わりうるため、導入時に公式ドキュメントで確認してください。

Node.js/Pythonでの計装実装：スパン定義とトークン・入出力の記録

スパンは「リクエスト全体(親)→ステップ → LLM呼び出し / ツール呼び出し」の階層で切ります。LLM呼び出しにはOpenTelemetryのGenAIセマンティック規約(gen_ai.*)の属性を使います。Langfuseはモデル名やトークン数をこれらの属性から読み取ります。入出力を載せる属性名を含め、マッピングの詳細は利用中のLangfuseのバージョンの仕様に従ってください。

Pythonでの最小例です。ツールごとに引数のハッシュを付けておくと、後でループ検知に使えます。

import hashlib, json
from collections import Counter
from opentelemetry import trace

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

class ToolGuard:
    def __init__(self, max_repeat=3):
        self.seen = Counter()
        self.max_repeat = max_repeat

    def check(self, name, args):
        key = (name, hashlib.sha256(
            json.dumps(args, sort_keys=True).encode()).hexdigest()[:12])
        self.seen[key] += 1
        return key[1], self.seen[key]

def call_tool(guard, name, args, step, fn):
    with tracer.start_as_current_span(f"tool.{name}") as span:
        args_hash, repeat = guard.check(name, args)
        span.set_attribute("tool.name", name)
        span.set_attribute("agent.step", step)
        span.set_attribute("tool.args_hash", args_hash)
        span.set_attribute("tool.repeat_count", repeat)
        if repeat > guard.max_repeat:
            span.set_attribute("agent.loop_suspected", True)
            raise RuntimeError("repeated tool call limit exceeded")
        result = fn(**args)
        span.set_attribute("tool.result_bytes", len(json.dumps(result)))
        return result

def record_llm(span, model, usage):
    span.set_attribute("gen_ai.request.model", model)
    span.set_attribute("gen_ai.usage.input_tokens", usage.input)
    span.set_attribute("gen_ai.usage.output_tokens", usage.output)

Node.jsでは、@opentelemetry/sdk-node と OTLP HTTPエクスポーターで同じ設計を組めます。

import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { trace } from "@opentelemetry/api";

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: `${process.env.LANGFUSE_HOST}/api/public/otel/v1/traces`,
    headers: { Authorization: `Basic ${process.env.LANGFUSE_AUTH_BASE64}` },
  }),
});
sdk.start();

const tracer = trace.getTracer("agent-app");

export async function callTool(name, args, step, fn) {
  return tracer.startActiveSpan(`tool.${name}`, async (span) => {
    span.setAttributes({ "tool.name": name, "agent.step": step });
    try {
      return await fn(args);
    } catch (e) {
      span.recordException(e);
      span.setStatus({ code: 2 });
      throw e;
    } finally {
      span.end();
    }
  });
}

ポイントは、ステップ番号、引数ハッシュ、結果サイズを自前の属性として必ず付けることです。これらは後段の検知ルールの材料になります。

異常検知の実践：想定外のループ実行とコンテキスト肥大化をアラートする

ループは「同一ツール・同一引数ハッシュの繰り返し」と「ステップ数の異常」で捉えます。先のコードのように、アプリ内で上限を超えたら打ち切るガードを入れるのが第一です。可視化だけでは、ループ中のコストは止まりません。

そのうえで、アラートはメトリクス側で組みます。Collectorの spanmetrics コネクタを使うと、スパンからツール別の呼び出し数とレイテンシ分布をPrometheusなどに出せます。tool.name をディメンションに指定し、agent.loop_suspected=true のスパン数や、1トレースあたりのステップ数が閾値を超えた件数をアラート条件にします。Langfuse側では、同じトレースを開いて、どの引数でループしたかを入出力つきで確認します。

コンテキスト肥大は、各LLMスパンの gen_ai.usage.input_tokens をステップ順に見ると判別できます。ステップごとにほぼ単調増加し、ツール結果が主因なら、次のような対策が効きます。

1. ツール結果を要約または切り詰めてから履歴に戻す。
2. 古いステップの結果を履歴から落とす。
3. 大きな結果は参照IDだけ返し、必要時に取得させる。

対策前後で「トレース1件あたりの総入力トークン」と「p95レイテンシ」を比べると、ツール呼び出し最適化の効果を数字で確認できます。閾値は一般論では決まらないので、まず1〜2週間の分布を見てから、p95やp99を基準に設定するのが現実的です。

本番運用の勘所：機密情報のマスキングとサンプリングによるオーバーヘッド低減

入出力を丸ごと記録すると、ユーザー入力やツール結果に含まれる個人情報や社内データがトレースに残ります。マスキングは二段構えにします。

1. アプリ側: スパン属性に載せる前に、メールアドレスやトークン類を正規表現などで置換します。Langfuse SDKにもマスキング関数を設定する機能があるので、SDK経由で送る場合は併用できます。
2. Collector側: attributes や transform プロセッサで、特定キーの削除やハッシュ化を行います。アプリ側の漏れに対する保険です。

入力全文が不要なツールは、結果サイズやハッシュだけを記録する設計も有効です。

サンプリングは、トラフィックが多い環境でオーバーヘッドとストレージを抑える手段です。単純な比率サンプリング(ヘッドサンプリング)は軽量ですが、障害時の重要なトレースを落とす可能性があります。そこでCollectorのテールサンプリング(tail_sampling プロセッサ)を使い、次の条件のトレースを優先して残します。

1. エラーを含むトレース。
2. 所要時間が閾値を超えたトレース。
3. ステップ数が多いトレース、またはループ疑いのトレース。

それ以外は数%に間引きます。テールサンプリングは、同一トレースの全スパンが同じCollectorインスタンスに届く必要があります。複数台構成では、トレースIDによるルーティングを先に用意してください。バッチ送信は非同期なので、計装自体のレイテンシへの影響は通常小さいですが、本番投入前に負荷試験で実測しておくと安心です。