agents-cli 可观测性实战复用遥测链路实现用户反馈收集机制结构化日志 → Log Sink → BigQuery【免费下载链接】agents-cliThe CLI and skills that turn any coding assistant into an expert at creating, evaluating, and deploying AI agents on Google Cloud.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-cli本文基于 agents-cli 可观测性技能google-agents-cli-observability中的参考文档 feedback-mechanism.md讲解如何把终端用户的反馈评分、点赞/点踩、自由文本收集起来并落地到 BigQuery 进行分析。读完本文你将掌握完整的三组件实现——Pydantic 请求模型、FastAPI 结构化日志端点、Terraform 日志 Sink 与 IAM 授权——并能结合 scaffold 模板中真实的遥测基础设施验证整条数据链路直接在自己的 scaffolded 项目中落地可用的反馈分析管道。整体思路复用 GenAI 日志的同一套模式该机制有一个明确前提项目由agents-cli scaffold/google-agents-cli-scaffold技能脚手架生成。它并非引入一套新的基础设施而是完全复用 prompt-response logging 已经铺设好的遥测链路模式与 GenAI 日志导出一致结构化日志jsonPayload → Cloud Logging log sink → BigQuery整个机制由三个组件构成分别对应一次代码改动请求模型 端点和一次基础设施改动Terraform sink IAM 绑定组件职责落点位置scaffolded 项目内1. 请求模型定义反馈载荷带固定的 discriminator 字段供 sink 过滤app/app_utils/如typing.py2. FastAPI 端点接收 POST 并以结构化日志写入 Cloud Loggingapp/fast_api_app.py3. Terraform Log Sink按 discriminator 字段过滤并路由到遥测 BigQuery dataset附带 IAM 写授权deployment/terraform/single-project/telemetry.tf及cicd/变体scaffold 模板中已经存在的遥测基础设施BigQuery dataset、GenAI 日志 sink、外部表等可以在本仓库直接查证Python 模板的单项目与 CI/CD 两套写法分别位于 single-project/telemetry.tf 对应模板 telemetry.tf 和 cicd/telemetry.tf。反馈 sink 就加在它们旁边。组件一请求模型Pydantic 固定 discriminator在应用存放请求/响应模型的位置例如app/app_utils/typing.py定义一个 Pydantic 模型。关键是带一个固定取值的 discriminator 字段让 log sink 的过滤条件可以稳定地匹配它import uuid from typing import Literal from pydantic import BaseModel, Field class Feedback(BaseModel): Represents feedback for a conversation. score: int | float text: str | None log_type: Literal[feedback] feedback service_name: Literal[project-name] project-name user_id: str Field(default_factorylambda: str(uuid.uuid4())) session_id: str Field(default_factorylambda: str(uuid.uuid4()))字段说明score评分值支持整数或浮点数如 1–5 星或 -1/0/1 的点赞/点踩text自由文本反馈可选默认为空字符串log_type固定为feedback是 sink 过滤的第一条件service_name固定为你的项目名注意是 agents-cli 项目名不是 GCP project ID是 sink 过滤的第二条件user_id/session_id默认由uuid.uuid4()随机生成保证不泄露真实身份信息。log_type和service_name两个字段是 log sink 过滤所依赖的锚点必须保持稳定——如果改了取值sink 的filter表达式也必须同步修改否则数据会静默丢失。隐私注意text是自由输入的用户文本会原样落到 Cloud Logging 和 BigQuery。如果无法接受保留 PII应对其做脱敏或直接省略保持user_id/session_id不透明默认的随机 UUID 即可如果确实要保留这类数据建议给遥测 dataset 设置表过期时间expiration。组件二FastAPI 端点写入结构化日志在app/fast_api_app.py中新增一个端点把载荷作为结构化日志条目写入——这样字段会进入jsonPayload而不是拼成一条纯文本 message。结构化是整条链路成立的前提sink 的过滤表达式正是匹配jsonPayload.log_type这类字段。from google.cloud import google_cloud_logging import logging as google_cloud_logging logging_client google_cloud_logging.Client() logger logging_client.logger(__name__) app.post(/feedback) def collect_feedback(feedback: Feedback) - dict[str, str]: Collect and log feedback. logger.log_struct(feedback.model_dump(), severityINFO) return {status: success}实现要点Cloud Logging 客户端在模块作用域创建一次避免每次请求都构造客户端使用logger.log_struct(...)而非logger.info(...)log_struct会把model_dump()的每个字段平铺进日志条目的jsonPayload与 sink 的过滤条件逐字段对应severityINFO是有意为之——如果用了 ERROR 级别反馈日志会触发错误告警污染监控信号端点返回简单的{status: success}把写日志成功作为成功语义。从源码结构看scaffold 生成的 ADK 项目里app实例本身由 ADK 的get_fast_api_app(...)工厂创建见模板 fast_api_app.py因此把app.post(/feedback)追加到该文件中、与框架自带的adk_api/a2a路由共存是模板结构上完全自然的扩展方式。组件三Terraform Log Sink 与 IAM 授权在deployment/terraform/single-project/telemetry.tfCI/CD 项目则在cicd/变体中添加一个 log sink把反馈条目路由到遥测 BigQuery dataset并授予 sink 自身writer_identity写权限resource google_logging_project_sink feedback_logs_to_bq { name ${var.project_name}-feedback project var.project_id destination bigquery.googleapis.com/projects/${var.project_id}/datasets/${google_bigquery_dataset.telemetry_dataset.dataset_id} filter jsonPayload.log_type\feedback\ jsonPayload.service_name\${var.project_name}\ unique_writer_identity true bigquery_options { use_partitioned_tables true } depends_on [google_bigquery_dataset.telemetry_dataset] } resource google_bigquery_dataset_iam_member feedback_logs_bq_writer { project var.project_id dataset_id google_bigquery_dataset.telemetry_dataset.dataset_id role roles/bigquery.dataEditor member google_logging_project_sink.feedback_logs_to_bq.writer_identity }关键参数逐一说明参数作用filter只放行同时满足jsonPayload.log_typefeedback且jsonPayload.service_name项目名的日志条目与请求模型的两个 discriminator 字段严格对应unique_writer_identity true为该 sink 生成独立的 writer 服务账号IAM 授权粒度到单个 sink而不是共享账号bigquery_options.use_partitioned_tablessink 自动创建的表按日期分区控制查询成本destination复用 scaffold 已创建的telemetry_dataset不新建 datasetIAM 绑定sink 的writer_identity需要roles/bigquery.dataEditor才能向 dataset 写数据缺这条绑定 sink 会写失败这套写法与模板中已有的 GenAI 日志 sinkgenai_logs_to_bq模式完全一致同样的unique_writer_identity 同样的google_bigquery_dataset_iam_member授权结构可以直接对照 telemetry.tf 中现有 sink 的写法。cicd 变体的多项目写法CI/CD 模板中所有遥测资源都按for_each local.deploy_project_ids展开到每个部署项目。新增反馈 sink 时同样要加for_each并用[each.key]/[each.value]索引引用资源与该文件其他 sink 保持一致可参考 cicd/telemetry.tf 中genai_logs_to_bq的写法project each.valuedestination ...datasets/${google_bigquery_dataset.telemetry_dataset[each.key].dataset_id}IAM 成员member google_logging_project_sink.feedback_logs_to_bq[each.key].writer_identity与 scaffold 遥测基础设施的衔接反馈机制落到的目标 dataset 并非凭空创建而是 scaffold 模板中telemetry.tf已经声明的google_bigquery_dataset.telemetry_dataset其中几个细节值得注意dataset 命名自动做连字符替换。模板中dataset_id replace(${var.project_name}_telemetry, -, _)见 telemetry.tf因为 BigQuery dataset 名不允许连字符。例如项目名my-agent对应 datasetmy_agent_telemetrysink 表自动创建第一次写入时Cloud Logging 会在该 dataset 中按日志命名自动创建一张日期分区表无需预先建表——这与模板中 GenAI 日志导出表genai_logs_table的预创建 schema 演进策略不同反馈表属于后者之外的常规 sink 建表路径同一 dataset 的汇聚效果反馈表与 GenAI 遥测数据completions 外部表、日志导出表、completions_view视图位于同一个project_name_telemetrydataset 下便于在同一分析环境中把用户反馈与 LLM 交互记录关联分析。另外应用侧的服务账号需要具备写日志能力才能成功调用log_struct。从源码结构看scaffold 中应用 SA 的角色统一由var.app_sa_roles驱动批量授权见 iam.tfAgent Runtime 目标下完整的遥测相关环境变量LOGS_BUCKET_NAME、OTEL_*等也在部署配置中声明见 service.tf。如果端点报权限错误优先检查应用 SA 是否持有日志写入相关角色。验证链路部署后按两步验证。第一步确认日志已结构化写入 Cloud Logging。POST 一条反馈载荷后执行gcloud logging read jsonPayload.log_typefeedback --limit 5 --project PROJECT_ID能查到条目且字段平铺在jsonPayload中说明组件一、二工作正常sink 过滤条件也有匹配对象。第二步确认 sink 已投递到 BigQuery。首次写入后几分钟sink 投递有延迟到project_name_telemetrydataset注意连字符已替换为下划线中查询 sink 自动创建的分区表确认有行# 先发现实际 dataset 名连字符 → 下划线 bq ls --project_id${PROJECT_ID} # 查询反馈导出表 bq query --use_legacy_sqlfalse \ SELECT score, COUNT(*) FROM \${PROJECT_ID}.${PROJECT_NAME//-/_}_telemetry.sink_table\ GROUP BY score如果日志能查到但 BQ 无行排查方向依次为sink 的filter与模型字段是否完全一致尤其是service_name是否用了 agents-cli 项目名而非 GCP project ID、IAM 绑定是否已 apply、以及是否等待了足够长的投递延迟。小结该机制的设计核心在于零新增基础设施请求模型提供两个稳定的 discriminator 字段FastAPI 端点用log_struct保证字段进入jsonPayloadTerraform 侧用一条filter表达式把数据从海量日志中精确挑出并复用既有的遥测 dataset。三个组件中任何一个字段值不一致都会导致链路静默断裂因此落地时把log_type/service_name的取值作为契约固定下来并用gcloud logging read加 BQ 计数查询完成端到端验证即可得到一条可持续运营的用户反馈分析管道。更多上下文可参考同技能目录下的 SKILL.md 与 cloud-trace-and-logging.md。【免费下载链接】agents-cliThe CLI and skills that turn any coding assistant into an expert at creating, evaluating, and deploying AI agents on Google Cloud.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-cli创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考