はじめに
AIエージェントをAgentCore以外でもECSや軽ければAppRunnerなどで作ることもあるのですが、
そうした際のデバッグのためにエージェント内で何が起きているか、ツールをどう呼んでいるかを確認したい場合が結構あります。
AgentCore Runtime以外と組み合わせることでと親和性が高く、大体使っていたのですが、
AgentCore外のエージェントでもAgentCore Observabilityを利用することができると知り、
どのような設定で利用できるのかを実際に実装しつつ、試してみました。
前提条件
- Python 3.10以上
-
AWSアカウントと認証情報の設定（
aws configure） - CloudWatch Transaction Searchの有効化（初回のみ、下記参照）
初回セットアップ: CloudWatch Transaction Searchの有効化
AgentCore Observabilityのトレースを表示するには、CloudWatch Transaction Searchを有効にする必要があります。これはアカウントごとに一度だけ実行します。
CloudWatchコンソールから有効化する方法
- CloudWatchコンソールを開く
- 左側ナビゲーションの Setup > Settings を選択
- Account を選択し、X-Ray traces タブを開く
- Transaction Search セクションで View settings を選択
- Edit を選択
- Enable Transaction Search を選択
- For X-Ray users を選択し、インデックスするトレースの割合を入力（1%は無料）
- Save を選択
有効化後、スパンが検索可能になるまで約10分かかります。
AWS CLIから有効化する方法
ステップ1: リソースベースポリシーの設定
X-RayがCloudWatch Logsにトレースを送信できるようにポリシーを設定します。
aws logs put-resource-policy \
--policy-name AWSXRayToCloudWatchLogs \
--policy-document '{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "xray.amazonaws.com"
},
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": [
"arn:aws:logs:*:*:log-group:aws/spans*",
"arn:aws:logs:*:*:log-group:/aws/application-signals/data*"
]
}
]
}'
ステップ2: Transaction Searchの有効化
スパン取り込み先をCloudWatch Logsに設定します。
aws xray update-trace-segment-destination --destination CloudWatchLogs
ステップ3: 設定の確認
aws xray get-trace-segment-destination
成功すると以下のレスポンスが返ります:
{
"Destination": "CloudWatchLogs",
"Status": "ACTIVE"
}
有効化後、スパンが検索可能になるまで約10分かかります。
1. 必要な設定やインストール、起動方法
1.1 依存関係
以下のパッケージをインストールします。
pip install aws-opentelemetry-distro bedrock-agentcore
または pyproject.toml / requirements.txt に追加:
dependencies = [
"aws-opentelemetry-distro",
"bedrock-agentcore",
]
Strands Agentsを使用する場合
Strands Agentsフレームワークを使用する場合は、OTEL対応版をインストールしてください:
pip install 'strands-agents[otel]'
1.2 CloudWatchロググループの作成
環境変数を設定する前に、CloudWatchにロググループとログストリームを作成します。
AWS CLIで作成する場合
# ロググループを作成
aws logs create-log-group --log-group-name /aws/agentcore/your-agent-name
# ログストリームを作成
aws logs create-log-stream \
--log-group-name /aws/agentcore/your-agent-name \
--log-stream-name your-agent-name
CloudWatchコンソールから作成する場合
- CloudWatchコンソールを開く
- 左側ナビゲーションの Logs > Log groups を選択
- Create log group を選択
- ロググループ名を入力（例:
/aws/agentcore/your-agent-name） - 作成したロググループを開き、Create log stream でログストリームを作成
1.3 環境変数
以下の環境変数を設定します。
| 環境変数 | 値 | 説明 |
|---|---|---|
AGENT_OBSERVABILITY_ENABLED |
true |
Observabilityの有効化 |
OTEL_PYTHON_DISTRO |
aws_distro |
AWS OpenTelemetryディストリビューション |
OTEL_PYTHON_CONFIGURATOR |
aws_configurator |
AWSコンフィギュレーター |
OTEL_EXPORTER_OTLP_PROTOCOL |
http/protobuf |
OTLPプロトコル設定 |
OTEL_EXPORTER_OTLP_LOGS_HEADERS |
下記参照 | CloudWatch連携ヘッダー |
OTEL_RESOURCE_ATTRIBUTES |
service.name=your-agent-name |
サービス識別情報（GenAI Observabilityダッシュボードでの識別名） |
AWS_DEFAULT_REGION |
us-east-1 等 |
AWSリージョン |
OTEL_EXPORTER_OTLP_LOGS_HEADERS の設定例
x-aws-log-group=/aws/agentcore/your-agent-name,x-aws-log-stream=your-agent-name,x-aws-metric-namespace=bedrock-agentcore
your-agent-name は1.2で作成したロググループ名・ログストリーム名に合わせてください。
1.4 起動方法
opentelemetry-instrument でアプリケーションをラップして起動します。
opentelemetry-instrument uvicorn agent.main:app --host 0.0.0.0 --port 8080
Dockerfileの例:
CMD ["opentelemetry-instrument", "uvicorn", "agent.main:app", "--host", "0.0.0.0", "--port", "8080"]
1.5 必要なIAMポリシー
実行ロール（タスクロール、インスタンスロール等）に以下の権限を付与します。
マネージドポリシー
CloudWatchAgentServerPolicyAWSXRayDaemonWriteAccess
カスタムポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:PutLogEvents",
"logs:DescribeLogStreams"
],
"Resource": "arn:aws:logs:*:*:log-group:/aws/agentcore/your-agent-name:*"
}
]
}
適用先
| サービス | ロール |
|---|---|
| App Runner | インスタンスロール |
| ECS/Fargate | タスクロール |
2. セッション使用について
AgentCore Observabilityでは、OpenTelemetry Baggageを使用してセッションIDを設定できます。これにより、同一セッション内のトレースをグルーピングできます。
実装例
from fastapi import FastAPI
from opentelemetry import baggage, context # (1)
import uuid
app = FastAPI()
@app.post("/invoke")
async def invoke_agent(request: InvokeRequest):
# セッションIDの取得または生成
session_id = request.session_id or str(uuid.uuid4()) # (2)
# OpenTelemetry Baggageにsession.idを設定
ctx = baggage.set_baggage("session.id", session_id) # (3)
# コンテキストをアタッチしてエージェントを実行
token = context.attach(ctx) # (4)
try:
result = await agent.invoke_async(request.prompt)
finally:
context.detach(token) # (5)
return {"response": result, "session_id": session_id}
コード解説
| 番号 | コード | 説明 |
|---|---|---|
| (1) | from opentelemetry import baggage, context |
baggage: トレース間でメタデータを伝播させる機能。context: OpenTelemetryのコンテキスト管理機能 |
| (2) | session_id = request.session_id or str(uuid.uuid4()) |
リクエストにセッションIDがあれば使用し、なければ新規生成。同じセッションIDを持つリクエストはCloudWatch上でグルーピングされる |
| (3) | baggage.set_baggage("session.id", session_id) |
session.id キーでセッションIDを設定。このキー名はAgentCore Observabilityが認識する予約キー |
| (4) | token = context.attach(ctx) |
コンテキストをアタッチし、トークンを取得。アタッチ後に実行されるすべてのスパンにセッションIDが自動的に伝播される |
| (5) | context.detach(token) |
処理完了後、トークンを使用してコンテキストをデタッチ。try/finallyで囲むことで例外発生時も確実にデタッチされる |
3. カスタムスパンの利用について
OpenTelemetryのトレーサーを使用して、カスタムスパンを作成できます。これにより、独自の処理をトレースに記録できます。
実装例: Web検索ツール
from opentelemetry import trace
# サービス名とバージョンを指定してトレーサーを取得
tracer = trace.get_tracer("web_search", "1.0.0") # (1)
def web_search(query: str, max_results: int = 5) -> str:
"""Web検索を実行し、結果をトレースに記録"""
with tracer.start_as_current_span("custom span web search tool") as span: # (2)
try:
# 検索パラメータを属性として記録
span.set_attribute("search.query", query) # (3)
span.set_attribute("search.max_results", max_results)
span.set_attribute("tool.name", "web_search")
span.set_attribute("search.provider", "duckduckgo")
# イベントで処理開始を記録
span.add_event("search_started", {"query": query}) # (4)
# 検索を実行
results = perform_search(query, max_results)
# 結果数を記録
span.set_attribute("search.results_count", len(results))
# 個別の結果属性を追加（上位3件のみ）
for i, result in enumerate(results[:3], 1): # (5)
span.set_attribute(f"search.result.{i}.title", result.get('title', '')[:100])
span.set_attribute(f"search.result.{i}.url", result.get('href', ''))
# イベントで処理完了を記録
span.add_event("search_completed", {
"status": "success",
"results_count": len(results)
})
return format_results(results)
except Exception as e:
# エラーを記録
span.set_attribute("search.error", str(e))
span.record_exception(e) # (6)
span.add_event("search_failed", {"error": str(e)})
raise
コード解説
| 番号 | コード | 説明 |
|---|---|---|
| (1) | trace.get_tracer("web_search", "1.0.0") |
サービス名とバージョンを指定してトレーサーを取得。トレースの整理・フィルタリングに有用 |
| (2) | start_as_current_span("custom span web search tool") |
スパンを開始。この名前はCloudWatchのトレースビューでスパンの識別子として表示される |
| (3) | span.set_attribute("search.query", query) |
スパンに属性（キー/値ペア）を追加。CloudWatchで検索・フィルタリングに使用できる |
| (4) | span.add_event("search_started", {...}) |
スパン内の特定時点にイベントを記録。処理のマイルストーンや状態変化を追跡できる |
| (5) | span.set_attribute(f"search.result.{i}.title", ...) |
動的なキー名で個別の結果を記録。データ量を抑えるため上位3件に限定 |
| (6) | span.record_exception(e) |
例外情報（スタックトレース含む）をスパンに記録。エラーのデバッグに有用 |
まとめ
よいAIエージェントを作成するためには内部挙動を理解することが必要で、そのためにAgentcore Observabilityは強力なツールとなります。
それをAgentcore Runtime以外と組み合わせることで、Runtime以外でもAIエージェントの改善に活かしてもらえればと思います。
(Backend Serverとエージェント処理を1コンテナで構成したいという需要はまだあると思うので。。。)