BigQuery 的 AI token 配额正式可用：每项目每天 2000 亿输入、200 亿输出

这次正式开放的是什么

这套功能是一组每日配额，限制 BigQuery 中生成式 AI 函数消耗的输入与输出 token 数量，在 IAM & Admin 的「配额与系统限制」里管理。工作负载需要更多容量时可以上调，也可以用 override 把用量压得比默认更紧。两个方向都重要：默认值保护你不被一条失控的查询拖走账单，更严格的 override 保护一个共享项目不被某个团队的实验占满。

覆盖范围很具体。AI.GENERATE、AI.GENERATE_TABLE、AI.GENERATE_TEXT、AI.CLASSIFY、AI.IF、AI.AGG、AI.SCORE 受配额约束。向量生成与语义搜索类函数，例如 AI.EMBED、ML.GENERATE_EMBEDDING，不受约束，因为它们跑在不同的、便宜得多的模型上，计费结构也不一样。

-- Estimate before you spend. AI.COUNT_TOKENS prices a dry run instead of a surprise.
SELECT
  COUNT(*)                    AS rows_to_process,
  SUM(tok.total_tokens)       AS estimated_tokens,
  SUM(tok.prompt_tokens)      AS estimated_input,
  SUM(tok.candidates_tokens)  AS estimated_output
FROM (
  SELECT
    review_id,
    AI.COUNT_TOKENS(
      'gemini-2.5-flash',
      CONCAT('Classify this review as positive, negative or mixed:\n', review_text)
    ) AS tok
  FROM `project.dataset.reviews`
  WHERE processed_at IS NULL
  LIMIT 1000
)
-- Multiply the per-1000-row average by your real row count, then compare against
-- the quota table: 200,000,000,000 input and 20,000,000,000 output tokens per day per project.

默认值换算成你能推理的单位

输入 token：每项目每天 2000 亿，每项目每用户每天 1500 亿。输出 token：每项目每天 200 亿，每项目每用户每天 150 亿。输出里包含内部推理阶段产生的思考 token，这个细节会让朴素的估算彻底失效——推理型 prompt 的可计费输出，可能远多于你看到的答案本身。

还有两个细节值得记。缓存命中的 token 不计入配额。配额以百万 token 为单位累积统计，所以虽然你可以设很精确的值，低于几百万的量级可能反映得并不完全准确。配额对整个项目的所有区域全局生效（成本控制本来就该这样），每日在太平洋时间午夜重置。

# Budget arithmetic in Python, because the quota units are easy to misread.
QUOTAS = {
    "input_per_project_per_day":  200_000_000_000,
    "input_per_user_per_day":     150_000_000_000,
    "output_per_project_per_day":  20_000_000_000,
    "output_per_user_per_day":     15_000_000_000,
}

def plan(rows, avg_input, avg_output, users=10, batch_size=10_000):
    in_total = rows * avg_input
    out_total = rows * avg_output
    batches = -(-rows // batch_size)
    return {
        "rows": rows,
        "input_tokens": f"{in_total:,}",
        "output_tokens": f"{out_total:,}",
        "input_pct_of_project_quota": f"{in_total / QUOTAS['input_per_project_per_day']:.1%}",
        "output_pct_of_project_quota": f"{out_total / QUOTAS['output_per_project_per_day']:.1%}",
        "output_pct_of_user_quota": f"{out_total / users / QUOTAS['output_per_user_per_day']:.1%}",
        "batches": batches,
    }

# 5 million rows, 700 input tokens and 90 output tokens each
print(plan(5_000_000, 700, 90))
# Output is usually the binding constraint: 450M output tokens is 2.25% of the project cap
# but only 15M of headroom per user on a shared service account is a different story.

执行行为才是出意外的地方

BigQuery 在执行前先查配额。如果相关配额已经耗尽，查询会以 QuotaExceeded 被直接拒绝，什么都不跑。这是好情况。

难的是运行中途耗尽。当一条正在跑的查询消耗 token 到把某个配额打满时，这条查询里新的 LLM 调用会被拒绝，其余依赖这些调用的行会遇到配额耗尽错误。对 AI.IF 来说，如果你用了 max_error_ratio，只要错误比例还在允许范围内，仍可能返回部分结果；否则整条查询失败。对 AI.GENERATE，作业可以跑完，但没能处理的行会返回空的生成结果，并在 status 字段里带一条错误信息。

最后这种行为在生产里最危险，因为它「软失败」。一条下游查询去对分类列求和，会顺利地把成功的那些行聚起来，然后静默地少算其余部分。而同一项目里后续的生成式 AI 查询，在每日重置之前都会以 QuotaExceeded 失败——一个糟糕的下午很容易变成一个糟糕的整天。

# Mid-query exhaustion is not an all-or-nothing failure. Handle the partial result.
from google.cloud import bigquery

QUERY = '''
SELECT
  review_id,
  AI.GENERATE(
    ('Classify the sentiment', review_text),
    connection_id => 'us.conn',
    endpoint => 'gemini-2.5-flash',
    output_schema => 'sentiment STRING, confidence FLOAT64'
  ) AS result
FROM `project.dataset.reviews`
WHERE processed_at IS NULL
'''

def run_with_partial_handling(client, query):
    job = client.query(query)          # rejected up front with QuotaExceeded if exhausted
    rows = list(job.result())
    ok = [r for r in rows if r["result"] and not r["result"].get("error")]
    failed = [r for r in rows if not (r["result"] and not r["result"].get("error"))]
    return {"ok": len(ok), "failed": len(failed), "job_id": job.job_id}

# AI.IF can return partial results when max_error_ratio is used; AI.GENERATE completes
# the job but leaves an error message on the rows it could not process.

输出配额比输入配额小两个数量级。猜猜哪个先顶到。

先估再跑，而不是跑完再算

最便宜的习惯是在样本上做一次「干跑」。AI.COUNT_TOKENS 能给出指定模型与 prompt 下的 prompt token 与候选 token 数量，于是你可以先给一千行定价、再乘上去，然后拿去和配额表对照，最后才碰全表。同一个函数还能用来做 prompt 优化：在每行 700 token 的情况下，把 prompt 砍掉 120 token，输入成本就降了 17%。

实践里，输出比输入更常成为瓶颈，因为默认输出配额小了两个数量级。如果你的负载是分类，这没问题，输出很短；但如果涉及推理模型输出带解释的结构化 JSON，200 亿输出 token 消耗起来会比你预期的快。

# Quotas can be raised, lowered or overridden. Do it as code so the value is reviewable.
import subprocess, json

def current_quota(project, quota_id="GenAiInputTokensPerDay"):
    out = subprocess.run(
        ["gcloud", "beta", "quotas", "info", "list",
         "--service=bigquery.googleapis.com", "--project", project, "--format=json"],
        capture_output=True, text=True, check=True,
    ).stdout
    for q in json.loads(out):
        if quota_id and quota_id in q.get("name", ""):
            return q
    return None

def desired_limits(monthly_rows, avg_input, avg_output, working_days=21):
    per_day_in = monthly_rows / working_days * avg_input
    per_day_out = monthly_rows / working_days * avg_output
    return {
        "GenAiInputTokensPerDay": int(per_day_in * 1.5),
        "GenAiOutputTokensPerDay": int(per_day_out * 1.5),
    }

print(desired_limits(3_000_000, 700, 90))     # headroom of 50% over expected daily usage
# Apply through the Cloud Quotas API, gcloud, or Terraform so the override is in git.

把配额当配置管，而不是控制台里的一个开关

只存在于控制台里的配额值，是一个没人复核的数字。Cloud Quotas API、gcloud 和 Terraform 都能把这些值当代码来管，意味着每次改动都会落进一个带理由的 pull request。这点接线成本值得付，因为配额改动往往发生在最糟的时刻——通常是在一次故障当中。

文档里还有一条提醒：如果你用的是带预置吞吐（provisioned throughput）的 Gemini Enterprise Agent Platform 模型，计费根本不由 token 用量决定。这种情况下应把 BigQuery 的 token 配额设得很高，否则一套为按 token 计费设计的配额，会把已经付过钱的查询挡在门外。

# Fail loudly when a job is about to sit on the quota instead of quietly burning it.
from dataclasses import dataclass

@dataclass
class Guardrail:
    name: str
    limit: int
    used: int

    @property
    def ratio(self):
        return self.used / self.limit if self.limit else 0.0

def preflight(guardrails, warn=0.7, block=0.95):
    findings = []
    for g in guardrails:
        status = "block" if g.ratio >= block else "warn" if g.ratio >= warn else "ok"
        findings.append({"guardrail": g.name, "used_pct": round(g.ratio * 100, 1), "status": status})
    return findings

print(preflight([
    Guardrail("input_per_project", 200_000_000_000, 46_000_000_000),
    Guardrail("output_per_project", 20_000_000_000, 19_200_000_000),
    Guardrail("output_per_user", 15_000_000_000, 4_100_000_000),
]))
# [(input_per_project, 23.0, ok), (output_per_project, 96.0, block), (output_per_user, 27.3, ok)]
# Output is the one that runs out first in almost every pipeline I have seen.

配额按项目全局统计，每日太平洋时间午夜重置。

我真正会装上的几道护栏

第一，预检：当某个护栏超过 95% 时让作业直接失败，而不是让它去试着跑、然后产出部分结果。第二，按用户或按服务账号设上限，作为爆炸半径控制，这样一个 notebook 吃不掉整个共享项目当天的输出预算。第三，跑完之后加一条断言，数一下未处理的行数，只要不为零就拒绝把结果写进下游。

这些都不新奇。这就是你对待一个有限流的第三方 API 时会有的纪律，只是最常缺失在那些「感觉像 SQL」的工具里。SQL 给人的印象是要么成功要么报错；而 SQL 里的生成式 AI 函数，是这套环境里第一个会「半成功」的东西——所以这笔账必须写明白。