Merge pull request #616 from UiPath/feature/eval-trajectory

feat: trajectory eval

Author: GabrielVasilescu04

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.1.66"
+version = "2.1.67"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.10"

src/uipath/_cli/_evals/_evaluator_factory.py

@@ -123,6 +123,22 @@ def _create_llm_as_judge_evaluator(
     @staticmethod
     def _create_trajectory_evaluator(
         base_params: EvaluatorBaseParams, data: Dict[str, Any]
-    ) -> TrajectoryEvaluator[Any]:
+    ) -> TrajectoryEvaluator:
         """Create a trajectory evaluator."""
-        raise NotImplementedError()
+        prompt = data.get("prompt", "")
+        if not prompt:
+            raise ValueError("Trajectory evaluator must include 'prompt' field")
+
+        model = data.get("model", "")
+        if not model:
+            raise ValueError("LLM evaluator must include 'model' field")
+        if model == "same-as-agent":
+            raise ValueError(
+                "'same-as-agent' model option is not supported by coded agents evaluations. Please select a specific model for the evaluator."
+            )
+
+        return TrajectoryEvaluator(
+            **base_params.model_dump(),
+            prompt=prompt,
+            model=model,
+        )

src/uipath/_cli/_evals/_models/_trajectory_span.py

@@ -0,0 +1,115 @@
+"""Trajectory evaluation span model for serializing span data in evaluations."""
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from opentelemetry.sdk.trace import ReadableSpan
+from pydantic import BaseModel
+
+
+@dataclass
+class TrajectoryEvaluationSpan:
+    """Simplified span representation for trajectory evaluation.
+
+    Contains span information needed for evaluating agent execution paths,
+    excluding timestamps which are not useful for trajectory analysis.
+    """
+
+    name: str
+    status: str
+    attributes: Dict[str, Any]
+    parent_name: Optional[str] = None
+    events: Optional[List[Dict[str, Any]]] = None
+
+    def __post_init__(self):
+        """Initialize default values."""
+        if self.events is None:
+            self.events = []
+
+    @classmethod
+    def from_readable_span(
+        cls, span: ReadableSpan, parent_spans: Optional[Dict[int, str]] = None
+    ) -> "TrajectoryEvaluationSpan":
+        """Convert a ReadableSpan to a TrajectoryEvaluationSpan.
+
+        Args:
+            span: The OpenTelemetry ReadableSpan to convert
+            parent_spans: Optional mapping of span IDs to names for parent lookup
+
+        Returns:
+            TrajectoryEvaluationSpan with relevant data extracted
+        """
+        # Extract status
+        status_map = {0: "unset", 1: "ok", 2: "error"}
+        status = status_map.get(span.status.status_code.value, "unknown")
+
+        # Extract attributes - keep all attributes for now
+        attributes = {}
+        if span.attributes:
+            attributes = dict(span.attributes)
+
+        # Get parent name if available
+        parent_name = None
+        if span.parent and parent_spans and span.parent.span_id in parent_spans:
+            parent_name = parent_spans[span.parent.span_id]
+
+        # Extract events (without timestamps)
+        events = []
+        if hasattr(span, "events") and span.events:
+            for event in span.events:
+                event_data = {
+                    "name": event.name,
+                    "attributes": dict(event.attributes) if event.attributes else {},
+                }
+                events.append(event_data)
+
+        return cls(
+            name=span.name,
+            status=status,
+            attributes=attributes,
+            parent_name=parent_name,
+            events=events,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            "name": self.name,
+            "status": self.status,
+            "parent_name": self.parent_name,
+            "attributes": self.attributes,
+            "events": self.events,
+        }
+
+
+class TrajectoryEvaluationTrace(BaseModel):
+    """Container for a collection of trajectory evaluation spans."""
+
+    spans: List[TrajectoryEvaluationSpan]
+
+    @classmethod
+    def from_readable_spans(
+        cls, spans: List[ReadableSpan]
+    ) -> "TrajectoryEvaluationTrace":
+        """Convert a list of ReadableSpans to TrajectoryEvaluationTrace.
+
+        Args:
+            spans: List of OpenTelemetry ReadableSpans to convert
+
+        Returns:
+            TrajectoryEvaluationTrace with converted spans
+        """
+        # Create a mapping of span IDs to names for parent lookup
+        span_id_to_name = {span.get_span_context().span_id: span.name for span in spans}
+
+        evaluation_spans = [
+            TrajectoryEvaluationSpan.from_readable_span(span, span_id_to_name)
+            for span in spans
+        ]
+
+        return cls(spans=evaluation_spans)
+
+    class Config:
+        """Pydantic configuration."""
+
+        arbitrary_types_allowed = True

src/uipath/_cli/_evals/_runtime.py

@@ -264,6 +264,7 @@ async def run_evaluator(
             agent_input=eval_item.inputs,
             agent_output=execution_output.result.output or {},
             agent_trace=execution_output.spans,
+            expected_agent_behavior=eval_item.expected_agent_behavior,
         )
 
         result = await evaluator.evaluate(

src/uipath/eval/evaluators/trajectory_evaluator.py

@@ -1,36 +1,159 @@
 """Trajectory evaluator for analyzing execution paths and decision sequences."""
 
-from typing import TypeVar
+import json
+from typing import Any, Optional
 
+from opentelemetry.sdk.trace import ReadableSpan
+from pydantic import field_validator
+
+from uipath._cli._evals._models._trajectory_span import TrajectoryEvaluationTrace
 from uipath.eval.models import EvaluationResult
 
-from ..models.models import AgentExecution
+from ..._services import UiPathLlmChatService
+from ..._utils.constants import COMMUNITY_agents_SUFFIX
+from ..models.models import AgentExecution, LLMResponse, NumericEvaluationResult
 from .base_evaluator import BaseEvaluator
 
-T = TypeVar("T")
-
 
-class TrajectoryEvaluator(BaseEvaluator[T]):
+class TrajectoryEvaluator(BaseEvaluator[dict[str, Any]]):
     """Evaluator that analyzes the trajectory/path taken to reach outputs."""
 
+    prompt: str
+    model: str
+    expected_agent_behavior_placeholder: str = "{{ExpectedAgentBehavior}}"
+    agent_run_history_placeholder: str = "{{AgentRunHistory}}"
+    llm: Optional[UiPathLlmChatService] = None
+
+    @field_validator("prompt")
+    @classmethod
+    def validate_prompt_placeholder(cls, v: str) -> str:
+        """Validate that prompt contains required placeholders."""
+        if "{{ExpectedAgentBehavior}}" not in v or "{{AgentRunHistory}}" not in v:
+            raise ValueError(
+                "Prompt must contain {ExpectedAgentBehavior} and {{AgentRunHistory}} placeholders"
+            )
+        return v
+
+    def model_post_init(self, __context):
+        """Initialize the LLM service after model creation."""
+        super().model_post_init(__context)
+        self._initialize_llm()
+
+    def _initialize_llm(self):
+        """Initialize the LLM used for evaluation."""
+        from uipath import UiPath
+
+        uipath = UiPath()
+        self.llm = uipath.llm
+
     async def evaluate(
-        self, agent_execution: AgentExecution, evaluation_criteria: T
+        self,
+        agent_execution: AgentExecution,
+        evaluation_criteria: dict[str, Any],
     ) -> EvaluationResult:
         """Evaluate using trajectory analysis.
 
-        Analyzes the execution path and decision sequence taken by the agent
-        to assess the quality of the reasoning process.
+        Analyzes the execution path and decision sequence taken by the agent.
 
         Args:
             agent_execution: The execution details containing:
                 - agent_input: The input received by the agent
                 - actual_output: The actual output from the agent
-                - spans: The execution spans to use for the evaluation
+                - agent_trace: The execution spans to use for the evaluation
+                - expected_agent_behavior: The expected agent behavior
             evaluation_criteria: The criteria to evaluate
         Returns:
             EvaluationResult: Score based on trajectory analysis
 
         Raises:
             NotImplementedError: This evaluator is not yet implemented
         """
-        raise NotImplementedError()
+        evaluation_prompt = self._create_evaluation_prompt(
+            expected_agent_behavior=agent_execution.expected_agent_behavior,
+            agent_run_history=agent_execution.agent_trace,
+        )
+
+        llm_response = await self._get_llm_response(evaluation_prompt)
+
+        return NumericEvaluationResult(
+            score=llm_response.score,
+            details=llm_response.justification,
+        )
+
+    def _create_evaluation_prompt(
+        self,
+        expected_agent_behavior: Any,
+        agent_run_history: Any,
+    ) -> str:
+        """Create the evaluation prompt for the LLM."""
+        formatted_prompt = self.prompt.replace(
+            self.expected_agent_behavior_placeholder,
+            str(expected_agent_behavior),
+        )
+
+        # Trim extra properties from the spans (such as timestamps which are not relevant to the eval)
+        if (
+            isinstance(agent_run_history, list)
+            and agent_run_history
+            and isinstance(agent_run_history[0], ReadableSpan)
+        ):
+            trajectory_trace = TrajectoryEvaluationTrace.from_readable_spans(
+                agent_run_history
+            )
+            agent_run_history = str(trajectory_trace.spans)
+        else:
+            agent_run_history = str(agent_run_history)
+
+        formatted_prompt = formatted_prompt.replace(
+            self.agent_run_history_placeholder,
+            agent_run_history,
+        )
+
+        return formatted_prompt
+
+    async def _get_llm_response(self, evaluation_prompt: str) -> LLMResponse:
+        """Get response from the LLM.
+
+        Args:
+            evaluation_prompt: The formatted prompt to send to the LLM
+
+        Returns:
+            LLMResponse with score and justification
+        """
+        if not self.llm:
+            raise ValueError("LLM service not initialized")
+
+        model = self.model
+        if model.endswith(COMMUNITY_agents_SUFFIX):
+            model = model.replace(COMMUNITY_agents_SUFFIX, "")
+
+        # Prepare the request
+        request_data = {
+            "model": model,
+            "messages": [{"role": "user", "content": evaluation_prompt}],
+            "response_format": {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": "evaluation_response",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "score": {
+                                "type": "number",
+                                "minimum": 0,
+                                "maximum": 100,
+                                "description": "Score between 0 and 100",
+                            },
+                            "justification": {
+                                "type": "string",
+                                "description": "Explanation for the score",
+                            },
+                        },
+                        "required": ["score", "justification"],
+                    },
+                },
+            },
+        }
+
+        response = await self.llm.chat_completions(**request_data)
+        return LLMResponse(**json.loads(response.choices[-1].message.content))

src/uipath/eval/models/models.py

@@ -15,6 +15,7 @@ class AgentExecution(BaseModel):
     agent_input: Optional[Dict[str, Any]]
     agent_output: Dict[str, Any]
     agent_trace: list[ReadableSpan]
+    expected_agent_behavior: Optional[str] = None
 
 
 class LLMResponse(BaseModel):