AIエージェント構築＆運用 Advent Calendar 2025のシリーズ１の24日目の投稿です
アドベントカレンダーに申し込んだものの、ネタがなく、他の人の投稿を見ていると、意外とOpenAIのAgents SDKに関するものがないことに気づきまして、これをBedrock AgentCoreで使うと特徴が出るかな、とおもってやってみましたの巻
BedrockとOpenAI Agents SDKを組み合わせて使う
BedrockではOpenAIのモデルが２つ提供されています。
- gpt-oss-20b
- gpt-oss-120b
BedrockはInvokeModel APIとConverse APIという２つのAPIで呼び出します。OpenAIモデルについてはこれらに加え、OpenAI社提供のAPIと同じ形式のChat completions APIとResponses APIでも呼び出すことが可能です。
Responses APIについては今月サポートが開始されました🎉🎉🎉
Amazon Bedrock が OpenAI の Responses API をサポートするようになりました
https://aws.amazon.com/jp/about-aws/whats-new/2025/12/amazon-bedrock-responses-api-from-openai/
まずは、BedrockのResponses APIを呼び出し、OpenAI Agents SDKを使用します。
BedrockでAPIキーを生成する
AWSマネジメントコンソールでBedrockの管理画面を開き、左のメニューの「API キー」を選択します。
「短期APIキー」タブにある「短期APIキーを生成」をクリックします。
表示されるAPIキーをメモします。
短期APIキーの有効期限は発行から12時間です。
APIキーはリージョンに紐づいて発行されます。バージニア北部リージョンで発行したAPIキーでは東京リージョンのBedrockにアクセスすることはできません。
Pythonのプロジェクトを作成する
Pythonのプロジェクトを作成します。
uv init
続いて、OpenAI Agents SDKをインストールします。
uv add openai-agents
Pythonコードを実装する
main.pyに実装していくのですが、exampleとして用意されているhello_world.pyをベースとします。
main.py（hello_world.pyから引用）
import asyncio
from agents import Agent, Runner
async def main():
agent = Agent(
name="Assistant",
instructions="You only respond in haikus.",
)
result = await Runner.run(
agent, "プログラミングにおける再帰について教えてください。"
)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
このままではOpenAI社のAPIにアクセスするので、これをBedrock宛に変更しましょう。
まず、インポート部分にいくつか追加します。
import asyncio
- from agents import Agent, Runner
+ from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
+ from openai import AsyncOpenAI
追加したset_default_openai_clientとset_default_openai_apiを使ってデフォルトのOpenAIクライアントと使用するAPIを指定します。
-
base_urlをhttps://bedrock-mantle.{リージョン}.api.aws/v1に -
set_default_openai_apiで使用するAPIのタイプをResponses APIに（デフォルト値はchat_completions）
OPENAI_API_KEY = "bedrock-api-key-*****" # Bedrock API Keyで発行したAPIキー
set_default_openai_client(
client=AsyncOpenAI(
base_url="https://bedrock-mantle.us-east-1.api.aws/v1",
api_key=OPENAI_API_KEY,
)
)
set_default_openai_api("responses")
最後に、呼び出すモデル名を指定します。
agent = Agent(
name="Assistant",
+ model="openai.gpt-oss-120b",
instructions="You only respond in haikus.",
)
Responses APIを使用するときだけ、モデル名の表記が違うのでご注意ください。
- Responses API
openai.gpt-oss-120bopenai.gpt-oss-20b
- InvokeModel API、Converse API、Chat completions API
openai.gpt-oss-120b-1:0openai.gpt-oss-20b-1:0
どういう意図なんでしょうね
これで完成です。
main.py
main.py
import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from openai import AsyncOpenAI
OPENAI_API_KEY = "bedrock-api-key-*****" # Bedrock API Keyで発行したAPIキー
set_default_openai_client(
client=AsyncOpenAI(
base_url="https://bedrock-mantle.us-east-1.api.aws/v1",
api_key=OPENAI_API_KEY,
)
)
set_default_openai_api("responses")
async def main():
agent = Agent(
name="Assistant",
model="openai.gpt-oss-120b",
instructions="You only respond in haikus.",
)
result = await Runner.run(
agent, "プログラミングにおける再帰について教えてください。"
)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
実行してみます。
uv run main.py
回答
## プログラミングにおける再帰 (Recursion) とは？
**再帰** とは、**関数（または手続き）が自分自身を呼び出すこと**です。
「自分自身を呼び出す」ことで、問題を **「より小さな同じ形の問題」** に分割し、最終的に **最も簡単な状態（＝終了条件）** まで到達させてから答えを組み立てていく手法です。
---
## 1. 再帰の基本構造
```text
function recursive(args):
if 終了条件 (base case) が満たされたら
→ 直接答えを返す
else
→ 問題を「小さく」した引数で自分自身を呼び出す
→ その戻り値を使って最終結果を作る
```
### 重要なポイント
| 要素 | 役割 | 例 |
|------|------|----|
| **終了条件 (base case)** | 再帰が止まる条件。無限に呼び出し続けないように必ず必要。 | `n == 0` のとき 0 を返す |
| **縮小 (reduction)** | 呼び出しごとに問題サイズを必ず **減らす**（あるいは変化させる）。 | `n-1`、`n/2`、配列の先頭を除く 等 |
| **スタック** | 各呼び出しの局所変数や戻りアドレスは **コールスタック** に積まれる。スタックが溢れると **StackOverflowError** が起きる。 | 再帰深さが 10⁶ 以上になると危険 |
---
## 2. 代表的な例
### 2‑1. 階乗 (Factorial)
```python
def factorial(n: int) -> int:
# 終了条件
if n == 0:
return 1
# 再帰呼び出し + 縮小
return n * factorial(n - 1)
print(factorial(5)) # 120
```
- `5! = 5 × 4!` → `4!` は `factorial(4)` が計算
- `0! = 1` が **base case**。
---
### 2‑2. フィボナッチ数列
**単純再帰（指数時間）**
```python
def fib(n: int) -> int:
if n <= 1: # base case
return n
return fib(n-1) + fib(n-2)
print([fib(i) for i in range(10)]) # [0,1,1,2,3,5,8,13,21,34]
```
**メモ化（キャッシュ）で高速化**
```python
from functools import lru_cache
@lru_cache(maxsize=None)
def fib_memo(n: int) -> int:
if n <= 1:
return n
return fib_memo(n-1) + fib_memo(n-2)
print(fib_memo(50)) # 12586269025
```
- キャッシュを使うと **線形時間 O(n)** に改善できる。
---
### 2‑3. 二分探索 (Binary Search)
```cpp
int binarySearch(const vector<int>& a, int left, int right, int target) {
if (left > right) return -1; // base case: not found
int mid = left + (right - left) / 2;
if (a[mid] == target) return mid; // found
if (a[mid] > target)
return binarySearch(a, left, mid - 1, target); // left half
else
return binarySearch(a, mid + 1, right, target); // right half
}
```
- 配列を **半分** に分割し続けるので、再帰深さは `log₂(N)`。
---
## 3. 再帰のメリットとデメリット
| メリット | デメリット |
|----------|------------|
| **コードがシンプル** - 同じ構造を繰り返す問題は自然に表現できる | **スタック消費** - 深すぎると `StackOverflow` |
| **問題分割が明確** - Divide‑and‑Conquer がそのまま書ける | **遅いことがある** - 重複計算や非最適化が起きやすい |
| **数学的な証明がしやすい**（帰納法と相性が良い） | **デバッグが難しい** - 呼び出し階層が多くなると追いにくい |
| **関数型言語と相性が良い**（副作用が少ない） | **尾再帰最適化が無い言語では非効率** |
---
## 4. 尾再帰 (Tail Recursion) と TCO (Tail Call Optimization)
### 4‑1. 尾再帰とは？
**最後の操作が再帰呼び出し** である形のこと。
例: 階乗の「累積結果」を引数に持たせる形
```python
def fact_tail(n: int, acc: int = 1) -> int:
if n == 0:
return acc # ここが最終結果、再帰は呼ばれない
return fact_tail(n-1, acc * n)
```
### 4‑2. TCO が有効な言語
- **Scheme, Haskell, OCaml, F#, Rust (some cases), C# (.NET 6+), Scala**
- **JavaScript** (ES2022 の一部実装で最適化が入っているが、保証はない)
- **C** / **C++** ではコンパイラの最適化オプション `-O2` 以上で「tail–call」最適化が入ることがあるが、保証は少ない。
**TCO が無い場合** はスタックを消費し続けるので、**ループに書き換える** か **手動でスタックをシミュレート**（`while` ループ + 変数）する必要があります。
---
## 5. 再帰 → 反復 (Iteration) への変換例
### 階乗のループ版（C++）
```cpp
long long factorial_iter(int n) {
long long acc = 1;
while (n > 0) {
acc *= n;
--n;
}
return acc;
}
```
### フィボナッチのループ版（Python）
```python
def fib_iter(n: int) -> int:
a, b = 0, 1
for _ in range(n):
a, b = b, a + b
return a
```
- **再帰的に書くと直感的** だけど、**反復はスタックを使わない**ので安全。
---
## 6. 再帰のデバッグテクニック
| 手法 | 説明 |
|------|------|
| **プリントデバッグ** | `print(f"call n={n}")` で入出力を追う（深さが大きいと膨大になる） |
| **スタックトレース** | 例外発生時に表示されるコールスタックを読む |
| **IDE のデバッガ** | ブレークポイントを再帰呼び出し行に置くと「再帰的にステップイン」できる |
| **再帰深さ制限** | Python なら `sys.setrecursionlimit()`、C++ なら `-Wstack-usage=` で上限を把握 |
| **メモ化やキャッシュ** | 同じ引数で何度も計算しないようにすると **実行回数が減り** デバッグが楽になる |
---
## 7. 再帰が特に有効な典型問題
| カテゴリ | 例 | 解法のヒント |
|----------|----|--------------|
| **木構造・グラフ** | 二分木の走査 (preorder/inorder/postorder) | ノードを「左子」「右子」へ再帰 |
| **分割統治** | クイックソート、マージソート | 配列を半分に分割し再帰的にソート |
| **組み合わせ列挙** | すべての列挙（subset, permutation, combination） | 現在の「選択/不選択」状態を引数で渡す |
| **DP（メモ化再帰）** | ナップサック、最長共通部分列 | 再帰で状態遷移を書き、`memo` で結果を保存 |
| **文字列処理** | パリンドローム判定、正規表現の簡易実装 | 文字列の先頭・末尾を取り除いて再帰 |
---
## 8. まとめ：再帰を使うかどうかの判断基準
| 条件 | 再帰が向いている | 再帰が向いていない（ループが好ましい） |
|------|------------------|---------------------------------------|
| **問題が自己相似**（同じ形が繰り返す） | ✅ | ❌ |
| **深さが予測でき、**`log n` か `n` 程度 | ✅（スタックが安全） | ❌ |
| **深さが `10⁴`‑`10⁶` 以上になる可能性** | ❌（スタックオーバーフロー） | ✅（ループ化） |
| **言語が尾再帰最適化を保証** | ✅ | ❌ |
| **計算量が指数的で、****メモ化が有効** | ✅（DP + 再帰） | ❌（手実装のループでも可） |
| **可読性や実装の簡潔さが重要** | ✅ | ❌ |
---
## 参考リソース（日本語）
| タイトル | URL |
|----------|-----|
| 「アルゴリズム入門」 第4章 再帰と分割統治 | https://algorithm-bootcamp.com/recursion |
| Python 公式ドキュメント – 再帰とスタック | https://docs.python.org/ja/3/tutorial/controlflow.html#recursive-functions |
| 「プログラミングコンテストで学ぶアルゴリズム」 – 再帰 & DP | https://atcoder.jp/contests/arc001/editorial/ |
| 「Tail Call Optimization」 – Wikipedia (日本語) | https://ja.wikipedia.org/wiki/末尾再帰最適化 |
---
**最後に**
再帰は「**問題を小さく分割して同じ手続きを繰り返す**」という自然な考え方です。
*「この問題は、最小単位まで細かくしたらどうなる？」* を常に意識すれば、再帰的な解法が浮かんできます。
ただし **スタック使用量** と **重複計算** に注意し、必要に応じて **メモ化** や **ループ化** を併用すれば、効率的で読みやすいコードが書けるようになります。
質問や実際に書きたいコードがあれば、遠慮なく聞いてください！
この時点の実装では、以下のようなエラーメッセージが複数表示されますが、後の手順で解消します。
[non-fatal] Tracing client error 401: {
"error": {
"message": "Incorrect API key provided: bedrock-*****. You can find your API key at https://platform.openai.com/account/api-keys.",
"type": "invalid_request_error",
"param": null,
"code": "invalid_api_key"
}
}
Bedrock AgentCoreを追加する
それでは更にAgentCoreを追加します。追加する機能は以下の２つです。（Runtimeのデプロイはちょっと力尽きたのでまたの機会に。。）
- AgentCore Identity
- AgentCore Runtime
- AgentCore Observability
OPENAI_API_KEYをAgentCore Identityで管理
先程の実装、OPENAI_API_KEYがソースコードに直書きされていて、あまり美しくありません。APIキーの管理に「AgentCore Identity」を使います。
AWSマネジメントコンソールでBedrock AgentCoreの管理画面を開き、左のメニューの「アイデンティティ」を選択します。
「OAuthクライアント/APIキーを追加」ボタンを押したうえで「APIキーを追加」をクリックします。
APIキーを追加ダイアログで「名前」「APIキー」を入力します。ここの名前は後で使用します。
AWS側はこれで完了です。Pythonのプログラムを修正していきます。まずは、Bedrock AgentCore SDKを追加します。
uv add bedrock-agentcore
Bedrock AgentCore SDKを使って、AgentCore IdentityからAPIキーを取得します。
requires_api_keyというデコレーターが用意されており、これを関数に付与することで、AgentCore IdentityからAPIキーを取得することができます。
provider_nameは、AgentCore IdentityにAPIキーを登録する際に指定した名前です。
import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
+ from bedrock_agentcore.identity import requires_api_key
from openai import AsyncOpenAI
+ @requires_api_key(provider_name="AWS_BEARER_TOKEN_BEDROCK")
+ async def need_api_key(*, api_key: str):
+ print("received api key for async func")
+ return api_key
APIキーが必要な部分で、作成した関数を呼び出します。（awaitをつけて呼び出す関係上、main()関数の中に移動しました）
- set_default_openai_client(
- client=AsyncOpenAI(
- base_url="https://bedrock-mantle.us-east-1.api.aws/v1",
- api_key=OPENAI_API_KEY,
- )
- )
- set_default_openai_api("responses")
async def main():
+ set_default_openai_client(
+ client=AsyncOpenAI(
+ base_url="https://bedrock-mantle.us-east-1.api.aws/v1",
+ api_key=await need_api_key(), # APIキーを取得してセット
+ )
+ )
+ set_default_openai_api("responses")
agent = Agent(
name="Assistant",
model="openai.gpt-oss-120b",
instructions="You only respond in haikus.",
)
これでAgentCore Identityの対応は完了です。
RuntimeとObservabilityを追加
最後にRuntimeとObservabilityを追加します。
まず、Bedrock AgentCore SDKを使ってRuntimeに合った構成へ変更します。
- import asyncio
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from bedrock_agentcore.identity import requires_api_key
+ from bedrock_agentcore.runtime import BedrockAgentCoreApp
from openai import AsyncOpenAI
～中略～
+ app = BedrockAgentCoreApp()
～中略～
- async def main():
+ @app.entrypoint
+ async def invoke(payload: dict):
+ prompt = payload.get("prompt")
～中略～
result = await Runner.run(
- agent, "プログラミングにおける再帰について教えてください。"
+ agent, prompt
)
- print(result.final_output)
+ return result.final_output
～中略～
if __name__ == "__main__":
- asyncio.run(main())
+ app.run()
これでWebサーバーとして起動するようになります。
サーバーとして起動
uv run main.py
別のターミナルからリクエストを送信します。
クライアントからリクエストを送信
curl -XPOST http://localhost:8080/invocations \
-d '{"prompt": "プログラミングにおける再帰について教えてください。"}'
では、Observabilityを追加しましょう。
こちらの「CloudWatchの設定変更」を参考に、CloudWatchのトランザクション検索を有効にしてください。
OpenAI Agents SDKのテレメトリーデータを収集するライブラリーとAWSに送信するライブラリーです。
uv add aws-opentelemetry-distro openinference-instrumentation-openai-agents
OpenAI Agents SDKのOpenTelemetry計装を有効にし、INFOレベルのログを出力する設定を行います。
+ import logging
from agents import Agent, Runner, set_default_openai_api, set_default_openai_client
from bedrock_agentcore.identity import requires_api_key
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from openai import AsyncOpenAI
+ from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor
+ OpenAIAgentsInstrumentor().instrument()
+ logging.basicConfig(level=logging.INFO)
+ logger = logging.getLogger(__name__)
その他の設定は環境変数で行います。サーバー側の起動時に色々環境変数をセットします。
CloudWatchのロググループとストリームは事前に作成しておきます
export AGENT_OBSERVABILITY_ENABLED=true
export OTEL_PYTHON_DISTRO=aws_distro
export OTEL_PYTHON_CONFIGURATOR=aws_configurator
export SERVICE_NAME=hello-world-agent
export LOG_GROUP=/aws/bedrock-agentcore/local
export LOG_STREAM=runtime-logs
export OTEL_RESOURCE_ATTRIBUTES="service.name=${SERVICE_NAME},aws.log.group.names=${LOG_GROUP}"
export OTEL_EXPORTER_OTLP_LOGS_HEADERS="x-aws-log-group=${LOG_GROUP},x-aws-log-stream=${LOG_STREAM},x-aws-metric-namespace=bedrock-agentcore"
export OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf
export OTEL_TRACES_EXPORTER=otlp
uv run opentelemetry-instrument python ./main.py
めでたく、トレースが確認できました。
Agents SDKの使い方はあまりわからないままなので、引き続きやってみようと思います