从原型到生产:借助 Agents CLI 实现费用审批代理
使用 Agents CLI 将 ADK 代理搭建、测试、评估并部署到 Cloud Run,过程中会涉及追踪功能、BigQuery 日志记录以及 CI/CD 和 RAG 扩展功能的介绍。
来源: Building AI Agents in 30 Minutes from Prototype to Production
最近尝试部署 AI 智能体的人都知道其中的规律。在开发阶段,它表现良好:能够回答问题、调用合适的工具、做出合理决策。但一旦有人询问何时可以上线,几天时间就会在 Docker、身份验证、可观测性以及部署流程中消耗掉。
智能体本身的代码只需几小时就能完成,而围绕它的基础设施却需要数周时间来搭建。
接下来将介绍如何使用开源工具包 Agents CLI 来构建和部署生产级智能体,该工具包旨在承担这些部署工作。示例是一个包含业务规则、评估机制以及云部署功能的费用审批智能体,根据该工具包的宣传,从开始到结束大约只需半小时。不过这个时间仅适用于准备充分的账户,在实际环境中可能并非如此。
部署差距
使用 Agent Development Kit(ADK)、LangChain 或 LlamaIndex 构建智能体相当简单,但将其投入生产环境才是真正的挑战。一个可用于生产的智能体通常需要具备以下所有要素。
本地开发
- 智能体代码,即最核心的部分
- 本地测试环境
- 模拟数据或 API 接口模拟对象
测试与评估
- 测试用例定义
- 性能指标与评分标准
- 以大语言模型作为评估者的配置
- 回归测试套件
生产环境基础设施
- 容器镜像
- 适用于 Cloud Run 或 Kubernetes 等平台的云部署配置
- 身份认证与授权机制
- 机密信息管理
- CI/CD 流水线
可观测性
- 请求追踪功能
- 提示词与响应的日志记录
每项功能都意味着需要学习新的工具、阅读额外的文档,并将不同的系统整合在一起。编程助手可以帮助处理智能体逻辑,但在部署方面却力不从心,因为它不了解你的基础设施规范,从而导致反复调试。
Agents CLI 是什么
Agents CLI 是一个工具包,为 Google Cloud 上的整个智能体开发生命周期提供统一接口。它有两种工作模式。
- 智能体模式。你将“技能”安装到 Gemini CLI、Claude Code 或 Cursor 等编程助手中。这些技能会教授助手部署模式,使其能够做出正确的决策。
其核心理念是让系统而非助手去推测部署规范,系统能够获取机器可读的上下文信息,从而判断何时使用特定的部署目标、如何构建评估体系以及哪种可观测性模式最为有效。
示例:费用审批智能体
该智能体会审核员工的费用报销单,并根据公司政策决定是否批准。它通过Vertex AI调用Gemini模型来进行推理。具体要求如下:
- 自动批准金额低于50美元的常规报销申请
- 识别重复提交的报销单
- 对周末消费或陌生供应商等异常情况发出警报
- 将不确定的案例转交人工审核员处理
前提条件
请确保已安装以下内容:
# Required tools
Python 3.11+
uv package manager
Node.js (for skills installation)
Google Cloud SDK (gcloud)
您还需要一个已开启计费功能的 Google Cloud 项目,可在 Google Cloud 控制台创建。
安装工具包:
uvx google-agents-cli setup
接着使用 Google Cloud 进行身份验证,设置项目并启用所需的 API:
# Authenticate with Google Cloud
gcloud auth application-default login
# Set your project ID
gcloud config set project YOUR_PROJECT_ID
# Enable required APIs
gcloud services enable aiplatform.googleapis.com
gcloud services enable run.googleapis.com
gcloud services enable artifactregistry.googleapis.com
这样即为该代理配置了 Vertex AI 访问权限,它将通过 Vertex AI API 调用 Gemini 模型。
步骤 1:搭建项目框架
无需手动创建目录和配置文件,让命令行工具自动生成完整的项目结构。
使用 Gemini CLI、Claude 或 Cursor 等编程助手时,只需输入相应提示语即可:
Build an expense approval agent using agents-cli.
Auto-approve under $50, flag duplicates, escalate suspicious cases.
Start with a prototype.
助手会读取已安装的技能,并生成结构合理的项目。手动操作则需要执行三个命令:
agents-cli create expense-agent --prototype --yes
cd expense-agent
agents-cli install
--prototype 参数很重要。它仅生成代理代码和测试结构,而不包含后续所需的 Docker、Terraform 以及 CI/CD 等部署相关内容。生成的框架结构如下所示:
expense-agent/
├── app/
│ ├── agent.py # Agent definition
│ └── app_utils/ # Telemetry helpers
├── tests/
│ ├── eval/
│ │ ├── evalsets/ # Test cases for LLM-as-judge
│ │ └── eval_config.json
│ ├── integration/
│ └── unit/
├── pyproject.toml # Dependencies managed by uv
└── GEMINI.md # Guidance for coding agents
第2步:编写业务逻辑
接下来,在app/agent.py中实现审批逻辑。与大多数智能体框架一样,在ADK中,这些工具都是普通的Python函数。文档字符串非常重要,因为模型会读取它来决定何时以及如何调用每个工具。
工具1:查看费用历史记录。
import json
from adk.models import Gemini
from adk.agents import Agent
def check_expense_history(
employee_id: str,
vendor: str,
date: str,
amount: float
) -> str:
"""Check if this expense was already submitted.
Args:
employee_id: Employee identifier
vendor: Vendor name
date: Expense date (YYYY-MM-DD)
amount: Expense amount in USD
Returns:
JSON string with duplicate check results
"""
# In production, this would query your expense database
# Using mock data for demonstration
mock_history = [
{"date": "2026-04-15", "vendor": "Coffee Shop", "amount": 8.50},
{"date": "2026-04-15", "vendor": "Uber", "amount": 24.00},
]
for past in mock_history:
if (past["vendor"].lower() == vendor.lower() and
past["date"] == date and
abs(past["amount"] - amount) < 0.01):
return json.dumps({
"is_duplicate": True,
"message": f"Duplicate found: ${amount} at {vendor} on {date}"
})
return json.dumps({
"is_duplicate": False,
"message": "No duplicate found"
})
工具2:获取公司政策。
def get_company_policy(category: str) -> str:
"""Retrieve expense policy for a given category.Args:
category: Expense category (meals, transport, lodging, etc.)
Returns:
JSON string with policy rules
"""
policies = {
"meals": {"auto_approve_threshold": 50.00},
"transport": {"auto_approve_threshold": 50.00},
"lodging": {"auto_approve_threshold": 0.00}, # Always escalate
"default": {"auto_approve_threshold": 25.00}
}
policy = policies.get(category.lower(), policies["default"])
return json.dumps(policy)
工具3:转交给人工处理。
def escalate_to_human(
expense_id: str,
reason: str,
context: str
) -> str:
"""Route expense to human approval queue.Args:
expense_id: Unique expense identifier
reason: Why this needs human review
context: Additional context for reviewer
Returns:
Confirmation message
"""
# In production, this creates a ticket or notification
print(f"\n🚨 ESCALATED: {expense_id}")
print(f"Reason: {reason}\n")
return json.dumps({
"status": "escalated",
"ticket_id": f"REVIEW-{expense_id}"
})
随后,这些工具会被接入一个具有明确指令和安全限制的智能体中:
root_agent = Agent(
name="expense_approval_agent",
model=Gemini(model="gemini-3.1-pro-preview"), # Using Vertex AI
instruction="""You are an expense approval agent. Review expenses and make
approval decisions based on company policy.
Process for every expense:
1. Call get_company_policy() to check the category threshold
2. Call check_expense_history() to verify no duplicate exists
3. Make a decision: AUTO_APPROVE, ESCALATE, or REJECT
Auto-approve only if ALL of these conditions are true:
- Amount is under the threshold for that category
- No duplicate found in expense history
- Expense is from a business day (Monday-Friday)
- Vendor appears legitimate
Escalate to human if ANY of these conditions are true:
- Amount is over the category threshold
- Duplicate submission detected
- Weekend expense without clear business justification
- Uncertain about vendor legitimacy
- Category is "lodging" (always requires human review)
Safety rules (NEVER violate these):
- NEVER auto-approve any expense over $50
- NEVER approve without checking for duplicates first
- ALWAYS provide clear reasoning for your decision
- When in doubt, ESCALATE (false escalation is better than false approval)
Response format:
Decision: [AUTO_APPROVE | ESCALATE | REJECT]
Reasoning: [Clear explanation of why you made this decision]
Tools called: [Which tools you used]
""",
tools=[check_expense_history, get_company_policy, escalate_to_human],
)
请仔细阅读说明中的安全规则。即便政策允许批准金额高达50美元的餐费,也存在明确限制:在未经人工审核的情况下,系统绝不会自动批准超过50美元的任何费用。这是为了避免出现高昂的错误。
第3步:在本地进行测试
启动ADK测试环境:
agents-cli playground
它会在本机主机的8080端口打开一个网页界面。三种情况涵盖了基本用法。
常规审批。
Review this expense:
Employee: emp-123
Category: meals
Vendor: Coffee Shop
Date: 2026-04-18
Amount: $12.50
预期行为:代理调用get_company_policy(),看到50美元的阈值后,再调用check_expense_history(),确认没有重复记录后予以批准。
超过阈值。
Review this expense:
Employee: emp-456
Category: meals
Vendor: Restaurant
Date: 2026-04-18
Amount: $85.00
预期行为:由于金额超过50美元,代理会进行升级处理。
重复提交。
Review this expense:
Employee: emp-789
Category: transport
Vendor: Uber
Date: 2026-04-15
Amount: $24.00
预期行为:代理在模拟历史记录中找到重复项并上报。
手动测试可以验证正常流程,但正式上线还需要更系统化的方法。
第4步:结构化评估
大多数开发者会跳过这一步,而这正是代理在正式环境中出问题的重要原因。你需要具备明确通过或失败标准的测试用例。问题不在于“这样感觉对吗?”,而在于“这是否符合规则?”
编辑 tests/eval/evalsets/basic.evalset.json:
{
"eval_set_id": "expense_approval_eval",
"name": "Expense Policy Validation",
"eval_cases": [
{
"eval_id": "routine_approval",
"conversation": [
{
"user_content": {
"parts": [{
"text": "Review: emp-101, meals, Coffee Shop, 2026-04-18, $15.00"
}]
}
}
],
"session_input": {
"app_name": "app",
"user_id": "eval_user",
"state": {}
},
"expected_tool_calls": ["get_company_policy", "check_expense_history"],
"expected_output_keywords": ["approve", "approved"]
},
{
"eval_id": "escalate_over_threshold",
"conversation": [
{
"user_content": {
"parts": [{
"text": "Review: emp-102, meals, Restaurant, 2026-04-18, $85.00"
}]
}
}
],
"session_input": {
"app_name": "app",
"user_id": "eval_user",
"state": {}
},
"expected_tool_calls": ["get_company_policy", "escalate_to_human"],
"expected_output_keywords": ["escalate", "human review"]
},
{
"eval_id": "reject_duplicate",
"conversation": [
{
"user_content": {
"parts": [{
"text": "Review: emp-103, transport, Uber, 2026-04-15, $24.00"
}]
}
}
],
"session_input": {
"app_name": "app",
"user_id": "eval_user",
"state": {}
},
"expected_tool_calls": ["check_expense_history"],
"expected_output_keywords": ["duplicate", "escalate"]
}
]
}
运行评估:
agents-cli eval run
示例输出:
Running expense_approval_eval...
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
✓ routine_approval policy_compliance: 1.0 safety: 1.0 tool_usage: 1.0
✓ escalate_over_threshold policy_compliance: 1.0 safety: 1.0 tool_usage: 1.0
✓ reject_duplicate policy_compliance: 0.9 safety: 1.0 tool_usage: 0.9
Overall Pass Rate: 100% (3/3 cases)
Average Scores: policy_compliance=0.97, safety=1.0, tool_usage=0.97
评分由充当裁判的大型语言模型根据eval_config.json中的评估标准来完成。当有案例测试失败时,就需要不断调整智能体的指令或工具逻辑,直到所有案例都通过。
这个循环就是评估-修正循环,也是打造可靠生产级智能体的关键。没有评估的话只能寄希望于智能体能正常工作,而有了评估就能明确知晓其表现。
第5步:部署
由于智能体已能正常运行且通过了所有评估,现在就可以进行部署了。首先需要搭建部署基础设施:
agents-cli scaffold enhance --deployment-target cloud_run
这将生成一个 Dockerfile、Cloud Run 部署配置以及 Terraform 文件。随后设置项目并进行部署:
gcloud config set project YOUR_PROJECT_ID
agents-cli deploy
预期输出如下:
Building container image...
Pushing to us-central1-docker.pkg.dev/your-project/agents/expense-agent...
Deploying to Cloud Run...
✓ Deployment successful!
Service URL: https://expense-agent-xyz123-uc.a.run.app
Logs: https://console.cloud.google.com/run/detail/.../logs
agents-cli create 将请求发送到实时服务所需的总时间约为 30 分钟,不过实际首次运行时间还会取决于账户设置和权限。
第 6 步:可观测性
Cloud Trace 会自动启用,无需任何配置。要测试已部署的代理程序:
agents-cli run "Review expense: emp-501, transport, Uber, 2026-04-18, $15.00"
agents-cli run "Review expense: emp-502, meals, Restaurant, 2026-04-18, $95.00"
agents-cli run "Review expense: emp-503, lodging, Hilton, 2026-04-18, $200.00"
然后打开 Cloud Trace 查看:
- 包含延迟明细的完整请求追踪信息
- 每次 LLM 调用的持续时间
- 工具的执行时间
- 每条请求的端到端时间线
为便于调试和分析,可开启内容日志记录:
agents-cli infra single-project --project YOUR_PROJECT_ID
这会提供一个用于日志记录的服务账户、一个用于存储提示和响应的云存储桶、一个用于分析的 BigQuery 数据集,以及让代理记录每段对话的更新功能。之后你便可以在 BigQuery 中查询实际运行时的行为表现:
SELECT
timestamp,
JSON_EXTRACT_SCALAR(request_data, '$.user_content.parts[0].text') as request,
JSON_EXTRACT_SCALAR(response_data, '$.decision') as decision,
latency_ms
FROM `your-project.agent_analytics.conversations`
WHERE service_name = 'expense-agent'
ORDER BY timestamp DESC
LIMIT 10;
这对调试而言至关重要。当有人询问为何智能体会将一顿30美元的餐食升级处理时,你可以通过查询BigQuery来准确了解它的决策依据及原因。
该工具包的独特之处
有四项设计特点尤为突出。
1. 将技能作为机器可读的上下文。 Agents CLI提供的不仅仅是命令,还有能够指导编码助手正确使用这些命令的技能。运行agents-cli setup即可安装其中的七项:
google-agents-cli-workflow,用于开发生命周期管理google-agents-cli-adk-code,用于ADK Python API模式google-agents-cli-scaffold,用于项目框架搭建google-agents-cli-eval,用于评估方法google-agents-cli-deploy,用于部署目标设置
google-agents-cli-publish 用于 Gemini Enterprise 注册google-agents-cli-observability 用于追踪与日志记录助手会在需要时查阅这些工具。它不会凭猜测行事,而是知道何时该选择 Agent Runtime、Cloud Run 或 GKE,也清楚哪些“以大语言模型作为评判者”的模式有效,因此无需通过反复试验来调整评估指标。
2. 默认进行评估。 每个模板化项目从一开始就包含评估基础设施:用于案例的 tests/eval/evalsets/ 文件夹、用于评分标准的 eval_config.json 文件,以及“以大语言模型作为评判者”的配置。这符合现实情况,因为没有经过评估的智能体很少能投入实际使用,而将评估设为默认流程意味着智能体在部署前就会经过测试。
3. 默认具备可观测性。每台已部署的代理都会自动启用 Cloud Trace。在没有追踪信息的情况下调试生产环境中的代理几乎是不可能的。内容日志记录和 BigQuery 分析功能需要通过 agents-cli infra single-project 才能启用,但基本追踪功能则是必须的。
4. 基础设施由系统处理。代理的逻辑部分很有趣,但 Docker 配置、认证中间件以及 Terraform 文件则不然。CLI 会负责处理基础设施相关事宜,这样你就可以把时间集中在让代理变得更智能上。
进一步深入
该费用代理已满足既定要求,但实际部署时还需要其他一些组件。
实时数据。可以用对费用数据库的查询来替代模拟的历史数据:
import psycopg2
def check_expense_history(employee_id, vendor, date, amount):
conn = psycopg2.connect(DATABASE_URL)
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM expenses
WHERE employee_id = %s
AND vendor = %s
AND date = %s
AND amount = %s
""", (employee_id, vendor, date, amount))
# Process results...
CI/CD。 为测试环境和生产环境生成 GitHub Actions 工作流:
agents-cli infra cicd --runner github-actions
它创建了一个多环境流水线,其中对 main 的提交会部署到预发布环境,而带有标签的版本则会被部署到生产环境。
会话存储。可添加 Cloud SQL 以实现持久化会话:
agents-cli scaffold enhance --session-type cloud_sql
多智能体系统。adk_a2a模板能够让专业智能体之间进行协调,非常适合处理诸如事件响应或代码迁移之类的复杂工作流程:
agents-cli create coordinator-agent --agent-type adk_a2a
检索增强生成。利用向量搜索构建智能体框架:
agents-cli create docs-agent \
--agent-type agentic_rag \
--datastore agent_platform_vector_search
这会生成数据接入代码、向量搜索集成以及检索逻辑。
技术概要
这里的每项技术都针对生命周期中的某个阶段:通过原型标志实现快速启动的框架搭建、以文档化的Python函数形式提供的工具、指令中设定的严格安全约束、本地测试环境下的验证、带有修复循环的结构化评估、用于部署的生成式基础设施,以及便于监控的追踪与内容日志功能。这些技术共同构成了从原型到正式部署的路径。
核心要点
部署难题确实存在。许多智能体项目停滞不前,并非因为智能体本身性能不佳,而是因为将其投入实际使用需要耗费大量的基础设施建设工作。
Agents CLI并不能解决所有问题。你仍然需要设计良好的提示词、合适的工具以及彻底的测试。它所消除的是那些让智能体困在开发环境中的基础设施相关繁琐工作。其目标是让智能体工作流程更像运行create-react-app,而非手动搭建基础设施。