Skip to content

Integrations

Connectors that expose RAGAS and DeepEval metrics through UAEF's metric interface. The metric names they add are listed in the Metrics Catalog.

uaef.integrations.registry

External metric connector registry and factory for managing external library integrations.

ExternalMetricConnector

Base interface for external metric connectors.

All external metric connectors should implement this interface to ensure consistent behavior and integration with UAEF.

Source code in uaef/integrations/registry.py 
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
class ExternalMetricConnector:
    """
    Base interface for external metric connectors.

    All external metric connectors should implement this interface to ensure
    consistent behavior and integration with UAEF.
    """

    def is_available(self) -> bool:
        """
        Check if external library is available.

        Returns:
            True if external library is installed and available, False otherwise
        """
        raise NotImplementedError("Subclasses must implement is_available()")

    def calculate_metrics(
        self,
        evaluation_input: EvaluationInput,
        metric_names: List[str],
        **kwargs: Any
    ) -> List[MetricScore]:
        """
        Calculate multiple metrics and return UAEF MetricScore objects.

        Args:
            evaluation_input: UAEF evaluation input
            metric_names: List of metric names to calculate
            **kwargs: Additional connector-specific parameters

        Returns:
            List of MetricScore objects

        Raises:
            RuntimeError: If external library is not available
        """
        raise NotImplementedError("Subclasses must implement calculate_metrics()")

    def list_available_metrics(self) -> List[str]:
        """
        List all metrics supported by this connector.

        Returns:
            List of metric names
        """
        raise NotImplementedError("Subclasses must implement list_available_metrics()")

calculate_metrics(evaluation_input, metric_names, **kwargs)

Calculate multiple metrics and return UAEF MetricScore objects.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required
metric_names List[str]

List of metric names to calculate

 required
**kwargs Any

Additional connector-specific parameters

 {}

Returns:

Type Description
List[MetricScore]

List of MetricScore objects

Raises:

Type Description
RuntimeError

If external library is not available
Source code in uaef/integrations/registry.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def calculate_metrics(
    self,
    evaluation_input: EvaluationInput,
    metric_names: List[str],
    **kwargs: Any
) -> List[MetricScore]:
    """
    Calculate multiple metrics and return UAEF MetricScore objects.

    Args:
        evaluation_input: UAEF evaluation input
        metric_names: List of metric names to calculate
        **kwargs: Additional connector-specific parameters

    Returns:
        List of MetricScore objects

    Raises:
        RuntimeError: If external library is not available
    """
    raise NotImplementedError("Subclasses must implement calculate_metrics()")

is_available()

Check if external library is available.

Returns:

Type Description
bool

True if external library is installed and available, False otherwise
Source code in uaef/integrations/registry.py 
23
24
25
26
27
28
29
30
def is_available(self) -> bool:
    """
    Check if external library is available.

    Returns:
        True if external library is installed and available, False otherwise
    """
    raise NotImplementedError("Subclasses must implement is_available()")

list_available_metrics()

List all metrics supported by this connector.

Returns:

Type Description
List[str]

List of metric names
Source code in uaef/integrations/registry.py 
54
55
56
57
58
59
60
61
def list_available_metrics(self) -> List[str]:
    """
    List all metrics supported by this connector.

    Returns:
        List of metric names
    """
    raise NotImplementedError("Subclasses must implement list_available_metrics()")

ExternalMetricRegistry

Registry for managing external metric library connectors.

This registry allows: - Registration of built-in connectors (RAGAS, DeepEval) - Registration of custom connectors - Retrieval of connectors by library name - Discovery of available connectors and their metrics - Graceful handling of unavailable libraries

Source code in uaef/integrations/registry.py 
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
class ExternalMetricRegistry:
    """
    Registry for managing external metric library connectors.

    This registry allows:
    - Registration of built-in connectors (RAGAS, DeepEval)
    - Registration of custom connectors
    - Retrieval of connectors by library name
    - Discovery of available connectors and their metrics
    - Graceful handling of unavailable libraries
    """

    def __init__(self):
        """Initialize the external metric registry."""
        self._connectors: Dict[str, Type[ExternalMetricConnector]] = {}
        self._connector_instances: Dict[str, ExternalMetricConnector] = {}
        self._register_builtin_connectors()

    def _register_builtin_connectors(self) -> None:
        """Register built-in external metric connectors."""
        # Register RAGAS connector
        try:
            from uaef.integrations.ragas_connector import RAGASConnector
            self.register("ragas", RAGASConnector)
            logger.info("Registered RAGAS connector")
        except ImportError as e:
            logger.debug(f"RAGAS connector not available: {e}")

        # Register DeepEval connector
        try:
            from uaef.integrations.deepeval_connector import DeepEvalConnector
            self.register("deepeval", DeepEvalConnector)
            logger.info("Registered DeepEval connector")
        except ImportError as e:
            logger.debug(f"DeepEval connector not available: {e}")

    def register(
        self,
        name: str,
        connector_class: Type[ExternalMetricConnector],
        override: bool = False
    ) -> None:
        """
        Register an external metric connector class.

        Args:
            name: Library name (e.g., "ragas", "deepeval")
            connector_class: Connector class that implements ExternalMetricConnector interface
            override: If True, allow overriding existing connectors

        Raises:
            ValueError: If connector is already registered and override is False
            TypeError: If connector_class does not implement required methods
        """
        # Validate connector class has required methods
        required_methods = ["is_available", "calculate_metrics", "list_available_metrics"]
        for method in required_methods:
            if not hasattr(connector_class, method):
                raise TypeError(
                    f"Connector class {connector_class.__name__} must implement {method}() method"
                )

        # Check for existing registration
        if name in self._connectors and not override:
            raise ValueError(
                f"Connector '{name}' is already registered. "
                f"Use override=True to replace it."
            )

        # Register the connector
        self._connectors[name] = connector_class
        logger.info(f"Registered external metric connector: {name}")

        # Clear cached instance if it exists
        if name in self._connector_instances:
            del self._connector_instances[name]

    def get_connector(
        self,
        name: str,
        connector_config: Optional[Dict[str, Any]] = None
    ) -> ExternalMetricConnector:
        """
        Get a connector instance by library name.

        Args:
            name: Library name (e.g., "ragas", "deepeval")
            connector_config: Optional configuration for connector initialization

        Returns:
            Connector instance

        Raises:
            ValueError: If connector is not registered
        """
        if name not in self._connectors:
            raise ValueError(
                f"Connector '{name}' is not registered. "
                f"Available connectors: {', '.join(self.list_connectors())}"
            )

        # Check if we need to create a new instance (for connectors with config)
        if connector_config:
            connector_class = self._connectors[name]
            return connector_class(**connector_config)

        # Return cached instance or create new one
        if name not in self._connector_instances:
            connector_class = self._connectors[name]
            self._connector_instances[name] = connector_class()

        return self._connector_instances[name]

    def list_connectors(self) -> List[str]:
        """
        List all registered connector names.

        Returns:
            List of connector names
        """
        return list(self._connectors.keys())

    def list_available_connectors(self) -> List[str]:
        """
        List connectors that are available (library is installed).

        Returns:
            List of available connector names
        """
        available = []
        for name in self._connectors.keys():
            try:
                connector = self.get_connector(name)
                if connector.is_available():
                    available.append(name)
            except Exception as e:
                logger.debug(f"Error checking availability of {name}: {e}")
        return available

    def list_all_metrics(self) -> Dict[str, List[str]]:
        """
        List all metrics from all available connectors.

        Returns:
            Dictionary mapping connector name to list of metric names
        """
        all_metrics = {}
        for name in self.list_available_connectors():
            try:
                connector = self.get_connector(name)
                metrics = connector.list_available_metrics()
                all_metrics[name] = metrics
            except Exception as e:
                logger.warning(f"Error listing metrics for {name}: {e}")
        return all_metrics

    def is_registered(self, name: str) -> bool:
        """
        Check if a connector is registered.

        Args:
            name: Library name

        Returns:
            True if connector is registered, False otherwise
        """
        return name in self._connectors

    def is_available(self, name: str) -> bool:
        """
        Check if a connector is registered and its library is available.

        Args:
            name: Library name

        Returns:
            True if connector is registered and library is available, False otherwise
        """
        if not self.is_registered(name):
            return False

        try:
            connector = self.get_connector(name)
            return connector.is_available()
        except Exception as e:
            logger.debug(f"Error checking availability of {name}: {e}")
            return False

    def unregister(self, name: str) -> None:
        """
        Unregister a connector.

        Args:
            name: Library name

        Raises:
            ValueError: If connector is not registered
        """
        if name not in self._connectors:
            raise ValueError(f"Connector '{name}' is not registered")

        del self._connectors[name]
        logger.info(f"Unregistered external metric connector: {name}")

        if name in self._connector_instances:
            del self._connector_instances[name]

    def calculate_metrics(
        self,
        connector_name: str,
        evaluation_input: EvaluationInput,
        metric_names: List[str],
        **kwargs: Any
    ) -> List[MetricScore]:
        """
        Calculate metrics using a specific connector.

        This is a convenience method that gets the connector and calls its
        calculate_metrics method. Handles errors gracefully per requirement 14.6.

        Args:
            connector_name: Name of the connector to use
            evaluation_input: UAEF evaluation input
            metric_names: List of metric names to calculate
            **kwargs: Additional connector-specific parameters

        Returns:
            List of MetricScore objects (may be empty if connector unavailable)
        """
        try:
            # Check if connector is registered
            if not self.is_registered(connector_name):
                logger.warning(
                    f"Connector '{connector_name}' is not registered. "
                    f"Available: {', '.join(self.list_connectors())}"
                )
                return []

            # Get connector instance
            connector = self.get_connector(connector_name)

            # Check if library is available
            if not connector.is_available():
                logger.warning(
                    f"Connector '{connector_name}' library is not available. "
                    f"Skipping metrics: {', '.join(metric_names)}"
                )
                return []

            # Calculate metrics
            scores = connector.calculate_metrics(
                evaluation_input,
                metric_names,
                **kwargs
            )

            logger.info(
                f"Calculated {len(scores)} metrics using {connector_name} connector"
            )
            return scores

        except Exception as e:
            # Handle errors gracefully per requirement 14.6
            logger.error(
                f"Error calculating metrics with {connector_name} connector: {e}",
                exc_info=True
            )
            return []

calculate_metrics(connector_name, evaluation_input, metric_names, **kwargs)

Calculate metrics using a specific connector.

This is a convenience method that gets the connector and calls its calculate_metrics method. Handles errors gracefully per requirement 14.6.

Parameters:

Name Type Description Default
connector_name str

Name of the connector to use

 required
evaluation_input EvaluationInput

UAEF evaluation input

 required
metric_names List[str]

List of metric names to calculate

 required
**kwargs Any

Additional connector-specific parameters

 {}

Returns:

Type Description
List[MetricScore]

List of MetricScore objects (may be empty if connector unavailable)
Source code in uaef/integrations/registry.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def calculate_metrics(
    self,
    connector_name: str,
    evaluation_input: EvaluationInput,
    metric_names: List[str],
    **kwargs: Any
) -> List[MetricScore]:
    """
    Calculate metrics using a specific connector.

    This is a convenience method that gets the connector and calls its
    calculate_metrics method. Handles errors gracefully per requirement 14.6.

    Args:
        connector_name: Name of the connector to use
        evaluation_input: UAEF evaluation input
        metric_names: List of metric names to calculate
        **kwargs: Additional connector-specific parameters

    Returns:
        List of MetricScore objects (may be empty if connector unavailable)
    """
    try:
        # Check if connector is registered
        if not self.is_registered(connector_name):
            logger.warning(
                f"Connector '{connector_name}' is not registered. "
                f"Available: {', '.join(self.list_connectors())}"
            )
            return []

        # Get connector instance
        connector = self.get_connector(connector_name)

        # Check if library is available
        if not connector.is_available():
            logger.warning(
                f"Connector '{connector_name}' library is not available. "
                f"Skipping metrics: {', '.join(metric_names)}"
            )
            return []

        # Calculate metrics
        scores = connector.calculate_metrics(
            evaluation_input,
            metric_names,
            **kwargs
        )

        logger.info(
            f"Calculated {len(scores)} metrics using {connector_name} connector"
        )
        return scores

    except Exception as e:
        # Handle errors gracefully per requirement 14.6
        logger.error(
            f"Error calculating metrics with {connector_name} connector: {e}",
            exc_info=True
        )
        return []

get_connector(name, connector_config=None)

Get a connector instance by library name.

Parameters:

Name Type Description Default
name str

Library name (e.g., "ragas", "deepeval")

 required
connector_config Optional[Dict[str, Any]]

Optional configuration for connector initialization

 None

Returns:

Type Description
ExternalMetricConnector

Connector instance

Raises:

Type Description
ValueError

If connector is not registered
Source code in uaef/integrations/registry.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def get_connector(
    self,
    name: str,
    connector_config: Optional[Dict[str, Any]] = None
) -> ExternalMetricConnector:
    """
    Get a connector instance by library name.

    Args:
        name: Library name (e.g., "ragas", "deepeval")
        connector_config: Optional configuration for connector initialization

    Returns:
        Connector instance

    Raises:
        ValueError: If connector is not registered
    """
    if name not in self._connectors:
        raise ValueError(
            f"Connector '{name}' is not registered. "
            f"Available connectors: {', '.join(self.list_connectors())}"
        )

    # Check if we need to create a new instance (for connectors with config)
    if connector_config:
        connector_class = self._connectors[name]
        return connector_class(**connector_config)

    # Return cached instance or create new one
    if name not in self._connector_instances:
        connector_class = self._connectors[name]
        self._connector_instances[name] = connector_class()

    return self._connector_instances[name]

is_available(name)

Check if a connector is registered and its library is available.

Parameters:

Name Type Description Default
name str

Library name

 required

Returns:

Type Description
bool

True if connector is registered and library is available, False otherwise
Source code in uaef/integrations/registry.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def is_available(self, name: str) -> bool:
    """
    Check if a connector is registered and its library is available.

    Args:
        name: Library name

    Returns:
        True if connector is registered and library is available, False otherwise
    """
    if not self.is_registered(name):
        return False

    try:
        connector = self.get_connector(name)
        return connector.is_available()
    except Exception as e:
        logger.debug(f"Error checking availability of {name}: {e}")
        return False

is_registered(name)

Check if a connector is registered.

Parameters:

Name Type Description Default
name str

Library name

 required

Returns:

Type Description
bool

True if connector is registered, False otherwise
Source code in uaef/integrations/registry.py 
220
221
222
223
224
225
226
227
228
229
230
def is_registered(self, name: str) -> bool:
    """
    Check if a connector is registered.

    Args:
        name: Library name

    Returns:
        True if connector is registered, False otherwise
    """
    return name in self._connectors

list_all_metrics()

List all metrics from all available connectors.

Returns:

Type Description
Dict[str, List[str]]

Dictionary mapping connector name to list of metric names
Source code in uaef/integrations/registry.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def list_all_metrics(self) -> Dict[str, List[str]]:
    """
    List all metrics from all available connectors.

    Returns:
        Dictionary mapping connector name to list of metric names
    """
    all_metrics = {}
    for name in self.list_available_connectors():
        try:
            connector = self.get_connector(name)
            metrics = connector.list_available_metrics()
            all_metrics[name] = metrics
        except Exception as e:
            logger.warning(f"Error listing metrics for {name}: {e}")
    return all_metrics

list_available_connectors()

List connectors that are available (library is installed).

Returns:

Type Description
List[str]

List of available connector names
Source code in uaef/integrations/registry.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def list_available_connectors(self) -> List[str]:
    """
    List connectors that are available (library is installed).

    Returns:
        List of available connector names
    """
    available = []
    for name in self._connectors.keys():
        try:
            connector = self.get_connector(name)
            if connector.is_available():
                available.append(name)
        except Exception as e:
            logger.debug(f"Error checking availability of {name}: {e}")
    return available

list_connectors()

List all registered connector names.

Returns:

Type Description
List[str]

List of connector names
Source code in uaef/integrations/registry.py 
177
178
179
180
181
182
183
184
def list_connectors(self) -> List[str]:
    """
    List all registered connector names.

    Returns:
        List of connector names
    """
    return list(self._connectors.keys())

register(name, connector_class, override=False)

Register an external metric connector class.

Parameters:

Name Type Description Default
name str

Library name (e.g., "ragas", "deepeval")

 required
connector_class Type[ExternalMetricConnector]

Connector class that implements ExternalMetricConnector interface

 required
override bool

If True, allow overriding existing connectors

 False

Raises:

Type Description
ValueError

If connector is already registered and override is False
TypeError

If connector_class does not implement required methods
Source code in uaef/integrations/registry.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def register(
    self,
    name: str,
    connector_class: Type[ExternalMetricConnector],
    override: bool = False
) -> None:
    """
    Register an external metric connector class.

    Args:
        name: Library name (e.g., "ragas", "deepeval")
        connector_class: Connector class that implements ExternalMetricConnector interface
        override: If True, allow overriding existing connectors

    Raises:
        ValueError: If connector is already registered and override is False
        TypeError: If connector_class does not implement required methods
    """
    # Validate connector class has required methods
    required_methods = ["is_available", "calculate_metrics", "list_available_metrics"]
    for method in required_methods:
        if not hasattr(connector_class, method):
            raise TypeError(
                f"Connector class {connector_class.__name__} must implement {method}() method"
            )

    # Check for existing registration
    if name in self._connectors and not override:
        raise ValueError(
            f"Connector '{name}' is already registered. "
            f"Use override=True to replace it."
        )

    # Register the connector
    self._connectors[name] = connector_class
    logger.info(f"Registered external metric connector: {name}")

    # Clear cached instance if it exists
    if name in self._connector_instances:
        del self._connector_instances[name]

unregister(name)

Unregister a connector.

Parameters:

Name Type Description Default
name str

Library name

 required

Raises:

Type Description
ValueError

If connector is not registered
Source code in uaef/integrations/registry.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def unregister(self, name: str) -> None:
    """
    Unregister a connector.

    Args:
        name: Library name

    Raises:
        ValueError: If connector is not registered
    """
    if name not in self._connectors:
        raise ValueError(f"Connector '{name}' is not registered")

    del self._connectors[name]
    logger.info(f"Unregistered external metric connector: {name}")

    if name in self._connector_instances:
        del self._connector_instances[name]

calculate_external_metrics(connector_name, evaluation_input, metric_names, **kwargs)

Calculate metrics using a specific connector from the global registry.

This is a convenience function that handles errors gracefully per requirement 14.6.

Parameters:

Name Type Description Default
connector_name str

Name of the connector to use

 required
evaluation_input EvaluationInput

UAEF evaluation input

 required
metric_names List[str]

List of metric names to calculate

 required
**kwargs Any

Additional connector-specific parameters

 {}

Returns:

Type Description
List[MetricScore]

List of MetricScore objects (may be empty if connector unavailable)
Source code in uaef/integrations/registry.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def calculate_external_metrics(
    connector_name: str,
    evaluation_input: EvaluationInput,
    metric_names: List[str],
    **kwargs: Any
) -> List[MetricScore]:
    """
    Calculate metrics using a specific connector from the global registry.

    This is a convenience function that handles errors gracefully per requirement 14.6.

    Args:
        connector_name: Name of the connector to use
        evaluation_input: UAEF evaluation input
        metric_names: List of metric names to calculate
        **kwargs: Additional connector-specific parameters

    Returns:
        List of MetricScore objects (may be empty if connector unavailable)
    """
    return _global_registry.calculate_metrics(
        connector_name,
        evaluation_input,
        metric_names,
        **kwargs
    )

get_connector(name, connector_config=None)

Get a connector instance from the global registry.

Parameters:

Name Type Description Default
name str

Library name (e.g., "ragas", "deepeval")

 required
connector_config Optional[Dict[str, Any]]

Optional configuration for connector initialization

 None

Returns:

Type Description
ExternalMetricConnector

Connector instance

Raises:

Type Description
ValueError

If connector is not registered
Source code in uaef/integrations/registry.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
def get_connector(
    name: str,
    connector_config: Optional[Dict[str, Any]] = None
) -> ExternalMetricConnector:
    """
    Get a connector instance from the global registry.

    Args:
        name: Library name (e.g., "ragas", "deepeval")
        connector_config: Optional configuration for connector initialization

    Returns:
        Connector instance

    Raises:
        ValueError: If connector is not registered
    """
    return _global_registry.get_connector(name, connector_config)

get_registry()

Get the global external metric registry instance.

Returns:

Type Description
ExternalMetricRegistry

Global ExternalMetricRegistry instance
Source code in uaef/integrations/registry.py 
462
463
464
465
466
467
468
469
def get_registry() -> ExternalMetricRegistry:
    """
    Get the global external metric registry instance.

    Returns:
        Global ExternalMetricRegistry instance
    """
    return _global_registry

is_connector_available(name)

Check if a connector is registered and its library is available in the global registry.

Parameters:

Name Type Description Default
name str

Library name

 required

Returns:

Type Description
bool

True if connector is registered and library is available, False otherwise
Source code in uaef/integrations/registry.py 
421
422
423
424
425
426
427
428
429
430
431
def is_connector_available(name: str) -> bool:
    """
    Check if a connector is registered and its library is available in the global registry.

    Args:
        name: Library name

    Returns:
        True if connector is registered and library is available, False otherwise
    """
    return _global_registry.is_available(name)

is_connector_registered(name)

Check if a connector is registered in the global registry.

Parameters:

Name Type Description Default
name str

Library name

 required

Returns:

Type Description
bool

True if connector is registered, False otherwise
Source code in uaef/integrations/registry.py 
408
409
410
411
412
413
414
415
416
417
418
def is_connector_registered(name: str) -> bool:
    """
    Check if a connector is registered in the global registry.

    Args:
        name: Library name

    Returns:
        True if connector is registered, False otherwise
    """
    return _global_registry.is_registered(name)

list_all_metrics()

List all metrics from all available connectors in the global registry.

Returns:

Type Description
Dict[str, List[str]]

Dictionary mapping connector name to list of metric names
Source code in uaef/integrations/registry.py 
398
399
400
401
402
403
404
405
def list_all_metrics() -> Dict[str, List[str]]:
    """
    List all metrics from all available connectors in the global registry.

    Returns:
        Dictionary mapping connector name to list of metric names
    """
    return _global_registry.list_all_metrics()

list_available_connectors()

List connectors that are available (library is installed) in the global registry.

Returns:

Type Description
List[str]

List of available connector names
Source code in uaef/integrations/registry.py 
388
389
390
391
392
393
394
395
def list_available_connectors() -> List[str]:
    """
    List connectors that are available (library is installed) in the global registry.

    Returns:
        List of available connector names
    """
    return _global_registry.list_available_connectors()

list_connectors()

List all registered connector names in the global registry.

Returns:

Type Description
List[str]

List of connector names
Source code in uaef/integrations/registry.py 
378
379
380
381
382
383
384
385
def list_connectors() -> List[str]:
    """
    List all registered connector names in the global registry.

    Returns:
        List of connector names
    """
    return _global_registry.list_connectors()

register_connector(name, connector_class, override=False)

Register a connector in the global registry.

Parameters:

Name Type Description Default
name str

Library name (e.g., "ragas", "deepeval")

 required
connector_class Type[ExternalMetricConnector]

Connector class that implements ExternalMetricConnector interface

 required
override bool

If True, allow overriding existing connectors

 False

Raises:

Type Description
ValueError

If connector is already registered and override is False
TypeError

If connector_class does not implement required methods
Source code in uaef/integrations/registry.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def register_connector(
    name: str,
    connector_class: Type[ExternalMetricConnector],
    override: bool = False
) -> None:
    """
    Register a connector in the global registry.

    Args:
        name: Library name (e.g., "ragas", "deepeval")
        connector_class: Connector class that implements ExternalMetricConnector interface
        override: If True, allow overriding existing connectors

    Raises:
        ValueError: If connector is already registered and override is False
        TypeError: If connector_class does not implement required methods
    """
    _global_registry.register(name, connector_class, override)

uaef.integrations.ragas_connector

RAGAS connector for external metric integration.

RAGASConnector

Connector for RAGAS (Retrieval Augmented Generation Assessment) library.

This connector wraps RAGAS library metrics in UAEF's interface, transforming UAEF data to RAGAS format, calling RAGAS metrics, and transforming results back to UAEF MetricScore objects.

RAGAS is an external evaluation library for RAG systems that provides metrics for answer quality and tool calling accuracy.

Source code in uaef/integrations/ragas_connector.py 
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
class RAGASConnector:
    """
    Connector for RAGAS (Retrieval Augmented Generation Assessment) library.

    This connector wraps RAGAS library metrics in UAEF's interface, transforming
    UAEF data to RAGAS format, calling RAGAS metrics, and transforming results
    back to UAEF MetricScore objects.

    RAGAS is an external evaluation library for RAG systems that provides metrics
    for answer quality and tool calling accuracy.
    """

    def __init__(self, llm=None, embeddings=None):
        """Initialize RAGAS connector.

        Args:
            llm: LangChain LLM to use as judge. If None, auto-creates a ChatBedrock
                 instance using UAEF config and environment AWS credentials.
            embeddings: LangChain embeddings model. If None, auto-creates a
                        BedrockEmbeddings instance using UAEF config.
        """
        self._ragas_available = False
        self._llm = llm
        self._embeddings = embeddings
        try:
            # The RAGAS metric path needs both ``ragas`` and ``datasets``
            # (both provided by the 'integrations' extra), so treat the
            # backend as available only when the full stack imports.
            import ragas  # noqa: F401
            import datasets  # noqa: F401
            self._ragas_available = True
            logger.info("RAGAS library loaded successfully")
        except ImportError:
            logger.warning(
                "RAGAS library not available. "
                "Install with: pip install 'uaef[integrations]'"
            )

        # Auto-configure Bedrock LLM/embeddings if not provided
        if self._ragas_available and (self._llm is None or self._embeddings is None):
            self._auto_configure_bedrock()

    def _ensure_available(self) -> None:
        """Raise an actionable ImportError when the RAGAS backend is missing.

        Requesting a RAGAS metric without the 'integrations' extra raises an
        ImportError naming the extra and the exact ``pip install`` command,
        and returns no metric result (Requirement 3.5).
        """
        if not self._ragas_available:
            raise integration_missing_error(_RAGAS_FEATURE)

    def _auto_configure_bedrock(self) -> None:
        """Auto-configure Bedrock LLM and embeddings from UAEF config."""
        try:
            from uaef.config import get_config
            import boto3
            from botocore.config import Config as BotoConfig

            config = get_config()
            boto_config = BotoConfig(
                region_name=config.aws.region,
                retries={"max_attempts": 3, "mode": "standard"},
            )
            bedrock_client = boto3.client(service_name="bedrock-runtime", config=boto_config)

            if self._llm is None:
                from langchain_aws import ChatBedrock
                self._llm = ChatBedrock(
                    client=bedrock_client,
                    model_id=config.llm_judge.model_id,
                    model_kwargs={
                        "max_tokens": config.llm_judge.max_tokens,
                        "temperature": config.llm_judge.temperature,
                    },
                )
                logger.info(f"Auto-configured RAGAS LLM judge: {config.llm_judge.model_id}")

            if self._embeddings is None:
                from langchain_aws import BedrockEmbeddings
                self._embeddings = BedrockEmbeddings(
                    client=bedrock_client,
                    model_id="amazon.titan-embed-text-v2:0",
                )
                logger.info("Auto-configured RAGAS embeddings: amazon.titan-embed-text-v2:0")

        except Exception as e:
            logger.warning(
                f"Could not auto-configure Bedrock for RAGAS: {e}. "
                "RAGAS will fall back to its default LLM (OpenAI). "
                "Set OPENAI_API_KEY or pass llm/embeddings to RAGASConnector()."
            )

    def is_available(self) -> bool:
        """
        Check if RAGAS library is available.

        Returns:
            True if RAGAS is installed and available, False otherwise
        """
        return self._ragas_available

    def _evaluate_kwargs(self) -> Dict[str, Any]:
        """Build extra kwargs for ragas evaluate() with optional llm/embeddings."""
        kwargs: Dict[str, Any] = {}
        # No progress bars: UAEF calls RAGAS from concurrent worker threads, and
        # tqdm bars there can deadlock on tqdm's global lock (seen in Jupyter
        # with DeepEval running alongside). Older RAGAS releases lack the flag.
        if _ragas_evaluate_accepts("show_progress"):
            kwargs["show_progress"] = False
        if self._llm is not None:
            kwargs["llm"] = self._llm
        if self._embeddings is not None:
            kwargs["embeddings"] = self._embeddings
        return kwargs

    @staticmethod
    def _extract_score(result: Any, metric_key: str) -> float:
        """Extract a single float score from a RAGAS evaluate result.

        RAGAS may return a single float or a list of floats depending on version.
        Returns nan if the score cannot be extracted or is nan.
        """
        value = result[metric_key]
        if isinstance(value, list):
            value = value[0] if value else 0.0
        return float(value)
    def list_available_metrics(self) -> List[str]:
        """
        List all metrics supported by this connector.

        Returns:
            List of metric names
        """
        return [
            "faithfulness",
            "context_precision",
            "context_recall",
            "answer_precision",
            "answer_recall",
            "answer_correctness",
            "tool_call_accuracy",
        ]


    def transform_to_ragas_format(
        self, evaluation_input: EvaluationInput
    ) -> Dict[str, Any]:
        """
        Transform UAEF EvaluationInput to RAGAS format.

        RAGAS expects data in a specific format with fields like:
        - question: The user's question/query
        - answer: The agent's response
        - contexts: List of context documents
        - ground_truth: Expected correct answer
        - reference: Expected tool calls and arguments

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            Dictionary in RAGAS format

        Raises:
            ValueError: If required fields are missing
        """
        self._ensure_available()

        trace = evaluation_input.trace
        ground_truth = evaluation_input.ground_truth

        # Extract question (first user message)
        question = None
        answer = None

        for message in trace.messages:
            if message.role == "user" and question is None:
                question = message.content
            elif message.role == "assistant" and answer is None:
                answer = message.content

        # Fallback: use first message content as question if no user role found
        if not question and trace.messages:
            question = trace.messages[0].content
            logger.warning("No user message found in trace, using first message as question")

        if not question:
            raise ValueError("No messages found in trace")

        # Fallback: use last message content as answer if no assistant role found
        if not answer and len(trace.messages) > 1:
            answer = trace.messages[-1].content
            logger.warning("No assistant message found in trace, using last message as answer")

        if not answer:
            raise ValueError("No assistant response found in trace")

        # Build RAGAS format
        ragas_data = {
            "question": question,
            "answer": answer,
            "contexts": evaluation_input.context,
        }

        # Add ground truth if available
        if ground_truth:
            if ground_truth.expected_output:
                ragas_data["ground_truth"] = ground_truth.expected_output

            if ground_truth.expected_tool_calls:
                ragas_data["reference_tool_calls"] = [
                    {
                        "name": tc.name,
                        "arguments": tc.arguments,
                    }
                    for tc in ground_truth.expected_tool_calls
                ]

        # Add actual tool calls
        if trace.tool_calls:
            ragas_data["tool_calls"] = [
                {
                    "name": tc.name,
                    "arguments": tc.arguments,
                }
                for tc in trace.tool_calls
            ]

        return ragas_data

    def calculate_faithfulness(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate faithfulness using RAGAS.

        Faithfulness measures whether every claim in the generated answer
        is supported by the retrieved context. A score of 1.0 means no
        hallucinations relative to the provided context.

        Args:
            evaluation_input: UAEF evaluation input (requires context)

        Returns:
            MetricScore with faithfulness score
        """
        self._ensure_available()

        if not evaluation_input.context:
            raise ValueError("Context is required for faithfulness")

        try:
            from ragas.metrics import faithfulness
            from ragas import evaluate
            from datasets import Dataset

            ragas_data = self.transform_to_ragas_format(evaluation_input)

            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "contexts": [ragas_data["contexts"]],
            })

            result = evaluate(dataset, metrics=[faithfulness], **self._evaluate_kwargs())
            score = self._extract_score(result, "faithfulness")

            return MetricScore(
                metric_name="ragas_faithfulness",
                score=float(score),
                reasoning="RAGAS faithfulness: measures if answer claims are supported by context",
                metadata={
                    "library": "ragas",
                    "metric": "faithfulness",
                    "contexts_count": len(ragas_data["contexts"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS faithfulness: {e}")
            raise

    def calculate_context_precision(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate context precision using RAGAS.

        Context precision measures whether the retrieved chunks ranked highest
        are the most relevant ones. Requires ground truth to determine which
        chunks are actually useful.

        Args:
            evaluation_input: UAEF evaluation input (requires context and ground truth)

        Returns:
            MetricScore with context precision score
        """
        self._ensure_available()

        if not evaluation_input.context:
            raise ValueError("Context is required for context precision")

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
            raise ValueError("Ground truth expected_output is required for context precision")

        try:
            from ragas.metrics import context_precision
            from ragas import evaluate
            from datasets import Dataset

            ragas_data = self.transform_to_ragas_format(evaluation_input)

            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "contexts": [ragas_data["contexts"]],
                "ground_truth": [ragas_data["ground_truth"]],
            })

            result = evaluate(dataset, metrics=[context_precision], **self._evaluate_kwargs())
            score = self._extract_score(result, "context_precision")

            return MetricScore(
                metric_name="ragas_context_precision",
                score=float(score),
                reasoning="RAGAS context precision: measures if top-ranked retrieved chunks are the most relevant",
                metadata={
                    "library": "ragas",
                    "metric": "context_precision",
                    "contexts_count": len(ragas_data["contexts"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS context precision: {e}")
            raise

    def calculate_context_recall(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate context recall using RAGAS.

        Context recall measures what percentage of the ground truth answer's
        claims are covered by the retrieved context. Low recall means the
        retriever is missing information the LLM needs.

        Args:
            evaluation_input: UAEF evaluation input (requires context and ground truth)

        Returns:
            MetricScore with context recall score
        """
        self._ensure_available()

        if not evaluation_input.context:
            raise ValueError("Context is required for context recall")

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
            raise ValueError("Ground truth expected_output is required for context recall")

        try:
            from ragas.metrics import context_recall
            from ragas import evaluate
            from datasets import Dataset

            ragas_data = self.transform_to_ragas_format(evaluation_input)

            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "contexts": [ragas_data["contexts"]],
                "ground_truth": [ragas_data["ground_truth"]],
            })

            result = evaluate(dataset, metrics=[context_recall], **self._evaluate_kwargs())
            score = self._extract_score(result, "context_recall")

            return MetricScore(
                metric_name="ragas_context_recall",
                score=float(score),
                reasoning="RAGAS context recall: measures coverage of ground truth claims in retrieved context",
                metadata={
                    "library": "ragas",
                    "metric": "context_recall",
                    "contexts_count": len(ragas_data["contexts"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS context recall: {e}")
            raise

    def calculate_answer_precision(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate answer precision using RAGAS.

        Answer precision measures how much of the generated answer is relevant
        and supported by the context documents.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with precision score

        Raises:
            ImportError: If the RAGAS backend ('integrations' extra) is not installed
            ValueError: If required fields are missing
        """
        self._ensure_available()

        try:
            from ragas.metrics import answer_relevancy
            from ragas import evaluate
            from datasets import Dataset

            # Transform to RAGAS format
            ragas_data = self.transform_to_ragas_format(evaluation_input)

            # Create dataset for RAGAS
            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "contexts": [ragas_data["contexts"]],
            })

            # Calculate metric
            result = evaluate(dataset, metrics=[answer_relevancy], **self._evaluate_kwargs())
            score = self._extract_score(result, "answer_relevancy")

            import math
            if math.isnan(score):
                score = 0.0

            return MetricScore(
                metric_name="ragas_answer_precision",
                score=float(score),
                reasoning="RAGAS answer precision: measures relevance and support from context",
                metadata={
                    "library": "ragas",
                    "metric": "answer_relevancy",
                    "contexts_count": len(ragas_data["contexts"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS answer precision: {e}")
            raise

    def calculate_answer_recall(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate answer recall using RAGAS.

        Answer recall measures how much of the ground truth information is
        present in the generated answer.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with recall score

        Raises:
            ImportError: If the RAGAS backend ('integrations' extra) is not installed
            ValueError: If ground truth is not provided
        """
        self._ensure_available()

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
            raise ValueError("Ground truth expected_output is required for answer recall")

        try:
            from ragas.metrics import answer_correctness
            from ragas import evaluate
            from datasets import Dataset

            # Transform to RAGAS format
            ragas_data = self.transform_to_ragas_format(evaluation_input)

            # Create dataset for RAGAS
            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "ground_truth": [ragas_data["ground_truth"]],
            })

            # Calculate metric
            result = evaluate(dataset, metrics=[answer_correctness], **self._evaluate_kwargs())
            score = self._extract_score(result, "answer_correctness")

            import math
            if math.isnan(score):
                score = 0.0

            return MetricScore(
                metric_name="ragas_answer_recall",
                score=float(score),
                reasoning="RAGAS answer recall: measures coverage of ground truth information",
                metadata={
                    "library": "ragas",
                    "metric": "answer_correctness",
                    "has_ground_truth": True,
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS answer recall: {e}")
            raise

    def calculate_answer_correctness(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate answer correctness using RAGAS.

        Answer correctness is a comprehensive metric that combines factual
        accuracy and semantic similarity with the ground truth.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with correctness score

        Raises:
            ImportError: If the RAGAS backend ('integrations' extra) is not installed
            ValueError: If ground truth is not provided
        """
        self._ensure_available()

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
            raise ValueError("Ground truth expected_output is required for answer correctness")

        try:
            from ragas.metrics import answer_correctness
            from ragas import evaluate
            from datasets import Dataset

            # Transform to RAGAS format
            ragas_data = self.transform_to_ragas_format(evaluation_input)

            # Create dataset for RAGAS
            dataset = Dataset.from_dict({
                "question": [ragas_data["question"]],
                "answer": [ragas_data["answer"]],
                "ground_truth": [ragas_data["ground_truth"]],
            })

            # Calculate metric
            result = evaluate(dataset, metrics=[answer_correctness], **self._evaluate_kwargs())
            score = self._extract_score(result, "answer_correctness")

            import math
            if math.isnan(score):
                score = 0.0

            return MetricScore(
                metric_name="ragas_answer_correctness",
                score=float(score),
                reasoning="RAGAS answer correctness: combines factual accuracy and semantic similarity",
                metadata={
                    "library": "ragas",
                    "metric": "answer_correctness",
                    "has_ground_truth": True,
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS answer correctness: {e}")
            raise

    def calculate_tool_call_accuracy(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate tool call accuracy using RAGAS.

        Tool call accuracy measures how accurately the agent selected and
        used tools compared to the expected tool calls.

        RAGAS 0.2.x requires a MultiTurnSample for this metric, so we build
        a conversation with HumanMessage → AIMessage (with tool_calls) →
        ToolMessage sequence.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with tool call accuracy score

        Raises:
            ImportError: If the RAGAS backend ('integrations' extra) is not installed
            ValueError: If ground truth tool calls are not provided
        """
        self._ensure_available()

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_tool_calls:
            raise ValueError("Ground truth expected_tool_calls is required for tool call accuracy")

        try:
            from ragas.metrics import ToolCallAccuracy
            from ragas.dataset_schema import MultiTurnSample, EvaluationDataset
            from ragas.messages import HumanMessage, AIMessage, ToolMessage
            from ragas.messages import ToolCall as RagasToolCall
            from ragas import evaluate

            # Transform to RAGAS format
            ragas_data = self.transform_to_ragas_format(evaluation_input)

            if "tool_calls" not in ragas_data or "reference_tool_calls" not in ragas_data:
                # Agent made no tool calls but ground truth expected some → score is 0
                actual = ragas_data.get("tool_calls", [])
                expected = ragas_data.get("reference_tool_calls", [])
                if not actual and expected:
                    return MetricScore(
                        metric_name="ragas_tool_call_accuracy",
                        score=0.0,
                        reasoning="Agent made no tool calls but ground truth expected tool usage",
                        metadata={
                            "library": "ragas",
                            "metric": "tool_call_accuracy",
                            "actual_tool_calls": 0,
                            "expected_tool_calls": len(expected),
                        },
                    )
                raise ValueError("Reference tool calls are required for tool call accuracy")

            # Build MultiTurnSample conversation
            # 1. User asks the question
            user_msg = HumanMessage(content=ragas_data["question"])

            # 2. AI responds with tool calls
            actual_tool_calls = [
                RagasToolCall(
                    name=tc["name"],
                    args=tc.get("arguments", tc.get("args", {})),
                )
                for tc in ragas_data["tool_calls"]
            ]
            ai_msg = AIMessage(
                content=ragas_data.get("answer", ""),
                tool_calls=actual_tool_calls,
            )

            # 3. Tool responses (one per tool call)
            tool_messages = []
            for tc in ragas_data["tool_calls"]:
                result = tc.get("result", tc.get("response", ""))
                tool_messages.append(ToolMessage(content=str(result) if result else ""))

            # Build reference tool calls
            reference_tool_calls = [
                RagasToolCall(
                    name=tc["name"],
                    args=tc.get("arguments", tc.get("args", {})),
                )
                for tc in ragas_data["reference_tool_calls"]
            ]

            # Create the multi-turn sample
            conversation = [user_msg, ai_msg] + tool_messages
            sample = MultiTurnSample(
                user_input=conversation,
                reference_tool_calls=reference_tool_calls,
            )

            # Evaluate using the multi-turn API
            tool_call_accuracy = ToolCallAccuracy()
            dataset = EvaluationDataset(samples=[sample])
            result = evaluate(dataset, metrics=[tool_call_accuracy], **self._evaluate_kwargs())
            score = self._extract_score(result, "tool_call_accuracy")

            return MetricScore(
                metric_name="ragas_tool_call_accuracy",
                score=float(score),
                reasoning="RAGAS tool call accuracy: measures correctness of tool selection and usage",
                metadata={
                    "library": "ragas",
                    "metric": "tool_call_accuracy",
                    "actual_tool_calls": len(ragas_data["tool_calls"]),
                    "expected_tool_calls": len(ragas_data["reference_tool_calls"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating RAGAS tool call accuracy: {e}")
            raise

    def calculate_metrics(
        self,
        evaluation_input: EvaluationInput,
        metric_names: List[str],
    ) -> List[MetricScore]:
        """
        Calculate multiple RAGAS metrics at once.

        Args:
            evaluation_input: UAEF evaluation input
            metric_names: List of metric names to calculate
                         (e.g., ["answer_precision", "answer_recall"])

        Returns:
            List of MetricScore objects

        Raises:
            ImportError: If the RAGAS backend ('integrations' extra) is not installed
            ValueError: If an unknown metric name is provided
        """
        self._ensure_available()

        metric_map = {
            "faithfulness": self.calculate_faithfulness,
            "context_precision": self.calculate_context_precision,
            "context_recall": self.calculate_context_recall,
            "answer_precision": self.calculate_answer_precision,
            "answer_recall": self.calculate_answer_recall,
            "answer_correctness": self.calculate_answer_correctness,
            "tool_call_accuracy": self.calculate_tool_call_accuracy,
        }

        scores = []
        for metric_name in metric_names:
            if metric_name not in metric_map:
                logger.warning(f"Unknown RAGAS metric: {metric_name}, skipping")
                continue

            try:
                score = metric_map[metric_name](evaluation_input)
                scores.append(score)
            except Exception as e:
                logger.error(f"Error calculating {metric_name}: {e}")
                # Continue with remaining metrics per requirement 14.6
                continue

        return scores

calculate_answer_correctness(evaluation_input)

Calculate answer correctness using RAGAS.

Answer correctness is a comprehensive metric that combines factual accuracy and semantic similarity with the ground truth.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with correctness score

Raises:

Type Description
ImportError

If the RAGAS backend ('integrations' extra) is not installed
ValueError

If ground truth is not provided
Source code in uaef/integrations/ragas_connector.py 
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
def calculate_answer_correctness(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate answer correctness using RAGAS.

    Answer correctness is a comprehensive metric that combines factual
    accuracy and semantic similarity with the ground truth.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with correctness score

    Raises:
        ImportError: If the RAGAS backend ('integrations' extra) is not installed
        ValueError: If ground truth is not provided
    """
    self._ensure_available()

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
        raise ValueError("Ground truth expected_output is required for answer correctness")

    try:
        from ragas.metrics import answer_correctness
        from ragas import evaluate
        from datasets import Dataset

        # Transform to RAGAS format
        ragas_data = self.transform_to_ragas_format(evaluation_input)

        # Create dataset for RAGAS
        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "ground_truth": [ragas_data["ground_truth"]],
        })

        # Calculate metric
        result = evaluate(dataset, metrics=[answer_correctness], **self._evaluate_kwargs())
        score = self._extract_score(result, "answer_correctness")

        import math
        if math.isnan(score):
            score = 0.0

        return MetricScore(
            metric_name="ragas_answer_correctness",
            score=float(score),
            reasoning="RAGAS answer correctness: combines factual accuracy and semantic similarity",
            metadata={
                "library": "ragas",
                "metric": "answer_correctness",
                "has_ground_truth": True,
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS answer correctness: {e}")
        raise

calculate_answer_precision(evaluation_input)

Calculate answer precision using RAGAS.

Answer precision measures how much of the generated answer is relevant and supported by the context documents.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with precision score

Raises:

Type Description
ImportError

If the RAGAS backend ('integrations' extra) is not installed
ValueError

If required fields are missing
Source code in uaef/integrations/ragas_connector.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def calculate_answer_precision(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate answer precision using RAGAS.

    Answer precision measures how much of the generated answer is relevant
    and supported by the context documents.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with precision score

    Raises:
        ImportError: If the RAGAS backend ('integrations' extra) is not installed
        ValueError: If required fields are missing
    """
    self._ensure_available()

    try:
        from ragas.metrics import answer_relevancy
        from ragas import evaluate
        from datasets import Dataset

        # Transform to RAGAS format
        ragas_data = self.transform_to_ragas_format(evaluation_input)

        # Create dataset for RAGAS
        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "contexts": [ragas_data["contexts"]],
        })

        # Calculate metric
        result = evaluate(dataset, metrics=[answer_relevancy], **self._evaluate_kwargs())
        score = self._extract_score(result, "answer_relevancy")

        import math
        if math.isnan(score):
            score = 0.0

        return MetricScore(
            metric_name="ragas_answer_precision",
            score=float(score),
            reasoning="RAGAS answer precision: measures relevance and support from context",
            metadata={
                "library": "ragas",
                "metric": "answer_relevancy",
                "contexts_count": len(ragas_data["contexts"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS answer precision: {e}")
        raise

calculate_answer_recall(evaluation_input)

Calculate answer recall using RAGAS.

Answer recall measures how much of the ground truth information is present in the generated answer.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with recall score

Raises:

Type Description
ImportError

If the RAGAS backend ('integrations' extra) is not installed
ValueError

If ground truth is not provided
Source code in uaef/integrations/ragas_connector.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
def calculate_answer_recall(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate answer recall using RAGAS.

    Answer recall measures how much of the ground truth information is
    present in the generated answer.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with recall score

    Raises:
        ImportError: If the RAGAS backend ('integrations' extra) is not installed
        ValueError: If ground truth is not provided
    """
    self._ensure_available()

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
        raise ValueError("Ground truth expected_output is required for answer recall")

    try:
        from ragas.metrics import answer_correctness
        from ragas import evaluate
        from datasets import Dataset

        # Transform to RAGAS format
        ragas_data = self.transform_to_ragas_format(evaluation_input)

        # Create dataset for RAGAS
        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "ground_truth": [ragas_data["ground_truth"]],
        })

        # Calculate metric
        result = evaluate(dataset, metrics=[answer_correctness], **self._evaluate_kwargs())
        score = self._extract_score(result, "answer_correctness")

        import math
        if math.isnan(score):
            score = 0.0

        return MetricScore(
            metric_name="ragas_answer_recall",
            score=float(score),
            reasoning="RAGAS answer recall: measures coverage of ground truth information",
            metadata={
                "library": "ragas",
                "metric": "answer_correctness",
                "has_ground_truth": True,
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS answer recall: {e}")
        raise

calculate_context_precision(evaluation_input)

Calculate context precision using RAGAS.

Context precision measures whether the retrieved chunks ranked highest are the most relevant ones. Requires ground truth to determine which chunks are actually useful.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input (requires context and ground truth)

 required

Returns:

Type Description
MetricScore

MetricScore with context precision score
Source code in uaef/integrations/ragas_connector.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def calculate_context_precision(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate context precision using RAGAS.

    Context precision measures whether the retrieved chunks ranked highest
    are the most relevant ones. Requires ground truth to determine which
    chunks are actually useful.

    Args:
        evaluation_input: UAEF evaluation input (requires context and ground truth)

    Returns:
        MetricScore with context precision score
    """
    self._ensure_available()

    if not evaluation_input.context:
        raise ValueError("Context is required for context precision")

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
        raise ValueError("Ground truth expected_output is required for context precision")

    try:
        from ragas.metrics import context_precision
        from ragas import evaluate
        from datasets import Dataset

        ragas_data = self.transform_to_ragas_format(evaluation_input)

        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "contexts": [ragas_data["contexts"]],
            "ground_truth": [ragas_data["ground_truth"]],
        })

        result = evaluate(dataset, metrics=[context_precision], **self._evaluate_kwargs())
        score = self._extract_score(result, "context_precision")

        return MetricScore(
            metric_name="ragas_context_precision",
            score=float(score),
            reasoning="RAGAS context precision: measures if top-ranked retrieved chunks are the most relevant",
            metadata={
                "library": "ragas",
                "metric": "context_precision",
                "contexts_count": len(ragas_data["contexts"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS context precision: {e}")
        raise

calculate_context_recall(evaluation_input)

Calculate context recall using RAGAS.

Context recall measures what percentage of the ground truth answer's claims are covered by the retrieved context. Low recall means the retriever is missing information the LLM needs.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input (requires context and ground truth)

 required

Returns:

Type Description
MetricScore

MetricScore with context recall score
Source code in uaef/integrations/ragas_connector.py 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
def calculate_context_recall(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate context recall using RAGAS.

    Context recall measures what percentage of the ground truth answer's
    claims are covered by the retrieved context. Low recall means the
    retriever is missing information the LLM needs.

    Args:
        evaluation_input: UAEF evaluation input (requires context and ground truth)

    Returns:
        MetricScore with context recall score
    """
    self._ensure_available()

    if not evaluation_input.context:
        raise ValueError("Context is required for context recall")

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
        raise ValueError("Ground truth expected_output is required for context recall")

    try:
        from ragas.metrics import context_recall
        from ragas import evaluate
        from datasets import Dataset

        ragas_data = self.transform_to_ragas_format(evaluation_input)

        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "contexts": [ragas_data["contexts"]],
            "ground_truth": [ragas_data["ground_truth"]],
        })

        result = evaluate(dataset, metrics=[context_recall], **self._evaluate_kwargs())
        score = self._extract_score(result, "context_recall")

        return MetricScore(
            metric_name="ragas_context_recall",
            score=float(score),
            reasoning="RAGAS context recall: measures coverage of ground truth claims in retrieved context",
            metadata={
                "library": "ragas",
                "metric": "context_recall",
                "contexts_count": len(ragas_data["contexts"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS context recall: {e}")
        raise

calculate_faithfulness(evaluation_input)

Calculate faithfulness using RAGAS.

Faithfulness measures whether every claim in the generated answer is supported by the retrieved context. A score of 1.0 means no hallucinations relative to the provided context.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input (requires context)

 required

Returns:

Type Description
MetricScore

MetricScore with faithfulness score
Source code in uaef/integrations/ragas_connector.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def calculate_faithfulness(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate faithfulness using RAGAS.

    Faithfulness measures whether every claim in the generated answer
    is supported by the retrieved context. A score of 1.0 means no
    hallucinations relative to the provided context.

    Args:
        evaluation_input: UAEF evaluation input (requires context)

    Returns:
        MetricScore with faithfulness score
    """
    self._ensure_available()

    if not evaluation_input.context:
        raise ValueError("Context is required for faithfulness")

    try:
        from ragas.metrics import faithfulness
        from ragas import evaluate
        from datasets import Dataset

        ragas_data = self.transform_to_ragas_format(evaluation_input)

        dataset = Dataset.from_dict({
            "question": [ragas_data["question"]],
            "answer": [ragas_data["answer"]],
            "contexts": [ragas_data["contexts"]],
        })

        result = evaluate(dataset, metrics=[faithfulness], **self._evaluate_kwargs())
        score = self._extract_score(result, "faithfulness")

        return MetricScore(
            metric_name="ragas_faithfulness",
            score=float(score),
            reasoning="RAGAS faithfulness: measures if answer claims are supported by context",
            metadata={
                "library": "ragas",
                "metric": "faithfulness",
                "contexts_count": len(ragas_data["contexts"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS faithfulness: {e}")
        raise

calculate_metrics(evaluation_input, metric_names)

Calculate multiple RAGAS metrics at once.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required
metric_names List[str]

List of metric names to calculate (e.g., ["answer_precision", "answer_recall"])

 required

Returns:

Type Description
List[MetricScore]

List of MetricScore objects

Raises:

Type Description
ImportError

If the RAGAS backend ('integrations' extra) is not installed
ValueError

If an unknown metric name is provided
Source code in uaef/integrations/ragas_connector.py 
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
def calculate_metrics(
    self,
    evaluation_input: EvaluationInput,
    metric_names: List[str],
) -> List[MetricScore]:
    """
    Calculate multiple RAGAS metrics at once.

    Args:
        evaluation_input: UAEF evaluation input
        metric_names: List of metric names to calculate
                     (e.g., ["answer_precision", "answer_recall"])

    Returns:
        List of MetricScore objects

    Raises:
        ImportError: If the RAGAS backend ('integrations' extra) is not installed
        ValueError: If an unknown metric name is provided
    """
    self._ensure_available()

    metric_map = {
        "faithfulness": self.calculate_faithfulness,
        "context_precision": self.calculate_context_precision,
        "context_recall": self.calculate_context_recall,
        "answer_precision": self.calculate_answer_precision,
        "answer_recall": self.calculate_answer_recall,
        "answer_correctness": self.calculate_answer_correctness,
        "tool_call_accuracy": self.calculate_tool_call_accuracy,
    }

    scores = []
    for metric_name in metric_names:
        if metric_name not in metric_map:
            logger.warning(f"Unknown RAGAS metric: {metric_name}, skipping")
            continue

        try:
            score = metric_map[metric_name](evaluation_input)
            scores.append(score)
        except Exception as e:
            logger.error(f"Error calculating {metric_name}: {e}")
            # Continue with remaining metrics per requirement 14.6
            continue

    return scores

calculate_tool_call_accuracy(evaluation_input)

Calculate tool call accuracy using RAGAS.

Tool call accuracy measures how accurately the agent selected and used tools compared to the expected tool calls.

RAGAS 0.2.x requires a MultiTurnSample for this metric, so we build a conversation with HumanMessage → AIMessage (with tool_calls) → ToolMessage sequence.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with tool call accuracy score

Raises:

Type Description
ImportError

If the RAGAS backend ('integrations' extra) is not installed
ValueError

If ground truth tool calls are not provided
Source code in uaef/integrations/ragas_connector.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
def calculate_tool_call_accuracy(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate tool call accuracy using RAGAS.

    Tool call accuracy measures how accurately the agent selected and
    used tools compared to the expected tool calls.

    RAGAS 0.2.x requires a MultiTurnSample for this metric, so we build
    a conversation with HumanMessage → AIMessage (with tool_calls) →
    ToolMessage sequence.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with tool call accuracy score

    Raises:
        ImportError: If the RAGAS backend ('integrations' extra) is not installed
        ValueError: If ground truth tool calls are not provided
    """
    self._ensure_available()

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_tool_calls:
        raise ValueError("Ground truth expected_tool_calls is required for tool call accuracy")

    try:
        from ragas.metrics import ToolCallAccuracy
        from ragas.dataset_schema import MultiTurnSample, EvaluationDataset
        from ragas.messages import HumanMessage, AIMessage, ToolMessage
        from ragas.messages import ToolCall as RagasToolCall
        from ragas import evaluate

        # Transform to RAGAS format
        ragas_data = self.transform_to_ragas_format(evaluation_input)

        if "tool_calls" not in ragas_data or "reference_tool_calls" not in ragas_data:
            # Agent made no tool calls but ground truth expected some → score is 0
            actual = ragas_data.get("tool_calls", [])
            expected = ragas_data.get("reference_tool_calls", [])
            if not actual and expected:
                return MetricScore(
                    metric_name="ragas_tool_call_accuracy",
                    score=0.0,
                    reasoning="Agent made no tool calls but ground truth expected tool usage",
                    metadata={
                        "library": "ragas",
                        "metric": "tool_call_accuracy",
                        "actual_tool_calls": 0,
                        "expected_tool_calls": len(expected),
                    },
                )
            raise ValueError("Reference tool calls are required for tool call accuracy")

        # Build MultiTurnSample conversation
        # 1. User asks the question
        user_msg = HumanMessage(content=ragas_data["question"])

        # 2. AI responds with tool calls
        actual_tool_calls = [
            RagasToolCall(
                name=tc["name"],
                args=tc.get("arguments", tc.get("args", {})),
            )
            for tc in ragas_data["tool_calls"]
        ]
        ai_msg = AIMessage(
            content=ragas_data.get("answer", ""),
            tool_calls=actual_tool_calls,
        )

        # 3. Tool responses (one per tool call)
        tool_messages = []
        for tc in ragas_data["tool_calls"]:
            result = tc.get("result", tc.get("response", ""))
            tool_messages.append(ToolMessage(content=str(result) if result else ""))

        # Build reference tool calls
        reference_tool_calls = [
            RagasToolCall(
                name=tc["name"],
                args=tc.get("arguments", tc.get("args", {})),
            )
            for tc in ragas_data["reference_tool_calls"]
        ]

        # Create the multi-turn sample
        conversation = [user_msg, ai_msg] + tool_messages
        sample = MultiTurnSample(
            user_input=conversation,
            reference_tool_calls=reference_tool_calls,
        )

        # Evaluate using the multi-turn API
        tool_call_accuracy = ToolCallAccuracy()
        dataset = EvaluationDataset(samples=[sample])
        result = evaluate(dataset, metrics=[tool_call_accuracy], **self._evaluate_kwargs())
        score = self._extract_score(result, "tool_call_accuracy")

        return MetricScore(
            metric_name="ragas_tool_call_accuracy",
            score=float(score),
            reasoning="RAGAS tool call accuracy: measures correctness of tool selection and usage",
            metadata={
                "library": "ragas",
                "metric": "tool_call_accuracy",
                "actual_tool_calls": len(ragas_data["tool_calls"]),
                "expected_tool_calls": len(ragas_data["reference_tool_calls"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating RAGAS tool call accuracy: {e}")
        raise

is_available()

Check if RAGAS library is available.

Returns:

Type Description
bool

True if RAGAS is installed and available, False otherwise
Source code in uaef/integrations/ragas_connector.py 
128
129
130
131
132
133
134
135
def is_available(self) -> bool:
    """
    Check if RAGAS library is available.

    Returns:
        True if RAGAS is installed and available, False otherwise
    """
    return self._ragas_available

list_available_metrics()

List all metrics supported by this connector.

Returns:

Type Description
List[str]

List of metric names
Source code in uaef/integrations/ragas_connector.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def list_available_metrics(self) -> List[str]:
    """
    List all metrics supported by this connector.

    Returns:
        List of metric names
    """
    return [
        "faithfulness",
        "context_precision",
        "context_recall",
        "answer_precision",
        "answer_recall",
        "answer_correctness",
        "tool_call_accuracy",
    ]

transform_to_ragas_format(evaluation_input)

Transform UAEF EvaluationInput to RAGAS format.

RAGAS expects data in a specific format with fields like: - question: The user's question/query - answer: The agent's response - contexts: List of context documents - ground_truth: Expected correct answer - reference: Expected tool calls and arguments

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
Dict[str, Any]

Dictionary in RAGAS format

Raises:

Type Description
ValueError

If required fields are missing
Source code in uaef/integrations/ragas_connector.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def transform_to_ragas_format(
    self, evaluation_input: EvaluationInput
) -> Dict[str, Any]:
    """
    Transform UAEF EvaluationInput to RAGAS format.

    RAGAS expects data in a specific format with fields like:
    - question: The user's question/query
    - answer: The agent's response
    - contexts: List of context documents
    - ground_truth: Expected correct answer
    - reference: Expected tool calls and arguments

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        Dictionary in RAGAS format

    Raises:
        ValueError: If required fields are missing
    """
    self._ensure_available()

    trace = evaluation_input.trace
    ground_truth = evaluation_input.ground_truth

    # Extract question (first user message)
    question = None
    answer = None

    for message in trace.messages:
        if message.role == "user" and question is None:
            question = message.content
        elif message.role == "assistant" and answer is None:
            answer = message.content

    # Fallback: use first message content as question if no user role found
    if not question and trace.messages:
        question = trace.messages[0].content
        logger.warning("No user message found in trace, using first message as question")

    if not question:
        raise ValueError("No messages found in trace")

    # Fallback: use last message content as answer if no assistant role found
    if not answer and len(trace.messages) > 1:
        answer = trace.messages[-1].content
        logger.warning("No assistant message found in trace, using last message as answer")

    if not answer:
        raise ValueError("No assistant response found in trace")

    # Build RAGAS format
    ragas_data = {
        "question": question,
        "answer": answer,
        "contexts": evaluation_input.context,
    }

    # Add ground truth if available
    if ground_truth:
        if ground_truth.expected_output:
            ragas_data["ground_truth"] = ground_truth.expected_output

        if ground_truth.expected_tool_calls:
            ragas_data["reference_tool_calls"] = [
                {
                    "name": tc.name,
                    "arguments": tc.arguments,
                }
                for tc in ground_truth.expected_tool_calls
            ]

    # Add actual tool calls
    if trace.tool_calls:
        ragas_data["tool_calls"] = [
            {
                "name": tc.name,
                "arguments": tc.arguments,
            }
            for tc in trace.tool_calls
        ]

    return ragas_data

uaef.integrations.deepeval_connector

DeepEval connector for external metric integration.

DeepEvalConnector

Connector for DeepEval library.

This connector wraps DeepEval library metrics in UAEF's interface, transforming UAEF data to DeepEval format, calling DeepEval metrics, and transforming results back to UAEF MetricScore objects.

DeepEval is an external evaluation library that provides metrics for contextual precision/recall, hallucination detection, tool correctness, and G-Eval (custom criteria evaluation).

Source code in uaef/integrations/deepeval_connector.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
class DeepEvalConnector:
    """
    Connector for DeepEval library.

    This connector wraps DeepEval library metrics in UAEF's interface, transforming
    UAEF data to DeepEval format, calling DeepEval metrics, and transforming results
    back to UAEF MetricScore objects.

    DeepEval is an external evaluation library that provides metrics for contextual
    precision/recall, hallucination detection, tool correctness, and G-Eval (custom
    criteria evaluation).
    """

    def __init__(self, model=None):
        """Initialize DeepEval connector.

        Args:
            model: DeepEval model to use as judge. If None, auto-creates an
                   AmazonBedrockModel instance using UAEF config.
        """
        self._deepeval_available = False
        self._model = model
        try:
            import deepeval  # noqa: F401
            self._deepeval_available = True
            logger.info("DeepEval library loaded successfully")
        except (ImportError, TypeError, Exception) as e:
            logger.warning(
                f"DeepEval library not available: {e}. "
                "Install with: pip install 'uaef[integrations]'"
            )

        # Auto-configure Bedrock model if not provided
        if self._deepeval_available and self._model is None:
            self._auto_configure_bedrock()

    def _ensure_available(self) -> None:
        """Raise an actionable ImportError when the DeepEval backend is missing.

        Requesting a DeepEval metric without the 'integrations' extra raises an
        ImportError naming the extra and the exact ``pip install`` command,
        and returns no metric result (Requirement 3.5).
        """
        if not self._deepeval_available:
            raise integration_missing_error(_DEEPEVAL_FEATURE)

    def _auto_configure_bedrock(self) -> None:
        """Auto-configure Amazon Bedrock model from UAEF config for DeepEval."""
        try:
            from uaef.config import get_config
            from deepeval.models import AmazonBedrockModel

            config = get_config()
            self._model = AmazonBedrockModel(
                model=config.llm_judge.model_id,
                region=config.aws.region,
            )
            logger.info(f"Auto-configured DeepEval Bedrock model: {config.llm_judge.model_id}")
        except Exception as e:
            logger.warning(
                f"Could not auto-configure Bedrock for DeepEval: {e}. "
                "DeepEval will fall back to OpenAI. "
                "Set OPENAI_API_KEY or pass a model to DeepEvalConnector()."
            )

    def is_available(self) -> bool:
        """
        Check if DeepEval library is available.

        Returns:
            True if DeepEval is installed and available, False otherwise
        """
        return self._deepeval_available
    def list_available_metrics(self) -> List[str]:
        """
        List all metrics supported by this connector.

        Returns:
            List of metric names
        """
        return [
            "contextual_precision",
            "contextual_recall",
            "contextual_relevancy",
            "hallucination",
            "faithfulness",
            "answer_relevancy",
            "tool_correctness",
            "geval",
        ]


    def transform_to_deepeval_format(
        self, evaluation_input: EvaluationInput
    ) -> Dict[str, Any]:
        """
        Transform UAEF EvaluationInput to DeepEval format.

        DeepEval expects data in a specific format with fields like:
        - input: The user's question/query
        - actual_output: The agent's response
        - retrieval_context: List of context documents
        - expected_output: Expected correct answer
        - context: List of context strings for hallucination detection
        - tools_called: List of tool calls made by the agent
        - expected_tools: List of expected tool calls

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            Dictionary in DeepEval format

        Raises:
            ValueError: If required fields are missing
        """
        self._ensure_available()

        trace = evaluation_input.trace
        ground_truth = evaluation_input.ground_truth

        # Extract question (first user message)
        question = None
        actual_output = None

        for message in trace.messages:
            if message.role == "user" and question is None:
                question = message.content
            elif message.role == "assistant" and actual_output is None:
                actual_output = message.content

        # Fallback: use first message content as question if no user role found
        if not question and trace.messages:
            question = trace.messages[0].content
            logger.warning("No user message found in trace, using first message as question")

        if not question:
            raise ValueError("No messages found in trace")

        # Fallback: use last message content as answer if no assistant role found
        if not actual_output and len(trace.messages) > 1:
            actual_output = trace.messages[-1].content
            logger.warning("No assistant message found in trace, using last message as answer")

        if not actual_output:
            raise ValueError("No assistant response found in trace")

        # Build DeepEval format
        deepeval_data = {
            "input": question,
            "actual_output": actual_output,
            "retrieval_context": evaluation_input.context,
            "context": evaluation_input.context,  # For hallucination detection
        }

        # Add ground truth if available
        if ground_truth:
            if ground_truth.expected_output:
                deepeval_data["expected_output"] = ground_truth.expected_output

            if ground_truth.expected_tool_calls:
                deepeval_data["expected_tools"] = [
                    {
                        "name": tc.name,
                        "arguments": tc.arguments,
                    }
                    for tc in ground_truth.expected_tool_calls
                ]

        # Add actual tool calls
        if trace.tool_calls:
            deepeval_data["tools_called"] = [
                {
                    "name": tc.name,
                    "arguments": tc.arguments,
                }
                for tc in trace.tool_calls
            ]

        return deepeval_data

    def calculate_contextual_precision(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate contextual precision using DeepEval.

        Contextual precision measures whether the retrieved context is relevant
        to the query and whether irrelevant context is ranked lower.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with contextual precision score

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If required fields are missing
        """
        self._ensure_available()

        if not evaluation_input.context:
            return MetricScore(
                metric_name="deepeval_contextual_precision",
                score=None,
                reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
                metadata={"warning": "not_applicable", "reason": "no_context"},
            )

        try:
            from deepeval.metrics import ContextualPrecisionMetric
            from deepeval.test_case import LLMTestCase

            # Transform to DeepEval format
            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            if not deepeval_data.get("expected_output"):
                raise ValueError("Ground truth expected_output is required for contextual precision")

            # Create test case for DeepEval
            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                expected_output=deepeval_data["expected_output"],
                retrieval_context=deepeval_data["retrieval_context"],
            )

            # Calculate metric
            metric = ContextualPrecisionMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_contextual_precision",
                score=float(metric.score),
                reasoning=f"DeepEval contextual precision: {metric.reason if hasattr(metric, 'reason') else 'measures relevance of retrieved context'}",
                metadata={
                    "library": "deepeval",
                    "metric": "contextual_precision",
                    "contexts_count": len(deepeval_data["retrieval_context"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval contextual precision: {e}")
            raise

    def calculate_contextual_recall(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate contextual recall using DeepEval.

        Contextual recall measures whether all relevant information from the
        expected output can be found in the retrieved context.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with contextual recall score

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If ground truth is not provided
        """
        self._ensure_available()

        if not evaluation_input.context:
            return MetricScore(
                metric_name="deepeval_contextual_recall",
                score=None,
                reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
                metadata={"warning": "not_applicable", "reason": "no_context"},
            )

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
            raise ValueError("Ground truth expected_output is required for contextual recall")

        try:
            from deepeval.metrics import ContextualRecallMetric
            from deepeval.test_case import LLMTestCase

            # Transform to DeepEval format
            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            # Create test case for DeepEval
            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                expected_output=deepeval_data["expected_output"],
                retrieval_context=deepeval_data["retrieval_context"],
            )

            # Calculate metric
            metric = ContextualRecallMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_contextual_recall",
                score=float(metric.score),
                reasoning=f"DeepEval contextual recall: {metric.reason if hasattr(metric, 'reason') else 'measures coverage of expected output in context'}",
                metadata={
                    "library": "deepeval",
                    "metric": "contextual_recall",
                    "has_ground_truth": True,
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval contextual recall: {e}")
            raise

    def calculate_hallucination(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate hallucination score using DeepEval.

        Hallucination metric detects claims in the agent's response that are not
        supported by the provided context documents.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with hallucination score (1.0 = no hallucination, 0.0 = high hallucination)

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If context is not provided
        """
        self._ensure_available()

        if not evaluation_input.context:
            return MetricScore(
                metric_name="deepeval_hallucination",
                score=None,
                reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
                metadata={"warning": "not_applicable", "reason": "no_context"},
            )

        try:
            from deepeval.metrics import HallucinationMetric
            from deepeval.test_case import LLMTestCase

            # Transform to DeepEval format
            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            # Create test case for DeepEval
            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                context=deepeval_data["context"],
            )

            # Calculate metric
            metric = HallucinationMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_hallucination",
                score=float(metric.score),
                reasoning=f"DeepEval hallucination: {metric.reason if hasattr(metric, 'reason') else 'detects unsupported claims in response'}",
                metadata={
                    "library": "deepeval",
                    "metric": "hallucination",
                    "contexts_count": len(deepeval_data["context"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval hallucination: {e}")
            raise

    def calculate_tool_correctness(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate tool correctness using DeepEval.

        Tool correctness measures how accurately the agent selected and used
        tools compared to the expected tool calls.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with tool correctness score

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If ground truth tool calls are not provided
        """
        self._ensure_available()

        if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_tool_calls:
            raise ValueError("Ground truth expected_tool_calls is required for tool correctness")

        try:
            from deepeval.metrics import ToolCorrectnessMetric
            from deepeval.test_case import LLMTestCase, ToolCall as DeepEvalToolCall

            # Transform to DeepEval format
            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            if "tools_called" not in deepeval_data or "expected_tools" not in deepeval_data:
                # Agent made no tool calls but ground truth expected some → score is 0
                actual = deepeval_data.get("tools_called", [])
                expected = deepeval_data.get("expected_tools", [])
                if not actual and expected:
                    return MetricScore(
                        metric_name="deepeval_tool_correctness",
                        score=0.0,
                        reasoning="Agent made no tool calls but ground truth expected tool usage",
                        metadata={
                            "library": "deepeval",
                            "metric": "tool_correctness",
                            "actual_tool_calls": 0,
                            "expected_tool_calls": len(expected),
                        },
                    )
                raise ValueError("Tool calls and expected tools are required")

            # Convert to DeepEval ToolCall objects
            tools_called = [
                DeepEvalToolCall(
                    name=tc["name"],
                    input_parameters=tc.get("arguments", {}),
                )
                for tc in deepeval_data["tools_called"]
            ]
            expected_tools = [
                DeepEvalToolCall(
                    name=tc["name"],
                    input_parameters=tc.get("arguments", {}),
                )
                for tc in deepeval_data["expected_tools"]
            ]

            # Create test case for DeepEval
            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                tools_called=tools_called,
                expected_tools=expected_tools,
            )

            # Calculate metric
            metric = ToolCorrectnessMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_tool_correctness",
                score=float(metric.score),
                reasoning=f"DeepEval tool correctness: {metric.reason if hasattr(metric, 'reason') else 'measures correctness of tool selection and usage'}",
                metadata={
                    "library": "deepeval",
                    "metric": "tool_correctness",
                    "actual_tool_calls": len(deepeval_data["tools_called"]),
                    "expected_tool_calls": len(deepeval_data["expected_tools"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval tool correctness: {e}")
            raise

    def calculate_contextual_relevancy(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate contextual relevancy using DeepEval.

        Contextual relevancy assesses whether the retrieved context is relevant
        to the user's input query.

        Args:
            evaluation_input: UAEF evaluation input (requires context)

        Returns:
            MetricScore with contextual relevancy score
        """
        self._ensure_available()

        if not evaluation_input.context:
            return MetricScore(
                metric_name="deepeval_contextual_relevancy",
                score=None,
                reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
                metadata={"warning": "not_applicable", "reason": "no_context"},
            )

        try:
            from deepeval.metrics import ContextualRelevancyMetric
            from deepeval.test_case import LLMTestCase

            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                retrieval_context=deepeval_data["retrieval_context"],
            )

            metric = ContextualRelevancyMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_contextual_relevancy",
                score=float(metric.score),
                reasoning=f"DeepEval contextual relevancy: {getattr(metric, 'reason', 'assesses relevance of retrieved context to query')}",
                metadata={
                    "library": "deepeval",
                    "metric": "contextual_relevancy",
                    "contexts_count": len(deepeval_data["retrieval_context"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval contextual relevancy: {e}")
            raise

    def calculate_faithfulness(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate faithfulness using DeepEval.

        Faithfulness measures whether the generated answer is factually consistent
        with the provided retrieval context (i.e., no hallucinated claims).

        Args:
            evaluation_input: UAEF evaluation input (requires context)

        Returns:
            MetricScore with faithfulness score
        """
        self._ensure_available()

        if not evaluation_input.context:
            return MetricScore(
                metric_name="deepeval_faithfulness",
                score=None,
                reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
                metadata={"warning": "not_applicable", "reason": "no_context"},
            )

        try:
            from deepeval.metrics import FaithfulnessMetric
            from deepeval.test_case import LLMTestCase

            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
                retrieval_context=deepeval_data["retrieval_context"],
            )

            metric = FaithfulnessMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_faithfulness",
                score=float(metric.score),
                reasoning=f"DeepEval faithfulness: {getattr(metric, 'reason', 'measures factual consistency with retrieval context')}",
                metadata={
                    "library": "deepeval",
                    "metric": "faithfulness",
                    "contexts_count": len(deepeval_data["retrieval_context"]),
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval faithfulness: {e}")
            raise

    def calculate_answer_relevancy(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """
        Calculate answer relevancy using DeepEval.

        Answer relevancy evaluates whether the generated answer is relevant
        to the user's input query.

        Args:
            evaluation_input: UAEF evaluation input

        Returns:
            MetricScore with answer relevancy score
        """
        self._ensure_available()

        try:
            from deepeval.metrics import AnswerRelevancyMetric
            from deepeval.test_case import LLMTestCase

            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
            )

            metric = AnswerRelevancyMetric(model=self._model)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_answer_relevancy",
                score=float(metric.score),
                reasoning=f"DeepEval answer relevancy: {getattr(metric, 'reason', 'evaluates relevance of answer to query')}",
                metadata={
                    "library": "deepeval",
                    "metric": "answer_relevancy",
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval answer relevancy: {e}")
            raise

    def calculate_geval(
        self,
        evaluation_input: EvaluationInput,
        criteria: str,
        evaluation_steps: Optional[List[str]] = None,
    ) -> MetricScore:
        """
        Calculate G-Eval score using DeepEval with custom criteria.

        G-Eval is a framework for LLM-based evaluation using custom criteria.
        It allows defining specific evaluation criteria and steps.

        Args:
            evaluation_input: UAEF evaluation input
            criteria: Custom evaluation criteria description
            evaluation_steps: Optional list of evaluation steps to follow

        Returns:
            MetricScore with G-Eval score

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If required fields are missing
        """
        self._ensure_available()

        if not criteria:
            raise ValueError("Criteria is required for G-Eval")

        try:
            from deepeval.metrics import GEval
            from deepeval.test_case import LLMTestCase

            # Transform to DeepEval format
            deepeval_data = self.transform_to_deepeval_format(evaluation_input)

            # Create test case for DeepEval
            test_case = LLMTestCase(
                input=deepeval_data["input"],
                actual_output=deepeval_data["actual_output"],
            )

            # Add expected output if available
            if deepeval_data.get("expected_output"):
                test_case.expected_output = deepeval_data["expected_output"]

            # Create G-Eval metric with custom criteria
            metric_kwargs = {
                "name": "custom_geval",
                "criteria": criteria,
                "evaluation_params": [
                    LLMTestCase.input,
                    LLMTestCase.actual_output,
                ],
            }

            if evaluation_steps:
                metric_kwargs["evaluation_steps"] = evaluation_steps

            metric = GEval(**metric_kwargs)
            _measure(metric, test_case)

            return MetricScore(
                metric_name="deepeval_geval",
                score=float(metric.score),
                reasoning=f"DeepEval G-Eval: {metric.reason if hasattr(metric, 'reason') else criteria}",
                metadata={
                    "library": "deepeval",
                    "metric": "geval",
                    "criteria": criteria,
                    "has_evaluation_steps": evaluation_steps is not None,
                },
            )

        except Exception as e:
            logger.error(f"Error calculating DeepEval G-Eval: {e}")
            raise

    def calculate_role_adherence(
        self, evaluation_input: EvaluationInput
    ) -> MetricScore:
        """Calculate role adherence using DeepEval's `RoleAdherenceMetric`.

        Reads the chatbot persona strictly from
        `ground_truth.expected_arguments["chatbot_role"]`. Builds a
        `ConversationalTestCase` from `trace.messages`, delegates to
        DeepEval, and surfaces the score along with per-turn out-of-character
        verdicts in metadata.

        Returns a `not_applicable` `MetricScore` if `chatbot_role` is missing
        or the filtered turn list is empty. Errors raised by DeepEval's
        `measure()` propagate to the caller (matches sibling methods).
        """
        if not self._deepeval_available:
            raise RuntimeError(
                "DeepEval library not available. Install with: pip install deepeval"
            )

        gt = evaluation_input.ground_truth
        role = (
            gt.expected_arguments.get("chatbot_role")
            if gt is not None and gt.expected_arguments
            else None
        )
        if not role:
            return MetricScore(
                metric_name="deepeval_role_adherence",
                score=None,
                reasoning="Cannot evaluate: expected_arguments['chatbot_role'] is required",
                metadata={"warning": "not_applicable"},
            )

        turns = _messages_to_deepeval_turns(evaluation_input.trace.messages)
        if not turns:
            return MetricScore(
                metric_name="deepeval_role_adherence",
                score=None,
                reasoning="Cannot evaluate: no user/assistant messages",
                metadata={"warning": "not_applicable"},
            )

        from deepeval.metrics import RoleAdherenceMetric
        from deepeval.test_case import ConversationalTestCase

        test_case = ConversationalTestCase(turns=turns, chatbot_role=role)
        metric = RoleAdherenceMetric(model=self._model)
        try:
            _measure(metric, test_case)
        except Exception as e:
            logger.error(f"Error calculating DeepEval role adherence: {e}")
            raise

        # Per-verdict access uses getattr for resilience against DeepEval field
        # renames; outer attributes are required and we want loud failures.
        verdicts = metric.out_of_character_verdicts.verdicts or []
        out_of_character_turns = [
            {
                "turn_index": getattr(v, "index", None),
                "ai_message": getattr(v, "ai_message", None),
                "reason": getattr(v, "reason", None),
            }
            for v in verdicts
        ]

        assistant_turn_count = sum(1 for t in turns if t.role == "assistant")

        if metric.score is None:
            return MetricScore(
                metric_name="deepeval_role_adherence",
                score=None,
                reasoning="Cannot evaluate: DeepEval returned no score (likely zero assistant turns)",
                metadata={
                    "warning": "not_applicable",
                    "library": "deepeval",
                    "metric": "role_adherence",
                    "out_of_character_turns": out_of_character_turns,
                    "assistant_turn_count": assistant_turn_count,
                },
            )

        return MetricScore(
            metric_name="deepeval_role_adherence",
            score=float(metric.score),
            reasoning=(
                getattr(metric, "reason", None)
                or "DeepEval role adherence (no reason returned)"
            ),
            metadata={
                "library": "deepeval",
                "metric": "role_adherence",
                "out_of_character_turns": out_of_character_turns,
                "assistant_turn_count": assistant_turn_count,
            },
        )

    def calculate_metrics(
        self,
        evaluation_input: EvaluationInput,
        metric_names: List[str],
        geval_criteria: Optional[str] = None,
    ) -> List[MetricScore]:
        """
        Calculate multiple DeepEval metrics at once.

        Args:
            evaluation_input: UAEF evaluation input
            metric_names: List of metric names to calculate
                         (e.g., ["contextual_precision", "hallucination"])
            geval_criteria: Optional criteria for G-Eval metric

        Returns:
            List of MetricScore objects

        Raises:
            ImportError: If the DeepEval backend ('integrations' extra) is not installed
            ValueError: If an unknown metric name is provided
        """
        self._ensure_available()

        metric_map = {
            "contextual_precision": self.calculate_contextual_precision,
            "contextual_recall": self.calculate_contextual_recall,
            "contextual_relevancy": self.calculate_contextual_relevancy,
            "hallucination": self.calculate_hallucination,
            "faithfulness": self.calculate_faithfulness,
            "answer_relevancy": self.calculate_answer_relevancy,
            "tool_correctness": self.calculate_tool_correctness,
        }

        scores = []
        for metric_name in metric_names:
            if metric_name == "geval":
                if not geval_criteria:
                    logger.warning("G-Eval requires criteria, skipping")
                    continue
                try:
                    score = self.calculate_geval(evaluation_input, geval_criteria)
                    scores.append(score)
                except Exception as e:
                    logger.error(f"Error calculating geval: {e}")
                    # Continue with remaining metrics per requirement 14.6
                    continue
            elif metric_name in metric_map:
                try:
                    score = metric_map[metric_name](evaluation_input)
                    scores.append(score)
                except Exception as e:
                    logger.error(f"Error calculating {metric_name}: {e}")
                    # Continue with remaining metrics per requirement 14.6
                    continue
            else:
                logger.warning(f"Unknown DeepEval metric: {metric_name}, skipping")
                continue

        return scores

calculate_answer_relevancy(evaluation_input)

Calculate answer relevancy using DeepEval.

Answer relevancy evaluates whether the generated answer is relevant to the user's input query.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with answer relevancy score
Source code in uaef/integrations/deepeval_connector.py 
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
def calculate_answer_relevancy(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate answer relevancy using DeepEval.

    Answer relevancy evaluates whether the generated answer is relevant
    to the user's input query.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with answer relevancy score
    """
    self._ensure_available()

    try:
        from deepeval.metrics import AnswerRelevancyMetric
        from deepeval.test_case import LLMTestCase

        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
        )

        metric = AnswerRelevancyMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_answer_relevancy",
            score=float(metric.score),
            reasoning=f"DeepEval answer relevancy: {getattr(metric, 'reason', 'evaluates relevance of answer to query')}",
            metadata={
                "library": "deepeval",
                "metric": "answer_relevancy",
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval answer relevancy: {e}")
        raise

calculate_contextual_precision(evaluation_input)

Calculate contextual precision using DeepEval.

Contextual precision measures whether the retrieved context is relevant to the query and whether irrelevant context is ranked lower.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with contextual precision score

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If required fields are missing
Source code in uaef/integrations/deepeval_connector.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
def calculate_contextual_precision(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate contextual precision using DeepEval.

    Contextual precision measures whether the retrieved context is relevant
    to the query and whether irrelevant context is ranked lower.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with contextual precision score

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If required fields are missing
    """
    self._ensure_available()

    if not evaluation_input.context:
        return MetricScore(
            metric_name="deepeval_contextual_precision",
            score=None,
            reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
            metadata={"warning": "not_applicable", "reason": "no_context"},
        )

    try:
        from deepeval.metrics import ContextualPrecisionMetric
        from deepeval.test_case import LLMTestCase

        # Transform to DeepEval format
        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        if not deepeval_data.get("expected_output"):
            raise ValueError("Ground truth expected_output is required for contextual precision")

        # Create test case for DeepEval
        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            expected_output=deepeval_data["expected_output"],
            retrieval_context=deepeval_data["retrieval_context"],
        )

        # Calculate metric
        metric = ContextualPrecisionMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_contextual_precision",
            score=float(metric.score),
            reasoning=f"DeepEval contextual precision: {metric.reason if hasattr(metric, 'reason') else 'measures relevance of retrieved context'}",
            metadata={
                "library": "deepeval",
                "metric": "contextual_precision",
                "contexts_count": len(deepeval_data["retrieval_context"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval contextual precision: {e}")
        raise

calculate_contextual_recall(evaluation_input)

Calculate contextual recall using DeepEval.

Contextual recall measures whether all relevant information from the expected output can be found in the retrieved context.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with contextual recall score

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If ground truth is not provided
Source code in uaef/integrations/deepeval_connector.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def calculate_contextual_recall(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate contextual recall using DeepEval.

    Contextual recall measures whether all relevant information from the
    expected output can be found in the retrieved context.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with contextual recall score

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If ground truth is not provided
    """
    self._ensure_available()

    if not evaluation_input.context:
        return MetricScore(
            metric_name="deepeval_contextual_recall",
            score=None,
            reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
            metadata={"warning": "not_applicable", "reason": "no_context"},
        )

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_output:
        raise ValueError("Ground truth expected_output is required for contextual recall")

    try:
        from deepeval.metrics import ContextualRecallMetric
        from deepeval.test_case import LLMTestCase

        # Transform to DeepEval format
        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        # Create test case for DeepEval
        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            expected_output=deepeval_data["expected_output"],
            retrieval_context=deepeval_data["retrieval_context"],
        )

        # Calculate metric
        metric = ContextualRecallMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_contextual_recall",
            score=float(metric.score),
            reasoning=f"DeepEval contextual recall: {metric.reason if hasattr(metric, 'reason') else 'measures coverage of expected output in context'}",
            metadata={
                "library": "deepeval",
                "metric": "contextual_recall",
                "has_ground_truth": True,
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval contextual recall: {e}")
        raise

calculate_contextual_relevancy(evaluation_input)

Calculate contextual relevancy using DeepEval.

Contextual relevancy assesses whether the retrieved context is relevant to the user's input query.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input (requires context)

 required

Returns:

Type Description
MetricScore

MetricScore with contextual relevancy score
Source code in uaef/integrations/deepeval_connector.py 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
def calculate_contextual_relevancy(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate contextual relevancy using DeepEval.

    Contextual relevancy assesses whether the retrieved context is relevant
    to the user's input query.

    Args:
        evaluation_input: UAEF evaluation input (requires context)

    Returns:
        MetricScore with contextual relevancy score
    """
    self._ensure_available()

    if not evaluation_input.context:
        return MetricScore(
            metric_name="deepeval_contextual_relevancy",
            score=None,
            reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
            metadata={"warning": "not_applicable", "reason": "no_context"},
        )

    try:
        from deepeval.metrics import ContextualRelevancyMetric
        from deepeval.test_case import LLMTestCase

        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            retrieval_context=deepeval_data["retrieval_context"],
        )

        metric = ContextualRelevancyMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_contextual_relevancy",
            score=float(metric.score),
            reasoning=f"DeepEval contextual relevancy: {getattr(metric, 'reason', 'assesses relevance of retrieved context to query')}",
            metadata={
                "library": "deepeval",
                "metric": "contextual_relevancy",
                "contexts_count": len(deepeval_data["retrieval_context"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval contextual relevancy: {e}")
        raise

calculate_faithfulness(evaluation_input)

Calculate faithfulness using DeepEval.

Faithfulness measures whether the generated answer is factually consistent with the provided retrieval context (i.e., no hallucinated claims).

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input (requires context)

 required

Returns:

Type Description
MetricScore

MetricScore with faithfulness score
Source code in uaef/integrations/deepeval_connector.py 
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
def calculate_faithfulness(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate faithfulness using DeepEval.

    Faithfulness measures whether the generated answer is factually consistent
    with the provided retrieval context (i.e., no hallucinated claims).

    Args:
        evaluation_input: UAEF evaluation input (requires context)

    Returns:
        MetricScore with faithfulness score
    """
    self._ensure_available()

    if not evaluation_input.context:
        return MetricScore(
            metric_name="deepeval_faithfulness",
            score=None,
            reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
            metadata={"warning": "not_applicable", "reason": "no_context"},
        )

    try:
        from deepeval.metrics import FaithfulnessMetric
        from deepeval.test_case import LLMTestCase

        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            retrieval_context=deepeval_data["retrieval_context"],
        )

        metric = FaithfulnessMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_faithfulness",
            score=float(metric.score),
            reasoning=f"DeepEval faithfulness: {getattr(metric, 'reason', 'measures factual consistency with retrieval context')}",
            metadata={
                "library": "deepeval",
                "metric": "faithfulness",
                "contexts_count": len(deepeval_data["retrieval_context"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval faithfulness: {e}")
        raise

calculate_geval(evaluation_input, criteria, evaluation_steps=None)

Calculate G-Eval score using DeepEval with custom criteria.

G-Eval is a framework for LLM-based evaluation using custom criteria. It allows defining specific evaluation criteria and steps.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required
criteria str

Custom evaluation criteria description

 required
evaluation_steps Optional[List[str]]

Optional list of evaluation steps to follow

 None

Returns:

Type Description
MetricScore

MetricScore with G-Eval score

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If required fields are missing
Source code in uaef/integrations/deepeval_connector.py 
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
def calculate_geval(
    self,
    evaluation_input: EvaluationInput,
    criteria: str,
    evaluation_steps: Optional[List[str]] = None,
) -> MetricScore:
    """
    Calculate G-Eval score using DeepEval with custom criteria.

    G-Eval is a framework for LLM-based evaluation using custom criteria.
    It allows defining specific evaluation criteria and steps.

    Args:
        evaluation_input: UAEF evaluation input
        criteria: Custom evaluation criteria description
        evaluation_steps: Optional list of evaluation steps to follow

    Returns:
        MetricScore with G-Eval score

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If required fields are missing
    """
    self._ensure_available()

    if not criteria:
        raise ValueError("Criteria is required for G-Eval")

    try:
        from deepeval.metrics import GEval
        from deepeval.test_case import LLMTestCase

        # Transform to DeepEval format
        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        # Create test case for DeepEval
        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
        )

        # Add expected output if available
        if deepeval_data.get("expected_output"):
            test_case.expected_output = deepeval_data["expected_output"]

        # Create G-Eval metric with custom criteria
        metric_kwargs = {
            "name": "custom_geval",
            "criteria": criteria,
            "evaluation_params": [
                LLMTestCase.input,
                LLMTestCase.actual_output,
            ],
        }

        if evaluation_steps:
            metric_kwargs["evaluation_steps"] = evaluation_steps

        metric = GEval(**metric_kwargs)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_geval",
            score=float(metric.score),
            reasoning=f"DeepEval G-Eval: {metric.reason if hasattr(metric, 'reason') else criteria}",
            metadata={
                "library": "deepeval",
                "metric": "geval",
                "criteria": criteria,
                "has_evaluation_steps": evaluation_steps is not None,
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval G-Eval: {e}")
        raise

calculate_hallucination(evaluation_input)

Calculate hallucination score using DeepEval.

Hallucination metric detects claims in the agent's response that are not supported by the provided context documents.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with hallucination score (1.0 = no hallucination, 0.0 = high hallucination)

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If context is not provided
Source code in uaef/integrations/deepeval_connector.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
def calculate_hallucination(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate hallucination score using DeepEval.

    Hallucination metric detects claims in the agent's response that are not
    supported by the provided context documents.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with hallucination score (1.0 = no hallucination, 0.0 = high hallucination)

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If context is not provided
    """
    self._ensure_available()

    if not evaluation_input.context:
        return MetricScore(
            metric_name="deepeval_hallucination",
            score=None,
            reasoning="Not applicable: no retrieval context provided (metric requires RAG context)",
            metadata={"warning": "not_applicable", "reason": "no_context"},
        )

    try:
        from deepeval.metrics import HallucinationMetric
        from deepeval.test_case import LLMTestCase

        # Transform to DeepEval format
        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        # Create test case for DeepEval
        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            context=deepeval_data["context"],
        )

        # Calculate metric
        metric = HallucinationMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_hallucination",
            score=float(metric.score),
            reasoning=f"DeepEval hallucination: {metric.reason if hasattr(metric, 'reason') else 'detects unsupported claims in response'}",
            metadata={
                "library": "deepeval",
                "metric": "hallucination",
                "contexts_count": len(deepeval_data["context"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval hallucination: {e}")
        raise

calculate_metrics(evaluation_input, metric_names, geval_criteria=None)

Calculate multiple DeepEval metrics at once.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required
metric_names List[str]

List of metric names to calculate (e.g., ["contextual_precision", "hallucination"])

 required
geval_criteria Optional[str]

Optional criteria for G-Eval metric

 None

Returns:

Type Description
List[MetricScore]

List of MetricScore objects

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If an unknown metric name is provided
Source code in uaef/integrations/deepeval_connector.py 
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
def calculate_metrics(
    self,
    evaluation_input: EvaluationInput,
    metric_names: List[str],
    geval_criteria: Optional[str] = None,
) -> List[MetricScore]:
    """
    Calculate multiple DeepEval metrics at once.

    Args:
        evaluation_input: UAEF evaluation input
        metric_names: List of metric names to calculate
                     (e.g., ["contextual_precision", "hallucination"])
        geval_criteria: Optional criteria for G-Eval metric

    Returns:
        List of MetricScore objects

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If an unknown metric name is provided
    """
    self._ensure_available()

    metric_map = {
        "contextual_precision": self.calculate_contextual_precision,
        "contextual_recall": self.calculate_contextual_recall,
        "contextual_relevancy": self.calculate_contextual_relevancy,
        "hallucination": self.calculate_hallucination,
        "faithfulness": self.calculate_faithfulness,
        "answer_relevancy": self.calculate_answer_relevancy,
        "tool_correctness": self.calculate_tool_correctness,
    }

    scores = []
    for metric_name in metric_names:
        if metric_name == "geval":
            if not geval_criteria:
                logger.warning("G-Eval requires criteria, skipping")
                continue
            try:
                score = self.calculate_geval(evaluation_input, geval_criteria)
                scores.append(score)
            except Exception as e:
                logger.error(f"Error calculating geval: {e}")
                # Continue with remaining metrics per requirement 14.6
                continue
        elif metric_name in metric_map:
            try:
                score = metric_map[metric_name](evaluation_input)
                scores.append(score)
            except Exception as e:
                logger.error(f"Error calculating {metric_name}: {e}")
                # Continue with remaining metrics per requirement 14.6
                continue
        else:
            logger.warning(f"Unknown DeepEval metric: {metric_name}, skipping")
            continue

    return scores

calculate_role_adherence(evaluation_input)

Calculate role adherence using DeepEval's RoleAdherenceMetric.

Reads the chatbot persona strictly from ground_truth.expected_arguments["chatbot_role"]. Builds a ConversationalTestCase from trace.messages, delegates to DeepEval, and surfaces the score along with per-turn out-of-character verdicts in metadata.

Returns a not_applicable MetricScore if chatbot_role is missing or the filtered turn list is empty. Errors raised by DeepEval's measure() propagate to the caller (matches sibling methods).

Source code in uaef/integrations/deepeval_connector.py 
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
def calculate_role_adherence(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """Calculate role adherence using DeepEval's `RoleAdherenceMetric`.

    Reads the chatbot persona strictly from
    `ground_truth.expected_arguments["chatbot_role"]`. Builds a
    `ConversationalTestCase` from `trace.messages`, delegates to
    DeepEval, and surfaces the score along with per-turn out-of-character
    verdicts in metadata.

    Returns a `not_applicable` `MetricScore` if `chatbot_role` is missing
    or the filtered turn list is empty. Errors raised by DeepEval's
    `measure()` propagate to the caller (matches sibling methods).
    """
    if not self._deepeval_available:
        raise RuntimeError(
            "DeepEval library not available. Install with: pip install deepeval"
        )

    gt = evaluation_input.ground_truth
    role = (
        gt.expected_arguments.get("chatbot_role")
        if gt is not None and gt.expected_arguments
        else None
    )
    if not role:
        return MetricScore(
            metric_name="deepeval_role_adherence",
            score=None,
            reasoning="Cannot evaluate: expected_arguments['chatbot_role'] is required",
            metadata={"warning": "not_applicable"},
        )

    turns = _messages_to_deepeval_turns(evaluation_input.trace.messages)
    if not turns:
        return MetricScore(
            metric_name="deepeval_role_adherence",
            score=None,
            reasoning="Cannot evaluate: no user/assistant messages",
            metadata={"warning": "not_applicable"},
        )

    from deepeval.metrics import RoleAdherenceMetric
    from deepeval.test_case import ConversationalTestCase

    test_case = ConversationalTestCase(turns=turns, chatbot_role=role)
    metric = RoleAdherenceMetric(model=self._model)
    try:
        _measure(metric, test_case)
    except Exception as e:
        logger.error(f"Error calculating DeepEval role adherence: {e}")
        raise

    # Per-verdict access uses getattr for resilience against DeepEval field
    # renames; outer attributes are required and we want loud failures.
    verdicts = metric.out_of_character_verdicts.verdicts or []
    out_of_character_turns = [
        {
            "turn_index": getattr(v, "index", None),
            "ai_message": getattr(v, "ai_message", None),
            "reason": getattr(v, "reason", None),
        }
        for v in verdicts
    ]

    assistant_turn_count = sum(1 for t in turns if t.role == "assistant")

    if metric.score is None:
        return MetricScore(
            metric_name="deepeval_role_adherence",
            score=None,
            reasoning="Cannot evaluate: DeepEval returned no score (likely zero assistant turns)",
            metadata={
                "warning": "not_applicable",
                "library": "deepeval",
                "metric": "role_adherence",
                "out_of_character_turns": out_of_character_turns,
                "assistant_turn_count": assistant_turn_count,
            },
        )

    return MetricScore(
        metric_name="deepeval_role_adherence",
        score=float(metric.score),
        reasoning=(
            getattr(metric, "reason", None)
            or "DeepEval role adherence (no reason returned)"
        ),
        metadata={
            "library": "deepeval",
            "metric": "role_adherence",
            "out_of_character_turns": out_of_character_turns,
            "assistant_turn_count": assistant_turn_count,
        },
    )

calculate_tool_correctness(evaluation_input)

Calculate tool correctness using DeepEval.

Tool correctness measures how accurately the agent selected and used tools compared to the expected tool calls.

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
MetricScore

MetricScore with tool correctness score

Raises:

Type Description
ImportError

If the DeepEval backend ('integrations' extra) is not installed
ValueError

If ground truth tool calls are not provided
Source code in uaef/integrations/deepeval_connector.py 
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def calculate_tool_correctness(
    self, evaluation_input: EvaluationInput
) -> MetricScore:
    """
    Calculate tool correctness using DeepEval.

    Tool correctness measures how accurately the agent selected and used
    tools compared to the expected tool calls.

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        MetricScore with tool correctness score

    Raises:
        ImportError: If the DeepEval backend ('integrations' extra) is not installed
        ValueError: If ground truth tool calls are not provided
    """
    self._ensure_available()

    if not evaluation_input.ground_truth or not evaluation_input.ground_truth.expected_tool_calls:
        raise ValueError("Ground truth expected_tool_calls is required for tool correctness")

    try:
        from deepeval.metrics import ToolCorrectnessMetric
        from deepeval.test_case import LLMTestCase, ToolCall as DeepEvalToolCall

        # Transform to DeepEval format
        deepeval_data = self.transform_to_deepeval_format(evaluation_input)

        if "tools_called" not in deepeval_data or "expected_tools" not in deepeval_data:
            # Agent made no tool calls but ground truth expected some → score is 0
            actual = deepeval_data.get("tools_called", [])
            expected = deepeval_data.get("expected_tools", [])
            if not actual and expected:
                return MetricScore(
                    metric_name="deepeval_tool_correctness",
                    score=0.0,
                    reasoning="Agent made no tool calls but ground truth expected tool usage",
                    metadata={
                        "library": "deepeval",
                        "metric": "tool_correctness",
                        "actual_tool_calls": 0,
                        "expected_tool_calls": len(expected),
                    },
                )
            raise ValueError("Tool calls and expected tools are required")

        # Convert to DeepEval ToolCall objects
        tools_called = [
            DeepEvalToolCall(
                name=tc["name"],
                input_parameters=tc.get("arguments", {}),
            )
            for tc in deepeval_data["tools_called"]
        ]
        expected_tools = [
            DeepEvalToolCall(
                name=tc["name"],
                input_parameters=tc.get("arguments", {}),
            )
            for tc in deepeval_data["expected_tools"]
        ]

        # Create test case for DeepEval
        test_case = LLMTestCase(
            input=deepeval_data["input"],
            actual_output=deepeval_data["actual_output"],
            tools_called=tools_called,
            expected_tools=expected_tools,
        )

        # Calculate metric
        metric = ToolCorrectnessMetric(model=self._model)
        _measure(metric, test_case)

        return MetricScore(
            metric_name="deepeval_tool_correctness",
            score=float(metric.score),
            reasoning=f"DeepEval tool correctness: {metric.reason if hasattr(metric, 'reason') else 'measures correctness of tool selection and usage'}",
            metadata={
                "library": "deepeval",
                "metric": "tool_correctness",
                "actual_tool_calls": len(deepeval_data["tools_called"]),
                "expected_tool_calls": len(deepeval_data["expected_tools"]),
            },
        )

    except Exception as e:
        logger.error(f"Error calculating DeepEval tool correctness: {e}")
        raise

is_available()

Check if DeepEval library is available.

Returns:

Type Description
bool

True if DeepEval is installed and available, False otherwise
Source code in uaef/integrations/deepeval_connector.py 
152
153
154
155
156
157
158
159
def is_available(self) -> bool:
    """
    Check if DeepEval library is available.

    Returns:
        True if DeepEval is installed and available, False otherwise
    """
    return self._deepeval_available

list_available_metrics()

List all metrics supported by this connector.

Returns:

Type Description
List[str]

List of metric names
Source code in uaef/integrations/deepeval_connector.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def list_available_metrics(self) -> List[str]:
    """
    List all metrics supported by this connector.

    Returns:
        List of metric names
    """
    return [
        "contextual_precision",
        "contextual_recall",
        "contextual_relevancy",
        "hallucination",
        "faithfulness",
        "answer_relevancy",
        "tool_correctness",
        "geval",
    ]

transform_to_deepeval_format(evaluation_input)

Transform UAEF EvaluationInput to DeepEval format.

DeepEval expects data in a specific format with fields like: - input: The user's question/query - actual_output: The agent's response - retrieval_context: List of context documents - expected_output: Expected correct answer - context: List of context strings for hallucination detection - tools_called: List of tool calls made by the agent - expected_tools: List of expected tool calls

Parameters:

Name Type Description Default
evaluation_input EvaluationInput

UAEF evaluation input

 required

Returns:

Type Description
Dict[str, Any]

Dictionary in DeepEval format

Raises:

Type Description
ValueError

If required fields are missing
Source code in uaef/integrations/deepeval_connector.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def transform_to_deepeval_format(
    self, evaluation_input: EvaluationInput
) -> Dict[str, Any]:
    """
    Transform UAEF EvaluationInput to DeepEval format.

    DeepEval expects data in a specific format with fields like:
    - input: The user's question/query
    - actual_output: The agent's response
    - retrieval_context: List of context documents
    - expected_output: Expected correct answer
    - context: List of context strings for hallucination detection
    - tools_called: List of tool calls made by the agent
    - expected_tools: List of expected tool calls

    Args:
        evaluation_input: UAEF evaluation input

    Returns:
        Dictionary in DeepEval format

    Raises:
        ValueError: If required fields are missing
    """
    self._ensure_available()

    trace = evaluation_input.trace
    ground_truth = evaluation_input.ground_truth

    # Extract question (first user message)
    question = None
    actual_output = None

    for message in trace.messages:
        if message.role == "user" and question is None:
            question = message.content
        elif message.role == "assistant" and actual_output is None:
            actual_output = message.content

    # Fallback: use first message content as question if no user role found
    if not question and trace.messages:
        question = trace.messages[0].content
        logger.warning("No user message found in trace, using first message as question")

    if not question:
        raise ValueError("No messages found in trace")

    # Fallback: use last message content as answer if no assistant role found
    if not actual_output and len(trace.messages) > 1:
        actual_output = trace.messages[-1].content
        logger.warning("No assistant message found in trace, using last message as answer")

    if not actual_output:
        raise ValueError("No assistant response found in trace")

    # Build DeepEval format
    deepeval_data = {
        "input": question,
        "actual_output": actual_output,
        "retrieval_context": evaluation_input.context,
        "context": evaluation_input.context,  # For hallucination detection
    }

    # Add ground truth if available
    if ground_truth:
        if ground_truth.expected_output:
            deepeval_data["expected_output"] = ground_truth.expected_output

        if ground_truth.expected_tool_calls:
            deepeval_data["expected_tools"] = [
                {
                    "name": tc.name,
                    "arguments": tc.arguments,
                }
                for tc in ground_truth.expected_tool_calls
            ]

    # Add actual tool calls
    if trace.tool_calls:
        deepeval_data["tools_called"] = [
            {
                "name": tc.name,
                "arguments": tc.arguments,
            }
            for tc in trace.tool_calls
        ]

    return deepeval_data