实战教程：使用 Antigravity SDK + Google Cloud 构建自主财务审计 AI 智能体团队

这是一份分步指南，介绍如何使用 Google Antigravity SDK 和 Google Cloud Platform 构建、测试并部署多 AI 智能体财务对账工作流。

由 Gemini Notebook 生成

在本教程中，你将构建一个自主的多 AI 智能体财务审计系统，用于将供应商交易记录与 PDF 发票进行核对。我们不会依赖单个大语言模型提示词，而是设计一支由专业 AI 智能体组成的团队。每个 AI 智能体都有明确的职责范围、严格的安全边界和不同的能力。

这支 AI 智能体团队包括：

审计编排 AI 智能体（Audit Orchestrator）——管理者。它负责掌握高层目标、向子 AI 智能体分派任务，并管理包含 4 个阶段的工作流状态机：触发、收集、核对和报告。

数据研究 AI 智能体（Data Researcher Agent）——只读型专业 AI 智能体，负责查询 BigQuery 中待处理的供应商交易记录。

发票分析 AI 智能体（Invoice Analyzer Agent）——只读型专业 AI 智能体，负责查找存储在 Google Cloud Storage（GCS）中的 PDF 发票，并从中提取结构化的明细项数据。

对账引擎 AI 智能体（Reconciliation Engine Agent）——分析核心。它接收来自数据研究 AI 智能体和发票分析 AI 智能体的数据，将交易与发票进行匹配、验证税费计算，并标记不一致项。

人工合规关卡（Human Compliance Gate）——它不是 AI，而是一个声明式策略钩子。对于金额超过 1,000 美元的任何差异，它会暂停执行并升级给人工合规专员，在写入结果前进行手动审核。

为什么不直接让一个 AI 智能体访问所有资源？因为让单个 AI 智能体同时拥有 BigQuery 写入权限、GCS 访问权限和命令执行能力，会产生不可接受的故障影响范围。通过将工作流组织为多 AI 智能体团队，我们可以落实最小权限原则、隔离复杂推理过程，并为每一项决策建立可验证的监管链。

本教程采用完全云原生的方案，并使用真实的 Google Cloud 服务：BigQuery、Cloud Storage、Cloud Logging 和 Cloud Trace。你将配置真实的基础设施，并根据实时数据验证结果。

为什么选择 Google Antigravity SDK？

如果直接基于原始 LLM API 调用构建生产级多 AI 智能体系统，那么在编写第一行业务逻辑之前，你就必须自行解决一长串基础设施问题：AI 智能体生命周期管理、工具注册与 Schema 生成、AI 智能体间通信、策略执行、可观测性钩子以及安全并发。

Google Antigravity SDK（google-antigravity）是一个专门为解决这些问题而构建的 Python 框架。它提供了一系列基础原语，让你能够专注于 AI 智能体应该做什么，而不是如何将它们连接起来：

该 SDK 的设计理念是关注点分离。AI 智能体的推理能力（系统指令 + 工具）与其治理机制（策略 + 钩子）相互独立。这意味着，你可以将宽松的开发环境策略替换为严格受限的生产环境策略，而不需要修改任何一行 AI 智能体代码。在本教程中，你会多次看到这种模式。

在本教程中，你将使用该 SDK 构建一支由四个 AI 智能体组成的团队，共同完成财务对账审计。每个 AI 智能体都有自己独立且限定范围的配置、工具和策略。编排 AI 智能体会创建专业子 AI 智能体，策略引擎负责执行最小权限控制，钩子系统则在 Cloud Logging 和 Cloud Trace 中生成完整的合规审计轨迹，而所有底层基础设施连接工作都由 SDK 负责处理。

你需要准备：

• 一个已启用结算功能的 Google Cloud 项目，并启用 Vertex AI API（或者准备一个 Gemini API 密钥）
• 已安装并完成身份验证的 gcloud CLI
• Python 3.11+ 和 pip。我们将使用 google-cloud-bigquery Python 客户端库
• 已安装 Antigravity SDK（pip install google-antigravity）
• 基本熟悉 Google Cloud BigQuery SQL 和 Python

获取代码并准备示例数据

本教程的完整源代码可在 GitHub 上获取。首先克隆代码仓库、安装依赖项，并生成供 AI 智能体处理的示例发票 PDF：

# 1. Clone the repository
git clone https://github.com/rominirani/financial-audit-agent-tutorial.git
cd financial-audit-agent-tutorial

# 2. Create a virtual environment and install dependencies
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# 3. Generate sample invoice PDFs
python scripts/generate_sample_invoices.py

✅ Generated 4 sample invoice PDFs in /path/to/data/invoices

⚠️ 重要提示：运行任何代码之前，请设置你的 Project ID。代码库会从环境变量中读取 PROJECT_ID。请在当前终端会话中设置一次：bash export PROJECT_ID="your-gcp-project-id"。该值将用于 BigQuery 查询、Cloud Storage 存储桶命名（$PROJECT_ID-audit-invoices）以及审计结果写入。运行 main.py 时，你还需要传入 --project-id=$PROJECT_ID。

该脚本会生成 4 份发票 PDF，其中刻意设置了供应商发票内容与 ERP 系统（BigQuery）记录之间的差异。AI 智能体需要找出这些差异：

代码仓库结构如下：

├── main.py # Entry point 
├── requirements.txt # Python dependencies
├── agents/ # Agent configurations 
│ ├── orchestrator.py
│ ├── data_researcher.py
│ ├── invoice_analyzer.py
│ └── reconciler.py
├── tools/ # Custom tools — BigQuery + GCS 
│ └── bigquery_tools.py
├── hooks/ # Observability hooks 
│ └── observability.py
├── policies/ # Safety policies 
│ └── audit_policies.py
├── scripts/ # Data generation
│ └── generate_sample_invoices.py
└── eval/ # Evaluation 
    ├── eval_dataset.jsonl
    └── run_eval.py

我们假设你已经克隆了该代码仓库，并将本教程作为代码导读使用。在运行 AI 智能体之前，你还需要按照后续章节的说明配置 Google Cloud 基础设施。

我们将使用的 GCP 服务

该企业级架构依赖多个 Google Cloud 服务，以提供数据、存储和可观测性能力。

在开始编写代码之前，理解整体架构至关重要。我们将关注点划分为 AI 智能体运行时、数据层和人工治理三个部分。

下面的系统架构图展示了不同的 AI 智能体，以及它们将与哪些 Google Cloud 服务交互。

实际的工作流是什么样的？请看下图。

那么数据流的安全性如何保障？下图展示了两层安全机制。SDK 策略层在工具层面控制每个 AI 智能体能够执行的操作，GCP IAM 层则在基础设施层面控制服务账号能够访问的资源。只有这两层都允许某项操作，该操作才能成功执行。

通过下表，我们来了解每个 AI 智能体能够对各项 GCP 服务执行哪些操作：

请注意这里采用了纵深防御机制。即使数据研究 AI 智能体的 LLM 遭到提示词注入，试图执行 DELETE FROM vendor_transactions，也会在两个层面受到阻止：SDK 策略引擎会拒绝非 SELECT 查询，而 BigQuery IAM 角色仅授予 financial_audit 数据集的 dataEditor 权限，并未授予 bigquery.admin 权限。编排 AI 智能体本身无法直接访问数据，只能将任务委派给权限范围受限的子 AI 智能体。

数据流受到严格控制。编排 AI 智能体负责管理状态和升级流程，但不会接触原始数据。数据研究 AI 智能体和发票分析 AI 智能体均为只读。对账 AI 智能体只能向特定的输出位置写入数据。

这展示了最小权限原则在 AI 中的应用方式。每个 AI 智能体的策略都只包含完成其特定工作所需的最小权限集合。数据研究 AI 智能体不需要读取 PDF，因此不会获得 GCS 访问权限。发票分析 AI 智能体不需要访问数据库，因此会被禁止访问 BigQuery。这种隔离机制可以确保，即使某个 AI 智能体产生幻觉，或者通过恶意发票 PDF 遭到提示词注入，也无法危及整个系统。

分步配置 GCP

首先创建一个专用的 GCP 项目，并启用 AI 智能体团队将使用的六项云服务。每个 API 都承担特定职责：BigQuery 用于数据处理，Cloud Storage 用于存储 PDF，Cloud Logging 和 Cloud Trace 用于可观测性，Cloud Run 则用于后续部署。

export PROJECT_ID="financial-audit-tutorial"
gcloud projects create $PROJECT_ID --name="Financial Audit Tutorial"
gcloud config set project $PROJECT_ID
gcloud services enable bigquery.googleapis.com storage.googleapis.com logging.googleapis.com cloudtrace.googleapis.com secretmanager.googleapis.com run.googleapis.com

我们只创建两个表：vendor_transactions（ERP 的事实来源）和 audit_results（AI 智能体写入审计结果的位置）。发票数据不会预先加载到 BigQuery 中，AI 智能体会从存储在 GCS 中的 PDF 文件里提取这些数据。

bq mk --dataset $PROJECT_ID:financial_audit

# Create vendor_transactions — the ERP source of truth
bq mk --table $PROJECT_ID:financial_audit.vendor_transactions \
vendor_id:STRING,vendor_name:STRING,invoice_num:STRING,amount:FLOAT64,currency:STRING,tax_rate:FLOAT64,status:STRING,quarter:STRING,transaction_date:DATE

# Create audit_results — the agent writes its findings here
bq mk --table $PROJECT_ID:financial_audit.audit_results \
execution_id:STRING,vendor_id:STRING,invoice_num:STRING,transaction_amount:FLOAT64,invoice_amount:FLOAT64,discrepancy_amount:FLOAT64,status:STRING,agent_notes:STRING,reviewed_by:STRING,timestamp:TIMESTAMP

注意：在下面的 SQL 脚本中，我们有意在数据里设置了特定差异，用来测试 AI 智能体团队的分析能力：

供应商 8492：交易记录中的税率为 8.5%，但发票是按 6.25% 计算的。

供应商 3301：交易以 USD 记录，但发票使用的是 EUR（数值金额完全相同，用于测试货币识别能力）。

供应商 5567：存在两条金额不同（$23,400 与 $24,100）但发票编号相同的重复发票，用于测试去重逻辑。

使用 ERP 数据填充交易表。这些数据代表公司会计系统所记录的内容，包括金额、税率、货币和发票编号。可以将其视为内部财务系统的“事实来源”：

-- Run this in the BigQuery Console or via bq query
INSERT INTO `YOUR_PROJECT_ID.financial_audit.vendor_transactions` 
(vendor_id, vendor_name, invoice_num, amount, currency, tax_rate, status, quarter, transaction_date) VALUES
('8492', 'TechCorp Solutions', 'INV-8492-Q3-001', 142300.00, 'USD', 0.085, 'PENDING', 'Q3', '2026-07-15'),
('1022', 'OfficeSupplies Co', 'INV-1022-Q3-014', 4500.00, 'USD', 0.05, 'PENDING', 'Q3', '2026-07-20'),
('3301', 'Global Services Ltd', 'INV-3301-Q3-099', 87500.00, 'USD', 0.10, 'PENDING', 'Q3', '2026-08-01'),
('5567', 'Consulting Group Inc', 'INV-5567-Q3-001', 23400.00, 'USD', 0.0, 'PENDING', 'Q3', '2026-08-10'),
('5567', 'Consulting Group Inc', 'INV-5567-Q3-001', 24100.00, 'USD', 0.0, 'PENDING', 'Q3', '2026-08-12');

请注意，我们没有向 BigQuery 插入任何发票数据。对应的发票 PDF 将上传到 GCS。AI 智能体必须读取并解析这些 PDF，以获取供应商一方的信息，然后与交易记录进行比较。这两个数据源之间的差异，正是 AI 智能体需要自主发现的内容。

如果你已经按照前面的说明操作，那么 data/invoices/ 中应该已有 4 份示例发票 PDF。现在创建一个 Cloud Storage 存储桶并上传它们。-l us-central1 标志会将存储桶与 BigQuery 数据集部署在同一区域，从而实现低延迟访问：

# Create a regional bucket for invoice PDFs
gsutil mb -l us-central1 gs://$PROJECT_ID-audit-invoices

# Upload all generated invoices into a Q3/ prefix
gsutil -m cp data/invoices/*.pdf gs://$PROJECT_ID-audit-invoices/Q3/

# Verify uploads — you should see 4 PDF files
gsutil ls gs://$PROJECT_ID-audit-invoices/Q3/

Q3/ 前缀用于按季度组织发票。AI 智能体的 list_invoices_in_gcs() 工具在搜索需要核对的发票时，会按此前缀进行筛选。

为审计 AI 智能体创建一个专用服务账号，并只授予其所需的最低限度 IAM 角色。每个角色都对应一项特定能力：dataViewer 用于读取 BigQuery 表，jobUser 用于运行查询，logWriter 用于写入结构化审计日志，cloudtrace.agent 用于发布跟踪 span，storage.objectViewer 用于从 GCS 读取发票 PDF。该 AI 智能体无法删除数据、修改表或访问其他项目。

# Create the service account
gcloud iam service-accounts create audit-agent-sa \
    --display-name="Audit Agent Service Account"
SA_EMAIL="audit-agent-sa@$PROJECT_ID.iam.gserviceaccount.com"
# BigQuery: read transactions + write audit results
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/bigquery.dataViewer" --condition=None
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/bigquery.jobUser" --condition=None
# Cloud Logging: write structured audit events
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/logging.logWriter" --condition=None
# Cloud Trace: publish distributed trace spans
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/cloudtrace.agent" --condition=None
# Cloud Storage: read-only access to invoice PDFs
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/storage.objectViewer" --condition=None

当项目中已经存在带条件的 IAM 策略时，必须使用 --condition=None 标志。否则，命令将失败并返回 Adding a binding without specifying a condition is prohibited 错误。

在编写任何代码之前，请确认所有基础设施均已正确配置。下面两条命令可确认 BigQuery 数据集是否存在，以及服务账号是否拥有预期的 IAM 角色：

# Verify BigQuery dataset and tables exist (audit_results & vendor_transactions)
bq ls $PROJECT_ID:financial_audit
# Verify IAM roles assigned to the service account
gcloud projects get-iam-policy $PROJECT_ID \
  --flatten="bindings[].members" \
  --format='table(bindings.role)' \
  --filter="bindings.members:audit-agent-sa"

你应该会在输出中看到 roles/bigquery.dataViewer、roles/bigquery.jobUser、roles/logging.logWriter、roles/cloudtrace.agent 和 roles/storage.objectViewer。

Antigravity SDK 代码——分步讲解

首先，我们来设置项目文件夹结构并准备必要的环境。

克隆代码仓库，设置 Python 虚拟环境并安装依赖项。google-antigravity 是 SDK 本身；google-cloud-bigquery 和 google-cloud-storage 提供工具将要封装的 GCP 客户端库；reportlab 用于生成示例发票 PDF；PyPDF2 则支持在运行时提取 PDF 文本：

git clone https://github.com/rominirani/financial-audit-agent-tutorial.git
cd financial-audit-agent-tutorial
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

AI 智能体需要借助工具与外部系统交互。在本教程中，我们会定义自定义 Python 函数来封装 Google Cloud 客户端库（google-cloud-bigquery 和 google-cloud-storage），并通过 LocalAgentConfig 中的 tools 参数将这些函数传递给 AI 智能体。

每个工具都是一个带有类型提示和描述性文档字符串的标准 Python 函数。SDK 会自动将其注册为 AI 智能体可以调用的工具。AI 智能体能够看到包含描述的可用函数列表，并根据需要调用这些函数。

（tools/bigquery_tools.py）

该文件包含全部四个工具函数。每个函数都会封装一次 Google Cloud 客户端库调用，并返回 AI 智能体可以解析和推理的 JSON 字符串。

查看包含全部四个工具的完整实现：[bigquery_tools.py]

from google.cloud import bigquery, storage
import json
import os

PROJECT_ID = os.environ.get("PROJECT_ID", "YOUR_PROJECT_ID")
DATASET = "financial_audit"
BUCKET_NAME = f"{PROJECT_ID}-audit-invoices"

_client = bigquery.Client(project=PROJECT_ID)

def query_vendor_transactions(quarter: str = "Q3") -> str:
    """Query BigQuery for all PENDING vendor transactions in a given quarter."""
    query = f"""
    SELECT vendor_id, vendor_name, invoice_num, amount, currency, tax_rate, status, quarter, transaction_date
    FROM `{PROJECT_ID}.{DATASET}.vendor_transactions`
    WHERE status = 'PENDING' AND quarter = '{quarter}'
    ORDER BY vendor_id, invoice_num
    """
    rows = _client.query(query).result()
    results = [{"vendor_id": row.vendor_id, "amount": row.amount, ...} for row in rows]
    return json.dumps({"total_transactions": len(results), "transactions": results}, indent=2)

def list_invoices_in_gcs(bucket_name: str, prefix: str = "Q3/") -> str:
    """List all invoice PDF files in a Google Cloud Storage bucket."""
    ...

def read_invoice_from_gcs(bucket_name: str, blob_name: str) -> str:
    """Read and extract structured data from an invoice PDF stored in GCS."""
    ...

第 3/3 部分翻译

def write_audit_result(vendor_id: str, invoice_num: str, status: str, ...) -> str:
    """Write an audit reconciliation result to the BigQuery audit_results table."""
    ...

AUDIT_TOOLS = [query_vendor_transactions, list_invoices_in_gcs, read_invoice_from_gcs, write_audit_result]

这四个函数为智能体提供了类型化、文档齐全的工具，具有明确的输入/输出契约。read_invoice_from_gcs 工具会下载 PDF，使用 PyPDF2 提取文本，并通过正则表达式解析结构化字段（金额、税率、货币）。智能体永远不会看到原始 PDF 字节，它接收的是干净的 JSON。

安装 PyPDF2 以进行 PDF 文本提取：pip install PyPDF2。每个工具函数的文档字符串会变成 LLM 可见的工具描述。编写清晰、具体的文档字符串，它们直接影响智能体选择正确工具的能力。

(policies/audit_policies.py)

安全策略是 SDK 用于控制智能体能做什么和不能做什么的机制。每个策略层级代表不同程度的信任。关键差异是：预发布环境需要人工在终端批准写入操作，而生产环境则完全自主运行，进行参数级验证。compliance_officer_approval_handler 模拟人工审核门槛，在真实部署中，你会将 input() 调用替换为 Slack 通知或审批工作流：

from google.antigravity.hooks import policy

VALID_STATUSES = {"MATCHED", "DISCREPANCY", "ESCALATED", "UNMATCHED"}

async def compliance_officer_approval_handler(tool_call) -> bool:
    """Escalation handler for high-risk actions in staging.

    Receives a types.ToolCall object. Use tool_call.name for the tool name
    and tool_call.args for the arguments dict.
    """
    print(f"\n🚨 ESCALATION REQUIRED 🚨")
    print(f"Action requested: {tool_call.name}")
    print(f"Arguments: {tool_call.args}")

    # In a real environment, this would ping Slack/Email and wait.
    # For this tutorial, we simulate a prompt.
    response = input("Compliance Officer, approve this action? (y/n): ")
    return response.lower() == 'y'

# --- Development: full permissions, no restrictions ---
DEVELOPMENT_POLICIES = [
    policy.allow_all(),
]

# --- Staging: reads auto-allowed, writes require human approval ---
# Use this when testing interactively. You sit at the terminal and
# approve/reject each write_audit_result call before it hits BigQuery.
STAGING_POLICIES = [
    policy.deny_all(),
    # Read-only tools — allowed without approval
    policy.allow("query_vendor_transactions"),
    policy.allow("list_invoices_in_gcs"),
    policy.allow("read_invoice_from_gcs"),
    # Write tools — human must approve each one
    policy.ask_user("write_audit_result", handler=compliance_officer_approval_handler),
]

# --- Production: fully autonomous, no human prompts ---
# Use this for unattended execution (Cloud Run, cron jobs).
# Writes are auto-allowed but ONLY if the status is valid.
# No ask_user — there's no human at the terminal in production.
PRODUCTION_POLICIES = [
    policy.deny_all(),
    # Read-only tools — allowed
    policy.allow("query_vendor_transactions"),
    policy.allow("list_invoices_in_gcs"),
    policy.allow("read_invoice_from_gcs"),
    # Write tools — auto-allowed but only with valid status values
    policy.allow("write_audit_result",
                 when=lambda args: args.get("status", "") in VALID_STATUSES,
                 name="allow_valid_audit_writes"),
]

注意三个不同的策略层级和预发布与生产环境之间的关键差异：

DEVELOPMENT_POLICIES ：使用 allow_all()，意味着智能体可以做任何事情。仅在本地机器上使用，用于快速迭代和调试。

STAGING_POLICIES ：读取操作自动允许，但 write_audit_result 会触发 compliance_officer_approval_handler，暂停智能体并通过标准输入提示你。你在每次写入前审查供应商 ID、状态和金额。

PRODUCTION_POLICIES ：针对 Cloud Run 或 cron 上的无人值守执行设计。没有 ask_user（没有人在终端）。相反，写入操作自动允许，但仅当状态参数是有效值（MATCHED、DISCREPANCY、ESCALATED 或 UNMATCHED）时。任何错误生成或格式不正确的状态会被静默拒绝。

策略工具名必须与注册的函数名匹配，而不是通用类别。由于我们的工具是 query_vendor_transactions、list_invoices_in_gcs、read_invoice_from_gcs 和 write_audit_result，这些正是在 policy.allow() 和 policy.ask_user() 中使用的确切字符串。使用错误的名称（例如 bigquery_query）会导致静默不匹配，工具会被 deny_all() 拒绝。

团队中的每个智能体都有自己的配置文件，定义其系统指令（