diff --git a/tests/test_tracing_content.py b/tests/test_tracing_content.py index 367ac6eb3..e390087b4 100644 --- a/tests/test_tracing_content.py +++ b/tests/test_tracing_content.py @@ -21,6 +21,7 @@ from veadk.tracing.telemetry import telemetry from veadk.tracing.telemetry.content_tracing import should_trace_content from veadk.tracing.telemetry.exporters.apmplus_exporter import MeterUploader +from veadk.tracing.telemetry.skill_observability import ActiveSkill, set_active_skill @dataclass @@ -130,6 +131,9 @@ def __init__(self): def record(self, value, attributes=None): self.records.append((value, attributes)) + def add(self, value, attributes=None): + self.records.append((value, attributes)) + def _start_test_span(name: str): provider = trace_sdk.TracerProvider() @@ -230,6 +234,56 @@ def test_apmplus_tool_metrics_skip_token_usage_when_tool_content_missing(): assert meter_uploader.apmplus_tool_token_usage.records == [] +def test_apmplus_llm_metrics_attribute_actual_tokens_to_active_skill(): + meter_uploader = object.__new__(MeterUploader) + meter_uploader.llm_invoke_counter = _FakeMetricRecorder() + meter_uploader.token_usage = _FakeMetricRecorder() + meter_uploader.skill_token_usage = _FakeMetricRecorder() + meter_uploader.duration_histogram = _FakeMetricRecorder() + meter_uploader.chat_exception_counter = _FakeMetricRecorder() + meter_uploader.apmplus_span_latency = _FakeMetricRecorder() + set_active_skill(ActiveSkill(name="pdf", invocation_id="invocation")) + + with _start_test_span("call_llm"): + meter_uploader.record_call_llm( + _FakeInvocationContext(), + "event-id", + _FakeLlmRequest(), + _FakeLlmResponse(), + ) + + assert [value for value, _ in meter_uploader.skill_token_usage.records] == [ + 11, + 7, + ] + assert all( + attributes["skill_name"] == "pdf" + for _, attributes in meter_uploader.skill_token_usage.records + ) + + +def test_skill_operation_metrics_record_count_error_and_both_durations(): + meter_uploader = object.__new__(MeterUploader) + meter_uploader.skill_invoke_counter = _FakeMetricRecorder() + meter_uploader.skill_error_counter = _FakeMetricRecorder() + meter_uploader.skill_invoke_latency = _FakeMetricRecorder() + meter_uploader.skill_duration_histogram = _FakeMetricRecorder() + + with _start_test_span("skill.run_script") as span: + meter_uploader.record_skill_operation( + span=span, + operation="run_script", + attributes={"skill_name": "pdf"}, + success=False, + error_type="skill_execution_error", + ) + + assert len(meter_uploader.skill_invoke_counter.records) == 1 + assert len(meter_uploader.skill_error_counter.records) == 1 + assert len(meter_uploader.skill_invoke_latency.records) == 1 + assert len(meter_uploader.skill_duration_histogram.records) == 1 + + def test_agent_root_span_skips_content_when_env_false(monkeypatch): monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_TRACE_CONTENT", "false") diff --git a/tests/tools/builtin_tools/test_run_sandbox_agent.py b/tests/tools/builtin_tools/test_run_sandbox_agent.py index 59391a023..a61cbbcb5 100644 --- a/tests/tools/builtin_tools/test_run_sandbox_agent.py +++ b/tests/tools/builtin_tools/test_run_sandbox_agent.py @@ -170,7 +170,12 @@ def test_custom_values_override_defaults_and_preserve_empty_values(self): ) def test_rejects_framework_managed_values(self): - for key in ["TOOL_USER_SESSION_ID", "USER_SESSION_ID"]: + for key in [ + "TOOL_USER_SESSION_ID", + "USER_SESSION_ID", + "TRACEPARENT", + "TRACESTATE", + ]: with self.subTest(key=key): with self.assertRaisesRegex(ValueError, "managed by VeADK"): self.module._merge_execution_env_vars({}, {key: "spoofed"}) @@ -199,6 +204,27 @@ def test_runner_code_overrides_the_sandbox_process_environment(self): self.assertNotIn("if key not in env", code) self.assertIn('srv_pythonpath = env.get("SRV_PYTHONPATH")', code) + def test_current_trace_context_is_mapped_to_protected_environment(self): + def inject(carrier): + carrier.update( + { + "traceparent": "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + "tracestate": "vendor=value", + "baggage": "secret=must-not-cross-the-boundary", + } + ) + + with patch.object(self.module.propagate, "inject", side_effect=inject): + result = self.module._current_trace_env_vars() + + self.assertEqual( + result, + { + "TRACEPARENT": "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + "TRACESTATE": "vendor=value", + }, + ) + class TestExecuteSkillsSkillApi(unittest.TestCase): def _tool_context(self): diff --git a/tests/tracing/test_skill_observability.py b/tests/tracing/test_skill_observability.py new file mode 100644 index 000000000..61945ccf8 --- /dev/null +++ b/tests/tracing/test_skill_observability.py @@ -0,0 +1,89 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from types import SimpleNamespace +from unittest.mock import patch + +from veadk.tracing.telemetry import skill_observability + + +class FakeSpan: + def __init__(self, invocation_id: str = "invocation-1") -> None: + self.attributes = {"invocation.id": invocation_id} + self.events = [] + + def set_attribute(self, name, value) -> None: + self.attributes[name] = value + + def add_event(self, name, attributes=None) -> None: + self.events.append((name, attributes or {})) + + +def test_google_skill_load_activates_skill_and_records_semantics(): + span = FakeSpan() + tool = SimpleNamespace(name="load_skill") + response = SimpleNamespace(response={"status": "success"}) + + with ( + patch.object( + skill_observability, + "get_event_function_responses", + return_value=[response], + ), + patch.object(skill_observability, "_record_skill_metrics") as record, + ): + skill_observability.observe_skill_tool_call( + span, tool, {"skill_name": "pdf"}, object() + ) + + assert span.attributes["skill.name"] == "pdf" + assert span.attributes["skill.operation"] == "load" + assert span.attributes["skill.phase"] == "completed" + assert span.events[0][0] == "skill.selected" + record.assert_called_once() + + +def test_active_skill_does_not_leak_into_another_invocation(): + skill_observability.set_active_skill( + skill_observability.ActiveSkill(name="pdf", invocation_id="invocation-previous") + ) + + assert skill_observability.active_skill_span_attributes("invocation-next") == {} + assert skill_observability.active_skill_metric_attributes("invocation-next") == {} + + +def test_skill_script_failure_is_annotated_without_replacing_active_skill(): + skill_observability.set_active_skill( + skill_observability.ActiveSkill(name="pdf", invocation_id="invocation-1") + ) + span = FakeSpan() + tool = SimpleNamespace(name="run_skill_script") + response = SimpleNamespace(response={"error": "script failed"}) + + with ( + patch.object( + skill_observability, + "get_event_function_responses", + return_value=[response], + ), + patch.object(skill_observability, "_record_skill_metrics") as record, + ): + skill_observability.observe_skill_tool_call( + span, tool, {"skill_name": "pdf"}, object() + ) + + assert span.attributes["skill.phase"] == "failed" + assert span.attributes["error.type"] == "skill_execution_error" + assert skill_observability.get_active_skill().name == "pdf" + assert record.call_args.args[2] is False diff --git a/veadk/tools/builtin_tools/run_sandbox_agent.py b/veadk/tools/builtin_tools/run_sandbox_agent.py index bde11cadf..474ce924b 100644 --- a/veadk/tools/builtin_tools/run_sandbox_agent.py +++ b/veadk/tools/builtin_tools/run_sandbox_agent.py @@ -18,6 +18,8 @@ from typing import Optional from google.adk.tools import ToolContext +from opentelemetry import propagate, trace +from opentelemetry.trace import SpanKind from veadk.tools.builtin_tools._agentkit import invoke_agentkit_run_code from veadk.utils.logger import get_logger @@ -25,7 +27,27 @@ logger = get_logger(__name__) _ENV_VAR_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") -_PROTECTED_ENV_VARS = frozenset({"TOOL_USER_SESSION_ID", "USER_SESSION_ID"}) +_TRACE_ENV_HEADERS = { + "traceparent": "TRACEPARENT", + "tracestate": "TRACESTATE", +} +_PROTECTED_ENV_VARS = frozenset( + {"TOOL_USER_SESSION_ID", "USER_SESSION_ID", *_TRACE_ENV_HEADERS.values()} +) + +tracer = trace.get_tracer("veadk.sandbox_agent") + + +def _current_trace_env_vars() -> dict[str, str]: + """Serialize the current W3C trace context for the sandbox subprocess.""" + + carrier: dict[str, str] = {} + propagate.inject(carrier) + return { + env_name: carrier[header] + for header, env_name in _TRACE_ENV_HEADERS.items() + if carrier.get(header) + } def _merge_execution_env_vars( @@ -196,33 +218,41 @@ def run_sandbox_agent( tool_user_session_id = agent_name + "_" + user_id + "_" + session_id logger.debug(f"tool_user_session_id: {tool_user_session_id}") - base_env_vars = { - "TOOL_USER_SESSION_ID": tool_user_session_id, - } - skill_space_id = os.getenv("SKILL_SPACE_ID", "") - if skill_space_id: - base_env_vars["SKILL_SPACE_ID"] = skill_space_id - env_vars = _merge_execution_env_vars(base_env_vars, extra_env_vars) - - logger.debug( - f"Run sandbox agent in session_id={session_id}, tool_id={tool_id}, timeout={timeout}, skills={skills}" - ) + with tracer.start_as_current_span( + "skill.sandbox.invoke", kind=SpanKind.CLIENT + ) as span: + span.set_attribute("sandbox.tool.id", tool_id) + span.set_attribute("sandbox.session.id", tool_user_session_id) + span.set_attribute("sandbox.invoke.mode", "invoke_tool") + + base_env_vars = { + "TOOL_USER_SESSION_ID": tool_user_session_id, + **_current_trace_env_vars(), + } + skill_space_id = os.getenv("SKILL_SPACE_ID", "") + if skill_space_id: + base_env_vars["SKILL_SPACE_ID"] = skill_space_id + env_vars = _merge_execution_env_vars(base_env_vars, extra_env_vars) + + logger.debug( + f"Run sandbox agent in session_id={session_id}, tool_id={tool_id}, timeout={timeout}, skills={skills}" + ) - cmd = _build_agent_command(workflow_prompt=workflow_prompt, skills=skills) - code = _build_agent_runner_code( - cmd=cmd, - timeout=timeout, - env_vars=env_vars, - working_dir=working_dir, - ) - res = invoke_agentkit_run_code( - tool_id=tool_id, - tool_user_session_id=tool_user_session_id, - code=code, - timeout=timeout, - kernel_name="python3", - tool_state=tool_context.state if tool_context else None, - ) + cmd = _build_agent_command(workflow_prompt=workflow_prompt, skills=skills) + code = _build_agent_runner_code( + cmd=cmd, + timeout=timeout, + env_vars=env_vars, + working_dir=working_dir, + ) + res = invoke_agentkit_run_code( + tool_id=tool_id, + tool_user_session_id=tool_user_session_id, + code=code, + timeout=timeout, + kernel_name="python3", + tool_state=tool_context.state if tool_context else None, + ) # The response can echo the submitted runner code, including custom env values. logger.debug("Invoke run sandbox agent completed") diff --git a/veadk/tools/skills_tools/skills_tool.py b/veadk/tools/skills_tools/skills_tool.py index 6b4a2e40f..bcc732085 100644 --- a/veadk/tools/skills_tools/skills_tool.py +++ b/veadk/tools/skills_tools/skills_tool.py @@ -26,6 +26,11 @@ from veadk.skills.skill import Skill from veadk.tools.skills_tools.session_path import get_session_path from veadk.tracing.telemetry.telemetry import set_common_attributes_on_tool_span +from veadk.tracing.telemetry.skill_observability import ( + ActiveSkill, + active_skill_metric_attributes, + set_active_skill, +) from veadk.utils.logger import get_logger tracer = trace.get_tracer("veadk.skills_tool") @@ -100,9 +105,24 @@ async def run_async( if not skill_name: return "Error: No skill name provided" - with tracer.start_as_current_span(f"execute_skill {skill_name}") as span: + with tracer.start_as_current_span(f"skill.load {skill_name}") as span: result = self._invoke_skill(skill_name, tool_context) self._add_skill_span_attributes(span, skill_name, result) + if not result.startswith("Error:"): + skill = self.skills.get(skill_name) + set_active_skill( + ActiveSkill( + name=skill_name, + skill_id=str(getattr(skill, "id", "") or ""), + space_id=str(getattr(skill, "skill_space_id", "") or ""), + version=str(getattr(skill, "version", "") or ""), + invocation_id=str( + (getattr(span, "attributes", None) or {}).get( + "invocation.id", "" + ) + ), + ) + ) self._upload_skill_metrics(span, skill_name, result) return result @@ -510,8 +530,13 @@ def _add_skill_span_attributes( span.set_status(Status(StatusCode.ERROR, result)) span.set_attribute("skill.name", skill_name) + span.set_attribute("skill.operation", "load") + span.set_attribute( + "skill.phase", + "failed" if result.startswith("Error:") else "completed", + ) span.set_attribute("tool.name", self.name) - span.set_attribute("gen_ai.operation.name", "execute_skill") + span.set_attribute("gen_ai.operation.name", "skill.load") span.set_attribute("gen_ai.span.kind", "tool") if skill_name in self.skills: skill = self.skills[skill_name] @@ -537,32 +562,33 @@ def _upload_skill_metrics(self, span: _Span, skill_name: str, result: str) -> No # 初始化属性，包含技能相关信息 skill = self.skills.get(skill_name) attributes = { + **active_skill_metric_attributes(), "skill_name": skill_name, "tool_name": self.name, "skill_space_id": ( skill.skill_space_id if skill and skill.skill_space_id else "" ), "skill_id": skill.id if skill and skill.id else "", - "gen_ai.operation.name": "execute_skill", - "error_type": ( - "skill_execution_error" if result.startswith("Error:") else "" - ), + "skill_operation": "load", } - - # 计算 span 执行耗时（秒） - latency_seconds = 0 - if hasattr(span, "start_time"): - # 计算耗时（秒） - latency_seconds = (time.time_ns() - span.start_time) / 1e9 # type: ignore - - # 记录技能执行延迟 - if hasattr(meter_uploader, "skill_invoke_latency"): - # 使用 skill_invoke_latency 记录技能执行延迟（秒） + failed = result.startswith("Error:") + error_type = "skill_execution_error" if failed else "" + if hasattr(meter_uploader, "record_skill_operation"): + meter_uploader.record_skill_operation( + span=span, + operation="load", + attributes=attributes, + success=not failed, + error_type=error_type, + ) + elif hasattr(meter_uploader, "skill_invoke_latency"): + latency_seconds = ( + (time.time_ns() - span.start_time) / 1e9 + if hasattr(span, "start_time") + else 0 + ) meter_uploader.skill_invoke_latency.record( latency_seconds, attributes ) - logger.debug( - f"Uploaded skill metrics for {skill_name} with latency {latency_seconds:.4f}s and attributes {attributes}" - ) except Exception as e: logger.warning(f"Failed to upload skill metrics: {e}") diff --git a/veadk/tracing/telemetry/exporters/apmplus_exporter.py b/veadk/tracing/telemetry/exporters/apmplus_exporter.py index 2b096aa4a..3e50cae0a 100644 --- a/veadk/tracing/telemetry/exporters/apmplus_exporter.py +++ b/veadk/tracing/telemetry/exporters/apmplus_exporter.py @@ -31,6 +31,7 @@ from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.trace import Span from pydantic import BaseModel, Field from typing_extensions import override @@ -153,6 +154,10 @@ class Meters: APMPLUS_TOOL_TOKEN_USAGE = "apmplus_tool_token_usage" # skill invoke latency GEN_AI_SKILL_INVOKE_LATENCY = "gen_ai_skill_invoke_latency" + GEN_AI_SKILL_INVOCATIONS = "gen_ai.skill.invocations" + GEN_AI_SKILL_ERRORS = "gen_ai.skill.errors" + GEN_AI_SKILL_OPERATION_DURATION = "gen_ai.skill.operation.duration" + GEN_AI_SKILL_TOKEN_USAGE = "gen_ai.skill.token.usage" class MeterUploader: @@ -277,6 +282,28 @@ def __init__( unit="s", explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, ) + self.skill_invoke_counter = self.meter.create_counter( + name=Meters.GEN_AI_SKILL_INVOCATIONS, + description="Number of skill operations", + unit="count", + ) + self.skill_error_counter = self.meter.create_counter( + name=Meters.GEN_AI_SKILL_ERRORS, + description="Number of failed skill operations", + unit="count", + ) + self.skill_duration_histogram = self.meter.create_histogram( + name=Meters.GEN_AI_SKILL_OPERATION_DURATION, + description="Duration of skill operations", + unit="s", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, + ) + self.skill_token_usage = self.meter.create_histogram( + name=Meters.GEN_AI_SKILL_TOKEN_USAGE, + description="Actual model token usage attributed to an active skill", + unit="count", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_TOKEN_USAGE_BUCKETS, + ) def record_call_llm( self, @@ -319,6 +346,14 @@ def record_call_llm( "server_address": server_address, } # required by Volcengine APMPlus + from veadk.tracing.telemetry.skill_observability import ( + active_skill_metric_attributes, + ) + + attributes.update( + active_skill_metric_attributes(invocation_context.invocation_id) + ) + if llm_response.usage_metadata: # llm invocation number += 1 self.llm_invoke_counter.add(1, attributes) @@ -330,9 +365,17 @@ def record_call_llm( if input_token: token_attributes = {**attributes, "gen_ai_token_type": "input"} self.token_usage.record(input_token, attributes=token_attributes) + if token_attributes.get("skill_name"): + self.skill_token_usage.record( + input_token, attributes=token_attributes + ) if output_token: token_attributes = {**attributes, "gen_ai_token_type": "output"} self.token_usage.record(output_token, attributes=token_attributes) + if token_attributes.get("skill_name"): + self.skill_token_usage.record( + output_token, attributes=token_attributes + ) # Get llm duration span = trace.get_current_span() @@ -387,6 +430,33 @@ def record_call_llm( duration = (time.time_ns() - span.start_time) / 1e9 # type: ignore self.apmplus_span_latency.record(duration, attributes=attributes) + def record_skill_operation( + self, + *, + span: Span, + operation: str, + attributes: dict[str, str], + success: bool, + error_type: str = "", + ) -> None: + """Record low-cardinality metrics for one semantic Skill operation.""" + + metric_attributes = { + **attributes, + "skill_operation": operation, + "status": "success" if success else "error", + } + if error_type: + metric_attributes["error_type"] = error_type + + self.skill_invoke_counter.add(1, metric_attributes) + if not success: + self.skill_error_counter.add(1, metric_attributes) + if hasattr(span, "start_time"): + duration = (time.time_ns() - span.start_time) / 1e9 # type: ignore + self.skill_invoke_latency.record(duration, metric_attributes) + self.skill_duration_histogram.record(duration, metric_attributes) + def record_tool_call( self, tool: BaseTool, diff --git a/veadk/tracing/telemetry/skill_observability.py b/veadk/tracing/telemetry/skill_observability.py new file mode 100644 index 000000000..6dbd9f62d --- /dev/null +++ b/veadk/tracing/telemetry/skill_observability.py @@ -0,0 +1,214 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Semantic observability helpers shared by VeADK and ADK-native skills.""" + +from __future__ import annotations + +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Any + +from opentelemetry.trace import Span + +from veadk.utils.adk_compat import get_event_function_responses + + +@dataclass(frozen=True) +class ActiveSkill: + """Low-cardinality metadata for the skill active in one invocation context.""" + + name: str + skill_id: str = "" + space_id: str = "" + version: str = "" + invocation_id: str = "" + + +_active_skill: ContextVar[ActiveSkill | None] = ContextVar( + "veadk_active_skill", default=None +) + +_SKILL_OPERATIONS = { + "skills_tool": "load", + "load_skill": "load", + "load_skill_resource": "load_resource", + "run_skill_script": "run_script", + "list_skills": "list", + "search_skills": "search", +} + + +def set_active_skill(skill: ActiveSkill) -> None: + """Set the skill used by subsequent model and tool calls in this context.""" + + _active_skill.set(skill) + + +def get_active_skill() -> ActiveSkill | None: + return _active_skill.get() + + +def active_skill_span_attributes(invocation_id: str = "") -> dict[str, str]: + skill = get_active_skill() + if skill is None or ( + invocation_id and skill.invocation_id and invocation_id != skill.invocation_id + ): + return {} + attributes = {"skill.name": skill.name} + if skill.skill_id: + attributes["skill.id"] = skill.skill_id + if skill.space_id: + attributes["skill.space.id"] = skill.space_id + if skill.version: + attributes["skill.version"] = skill.version + return attributes + + +def active_skill_metric_attributes(invocation_id: str = "") -> dict[str, str]: + skill = get_active_skill() + if skill is None or ( + invocation_id and skill.invocation_id and invocation_id != skill.invocation_id + ): + return {} + return { + "skill_name": skill.name, + "skill_id": skill.skill_id, + "skill_space_id": skill.space_id, + "skill_version": skill.version, + } + + +def set_active_skill_attributes(span: Span) -> None: + invocation_id = str( + (getattr(span, "attributes", None) or {}).get("invocation.id", "") + ) + for name, value in active_skill_span_attributes(invocation_id).items(): + span.set_attribute(name, value) + + +def observe_skill_tool_call( + span: Span, + tool: Any, + args: dict[str, Any], + function_response_event: Any, +) -> None: + """Annotate ADK tool spans with stable Skill semantics. + + Google ADK exposes SkillToolset operations as ordinary tools, while VeADK's + legacy implementation exposes ``skills_tool``. Recognizing both here keeps + the observability contract independent of the selected Skill runtime. + """ + + tool_name = getattr(tool, "name", "") + operation = _SKILL_OPERATIONS.get(tool_name) + if operation is None: + set_active_skill_attributes(span) + return + + skill = _skill_from_tool_call(tool, args) + failed, error_type = _tool_call_failed(function_response_event) + invocation_id = str( + (getattr(span, "attributes", None) or {}).get("invocation.id", "") + ) + if skill and not failed and operation in {"load", "load_resource", "run_script"}: + skill = ActiveSkill( + name=skill.name, + skill_id=skill.skill_id, + space_id=skill.space_id, + version=skill.version, + invocation_id=invocation_id, + ) + set_active_skill(skill) + + span.set_attribute("skill.operation", operation) + span.set_attribute("skill.phase", "completed" if not failed else "failed") + span.set_attribute("gen_ai.operation.name", f"skill.{operation}") + set_active_skill_attributes(span) + if error_type: + span.set_attribute("error.type", error_type) + + event_attributes = active_skill_span_attributes(invocation_id) + event_attributes["skill.operation"] = operation + event_name = { + "load": "skill.selected" if not failed else "skill.load_failed", + "run_script": "skill.completed" if not failed else "skill.failed", + "load_resource": "skill.resource_loaded" + if not failed + else "skill.resource_load_failed", + }.get(operation, f"skill.{operation}") + span.add_event(event_name, attributes=event_attributes) + + # The legacy SkillsTool records its metric inside run_async so direct uses + # that bypass ADK telemetry remain observable. Avoid double counting here. + if tool_name != "skills_tool": + _record_skill_metrics(span, operation, not failed, error_type) + + +def _skill_from_tool_call(tool: Any, args: dict[str, Any]) -> ActiveSkill | None: + name = str(args.get("skill_name") or args.get("command") or "").strip() + current = get_active_skill() + if not name: + return current + + skill_id = "" + space_id = "" + version = "" + skills = getattr(tool, "skills", None) + if isinstance(skills, dict): + skill = skills.get(name) + if skill is not None: + skill_id = str(getattr(skill, "id", "") or "") + space_id = str(getattr(skill, "skill_space_id", "") or "") + version = str(getattr(skill, "version", "") or "") + return ActiveSkill(name, skill_id=skill_id, space_id=space_id, version=version) + + +def _tool_call_failed(function_response_event: Any) -> tuple[bool, str]: + responses = get_event_function_responses(function_response_event) + if not responses: + return False, "" + response = getattr(responses[0], "response", None) + if response is None and isinstance(responses[0], dict): + response = responses[0].get("response") + + if isinstance(response, dict): + error = response.get("error") + status = str(response.get("status", "")).lower() + if error: + return True, "skill_execution_error" + if status in {"error", "failed", "failure"}: + return True, status + if isinstance(response, str) and response.lstrip().lower().startswith( + ("error:", "execution failed:") + ): + return True, "skill_execution_error" + return False, "" + + +def _record_skill_metrics( + span: Span, operation: str, success: bool, error_type: str = "" +) -> None: + from veadk.tracing.telemetry.telemetry import meter_uploader + + if meter_uploader and hasattr(meter_uploader, "record_skill_operation"): + meter_uploader.record_skill_operation( + span=span, + operation=operation, + attributes=active_skill_metric_attributes( + str((getattr(span, "attributes", None) or {}).get("invocation.id", "")) + ), + success=success, + error_type=error_type, + ) diff --git a/veadk/tracing/telemetry/telemetry.py b/veadk/tracing/telemetry/telemetry.py index 1eca260d6..78d3a13af 100644 --- a/veadk/tracing/telemetry/telemetry.py +++ b/veadk/tracing/telemetry/telemetry.py @@ -299,6 +299,11 @@ def set_common_attributes_on_model_span( for attr_name, attr_extractor in common_attributes.items(): value = attr_extractor(**kwargs) current_span.set_attribute(attr_name, value) + from veadk.tracing.telemetry.skill_observability import ( + set_active_skill_attributes, + ) + + set_active_skill_attributes(current_span) except Exception as e: logger.error(f"Failed to set common attributes for spans: {e}") @@ -362,6 +367,10 @@ def trace_tool_call( response: ExtractorResponse = attr_extractor(params) ExtractorResponse.update_span(span, attr_name, response) + from veadk.tracing.telemetry.skill_observability import observe_skill_tool_call + + observe_skill_tool_call(span, tool, args, function_response_event) + _upload_tool_call_metrics(tool, args, function_response_event)