AgentOps 集成 Google GenAIGeminiSDK同步、流式与 Token 计数的自动观测实践【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops本文以 examples/google_genai/README.md 为核心讲解如何使用 AgentOps 自动为 Google Generative AIGeminiSDK 的调用埋点从环境准备、可直接运行的示例代码到 instrumentor 底层包装了哪些方法、捕获了哪些 span 属性与指标帮助你把 Gemini 的同步生成、流式生成与 token 计数完整纳入可观测体系。前置条件与依赖安装根据示例目录说明运行 gemini_example 需要Python 版本 3.10 3.13依赖安装pip install agentops google-generativeai示例目录的 requirements.txt 中同时列出了google-generativeai与google-genai两个包。需要注意仓库中实际的 instrumentation 实现instrumentor.py 声明的依赖为google-genai 0.1.0包装的是新版google-genai客户端from google import genai示例代码也正是基于该客户端编写因此建议以google-genai为准。另外运行示例需要两个环境变量环境变量用途AGENTOPS_API_KEYAgentOps 平台鉴权agentops.init时读取GEMINI_API_KEYGoogle GenAI 客户端genai.Client()鉴权示例通过dotenv加载本地.env文件并为两个变量提供了占位默认值见 gemini_example.py。示例代码逐段解析同步、流式与 Token 计数以下是对 gemini_example.py 的完整走读该脚本覆盖了三类调用形态同步生成、流式生成和 token 计数并用validate_trace_spans做程序化校验。初始化 AgentOps 与 Gemini 客户端agentops.init(trace_nameGoogle Gemini Example, tags[gemini-example, agentops-example]) client genai.Client()agentops.init()入口定义于 agentops/init.py内部委托给 AgentOpsClient.init完成初始化后AgentOps 会对已安装的第三方库执行自动插桩——其中就包括对google-genai客户端方法的包装示例代码本身不需要任何显式的埋点调用。trace_name与tags用于标识本次运行便于在 AgentOps 侧按标签检索。同步生成response client.models.generate_content( modelgemini-1.5-flash, contentsWhat are the three laws of robotics? ) print(response.text)client.models.generate_content是被插桩的核心同步入口之一。每次调用都会产生一个名为gemini.generate_content的 span见 instrumentor.py 中的 WrapConfig 定义自动记录模型名、prompt 内容、响应文本、token 用量等属性无需手动处理。流式生成response_stream client.models.generate_content_stream( modelgemini-1.5-flash, contentsExplain the concept of machine learning in simple terms. ) for chunk in response_stream: print(chunk.text, end)流式方法generate_content_stream不走通用的WrapConfig通道而是由专门的流式包装器处理STREAMING_METHODS 定义。其实现原理见下文 流式调用的 span 生命周期 一节。对调用方而言体验不变拿到的仍然是一个逐块 yield 文本的生成器。Token 计数token_response client.models.count_tokens( modelgemini-1.5-flash, contentsThis is a test sentence to count tokens. ) print(fToken count: {token_response.total_tokens})count_tokens以及compute_tokens同样在插桩列表内对应 span 名为gemini.count_tokens并从返回对象中提取total_tokens写入 token 用量属性get_token_counting_attributes。程序化校验 span 是否上报示例最后用 AgentOps 提供的校验接口确认所有 LLM span 已被正确记录try: agentops.validate_trace_spans(trace_contextNone) print(\n✅ Success! All LLM spans were properly recorded in AgentOps.) except agentops.ValidationError as e: print(f\n❌ Error validating spans: {e}) raisevalidate_trace_spans的实现在 agentops/validation.py。这一做法适合放进集成测试或 CI 中验证 instrumentation 确实生效、span 确实到达了 AgentOps 后端而不仅仅是打印日志“看起来正常”。Instrumentor 原理哪些方法被包装GoogleGenaiInstrumentor 继承自 CommonInstrumentor通过wrapt.wrap_function_wrapper对google.genai.models模块下的方法做运行时包装。从 WRAPPED_METHODS 列表 可以确认覆盖范围同步与异步成对被包装方法类.方法span 名属性提取器Models.generate_content/AsyncModels.generate_contentgemini.generate_contentget_generate_content_attributesModels.count_tokens/AsyncModels.count_tokensgemini.count_tokensget_token_counting_attributesModels.compute_tokens/AsyncModels.compute_tokensgemini.compute_tokensget_token_counting_attributesModels.generate_content_stream/AsyncModels.generate_content_streamgemini.generate_content_stream异步为_async后缀专用流式包装器两点设计值得注意流式方法单独处理。流式接口返回的是生成器普通的“调用前开始 span、调用后结束 span”模式无法覆盖整个消费周期因此 _custom_wrap 使用STREAMING_METHODS配置单独挂载包装器异步版本则挂generate_content_stream_async_wrapper。容错包装。_custom_wrap与 _custom_unwrap 均捕获AttributeError/ModuleNotFoundError若google-genai未安装或方法签名不匹配插桩会静默跳过并记录 debug 日志不会让宿主应用崩溃。此外instrumentor 通过 StandardMetrics.create_standard_metrics 注册了标准 LLM 指标输入/输出/总 token、操作耗时、异常计数等 OpenTelemetry metrics与 span 属性互补可用于聚合看板。Span 属性详解从请求参数到响应元数据请求侧模型名与生成参数映射extract_request_attributes 负责解析请求参数model参数支持字符串或带_model_name/name属性的模型对象统一写入LLM_REQUEST_MODELstream参数写入LLM_REQUEST_STREAMINGconfig即GenerateContentConfig对象通过 REQUEST_CONFIG_ATTRIBUTES 映射表 反射提取生成参数Google SDK 参数映射到的 span 属性temperatureLLM_REQUEST_TEMPERATUREmax_output_tokensLLM_REQUEST_MAX_TOKENStop_pLLM_REQUEST_TOP_Ptop_kLLM_REQUEST_TOP_KseedLLM_REQUEST_SEEDsystem_instructionLLM_REQUEST_SYSTEM_INSTRUCTIONpresence_penalty/frequency_penalty对应 penalty 属性stop_sequencesLLM_REQUEST_STOP_SEQUENCEScandidate_countLLM_REQUEST_CANDIDATE_COUNT也就是说如果你在示例中这样调用from google.genai import types response client.models.generate_content( modelgemini-1.5-flash, contents..., configtypes.GenerateContentConfig(temperature0.7, max_output_tokens1024), )temperature0.7与max_output_tokens1024会自动出现在对应 span 的属性中无需任何额外代码。Prompt 与响应侧多形态内容归一化_extract_content_from_prompt 处理 Gemini SDK 接受的各种contents形态——纯字符串、parts 列表、ContentDict字典、带parts属性的对象等——统一归一为文本_set_prompt_attributes 再按消息序号写入PROMPT_CONTENT/PROMPT_ROLE多轮对话时逐项编号。响应侧由 _set_response_attributes 处理response.model→ 响应实际模型response.usage_metadata中的prompt_token_count/candidates_token_count/total_token_count→ 对应 token 用量属性这是成本统计的数据来源响应文本优先取response.text若走candidates结构则逐 candidate 拼接parts中的文本并记录finish_reason。所有 span 都会带上LLM_SYSTEM Gemini以及库名/库版本LIBRARY_NAME等公共属性方便在平台侧按 provider 过滤。流式调用的 span 生命周期流式包装器的完整实现见 stream_wrapper.py。其关键逻辑span 提前创建调用generate_content_stream时立即以SpanKind.CLIENT打开 spanL50-L62写入请求属性并标记LLM_REQUEST_STREAMING True。逐块透传并累积内部instrumented_stream()生成器对原始流逐块yield同时拼接chunk.text到full_text并记住最后一个携带usage_metadata的 chunkL78-L99。流结束后收尾流正常耗尽时把累积的full_text写入完成内容属性从最后一个带元数据的 chunk 提取 prompt/candidates/total 三个 token 计数并将 span 状态置为OKspan 在finally中结束L101-L124。异常路径流消费中抛错会record_exception并写入错误消息/类型属性span 置为ERROR状态后原样抛出不影响业务代码的异常处理。抑制开关包装器检查 OpenTelemetry 的_SUPPRESS_INSTRUMENTATION_KEYL47-L48支持在特定上下文里临时关闭插桩。异步版 generate_content_stream_async_wrapper 逻辑完全对称span 名为gemini.generate_content_stream_async。验证与延伸阅读跑通 examples/google_genai/gemini_example.py 后validate_trace_spans返回成功即说明同步、流式、token 计数三类 span 均已正确记录对应的 notebook 版本为 gemini_example.ipynb。模块级说明见 google_genai 插桩 README其中列出的指标输入/输出/总 token、耗时、异常数即上文StandardMetrics注册的内容。如果你想为其他 providerOpenAI、Anthropic 等搭建同类观测可参考同级的 providers 目录结构 与 instrumentation 总览文档。【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
AgentOps 集成 Google GenAIGeminiSDK同步、流式与 Token 计数的自动观测实践【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops本文以 examples/google_genai/README.md 为核心讲解如何使用 AgentOps 自动为 Google Generative AIGeminiSDK 的调用埋点从环境准备、可直接运行的示例代码到 instrumentor 底层包装了哪些方法、捕获了哪些 span 属性与指标帮助你把 Gemini 的同步生成、流式生成与 token 计数完整纳入可观测体系。前置条件与依赖安装根据示例目录说明运行 gemini_example 需要Python 版本 3.10 3.13依赖安装pip install agentops google-generativeai示例目录的 requirements.txt 中同时列出了google-generativeai与google-genai两个包。需要注意仓库中实际的 instrumentation 实现instrumentor.py 声明的依赖为google-genai 0.1.0包装的是新版google-genai客户端from google import genai示例代码也正是基于该客户端编写因此建议以google-genai为准。另外运行示例需要两个环境变量环境变量用途AGENTOPS_API_KEYAgentOps 平台鉴权agentops.init时读取GEMINI_API_KEYGoogle GenAI 客户端genai.Client()鉴权示例通过dotenv加载本地.env文件并为两个变量提供了占位默认值见 gemini_example.py。示例代码逐段解析同步、流式与 Token 计数以下是对 gemini_example.py 的完整走读该脚本覆盖了三类调用形态同步生成、流式生成和 token 计数并用validate_trace_spans做程序化校验。初始化 AgentOps 与 Gemini 客户端agentops.init(trace_nameGoogle Gemini Example, tags[gemini-example, agentops-example]) client genai.Client()agentops.init()入口定义于 agentops/init.py内部委托给 AgentOpsClient.init完成初始化后AgentOps 会对已安装的第三方库执行自动插桩——其中就包括对google-genai客户端方法的包装示例代码本身不需要任何显式的埋点调用。trace_name与tags用于标识本次运行便于在 AgentOps 侧按标签检索。同步生成response client.models.generate_content( modelgemini-1.5-flash, contentsWhat are the three laws of robotics? ) print(response.text)client.models.generate_content是被插桩的核心同步入口之一。每次调用都会产生一个名为gemini.generate_content的 span见 instrumentor.py 中的 WrapConfig 定义自动记录模型名、prompt 内容、响应文本、token 用量等属性无需手动处理。流式生成response_stream client.models.generate_content_stream( modelgemini-1.5-flash, contentsExplain the concept of machine learning in simple terms. ) for chunk in response_stream: print(chunk.text, end)流式方法generate_content_stream不走通用的WrapConfig通道而是由专门的流式包装器处理STREAMING_METHODS 定义。其实现原理见下文 流式调用的 span 生命周期 一节。对调用方而言体验不变拿到的仍然是一个逐块 yield 文本的生成器。Token 计数token_response client.models.count_tokens( modelgemini-1.5-flash, contentsThis is a test sentence to count tokens. ) print(fToken count: {token_response.total_tokens})count_tokens以及compute_tokens同样在插桩列表内对应 span 名为gemini.count_tokens并从返回对象中提取total_tokens写入 token 用量属性get_token_counting_attributes。程序化校验 span 是否上报示例最后用 AgentOps 提供的校验接口确认所有 LLM span 已被正确记录try: agentops.validate_trace_spans(trace_contextNone) print(\n✅ Success! All LLM spans were properly recorded in AgentOps.) except agentops.ValidationError as e: print(f\n❌ Error validating spans: {e}) raisevalidate_trace_spans的实现在 agentops/validation.py。这一做法适合放进集成测试或 CI 中验证 instrumentation 确实生效、span 确实到达了 AgentOps 后端而不仅仅是打印日志“看起来正常”。Instrumentor 原理哪些方法被包装GoogleGenaiInstrumentor 继承自 CommonInstrumentor通过wrapt.wrap_function_wrapper对google.genai.models模块下的方法做运行时包装。从 WRAPPED_METHODS 列表 可以确认覆盖范围同步与异步成对被包装方法类.方法span 名属性提取器Models.generate_content/AsyncModels.generate_contentgemini.generate_contentget_generate_content_attributesModels.count_tokens/AsyncModels.count_tokensgemini.count_tokensget_token_counting_attributesModels.compute_tokens/AsyncModels.compute_tokensgemini.compute_tokensget_token_counting_attributesModels.generate_content_stream/AsyncModels.generate_content_streamgemini.generate_content_stream异步为_async后缀专用流式包装器两点设计值得注意流式方法单独处理。流式接口返回的是生成器普通的“调用前开始 span、调用后结束 span”模式无法覆盖整个消费周期因此 _custom_wrap 使用STREAMING_METHODS配置单独挂载包装器异步版本则挂generate_content_stream_async_wrapper。容错包装。_custom_wrap与 _custom_unwrap 均捕获AttributeError/ModuleNotFoundError若google-genai未安装或方法签名不匹配插桩会静默跳过并记录 debug 日志不会让宿主应用崩溃。此外instrumentor 通过 StandardMetrics.create_standard_metrics 注册了标准 LLM 指标输入/输出/总 token、操作耗时、异常计数等 OpenTelemetry metrics与 span 属性互补可用于聚合看板。Span 属性详解从请求参数到响应元数据请求侧模型名与生成参数映射extract_request_attributes 负责解析请求参数model参数支持字符串或带_model_name/name属性的模型对象统一写入LLM_REQUEST_MODELstream参数写入LLM_REQUEST_STREAMINGconfig即GenerateContentConfig对象通过 REQUEST_CONFIG_ATTRIBUTES 映射表 反射提取生成参数Google SDK 参数映射到的 span 属性temperatureLLM_REQUEST_TEMPERATUREmax_output_tokensLLM_REQUEST_MAX_TOKENStop_pLLM_REQUEST_TOP_Ptop_kLLM_REQUEST_TOP_KseedLLM_REQUEST_SEEDsystem_instructionLLM_REQUEST_SYSTEM_INSTRUCTIONpresence_penalty/frequency_penalty对应 penalty 属性stop_sequencesLLM_REQUEST_STOP_SEQUENCEScandidate_countLLM_REQUEST_CANDIDATE_COUNT也就是说如果你在示例中这样调用from google.genai import types response client.models.generate_content( modelgemini-1.5-flash, contents..., configtypes.GenerateContentConfig(temperature0.7, max_output_tokens1024), )temperature0.7与max_output_tokens1024会自动出现在对应 span 的属性中无需任何额外代码。Prompt 与响应侧多形态内容归一化_extract_content_from_prompt 处理 Gemini SDK 接受的各种contents形态——纯字符串、parts 列表、ContentDict字典、带parts属性的对象等——统一归一为文本_set_prompt_attributes 再按消息序号写入PROMPT_CONTENT/PROMPT_ROLE多轮对话时逐项编号。响应侧由 _set_response_attributes 处理response.model→ 响应实际模型response.usage_metadata中的prompt_token_count/candidates_token_count/total_token_count→ 对应 token 用量属性这是成本统计的数据来源响应文本优先取response.text若走candidates结构则逐 candidate 拼接parts中的文本并记录finish_reason。所有 span 都会带上LLM_SYSTEM Gemini以及库名/库版本LIBRARY_NAME等公共属性方便在平台侧按 provider 过滤。流式调用的 span 生命周期流式包装器的完整实现见 stream_wrapper.py。其关键逻辑span 提前创建调用generate_content_stream时立即以SpanKind.CLIENT打开 spanL50-L62写入请求属性并标记LLM_REQUEST_STREAMING True。逐块透传并累积内部instrumented_stream()生成器对原始流逐块yield同时拼接chunk.text到full_text并记住最后一个携带usage_metadata的 chunkL78-L99。流结束后收尾流正常耗尽时把累积的full_text写入完成内容属性从最后一个带元数据的 chunk 提取 prompt/candidates/total 三个 token 计数并将 span 状态置为OKspan 在finally中结束L101-L124。异常路径流消费中抛错会record_exception并写入错误消息/类型属性span 置为ERROR状态后原样抛出不影响业务代码的异常处理。抑制开关包装器检查 OpenTelemetry 的_SUPPRESS_INSTRUMENTATION_KEYL47-L48支持在特定上下文里临时关闭插桩。异步版 generate_content_stream_async_wrapper 逻辑完全对称span 名为gemini.generate_content_stream_async。验证与延伸阅读跑通 examples/google_genai/gemini_example.py 后validate_trace_spans返回成功即说明同步、流式、token 计数三类 span 均已正确记录对应的 notebook 版本为 gemini_example.ipynb。模块级说明见 google_genai 插桩 README其中列出的指标输入/输出/总 token、耗时、异常数即上文StandardMetrics注册的内容。如果你想为其他 providerOpenAI、Anthropic 等搭建同类观测可参考同级的 providers 目录结构 与 instrumentation 总览文档。【免费下载链接】agentopsPython SDK for AI agent monitoring, LLM cost tracking, benchmarking, and more. Integrates with most LLMs and agent frameworks including CrewAI, Agno, OpenAI Agents SDK, Langchain, Autogen, AG2, and CamelAI项目地址: https://gitcode.com/GitHub_Trending/ag/agentops创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
