异步工具调用 | OpenAI API

异步工具调用让模型在调用工具后继续工作，无需等待该工具的结果。您可以用它提前发起耗时的查询请求，回答请求中不依赖这些结果的部分，并在应用获取结果后传回结果。

异步工具的工作原理

普通的函数调用会暂停模型的当前轮次，以等待工具响应。在函数或自定义工具的定义中设置 async: true，即可让模型在发出调用后、您的应用返回输出前继续工作。

工具仍由您的应用执行。异步工具不会将执行工作转移到 OpenAI，也不会管理您的后台作业。

这与后台模式不同，后台模式以异步方式生成响应。异步工具调用则让模型在您的应用运行工具时继续工作。

作业完成后，在后续的 Responses 请求中包含其输出。使用原始 API call_id 将结果与对应的调用匹配：

工具类型	调用项	输出项
函数	`function_call`	`function_call_output`
自定义	`custom_tool_call`	`custom_tool_call_output`

调用异步工具

在工具定义中添加 async: true。response.output 中的对应调用项会包含 async: true。

在后台查询天气

Python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79import OpenAI from "openai";

const client = new OpenAI();
const model = "gpt-6-astra";

const tools = [
  {
    type: "function",
    name: "get_weather",
    description: "Read a demo weather snapshot for a city.",
    async: true,
    strict: true,
    parameters: {
      type: "object",
      properties: { city: { type: "string" } },
      required: ["city"],
      additionalProperties: false,
    },
  },
];
async function getWeather(city) {
  const snapshots = {
    Paris: {
      city: "Paris",
      temperature_c: 22,
      condition: "Clear",
      source: "demo weather snapshot",
    },
  };
  if (typeof city !== "string" || !Object.hasOwn(snapshots, city)) {
    throw new Error(`No demo weather snapshot for ${city}.`);
  }
  return snapshots[city];
}
const instructions =
  "Start the weather lookup and answer the independent packing question " +
  "without waiting. Use the demo weather result when it arrives; never invent it.";

let response = await client.responses.create({
  model,
  tools,
  instructions,
  input:
    "Check the demo weather snapshot for Paris. Meanwhile, " +
    "list three essentials for any city trip.",
});

const call = response.output.find((item) => item.type === "function_call");
if (!call || call.name !== "get_weather") {
  throw new Error("The response did not include a weather call.");
}
const { city } = JSON.parse(call.arguments);
let latestResponseId = response.id;

// Calling an async function starts the application's job immediately.
const job = getWeather(city).catch((error) => ({ error: error.message }));
if (!call.async) {
  // Ordinary synchronous calls must finish before the model resumes.
  await job;
}
console.log(response.output);
// Independent work or conversation turns can happen here.
// Update latestResponseId after each continuation.
const result = await job;
response = await client.responses.create({
  model,
  tools,
  instructions,
  previous_response_id: latestResponseId,
  input: [
    {
      type: "function_call_output",
      call_id: call.call_id,
      output: JSON.stringify(result),
    },
  ],
});
latestResponseId = response.id;
console.log(response.output);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93import json
from concurrent.futures import ThreadPoolExecutor

from openai import OpenAI
from openai.types.responses import FunctionToolParam


def get_weather(city):
    # Demo data. Replace this function with your weather service.
    weather = {
        "Paris": {
            "city": "Paris",
            "temperature_c": 22,
            "condition": "Clear",
            "source": "demo weather snapshot",
        }
    }
    return weather[city]


worker = ThreadPoolExecutor()


def main():
    client = OpenAI()
    model = "gpt-6-astra"
    tools: list[FunctionToolParam] = [
        {
            "type": "function",
            "name": "get_weather",
            "description": "Read the demo weather snapshot for a city.",
            "async": True,
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
                "additionalProperties": False,
            },
        },
    ]

    instructions = (
        "Start the weather lookup and answer the independent packing "
        "question without waiting. Use the actual tool result when it "
        "arrives; never invent it. Identify the weather as demo data."
    )
    response = client.responses.create(
        model=model,
        tools=tools,
        instructions=instructions,
        input=(
            "Check the demo weather in Paris. Meanwhile, "
            "list three essentials for any city trip."
        ),
    )

    call = next(item for item in response.output if item.type == "function_call")
    arguments = json.loads(call.arguments)
    if call.name != "get_weather" or arguments != {"city": "Paris"}:
        raise ValueError("Expected a weather lookup for Paris")

    latest_response_id = response.id
    if call.async_:
        job = worker.submit(get_weather, **arguments)
        print(response.output_text)
        # Independent work or conversation turns can happen here.
        # Update latest_response_id after each continuation.
        result = job.result()
    else:
        result = get_weather(**arguments)

    response = client.responses.create(
        model=model,
        tools=tools,
        instructions=instructions,
        previous_response_id=latest_response_id,
        input=[
            {
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result),
            },
        ],
    )
    print(response.output_text)


if __name__ == "__main__":
    try:
        main()
    finally:
        worker.shutdown(wait=True)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/openai/openai-go/v3"
	"github.com/openai/openai-go/v3/responses"
)

type weatherArguments struct {
	City string `json:"city"`
}

type weatherSnapshot struct {
	City         string `json:"city"`
	TemperatureC int    `json:"temperature_c"`
	Condition    string `json:"condition"`
	Source       string `json:"source"`
}

func getWeather(city string) weatherSnapshot {
	// Demo data. Replace this function with your weather service.
	if city != "Paris" {
		panic("No demo weather snapshot for " + city)
	}
	return weatherSnapshot{
		City: city, TemperatureC: 22, Condition: "Clear", Source: "demo weather snapshot",
	}
}

func main() {
	client := openai.NewClient()
	ctx := context.Background()
	tool := responses.ToolParamOfFunction("get_weather", map[string]any{
		"type":                 "object",
		"properties":           map[string]any{"city": map[string]string{"type": "string"}},
		"required":             []string{"city"},
		"additionalProperties": false,
	}, true)
	tool.OfFunction.Description = openai.String("Read the demo weather snapshot for a city.")
	tool.OfFunction.Async = openai.Bool(true)
	tools := []responses.ToolUnionParam{tool}
	instructions := "Start the weather lookup and answer the independent packing question " +
		"without waiting. Use the actual tool result when it arrives; never invent it. " +
		"Identify the weather as demo data."
	response, err := client.Responses.New(ctx, responses.ResponseNewParams{
		Model:        "gpt-6-astra",
		Tools:        tools,
		Instructions: openai.String(instructions),
		Input:        responses.ResponseNewParamsInputUnion{OfString: openai.String("Check the demo weather in Paris. Meanwhile, list three essentials for any city trip.")},
	})
	if err != nil {
		panic(err)
	}
	var call responses.ResponseFunctionToolCall
	for _, item := range response.Output {
		if item.Type == "function_call" && item.AsFunctionCall().Name == "get_weather" {
			call = item.AsFunctionCall()
			break
		}
	}
	if call.CallID == "" {
		panic("The response did not include a weather call.")
	}
	var arguments weatherArguments
	if err := json.Unmarshal([]byte(call.Arguments), &arguments); err != nil {
		panic(err)
	}
	latestResponseID := response.ID
	var result weatherSnapshot
	if call.Async {
		job := make(chan weatherSnapshot, 1)
		go func() { job <- getWeather(arguments.City) }()
		fmt.Println(response.OutputText())
		// Independent work or conversation turns can happen here.
		// Update latestResponseID after each continuation.
		result = <-job
	} else {
		result = getWeather(arguments.City)
	}
	output, err := json.Marshal(result)
	if err != nil {
		panic(err)
	}
	functionOutput := responses.ResponseInputItemParamOfFunctionCallOutput(string(output))
	functionOutput.OfFunctionCallOutput.CallID = openai.String(call.CallID)
	response, err = client.Responses.New(ctx, responses.ResponseNewParams{
		Model:              "gpt-6-astra",
		Tools:              tools,
		Instructions:       openai.String(instructions),
		PreviousResponseID: openai.String(latestResponseID),
		Input:              responses.ResponseNewParamsInputUnion{OfInputItemList: responses.ResponseInputParam{functionOutput}},
	})
	if err != nil {
		panic(err)
	}
	fmt.Println(response.OutputText())
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.openai.core.JsonValue;
import com.openai.models.responses.FunctionTool;
import com.openai.models.responses.Response;
import com.openai.models.responses.ResponseCreateParams;
import com.openai.models.responses.ResponseFunctionToolCall;
import com.openai.models.responses.ResponseInputItem;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

record WeatherArguments(String city) {}

record WeatherSnapshot(
    String city,
    @JsonProperty("temperature_c") int temperatureC,
    String condition,
    String source) {}

static WeatherSnapshot getWeather(String city) {
  // Demo data. Replace this function with your weather service.
  if (!city.equals("Paris")) {
    throw new IllegalArgumentException("No demo weather snapshot for " + city);
  }
  return new WeatherSnapshot(city, 22, "Clear", "demo weather snapshot");
}

FunctionTool tool =
    FunctionTool.builder()
        .name("get_weather")
        .description("Read the demo weather snapshot for a city.")
        .async(true)
        .strict(true)
        .parameters(
            FunctionTool.Parameters.builder()
                .putAdditionalProperty("type", JsonValue.from("object"))
                .putAdditionalProperty(
                    "properties", JsonValue.from(Map.of("city", Map.of("type", "string"))))
                .putAdditionalProperty("required", JsonValue.from(List.of("city")))
                .putAdditionalProperty("additionalProperties", JsonValue.from(false))
                .build())
        .build();
String instructions =
    "Start the weather lookup and answer the independent packing question without waiting. Use the actual tool result when it arrives; never invent it. Identify the weather as demo data.";
Response response =
    client
        .responses()
        .create(
            ResponseCreateParams.builder()
                .model("gpt-6-astra")
                .addTool(tool)
                .instructions(instructions)
                .input(
                    "Check the demo weather in Paris. Meanwhile, list three essentials for any city trip.")
                .build());
ResponseFunctionToolCall call =
    response.output().stream()
        .flatMap(item -> item.functionCall().stream())
        .filter(item -> item.name().equals("get_weather"))
        .findFirst()
        .orElseThrow(
            () -> new IllegalStateException("The response did not include a weather call."));
WeatherArguments arguments = call.arguments(WeatherArguments.class);
String latestResponseId = response.id();
WeatherSnapshot result;
if (call.async().orElse(false)) {
  CompletableFuture<WeatherSnapshot> job =
      CompletableFuture.supplyAsync(() -> getWeather(arguments.city()));
  System.out.println(response.output());
  // Independent work or conversation turns can happen here.
  // Update latestResponseId after each continuation.
  result = job.join();
} else {
  result = getWeather(arguments.city());
}
response =
    client
        .responses()
        .create(
            ResponseCreateParams.builder()
                .model("gpt-6-astra")
                .addTool(tool)
                .instructions(instructions)
                .previousResponseId(latestResponseId)
                .inputOfResponse(
                    List.of(
                        ResponseInputItem.ofFunctionCallOutput(
                            ResponseInputItem.FunctionCallOutput.builder()
                                .callId(call.callId())
                                .output(new ObjectMapper().writeValueAsString(result))
                                .build())))
                .build());
response.output().stream()
    .flatMap(item -> item.message().stream())
    .flatMap(message -> message.content().stream())
    .flatMap(content -> content.outputText().stream())
    .forEach(text -> System.out.println(text.text()));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70require "json"
require "openai"

def get_weather(city)
  # Demo data. Replace this function with your weather service.
  raise "No demo weather snapshot for #{city}" unless city == "Paris"

  {
    city: city,
    temperature_c: 22,
    condition: "Clear",
    source: "demo weather snapshot"
  }
end

client = OpenAI::Client.new
tools = [
  OpenAI::Models::Responses::FunctionTool.new(
    name: "get_weather",
    description: "Read the demo weather snapshot for a city.",
    async: true,
    strict: true,
    parameters: {
      type: "object",
      properties: { city: { type: "string" } },
      required: ["city"],
      additionalProperties: false
    }
  )
]
instructions = "Start the weather lookup and answer the independent packing question " \
  "without waiting. Use the actual tool result when it arrives; never invent it. " \
  "Identify the weather as demo data."
response = client.responses.create(
  model: "gpt-6-astra",
  tools: tools,
  instructions: instructions,
  input: "Check the demo weather in Paris. Meanwhile, list three essentials for any city trip."
)
call = response.output.find do |item|
  item.is_a?(OpenAI::Models::Responses::ResponseFunctionToolCall) && item.name == "get_weather"
end
unless call.is_a?(OpenAI::Models::Responses::ResponseFunctionToolCall)
  raise "The response did not include a weather call."
end

city = JSON.parse(call.arguments).fetch("city")
latest_response_id = response.id
result = if call.async
           job = Thread.new { get_weather(city) }
           puts(response.output_text)
           # Independent work or conversation turns can happen here.
           # Update latest_response_id after each continuation.
           job.value
         else
           get_weather(city)
         end
response = client.responses.create(
  model: "gpt-6-astra",
  tools: tools,
  instructions: instructions,
  previous_response_id: latest_response_id,
  input: [
    OpenAI::Models::Responses::ResponseInputItem::FunctionCallOutput.new(
      call_id: call.call_id,
      output: JSON.generate(result)
    )
  ]
)
puts(response.output_text)

响应可以同时包含异步调用和回答。如果在作业完成前又进行了其他对话轮次，请更新 latest_response_id，以从最新响应继续，同时保留原始工具调用的 call_id。

使用流式传输时，若要更早启动作业，可在收到完整的调用项后立即启动该作业，同时继续接收并处理响应。

添加等待工具

等待工具让模型自行决定何时需要尚未返回的结果。例如，模型可以发起两个价格查询请求，先处理不依赖查询结果的工作，直到准备比较价格时才等待结果。

为每个异步工具添加 task_handle 参数。模型会为每次调用分配一个句柄，您的应用则将该句柄绑定到原始 API call_id 和正在运行的作业。请确保句柄在整个对话中保持唯一，包括已完成的任务和重复的查询请求。

将等待工具定义为普通的同步函数：省略 async，或将其设为 false。它的模式和行为由您的应用定义。wait_for_tasks 不是 Responses 的内置工具。

在请求的 tools 数组中使用这些定义：

1234567891011121314151617181920212223242526272829303132333435[
  {
    "type": "function",
    "name": "lookup_price",
    "async": true,
    "description": "Look up a product price in the background. Choose a fresh task_handle unique within this conversation, including completed tasks.",
    "strict": true,
    "parameters": {
      "type": "object",
      "properties": {
        "sku": { "type": "string" },
        "task_handle": { "type": "string" }
      },
      "required": ["sku", "task_handle"],
      "additionalProperties": false
    }
  },
  {
    "type": "function",
    "name": "wait_for_tasks",
    "description": "Wait for selected tasks whose results you need. Pass a nonempty list of distinct task_handles from your earlier lookup_price calls. Results arrive on their original calls; this tool returns status only. Do not wait again for results that have already arrived.",
    "strict": true,
    "parameters": {
      "type": "object",
      "properties": {
        "task_handles": {
          "type": "array",
          "items": { "type": "string" }
        }
      },
      "required": ["task_handles"],
      "additionalProperties": false
    }
  }
]

注册每个作业

在处理依赖某些作业的等待调用前，请先注册并启动这些作业。调用可能在同一个响应中到达，也可能分散在多个响应中。以下输出项示例展示了两次启动调用，以及一个依赖这两次调用的等待调用：

12345678910111213141516171819202122[
  {
    "type": "function_call",
    "name": "lookup_price",
    "async": true,
    "call_id": "call_widget",
    "arguments": "{\"sku\":\"WIDGET\",\"task_handle\":\"widget_price_1\"}"
  },
  {
    "type": "function_call",
    "name": "lookup_price",
    "async": true,
    "call_id": "call_gadget",
    "arguments": "{\"sku\":\"GADGET\",\"task_handle\":\"gadget_price_1\"}"
  },
  {
    "type": "function_call",
    "name": "wait_for_tasks",
    "call_id": "call_wait",
    "arguments": "{\"task_handles\":[\"widget_price_1\",\"gadget_price_1\"]}"
  }
]

应用的注册表将每个句柄绑定到其原始调用和正在运行的作业：

任务句柄	原始调用 ID	作业
`widget_price_1`	`call_widget`	WIDGET 价格查询
`gadget_price_1`	`call_gadget`	GADGET 价格查询

请在整个对话期间保留注册表，防止重复使用已完成任务的句柄。

先传回结果，再传回等待状态

在注册表中查找请求指定的句柄，并且仅等待这些句柄对应的作业。先使用各自的原始 call_id 返回每个新完成的结果，再使用等待调用自身的 call_id 返回状态。按此顺序操作，可让模型在恢复运行时获得结果。

例如，在下一个请求的 input 数组中发送以下输出项。其中的价格仅作示例：

1234567891011121314151617[
  {
    "type": "function_call_output",
    "call_id": "call_widget",
    "output": "{\"task_handle\":\"widget_price_1\",\"price_cents\":1200,\"currency\":\"USD\"}"
  },
  {
    "type": "function_call_output",
    "call_id": "call_gadget",
    "output": "{\"task_handle\":\"gadget_price_1\",\"price_cents\":1500,\"currency\":\"USD\"}"
  },
  {
    "type": "function_call_output",
    "call_id": "call_wait",
    "output": "{\"status\":\"completed\",\"completed_task_handles\":[\"widget_price_1\",\"gadget_price_1\"]}"
  }
]

将 previous_response_id 设为最新响应的 ID，并在后续请求中包含工具和指令。您的应用也可以在结果就绪时直接传回结果，无需等待调用。只有当模型的下一步依赖尚未返回的结果时，才使用等待工具。

兼容性

GPT-6 Astra 及后续模型支持异步工具调用。

异步执行适用于由您的应用运行的函数工具和自定义工具，不适用于托管的内置工具。请直接调用工具，不要为程序化工具调用配置异步工具。

在多智能体模式下，请勿将异步工具与并行工具调用结合使用。

建议

开始使用

核心概念

SDK 与 CLI

资源

旧版 API

选择模型

文本与代码

提示词

推理

图像与视频

实时交互与音频

专用模型

智能体 API

智能体 SDK

ChatKit

搜索与检索

连接工具与数据

构建工具工作流

计算机与代码

媒体

GPT-Live

Realtime API

构建语音应用

连接

音频处理

上线

性能与质量

成本与吞吐量

安全与治理

基础设施与访问

运营

核心概念

Plan

Build

Test and publish

Conversion specs

指南

资源

开始使用

指南

File Upload

API

Measurement

Advertiser API

API 参考

开始使用

基础

探索

可用平台

发布

工作流

能力

参考资料

自定义

配置文件

智能体配置

扩展 ChatGPT 和 Codex 的功能

Linux

Windows

开发工作流

扩展并实现自动化

环境

使用 Codex 构建

第三方集成

参考资料

权限

Codex Security

网络安全

入门

ChatGPT Work

身份与身份验证

工作空间访问权限、策略和模型

插件和连接器控制

使用情况、治理与合规

部署和模型提供商

社区

博客

社区

博客