并行调用函数
处理模型同时调用多个函数的响应，使用 asyncio 并行执行这些函数，并将结果批量合并到一次后续 API 调用中。
并行调用函数 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
什么是并行函数调用
当答案需要来自多个独立来源的信息时，OpenAI 的模型可以在一次响应中同时调用多个函数。工具调用不再按顺序链式执行、每次都等待上一次完成，而是由模型一次发出多个工具调用。您的应用程序可以并行执行它们，然后一并发送所有结果，从而显著降低延迟。
识别并行工具调用
模型发出并行工具调用时，响应消息会包含一个 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。请在一次后续 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 处理同步工具
如果您的工具函数是同步的（使用 requests 而不是异步的 httpx），仍然可以使用 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处理单个工具失败
并行执行工具调用时，某个调用可能失败，而其他调用成功。绝不要让一个失败阻塞其他调用。请在 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 不堪重负。请使用信号量限制并发工具调用的最大数量。这样既能遵守 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.'衡量延迟改善
并行工具调用可以显著降低延迟。如果三个工具调用按顺序执行且每个耗时 500 毫秒，总耗时就是 1500 毫秒。并行运行则可将耗时降至约 500 毫秒，速度提升 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。接下来，我们将使用函数调用构建一个自然语言数据库接口，把普通英语查询转换为 SQL。
常见问题解答
「并行调用函数」课时是免费的吗？
是的 — 「并行调用函数」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「并行调用函数」这节课中我会学到什么？
处理模型同时调用多个函数的响应，使用 asyncio 并行执行这些函数，并将结果批量合并到一次后续 API 调用中。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「并行调用函数」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。