railtracks.evaluations

135def evaluate(
136    data: AgentDataPoint | list[AgentDataPoint],
137    evaluators: list[Evaluator],
138    agent_selection: bool = True,
139    agents: list[str] | None = None,
140    name: str | None = None,
141    payload_callback: Callable[[dict[str, Any]], Any] | None = None,
142):
143    """Evaluate agent data using the provided evaluators.
144
145    Args:
146        data: The agent data to be evaluated. Can be a single AgentDataPoint, a list of AgentDataPoints, or an EvaluationDataset.
147        evaluators: A list of Evaluator instances to run on the data.
148        agent_selection: If True and multiple agents are found in the data, prompts the user to select which agents to evaluate.
149                         If False, evaluates all agents without prompting.
150        agents: An optional list of agent names to evaluate. If provided, only these agents will be evaluated. Overrides agent_selection if both are provided.
151        name: An optional name for the evaluation, which will be included in the EvaluationResult.
152        payload_callback: An optional callback function that will be called with the evaluation results payload after evaluation is complete. Can be used for custom logging, notifications, etc.
153    Returns:
154        A list of EvaluationResult instances containing the results from each evaluator.
155    """
156    _check_evaluators(evaluators)
157
158    data_dict, agents = _setup_agent_data(data, agent_selection, agents)
159
160    evaluation_results: list[EvaluationResult] = []
161
162    for agent_name in agents:
163        logger.info(
164            f"Evaluation for {agent_name} with {len(data_dict[agent_name])} data points CREATED"
165        )
166
167        evaluator_results: list[EvaluatorResult] = []
168
169        start_time = datetime.now(timezone.utc)
170        for evaluator in evaluators:
171            logger.info(f"Evaluator: {evaluator.__class__.__name__} CREATED")
172            try:
173                result = evaluator.run(data_dict[agent_name])
174            except Exception as e:
175                logger.error(f"Evaluator {evaluator.__class__.__name__} FAILED: {e}")
176                continue
177
178            evaluator_results.append(result)
179            logger.info(f"Evaluator: {evaluator.__class__.__name__} DONE")
180
181        logger.info(f"Evaluation for {agent_name} DONE.")
182
183        metrics_map = {}
184        for er in evaluator_results:
185            metrics = er.metrics
186            for metric in metrics:
187                metrics_map[metric.identifier] = metric
188
189        end_time = datetime.now(timezone.utc)
190
191        evaluation_results.append(
192            EvaluationResult(
193                evaluation_name=name or None,
194                created_at=start_time,
195                completed_at=end_time,
196                agents=[
197                    {
198                        "agent_name": agent_name,
199                        "agent_node_ids": [
200                            {
201                                "session_id": adp.session_id,
202                                "agent_node_id": adp.identifier,
203                            }
204                            for adp in data_dict[agent_name]
205                        ],
206                    }
207                ],
208                metrics_map=metrics_map,
209                evaluator_results=evaluator_results,
210            )
211        )
212
213    logger.info("Evaluation DONE.")
214
215    if payload_callback is not None:
216        try:
217            for result in evaluation_results:
218                payload_callback(payload(result))
219        except Exception as e:
220            logger.error(f"Failed to execute payload callback: {e}")
221
222    try:
223        save(evaluation_results)
224    except Exception as e:
225        logger.error(f"Failed to save evaluation results: {e}")
226    return evaluation_results

401def extract_agent_data_points(
402    sources: list[str] | str | list[dict],
403) -> list[AgentDataPoint]:
404    """Extract AgentDataPoint instances from session payloads or session JSON files.
405
406    Args:
407        sources: One of:
408            - list[dict]: in-memory session payloads (e.g. from railtownai.get_agent_runs).
409            - list[str]: file paths to session JSON files.
410            - str: a directory path; all files inside are loaded.
411            A single file or single payload must be wrapped in a list. Mixed lists of
412            files and payloads are not supported.
413
414    Returns:
415        List of AgentDataPoint instances, one per agent execution found across all
416        provided payloads. Returns an empty list if no valid agent data is found.
417    """
418    if isinstance(sources, list) and sources and isinstance(sources[0], dict):
419        payloads: list[dict] = cast(list[dict], sources)
420    else:
421        file_sources = cast("list[str] | str", sources)
422        payloads = []
423        for file_path in resolve_file_paths(file_sources):
424            try:
425                payloads.append(load_session(file_path))
426            except (FileNotFoundError, ValueError) as e:
427                logger.error(str(e))
428
429    data_points: list[AgentDataPoint] = []
430    for payload in payloads:
431        data_points.extend(_data_points_from_payload(payload))
432    return data_points

 37class JudgeEvaluator(Evaluator):
 38    def __init__(
 39        self,
 40        llm: rt.llm.ModelBase,
 41        metrics: list[Metric],
 42        system_prompt: str | None = None,
 43        timeout: float | None = None,
 44        reasoning: bool = True,
 45    ):
 46        """
 47        The JudgeEvaluator with a system prompt, LLM, metric, and reasoning flag.
 48
 49        Args:
 50            system_prompt: The system prompt template for the judge LLM.
 51            llm: The LLM model to be used as the judge.
 52            metrics: A list of Metrics to guide the evaluation.
 53            reasoning: A flag indicating whether the judge should provide reasoning for its evaluations.
 54        """
 55        # These are config not state
 56        self._metrics: dict[str, Metric] = {m.identifier: m for m in metrics}
 57        for m in metrics:
 58            if isinstance(m, Categorical):
 59                self._metrics[m.identifier] = m
 60            else:
 61                logger.warning(
 62                    f"JudgeEvaluator currently only supports Categorical metrics, metric {m.name} of type {type(m)} will be skipped."
 63                )
 64        self._llm = llm
 65        self._reasoning: bool = reasoning
 66        self._template = self._load_yaml()
 67        self._system_prompt = (
 68            system_prompt
 69            if system_prompt is not None
 70            else self._template["system_prompt"]
 71        )
 72        super().__init__()
 73
 74        self.timeout = timeout
 75        self._judge = rt.agent_node(
 76            llm=self._llm,
 77            output_schema=JudgeResponseSchema,
 78            tool_nodes=[],
 79        )
 80
 81    def run(
 82        self, data: list[AgentDataPoint]
 83    ) -> EvaluatorResult[Metric, MetricResult, CategoricalAggregateNode]:
 84        judge_outputs: list[JudgeOutput] = self._invoke(data)
 85
 86        self.agent_data_ids = {adp.identifier for adp in data}
 87        results: dict[Metric, list[MetricResult]] = defaultdict(list)
 88        forest = AggregateForest[CategoricalAggregateNode, MetricResult]()
 89
 90        for output in judge_outputs:
 91            metric = self._metrics[output.metric_id]
 92
 93            metric_result = MetricResult(
 94                result_name=f"JudgeResult/{metric.name}",
 95                metric_id=metric.identifier,
 96                agent_data_id=[UUID(output.adp_id)],
 97                value=output.response.metric_value,
 98            )
 99            results[metric].append(metric_result)
100            forest.add_node(metric_result)
101
102            if self._reasoning:
103                reasoning_metric = Metric(name=f"{metric.name}_reasoning")
104                if output.response.reasoning is not None:
105                    results[reasoning_metric].append(
106                        MetricResult(
107                            result_name=f"JudgeReasoning/{metric.name}",
108                            metric_id=reasoning_metric.identifier,
109                            agent_data_id=[UUID(output.adp_id)],
110                            value=output.response.reasoning,
111                        )
112                    )
113                else:
114                    logger.warning(
115                        f"No reasoning returned for Judge Evaluator Metric: {metric.name}, AgentDataPoint ID: {output.adp_id}"
116                    )
117
118        self._aggregate_metrics(results, forest)
119
120        self._result = EvaluatorResult(
121            evaluator_name=self.name,
122            evaluator_id=self.identifier,
123            agent_data_ids=self.agent_data_ids,
124            metric_results=[item for sublist in results.values() for item in sublist],
125            aggregate_results=forest,
126            metrics=list(self._metrics.values()),
127        )
128        return self._result
129
130    def __repr__(self) -> str:
131        return (
132            f"JudgeEvaluator, "
133            f"llm={self._llm}, "
134            f"metrics={list(self._metrics.values())}, "
135            f"reasoning={self._reasoning})"
136        )
137
138    def _invoke(self, data: list[AgentDataPoint]) -> list[JudgeOutput]:
139        @rt.function_node
140        async def judge_flow():
141            output: list[JudgeOutput] = []
142            for metric in self._metrics.values():
143                logger.info(
144                    f"START Evaluating Metric: {metric.name} for {len(data)} AgentDataPoints"
145                )
146
147                for idx, adp in enumerate(data):
148                    user_message = self._generate_user_prompt(adp)
149                    system_message = self._generate_system_prompt(metric)
150                    message_history = rt.llm.MessageHistory(
151                        [
152                            rt.llm.SystemMessage(system_message),
153                            rt.llm.UserMessage(user_message),
154                        ]
155                    )
156                    res = await rt.call(
157                        self._judge,
158                        message_history,
159                    )
160                    output.append(
161                        JudgeOutput(
162                            metric_id=metric.identifier,
163                            adp_id=str(adp.identifier),
164                            response=res.structured,
165                        )
166                    )
167
168                    logger.info(
169                        f"AgentDataPoint ID: {adp.identifier} {idx + 1}/{len(data)} DONE"
170                    )
171
172            return output
173
174        judge_evaluator_flow = rt.Flow(
175            name="JudgeEvaluatorFlow",
176            entry_point=judge_flow,
177            timeout=self.timeout,
178            save_state=False,
179        )
180
181        return judge_evaluator_flow.invoke()
182
183    def _aggregate_metrics(
184        self,
185        results: dict[Metric, list[MetricResult]],
186        forest: AggregateForest[CategoricalAggregateNode, MetricResult],
187    ) -> None:
188        for metric in results:
189            if isinstance(metric, Numerical):
190                continue
191            elif isinstance(metric, Categorical):
192                aggregate_node = CategoricalAggregateNode(
193                    name=f"Aggregate/{metric.name}",
194                    metric=metric,
195                    children=[val.identifier for val in results[metric]],
196                    forest=forest,
197                )
198
199                forest.roots.append(aggregate_node.identifier)
200                forest.add_node(aggregate_node)
201
202    def _generate_user_prompt(self, data: AgentDataPoint) -> str:
203        return self._template["user"].format(
204            agent_input=data.agent_input,
205            agent_output=data.agent_output.get("message_history", ""),
206        )
207
208    def _generate_system_prompt(self, metric: Metric) -> str:
209        system_prompt: str = self._template["system_prompt"]
210
211        system_prompt += "\n" + self._template["metric"].format(metric=str(metric))
212
213        if isinstance(metric, Categorical):
214            category_names = ", ".join(c for c in metric.category_names)
215            system_prompt += (
216                f"\nYour metric_value must be exactly one of these category "
217                f"names: {category_names}."
218            )
219
220        if self._reasoning:
221            system_prompt += self._template["reasoning"]
222
223        return system_prompt
224
225    def _load_yaml(self):
226        yaml_path = Path(__file__).parent / "judge_evaluator.yaml"
227        with open(yaml_path, "r") as f:
228            template = yaml.safe_load(f)
229
230        return template
231
232    def _get_config(self) -> dict:
233        return {
234            "llm": self._llm.model_name(),
235            "llm_provider": self._llm.model_provider(),
236            "system_prompt": self._system_prompt,
237            "metrics": sorted(self._metrics.keys()),
238            "reasoning": self._reasoning,
239        }

 48class ToolUseEvaluator(Evaluator):
 49    """
 50    Evaluator that analyzes tool usage patterns across agent runs.
 51
 52    Computes per-call and aggregated metrics for each tool, including
 53    runtime, failure rate, and usage count.
 54    """
 55
 56    def __init__(
 57        self,
 58    ):
 59        super().__init__()
 60
 61    def run(
 62        self, data: list[AgentDataPoint]
 63    ) -> EvaluatorResult[ToolMetric, ToolMetricResult, ToolAggregateNode]:
 64        """
 65        Run the evaluator over a list of agent data points.
 66
 67        Args:
 68            data: A list of AgentDataPoint instances to evaluate.
 69
 70        Returns:
 71            An EvaluatorResult containing per-call metric results and
 72            aggregated nodes across runs.
 73        """
 74        agent_data_ids: set[UUID] = {adp.identifier for adp in data}
 75        forest = AggregateForest[ToolAggregateNode, ToolMetricResult]()
 76
 77        results = self._extract_tool_stats(data, forest)
 78        self._aggregate_per_run(results, forest)
 79        self._aggregate_across_runs(results, forest)
 80
 81        metrics = list(results.keys())
 82
 83        return EvaluatorResult(
 84            evaluator_name=self.name,
 85            evaluator_id=self.identifier,
 86            agent_data_ids=agent_data_ids,
 87            metrics=metrics,
 88            metric_results=[item for sublist in results.values() for item in sublist],
 89            aggregate_results=forest,
 90        )
 91
 92    def _extract_tool_stats(
 93        self,
 94        data: list[AgentDataPoint],
 95        forest: AggregateForest[ToolAggregateNode, ToolMetricResult],
 96    ) -> dict[ToolMetric, list[ToolMetricResult]]:
 97        """
 98        Retrieve tool usage statistics from the agent data points.
 99        There is no aggregation at this level, so results are at the tool call level.
100
101        Args:
102            data: A list of AgentDataPoint instances.
103        """
104
105        results: dict[ToolMetric, list[ToolMetricResult]] = defaultdict(list)
106        # (agent_datapoint_id, tool_name): stats_dict
107        stats: dict[tuple[UUID, str], ToolStats] = defaultdict(
108            lambda: {"usage_count": 0, "failure_count": 0, "runtimes": []}
109        )
110
111        for datapoint in data:
112            for tool in datapoint.tool_details.calls:
113                tool_name = tool.name
114                key = (datapoint.identifier, tool_name)
115
116                stats[key]["usage_count"] += 1
117
118                metric_result = ToolMetricResult(
119                    result_name=f"{METRICS['Runtime'].name}/{tool_name}",
120                    agent_data_id=[datapoint.identifier],
121                    metric_id=METRICS["Runtime"].identifier,
122                    tool_name=tool_name,
123                    tool_node_id=tool.identifier,
124                    value=tool.runtime if tool.runtime is not None else 0.0,
125                )
126                forest.add_node(metric_result)
127                results[METRICS["ToolFailure"]].append(metric_result)
128
129                if tool.status == Status.FAILED:
130                    stats[key]["failure_count"] += 1
131                runtime = tool.runtime
132
133                if runtime is not None:
134                    stats[key]["runtimes"].append(runtime)
135
136                    metric_result = ToolMetricResult(
137                        result_name=f"{METRICS['Runtime'].name}/{tool_name}",
138                        agent_data_id=[datapoint.identifier],
139                        metric_id=METRICS["Runtime"].identifier,
140                        tool_name=tool_name,
141                        tool_node_id=tool.identifier,
142                        value=runtime,
143                    )
144                    forest.add_node(metric_result)
145                    results[METRICS["Runtime"]].append(metric_result)
146
147        for key, tool_data in stats.items():
148            adp_id, tool_name = key
149
150            failure_rate = (
151                tool_data["failure_count"] / tool_data["usage_count"]
152                if tool_data["usage_count"] > 0
153                else 0.0
154            )
155
156            tmr = ToolMetricResult(
157                result_name=f"{METRICS['FailureRate'].name}/{tool_name}",
158                agent_data_id=[adp_id],
159                metric_id=METRICS["FailureRate"].identifier,
160                tool_name=tool_name,
161                tool_node_id=None,
162                value=failure_rate,
163            )
164            forest.add_node(tmr)
165            results[METRICS["FailureRate"]].append(tmr)
166
167            tmr = ToolMetricResult(
168                result_name=f"{METRICS['UsageCount'].name}/{tool_name}",
169                agent_data_id=[adp_id],
170                metric_id=METRICS["UsageCount"].identifier,
171                tool_name=tool_name,
172                tool_node_id=None,
173                value=tool_data["usage_count"],
174            )
175            forest.add_node(tmr)
176            results[METRICS["UsageCount"]].append(tmr)
177
178        return results
179
180    def _aggregate_per_run(
181        self,
182        results: dict[ToolMetric, list[ToolMetricResult]],
183        forest: AggregateForest[ToolAggregateNode, ToolMetricResult],
184    ) -> None:
185        metric_results = results[METRICS["Runtime"]]
186        metric_results_by_adp_id: dict[UUID, list[ToolMetricResult]] = defaultdict(list)
187
188        values: dict[UUID, dict[str, list[ToolMetricResult]]] = defaultdict(dict)
189
190        for result in metric_results:
191            for adp_id in result.agent_data_id:
192                metric_results_by_adp_id[adp_id].append(result)
193
194        for adp_id in metric_results_by_adp_id:
195            values[adp_id] = defaultdict(list)
196
197            for tmr in metric_results_by_adp_id[adp_id]:
198                values[adp_id][tmr.tool_name].append(tmr)
199
200            for tool_name in values[adp_id]:
201                aggregate_node = ToolAggregateNode(
202                    name=f"Aggregate/{METRICS['Runtime'].name}",
203                    metric=METRICS["Runtime"],
204                    tool_name=tool_name,
205                    children=[tmr.identifier for tmr in values[adp_id][tool_name]],
206                    forest=forest,
207                )
208                forest.roots.append(aggregate_node.identifier)
209                forest.add_node(aggregate_node)
210
211    def _aggregate_across_runs(
212        self,
213        results: dict[ToolMetric, list[ToolMetricResult]],
214        forest: AggregateForest[ToolAggregateNode, ToolMetricResult],
215    ) -> None:
216        """
217        Aggregates the ToolUseEvaluator metrics across runs on an agent level.
218        This is a separate step from the initial extraction to allow for more flexible aggregation strategies in the future.
219
220        Args:
221            results: A dictionary of ToolMetric to list of ToolMetricResult at the tool call level.
222
223        Returns:
224            A list of ToolAggregateNode instances containing the aggregated results at the run level.
225        """
226
227        for metric in [METRICS["FailureRate"], METRICS["UsageCount"]]:
228            metric_results = results[metric]
229            values: dict[str, list[ToolMetricResult]] = defaultdict(list)
230
231            for tmr in metric_results:
232                values[tmr.tool_name].append(tmr)
233
234            for tool_name, vals in values.items():
235                aggregate_node = ToolAggregateNode(
236                    name=f"Aggregate/{metric.name}",
237                    metric=metric,
238                    tool_name=tool_name,
239                    children=[val.identifier for val in vals],
240                    forest=forest,
241                )
242                forest.roots.append(aggregate_node.identifier)
243                forest.add_node(aggregate_node)
244
245        ## Aggregation of Runtime ------------------------------
246        tool_breakdown = defaultdict(list)
247        for root_id in forest.roots:
248            agg = forest.get(root_id)
249            if isinstance(agg, ToolMetricResult):
250                raise ValueError(
251                    f"Expected root nodes in the forest to be ToolAggregateNodes, but got {type(agg)}"
252                )
253            if agg.metric == METRICS["Runtime"]:
254                tool_breakdown[agg.tool_name].append(agg)
255
256        for tool_name in tool_breakdown:
257            parent = ToolAggregateNode(
258                name=f"Aggregate/{METRICS['Runtime'].name}",
259                metric=METRICS["Runtime"],
260                tool_name=tool_name,
261                children=[
262                    tool_agg.identifier for tool_agg in tool_breakdown[tool_name]
263                ],
264                forest=forest,
265            )
266            forest.add_node(parent)
267            forest.roots.append(parent.identifier)

 19class LLMInferenceEvaluator(Evaluator):
 20    """
 21    Evaluator that analyzes LLM inference statistics across agent runs.
 22
 23    Computes per-call and aggregated metrics for each LLM invocation,
 24    including input/output token counts, token cost, and latency.
 25    """
 26
 27    def __init__(
 28        self,
 29    ):
 30        super().__init__()
 31
 32    def run(
 33        self, data: list[AgentDataPoint]
 34    ) -> EvaluatorResult[LLMMetric, LLMMetricResult, LLMInferenceAggregateNode]:
 35        """
 36        Run the evaluator over a list of agent data points.
 37
 38        Args:
 39            data: A list of AgentDataPoint instances to evaluate.
 40
 41        Returns:
 42            An EvaluatorResult containing per-call metric results and
 43            aggregated nodes grouped by model and call index.
 44        """
 45        agent_data_ids: set[UUID] = {adp.identifier for adp in data}
 46        forest = AggregateForest[LLMInferenceAggregateNode, LLMMetricResult]()
 47
 48        results = self._retrieve_llm_states(data, forest)
 49        self._aggregate_metrics(results, forest)
 50
 51        metrics = list(results.keys())
 52
 53        return EvaluatorResult(
 54            evaluator_name=self.name,
 55            evaluator_id=self.identifier,
 56            agent_data_ids=agent_data_ids,
 57            metrics=metrics,
 58            metric_results=[item for sublist in results.values() for item in sublist],
 59            aggregate_results=forest,
 60        )
 61
 62    def _retrieve_llm_states(
 63        self,
 64        data: list[AgentDataPoint],
 65        forest: AggregateForest[LLMInferenceAggregateNode, LLMMetricResult],
 66    ) -> dict[LLMMetric, list[LLMMetricResult]]:
 67        results: dict[LLMMetric, list[LLMMetricResult]] = defaultdict(list)
 68
 69        for datapoint in data:
 70            llm_details = datapoint.llm_details
 71
 72            for call in llm_details.calls:
 73                # Input Tokens
 74                metric = LLMMetric(
 75                    name="InputTokens",
 76                    min_value=0,
 77                )
 78
 79                metric_result = LLMMetricResult(
 80                    result_name="InputTokens",
 81                    metric_id=metric.identifier,
 82                    agent_data_id=[datapoint.identifier],
 83                    value=call.input_tokens,
 84                    llm_call_index=call.index,
 85                    model_name=call.model_name,
 86                    model_provider=call.model_provider,
 87                )
 88                results[metric].append(metric_result)
 89                forest.add_node(metric_result)
 90
 91                # Output Tokens
 92                metric = LLMMetric(
 93                    name="OutputTokens",
 94                    min_value=0,
 95                )
 96
 97                metric_result = LLMMetricResult(
 98                    result_name="OutputTokens",
 99                    metric_id=metric.identifier,
100                    agent_data_id=[datapoint.identifier],
101                    value=call.output_tokens,
102                    llm_call_index=call.index,
103                    model_name=call.model_name,
104                    model_provider=call.model_provider,
105                )
106                results[metric].append(metric_result)
107                forest.add_node(metric_result)
108
109                # Total Cost
110                metric = LLMMetric(
111                    name="TokenCost",
112                    min_value=0.0,
113                )
114
115                metric_result = LLMMetricResult(
116                    result_name="TokenCost",
117                    metric_id=metric.identifier,
118                    agent_data_id=[datapoint.identifier],
119                    value=call.total_cost,
120                    llm_call_index=call.index,
121                    model_name=call.model_name,
122                    model_provider=call.model_provider,
123                )
124                results[metric].append(metric_result)
125                forest.add_node(metric_result)
126
127                # Latency
128                metric = LLMMetric(
129                    name="Latency",
130                    min_value=0.0,
131                )
132                metric_result = LLMMetricResult(
133                    result_name="Latency",
134                    metric_id=metric.identifier,
135                    agent_data_id=[datapoint.identifier],
136                    value=call.latency,
137                    llm_call_index=call.index,
138                    model_name=call.model_name,
139                    model_provider=call.model_provider,
140                )
141                results[metric].append(metric_result)
142                forest.add_node(metric_result)
143
144        return results
145
146    def _aggregate_metrics(
147        self,
148        results: dict[LLMMetric, list[LLMMetricResult]],
149        forest: AggregateForest[LLMInferenceAggregateNode, LLMMetricResult],
150    ) -> None:
151        for metric in results:
152            metric_results = results[metric]
153            values: dict[tuple[str, str, int], list[LLMMetricResult]] = defaultdict(
154                list
155            )
156            for mr in metric_results:
157                if isinstance(mr.value, (int, float)):
158                    key = (mr.model_name, mr.model_provider, mr.llm_call_index)
159                    values[key].append(mr)
160
161            for (model_name, model_provider, llm_call_index), vals in values.items():
162                aggregate_node = LLMInferenceAggregateNode(
163                    name=f"Aggregate/{metric.name}/{model_name}/{model_provider}/Call_{llm_call_index}",
164                    metric=metric,
165                    children=[val.identifier for val in vals],
166                    model_name=model_name,
167                    model_provider=model_provider,
168                    llm_call_index=llm_call_index,
169                    forest=forest,
170                )
171
172                forest.roots.append(aggregate_node.identifier)
173                forest.add_node(aggregate_node)