railtracks.evaluations

137def evaluate(
138    data: AgentDataPoint | list[AgentDataPoint],
139    evaluators: list[Evaluator],
140    agent_selection: bool = True,
141    agents: list[str] | None = None,
142    name: str | None = None,
143    payload_callback: Callable[[dict[str, Any]], Any] | None = None,
144):
145    """Evaluate agent data using the provided evaluators.
146
147    Args:
148        data: The agent data to be evaluated. Can be a single AgentDataPoint, a list of AgentDataPoints, or an EvaluationDataset.
149        evaluators: A list of Evaluator instances to run on the data.
150        agent_selection: If True and multiple agents are found in the data, prompts the user to select which agents to evaluate.
151                         If False, evaluates all agents without prompting.
152        agents: An optional list of agent names to evaluate. If provided, only these agents will be evaluated. Overrides agent_selection if both are provided.
153        name: An optional name for the evaluation, which will be included in the EvaluationResult.
154        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.
155    Returns:
156        A list of EvaluationResult instances containing the results from each evaluator.
157    """
158    _check_evaluators(evaluators)
159
160    data_dict, agents = _setup_agent_data(data, agent_selection, agents)
161
162    evaluation_results: list[EvaluationResult] = []
163
164    for agent_name in agents:
165        logger.info(
166            f"Evaluation for {agent_name} with {len(data_dict[agent_name])} data points CREATED"
167        )
168
169        evaluator_results: list[EvaluatorResult] = []
170
171        start_time = datetime.now(timezone.utc)
172        for evaluator in evaluators:
173            logger.info(f"Evaluator: {evaluator.__class__.__name__} CREATED")
174            try:
175                result = evaluator.run(data_dict[agent_name])
176            except Exception as e:
177                logger.error(f"Evaluator {evaluator.__class__.__name__} FAILED: {e}")
178                continue
179
180            evaluator_results.append(result)
181            logger.info(f"Evaluator: {evaluator.__class__.__name__} DONE")
182
183        logger.info(f"Evaluation for {agent_name} DONE.")
184
185        metrics_map = {}
186        for er in evaluator_results:
187            metrics = er.metrics
188            for metric in metrics:
189                metrics_map[metric.identifier] = metric
190
191        end_time = datetime.now(timezone.utc)
192
193        evaluation_results.append(
194            EvaluationResult(
195                evaluation_name=name or None,
196                created_at=start_time,
197                completed_at=end_time,
198                agents=[
199                    {
200                        "agent_name": agent_name,
201                        "agent_node_ids": [
202                            {
203                                "session_id": adp.session_id,
204                                "agent_node_id": adp.identifier,
205                            }
206                            for adp in data_dict[agent_name]
207                        ],
208                    }
209                ],
210                metrics_map=metrics_map,
211                evaluator_results=evaluator_results,
212            )
213        )
214
215    logger.info("Evaluation DONE.")
216
217    if payload_callback is not None:
218        try:
219            for result in evaluation_results:
220                payload_callback(payload(result))
221        except Exception as e:
222            logger.error(f"Failed to execute payload callback: {e}")
223
224    try:
225        save(evaluation_results)
226    except Exception as e:
227        logger.error(f"Failed to save evaluation results: {e}")
228    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] = {}
 57        for m in metrics:
 58            # LLMMetric/ToolMetric subclass Numerical but are usage-stat metrics,
 59            # not meant to be scored by the LLM judge.
 60            if isinstance(m, (LLMMetric, ToolMetric)) or not isinstance(
 61                m, (Categorical, Numerical)
 62            ):
 63                logger.warning(
 64                    f"JudgeEvaluator currently only supports Categorical and Numerical metrics, metric {m.name} of type {type(m)} will be skipped."
 65                )
 66            else:
 67                self._metrics[m.identifier] = m
 68        self._llm = llm
 69        self._reasoning: bool = reasoning
 70        self._template = self._load_yaml()
 71        self._system_prompt = (
 72            system_prompt
 73            if system_prompt is not None
 74            else self._template["system_prompt"]
 75        )
 76        super().__init__()
 77
 78        self.timeout = timeout
 79        self._judge = rt.agent_node(
 80            llm=self._llm,
 81            output_schema=JudgeResponseSchema,
 82        )
 83
 84    def run(
 85        self, data: list[AgentDataPoint]
 86    ) -> EvaluatorResult[Metric, MetricResult, CategoricalAggregateNode]:
 87        judge_outputs: list[JudgeOutput] = self._invoke(data)
 88
 89        self.agent_data_ids = {adp.identifier for adp in data}
 90        results: dict[Metric, list[MetricResult]] = defaultdict(list)
 91        forest = AggregateForest[CategoricalAggregateNode, MetricResult]()
 92
 93        for output in judge_outputs:
 94            metric = self._metrics[output.metric_id]
 95
 96            metric_result = MetricResult(
 97                result_name=f"JudgeResult/{metric.name}",
 98                metric_id=metric.identifier,
 99                agent_data_id=[UUID(output.adp_id)],
100                value=output.response.metric_value,
101            )
102            results[metric].append(metric_result)
103            forest.add_node(metric_result)
104
105            if self._reasoning:
106                reasoning_metric = Metric(name=f"{metric.name}_reasoning")
107                if output.response.reasoning is not None:
108                    results[reasoning_metric].append(
109                        MetricResult(
110                            result_name=f"JudgeReasoning/{metric.name}",
111                            metric_id=reasoning_metric.identifier,
112                            agent_data_id=[UUID(output.adp_id)],
113                            value=output.response.reasoning,
114                        )
115                    )
116                else:
117                    logger.warning(
118                        f"No reasoning returned for Judge Evaluator Metric: {metric.name}, AgentDataPoint ID: {output.adp_id}"
119                    )
120
121        self._aggregate_metrics(results, forest)
122
123        self._result = EvaluatorResult(
124            evaluator_name=self.name,
125            evaluator_id=self.identifier,
126            agent_data_ids=self.agent_data_ids,
127            metric_results=[item for sublist in results.values() for item in sublist],
128            aggregate_results=forest,
129            metrics=list(self._metrics.values()),
130        )
131        return self._result
132
133    def __repr__(self) -> str:
134        return (
135            f"JudgeEvaluator, "
136            f"llm={self._llm}, "
137            f"metrics={list(self._metrics.values())}, "
138            f"reasoning={self._reasoning})"
139        )
140
141    def _invoke(self, data: list[AgentDataPoint]) -> list[JudgeOutput]:
142        @rt.function_node
143        async def judge_flow():
144            output: list[JudgeOutput] = []
145            for metric in self._metrics.values():
146                logger.info(
147                    f"START Evaluating Metric: {metric.name} for {len(data)} AgentDataPoints"
148                )
149
150                for idx, adp in enumerate(data):
151                    user_message = self._generate_user_prompt(adp)
152                    system_message = self._generate_system_prompt(metric)
153                    message_history = rt.llm.MessageHistory(
154                        [
155                            rt.llm.SystemMessage(system_message),
156                            rt.llm.UserMessage(user_message),
157                        ]
158                    )
159                    res = await rt.call(
160                        self._judge,
161                        message_history,
162                    )
163                    output.append(
164                        JudgeOutput(
165                            metric_id=metric.identifier,
166                            adp_id=str(adp.identifier),
167                            response=res.structured,
168                        )
169                    )
170
171                    logger.info(
172                        f"AgentDataPoint ID: {adp.identifier} {idx + 1}/{len(data)} DONE"
173                    )
174
175            return output
176
177        judge_evaluator_flow = rt.Flow(
178            name="JudgeEvaluatorFlow",
179            entry_point=judge_flow,
180            timeout=self.timeout,
181            save_state=False,
182        )
183
184        return judge_evaluator_flow.invoke()
185
186    def _aggregate_metrics(
187        self,
188        results: dict[Metric, list[MetricResult]],
189        forest: AggregateForest[CategoricalAggregateNode, MetricResult],
190    ) -> None:
191        for metric in results:
192            if isinstance(metric, Numerical):
193                continue
194            elif isinstance(metric, Categorical):
195                aggregate_node = CategoricalAggregateNode(
196                    name=f"Aggregate/{metric.name}",
197                    metric=metric,
198                    children=[val.identifier for val in results[metric]],
199                    forest=forest,
200                )
201
202                forest.roots.append(aggregate_node.identifier)
203                forest.add_node(aggregate_node)
204
205    def _generate_user_prompt(self, data: AgentDataPoint) -> str:
206        return self._template["user"].format(
207            agent_input=data.agent_input,
208            agent_output=data.agent_output.get("message_history", ""),
209        )
210
211    def _generate_system_prompt(self, metric: Metric) -> str:
212        system_prompt: str = self._template["system_prompt"]
213
214        system_prompt += "\n" + self._template["metric"].format(metric=str(metric))
215
216        if isinstance(metric, Categorical):
217            category_names = ", ".join(c for c in metric.category_names)
218            system_prompt += (
219                f"\nYour metric_value must be exactly one of these category "
220                f"names: {category_names}."
221            )
222        elif isinstance(metric, Numerical):
223            system_prompt += f"\n{self._numeric_bounds_text(metric)}"
224            if metric.shots:
225                system_prompt += self._numeric_shots_text(metric)
226
227        if self._reasoning:
228            system_prompt += self._template["reasoning"]
229
230        return system_prompt
231
232    def _numeric_bounds_text(self, metric: Numerical) -> str:
233        if metric.min_value is not None and metric.max_value is not None:
234            return (
235                f"Your metric_value must be a single number between "
236                f"{metric.min_value} and {metric.max_value} inclusive."
237            )
238        if metric.min_value is not None:
239            return (
240                f"Your metric_value must be a single number of at least "
241                f"{metric.min_value}."
242            )
243        if metric.max_value is not None:
244            return (
245                f"Your metric_value must be a single number of at most "
246                f"{metric.max_value}."
247            )
248        return "Your metric_value must be a single number."
249
250    def _numeric_shots_text(self, metric: Numerical) -> str:
251        text = "\nUse the following anchor points to calibrate your scoring:"
252        for value, description in metric.shots or []:
253            text += f"\n- A score of {value} means: {description}"
254        text += (
255            "\nFor scores between the provided anchor points, "
256            "interpolate based on how closely the agent's output "
257            "matches the descriptions of the nearest anchors."
258        )
259        return text
260
261    def _load_yaml(self):
262        yaml_path = Path(__file__).parent / "judge_evaluator.yaml"
263        with open(yaml_path, "r") as f:
264            template = yaml.safe_load(f)
265
266        return template
267
268    def _get_config(self) -> dict:
269        return {
270            "llm": self._llm.model_name(),
271            "llm_provider": self._llm.model_provider(),
272            "system_prompt": self._system_prompt,
273            "metrics": sorted(self._metrics.keys()),
274            "reasoning": self._reasoning,
275        }

 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                failed = tool.status == Status.FAILED
119                if failed:
120                    stats[key]["failure_count"] += 1
121
122                # not added to the forest, nothing aggregates per-call failures
123                results[METRICS["ToolFailure"]].append(
124                    ToolMetricResult(
125                        result_name=f"{METRICS['ToolFailure'].name}/{tool_name}",
126                        agent_data_id=[datapoint.identifier],
127                        metric_id=METRICS["ToolFailure"].identifier,
128                        tool_name=tool_name,
129                        tool_node_id=tool.identifier,
130                        value=1 if failed else 0,
131                    )
132                )
133
134                runtime = tool.runtime
135
136                if runtime is not None:
137                    stats[key]["runtimes"].append(runtime)
138
139                    metric_result = ToolMetricResult(
140                        result_name=f"{METRICS['Runtime'].name}/{tool_name}",
141                        agent_data_id=[datapoint.identifier],
142                        metric_id=METRICS["Runtime"].identifier,
143                        tool_name=tool_name,
144                        tool_node_id=tool.identifier,
145                        value=runtime,
146                    )
147                    forest.add_node(metric_result)
148                    results[METRICS["Runtime"]].append(metric_result)
149
150        for key, tool_data in stats.items():
151            adp_id, tool_name = key
152
153            failure_rate = (
154                tool_data["failure_count"] / tool_data["usage_count"]
155                if tool_data["usage_count"] > 0
156                else 0.0
157            )
158
159            tmr = ToolMetricResult(
160                result_name=f"{METRICS['FailureRate'].name}/{tool_name}",
161                agent_data_id=[adp_id],
162                metric_id=METRICS["FailureRate"].identifier,
163                tool_name=tool_name,
164                tool_node_id=None,
165                value=failure_rate,
166            )
167            forest.add_node(tmr)
168            results[METRICS["FailureRate"]].append(tmr)
169
170            tmr = ToolMetricResult(
171                result_name=f"{METRICS['UsageCount'].name}/{tool_name}",
172                agent_data_id=[adp_id],
173                metric_id=METRICS["UsageCount"].identifier,
174                tool_name=tool_name,
175                tool_node_id=None,
176                value=tool_data["usage_count"],
177            )
178            forest.add_node(tmr)
179            results[METRICS["UsageCount"]].append(tmr)
180
181        return results
182
183    def _aggregate_per_run(
184        self,
185        results: dict[ToolMetric, list[ToolMetricResult]],
186        forest: AggregateForest[ToolAggregateNode, ToolMetricResult],
187    ) -> None:
188        metric_results = results[METRICS["Runtime"]]
189        metric_results_by_adp_id: dict[UUID, list[ToolMetricResult]] = defaultdict(list)
190
191        values: dict[UUID, dict[str, list[ToolMetricResult]]] = defaultdict(dict)
192
193        for result in metric_results:
194            for adp_id in result.agent_data_id:
195                metric_results_by_adp_id[adp_id].append(result)
196
197        for adp_id in metric_results_by_adp_id:
198            values[adp_id] = defaultdict(list)
199
200            for tmr in metric_results_by_adp_id[adp_id]:
201                values[adp_id][tmr.tool_name].append(tmr)
202
203            for tool_name in values[adp_id]:
204                aggregate_node = ToolAggregateNode(
205                    name=f"Aggregate/{METRICS['Runtime'].name}",
206                    metric=METRICS["Runtime"],
207                    tool_name=tool_name,
208                    children=[tmr.identifier for tmr in values[adp_id][tool_name]],
209                    forest=forest,
210                )
211                forest.roots.append(aggregate_node.identifier)
212                forest.add_node(aggregate_node)
213
214    def _aggregate_across_runs(
215        self,
216        results: dict[ToolMetric, list[ToolMetricResult]],
217        forest: AggregateForest[ToolAggregateNode, ToolMetricResult],
218    ) -> None:
219        """
220        Aggregates the ToolUseEvaluator metrics across runs on an agent level.
221        This is a separate step from the initial extraction to allow for more flexible aggregation strategies in the future.
222
223        Args:
224            results: A dictionary of ToolMetric to list of ToolMetricResult at the tool call level.
225
226        Returns:
227            A list of ToolAggregateNode instances containing the aggregated results at the run level.
228        """
229
230        for metric in [METRICS["FailureRate"], METRICS["UsageCount"]]:
231            metric_results = results[metric]
232            values: dict[str, list[ToolMetricResult]] = defaultdict(list)
233
234            for tmr in metric_results:
235                values[tmr.tool_name].append(tmr)
236
237            for tool_name, vals in values.items():
238                aggregate_node = ToolAggregateNode(
239                    name=f"Aggregate/{metric.name}",
240                    metric=metric,
241                    tool_name=tool_name,
242                    children=[val.identifier for val in vals],
243                    forest=forest,
244                )
245                forest.roots.append(aggregate_node.identifier)
246                forest.add_node(aggregate_node)
247
248        ## Aggregation of Runtime ------------------------------
249        tool_breakdown = defaultdict(list)
250        for root_id in forest.roots:
251            agg = forest.get(root_id)
252            if isinstance(agg, ToolMetricResult):
253                raise ValueError(
254                    f"Expected root nodes in the forest to be ToolAggregateNodes, but got {type(agg)}"
255                )
256            if agg.metric == METRICS["Runtime"]:
257                tool_breakdown[agg.tool_name].append(agg)
258
259        for tool_name in tool_breakdown:
260            parent = ToolAggregateNode(
261                name=f"Aggregate/{METRICS['Runtime'].name}",
262                metric=METRICS["Runtime"],
263                tool_name=tool_name,
264                children=[
265                    tool_agg.identifier for tool_agg in tool_breakdown[tool_name]
266                ],
267                forest=forest,
268            )
269            forest.add_node(parent)
270            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)