# Skill（技能）观测数据上报 当智能体引入Skill（技能）能力后，Skill的运行状态往往是影响整体体验的关键环节。常见的线上问题包括：某个Skill加载变慢拖累了整体响应、Skill频繁执行失败但没有告警、Skill内容过长导致Token成本异常攀升。由于Skill的调用过程嵌套在智能体链路内部，这些问题在没有专项观测的情况下很难被发现和定位。 通过将Skill的调用链路（Trace）与运行指标（Metric）上报至AgentArts观测平台，可以实现： - 链路定位：在调用链分析中还原 Skill 的加载与释放过程，定位具体异常发生在哪个步骤。 - 稳定性监控：通过Skill调用次数与成功率趋势，及时发现Skill执行失败的规律。 - 性能分析：通过Skill调用平均耗时与Top5排行，识别耗时异常的Skill。 - 成本治理：统计Skill内容注入LLM时消耗的Token，量化Skill带来的额外成本。 #### 实现原理 Skill 的一次完整生命周期包含两个阶段：加载（skill_tool）和释放（skill_complete）。这两个阶段本质上是工具调用，因此在 Trace 中不单独创建 Skill Span，而是在对应的工具执行 Span 上附加 Skill 属性与事件，避免链路结构冗余。 一次包含 Skill 的完整调用链结构如下： ``` Root Span（agent_dialogue_session） ├── Model Span（llm_decision_call） ← LLM 决策：是否调用 Skill ├── Tool Span（skill_tool） ← Skill 加载 │ ├── gen_ai.operation.name = load_skill │ ├── gen_ai.skill.name = │ ├── gen_ai.skill.id = │ └── Event: skill.loaded ← 加载成功事件，记录路径 ├── Tool Span（skill_complete） ← Skill 释放 │ ├── gen_ai.operation.name = release_skill │ ├── gen_ai.skill.name = │ └── Event: skill.released ← 释放完成事件 └── Model Span（llm_final_answer） ← 携带 Skill 内容的最终 LLM 调用 ``` #### 上报机制 表1上报机制说明 | 数据类型 | 上报协议 | 接收端 | 说明 | |:---|:---|:---|:---| | Trace（调用链） | OTel gRPC（OTLP） | 华为云APM | 记录单次请求的完整调用路径与节点详情 | | Metric（指标） | OTel gRPC（OTLP） | 华为云AOM | 记录调用次数、耗时、成功率等聚合统计 | 两路数据均通过OpenTelemetry Python SDK上报，Trace使用BatchSpanProcessor批量发送，Metric使用PeriodicExportingMetricReader定期批量发送。 本地调试阶段同时启用ConsoleSpanExporter和ConsoleMetricExporter，将数据同步写入本地otel_debug.log，用于在看板出数据前确认本地采集是否正常。 ![](https://support.huaweicloud.com/bestpractice-agentarts/public_sys-resources/note_3.0-zh-cn.png) Trace数据写入APM，Metric数据写入AOM，两者均按实际使用量计费，并提供一定免费额度。生产环境部署前请参阅[APM计费说明](https://support.huaweicloud.com/price-apm2/apm_07_0019.html)和[AOM计费说明](https://support.huaweicloud.com/price-aom2/aom_07_0004.html)。 #### 指标上报逻辑 表2Skill专项指标 | 指标名 | 类型 | 触发时机 | 说明 | |:---|:---|:---|:---| | gen_ai.skill.call.count | Counter | Skill加载阶段完成时（skill_tool Span结束后）。 | Skill调用次数，成功与失败均计入，是计算Skill调用成功率的基础数据。 | | gen_ai.skill.duration | Histogram | Skill释放阶段完成时（skill_complete Span结束后）。 | 记录从Skill加载开始到释放完成的完整耗时，平台据此计算Skill调用平均耗时。 | | gen_ai.skill.error.count | Counter | Skill执行结果含错误时。 | 仅在Skill执行失败时累加，平台以call.count - error.count计算成功次数，进而得出成功率。 | | gen_ai.skill.token.usage | Counter | Skill内容加载完成后、LLM最终调用前。 | Skill内容（SKILL.md 正文）注入LLM时消耗的Token估算值。 | 上报上述指标时，必须在 Attributes 中携带以下维度字段，否则看板无法正确聚合与展示对应数据： 表3关键维度（Attributes） | 维度 | 说明 | 适用指标 | |:---|:---|:---| | gen_ai.resource.id | 智能体 ID | 全部指标必填。平台通过此字段将Skill指标与对应的智能体关联，若缺失则该条指标数据无法归属到任何智能体，看板中该智能体下的所有Skill数据均无法展示。 | | gen_ai.skill.name | Skill 名称 | 全部指标必填。平台通过此字段对Skill进行去重统计（Skill数）和分组聚合（调用次数排行、耗时排行、成功率排行），若缺失则平台无法区分不同Skill，所有排行榜类看板均无数据。 | | domain.id | 账号 ID | 全部指标必填。用于租户之间的数据安全隔离，平台查询时会以此字段过滤出当前账号的数据，若填写错误或缺失，数据会被平台过滤掉，看板查询不到任何结果。 | | gen_ai.call.status | 调用是否成功（True/False） | call.count、duration、error.count 必填。平台通过此维度区分成功与失败的调用，用于计算Skill调用成功率（成功率 = status 为 True 的 call.count / 总 call.count）及成功率排行；duration携带此字段可支持按成功/失败分别分析耗时分布；若缺失，平台无法计算成功率，成功率相关看板将无数据。 | | gen_ai.request.model | 模型名称 | token.usage必填，其余选填。上报gen_ai.skill.token.usage时携带，用于支撑按模型维度分析Skill带来的Token成本，例如对比不同模型下同一Skill的Token消耗差异；其余指标携带此字段可支持按模型过滤，但非强制要求。 | #### 前置准备：在AgentArts控制台创建接入智能体 1. 登录[AgentArts智能体平台](https://console.huaweicloud.com/agentarts/#/home/overview)。 2. 在左侧导航栏中选择"运营运维 \> 观测"，在"智能体列表"页签，单击"智能体接入"。 图1智能体接入 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549195.png "点击放大") 3. 填写智能体名称，智能体类型选择"单智能体"。 4. 创建完成后，单击"确定"，在页面接入指南中获取以下信息。 表4接入信息说明 | 参数 | 说明 | |:---|:---| | agent_id trace_endpoint trace_token | 智能体 ID。 Trace 数据接入地址。 Trace 上报鉴权 Token。 图2获取接入地址、鉴权信息、智能体ID等信息 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549207.png "点击放大") | | metric_endpoint metric_token project_id promID | Metric 数据接入地址。 Metric 上报鉴权 Token。 华为云项目 ID。 AOM Prometheus 实例 ID。 图3获取接入地址、鉴权信息、项目ID等信息 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549199.png "点击放大") | 同时在控制台"[我的凭证 \> API凭证](https://console.huaweicloud.com/iam/#/mine/apiCredential)"页面获取： - domain_id（账号ID） - user_id（IAM用户ID） 图4获取domain id、user id ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389315.png "点击放大") #### 准备Skill文件 示例使用一个名为skill-creator的Skill作为演示对象，可以使用该Skill创建新的技能。脚本在运行时会检测该文件是否存在、读取其内容并将其作为上下文注入最终的LLM调用中。整个Skill的加载、执行、释放过程，正是脚本所要观测和上报的对象。因此在运行脚本之前，需要在本地创建好该Skill文件，否则脚本无法触发Skill调用路径，也就无法产生Skill相关的Trace和Metric数据。 准备Skill文件，目录结构如下，创建skill-creator文件夹，并在文件夹中创建SKILL.md文件。 ``` skill-creator/ SKILL.md ``` SKILL.md 示例内容如下： ``` --- name: skill-creator description: 创建新 Skill 或改进已有 Skill。当用户想创建一个可复用的专项能力、 固化某种工作流程、或希望智能体具备某项专项能力时使用。 --- # Skill Creator 根据用户需求，直接输出完整的 Skill 内容（SKILL.md 文本）供用户复制使用。 > 注意：直接在回复中输出 Skill 文本，不要尝试执行命令或写入文件。 ## 工作方式 1. **理解需求**：明确 Skill 要解决什么问题、适用场景和输出格式。信息足够则直接开始写，不要过度追问。 2. **输出 Skill**：将完整 SKILL.md 放在代码块中输出，用户直接复制保存为 `/SKILL.md`。 3. **按需迭代**：根据用户反馈修改，每次输出完整文件，方便用户直接替换。 ## Skill 结构模板 ```markdown --- name: skill 名称（英文小写，连字符分隔） description: 说明做什么、何时触发，描述应涵盖用户可能使用的不同表述方式。 --- # Skill 标题 ## 技能说明 ## 工作流程 ## 输出格式 ``` ## 写作原则 - 说明原因而非只下指令：模型理解背后逻辑后，在边界情况下也能正确处理 - 示例胜过规则：给出 1～2 个典型输入输出，比文字规则更有效 - 保持简洁：去掉不影响输出质量的内容 ``` #### 创建Skill数据上报脚本 安装运行依赖，Python建议3.9以上版本。 ``` python -V pip install opentelemetry-api==1.41.1 \ opentelemetry-sdk==1.41.1 \ opentelemetry-exporter-otlp==1.41.1 \ openai==1.14.0 \ python-dotenv==1.0.0 ``` 与skill-creator文件夹同目录文件下，创建环境变量文件以及skill脚本。 ``` skill-creator/ SKILL.md skill.py .env ``` ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389331.png) 创建.env文件，环境变量值请按照实际进行替换。 ``` # ===== 华为云 MaaS 模型凭证 ===== MODEL_NAME=deepseek-v4-flash MODEL_URL=https://api.modelarts-maas.com/openai/v1 MODEL_API_KEY=替换为你在MaaS申请的API Key # ===== AgentArts 接入凭证 ===== AGENT_ID=替换为agent_id AGENT_NAME=替换为智能体名称 # ===== Trace 上报凭证 ===== TRACE_ENDPOINT=替换为trace_endpoint TRACE_TOKEN=替换为trace_token # ===== Metric 上报凭证 ===== METRIC_ENDPOINT=替换为metric_endpoint METRIC_TOKEN=替换为metric_token PROJECT_ID=替换为project_id PROM_ID=替换为promID # ===== 身份字段（必须填写真实值）===== DOMAIN_ID=替换为华为云账号ID USER_ID=替换为IAM用户ID ``` 华为云MaaS服务的模型API Key，请登录[MaaS服务](https://console.huaweicloud.com/modelarts/#/model-studio/homepage)获取。 图5模型API Key ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389323.png "点击放大") 创建skill.py文件，下面是本案例的完整实现脚本。启动时从.env文件加载接入凭证，同步初始化Trace和 Metric两路上报通道，并开启本地日志输出（otel_debug.log）用于调试；对话过程中，每次调用依次经历LLM决策、Skill加载、Skill释放、LLM最终回答四个环节，各节点完成后即时上报对应Metric，Root Span在关闭前汇总所有Model Span的Token消耗并统一写入。 ``` import os import time import uuid import json import subprocess from openai import OpenAI from dotenv import load_dotenv from opentelemetry import trace, metrics from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, SimpleSpanProcessor, ConsoleSpanExporter, ) from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import ( PeriodicExportingMetricReader, ConsoleMetricExporter, ) from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter load_dotenv() # ============================================================ # 1. 凭证读取 # ============================================================ AGENT_ID = os.getenv("AGENT_ID", "") AGENT_NAME = os.getenv("AGENT_NAME", "") DOMAIN_ID = os.getenv("DOMAIN_ID", "") USER_ID = os.getenv("USER_ID", "") PROJECT_ID = os.getenv("PROJECT_ID", "") PROM_ID = os.getenv("PROM_ID", "") MODEL_NAME = os.getenv("MODEL_NAME", "") # ============================================================ # 2. OTel 初始化 # ============================================================ log_file = open("otel_debug.log", "a", encoding="utf-8") # Trace 环境变量 os.environ["OTEL_SERVICE_NAME"] = f"AgentArts.{AGENT_ID}.default" os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = ( f"Authentication={os.getenv('TRACE_TOKEN')}" ) os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"] = os.getenv("TRACE_ENDPOINT", "") os.environ["OTEL_EXPORTER_OTLP_TRACES_INSECURE"] = "true" # Metric 环境变量 # Bearer 前缀处理：去重后统一补上，避免 "Bearer Bearer xxx" 问题 metric_token = os.getenv("METRIC_TOKEN", "").strip() metric_token = metric_token.removeprefix("Bearer ").strip() metric_token = f"Bearer {metric_token}" os.environ["OTEL_EXPORTER_OTLP_METRICS_HEADERS"] = ( f"Authorization={metric_token}," f"projectID={PROJECT_ID}," f"promID={PROM_ID}," f"Content-Type=application/x-protobuf" ) os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = os.getenv("METRIC_ENDPOINT", "") os.environ["OTEL_EXPORTER_OTLP_METRICS_INSECURE"] = "true" # Trace Provider（双路：OTLP 上报 + 本地日志） tracer_provider = TracerProvider() tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) tracer_provider.add_span_processor( SimpleSpanProcessor(ConsoleSpanExporter(out=log_file)) ) trace.set_tracer_provider(tracer_provider) tracer = trace.get_tracer(__name__) # Metric Provider（双路：OTLP 上报 + 本地日志） meter_provider = MeterProvider(metric_readers=[ PeriodicExportingMetricReader(OTLPMetricExporter(), export_interval_millis=60000), PeriodicExportingMetricReader(ConsoleMetricExporter(out=log_file), export_interval_millis=15000), ]) metrics.set_meter_provider(meter_provider) meter = metrics.get_meter(__name__) # ============================================================ # 3. 指标注册（全局单例，不可在循环内重复注册） # ============================================================ # 应用级指标：支撑用户数、会话数、应用调用次数、QPS/QPM、响应成功率、链路平均耗时 usage_status_counter = meter.create_up_down_counter( "gen_ai.usage.status", unit="1", description="用户数/会话数统计") app_request_counter = meter.create_counter( "requests_total", unit="1", description="应用调用总次数/响应成功率/QPS") app_duration_histogram = meter.create_histogram( "client.operation.duration", unit="s", description="应用端到端调用耗时/链路平均耗时") # 标准大模型指标 request_counter = meter.create_counter( "gen_ai.total.requests", unit="1", description="大模型调用次数") input_tokens_counter = meter.create_counter( "gen_ai.usage.input_tokens", unit="1", description="输入Token消耗") output_tokens_counter = meter.create_counter( "gen_ai.usage.output_tokens", unit="1", description="输出Token消耗") # Skill 专项指标 skill_call_counter = meter.create_counter( "gen_ai.skill.call.count", unit="1", description="Skill调用次数") skill_duration_histogram = meter.create_histogram( "gen_ai.skill.duration", unit="s", description="Skill调用耗时") skill_error_counter = meter.create_counter( "gen_ai.skill.error.count", unit="1", description="Skill错误次数") skill_token_counter = meter.create_counter( "gen_ai.skill.token.usage", unit="1", description="Skill内容Token消耗") # ============================================================ # 4. 智能体核心逻辑 # ============================================================ def chat_with_agent( user_query: str, session_id: str, conversation_history: list, ) -> str: client = OpenAI( api_key=os.getenv("MODEL_API_KEY"), base_url=os.getenv("MODEL_URL"), ) # 公共 Metric 维度（同时携带多种形式确保各看板均能命中） base_metric_attrs = { "gen_ai.resource.id": AGENT_ID, "gen_ai.resource.type": "agent", "gen_ai.model.id": MODEL_NAME, "gen_ai.session.id": session_id, "gen_ai.conversation.id": session_id, "gen_ai.domain.id": DOMAIN_ID, "domain.id": DOMAIN_ID, "gen_ai.project.id": PROJECT_ID, "gen_ai.user.id": USER_ID, "user.id": USER_ID, "gen_ai.space.id": "default", } # LLM 调用 Metric 维度（提前定义，供两次 LLM 调用共用） llm_attrs = {**base_metric_attrs, "gen_ai.call.status": True} # Token 累加器：汇总本次请求中所有 Model Span 的 Token 消耗 # Root Span 上的 Token 字段需为所有 Model Span 消耗的总和 total_input_tokens = 0 total_output_tokens = 0 total_tokens = 0 # 记录应用级整体调用开始时间，用于上报 client.operation.duration app_start_time = time.time() # 上报 gen_ai.usage.status：标记本次调用的用户与会话活跃状态 # 用于支撑看板"用户数"、"会话数"、"在线应用数"统计 usage_attrs = { "gen_ai.resource.id": AGENT_ID, "gen_ai.resource.type": "agent", "domain.id": DOMAIN_ID, "user.id": USER_ID, "gen_ai.conversation.id": session_id, } usage_status_counter.add(1, attributes=usage_attrs) with tracer.start_as_current_span("agent_dialogue_session") as root_span: root_span.set_attribute("gen_ai.span.type", "root") root_span.set_attribute("gen_ai.resource.id", AGENT_ID) root_span.set_attribute("gen_ai.resource.type", "agent") root_span.set_attribute("gen_ai.agent.name", AGENT_NAME) root_span.set_attribute("gen_ai.call.type", "API") root_span.set_attribute("gen_ai.application_name", "default") root_span.set_attribute("gen_ai.environment", "default") root_span.set_attribute("resource_version", "default") root_span.set_attribute("domain.id", DOMAIN_ID) root_span.set_attribute("user.id", USER_ID) root_span.set_attribute("gen_ai.conversation.id", session_id) root_span.set_attribute("input.value", user_query) tools = [{ "type": "function", "function": { "name": "skill-creator", "description": "用于自动生成、辅助编写新 Skill 技能代码的工具。", "parameters": { "type": "object", "properties": {"prompt": {"type": "string"}}, "required": ["prompt"], }, }, }] # 本轮消息 = 历史消息 + 当前用户输入 current_messages = conversation_history + [ {"role": "user", "content": user_query} ] # ---------- 节点1：LLM 决策调用 ---------- with tracer.start_as_current_span("llm_decision_call") as model_span: model_span.set_attribute("gen_ai.span.type", "model") model_span.set_attribute("gen_ai.resource.id", AGENT_ID) model_span.set_attribute("gen_ai.resource.type", "agent") model_span.set_attribute("gen_ai.agent.name", AGENT_NAME) model_span.set_attribute("domain.id", DOMAIN_ID) model_span.set_attribute("user.id", USER_ID) model_span.set_attribute("gen_ai.conversation.id", session_id) model_span.set_attribute("traceloop.workflow.type", "LLM") model_span.set_attribute("gen_ai.request.model", MODEL_NAME) model_span.set_attribute("input.value", user_query) start_time = time.time() response = client.chat.completions.create( model=MODEL_NAME, messages=current_messages, tools=tools, ) duration = time.time() - start_time message = response.choices[0].message usage = response.usage model_span.set_attribute("gen_ai.usage.input_tokens", usage.prompt_tokens) model_span.set_attribute("gen_ai.usage.output_tokens", usage.completion_tokens) model_span.set_attribute("gen_ai.usage.total_tokens", usage.total_tokens) model_span.set_attribute("gen_ai.client.operation.duration", str(duration)) model_span.set_attribute( "output.value", "准备调用工具" if message.tool_calls else (message.content or "") ) request_counter.add(1, attributes=llm_attrs) input_tokens_counter.add(usage.prompt_tokens, attributes=llm_attrs) output_tokens_counter.add(usage.completion_tokens, attributes=llm_attrs) # 将本次 Model Span 的 Token 消耗累加至总计 total_input_tokens += usage.prompt_tokens total_output_tokens += usage.completion_tokens total_tokens += usage.total_tokens if message.tool_calls: tool_call = message.tool_calls[0] skill_name = tool_call.function.name skill_args = json.loads(tool_call.function.arguments) prompt_text = skill_args.get("prompt", "") skill_path = f"{skill_name}/SKILL.md" skill_scripts = [ f"{skill_name}/scripts/main.py", f"{skill_name}/main.py", ] script_path = next( (p for p in skill_scripts if os.path.exists(p)), None ) # ---------- 节点2：skill_tool（加载 Skill）---------- # skill_activate_time 在 skill_tool 之前开始计时， # 确保 skill_duration 覆盖从加载到释放的完整生命周期 skill_activate_time = time.time() call_status = True skill_result_text = "" skill_body = "" with tracer.start_as_current_span("skill_tool") as skill_span: skill_span.set_attribute("gen_ai.span.type", "model") skill_span.set_attribute("gen_ai.resource.id", AGENT_ID) skill_span.set_attribute("gen_ai.resource.type", "agent") skill_span.set_attribute("gen_ai.agent.name", AGENT_NAME) skill_span.set_attribute("domain.id", DOMAIN_ID) skill_span.set_attribute("user.id", USER_ID) skill_span.set_attribute("gen_ai.conversation.id", session_id) skill_span.set_attribute("gen_ai.operation.name", "load_skill") skill_span.set_attribute("gen_ai.skill.name", skill_name) skill_span.set_attribute( "gen_ai.skill.id", f"skill_{abs(hash(skill_name)) % (10 ** 8)}" ) skill_span.set_attribute("input.value", f"加载技能参数: {prompt_text}") if os.path.exists(skill_path): skill_span.add_event("skill.loaded", { "skill.name": skill_name, "skill.path": skill_path, }) try: if script_path: print(f"\n[Skill执行中] 检测到代码型技能，正在运行 {script_path}...") result = subprocess.run( ["python", script_path, prompt_text], capture_output=True, text=True, check=True, encoding="utf-8", ) skill_result_text = result.stdout elif os.path.exists(skill_path): print(f"\n[Skill执行中] 检测到提示词型技能，正在加载 {skill_path}...") with open(skill_path, "r", encoding="utf-8") as f: skill_body = f.read() skill_result_text = ( "【已成功加载该技能规范，请严格遵循以下内容处理用户请求】:\n\n" + skill_body ) else: skill_result_text = f"Skill 执行失败：未找到 {skill_path} 或执行脚本。" call_status = False except Exception as e: skill_result_text = f"Skill 执行时发生故障: {e}" call_status = False skill_span.set_attribute("output.value", skill_result_text[:500]) # ---- Skill call.count 在 skill_tool 结束后上报 ---- # Skill数、Skill调用次数、Skill调用成功率及各排行 skill_attrs = { **base_metric_attrs, "gen_ai.skill.name": skill_name, "gen_ai.skill.version": "1.0.0", "gen_ai.system": "python-sdk", "gen_ai.call.status": call_status, } skill_call_counter.add(1, attributes=skill_attrs) if not call_status: skill_error_counter.add(1, attributes=skill_attrs) # Skill 内容 Token 估算 skill_content = skill_body if skill_body else skill_result_text skill_content_tokens = max(1, len(skill_content.encode("utf-8")) // 4) skill_token_counter.add(skill_content_tokens, attributes={ **skill_attrs, "gen_ai.request.model": MODEL_NAME, }) # ---------- 节点3：skill_complete（释放 Skill）---------- with tracer.start_as_current_span("skill_complete") as release_span: release_span.set_attribute("gen_ai.span.type", "model") release_span.set_attribute("gen_ai.resource.id", AGENT_ID) release_span.set_attribute("gen_ai.resource.type", "agent") release_span.set_attribute("gen_ai.agent.name", AGENT_NAME) release_span.set_attribute("domain.id", DOMAIN_ID) release_span.set_attribute("user.id", USER_ID) release_span.set_attribute("gen_ai.conversation.id", session_id) release_span.set_attribute("gen_ai.operation.name", "release_skill") release_span.set_attribute("gen_ai.skill.name", skill_name) release_span.set_attribute( "gen_ai.skill.id", f"skill_{abs(hash(skill_name)) % (10 ** 8)}" ) release_span.set_attribute("input.value", "释放技能") release_span.set_attribute("output.value", "释放成功") release_span.add_event("skill.released", {"skill.name": skill_name}) # ---- skill_duration 在 skill_complete 结束后计算并上报 ---- # 覆盖从 skill_tool 开始到 skill_complete 结束的完整生命周期 # 支撑：Skill调用平均耗时、Skill调用平均耗时排行Top5 skill_duration = time.time() - skill_activate_time skill_duration_histogram.record(skill_duration, attributes=skill_attrs) print( f"[{time.strftime('%H:%M:%S')}] " f"Skill={skill_name} | ok={call_status} | " f"duration={skill_duration:.2f}s | " f"skill_tokens≈{skill_content_tokens}" ) # ---------- 节点4：LLM 最终回答 ---------- with tracer.start_as_current_span("llm_final_answer") as final_span: final_span.set_attribute("gen_ai.span.type", "model") final_span.set_attribute("gen_ai.resource.id", AGENT_ID) final_span.set_attribute("gen_ai.resource.type", "agent") final_span.set_attribute("gen_ai.agent.name", AGENT_NAME) final_span.set_attribute("domain.id", DOMAIN_ID) final_span.set_attribute("user.id", USER_ID) final_span.set_attribute("gen_ai.conversation.id", session_id) final_span.set_attribute("traceloop.workflow.type", "LLM") final_span.set_attribute("gen_ai.request.model", MODEL_NAME) final_span.set_attribute( "input.value", f"工具反馈结果长度: {len(skill_result_text)} 字符" ) assistant_message = message.model_dump(exclude_none=True) if not assistant_message.get("content"): assistant_message["content"] = "" # 最终回答消息 = 历史 + 本轮用户问题 + 助手决策 + 工具结果 final_messages = conversation_history + [ {"role": "user", "content": user_query}, assistant_message, { "role": "tool", "tool_call_id": tool_call.id, "content": skill_result_text, }, ] start_time = time.time() final_response = client.chat.completions.create( model=MODEL_NAME, messages=final_messages, ) duration = time.time() - start_time usage = final_response.usage final_answer = final_response.choices[0].message.content final_span.set_attribute("gen_ai.usage.input_tokens", usage.prompt_tokens) final_span.set_attribute("gen_ai.usage.output_tokens", usage.completion_tokens) final_span.set_attribute("gen_ai.usage.total_tokens", usage.total_tokens) final_span.set_attribute("gen_ai.client.operation.duration", str(duration)) final_span.set_attribute("output.value", final_answer) request_counter.add(1, attributes=llm_attrs) input_tokens_counter.add(usage.prompt_tokens, attributes=llm_attrs) output_tokens_counter.add(usage.completion_tokens, attributes=llm_attrs) # 将本次 Model Span 的 Token 消耗累加至总计 total_input_tokens += usage.prompt_tokens total_output_tokens += usage.completion_tokens total_tokens += usage.total_tokens # Root Span 关闭前统一写入：所有 Model Span Token 消耗的累加总和 # 本次请求包含两次 LLM 调用（决策 + 最终回答），Token 为两次之和 root_span.set_attribute("output.value", final_answer) root_span.set_attribute("gen_ai.usage.input_tokens", total_input_tokens) root_span.set_attribute("gen_ai.usage.output_tokens", total_output_tokens) root_span.set_attribute("gen_ai.usage.total_tokens", total_tokens) # 上报应用级指标：计算本次调用整体耗时 # 应用调用次数、QPS/QPM、响应成功率、链路整体平均耗时 app_total_duration = time.time() - app_start_time app_attrs = { "gen_ai.resource.id": AGENT_ID, "gen_ai.resource.type": "agent", "domain.id": DOMAIN_ID, "gen_ai.conversation.id": session_id, "gen_ai.call.status": True, } app_request_counter.add(1, attributes=app_attrs) app_duration_histogram.record(app_total_duration, attributes=app_attrs) # 将本轮完整对话追加至历史，供下一轮携带 conversation_history.append({"role": "user", "content": user_query}) conversation_history.append(assistant_message) conversation_history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": skill_result_text, }) conversation_history.append({ "role": "assistant", "content": final_answer, }) return final_answer else: # 非工具调用分支：模型直接回答，无 Skill 介入 final_answer = message.content or "" # Root Span 关闭前写入 Token 字段 # 此分支只有一次 LLM 调用，直接取该次的 Token 值 root_span.set_attribute("output.value", final_answer) root_span.set_attribute("gen_ai.usage.input_tokens", total_input_tokens) root_span.set_attribute("gen_ai.usage.output_tokens", total_output_tokens) root_span.set_attribute("gen_ai.usage.total_tokens", total_tokens) # 非工具调用分支同样上报应用级指标 app_total_duration = time.time() - app_start_time app_attrs = { "gen_ai.resource.id": AGENT_ID, "gen_ai.resource.type": "agent", "domain.id": DOMAIN_ID, "gen_ai.conversation.id": session_id, "gen_ai.call.status": True, } app_request_counter.add(1, attributes=app_attrs) app_duration_histogram.record(app_total_duration, attributes=app_attrs) conversation_history.append({"role": "user", "content": user_query}) conversation_history.append({"role": "assistant", "content": final_answer}) return final_answer # ============================================================ # 5. 启动入口 # ============================================================ if __name__ == "__main__": session_id = f"sess-{uuid.uuid4().hex[:12]}" conversation_history = [] print("==================================================") print(" AgentArts Skill Telemetry Verify 已启动") print(f" agent_id = {AGENT_ID}") print(f" agent_name = {AGENT_NAME}") print(f" session_id = {session_id}") print(" 本地调试：OTel 数据同步写入 otel_debug.log") print(" 指标每 60s 批量上报，请对话 3 分钟以上再查看看板") print("==================================================") try: while True: user_input = input("\nUser: ").strip() if not user_input: continue if user_input.lower() in ("exit", "quit", "q"): break reply = chat_with_agent(user_input, session_id, conversation_history) print(f"\nAgent: {reply}") except KeyboardInterrupt: print("\n检测到 Ctrl+C，正在准备退出...") finally: print("\n正在强制刷新残留数据至 AgentArts 平台...") tracer_provider.shutdown() meter_provider.shutdown() log_file.close() print("数据上报完成！") ``` #### 测试Skill观测数据上报 1. 在脚本同目录下，执行python skill.py命令，并与智能体进行对话。 示例中创建的skill-creator是一个skill生成器，因此对话过程汇总可以让智能体生成各式各样的skill。 图6运行示例 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389319.png "点击放大") 2. 与智能体进行多次会话后，登录[AgentArts智能体平台](https://console.huaweicloud.com/agentarts/#/home/overview)，在"运营运维 \> 观测 \> 智能体列表"页面，单击对应的智能体名称查看上报的数据。 图7Skill数据 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949566.png "点击放大") 图8调用链数据 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949574.png "点击放大") #### 常见问题 - **UNAUTHENTICATED：鉴权失败报错** 检查METRIC_TOKEN是否正确；脚本会自动处理Bearer前缀，不需要手动添加；检查projectID/promID大小写是否与接入指南完全一致。 - **看板有数据但没有Skill指标** 确认Skill相关维度（gen_ai.resource.id、gen_ai.skill.name、domain.id）均已正确携带；指标每60秒批量上报一次，建议持续对话3分钟以上再查看。 - **LLM没有触发Skill，控制台只有LLM指标** 模型并非每次都会调用工具，建议在提问中明确说明需要创建或使用Skill，引导模型触发工具调用。 Invalid type NoneType for attribute报错 .env文件中缺少对应的环境变量配置，导致读取结果为None。常见缺失项为AGENT_NAME，请检查.env文件中所有必填参数是否均已填写真实值。 - **UNAVAILABLE：连不上服务器报错** 检查METRIC_ENDPOINT/TRACE_ENDPOINT格式，确保与接入指南给出的格式保持一致（纯域名，不带https://）。 - **会话分析或调用链分析中Tokens列显示为0** Token相关字段（gen_ai.usage.input_tokens、gen_ai.usage.output_tokens、gen_ai.usage.total_tokens）仅在Model Span上设置，未在Root Span上同步设置。平台统计会话/Trace维度的Token消耗时读取的是Root Span上的字段，若缺失则显示为0。正确做法是将本次请求中所有Model Span的Token消耗累加后，在Root Span关闭前统一写入。 - **看板中Skill调用次数显示为小数** AOM底层使用Prometheus的increase()函数计算Counter类指标在时间窗口内的增长量，当查询时间窗口的边界与指标上报时间点不完全对齐时，会对首尾两端进行线性插值外推，从而产生小数。实际累计调用次数仍为整数，小数是速率估算的结果。若需查看精确整数，可将查询时间窗口对齐为60秒的整数倍（与脚本上报周期一致）。 - **domain.id或user.id填写了假数据，看板查询不到任何数据** 平台存在租户隔离机制，查询时会以当前登录账号的真实domain.id进行过滤。若上报的指标中domain.id与实际账号ID不一致，数据会被直接过滤，看板无法展示任何结果。请登录"[我的凭证 \> API凭证](https://console.huaweicloud.com/iam/#/mine/apiCredential)"页面获取真实的domain_id和user_id并填入.env文件。