Ambient Expense Agent 实战基于 ADK 2.0 Workflow 的事件驱动费用审批系统Pub/Sub 触发 人机协同审批【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples导读本文围绕 adk-samples 仓库中的 core/python/ambient-expense-agent 示例深入讲解如何用 Agent Development KitADK2.0 构建一个没有聊天界面的 ambient agent费用报销单以 Pub/Sub 事件的形式到达被推送到 ADK trigger 端点后流入图结构的Workflow。业务规则$100阈值路由由代码确定性执行只有高价值报销才进入 LLMreview_agent做风险评估并借助RequestInput暂停工作流等待人工审批HITL。读完本文你将掌握事件驱动 agent 的触发接入、图工作流条件路由、基于 ADK 会话事件的 HITL 审批发现与恢复以及 Cloud Run Pub/Sub IAP Cloud Monitoring 的一键部署方案。一、项目定位什么是 Ambient Agent这个示例的核心设计意图见 AGENTS.md是做一个克隆即学习clone-and-study的 ambient agent——它与传统 request/response 聊天 agent 的关键区别在于没有交互式聊天循环用户不会对着它打字费用报销单以 Pub/Sub 事件的形式到来事件驱动入口事件被推送到 ADK trigger 端点POST /apps/{app}/trigger/pubsub业务规则在代码中、LLM 只做判断$100阈值路由是确定性的deterministicLLM 仅在高价值场景下做风险分析人机协同HITL高价值报销在RequestInput处暂停稍后由独立的审批 UI 恢复工作流。从源码看整个图本身非常紧凑expense_agent/agent.py 中root_agent只有三条边真正值得研究的是外围的 ambient plumbing认证的 Pub/Sub 推送、基于会话session的 HITL 发现机制、IAP 保护的审批 UI以及由 Terraform 统一供给的 Cloud Monitoring 邮件告警。适用场景继承自原文档的 When To Use 小节需要事件驱动 / ambient的 agent由 Pub/Sub 触发无人值守而非一问一答的聊天 agent希望业务规则用代码实现、LLM 只处理需要判断的场景确定性阈值路由LLM 仅做风险分析需要人机协同审批工作流运行中暂停稍后从独立 UI 恢复需要一个可复现的Cloud Run 双服务 Pub/Sub IAP Cloud Monitoring部署起点。二、端到端流程从 Pub/Sub 事件到审批闭环原文档用一张 ASCII 流程图精确描述了完整链路这里完整继承并结合源码逐步解读expense published to Pub/Sub topic expense-reports - authenticated OIDC push - POST /apps/expense_agent/trigger/pubsub - parse_expense_email (base64/plain JSON - ExpenseData fields) - route_by_amount ($100 threshold, stashes expense_data in ctx.state) | | $100 $100 | | auto_approve review_agent (LLM) (logs INFO, done) - emit_expense_alert (JSON stdout) - request_approval (RequestInput - PAUSE)工作流暂停期间一条结构化的WARNING日志会触发基于日志的指标 → 告警策略 → 邮件的链路通知经理。经理打开IAP 保护的审批 UI该 UI 通过扫描 ADK 会话事件找到挂起的adk_request_input并通过POST /run恢复工作流... PAUSED ... - manager approves/rejects in approval UI - frontend POST /approve - backend POST /run (functionResponse) - process_decision (logs approved/rejected, emits final summary) - done一个容易忽略但重要的设计细节Pub/Sub 推送在工作流一暂停PAUSE时就已完成——ack 期限不会为等待人类而保持HITL 恢复完全发生在带外out-of-band。这与 terraform/pubsub.tf 中ack_deadline_seconds 600推送订阅的 10 分钟上限的设置相互印证即使推送请求在几秒内完成订阅仍预留了足够的重试窗口。2.1 触发入口ADK trigger 端点expense_agent/fast_api_app.py 通过get_fast_api_app(..., trigger_sources[pubsub])构建 ADK 应用从而开放 Pub/Sub 触发端点app get_fast_api_app( agents_dirAGENTS_DIR, webFalse, trigger_sources[pubsub], )其中AGENTS_DIR指向项目根目录使 ADK 能发现expense_agent/作为 agent 包包含agent.py与__init__.py。webFalse表示不启用 ADK 自带 Web UI仅保留 API 能力。2.2 订阅名归一化中间件真实 Pub/Sub 推送请求中的subscription字段是完整资源路径projects/.../subscriptions/NAME而 ADK trigger 处理器会把订阅名用作会话的user_id。为了让前端能用同一个短名字查询会话fast_api_app.py增加了 Starlette 中间件将完整路径截断为短名源码位置app.middleware(http) async def normalize_pubsub_subscription(request: Request, call_next): if ( request.url.path.endswith(/trigger/pubsub) and request.method POST ): body await request.body() try: data json.loads(body) sub data.get(subscription, ) if / in sub: data[subscription] sub.rsplit(/, 1)[-1] request._body json.dumps(data).encode() except (json.JSONDecodeError, KeyError): pass return await call_next(request)这一设计的影响在 tests/test_integration.py 中有行为级验证test_subscription_normalization提交projects/my-project/subscriptions/test-sub完整路径后断言可用短名test-sub查询到会话列表。三、图工作流函数节点 LLM 子代理 HITL 暂停3.1 数据结构ExpenseDataPydantic 模型数据在此示例中指在图节点间流动的费用事件负载由 expense_agent/agent.py 中的ExpenseData模型定义class ExpenseData(BaseModel): amount: float Field(descriptionExpense amount in USD) submitter: str Field(descriptionEmail of the person who submitted) category: str Field(descriptionExpense category, e.g. travel, meals) description: str Field(descriptionWhat the expense is for) date: str Field(descriptionDate of the expense (YYYY-MM-DD))3.2 函数节点逐个拆解parse_expense_email源码接受 Pub/Sub trigger 的原始 JSON其data字段可能是 base64 编码真实 Pub/Sub或纯 JSON本地测试并做了安全默认值兜底def parse_expense_email(node_input: str) - Event: try: event json.loads(node_input) except json.JSONDecodeError: return Event(output{error: fInvalid JSON: {node_input[:200]}}) data event.get(data, {}) if isinstance(data, str): try: data json.loads(base64.b64decode(data)) except Exception: return Event(output{error: fFailed to decode data: {data[:200]}}) return Event(output{ amount: float(data.get(amount, 0)), submitter: data.get(submitter, unknown), category: data.get(category, other), description: data.get(description, ), date: data.get(date, ), })route_by_amount源码是条件路由核心把解析出的 dict 存入ctx.state[expense_data]供 HITL 节点与前端在暂停后读取再依据config.review_threshold返回路由事件def route_by_amount(node_input: dict, ctx: Context) - Event: ctx.state[expense_data] node_input amount node_input.get(amount, 0) if amount config.review_threshold: return Event(routeNEEDS_REVIEW, outputnode_input) return Event(routeAUTO_APPROVE, outputnode_input)auto_approve源码对低价值报销直接放行并输出结构化 INFO 日志这正对应severity: INFO的log_entry结构decision、amount、submitter、category字段。3.3 LLM 子代理review_agent只有 $100的报销才会进入 review_agent。它以modesingle_turn运行输入模式为ExpenseData仅暴露一个工具emit_expense_alert提示词要求检查异常类别、模糊描述、可疑整数金额、超高价值$1000或潜在违规等风险因素并输出包含金额、提交人、类别、风险等级low/medium/high、风险因素与建议approve / request-more-info / escalate的结构化审查。emit_expense_alert工具源码打印一条alert_type: expense_review的 WARNING 结构化日志——这是整个告警链路的源头def emit_expense_alert(submitter, amount, category, risk_summary) - dict: log_entry { severity: WARNING, message: fExpense review alert: ${amount:.2f} from {submitter} — {risk_summary}, alert_type: expense_review, submitter: submitter, amount: amount, category: category, risk_summary: risk_summary, } print(json.dumps(log_entry), flushTrue) return {status: alert_emitted, submitter: submitter, amount: amount}3.4 HITL 暂停与恢复request_approval源码是 HITL 的核心它 yield 一个RequestInput让 ADK 运行时将暂停面暴露给 UI工作流保持暂停直到有人恢复会话通过审批 UI 或POST /rundef request_approval(node_input, ctx: Context): expense ctx.state.get(expense_data, {}) yield RequestInput( messageExpense requires manager approval. Approve or reject., payloadexpense, )process_decision源码接收人类响应{decision: approve}或{decision: reject}判定通过/拒绝后输出结构化日志与最终摘要事件。3.5Workflow图装配agent.py 末尾 用三条边把上述节点织成一个混合函数/LLM 图root_agent Workflow( nameexpense_processor, edges[ (START, parse_expense_email, route_by_amount), ( route_by_amount, { AUTO_APPROVE: auto_approve, NEEDS_REVIEW: review_agent, }, ), (review_agent, request_approval, process_decision), ], )route_by_amount节点的路由表正是对Event(routeAUTO_APPROVE)/Event(routeNEEDS_REVIEW)的消费端低价值直接auto_approve结束高价值则依次经过review_agent → request_approval → process_decision。四、配置与认证引导config.pyexpense_agent/config.py 在导入时完成认证模式引导并暴露两个配置旋钮若设置了GOOGLE_API_KEY则强制GOOGLE_GENAI_USE_VERTEXAIFalse走AI Studio模式否则调用google.auth.default()应用默认凭据 ADC设置GOOGLE_CLOUD_PROJECT与GOOGLE_CLOUD_LOCATIONglobal强制GOOGLE_GENAI_USE_VERTEXAITrue走Vertex AI模式model从MODEL_NAME环境变量读取无默认值review_threshold 100.0是路由阈值。expense_agent/__init__.py在子模块导入前执行load_dotenv()因此出现# noqa: E402注释确保.env中的变量在config.py导入时已就位。4.1 关键配置项速查表配置项来源默认值说明MODEL_NAME环境变量无未设置则为None指定 Gemini 模型.env.example中为gemini-3.5-flashGOOGLE_API_KEY环境变量无设置则启用 AI Studio 模式review_threshold代码常量100.0后端路由的金额阈值进入审查路径REVIEW_THRESHOLDfrontend/main.py常量100前端预过滤会话的阈值必须与后端保持一致五、部署管线Terraform 供给的生产级 ambient 基础设施5.1 Pub/Sub 触发接线pubsub.tfterraform/pubsub.tf 定义了一个expense-reportstopic 和认证推送订阅push_config指向后端 trigger URL 并携带oidc_tokenack_deadline_seconds 600推送订阅的最大 10 分钟 ack 期限retry_policy指数退避minimum_backoff 10smaximum_backoff 600sdead_letter_policy5 次投递失败后进入expense-reports-dead-letter死信 topicexpiration_policy.ttl 订阅不过期。resource google_pubsub_subscription expense_push { name expense-reports-push project var.project_id topic google_pubsub_topic.expense_reports.id push_config { push_endpoint ${google_cloud_run_v2_service.backend.uri}/apps/${var.agent_name}/trigger/pubsub oidc_token { service_account_email google_service_account.pubsub_invoker.email audience google_cloud_run_v2_service.backend.uri } } ack_deadline_seconds 600 retry_policy { minimum_backoff 10s maximum_backoff 600s } dead_letter_policy { dead_letter_topic google_pubsub_topic.dead_letter.id max_delivery_attempts 5 } expiration_policy { ttl } }5.2 双 Cloud Run 服务cloud_run.tfterraform/cloud_run.tf 定义了两个服务backend运行 ADK agentmin_instance_count 1保持一个实例常驻与内存会话相关见下文 Gotchasfrontendiap_enabled true通过环境变量注入BACKEND_URL自动指向 backend 的 URI、APP_NAME、PUBSUB_SUBSCRIPTION与USE_SERVICE_AUTHtrue。5.3 日志指标 → 邮件告警monitoring.tfterraform/monitoring.tf 把 stdout 日志变成经理邮箱里的告警邮件google_logging_metric用过滤器jsonPayload.alert_typeexpense_review从 Cloud Run 日志中计数DELTA / INT64google_monitoring_alert_policy在指标 0且持续 0s 时触发对齐周期 60s告警通过 email 通知渠道发送其 markdown 文档内嵌直达审批 UI 的链接${frontend.uri}/approval并写明操作步骤检查金额/提交人/类别/LLM 风险评估点击 Approve 或 Reject。5.4 最小权限 IAMiam.tfterraform/iam.tf 为每个调用方向准备独立服务账号pubsub_invokerexpense-agent-invoker持有 backend 的roles/run.invoker并授予 GCP 托管的 Pub/Sub 服务代理roles/iam.serviceAccountTokenCreator使其能为 OIDC 推送签发令牌frontend_invokerapproval-ui-invoker持有 backend 的roles/run.invoker供审批 UI 调用后端会话 APIIAP 绑定通知邮箱用户获得 frontend 的roles/iap.httpsResourceAccessor保证收到告警邮件的经理无需额外 IAM 配置即可打开审批 UI同时 IAP 服务代理获得 frontend 的roles/run.invoker以便代理请求backend 的默认计算服务账号被授予roles/aiplatform.user使其能调用 Gemini。六、审批 UI基于 ADK 会话 API 的 HITL 前端frontend/main.py 是一个轻量代理而非 agent属于独立的 uv 子项目有自己的pyproject.toml、uv.lock与Dockerfile。它与后端唯一的耦合是 ADK 会话 API 契约可指向任何 ADK agent。6.1 发现挂起审批GET /pending-approvals流程如下源码用PUBSUB_SUBSCRIPTION默认test-sub作为user_id查询后端的GET /apps/{app}/users/{user}/sessions列出该用户的所有会话预过滤只保留state.expense_data.amount REVIEW_THRESHOLD的会话对应REVIEW_THRESHOLD 100常量并发拉取这些会话的完整详情含 events调用_extract_pending_approval源码逐事件扫描寻找adk_request_input函数调用且没有匹配的 functionResponse同时顺带收集emit_expense_alert的风险摘要。def _extract_pending_approval(session: dict, user_id: str) - dict | None: request_input None responded False review_summary None for event in session.get(events, []): content event.get(content) or {} parts content.get(parts) or [] for part in parts: fc part.get(functionCall) if fc: name fc.get(name) if name emit_expense_alert: args fc.get(args, {}) if args.get(risk_summary): review_summary args[risk_summary] elif name adk_request_input: args fc.get(args, {}) payload args.get(payload) if isinstance(payload, str): try: payload json.loads(payload) except (json.JSONDecodeError, ValueError): pass request_input { interrupt_id: fc.get(id, ), message: args.get(message, ), payload: payload, } fr part.get(functionResponse) if fr and fr.get(name) adk_request_input: responded True if not request_input or responded: return None ...6.2 恢复工作流POST /approvePOST /approve源码把决策以functionResponse的形式转发到后端的POST /run以恢复暂停的工作流app.post(/approve) async def approve(request: Request): body await request.json() headers await _get_auth_headers() headers[Content-Type] application/json async with httpx.AsyncClient() as client: resp await client.post( f{BACKEND_URL}/run, jsonbody, headersheaders, timeout30.0 ) return resp.json()前端还实现了服务间认证Cloud Run 上通过USE_SERVICE_AUTHtrue走元数据服务器获取 ID token要求前端服务账号对 backend 持有roles/run.invoker本地则无认证直连_get_auth_headers返回空 dict。6.3 本地端口约定main.py底部源码的端口解析逻辑Cloud Run 注入的PORT必须被遵守本地场景下FRONTEND_PORT是本地专属逃生口用于避免前后端共享单一PORT冲突默认 8081 保持make dev-frontend的原始行为。七、行为验证in-process 集成测试core/python/ambient-expense-agent 不附带 eval 数据集行为覆盖主要来自tests/tests/test_runnability.py导入expense_agent.agentpatch 掉google.auth.default()断言root_agent已定义冒烟测试tests/test_integration.py通过httpx.ASGITransport在进程内驱动完整流程不启动真实服务器覆盖四种场景源码test_auto_approve$45报销直接走 auto-approve 路径test_review_and_hitl_approval$250报销触发 review → HITL 暂停 → 前端发现挂起审批 → 携带interrupt_id的functionResponse审批 → 最后一个事件为process_decision且status approved→ 挂起列表清空test_review_and_hitl_rejection$500报销被 reject 后日志状态为rejectedtest_subscription_normalization完整订阅路径被归一化为短名后可按短名查询会话。集成测试构造的审批请求体appName、userId、sessionId、newMessage.parts[0].functionResponse是理解POST /run契约的最佳范例approval_body { appName: expense_agent, userId: item[user_id], sessionId: item[session_id], newMessage: { role: user, parts: [ { functionResponse: { id: item[interrupt_id], name: adk_request_input, response: { result: json.dumps({decision: approve}), }, } } ], }, }八、数据处理要点无文档、无摄取本示例的数据即图节点间流动的费用事件负载由ExpenseData模型定型解码parse_expense_email同时接受 base64真实 Pub/Sub与纯 JSON本地测试两种data形态并用安全默认值兜底字段状态交接route_by_amount把解析后的 dict 存入ctx.state[expense_data]暂停后的 HITL 节点与前端都从这里读取会话是持久化基座挂起审批通过扫描 ADK 会话事件中未应答的adk_request_input发现会话user_id即 Pub/Sub 订阅名经中间件归一化未配置会话服务 URI因此使用 ADK 默认的内存会话服务结构化日志即输出auto_approve、emit_expense_alert、process_decision均以print(json.dumps(...))输出 stdoutCloud Run 将其捕获为 Cloud Logging 结构化条目驱动告警指标。九、Gotchas必须知道的坑原文档总结的七条实战经验逐条保留并补充源码依据会话是内存态的未配置持久化会话服务时挂起审批只存在于后端实例内存中——min_instance_count 1能保活一个实例但重启/重新部署会丢失进行中的审批。真实使用请接入持久化会话服务如 core/python/cross-session-memory 所演示的方向。$100阈值存在于两处config.review_threshold后端路由与frontend/main.py的REVIEW_THRESHOLD 100前端预过滤。两者不同步会导致 UI 与 agent 判定不一致。user_id 订阅名前端用PUBSUB_SUBSCRIPTION本地默认test-sub查询挂起审批若不与 trigger 请求中的subscription字段匹配UI 将什么都不显示——这正是fast_api_app.py归一化完整订阅路径的原因。模型来自MODEL_NAMEconfig.py读取os.getenv(MODEL_NAME)且无默认值未设置则为None。.env.example设为gemini-3.5-flash测试会显式设置。不要硬编码模型名也不要使用已弃用模型如gemini-2.0-flash、gemini-2.5-flash。导入期凭据没有GOOGLE_API_KEY时导入config会立即调用google.auth.default()——没有 ADC 的情况下无凭据导入会失败test_runnability.py通过 patch 规避。告警仅在部署后生效emit_expense_alert本地只是打印邮件路径日志指标 → 告警策略 → 通知渠道只存在于部署后的 Cloud Monitoring 配置中。IAP 传播延迟make deploy之后 IAP 可能需要5–10 分钟才完成传播审批 UI 早期出现403 Forbidden属正常现象。十、运行方式Makefile 目标速查原文档提供的 Makefile 目标完整继承如下对应 Makefile 中的具体实现命令作用make install/make install-frontenduv sync后端 / 前端子项目依赖make dev运行后端 trigger 服务器fast_api_app.py端口 8080make dev-frontend以BACKEND_URLlocalhost:8080运行审批 UI端口 8081make playground本地 ADK Web UIadk web端口 8501make testpytest tests/ -xvs冒烟 in-process 集成测试make lintcodespellruff check --fixruff formatmypymake deploy NOTIFICATION_EMAIL...通过 Cloud Build 构建双镜像再terraform apply整个栈make remote-test向已部署的 topic 发布一条$250测试报销make clean NOTIFICATION_EMAIL...terraform destroy告警策略传播原因 60s 后自动重试一次make deploy的四步流程Makefile 第 50-108 行启用 Artifact Registry 与 Cloud Build API → 创建镜像仓库并授予计算默认 SA 写入/读取权限 → 并行gcloud builds submit构建 backend 与 frontend 镜像 →terraform apply并回显 backend/frontend/approval URL 与 topic 名。注意PROJECT_ID取自gcloud config get-value project未设置时会直接报错退出。make remote-test发布的测试消息是理解 Pub/Sub 负载格式的最佳样例Makefile 第 117-120 行gcloud pubsub topics publish expense-reports \ --project$(PROJECT_ID) \ --message{amount:250.00,submitter:alicecompany.com,category:travel,description:Flight to NYC for client meeting,date:2026-04-10} \ --attributesourcemake-test十一、复用指南把这份模板搬进自己的项目原文档明确了两块可以原样复制的资产terraform/目录自包含两个 Cloud Run 服务、Pub/Sub含死信、IAM、IAP 与 Cloud Monitoring。复制目录后只需设置project_id、region、notification_email以及backend_image/frontend_image变量。它期望镜像已预先构建因此应先构建镜像Makefile 用 Cloud Build再terraform apply。frontend/是独立的 uv 子项目一个通用的ADK HITL 审批代理除 ADK 会话 API 契约外与后端零代码耦合通过BACKEND_URL、APP_NAME、PUBSUB_SUBSCRIPTION三个环境变量即可指向任意 ADK agent。expense_agent/内部零外部耦合agent 完全通过环境变量MODEL_NAME、GOOGLE_API_KEY/GOOGLE_CLOUD_*配置前端仅通过 HTTP 访问它。十二、小结Ambient Expense Agent 提供了一个值得反复研读的参考实现它把事件驱动触发、确定性业务规则路由、LLM 判断、HITL 暂停恢复、日志驱动告警这五件事清晰地拆解到 agent.py、fast_api_app.py、frontend/main.py 与 terraform 各层中。推荐按原文档建议的顺序研读源码先看 trigger 服务器ambient 接入缝、再看 Terraform 部署管线生产化基础设施、然后看审批 UIHITL 发现与恢复、最后回到 agent 图与配置才能完整理解事件如何抵达、审批如何回流。【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考