跳至内容

GPT-6 Astra API 教程:使用异步工具与转向构建发布检查代理

通过 OpenAI API 使用 GPT-6 Astra 构建一个 Python 发布检查代理,具备异步工具、推理控制、结构化输出与成本跟踪,并测试电脑操作与转向。
更新 2026年9月7日  · 14分钟

用 AI 探索

ChatGPTClaudePerplexity

我第一次在同一轮里给 GPT-6 Astra 同时提供一个慢工具和一个快工具时,本以为会走传统的同步循环:它会请求慢工具,然后在所有人等待时阻塞我的代码。OpenAI 的 异步工具调用文档 说 Astra 可以继续工作,但我并不确定。应用仍然要管理后台工作,因此异步工具调用并未移除编排。问题在于它是否改变得足够多,从而值得采用。

我们的 GPT-6 Astra 概览 涵盖了发布与基准测试,GPT-6 Astra vs. Claude Fable 5.1 指南则将其性能与价格与主要竞争对手进行了比较。本教程将让 GPT-6 Astra 运行起来,构建一个包含测试套件、健康检查端点,以及固定浏览器检查的发布检查工作流。另有独立演示涵盖模型生成的电脑操作与轮次中途转向。

我们将介绍如何:

  • 发起一次 GPT-6 Astra API 调用
  • 搭建一个作为基线的同步工具调用循环
  • 将慢检查切换为异步工具调用
  • 运行一次有界的电脑操作检查
  • 通过 WebSocket 将转向与“完成后重启”的基线进行比较
  • 仅在最终诊断时提高推理投入
  • 使用 结构化输出 返回经验证的 go/no-go 报告
  • 正确计算 API 成本,包括缓存写入
  • 在 Streamlit 中实时观看运行
  • 处理由 异步 任务引发的边缘情况

要点速览

GPT-6 Astra 在标准 Responses 循环上新增了三项 API 能力:异步工具调用、通过 WebSocket 的轮次中途转向,以及在对话中途调整推理投入。将三者合并进一个发布检查代理后得到的四个结论改变了我的实现思路。

  • 异步工具调用减少的是等待时间,而非模型工作量:回合数仍取决于模型的调用序列。
  • 转向耗时少于“完成后再重启”的基线,尽管测试未比较所有重启策略。
  • 更高的推理投入不一定改变诊断结论,即使会消耗更多推理令牌。
  • 并行运行检查可能暴露竞态条件,而顺序版本会将其掩盖。

这些结果适用于本次发布检查,而非所有代理工作负载。工具时长、共享状态以及模型采用的回合数都可能改变结果。

什么是 GPT-6 Astra API?

GPT-6 Astra API 是通过 Responses API 访问 OpenAI 新旗舰模型(于 2026 年 9 月 3 日发布)的方式。就本教程而言,重点在于使用方式:gpt-6-astra 通过 OpenAI 的 Responses API 接收文本和图像输入。其推理投入范围从 lowmax,不支持 none 选项。

本教程中的示例使用 client.responses.create 而不是 client.chat.completions.create,但迁移还需要调整请求、输出与工具结果的格式。还需移除自定义的 temperaturetop_p 和对数概率设置,因为 Astra 不支持它们。在首次调用前,先看看定价。

GPT-6 Astra 的价格是多少?

对于输入令牌不超过 272,000 的请求,标准定价 为每百万普通输入令牌 $10、每百万输出令牌 $50。缓存命中输入为每百万 $1,缓存写入为每百万 $12.50。

一旦请求超过该阈值,OpenAI 会对输入与缓存费率应用 2 倍系数,对输出费率应用 1.5 倍系数。更高费率适用于整个请求,而不仅仅是超出阈值的部分。本教程中的任何一次运行都未接近该阈值。

我们将用 GPT-6 Astra API 构建什么?

预发布应用是一个小型 Flask 任务看板:一个主页、一个添加任务的表单、一个将任务标记为完成的按钮,以及一个 /health 端点。完整代码(包含预发布应用)见此GitHub 仓库

代理检查的预发布任务看板,展示任务列表与添加任务表单

测试前会预置三个任务。图片来源:作者。

该应用有一个有意为之的缺陷。代理有三个检查,但只有测试套件被设计为能够捕捉到它。

为什么应用会接受空白任务标题?

创建任务的端点不会拒绝空白标题。我保留了这种行为,以便代理在不从提示中直接得知的情况下发现一个已知故障。

代理可以运行哪些发布检查?

代理可以调用三个工具:

  • run_test_suite 运行 pytest,包括约 250 次 HTTP 请求的批量导入测试。

  • check_ui_flow 使用 Playwright 添加一个任务并确认其显示。

  • check_staging_health/health 发送 GET 请求。

三者均对预发布环境运行,无任何 Mock。浏览器检查使用固定代码;模型生成的电脑操作演示稍后介绍。

如何在 Python 中设置 GPT-6 Astra API

您需要一个具有 gpt-6-astra 访问权限的 OpenAI API 密钥。在 platform.openai.com/api-keys 创建密钥,并检查您的项目是否启用了 gpt-6-astra。企业工作区在发布时默认关闭 Astra。

下面的命令使用 Windows PowerShell,并安装本教程所需的全部依赖,包括用于转向演示的 OpenAI realtime

python -m venv .venv
.venv\Scripts\Activate.ps1
Copy-Item .env.example .env
pip install "openai[realtime]" flask pytest playwright pydantic matplotlib python-dotenv streamlit
playwright install chromium

在 macOS 或 Linux 上,将激活命令替换为 source .venv/bin/activate,复制命令替换为 cp .env.example .env

然后将 API 密钥添加到新的 .env 文件:打开刚复制的 .env 文件,添加 OPENAI_API_KEY=sk-...python-dotenv 会加载该值,SDK 会自动读取,因而您无需在代码中传递密钥。

如果项目依赖或 .env 文件对您来说是新概念,我们的 虚拟环境环境变量 指南有详细说明。构建之前请先确认密钥是否可用。

如果您的 API 密钥已可用于 Responses API,可跳过下一小节,直接从同步工具循环开始。首次请求仅用于验证环境。

发出您的首个 GPT-6 Astra API 调用

将密钥就位后,需要通过 dotenv 加载。之后,您可以创建 OpenAI 客户端,并使用 client.responses.create() 发送首个请求。最小可行的请求如下:

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI()
response = client.responses.create(
    model="gpt-6-astra",
    reasoning={"effort": "low"},
    input="In one sentence, what is a staging environment used for?",
)
print(response.output_text)

响应的 usage 字段列出了稍后计算成本所需的令牌计数。

构建一个同步的 GPT-6 Astra 工具循环

下方代码片段为节选;可运行版本见配套的 GitHub 仓库

在碰异步之前,我先构建了普通版本:

  1. 调用模型

  2. 检查是否有 function_call

  3. 运行匹配的工具

  4. 使用 previous_response_id 发送结果

  5. 重复,直到模型不再请求工具。

该阻塞循环是进行异步对比的基线。

for turn in range(max_turns):
    response = client.responses.create(
        model="gpt-6-astra",
        reasoning={"effort": "low"},
        tools=TOOLS,
        input=next_input,
        previous_response_id=previous_response_id,
    )
    calls = [item for item in response.output if item.type == "function_call"]
    if not calls:
        final_text = response.output_text
        break
    outputs = [
        {"type": "function_call_output", "call_id": call.call_id, "output": run_tool(call)}
        for call in calls
    ]
    next_input, previous_response_id = outputs, response.id

基线运行发现了什么

在基线运行中,模型依次调用了每个工具。它发现了已知的校验失败,并返回了“不可发布”的决策。三次运行的总墙钟时间平均为 23.40 秒。该平均值覆盖了整个运行,而非单个工具时间。

GPT-6 Astra 异步工具调用如何工作?

在工具的 schema 中标记 "async": true,您的应用即可在模型继续工作或等待的同时延后返回该结果。您的应用仍需运行该工具并管理后台任务。

如何并发运行相互独立的检查

我将 run_test_suitecheck_ui_flow 标记为异步,并新增了一个应用自定义的 wait_for_tasks 工具且不含参数,因为本演示每次只会有一个待处理批次。这个无参数的等待让示例足够小。

在生产运行器中,请用任务句柄标识作业,将每个句柄绑定到其原始 call_id,并仅在下一步依赖某个未完成结果时再进行等待。

按各自的 call_id 返回每个已完成的结果,然后在等待工具自身的 call_id 上返回等待状态。

TOOLS = [
    {"type": "function", "name": "run_test_suite", "async": True, ...},
    {"type": "function", "name": "check_ui_flow", "async": True, ...},
    {"type": "function", "name": "check_staging_health", ...},
    {"type": "function", "name": "wait_for_tasks", ...},
]

在该次运行中,应用将每个已标记的调用提交到一个线程池。在后台检查运行时,它执行了快速的同步健康检查。然后在 wait_for_tasks 处阻塞,直到待处理检查完成。

异步工具调用能减少墙钟时间吗?

三次运行下来,异步将平均墙钟时间减少了 19.1%,从 23.40 秒降至 18.94 秒。平均值看起来直观,但单次运行差异很大;下图展示了它们的实际波动。三次运行足以说明此处发生的情况,但不足以预测生产时延。

折线图对比同一发布检查的三次同步运行与三次异步运行的墙钟秒数

两种模式下的运行时间均有波动。图片来源:作者。

为何并发检查会引发竞态条件?

同时运行浏览器检查与批量导入测试,有时会导致批量导入断言失败,因为两者都修改了同一个内存任务列表。这破坏了测试对独占访问的假设。在运行器或测试中隔离数据可避免该竞态。

用于 UI 测试的有界 GPT-6 Astra 电脑操作

对于电脑操作,GPT-6 Astra 的文档推荐使用代码执行,而结构化的 computer 工具仍作为替代方案受支持。使用代码执行时,一次调用可以组合多个动作、循环与条件逻辑;而 computer 工具则一次返回一个结构化的鼠标或键盘动作,由您的应用进行翻译与回放。

如何限制电脑操作运行器

我把该类命名为 BrowserSandbox,但这个名字夸大了其防护能力。它向模型生成的代码提供了一个 Playwright 的 page、一个 log() 函数,以及一个 expect_text() 辅助函数。Python 仍可向 exec() 注入内置对象,而页面也可导航到其他来源。

sandbox_globals = {"page": self.page, "log": log, "expect_text": expect_text}
exec(code, sandbox_globals)

请将其视为演示运行器,而非安全边界。模型生成的代码需要在隔离的进程或容器中运行,并限制文件系统、进程、网络与来源。

UI 检查中发生了什么?

我将循环上限设为 8 轮,因为有界检查需要硬性停止条件。模型检查了页面,找到了表单输入,创建了标题“UI flow check, cobalt otter 73921”,提交了任务,并确认标题已出现,全程耗时 13 秒。我们的 GPT-5.4 电脑操作教程 使用 Astra 的前代模型展示了更详细的示例。

GPT-6 Astra 的轮次中途转向如何工作?

轮次中途转向 仅可在 gpt-6-astra 且通过 WebSocket 连接时使用。

建立连接并创建响应,然后在响应生成过程中发送 response.steer 事件以附加新指令。 response.steer.accepted 事件意味着更新已排队,而非已应用。在创建自动续写之前,服务器会先完成当前输出项以及任何已在运行的托管工具工作。

如果仍需要客户端的工具结果或批准,response.steer.pending 会指出缺失的输入。

async with client.responses.connect() as connection:
    await connection.response.create(model="gpt-6-astra", input=TASK)
    async for event in connection:
        if event.type == "response.created" and initial_response_id is None:
            initial_response_id = event.response.id
            await asyncio.sleep(1.0)
            await connection.response.steer(
                previous_response_id=initial_response_id,
                input="Also add a rollback plan, but skip mobile.",
            )

这一秒的延迟与实验代码一致,给首个响应留出启动时间。若无该延迟,测试点将落在响应生命周期的不同位置。

中途转向有哪些不改变之处?

转向不会改写原始响应。若更新中断了它,该响应会以 status: "incomplete" 结束,且 incomplete_details.reason: "steered"。随后一个后续响应会在新指令下继续。

若首个响应在更新生效前完成,它将保持完成状态。转向不会逆转或取消已开始的客户端侧动作;您的应用仍需负责该行为。排队中的转向仅存在于当前 WebSocket 连接上,因此在尝试重连前请记录更新。

对比路径允许首个响应完成,然后用合并后的指令发送新请求。两次运行中,转向平均 26.91 秒,而“完成后再重启”路径为 50.02 秒。此对比未涵盖“取消并重启”策略,也未对最终文本正确性评分。

两张柱状图对比转向与重启的平均时长与平均成本

两次运行的时间与成本平均值。图片来源:作者。

重启路径会生成两次完整响应,覆盖部分相同请求。这一设置解释了部分时间差,因此我不会将这两次运行视为转向的一般性基准。

如何在对话中途更改 GPT-6 Astra 的推理投入

gpt-6-astra 支持从 lowmax 的推理投入。更高投入可能增加推理令牌使用,但并不保证得到不同答案。通过 configuration_update 可更改接下来的响应与后续响应,直到另一次更新覆盖它;而请求级设置保持不变。

在本流水线中,报告会结束对话,因此更新仅影响最后一步。

response = client.responses.create(
    model="gpt-6-astra",
    previous_response_id=previous_id,
    reasoning={"effort": "low"},  # default remains low
    input=[
        {"type": "configuration_update", "reasoning": {"effort": "high"}},
        {"role": "user", "content": "Diagnose the root cause and recommend a fix."},
    ],
)

我在两种投入等级下使用了相同的失败追踪,并比较了诊断与令牌使用。

有个细节:响应的 reasoning.effort 字段仍报告请求级设置,而非由 configuration_update 选定的投入等级。不要用该字段判断更新是否生效。

高推理投入是否改变了诊断?

在完整合并运行中,提升投入的步骤使用了 108 个推理令牌。其他在 low 等级的步骤都使用了 0 个。在更早的一个隔离测试中,失败追踪更长,低投入使用了 0 个推理令牌,而高投入使用了 318 个。两种版本都识别出了校验缺陷,因此更高投入改变了令牌用量,但未改变诊断结论。

配置更新仅适用于标准的单代理请求。API 会拒绝相邻的更新,且它们不能与自动压缩或自动截断同时使用。

如何使用 GPT-6 Astra 结构化输出

流水线的最后一步调用 client.responses.parse,将自由文本替换为经验证的 Pydantic 模型。

class GoNoGoReport(BaseModel):
    decision: str
    summary: str
    checks_completed: list[str]
    failures: list[str]
    risks: list[str]
    follow_up_actions: list[str]
    confidence: float

response = client.responses.parse(
    model="gpt-6-astra",
    previous_response_id=previous_id,
    text_format=GoNoGoReport,
    input=[...],
)

Pydantic 会检查此处声明的字段类型。它不能证明报告与证据一致,且该版本未将 decision 限定为两个取值,或将 confidence 限定在某个范围内。

go/no-go 报告捕捉到了什么?

报告返回了 decision: "no_go"。它将前述的校验失败归入 failures,将并发问题归入 risks。针对后者,它写道:“可能存在针对共享预发布数据的并发 UI 检查相互干扰。”提示中并未点名该风险。

将时延与成本保留在该 schema 之外,并在代码中计算。模型会根据工具证据填充报告字段。

Streamlit 界面与命令行运行器消费相同的生成器。它会在每个工具事件抵达时进行渲染,然后在ReportJSON 选项卡中展示解析后的报告。

实时展示代理进度与最终报告。视频来源:作者。

该仪表板运行合并后的异步发布流水线,包含固定浏览器检查、推理更新、结构化报告与时间显示。其成本面板读取与命令行运行器相同的台账。它不运行独立的模型生成电脑操作或转向演示。

如何跟踪 GPT-6 Astra API 的令牌用量与成本

在这些运行的模型令牌部分中,usage 字段包含为每次响应定价所需的四项令牌计数。缓存写入有单独费率,因此不要将其与普通输入合并。此处的工具运行在您的应用中;如果您添加了需单独计费的托管工具,也要将该费用计入。

details = usage.input_tokens_details
cached_tokens = details.cached_tokens
cache_write_tokens = details.cache_write_tokens
ordinary_tokens = usage.input_tokens - cached_tokens - cache_write_tokens

cost = (
    ordinary_tokens * PRICE_INPUT
    + cached_tokens * PRICE_CACHED_INPUT
    + cache_write_tokens * PRICE_CACHE_WRITE
    + usage.output_tokens * PRICE_OUTPUT
) / 1_000_000

该计算覆盖一次响应。台账会在每次响应后应用之,然后汇总调用总计。

即使 cache_write_tokens 值为 0 也应记录。否则,未来的缓存写入可能隐藏在普通输入计数里,这是发现计费错误的一种恼人方式。

GPT-6 Astra 代理的生产注意事项

此演示在用于部署门禁前需要进行如下更改。

工具权限与隔离

保持预发布边界,并按上述方式隔离 BrowserSandbox。不要为其提供数据库凭据或生产访问。

异步任务生命周期

在演示中,每个待处理任务都会完成,且无其他进程接触它。已部署的运行器必须能在两者都不成立时存活下来。对于每个待处理条目,它需要:

  • 截止时间与最终状态,避免任务永久处于待处理
  • 投递标记,避免回调与重试将同一结果发送两次
  • 开始与完成时间戳,以区分超时与迟到但有效的完成
  • 重复调用拒绝,防止同一任务被启动两次
  • 后台线程的错误处理,确保抛出的异常不会在池中悄然消失
  • 取消功能,即忽略迟到结果,并停止任何已在进行的外部工作

转向与不可逆操作

转向可以改变未来的指令,但无法逆转已完成的副作用。若工具已更改某个外部系统,则必须通过单独的工具动作进行补偿。

失准监控

自动停止适用于使用持久化推理、WebSocket 或 OpenAI 压缩的 Responses API 请求。其他请求可以触发告警,但不会被自动停止。

在流式传输之前,失准监控 可用 HTTP 403 及代码 misalignment_policy_violation 拦截受管控的运行。流式客户端则可能在输出已开始后收到错误。API 未为被停止的对话提供通用的恢复路径。

GPT-6 Astra 代理部署清单

在将该代理从本地演示迁移到部署之前,请加入以下控制。它们应位于应用代码中,而非模型指令中。

  • 为预发布应用的 HTTP 调用与 WebSocket 连接设置明确的超时

  • 记录普通输入、缓存输入、缓存写入、输出、响应 ID 与回合数

  • 对未完成的运行或触及回合上限的运行发出告警。为预算超支设置单独告警

  • 固定 openai SDK 版本,并在升级前重新验证异步、转向与 configuration_update 行为

何时应使用 GPT-6 Astra 的异步工具或转向?

选择能满足任务的最简单路径。

  • 当工具返回很快且需求保持固定时,从一次同步请求配合结构化输出开始。
  • 当模型或其他工具能在慢调用期间继续完成有用工作,且节省的时间足以抵消额外的任务管理开销时,添加异步工具调用。
  • 当需求在运行过程中发生变化时,使用轮次中途转向。

结语

在慢工具可重叠运行之后,这个同步的发布检查变得更有用,尽管结果并非彻底胜出。三次运行中,异步将平均时间从 23.40 秒降至 18.94 秒,同时也暴露了顺序循环曾经掩盖的共享状态竞态。

我会隔离浏览器与测试数据,将常规回合维持在 low,仅在需要更仔细审视证据时提高推理投入。若工具完成很快且需求固定,停留在带结构化输出的同步循环即可。当独立工作可以重叠时使用异步;当响应生成过程中指令发生变化时使用转向。

关于 API 基础知识,我推荐参加我们的 Working with the OpenAI API 课程。对于更大型的代理系统,请参阅我们的 Building Scalable Agentic Systems 课程。

常见问题

我可以用 Chat Completions 搭配 GPT-6 Astra 吗?

对于纯文本,可以。对于工具调用,不行:Astra 需要使用 Responses API,因此此处所有示例都使用 client.responses.create

哪些模型支持异步工具调用与转向?

异步工具调用是在 GPT-6 Astra 中引入的。轮次中途转向仅限 Astra 且仅支持 WebSocket;GPT-5.6 及更早版本完全不支持。

将现有请求切换到 gpt-6-astra 时会破坏什么?

三件事。reasoning.effort: "none" 会返回 HTTP 400,因此请从 low 开始。需要移除 temperaturetop_p 与对数概率设置。且若工具调用尚未迁移到 Responses,就必须迁过去。

异步工具调用会取代并行工具调用吗?

不会,它们解决的是不同问题。并行工具调用允许模型在一轮中请求多个工具;异步则允许您的应用在模型继续前进时延后某个工具的结果。

为何我的异步工具调用报错缺少 function_call_output?

当同一批中存在未解决的非异步工具调用时,可能会出现该错误。异步只会延后被标记的调用;其他所有工具调用仍需先提供输出。

异步工具调用是否需要 WebSocket?

不需要。上述异步实现使用的是常规 Responses API 调用。

空标题缺陷是否曾被 UI 检查而非测试套件捕捉到?

没有。只有测试套件覆盖了空标题校验;UI 与健康检查测试的是其他行为。

为什么我们在工具循环中使用 previous_response_id?

它将每个工具结果与请求它的响应关联起来。该循环可以在同一 Responses API 会话中继续,而无需在每次调用中重发完整对话记录。

主题
人工智能
大语言模型
AI 代理

在 DataCamp 学习 AI!

Tracks

面向开发者的 AI 工程师助理

26小时
了解如何使用 API 和开源库将 AI 集成到软件应用程序中。 今天就开始你的 AI 工程师之旅吧!
查看详情Right Arrow
开始课程
查看更多Right Arrow