Add SmoLAgents instrumentation

Author: Dwij1704

agentops/instrumentation/__init__.py

@@ -87,6 +87,11 @@ def get_instance(self) -> BaseInstrumentor:
         class_name="AG2Instrumentor",
         provider_import_name="autogen",
     ),
+    InstrumentorLoader(
+        module_name="agentops.instrumentation.smolagents",
+        class_name="SmoLAgentsInstrumentor",
+        provider_import_name="smolagents",
+    ),
 ]
 
 

agentops/instrumentation/common/attributes.py

@@ -252,3 +252,27 @@ def get_base_span_attributes(span: Any) -> AttributeMap:
         attributes[CoreAttributes.PARENT_ID] = parent_id
 
     return attributes
+
+
+def extract_token_usage(response: Any) -> Dict[str, int]:
+    """Extract token usage information from a response.
+
+    Args:
+        response: The response object to extract token usage from
+
+    Returns:
+        Dictionary containing token usage information
+    """
+    usage = {}
+
+    # Try to extract token counts from response
+    if hasattr(response, "usage"):
+        usage_data = response.usage
+        if hasattr(usage_data, "prompt_tokens"):
+            usage["prompt_tokens"] = usage_data.prompt_tokens
+        if hasattr(usage_data, "completion_tokens"):
+            usage["completion_tokens"] = usage_data.completion_tokens
+        if hasattr(usage_data, "total_tokens"):
+            usage["total_tokens"] = usage_data.total_tokens
+
+    return usage

agentops/instrumentation/smolagents/README.md

@@ -0,0 +1,88 @@
+# SmoLAgents Instrumentation
+
+This module provides OpenTelemetry instrumentation for the SmoLAgents framework. It captures telemetry data from model operations, agent executions, and tool usage.
+
+## Features
+
+- Model operation tracking
+  - Text generation
+  - Token usage
+  - Streaming responses
+  - Latency metrics
+
+- Agent execution monitoring
+  - Step-by-step execution
+  - Planning phases
+  - Tool usage
+  - Execution time
+
+- Tool usage analytics
+  - Tool call patterns
+  - Success/failure rates
+  - Execution time
+  - Error tracking
+
+## Usage
+
+```python
+from agentops import init
+from agentops.instrumentation.smolagents import SmoLAgentsInstrumentor
+
+# Initialize AgentOps with your API key
+init(api_key="your-api-key")
+
+# The instrumentation will be automatically activated
+# All SmoLAgents operations will now be tracked
+```
+
+## Metrics Collected
+
+1. Token Usage
+   - Input tokens
+   - Output tokens
+   - Total tokens per operation
+
+2. Timing Metrics
+   - Operation duration
+   - Time to first token (streaming)
+   - Tool execution time
+   - Planning phase duration
+
+3. Agent Metrics
+   - Step counts
+   - Planning steps
+   - Tools used
+   - Success/failure rates
+
+4. Error Tracking
+   - Generation errors
+   - Tool execution errors
+   - Parsing errors
+
+## Architecture
+
+The instrumentation is built on OpenTelemetry and follows the same pattern as other AgentOps instrumentors:
+
+1. Attribute Extractors
+   - Model attributes
+   - Agent attributes
+   - Tool call attributes
+
+2. Wrappers
+   - Method wrappers for sync operations
+   - Stream wrappers for async operations
+   - Context propagation handling
+
+3. Metrics
+   - Histograms for distributions
+   - Counters for events
+   - Custom attributes for filtering
+
+## Contributing
+
+When adding new features or modifying existing ones:
+
+1. Follow the established pattern for attribute extraction
+2. Maintain context propagation
+3. Add appropriate error handling
+4. Update tests and documentation 

agentops/instrumentation/smolagents/__init__.py

@@ -0,0 +1,7 @@
+"""SmoLAgents instrumentation for AgentOps."""
+from .instrumentor import SmoLAgentsInstrumentor
+
+LIBRARY_NAME = "smolagents"
+LIBRARY_VERSION = "1.16.0"
+
+__all__ = ["SmoLAgentsInstrumentor"]

agentops/instrumentation/smolagents/attributes/agent.py

@@ -0,0 +1,145 @@
+"""Attribute extractors for SmoLAgents agent operations."""
+
+from typing import Any, Dict, Optional, Tuple
+import uuid
+import time
+
+from agentops.instrumentation.common.attributes import get_common_attributes
+from agentops.semconv.agent import AgentAttributes
+from agentops.semconv.tool import ToolAttributes
+
+
+def get_agent_attributes(
+    args: Optional[Tuple] = None,
+    kwargs: Optional[Dict] = None,
+    return_value: Optional[Any] = None,
+) -> Dict[str, Any]:
+    """Extract attributes from an agent execution.
+
+    Args:
+        args: Optional tuple of positional arguments
+        kwargs: Optional dict of keyword arguments
+        return_value: Optional return value from the wrapped function
+
+    Returns:
+        Dict containing extracted attributes
+    """
+    attributes = get_common_attributes()
+
+    # Extract agent info from instance
+    if args and len(args) > 0:
+        instance = args[0]
+        agent_type = instance.__class__.__name__
+        tools = [t.name for t in instance.tools] if hasattr(instance, "tools") else []
+        attributes.update(
+            {
+                AgentAttributes.AGENT_ID: str(uuid.uuid4()),
+                AgentAttributes.AGENT_NAME: agent_type,
+                AgentAttributes.AGENT_ROLE: "executor",
+                AgentAttributes.AGENT_TOOLS: tools,
+            }
+        )
+
+    # Extract task from kwargs or args
+    task = kwargs.get("task", args[1] if len(args) > 1 else "unknown") if kwargs else "unknown"
+    attributes[AgentAttributes.AGENT_REASONING] = task
+
+    return attributes
+
+
+def get_tool_call_attributes(
+    args: Optional[Tuple] = None,
+    kwargs: Optional[Dict] = None,
+    return_value: Optional[Any] = None,
+) -> Dict[str, Any]:
+    """Extract attributes from a tool call.
+
+    Args:
+        args: Optional tuple of positional arguments
+        kwargs: Optional dict of keyword arguments
+        return_value: Optional return value from the wrapped function
+
+    Returns:
+        Dict containing extracted attributes
+    """
+    attributes = get_common_attributes()
+
+    # Extract tool info from instance and args
+    if args and len(args) > 0:
+        instance = args[0]
+        tool_name = instance.name if hasattr(instance, "name") else "unknown"
+        tool_description = instance.description if hasattr(instance, "description") else "unknown"
+
+        # Get arguments from args/kwargs
+        arguments = {}
+        if len(args) > 1:
+            arguments = args[1]
+        elif kwargs:
+            arguments = kwargs
+
+        # Track execution time and success
+        start_time = time.time()
+        error = None
+
+        try:
+            if return_value is not None:
+                execution_time = time.time() - start_time
+                attributes.update(
+                    {
+                        ToolAttributes.TOOL_ID: str(uuid.uuid4()),
+                        ToolAttributes.TOOL_NAME: tool_name,
+                        ToolAttributes.TOOL_DESCRIPTION: tool_description,
+                        ToolAttributes.TOOL_PARAMETERS: arguments,
+                        ToolAttributes.TOOL_STATUS: "success",
+                        ToolAttributes.TOOL_RESULT: str(return_value),
+                        "tool.execution_time": execution_time,
+                    }
+                )
+        except Exception as e:
+            error = str(e)
+            attributes.update(
+                {
+                    ToolAttributes.TOOL_ID: str(uuid.uuid4()),
+                    ToolAttributes.TOOL_NAME: tool_name,
+                    ToolAttributes.TOOL_DESCRIPTION: tool_description,
+                    ToolAttributes.TOOL_PARAMETERS: arguments,
+                    ToolAttributes.TOOL_STATUS: "error",
+                    ToolAttributes.TOOL_ERROR: error,
+                }
+            )
+
+    return attributes
+
+
+def get_planning_step_attributes(
+    args: Optional[Tuple] = None,
+    kwargs: Optional[Dict] = None,
+    return_value: Optional[Any] = None,
+) -> Dict[str, Any]:
+    """Extract attributes from a planning step.
+
+    Args:
+        args: Optional tuple of positional arguments
+        kwargs: Optional dict of keyword arguments
+        return_value: Optional return value from the wrapped function
+
+    Returns:
+        Dict containing extracted attributes
+    """
+    attributes = get_common_attributes()
+
+    # Extract step info from kwargs
+    if kwargs:
+        step_number = kwargs.get("step_number", 0)
+        is_first_step = kwargs.get("is_first_step", False)
+        task = kwargs.get("task", "unknown")
+
+        attributes.update(
+            {
+                AgentAttributes.AGENT_REASONING: task,
+                "planning.step_number": step_number,
+                "planning.is_first_step": is_first_step,
+            }
+        )
+
+    return attributes