Merge pull request volcengine#529 from wujiaming-ai/feat/skill_observe

feat: 新增skill observe功能

Author: wujiaming-ai

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "veadk-python"
-version = "0.5.22"
+version = "0.5.23"
 description = "Volcengine agent development kit, integrations with Volcengine cloud services."
 readme = "README.md"
 requires-python = ">=3.10"

veadk/skills/skill.py

@@ -23,6 +23,7 @@ class Skill(BaseModel):
     skill_space_id: Optional[str] = None
     bucket_name: Optional[str] = None
     checklist: List[Dict[str, str]] = []
+    id: Optional[str] = None
 
     def get_checklist_items(self) -> List[str]:
         return [item.get("item", item.get("id", "")) for item in self.checklist]

veadk/skills/utils.py

@@ -197,6 +197,7 @@ def load_skills_from_cloud(skill_space_ids: str) -> list[Skill]:
                 skill_description = item.get("Description")
                 tos_bucket = item.get("BucketName")
                 tos_path = item.get("TosPath")
+                skill_id = item.get("SkillId")
                 if not skill_name:
                     continue
 
@@ -206,6 +207,7 @@ def load_skills_from_cloud(skill_space_ids: str) -> list[Skill]:
                     path=tos_path,
                     skill_space_id=skill_space_id,
                     bucket_name=tos_bucket,
+                    id=skill_id,
                 )
 
                 skills.append(skill)

veadk/tools/builtin_tools/execute_skills.py

@@ -170,6 +170,7 @@ def execute_skills(
         "TOS_SKILLS_DIR": f"tos://agentkit-platform-{account_id}/skills/",
         "SKILL_SPACE_ID": skill_space_id,
         "TOOL_USER_SESSION_ID": tool_user_session_id,
+        "PYTHONPATH": "$SRV_PYTHONPATH:$PYTHONPATH",
     }
 
     code = f"""

veadk/tools/skills_tools/skills_tool.py

@@ -19,11 +19,17 @@
 
 from google.adk.tools import BaseTool, ToolContext
 from google.genai import types
+from opentelemetry import trace
+from opentelemetry.sdk.trace import _Span
+from opentelemetry.trace.status import Status, StatusCode
 
 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.utils.logger import get_logger
 
+tracer = trace.get_tracer("veadk.skills_tool")
+
 logger = get_logger(__name__)
 
 
@@ -94,7 +100,11 @@ async def run_async(
         if not skill_name:
             return "Error: No skill name provided"
 
-        return self._invoke_skill(skill_name, tool_context)
+        with tracer.start_as_current_span(f"execute_skill {skill_name}") as span:
+            result = self._invoke_skill(skill_name, tool_context)
+            self._add_skill_span_attributes(span, skill_name, result)
+            self._upload_skill_metrics(span, skill_name, result)
+            return result
 
     def _invoke_skill(self, skill_name: str, tool_context: ToolContext) -> str:
         """Load and return the full content of a skill."""
@@ -345,3 +355,75 @@ def _format_skill_content(self, skill_name: str, content: str, skill_dir) -> str
             "The skill has been loaded. Follow the instructions above and use the bash tool to execute commands."
         )
         return header + content + footer
+
+    def _add_skill_span_attributes(
+        self,
+        span: _Span,
+        skill_name: str,
+        result: str,
+    ) -> None:
+        """Add attributes to the skill execution span."""
+        try:
+            set_common_attributes_on_tool_span(current_span=span)
+
+            if result:
+                if result.startswith("Error:"):
+                    span.set_status(Status(StatusCode.ERROR, result))
+
+            span.set_attribute("skill.name", skill_name)
+            span.set_attribute("tool.name", self.name)
+            span.set_attribute("gen_ai.operation.name", "execute_skill")
+            span.set_attribute("gen_ai.span.kind", "tool")
+            if skill_name in self.skills:
+                skill = self.skills[skill_name]
+                if hasattr(skill, "skill_space_id") and skill.skill_space_id:
+                    span.set_attribute("skill.space_id", skill.skill_space_id)
+                if hasattr(skill, "bucket_name") and skill.bucket_name:
+                    span.set_attribute("skill.bucket_name", skill.bucket_name)
+                if hasattr(skill, "path") and skill.path:
+                    span.set_attribute("skill.path", skill.path)
+                if hasattr(skill, "id") and skill.id:
+                    span.set_attribute("skill.id", skill.id)
+            logger.debug(f"Added skill span attributes for {skill_name}")
+        except Exception as e:
+            logger.warning(f"Failed to add skill span attributes: {e}")
+
+    def _upload_skill_metrics(self, span: _Span, skill_name: str, result: str) -> None:
+        """Upload skill metrics to the telemetry system."""
+        try:
+            import time
+            from veadk.tracing.telemetry.telemetry import meter_uploader
+
+            if meter_uploader:
+                # 初始化属性，包含技能相关信息
+                skill = self.skills.get(skill_name)
+                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 ""
+                    ),
+                }
+
+                # 计算 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 记录技能执行延迟（秒）
+                    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}")

veadk/tracing/telemetry/exporters/apmplus_exporter.py

@@ -151,6 +151,8 @@ class Meters:
     APMPLUS_SPAN_LATENCY = "apmplus_span_latency"
     # tool token usage
     APMPLUS_TOOL_TOKEN_USAGE = "apmplus_tool_token_usage"
+    # skill invoke latency
+    GEN_AI_SKILL_INVOKE_LATENCY = "gen_ai_skill_invoke_latency"
 
 
 class MeterUploader:
@@ -269,6 +271,12 @@ def __init__(
             unit="count",
             explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_TOKEN_USAGE_BUCKETS,
         )
+        self.skill_invoke_latency = self.meter.create_histogram(
+            name=Meters.GEN_AI_SKILL_INVOKE_LATENCY,
+            description="Latency of skill invocations",
+            unit="s",
+            explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS,
+        )
 
     def record_call_llm(
         self,

veadk/version.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-VERSION = "0.5.22"
+VERSION = "0.5.23"