1.はじめに
AI Agentの台頭とAgentSDKの登場
昨今流行の「AI Agent」。ChatGPTやGeminiのAPIが登場して久しいですが、Agentの登場により、AIは単なるチャットボットから、複雑な業務プロセスを自律的に判断・遂行する優れモノになりました。
Agentの最大の特徴は、ゴールを与えるだけで自ら計画を立て、ツールを使いこなしながらタスクを完了してくれる「自律的な動き」にあります。しかしこの「自律的な動き」をゼロから実装するのは難しく、会話履歴の管理や、ツール実行結果をLLMに戻すループ処理など裏には複雑な実装が求められます。
そこで登場したのがOpenAIやMicrosoftが提供するAzure AI Agent SDKです。これはpython環境下でAgent構築に必要なオーケストレーション処理をカプセル化したフレームワークであり、「命令(Instructions)」と「使える道具(Tools)」を渡すだけで、数行のコードだけでまるで魔法のように高度なAgentが動いてくれ、だれでも楽々にAgentを開発することができます。
AgentSDKによる完全なブラックボックス化の懸念
AgentSDKを使えば、複雑な処理を意識せずに開発できます。しかし裏を返せばそれは「処理の完全なブラックボックス化」を意味します。
このAgentSDKを業務や本番環境に実装したらどうでしょうか。
インフラエンジニアやバックエンドエンジニアの方なら共感していただけると思いますが、
「本番環境でなぜか動くブラックボックス」
ほど怖いものはありません。例えば代表的な不透明性としては以下があげられます。
- なぜそのツールを選んだのか？ → 判断ロジックの不透明さ
- 並列処理の判断は正しいのか？ → 依存関係の無視リスク
- 見えないコストは発生していないか？ → 隠れたAPIコール
これが一切見えないのです。ちょっと怖すぎますね。
そこで本記事はこのAzure AI Agent SDKの「ブラックボックス」を、ロギング機能を用いて、上記の３つの不透明性に対してそれぞれ検証していきます。
2.検証環境の構築
検証のためのアーキテクチャ
早速AgentSDKの内部処理を可視化するための検証環境を構築します。
今回はツールの選択や、並列処理の判断の思考過程も確認するために「階層型自律エージェント」を構築しました。構築のベースには以下の記事を参考にしました。
構築したのは、日本語から他言語へ翻訳する「上司(Orchestrator)」と「部下(Sub-Agents)」のチーム型エージェントです。
- 上司(Orchestrator) : ユーザーからの指示を受け取り、タスクを分解・計画し適切なSub-Agentsを呼び出す。
- 部下(Sub-Agents) : 「スペイン語翻訳担当」、「フランス語翻訳担当」など特定の機能を担う。また単なる関数ではなく、裏側で実際にChat Completions APIを叩く独立したLLMとして再帰的な構造を持っている。
最初にOrchestratorに特定言語への翻訳の依頼を投げ、それを解釈したOrchestratorが適切なSub-Agentsに投げるという構成です。この構成によりAgent SDKが「複数のツールをどう使い分けるか」、「依存関係のないタスクをどう処理するか」というタスクに対して、どのような判断ロジックをしているか検証します。
ブラックボックスを開ける鍵：内部挙動を追跡するためのロギング設定
ただAgentを動かすだけでは出力結果のみで思考過程は分かりません。そこで内部挙動を可視化するために２つの「仕掛け」を組み込みます。
仕掛けその１：通信ログの可視化
Azure AI Agent SDKは内部でHTTPクライアントライブラリhttpxを使用しています。標準のloggingライブラリを使ってこのhttpxとopenaiのログレベルをDEBUGに引き下げることで、SDKが隠蔽している「生のリクエスト/レスポンス（HTTP Trace）」を全てファイルに出力させます。
# ロガーの設定
import logging
# ファイルに生ログを全て吐き出す設定
logging.basicConfig(
filename='agent_trace.log',
level=logging.DEBUG, # DEBUGレベルで全通信を取得
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# SDKが依存するライブラリのログも強制的に取得
logging.getLogger("httpx").setLevel(logging.DEBUG)
logging.getLogger("openai").setLevel(logging.DEBUG)
またAzure AI Studioには標準でTracing機能があり、AgentSDKの挙動をGUIで追うことは可能です。またRun Steps APIを使えば実行履歴を取得することもできます。しかしこれらはあくまで「SDKによって抽象化されたログ」です。今回はSDKが裏側でどのようなHTTPリクエストを投げ動いているか？という「生の挙動」を全て解明するため、ローカルレベルでの検証を行いました。
仕掛けその２：思考の強制書き出し
今回使用するモデルであるGPT-4oやそれ以降の最新モデルでは思考過程を内部で済ませてしまい、いきなり結果のみを出力する傾向があります。よって「なぜそのツールを選んだか」などの推論プロセスを
System Prompt(instructions)にChain of Thoughtの指示を埋め込み、思考過程の可視化を強制しました。
# 思考を出力してから行動するよう指示
instructions = """
あなたは翻訳チームのオーケストレーターです。
ユーザーの依頼を分析し、適切な専門エージェント(ツール)を呼び出して解決してください。
【重要：思考プロセスを出力すること】
ツールを使用する直前に、必ず以下の形式で「自分の考え」を出力してください。
Thought: ここに、なぜそのツールを選ぶのかという理由を書く。
その後にツールを実行してください。
"""
補足として、GPT-4oはデフォルトでは思考過程を出力しませんが、昨今のOpenAI APIにはモデル内部で推論を行い、その過程を要約して出力できるものも存在します。
しかし、それらのモデルであっても「生の思考トークン」は隠蔽されているケースが多く、完全に透明とは言えません。 システムの挙動を完全に可視化するには生ログを取ってくる手法が有効です。
全体の実装コード
上記の内容をまとめた全体の実装コードは以下になります。
import os
import sys
import time
import json
import logging
from dotenv import load_dotenv
from openai import AzureOpenAI
# ==========================================
# 0. 初期設定
# ==========================================
load_dotenv()
# 必須変数の確認
required_vars = ["AZURE_OPENAI_API_KEY", "AZURE_OPENAI_ENDPOINT", "AZURE_OPENAI_DEPLOYMENT_NAME"]
missing_vars = [var for var in required_vars if not os.getenv(var)]
if missing_vars:
# 変数名が異なる場合のエラー処理
if os.getenv("AZURE_OPENAI_DEPLOYMENT"):
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT")
else:
raise ValueError(f"エラー: .env に以下の変数が足りません: {', '.join(missing_vars)}")
else:
deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")
api_key = os.getenv("AZURE_OPENAI_API_KEY")
endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
api_version = os.getenv("AZURE_OPENAI_API_VERSION", "2024-05-01-preview")
# --- ロガー設定 ---
logger = logging.getLogger("DeepDiveLog")
logger.setLevel(logging.DEBUG)
# ファイル出力（詳細ログ）
file_handler = logging.FileHandler("agent_trace_real.log", mode='w', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
# コンソール出力（進行状況のみ）
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# 裏側の通信ログも取得
logging.getLogger("httpx").setLevel(logging.DEBUG)
logging.getLogger("httpx").addHandler(file_handler)
logging.getLogger("openai").setLevel(logging.DEBUG)
logging.getLogger("openai").addHandler(file_handler)
client = AzureOpenAI(
api_key=api_key,
api_version=api_version,
azure_endpoint=endpoint,
)
# ==========================================
# 1. ツール定義 (サブエージェントの実装)
# ==========================================
def call_spanish_agent(text):
"""スペイン語翻訳エージェント（実体：LLM呼び出し）"""
logger.info(f" [Sub-Agent起動] 🇪🇸 スペイン語担当AIに依頼中... Input: {text}")
# 実際にAPIを叩いて翻訳させる
response = client.chat.completions.create(
model=deployment_name,
messages=[
{"role": "system", "content": "あなたはプロの翻訳家です。入力されたテキストをスペイン語に翻訳してください。解説は不要で、翻訳結果のみを出力してください。"},
{"role": "user", "content": text}
],
temperature=0.3
)
translated_text = response.choices[0].message.content
logger.info(f" [Sub-Agent完了] Output: {translated_text}")
return json.dumps({"translated_text": translated_text, "status": "success"})
def call_french_agent(text):
"""フランス語翻訳エージェント（実体：LLM呼び出し）"""
logger.info(f" [Sub-Agent起動] 🇫🇷 フランス語担当AIに依頼中... Input: {text}")
response = client.chat.completions.create(
model=deployment_name,
messages=[
{"role": "system", "content": "あなたはプロの翻訳家です。入力されたテキストをフランス語に翻訳してください。解説は不要で、翻訳結果のみを出力してください。"},
{"role": "user", "content": text}
],
temperature=0.3
)
translated_text = response.choices[0].message.content
logger.info(f" [Sub-Agent完了] Output: {translated_text}")
return json.dumps({"translated_text": translated_text, "status": "success"})
# Agentに渡すツール定義スキーマ
tools_schema = [
{
"type": "function",
"function": {
"name": "call_spanish_agent",
"description": "スペイン語への翻訳が必要な場合にこのエージェントを呼び出す。",
"parameters": {
"type": "object",
"properties": {"text": {"type": "string", "description": "翻訳元のテキスト"}},
"required": ["text"]
}
}
},
{
"type": "function",
"function": {
"name": "call_french_agent",
"description": "フランス語への翻訳が必要な場合にこのエージェントを呼び出す。",
"parameters": {
"type": "object",
"properties": {"text": {"type": "string", "description": "翻訳元のテキスト"}},
"required": ["text"]
}
}
}
]
# ==========================================
# 2. 解析用関数
# ==========================================
def analyze_run_steps(thread_id, run_id):
logger.info("\n--- 🔍 Agentの思考プロセス解析 (Run Steps) ---")
# API呼び出し時のエラー処理を追加
try:
steps = client.beta.threads.runs.steps.list(thread_id=thread_id, run_id=run_id, order="asc")
for step in steps.data:
step_type = step.type
logger.info(f"\nStep ID: {step.id} [{step_type}]")
if step_type == "message_creation":
msg_id = step.step_details.message_creation.message_id
msg_obj = client.beta.threads.messages.retrieve(thread_id=thread_id, message_id=msg_id)
logger.info(f" 📝 [Thought/Message]:\n {msg_obj.content[0].text.value}")
elif step_type == "tool_calls":
for tool_call in step.step_details.tool_calls:
logger.info(f" 🛠️ [Action]: Tool '{tool_call.function.name}' called with args: {tool_call.function.arguments}")
except Exception as e:
logger.error(f"解析中にエラーが発生しました: {e}")
# ==========================================
# 3. メイン処理
# ==========================================
def run_main():
logger.info("--- プロセス開始 ---")
# Orchestratorの指示（思考プロセスの出力強制）
instructions = """
あなたは翻訳チームのオーケストレーターです。
ユーザーの依頼を分析し、適切な専門エージェント(ツール)を呼び出して解決してください。
【重要：思考プロセスを出力すること】
ツールを使用する直前に、必ず以下の形式で「自分の考え」を出力してください。
Thought: ここに、なぜそのツールを選ぶのかという理由を書く。
その後にツールを実行してください。
"""
assistant = client.beta.assistants.create(
name="OrchestratorBot",
instructions=instructions,
tools=tools_schema,
model=deployment_name,
)
logger.info(f"Agent作成完了: {assistant.id}")
try:
thread = client.beta.threads.create()
# 検証用入力
user_input = "『ありがとう』を、スペイン語とフランス語の両方に翻訳して。"
logger.info(f"\nUser: {user_input}")
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=user_input
)
run = client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id)
# ポーリングループ（time.sleep必須）
while True:
run_status = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
if run_status.status == 'completed':
logger.info("Run完了。")
break
elif run_status.status == 'requires_action':
logger.info("\n>>> Agentがツール利用を要求 (Requires Action) <<<")
tool_outputs = []
for tool_call in run_status.required_action.submit_tool_outputs.tool_calls:
fname = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# 実際にAPIを叩くサブエージェントツールを実行
result_json = ""
if fname == "call_spanish_agent":
result_json = call_spanish_agent(args["text"])
elif fname == "call_french_agent":
result_json = call_french_agent(args["text"])
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": result_json
})
# 結果提出
client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=tool_outputs
)
time.sleep(1) # 待機時間を確保
elif run_status.status in ['failed', 'cancelled', 'expired']:
logger.error(f"Run失敗: {run_status.last_error}")
break
time.sleep(1) # ポーリング間隔
# 解析結果の表示
analyze_run_steps(thread.id, run.id)
logger.info("\n--- 最終回答 ---")
msgs = client.beta.threads.messages.list(thread_id=thread.id, limit=1)
for m in msgs.data:
if m.role == "assistant":
logger.info(m.content[0].text.value)
finally:
client.beta.assistants.delete(assistant.id)
logger.info("\nAgent削除完了")
if __name__ == "__main__":
run_main()
また実行の際にはAzureOpenAIのAPI情報を入れた.envファイルを作成してください。
AZURE_OPENAI_ENDPOINT=https://your-resource-name.openai.azure.com/
AZURE_OPENAI_API_KEY=your-azure-api-key-here
AZURE_OPENAI_DEPLOYMENT=your-deployment-name
AZURE_OPENAI_API_VERSION=2024-02-15-preview
3.ログ解析その１：ReActパターンの確認
検証環境が整ったので、実際にAgentを動かしてみましょう。
今回モデルはgpt-4oを使用します。
Orchestoratorには以下のプロンプトを渡しました。
user_input = "『ありがとう』を、スペイン語とフランス語の両方に翻訳して。"
その結果、Agentの出力は以下が得られました。
翻訳結果です：
- スペイン語: Gracias
- フランス語: Merci
しっかり入力の意図に沿った出力が得られていることが確認できます。
実際のログを見てみる
では、出力が導き出された過程はどうでしょうか？Agentの出力がなぜそうなったかを探るためにログを見ていきましょう。添付コードを動かすとagent_trace_real.logというファイルにログが書き込まれます。
その中でmessage_creationのステップで以下のログが出力されました。
Step ID: step_RU5cnkPtGOm1YJk9iJPtVS1a [message_creation]
# 途中省略
📝 [Thought/Message]:
Thought: この依頼は、「ありがとう」をスペイン語とフランス語両方に翻訳する必要があるため、並行してスペイン語翻訳エージェントとフランス語翻訳エージェントを利用する。
プロンプトで思考を強制させた結果、Agentの思考過程を引き出すことができ、判断ロジックを可視化することができました。
これによりAgentがいきなりツールを実行するのではなく、その直前に明確な推論を行っていることが確認できました。
分析：ReActパターンの確認
このログの挙動は、「ReAct (Reasoning + Acting)」 と呼ばれるパターンそのものです。
ReActとは、2022年に提案されたプロンプトエンジニアリングの手法であり、「Reasoning（推論）」と「Acting（行動）」を組み合わせることで、LLMに複雑なタスクを解決させるアプローチです。
ReActでは、Agentは以下の3つのステップを繰り返すことでタスクを遂行します。
- Reasoning (推論): ゴールに対して、今何をするべきかを考える。
- Acting (行動): 推論に基づき、具体的なツールを実行する。
- Observation (観察): ツールの結果を受け取り、次の手を考える。
ログで見られたAgentの挙動はまさに推論から次の行動に移っていることから、実際は 「ReActループ」を裏側で回し制御していると言えます。
4.ログ解析その２：Parallel Function Callingによる並列処理
前節ではAgentの判断ロジックを可視化しました。この節ではAgentの依存関係の無視リスクに対して、処理の順番を調べることで検証していきます。
先ほどはOrchestoratorに対して以下のタスクを投げました。
user_input = "『ありがとう』を、スペイン語とフランス語の両方に翻訳して。"
結果として正しく回答が返ってきたのですが、この回答の処理のフローはどうなっているのでしょうか？
スペイン語の処理が先なのでしょうか？フランス語の処理が先なのでしょうか？
実際のログを見てみる
というわけでまたログを見てます。
ログのtool_callsステップを確認すると以下の挙動が記録されていました。
Step ID: step_K42j6UhCfHLCbBiohJ6U0vhk [tool_calls]
2026-01-23 02:39:49,854 - DeepDiveLog - INFO - 🛠️ [Action]: Tool 'call_spanish_agent' called with args: {"text": "ありがとう"}
2026-01-23 02:39:49,854 - DeepDiveLog - INFO - 🛠️ [Action]: Tool 'call_french_agent' called with args: {"text": "ありがとう"}
注目すべき点はスペイン語のエージェントとフランス語のエージェントの呼び出しが全く同じStep ID内かつ同じタイミングであることです。
これはAgentが「スペイン語への翻訳」と「フランス語への翻訳」を、順番ではなく同時に実行したことを証明しています。
分析：Parallel Function Callingによる並列処理
この挙動は、OpenAIモデルの機能であるParallel Function Callingと呼ばれるものです。
Agentは直前の「推論（Reasoning）」フェーズにおいて、以下の論理判断を行っています。
- タスク分解: 依頼は「スペイン語訳」と「フランス語訳」の2つだ。
- 依存関係チェック: フランス語訳をするために、スペイン語訳の結果を待つ必要はあるか？ → No（独立している）。
- 実行計画: ならば、1回のターンで両方のツールを呼んでしまった方が速い。
逆説的に言うならば、Parallel Function Callingを使うか判断しているのは、Orchestoratorがタスク間での依存関係を考慮していると言えます。今回の場合はスペイン語とフランス語それぞれへの翻訳作業は依存関係のない並列な作業といえるので、Agentの判断は妥当かつ正しいと考えられます。よってAgentはタスク間での依存関係をしっかり考慮できていることが確認できます。
5.ログ解析その３：内部APIコールによる再帰構造
前節でOrchestratorが「並列実行」を決断した後、実際にSub-Agentsが実行されました。ここで取得したhttpxの生ログに以下の内容が記録されていました。
実際のログを見てみる
そのログが以下になります。
2026-01-23 02:39:35,497 - DeepDiveLog - INFO - [Sub-Agent起動] 🇪🇸 スペイン語担当AIに依頼中... Input: ありがとう
2026-01-23 02:39:35,503 - openai._base_client - DEBUG - Request options: {'method': 'post', 'url': '/deployments/gpt-4o/chat/completions', 'headers': {'api-key': '<redacted>'}, 'files': None, 'idempotency_key': 'stainless-python-retry-e30e1e63-aa05-422a-bef3-66e9772811e6', 'json_data': {'messages': [{'role': 'system', 'content': 'あなたはプロの翻訳家です。入力されたテキストをスペイン語に翻訳してください。解説は不要で、翻訳結果のみを出力してください。'}, {'role': 'user', 'content': 'ありがとう'}], 'model': 'gpt-4o', 'temperature': 0.3}}
Orchestrator自体は Assistants API (/assistants) で動いていますが、ツール実行の裏側で、サブエージェントが独立したChat Completions API (/chat/completions)を叩いていることが分かります。つまりここに隠れたAPIコールが存在しています。
分析：内部APIコールの連鎖
これはOrchestratorから見れば単なる「関数呼び出し」ですが、システム全体で見れば 「AIがAIを呼び出す」再帰的な構造になっています。
- Orchestrator: 「翻訳関数」を実行したつもり。
- Pythonコード: その裏で、別のプロンプト（"あなたは翻訳家です..."）を持ったLLMを起動。
- Sub-Agent: 翻訳を実行し、結果を返す。
このようにAgentの中に別のAgentが含まれる 「マトリョーシカ構造」 が、SDKの抽象化によってあいまいになってしまっています。実際に「なぜか応答が遅い」という時、親の思考が遅いのか、裏で呼ばれたサブエージェントのAPI応答が遅いのか？ ログでこの「連鎖」を追跡することでボトルネックの特定が可能です。
6.検証の結果まとめ
前述の3つの不透明性に対する検証を通して、Azure AI Agent SDKというブラックボックスの中身を生ログを用いて可視化してきました。そこで見えた正体は、決して理解不能な処理などではなく、ReActという堅実な理論に基づき、並列実行や再帰構造を駆使してタスクを遂行する、極めて論理的なオーケストレーションエンジンと言えます。
7.おわりに
今回はAgentでブラックボックスとなる「思考」というプロセスを、ユーザーに見えない内部処理が多いAgentSDKを通じて検証しました。
Agentの挙動が分からなくなってしまった場合は「ログの可視化」「思考過程の可視化」してブラックボックスをこじ開けられるか試してみるのがおすすめです。
そしてもしAgentを開発される場合は、これらの内部挙動を理解・意識した上で効率的に設計をしてみてはいかがでしょうか。
Discussion