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 = <skill名称>
│   ├── gen_ai.skill.id = <skill标识>
│   └── Event: skill.loaded               ← 加载成功事件，记录路径
├── Tool Span（skill_complete）            ← Skill 释放
│   ├── gen_ai.operation.name = release_skill
│   ├── gen_ai.skill.name = <skill名称>
│   └── Event: skill.released             ← 释放完成事件
└── Model Span（llm_final_answer）         ← 携带 Skill 内容的最终 LLM 调用

上报机制

两路数据均通过OpenTelemetry Python SDK上报，Trace使用BatchSpanProcessor批量发送，Metric使用PeriodicExportingMetricReader定期批量发送。

本地调试阶段同时启用ConsoleSpanExporter和ConsoleMetricExporter，将数据同步写入本地otel_debug.log，用于在看板出数据前确认本地采集是否正常。

指标上报逻辑

上报上述指标时，必须在 Attributes 中携带以下维度字段，否则看板无法正确聚合与展示对应数据：

前置准备：在AgentArts控制台创建接入智能体

1. 登录AgentArts智能体平台。
2. 在左侧导航栏中选择“运营运维 > 观测”，在“智能体列表”页签，单击“智能体接入”。
   图1 智能体接入
   

3. 填写智能体名称，智能体类型选择“单智能体”。
4. 创建完成后，单击“确定”，在页面接入指南中获取以下信息。

   表4 接入信息说明

   参数	说明
   agent_id trace_endpoint trace_token	智能体 ID。 Trace 数据接入地址。 Trace 上报鉴权 Token。 图2 获取接入地址、鉴权信息、智能体ID等信息
   metric_endpoint metric_token project_id promID	Metric 数据接入地址。 Metric 上报鉴权 Token。 华为云项目 ID。 AOM Prometheus 实例 ID。 图3 获取接入地址、鉴权信息、项目ID等信息

   同时在控制台“我的凭证 > API凭证”页面获取：

   • domain_id（账号ID）
   • user_id（IAM用户ID）
   图4 获取domain id、user id
   

准备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-name>/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

创建.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服务获取。

创建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 运行示例
   

2. 与智能体进行多次会话后，登录AgentArts智能体平台，在“运营运维 > 观测 > 智能体列表”页面，单击对应的智能体名称查看上报的数据。
   图7 Skill数据
   
   图8 调用链数据
   

常见问题

• 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凭证”页面获取真实的domain_id和user_id并填入.env文件。