🌊

Strands Agentsはどのようにトレースを送信しているのか？

2025/12/10に公開

本記事はAIエージェント構築＆運用 Advent Calendar 2025の10日目の記事です。
https://qiita.com/advent-calendar/2025/agents
Strands Agentsはトレースを自動でOpenTelemetry(OTEL)に沿って送ってくれるのですが、仕組みを理解したいなと思い、sdk-python(https://github.com/strands-agents/sdk-python) の中を見て、トレースの仕組みを少し深掘りしました。
https://github.com/strands-agents/sdk-python
ちなみに、OTELまったくわかっておらず雰囲気理解をしています

 仕組みの概要簡単にいうとエージェントループの過程で情報(モデルID、入出力、トークン数...)をスパンとして生成し、出力先に送信するという仕組みです。
いくつかの大きな役割の単位がありますが、重要なのはTracerです。
Tracer: スパンの作成と終了
SpanProcessor: スパンのバッファリング
Exporter: 実際にOTELエンドポイントにネットワーク通信

 スパンとイベントStrandsではスパンという単位と、その下のイベントという単位があります。

例えば、LLM実行がひとつのスパンで、そのスパンの中にuser messageやassistant messageという複数のイベントが入っているような構造のイメージです。
スパン: "LLM Call"

├─ イベント1: "gen_ai.user.message"（時刻: 10:00:00.100）

│   データ: { "content": "ユーザーの質問内容..." }

├─ イベント2: "gen_ai.assistant.message"（時刻: 10:00:02.500）

│   データ: { "content": "AIの応答内容..." }

 Tracerの機能と動作Strandsのトレースの仕組みの根幹を担っているのがTracerクラスです。

Tracerはスパンを作成(start_xxx_span)して送信する(end_xxx_span)機能と、スパンにイベントや属性情報を追加する機能(_set_attributes, _add_event)を持っています。この機能がエージェントのループの各ステップで実行されます。
また、スパンの終了はつまり送信(フラッシュ)とイコールです。バッファしておいたものをフラッシュするイメージです。

 トレースの階層構造トレースの属性情報には階層構造があります。

例えば、一番上の階層には実行するエージェント名(gen_ai.agent.name)などが紐付き、その下に複数回のモデル実行(gen_ai.input.messages)が紐づいています。
つまり、スパンには親子関係があります。この親子関係をデータ構造として持ちながらスパンを作成していきます。

 全体の流れClaude に書かせた図です
  ┌─────────────┐
  │   Agent()   │ start_agent_span()
  └──────┬──────┘
         │
         ├─ trace_span を作成（最上位）
         │
         ▼
  ┌──────────────────┐
  │ event_loop_cycle │ start_event_loop_cycle_span(parent_span=agent.trace_span)
  └────────┬─────────┘
           │
           ├─ cycle_span を作成
           │
           ▼
     ┌─────────────┐
     │ Model Call  │ start_model_invoke_span(parent_span=cycle_span)
     └──────┬──────┘
            │
            ├─ model_invoke_span を作成
            ├─ stream_messages() 実行
            ├─ end_model_invoke_span() → force_flush()
            │
            ▼
     ┌─────────────┐
     │ Tool Call   │ start_tool_call_span(parent_span=cycle_span)
     └──────┬──────┘
            │
            ├─ tool_call_span を作成
            ├─ ツール実行
            ├─ end_tool_call_span() → force_flush()
            │
            ▼
  ┌──────────────────┐
  │ Cycle終了        │ end_event_loop_cycle_span()
  └────────┬─────────┘
           │
           └─ force_flush()

  ┌──────────────┐
  │ Agent終了    │ end_agent_span()
  └──────┬───────┘
         │
         └─ force_flush()

 1. Agentスタート(agent.py)エージェント実行時に最上位階層のスパンを作成します

self.trace_span = self._start_agent_trace_span(messages)

そして、with trace_api.use_span(self.trace_span):の中でエージェントを実行することで、その中で作成されたスパンは全て最初に作ったスパンを親に持つことになります。
self.trace_span = self._start_agent_trace_span(messages)
with trace_api.use_span(self.trace_span):
  # ... Agent実行 ...

 2. EventLoop開始(event_loop.py)イベントループの最初にスパンを作成します

記録する情報
サイクルID (event_loop.cycle_id)
全メッセージ履歴（gen_ai.input.messagesイベント）
# Event loop cycle開始
  tracer = get_tracer()
  cycle_span = tracer.start_event_loop_cycle_span(
      invocation_state=invocation_state,
      messages=agent.messages,
      parent_span=agent.trace_span,  # Agent spanを親として指定
      custom_trace_attributes=agent.trace_attributes,
  )

 3. モデル実行(event_loop.py)次に、ループの中でモデル実行が行われるタイミングでスパンを作成し、イベントとして下記の情報を追加します。
リクエスト: gen_ai.request.model, gen_ai.operation.name
レスポンス: メッセージ内容（イベント）
メトリクス:
gen_ai.usage.input_tokens
gen_ai.usage.output_tokens
gen_ai.usage.total_tokens
gen_ai.usage.cache_read_input_tokens
gen_ai.server.time_to_first_token
gen_ai.server.request.duration

停止理由: finish_reason
model_invoke_span = tracer.start_model_invoke_span(
      messages=agent.messages,
      parent_span=cycle_span,  # Cycle spanを親として指定
      model_id=model_id,
      custom_trace_attributes=agent.trace_attributes,
  )

  with trace_api.use_span(model_invoke_span):
      await agent.hooks.invoke_callbacks_async(BeforeModelCallEvent(...))

      # モデル実行
      async for event in stream_messages(...):
          yield event

      stop_reason, message, usage, metrics = event["stop"]

      # Model呼び出し終了（成功時）
      tracer.end_model_invoke_span(
          model_invoke_span,
          message,      # モデルの応答
          usage,        # トークン使用量
          metrics,      # レイテンシ等
          stop_reason   # 停止理由
      )

 4. ツール実行(_executor.py)ツール実行についてもモデル実行と同様の流れでスパンを作成しイベント情報を登録します。
ツール名: gen_ai.tool.name
呼び出しID: gen_ai.tool.call.id
入力: gen_ai.tool.messageイベント
結果: gen_ai.choiceイベント
ステータス: gen_ai.tool.status
# Tool呼び出し開始
  tool_call_span = tracer.start_tool_call_span(
      tool_use,
      cycle_span,  # Cycle spanを親として指定
      custom_trace_attributes=agent.trace_attributes
  )

  with trace_api.use_span(tool_call_span):
      # ツール実行
      async for event in ...:
          yield event

      result = result_event.tool_result

      # Tool呼び出し終了
      tracer.end_tool_call_span(tool_call_span, result)

 5. EventLoop終了end_event_loop_cycle_span()を呼び出し、EventLoop階層のスパンを終了します

 6. Agent終了end_agent_span()を呼び出し、Agent階層のスパンを終了します

 今後の課題: トレーシングのカスタマイズ仕組みがわかったので、トレースの仕方をカスタマイズできるようになることが次の目標です。

Strandsにはデフォルトでトレースをカスタマイズする機能がありますが、この属性はエージェント内で作成される全てのスパンに追加されます。細かい単位や動的なカスタマイズには対応していません。
agent = Agent(
      tools=...,
      trace_attributes={
          "experiment.name": experiment_name,
          "user.id": user_id,
          ...
      }
  )
そこで、例えば、StrandsのHooks機能を使うことでトレースで送信する情報の種類やタイミングを増やせるんじゃないかという仮説があります。それを検証したいです。

仕組みの概要

スパンとイベント

Tracerの機能と動作

トレースの階層構造

全体の流れ

1. Agentスタート(agent.py)

2. EventLoop開始(event_loop.py)

3. モデル実行(event_loop.py)

4. ツール実行(_executor.py)

5. EventLoop終了

6. Agent終了

今後の課題: トレーシングのカスタマイズ

Discussion