ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

AgentOps 中的 IBM Watsonx AI 可观测性实战:三类示例与源码级埋点解析

AgentOps 中的 IBM Watsonx AI 可观测性实战:三类示例与源码级埋点解析 AgentOps 中的 IBM Watsonx AI 可观测性实战三类示例与源码级埋点解析【免费下载链接】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本文基于 AgentOps 仓库中 examples/watsonx/README.md 及其配套的三个示例脚本完整讲解如何在 IBM Watsonx AI 上完成文本生成、流式对话与分词/模型详情三类任务的监控接入你将掌握环境与凭据配置方法、可直接运行的完整示例代码以及 AgentOps 底层WatsonxInstrumentor如何通过 OpenTelemetry 包装ModelInference各方法、在流式场景下如何还原 token 用量与完整输出内容的实现原理。一、示例目录结构与前置条件examples/watsonx/目录包含三组“Notebook 等价 Python 脚本”的示例watsonx-text-chat.py、watsonx-streaming.py、watsonx-tokeniation-model.py各自对应同名.ipynb覆盖 IBM Watsonx AI 最常见的三类使用场景场景示例文件演示内容基础文本生成与对话补全watsonx-text-chat.py基础文本生成、带 system/user 消息的 chat completion、多轮不同主题的对话流式生成watsonx-streaming.py流式文本生成、流式 chat completion、流式响应 chunk 的解析与拼接分词与模型详情watsonx-tokeniation-model.py用 Watsonx AI 模型对文本分词、获取模型详情、对比不同模型的 tokenization 结果1.1 依赖与安装按照 examples/watsonx/README.md 的说明运行前提为一个带有 API key 的 IBM Watsonx AI 账号Python 版本要求 3.10 3.13安装依赖pip install agentops ibm-watsonx-ai python-dotenv其中 examples/watsonx/requirements.txt 声明的核心依赖即ibm-watsonx-ai。需要注意版本约束从 instrumentor.py 中InstrumentorConfig的dependencies字段可以看到AgentOps 声明的最低依赖版本为ibm-watsonx-ai 1.3.11低于该版本时埋点逻辑无法正常匹配被包装的方法。1.2 环境变量配置在项目根目录创建.env文件示例中通过python-dotenv的load_dotenv()加载WATSONX_URLhttps://your-region.ml.cloud.ibm.com WATSONX_API_KEYyour-api-key-here WATSONX_PROJECT_IDyour-project-id-here此外三个示例脚本还会显式加载 AgentOps 侧的AGENTOPS_API_KEY未设置时回退为占位符字符串并在初始化前写入WATSONX_API_KEY。二、示例一基础文本生成与对话补全watsonx-text-chat.py 完整展示了“初始化 AgentOps → 配置 Watsonx 凭据 → 生成 → 对话 → 清理连接 → 校验 span”的标准流程核心代码如下import agentops from ibm_watsonx_ai import Credentials from ibm_watsonx_ai.foundation_models import ModelInference from dotenv import load_dotenv import os # 加载环境变量 load_dotenv() os.environ[AGENTOPS_API_KEY] os.getenv(AGENTOPS_API_KEY, your_api_key_here) # 初始化 AgentOpstrace_name 与 tags 用于在控制台中标识这次运行 agentops.init(trace_nameWatsonX Text Chat Example, tags[watsonx-text-chat, agentops-example]) # 凭据配置URL 未设置时默认指向 eu-de 区域 os.environ[WATSONX_API_KEY] os.getenv(WATSONX_API_KEY, your_watsonx_api_key_here) credentials Credentials( urlos.getenv(WATSONX_URL, https://eu-de.ml.cloud.ibm.com), api_keyos.environ[WATSONX_API_KEY], ) project_id os.getenv(WATSONX_PROJECT_ID, your-project-id-here) # 文本生成google/flan-ul2 模型 gen_model ModelInference(model_idgoogle/flan-ul2, credentialscredentials, project_idproject_id) response gen_model.generate_text(Write a short poem about artificial intelligence:) print(fGenerated Text:\n{response}) # 对话补全meta-llama/llama-3-3-70b-instruct 模型 chat_model ModelInference( model_idmeta-llama/llama-3-3-70b-instruct, credentialscredentials, project_idproject_id ) messages [ {role: system, content: You are a helpful AI assistant.}, {role: user, content: What are the three laws of robotics?}, ] chat_response chat_model.chat(messages) print(fChat Response:\n{chat_response[choices][0][message][content]}) # 脚本换一组 system/user 消息再做一次对话验证多次调用都能被记录 # 清理关闭与模型服务建立的持久连接 gen_model.close_persistent_connection() chat_model.close_persistent_connection()脚本末尾还有一个值得注意的动作——程序化校验本次 trace 的 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}) raise这一模式在三个示例中完全一致适合在 CI 或集成测试中快速验证“SDK 是否真的把 LLM 调用记录下来”而不必人工登录控制台查看。三、示例二流式生成与流式对话watsonx-streaming.py 演示了两类流式接口的消费方式。其关键差异在于两种流返回的 chunk 结构不同# 流式文本生成generate_text_stream 逐块 yield 字符串 prompt List 3 benefits of machine learning: stream_response gen_model.generate_text_stream(prompt) full_stream_response for chunk in stream_response: if isinstance(chunk, str): print(chunk, end, flushTrue) full_stream_response chunk # 流式对话chat_stream 逐块 yield 带 choices/delta 结构的 dict chat_stream_messages [ {role: system, content: You are a concise assistant.}, {role: user, content: Explain the concept of photosynthesis in one sentence.}, ] for chunk in chat_model.chat_stream(messageschat_stream_messages): if chunk and choices in chunk and chunk[choices]: delta chunk[choices][0].get(delta, {}) content_chunk delta.get(content) if content_chunk: print(content_chunk, end, flushTrue) full_chat_stream_response content_chunk也就是说generate_text_stream产出的是纯文本片段字符串而chat_stream产出的是 OpenAI 风格的{choices: [{delta: {content: ...}}]}字典结构。理解这一差异是阅读下文 AgentOps 流式埋点实现的前提。四、示例三分词与模型详情对比watsonx-tokeniation-model.py 围绕两个方法展开tokenize与get_details。示例先对短句和一段长文本分别分词再获取google/flan-ul2的模型详情然后初始化meta-llama/llama-3-3-70b-instruct获取其详情并做同样的分词最后用同一句The quick brown fox jumps over the lazy dog.分别经两个模型 tokenize直观对比两者 tokenizer 的差异model ModelInference(model_idgoogle/flan-ul2, credentialscredentials, project_idproject_id) tokens model.tokenize(Hello, how are you today?) model_details model.get_details() llama_model ModelInference( model_idmeta-llama/llama-3-3-70b-instruct, credentialscredentials, project_idproject_id ) llama_tokens llama_model.tokenize(The quick brown fox jumps over the lazy dog.) flan_tokens model.tokenize(The quick brown fox jumps over the lazy dog.)这个示例的实际价值在于分词差异直接决定 token 计费与上下文长度估算而 AgentOps 会把tokenize返回的token_count记录为 span 属性帮助你在 Agent 应用中定位“为什么同一个 prompt 在不同模型下 token 数不同”。五、源码级实现WatsonxInstrumentor 如何自动埋点上面三个示例中并没有任何一行显式的“埋点代码”——这正是 AgentOps 的设计agentops.init()之后agentops/instrumentation/init.py 注册的WatsonxInstrumentor会基于wrapt.wrap_function_wrapper对ibm_watsonx_aiSDK 的方法进行运行时包装。5.1 被包装的六个端点从 instrumentor.py 的WRAPPED_METHODS列表可以看到全部包装对象都作用于ibm_watsonx_ai.foundation_models.inference.ModelInference类被包装方法生成的 Span 名属性提取 handlergeneratewatsonx.generateget_generate_attributesgenerate_text_streamwatsonx.generate_text_stream专用流式 wrapperchatwatsonx.chatget_chat_attributeschat_streamwatsonx.chat_stream专用流式 wrappertokenizewatsonx.tokenizeget_tokenize_attributesget_detailswatsonx.get_detailsget_model_details_attributes普通方法走CommonInstrumentor的标准包装路径而两个流式方法因为返回的是生成器、需要在迭代过程中持续收集数据所以在WatsonxInstrumentor._custom_wrapinstrumentor.py中被从标准列表剔除改用generate_text_stream_wrapper/chat_stream_wrapper做自定义包装_custom_unwrap则通过 OpenTelemetry 的unwrap对称地移除包装保证 instrumentation 可逆。5.2 流式场景下的 TracedStream边消费边记账流式埋点的核心实现是 stream_wrapper.py 中的TracedStream类。它包装原始流迭代器在每次yield之前做三件事提取增量内容优先尝试读取底层生成器帧的局部变量gi_frame.f_locals中以data:开头的 SSE 原始载荷解析出model_id、input_token_count、generated_token_count若该内部结构不可得则回退到直接解析用户可见的 chunk——字符串直接作为生成片段字典则取choices[0].delta.content。这与第四小节中两种流的 chunk 形态一一对应。实时回写 span 属性解析到finish_reason stop的终结块时进一步尝试从内部状态读取最终usageprompt_tokens/completion_tokens立即更新LLM_USAGE_PROMPT_TOKENS、LLM_USAGE_COMPLETION_TOKENS与LLM_USAGE_TOTAL_TOKENS等 span 属性迭代过程中也会持续刷新累计值。收尾记录在finally块中把累积的完整输出写入COMPLETION_CONTENT角色标记为assistant、类型为text再次确认 token 计数后调用span.end()。由于finally语义即使用户中途break跳出迭代span 也一定被关闭且已带上当前累计的输出内容。两个入口 wrappergenerate_text_stream_wrapper与chat_stream_wrapperstream_wrapper.py分别在调用前记录请求侧信息前者写入 prompt 内容与LLM_REQUEST_TYPEcompletion后者遍历 messages 逐条写入角色与内容并对 list 形式的复杂 content 做了文本抽取两者都打上LLM_REQUEST_STREAMINGtrue标记若底层调用抛异常则会record_exception并记录ERROR_MESSAGE、ERROR_TYPE后再原样抛出即监控不会吞掉业务异常。5.3 非流式响应的属性提取对于同步方法属性提取集中在 attributes/attributes.py。以get_generate_attributes为例attributes.py请求侧从参数提取 prompt写入PROMPT_ROLEuser、PROMPT_CONTENT、PROMPT_TYPE响应侧从返回值的results数组逐条提取generated_text作为 completion 内容把input_token_count/generated_token_count映射为 prompt/completion token 用量并计算 total同时记录model_id与stop_reasonget_tokenize_attributes则记录ibm.watsonx.tokenize.result并把token_count作为 prompt tokens 用量get_model_details_attributes会挑选model_id、provider、number_params、input_tier/output_tier等字段以ibm.watsonx.model.*前缀写入 span。这些属性最终决定了 AgentOps 控制台中每次调用展示的模型、耗时、token 用量与输入/输出内容——也就是 README 中“monitor and analyze your AI applications”一句背后的具体机制。六、使用要点与适用前提模型选择三个示例使用google/flan-ul2文本生成与meta-llama/llama-3-3-70b-instruct对话补全两个模型model_id是ModelInference构造参数可替换为你在 IBM Watsonx 平台上可用的其他模型区域与项目WATSONX_URL决定服务区域示例默认eu-deWATSONX_PROJECT_ID决定配额归属两者均从环境变量读取请勿硬编码版本前提Python 3.10 3.13、ibm-watsonx-ai 1.3.11否则自动埋点可能因 SDK 内部结构差异而失效可验证性每个示例结尾的agentops.validate_trace_spans调用提供了“埋点是否生效”的程序化断言建议保留到自动化流程中连接管理示例显式调用close_persistent_connection()关闭与模型服务的持久连接长运行服务中应注意同样的资源释放。通过这份目录下的示例与对应的 instrumentor.py、stream_wrapper.py 源码你可以完整复现从环境配置、三类典型调用到 span 校验的闭环并理解 AgentOps 在 IBM Watsonx AI 场景下的埋点边界与数据流向。【免费下载链接】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),仅供参考
返回列表