Spaces:
Running
Running
File size: 4,343 Bytes
fea07c2 a9fb876 fea07c2 a9fb876 fea07c2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
from typing import Any, Dict, List
import json
from any_agent import AgentFramework
from surf_spot_finder.evaluation.telemetry import TelemetryProcessor
class SmolagentsTelemetryProcessor(TelemetryProcessor):
"""Processor for SmoL Agents telemetry data."""
def _get_agent_framework(self) -> AgentFramework:
return AgentFramework.SMOLAGENTS
def extract_hypothesis_answer(self, trace: List[Dict[str, Any]]) -> str:
for span in reversed(trace):
if span["attributes"]["openinference.span.kind"] == "AGENT":
content = span["attributes"]["output.value"]
return content
raise ValueError("No agent final answer found in trace")
def _extract_telemetry_data(self, telemetry: List[Dict[str, Any]]) -> List[Dict]:
"""Extract LLM calls and tool calls from SmoL Agents telemetry."""
calls = []
for span in telemetry:
# Skip spans without attributes
if "attributes" not in span:
continue
attributes = span["attributes"]
# Extract tool information
if "tool.name" in attributes or span.get("name", "").startswith(
"SimpleTool"
):
tool_info = {
"tool_name": attributes.get(
"tool.name", span.get("name", "Unknown tool")
),
"status": "success"
if span.get("status", {}).get("status_code") == "OK"
else "error",
"error": span.get("status", {}).get("description", None),
}
# Extract input if available
if "input.value" in attributes:
try:
input_value = json.loads(attributes["input.value"])
if "kwargs" in input_value:
# For SmoLAgents, the actual input is often in the kwargs field
tool_info["input"] = input_value["kwargs"]
else:
tool_info["input"] = input_value
except (json.JSONDecodeError, TypeError):
tool_info["input"] = attributes["input.value"]
# Extract output if available
if "output.value" in attributes:
try:
# Try to parse JSON output
output_value = (
json.loads(attributes["output.value"])
if isinstance(attributes["output.value"], str)
else attributes["output.value"]
)
tool_info["output"] = output_value
except (json.JSONDecodeError, TypeError):
tool_info["output"] = attributes["output.value"]
else:
tool_info["output"] = "No output found"
calls.append(tool_info)
# Extract LLM calls to see reasoning
elif "LiteLLMModel.__call__" in span.get("name", ""):
# The LLM output may be in different places depending on the implementation
output_content = None
# Try to get the output from the llm.output_messages.0.message.content attribute
if "llm.output_messages.0.message.content" in attributes:
output_content = attributes["llm.output_messages.0.message.content"]
# Or try to parse it from the output.value as JSON
elif "output.value" in attributes:
try:
output_value = json.loads(attributes["output.value"])
if "content" in output_value:
output_content = output_value["content"]
except (json.JSONDecodeError, TypeError):
pass
if output_content:
calls.append(
{
"model": attributes.get("llm.model_name", "Unknown model"),
"output": output_content,
"type": "reasoning",
}
)
return calls
|