AIコーディング 2026.07.16
OpenAI Agents SDK for Python 完全ガイド——Function Calling実装とプラクティス【2026年版】
Agents SDKとは——OpenAI最新エージェント実装の仕組み
OpenAI Agents SDKは、ChatGPT APIを使ってAIエージェント（自動的に判断・実行するプログラム）を構築するための公式Pythonライブラリです。Function Calling機能を活用して、AIモデルが関数を選んで実行し、その結果をもとに次のアクションを決定するフローを実現します。
従来のChatGPT APIは「質問と回答」のシンプルなやり取りが中心でしたが、Agents SDKを使うと以下のような複雑な処理が可能になります。
- 複数の関数から最適なものを選択させる: ユーザーの質問に応じて、天気取得API・カレンダー検索・メール送信など、複数の外部機能の中から適切なものを自動選択
- 逐次処理（ステップバイステップ実行）: 1つの質問に対して複数のステップが必要な場合、AIが自動的に順番に関数を呼び出す
- エラー時の自動回復: APIエラーやタイムアウトが起きても、AIが自動的に別の方法を試す
- 監査可能なログ: すべてのエージェント実行ステップを記録でき、何をどの順序で実行したかが追跡可能
2026年現在、ChatGPT APIの開発者向け検索需要の大部分がこのAgents SDK実装に集中しており、従来のFunction Callingの書き方（openai.ChatCompletion）は既にバージョン1.0.0以上では非推奨になっています。
また、複数のエージェントが同時に動作するマルチエージェント構成の需要も急増しており、その実装では会話履歴やコンテキスト管理に起因するメモリ問題が頻発しています。本記事では基本実装に加え、マルチエージェント環境特有の問題と解決パターンについても解説します。
必要な環境と準備作業
システム要件
- Python 3.8以上
- pip（Pythonパッケージマネージャー）
- OpenAI APIキー（OpenAI公式サイトで無料アカウント登録時に取得可能）
Python環境の確認
ターミナル/コマンドプロンプトで以下を実行し、Pythonのバージョンを確認します。
python3 --version
バージョン3.8以上が出力されればOKです。
OpenAI APIキーの取得と設定
- OpenAI公式サイトにアクセスし、アカウントにログインします
- 左メニューから「API keys」をクリック
- 「Create new secret key」ボタンで新しいキーを生成
- 生成されたキーをコピー（二度と表示されないため必ずメモ）
APIキーをPython環境に設定するには、以下いずれかの方法を使います。
方法1：環境変数として設定（推奨）
# macOS / Linux
export OPENAI_API_KEY="sk-..."
# Windows PowerShell
$env:OPENAI_API_KEY="sk-..."
# Windows コマンドプロンプト
set OPENAI_API_KEY=sk-...
方法2：Pythonコード内で直接設定
import os
os.environ['OPENAI_API_KEY'] = 'sk-...'
方法3：.envファイルで管理（本番環境推奨）
プロジェクトルートに.envファイルを作成：
OPENAI_API_KEY=sk-...
Python内で読み込み：
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv('OPENAI_API_KEY')
必要なライブラリのインストール
pip install openai python-dotenv
2026年現在、openaiパッケージのバージョンは1.0.0以上が標準です。バージョンを確認するには：
pip show openai
Function Callingの基本——シンプルな例から始める
Function Callingの動作フロー
Function Callingの本質は「AIにコード実行させる権限を与える」ことです。処理フローは以下の通りです。
- ユーザーが質問を入力（例：「今日の天気は？」）
- AIが「この質問に対しては weather_api を呼び出すべき」と判断
- AIが天気関数を呼び出すように指示
- プログラムが実際に天気APIを呼び出し
- 結果をAIに返す
- AIが結果を人間が理解しやすい文章にまとめて返答
最小限の実装例
以下は、シンプルな計算機能を持つエージェントです。
from openai import OpenAI
import json
# OpenAIクライアント初期化
client = OpenAI()
# ステップ1: 実行可能な関数を定義
def multiply(a: int, b: int) -> int:
"""2つの数を掛け算する"""
return a * b
def add(a: int, b: int) -> int:
"""2つの数を足す"""
return a + b
# ステップ2: 関数の定義をAIに伝える形式に変換
tools = [
{
"type": "function",
"function": {
"name": "multiply",
"description": "2つの整数を掛け算します",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "integer",
"description": "最初の数"
},
"b": {
"type": "integer",
"description": "2番目の数"
}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "add",
"description": "2つの整数を足します",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "integer",
"description": "最初の数"
},
"b": {
"type": "integer",
"description": "2番目の数"
}
},
"required": ["a", "b"]
}
}
}
]
# ステップ3: ユーザーの質問を送信
messages = [
{"role": "user", "content": "12と5を掛けてください"}
]
response = client.chat.completions.create(
model="gpt-4o", # 2026年のデフォルトモデル
messages=messages,
tools=tools,
tool_choice="auto"
)
print("AI返答：", response.choices[0].message.content)
# ステップ4: AIが関数呼び出しを指示してきた場合、実際に実行
if response.choices[0].message.tool_calls:
for tool_call in response.choices[0].message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name == "multiply":
result = multiply(function_args["a"], function_args["b"])
elif function_name == "add":
result = add(function_args["a"], function_args["b"])
print(f"関数実行: {function_name}({function_args}) = {result}")
実行結果：
関数実行: multiply({'a': 12, 'b': 5}) = 60
AI返答： 12と5を掛けると60になります。
実践的なエージェント実装——複数ステップの自動処理
ユースケース：天気情報＋アラート機能
実際の開発では、単一の関数呼び出しではなく、複数のステップを組み合わせる必要があります。以下は、天気情報を取得して、気温が高い場合は警告メールを送る例です。
from openai import OpenAI
import json
from typing import Any
client = OpenAI()
# 実装する関数たち
def get_weather(city: str) -> dict:
"""指定都市の天気情報を取得（ダミー実装）"""
weather_data = {
"Tokyo": {"temp": 28, "condition": "晴れ"},
"Osaka": {"temp": 25, "condition": "曇り"},
"Hokkaido": {"temp": 15, "condition": "雨"}
}
return weather_data.get(city, {"temp": 20, "condition": "不明"})
def send_alert_email(recipient: str, message: str) -> bool:
"""アラートメールを送信（ダミー実装）"""
print(f"📧 メール送信: {recipient} に '{message}' を送信")
return True
def check_alert_threshold(temperature: int) -> bool:
"""気温が警告しきい値を超えているか確認"""
threshold = 27
return temperature > threshold
# AIに提供する関数定義
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "指定された都市の天気情報を取得します",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名（例：Tokyo, Osaka）"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "send_alert_email",
"description": "警告メールを送信します",
"parameters": {
"type": "object",
"properties": {
"recipient": {
"type": "string",
"description": "受信者のメールアドレス"
},
"message": {
"type": "string",
"description": "メールの内容"
}
},
"required": ["recipient", "message"]
}
}
},
{
"type": "function",
"function": {
"name": "check_alert_threshold",
"description": "気温が警告しきい値を超えているか確認",
"parameters": {
"type": "object",
"properties": {
"temperature": {
"type": "integer",
"description": "現在の気温（摂氏）"
}
},
"required": ["temperature"]
}
}
}
]
# メイン処理：複数ステップのエージェント実行
def run_weather_agent(user_query: str, user_email: str):
"""
ユーザーの質問に基づいて天気確認＆アラート判定を自動実行
"""
messages = [
{
"role": "user",
"content": f"{user_query}。気温が27℃を超えていたら {user_email} にアラートメールを送ってください。"
}
]
print(f"\n🤖 エージェント開始: {user_query}\n")
# ループ：AIが「もう関数は不要」と判断するまで繰り返す
while True:
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
# AIの応答をメッセージ履歴に追加
assistant_message = response.choices[0].message
messages.append({"role": "assistant", "content": assistant_message.content})
# Function Callがない場合はループを抜ける
if not assistant_message.tool_calls:
print(f"✅ AI最終返答: {assistant_message.content}\n")
break
# Function Callを実行
for tool_call in assistant_message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"🔧 関数実行: {function_name}({function_args})")
# 関数を実行して結果を取得
if function_name == "get_weather":
result = get_weather(function_args["city"])
print(f" 結果: {result}")
elif function_name == "send_alert_email":
result = send_alert_email(
function_args["recipient"],
function_args["message"]
)
print(f" 結果: メール送信完了")
elif function_name == "check_alert_threshold":
result = check_alert_threshold(function_args["temperature"])
print(f" 結果: 警告が必要={result}")
# 関数実行結果をメッセージ履歴に追加
messages.append({
"role": "user",
"content": json.dumps({"function_result": result})
})
# 実行例
run_weather_agent("東京の天気を教えてください", "admin@example.com")
run_weather_agent("大阪の天気を教えてください", "admin@example.com")
実行出力例：
🤖 エージェント開始: 東京の天気を教えてください
🔧 関数実行: get_weather({'city': 'Tokyo'})
結果: {'temp': 28, 'condition': '晴れ'}
🔧 関数実行: check_alert_threshold({'temperature': 28})
結果: 警告が必要=True
🔧 関数実行: send_alert_email({'recipient': 'admin@example.com', 'message': '東京の気温が28℃に達しています。高温注意'})
📧 メール送信: admin@example.com に '東京の気温が28℃に達しています。高温注意' を送信
結果: メール送信完了
✅ AI最終返答: 東京は現在晴れで、気温は28℃です。警告しきい値を超えているため、アラートメールを送信しました。
🤖 エージェント開始: 大阪の天気を教えてください
🔧 関数実行: get_weather({'city': 'Osaka'})
結果: {'temp': 25, 'condition': '曇り'}
✅ AI最終返答: 大阪は現在曇りで、気温は25℃です。警告しきい値以下のため、アラートメールは送信していません。
複数エージェント実装で発生するメモリ問題
複数のエージェントが同時に動作するマルチエージェント環境では、会話履歴やコンテキスト管理に関するエラーが頻繁に発生します。特に以下のような問題が報告されています。
- トークン超過エラー: 複数エージェントの会話履歴が蓄積され、APIリクエストのトークン数が上限を超える
- メモリリーク: エージェント間で会話履歴が共有され、不正なコンテキストが混入する
- 状態の不整合: 複数エージェントが同じ会話履歴を参照・更新しようとして競合が発生する
- API呼び出しコストの増加: 無駄な履歴データをリクエストに含める結果、コストが急増する
メモリ問題が発生する根本的な原因
単一エージェントで問題なく動作する基本的な会話管理は、マルチエージェント環境では以下の課題を引き起こします。
課題1: メッセージリストの共有による競合
複数エージェントが同じメッセージリストを参照する場合、非同期処理環境で特に顕著な不整合が発生します。
# 問題のあるマルチエージェント実装
messages = [] # すべてのエージェントが共有
async def agent_a_process():
messages.append({"role": "user", "content": "Agent Aからのメッセージ"})
# この間にAgent Bが同じmessagesを修正するかもしれない
response = await openai.ChatCompletion.create(model="gpt-4", messages=messages)
messages.append({"role": "assistant", "content": response['choices'][0]['message']})
async def agent_b_process():
messages.append({"role": "user", "content": "Agent Bからのメッセージ"})
# Agent Aが同時にmessagesを修正するかもしれない
response = await openai.ChatCompletion.create(model="gpt-4", messages=messages)
messages.append({"role": "assistant", "content": response['choices'][0]['message']})
課題2: トークン数の累積によるAPI上限超過
複数エージェントの会話が全て1つのメッセージリストに格納されると、リクエストごとにすべての履歴を送信することになり、トークン数が急増します。
課題3: エージェント間での文脈の混在
複数エージェントが同じ会話履歴を参照すると、本来無関係なやり取りが履歴に含まれ、エージェントが不適切なコンテキストで応答する可能性があります。
マルチエージェント向けメモリ管理の設計パターン
パターン1: エージェント単位の会話履歴分離
最も簡潔で実効的な解決方法は、各エージェントが独立した会話履歴を持つことです。
from dataclasses import dataclass, field
from typing import List
@dataclass
class AgentMemory:
agent_id: str
conversation_history: List[dict] = field(default_factory=list)
def add_user_message(self, content: str):
self.conversation_history.append({
"role": "user",
"content": content
})
def add_assistant_message(self, content: str):
self.conversation_history.append({
"role": "assistant",
"content": content
})
def get_conversation(self) -> List[dict]:
return self.conversation_history.copy()
class MultiAgentSystem:
def __init__(self):
self.agents = {}
def create_agent(self, agent_id: str):
self.agents[agent_id] = AgentMemory(agent_id=agent_id)
def get_agent_memory(self, agent_id: str) -> AgentMemory:
return self.agents[agent_id]
# 使用例
system = MultiAgentSystem()
system.create_agent("agent_a")
system.create_agent("agent_b")
memory_a = system.get_agent_memory("agent_a")
memory_a.add_user_message("Agent Aへの質問")
memory_a.add_assistant_message("Agent Aからの応答")
# Agent Bの会話（Agent Aと完全に独立）
memory_b = system.get_agent_memory("agent_b")
memory_b.add_user_message("Agent Bへの質問")
memory_b.add_assistant_message("Agent Bからの応答")
パターン2: 会話履歴の段階的な削減（スライディングウィンドウ）
エージェント単位でメモリを分離しても、1つのエージェントの会話が長くなるとトークン超過の問題が発生します。最新のN個のメッセージのみを保持し、古いメッセージを削除する方法が有効です。
from collections import deque
@dataclass
class BoundedAgentMemory:
agent_id: str
max_messages: int = 20 # 最新20メッセージまで保持
conversation_history: deque = field(default_factory=deque)
def __post_init__(self):
self.conversation_history = deque(maxlen=self.max_messages)
def add_message(self, role: str, content: str):
self.conversation_history.append({
"role": role,
"content": content
})
def get_conversation(self) -> List[dict]:
return list(self.conversation_history)
スライディングウィンドウを使う場合でも、システムプロンプトで基本的なコンテキストを保持することを推奨します。
SYSTEM_PROMPT = """
あなたは顧客サービスエージェントです。
以下の基本情報を踏まえて応答してください：
- 企業名: XYZ Corporation
- 対応時間: 平日9:00-18:00
- 主な商品: A, B, C
"""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
# ... 最新20個のユーザー・アシスタントメッセージ
]
パターン3: 会話の要約と階層的な履歴管理
長期間の会話を保持する必要がある場合、古い会話を要約に置き換えることで、完全な履歴情報を失わずにトークン数を削減できます。
class SummarizingAgentMemory:
def __init__(self, agent_id: str, message_threshold: int = 30):
self.agent_id = agent_id
self.message_threshold = message_threshold
self.conversation_history = []
self.summary = ""
def add_message(self, role: str, content: str):
self.conversation_history.append({
"role": role,
"content": content
})
if len(self.conversation_history) > self.message_threshold:
self._create_summary()
def _create_summary(self):
messages_to_summarize = self.conversation_history[:10]
summary_prompt = f"""
以下の会話を簡潔に要約してください：
{self._format_messages(messages_to_summarize)}
"""
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=500
)
self.summary = response['choices'][0]['message']['content']
self.conversation_history = self.conversation_history[10:]
def _format_messages(self, messages: List[dict]) -> str:
return "\n".join([
f"{msg['role']}: {msg['content']}"
for msg in messages
])
def get_conversation_with_summary(self) -> List[dict]:
result = []
if self.summary:
result.append({
"role": "system",
"content": f"前の会話の要約:\n{self.summary}"
})
result.extend(self.conversation_history)
return result
Agents SDKで起きやすいエラーと対処法
エラー1：「InvalidRequestError: Unrecognized request argument supplied: messages」
このエラーは通常、古いバージョンのopenaiライブラリで新しいAPI仕様を使う場合に発生します。
原因
openai < 1.0.0（古いバージョン）を使っている場合、APIコールの方法が異なります。
解決策
pip install --upgrade openai
アップグレード後、以下の書き方を使用してください：
from openai import OpenAI
client = OpenAI() # 新しい書き方
response = client.chat.completions.create(
model="gpt-4o",
messages=[...],
tools=[...]
)
古い書き方（非推奨）:
# これは使わない（バージョン1.0.0以上では動作しません）
import openai
openai.ChatCompletion.create(...)
エラー2：Function CallがNoneで返される
AIが関数呼び出しを判断せず、常にtool_calls=Noneになる場合があります。
原因
- 関数定義（tools）がAIに正しく伝わっていない
- ユーザー質問が関数呼び出しを必要としない内容
- tool_choiceパラメータの設定が「auto」になっていない
解決策
# ✅ 推奨：関数呼び出しを強制
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="required" # 必ず関数を呼び出させる
)
# または「auto」で自動判定（デフォルト）
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
エラー3：関数の引数パースエラー
Function Callの引数がJSON形式で返されるため、パースに失敗することがあります。
原因
json.loads()で不正なJSON文字列を処理しようとしている
安全な実装
import json
try:
function_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
print(f"JSON解析エラー: {e}")
print(f"引数文字列: {tool_call.function.arguments}")
function_args = {}
エラー4：「Too many requests」が頻発する
API呼び出しが多すぎる場合、レート制限に引っかかります。マルチエージェント環境では複数エージェントが同時にAPIを呼び出すためこの問題が特に顕著です。
原因
短時間に大量のAPI呼び出しをしている
解決策（単一エージェント）
import time
# 関数実行の間に遅延を挿入
for tool_call in assistant_message.tool_calls:
# 処理...
time.sleep(0.5) # 500ms待機
解決策（マルチエージェント環境）
import asyncio
import time
from typing import Coroutine
class RateLimitedAPIClient:
def __init__(self, requests_per_minute: int = 3500):
self.requests_per_minute = requests_per_minute
self.request_times = []
async def call_api(self, coro: Coroutine):
now = time.time()
self.request_times = [t for t in self.request_times if now - t < 60]
if len(self.request_times) >= self.requests_per_minute:
wait_time = 60 - (now - self.request_times[0])
await asyncio.sleep(wait_time)
self.request_times.append(now)
return await coro
エラー5：「Conversation is too long for the specified model」
マルチエージェント環境で特に頻発するエラーです。
原因: メッセージ総数またはトークン数がモデルの上限を超えている
対処法:
import tiktoken
def count_tokens(messages: List[dict], model: str = "gpt-4") -> int:
encoding = tiktoken.encoding_for_model(model)
total_tokens = 0
for message in messages:
total_tokens += len(encoding.encode(message["content"]))
return total_tokens
def trim_messages(messages: List[dict], max_tokens: int = 6000) -> List[dict]:
while count_tokens(messages) > max_tokens and len(messages) > 1:
messages = messages[1:] # 最初のメッセージから削除
return messages
# 使用例
messages = memory.get_conversation()
messages = trim_messages(messages, max_tokens=6000)
response = client.chat.completions.create(model="gpt-4o", messages=messages)
参考：Claude APIでのFunction Calling相当実装
Claude Tool Useを使った、同等のエージェント実装パターンです。
import anthropic
import json
client = anthropic.Anthropic()
TOOLS = [
{
"name": "get_weather",
"description": "指定した都市の天気を取得する",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
},
{
"name": "send_calendar_invite",
"description": "カレンダーに予定を追加する",
"input_schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"date": {"type": "string"}
},
"required": ["title", "date"]
}
}
]
def execute_tool(name: str, inputs: dict) -> str:
if name == "get_weather":
return json.dumps({"city": inputs["city"], "weather": "晴れ", "temp": 25})
elif name == "send_calendar_invite":
return json.dumps({"status": "added", "title": inputs["title"]})
return json.dumps({"error": "unknown tool"})
def run_agent(user_request: str, max_steps: int = 5) -> str:
"""複数ステップの自動処理（監査ログ付き）"""
messages = [{"role": "user", "content": user_request}]
audit_log = []
for step in range(max_steps):
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=TOOLS,
messages=messages
)
if response.stop_reason == "end_turn":
return response.content[0].text
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_tool(block.name, block.input)
audit_log.append({"step": step, "tool": block.name, "input": block.input, "result": result})
tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
return "最大ステップ数に到達しました"
# 使用例
result = run_agent("東京の天気を確認して、晴れなら明日のピクニックの予定を追加して")
print(result)
Claude APIのTool Useは、OpenAI Function Callingとほぼ同等の構造を持ち、stop_reasonをチェックしながら複数ステップの処理を反復実行する点も共通しています。
ベストプラクティス——本番環境での実装ガイドライン
あわせて読みたい
- Claude OpusとSonnetを1つのAPIで連携させる方法｜コスト削減＆高速化
- PythonでMCPサーバーを構築する方法【2026年版】5ステップ実装ガイド
- Claude Codeで請求書自動処理システムを構築｜1日2,000件対応コード実装ガイド
参考ソース
- 10 AI Tools Every Developer Should Try in 2026
- Auditable Agent Context
- OpenAI API error: You tried to access openai.ChatCompletion, but this is no longer supported in openai>=1.0.0
- Upload an image to chat gpt using the API?
- How to continue incomplete response of openai API
- How to send longer text inputs to ChatGPT API?
- OpenAI Chat Completions API error: InvalidRequestError: Unrecognized request argument supplied: messages
