OpenAI Responses API 写作如何处理并行工具调用

2026-08-28 9 分钟阅读 AI写作教程海艺思创原创

准备

本文适合已有 OpenAI Python SDK、并准备把资料检索与内容校验接入写作流程的开发者。Responses API 的创建请求提供 parallel_tool_calls 参数，用于允许模型并行发起工具调用。应用侧仍要自行执行每一项函数，并把结果关联回原调用。

准备异步客户端与可用模型。
把每个工具定义为参数明确的函数，例如 search_sources 与 check_required_fields。
为工具结果准备统一 JSON 结构，至少包含 ok、data 和 error。
限制工具的输入范围，资料检索只接收查询词，字段校验只接收待检查内容。

若你还没有接过基础工具循环，可先看工具调用接入 AI 写作流程，再处理并行场景。

分步操作

声明工具并允许并行调用。在首次 responses.create 请求中传入函数工具，并设置 parallel_tool_calls=True。这表示可以允许模型在同一轮提出多项调用，不表示应用会自动执行这些函数。
收集本轮全部函数调用。遍历 first.output，筛选 type == "function_call" 的项目。不要拿到第一项就立刻续写，否则同轮其余检索或校验请求会被遗漏。
按调用分别解析和执行。每个调用都有自己的名称、参数和调用 ID。根据名称分发到受控的本地函数，未知名称直接返回失败结果，不要动态执行模型给出的任意名称。
并发等待工具完成。对彼此独立的调用使用 asyncio.gather。每项任务内部捕获异常，确保一个资料源超时不会让整轮写作中断。
回传全部工具结果。把每个结果包装为 function_call_output，并把其 call_id 设置为对应函数调用的 ID。随后将上一轮输出和本轮工具结果一起作为下一次请求的输入。
循环直到没有函数调用。模型收到资料与校验结果后，可能继续请求工具，也可能直接返回文章。只有当前响应不再包含函数调用时，才读取最终正文并进入发布检查。

可复制的异步模板

以下示例用两个独立工具模拟“查资料”和“检查必填字段”。将其中的模拟逻辑替换为你的数据库、检索服务或审核规则即可。

import asyncio
import json
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def search_sources(query):
    await asyncio.sleep(0)
    return {"ok": True, "data": [{"title": "资料摘要", "text": query}]}

async def check_required_fields(title, body):
    await asyncio.sleep(0)
    missing = []
    if not title:
        missing.append("title")
    if not body:
        missing.append("body")
    return {"ok": not missing, "data": {"missing": missing}}

TOOLS = [
    {
        "type": "function",
        "name": "search_sources",
        "description": "查询允许使用的资料库",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
            "additionalProperties": False
        }
    },
    {
        "type": "function",
        "name": "check_required_fields",
        "description": "检查文章标题和正文是否为空",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "body": {"type": "string"}
            },
            "required": ["title", "body"],
            "additionalProperties": False
        }
    }
]

async def run_call(call):
    try:
        args = json.loads(call.arguments)
        if call.name == "search_sources":
            result = await search_sources(**args)
        elif call.name == "check_required_fields":
            result = await check_required_fields(**args)
        else:
            result = {"ok": False, "error": "unsupported_tool"}
    except Exception as exc:
        result = {"ok": False, "error": str(exc)}

    return {
        "type": "function_call_output",
        "call_id": call.call_id,
        "output": json.dumps(result, ensure_ascii=False)
    }

async def write_article(task):
    response = await client.responses.create(
        model="gpt-4.1",
        input=task,
        tools=TOOLS,
        parallel_tool_calls=True
    )

    while True:
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:
            return response.output_text

        tool_outputs = await asyncio.gather(*(run_call(call) for call in calls))
        response = await client.responses.create(
            model="gpt-4.1",
            input=[*response.output, *tool_outputs],
            tools=TOOLS,
            parallel_tool_calls=True
        )

article = asyncio.run(write_article("根据资料写一篇产品说明，并检查标题和正文。"))
print(article)

工具结果示例

建议每个工具都返回同样的外层字段，让后续模型与日志系统容易判断是否可用。资料为空、校验不通过和服务异常都应明确返回，而不是伪造成功结果。

{
  "ok": false,
  "data": {"missing": ["body"]},
  "error": null
}

翻车怎么改

常见故障：模型明明请求了两个工具，最终却只使用了一份资料。
原因：应用只读取了第一个 function_call，或在循环中提前发送了下一次请求。
修正动作：先完整筛选当前响应中的全部函数调用，再用 asyncio.gather 执行；确认每一个 call_id 都有一条对应的 function_call_output 后，才发起后续请求。

常见故障：某个资料服务报错后，整篇文章不再生成。
原因：并发任务的异常直接向外传播。
修正动作：在单个工具执行函数中捕获异常，将失败信息作为该调用的结果回传；模型可据此改用已有资料或提示信息不足。