関数の並列呼び出し
モデルが複数の関数を同時に呼び出すレスポンスを処理し、asyncioで並列実行したうえで、結果を1回の後続API呼び出しにまとめます。
「関数の並列呼び出し」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。
並列関数呼び出しとは
OpenAIのモデルは、複数の独立した情報源から情報が必要な場合、1つのレスポンスで複数の関数を同時に呼び出すことができます。前のツール呼び出しの完了を待ちながら順番に呼び出すのではなく、モデルは複数のツール呼び出しを一度に生成します。アプリケーションはそれらを並列実行し、すべての結果をまとめて返すことで、遅延を大幅に短縮できます。
並列ツール呼び出しを認識する
モデルが並列ツール呼び出しを行うと、レスポンスメッセージのtool_callsリストに複数のエントリが含まれます。各エントリには、一意のid、関数名、引数があります。後続のAPI呼び出しを行う前に、すべての呼び出しを処理する必要があります。モデルは、発行したすべてのツール呼び出しに対する結果を期待しているためです。
from openai import OpenAI
import json
client = OpenAI()
# A question that naturally requires two independent lookups
response = client.chat.completions.create(
model='gpt-4o',
messages=[{'role': 'user', 'content': 'Compare the weather in London and Tokyo right now.'}],
tools=tools
)
message = response.choices[0].message
print('Number of tool calls:', len(message.tool_calls))
# Might print: Number of tool calls: 2
for tc in message.tool_calls:
print(f' {tc.function.name}({tc.function.arguments})')
# get_current_weather({"location": "London"})
# get_current_weather({"location": "Tokyo"})asyncioでツール呼び出しを実行する
asyncio.gather()を使用して、複数のツール呼び出しを並行して実行してください。各ツール関数は別々のコルーチンで実行され、すべてのコルーチンが完了すると結果が収集されます。各ツール呼び出しがネットワークリクエストを行う場合、順次実行するよりもはるかに高速です。
import asyncio
import json
async def execute_tool_call_async(tool_call) -> tuple:
'''Execute a single tool call and return (tool_call_id, result).'''
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# Async version of your tool (uses httpx, aiohttp, etc.)
if name == 'get_current_weather':
result = await async_get_weather(**args)
elif name == 'get_stock_price':
result = await async_get_stock(**args)
else:
result = f'Unknown tool: {name}'
return tool_call.id, str(result)
async def execute_all_parallel(tool_calls) -> list:
'''Execute all tool calls concurrently.'''
tasks = [execute_tool_call_async(tc) for tc in tool_calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return resultsすべての結果をまとめて返す
並列ツール呼び出しをすべて実行したら、各結果を個別のrole='tool'メッセージとして会話に追加してください。各メッセージには対応するtool_call_idを含める必要があります。モデルがすべての結果を統合して一貫した回答を生成できるよう、これらを1回の後続API呼び出しでまとめて送信してください。
async def run_parallel_tool_calls(user_message: str) -> str:
messages = [{'role': 'user', 'content': user_message}]
response = client.chat.completions.create(
model='gpt-4o', messages=messages, tools=tools
)
assistant_message = response.choices[0].message
messages.append(assistant_message) # Add assistant's tool_calls
if response.choices[0].finish_reason == 'tool_calls':
# Execute all tool calls in parallel
results = await execute_all_parallel(assistant_message.tool_calls)
# Add all results to conversation
for tool_call_id, result in results:
messages.append({
'role': 'tool',
'tool_call_id': tool_call_id,
'content': result
})
# One more API call to synthesize results
final = client.chat.completions.create(model='gpt-4o', messages=messages)
return final.choices[0].message.content
return assistant_message.content同期ツールにThreadPoolExecutorを使用する
ツール関数が同期処理（非同期のhttpxではなくrequestsを使用）であっても、concurrent.futures.ThreadPoolExecutorを使って並列実行できます。設定は簡単ですが、完全な非同期コードよりもやや効率が落ちます。
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
def execute_all_with_threads(tool_calls) -> list:
results = []
with ThreadPoolExecutor(max_workers=len(tool_calls)) as executor:
future_to_id = {
executor.submit(execute_tool_call, tc): tc.id
for tc in tool_calls
}
for future in as_completed(future_to_id):
tool_call_id = future_to_id[future]
try:
result = future.result(timeout=15)
except Exception as e:
result = f'Tool failed: {str(e)}'
results.append((tool_call_id, str(result)))
return results個別のツール失敗を処理する
ツール呼び出しを並列実行すると、一部が失敗しても他が成功することがあります。1つの失敗によって他の処理を止めてはいけません。asyncio.gather()でreturn_exceptions=Trueを使用すると、一部で例外が発生してもすべての結果を収集できます。例外をエラー文字列に変換して、モデルがすべての結果を受け取り、部分的な失敗について推論できるようにしてください。
async def safe_execute_all(tool_calls) -> list:
tasks = [execute_tool_call_async(tc) for tc in tool_calls]
raw_results = await asyncio.gather(*tasks, return_exceptions=True)
results = []
for tc, result in zip(tool_calls, raw_results):
if isinstance(result, Exception):
results.append((tc.id, f'Tool error: {str(result)}'))
else:
tool_call_id, output = result
results.append((tool_call_id, output))
return resultsモデルが並列呼び出しを使用する場合
モデルは、必要な情報を独立して取得できると判断した場合に、並列ツール呼び出しを行います。つまり、一方の結果が他方に依存しない場合です。例として、複数のティッカーの株価を取得する、複数の都市の天気を取得する、複数のデータベーステーブルをクエリする、といった処理があります。結果に依存関係がある場合は順次呼び出しになります。たとえば、まずユーザーIDを検索し、そのIDの注文を取得する場合です。
並列呼び出しをレート制限する
多数のツール呼び出しを並列実行すると、同時リクエストによって外部APIに過大な負荷がかかることがあります。セmaphoreを使用して、同時実行できるツール呼び出し数の上限を設定してください。これにより、APIのレート制限を守りながら、完全な順次処理より効率的に実行できます。
import asyncio
async def rate_limited_execute_all(tool_calls, max_concurrent: int = 5) -> list:
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_call(tc):
async with semaphore:
return await execute_tool_call_async(tc)
tasks = [limited_call(tc) for tc in tool_calls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
(tc.id, str(r) if not isinstance(r, Exception) else f'Error: {r}')
for tc, r in zip(tool_calls, results)
]順次呼び出しと並列呼び出しを組み合わせる
実際のエージェントワークフローでは、順次呼び出しと並列呼び出しを組み合わせることがよくあります。モデルはまずlookup_user(email)を呼び出し、返されたuser_idを使ってget_orders(user_id)とget_preferences(user_id)を並列に呼び出す場合があります。外側のループで、並列化できる呼び出し（独立）と順次実行が必要な呼び出し（依存）を判別するよう実装してください。
async def multi_round_agent(user_message: str) -> str:
messages = [{'role': 'user', 'content': user_message}]
MAX_ROUNDS = 5
for _ in range(MAX_ROUNDS):
response = client.chat.completions.create(
model='gpt-4o', messages=messages, tools=tools
)
choice = response.choices[0]
messages.append(choice.message)
if choice.finish_reason == 'stop':
return choice.message.content # Done
if choice.finish_reason == 'tool_calls':
# Execute all tool calls in parallel (may be 1 or many)
results = await safe_execute_all(choice.message.tool_calls)
for tc_id, result in results:
messages.append({'role': 'tool', 'tool_call_id': tc_id, 'content': result})
# Loop continues for potentially sequential next call
return 'Max rounds reached.'遅延の改善を測定する
並列ツール呼び出しによって、遅延を大幅に短縮できます。3つのツール呼び出しを順次実行すると、それぞれ500msの場合、合計は1500msです。並列実行すれば約500msとなり、3倍高速になります。並行処理のオーバーヘッドやレート制限の制約を考慮しながら、実際のシナリオで順次実行と並列実行を必ず測定・比較してください。
import time
import asyncio
async def benchmark_parallel_vs_sequential():
tool_calls = [...] # 5 independent tool calls
# Sequential
start = time.time()
for tc in tool_calls:
await execute_tool_call_async(tc)
sequential_time = time.time() - start
# Parallel
start = time.time()
await asyncio.gather(*[execute_tool_call_async(tc) for tc in tool_calls])
parallel_time = time.time() - start
print(f'Sequential: {sequential_time:.2f}s')
print(f'Parallel: {parallel_time:.2f}s')
print(f'Speedup: {sequential_time/parallel_time:.1f}x')並列ツール呼び出しをデバッグする
並列ツール呼び出しの問題をデバッグするときは、完全なリクエストとレスポンスの流れをログに記録してください。これには、すべてのtool_callsを含むモデルのメッセージ、会話に追加された各結果、最終的なモデルのレスポンスが含まれます。ツールの結果が欠落していたり順序が違っていたりすると、モデルが一貫性のない回答を生成する可能性があります。tool_call_idを相関キーとする構造化ログを使うと、問題を簡単に追跡できます。
クイックチェック
OpenAIでの並列関数呼び出しについて、理解度を確認しましょう。
レッスンのまとめ
このレッスンでは、並列のツール呼び出しは tool_calls リスト内の複数のエントリとして現れること、asyncio.gather は最大限の速度でそれらを同時実行すること、そして各結果をモデルに返す際には、対応する tool_call_id が必要であることを学びました。次は、function calling を使って自然言語のクエリを SQL に変換する、自然言語データベースインターフェースを構築します。
よくある質問
「関数の並列呼び出し」レッスンは無料ですか？
はい。「関数の並列呼び出し」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。
「関数の並列呼び出し」で何を学びますか？
モデルが複数の関数を同時に呼び出すレスポンスを処理し、asyncioで並列実行したうえで、結果を1回の後続API呼び出しにまとめます。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Engineering Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「関数の並列呼び出し」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Engineering Academyレッスンでコードを書いて実行できますか？
はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。