Implementation:Explodinggradients Ragas Discrete Metric Decorator
| Knowledge Sources | Domains | Last Updated |
|---|---|---|
| explodinggradients/ragas | LLM Evaluation, Metric Design | 2026-02-10 |
Overview
The Discrete Metric Decorator transforms plain Python functions into full metric objects with discrete categorical outputs, automatic input validation, and standardized scoring interfaces.
Description
The discrete_metric decorator function (lines 128-178 of src/ragas/metrics/discrete.py) is a shorthand that delegates to create_metric_decorator() (lines 83-358 of src/ragas/metrics/decorator.py). The decorator factory introspects the target function's signature, creates a dynamic Pydantic model for input validation, and wraps the function in a dynamically defined CustomMetric dataclass that inherits from SimpleBaseMetric and the appropriate validator mixin. The resulting metric instance exposes score(**kwargs), ascore(**kwargs), batch_score(), abatch_score(), and direct __call__ methods. Output values are validated against the specified allowed_values list, and plain return values are automatically wrapped in MetricResult objects.
Usage
Use the Discrete Metric Decorator when:
- Defining evaluation metrics with pure Python logic (no LLM dependency)
- Creating pass/fail or multi-category metrics from simple functions
- Wanting automatic input validation based on function type annotations
- Needing metrics that integrate into Ragas evaluation pipelines with standard interfaces
Code Reference
Source Location: src/ragas/metrics/discrete.py, lines 128-178 (discrete_metric shorthand); src/ragas/metrics/decorator.py, lines 83-358 (create_metric_decorator factory)
Signature:
def discrete_metric(
*,
name: Optional[str] = None,
allowed_values: Optional[List[str]] = None,
**metric_params: Any,
) -> Callable[[Callable[..., Any]], DiscreteMetricProtocol]
Import:
from ragas.metrics import discrete_metric
Result Type Protocol:
class DiscreteMetricProtocol(Protocol):
name: str
allowed_values: List[str]
def score(self, **kwargs) -> MetricResult: ...
async def ascore(self, **kwargs) -> MetricResult: ...
def batch_score(self, inputs: List[Dict[str, Any]]) -> List[MetricResult]: ...
async def abatch_score(self, inputs: List[Dict[str, Any]]) -> List[MetricResult]: ...
def __call__(self, *args, **kwargs): ...
I/O Contract
Inputs (decorator parameters):
| Parameter | Type | Required | Description |
|---|---|---|---|
name |
Optional[str] |
No | Name for the metric (defaults to the decorated function's name) |
allowed_values |
Optional[List[str]] |
No | Allowed output categories (default: ["pass", "fail"])
|
**metric_params |
Any |
No | Additional parameters passed to the metric initialization |
Inputs (score method):
| Parameter | Type | Required | Description |
|---|---|---|---|
**kwargs |
Any |
Yes | Keyword arguments matching the decorated function's parameter names and types |
Outputs:
| Output | Type | Description |
|---|---|---|
| Metric result | MetricResult |
Contains .value (one of the allowed values or None on error) and .reason (explanation or error message)
|
Usage Examples
Basic pass/fail metric:
from ragas.metrics import discrete_metric
@discrete_metric(name="has_greeting", allowed_values=["pass", "fail"])
def has_greeting(response: str) -> str:
"""Check if the response contains a greeting."""
greetings = ["hello", "hi", "hey", "greetings"]
if any(g in response.lower() for g in greetings):
return "pass"
return "fail"
# Use via score() with keyword arguments (validated)
result = has_greeting.score(response="Hello! How can I help you?")
print(result.value) # "pass"
print(result.reason) # None (pure logic, no LLM reasoning)
# Or use as direct function call
raw_result = has_greeting(response="Hello! How can I help you?")
Multi-category sentiment metric:
from ragas.metrics import discrete_metric
@discrete_metric(name="sentiment", allowed_values=["positive", "neutral", "negative"])
def sentiment_analysis(user_input: str, response: str) -> str:
"""Analyze sentiment of the response."""
positive_words = ["great", "good", "excellent", "wonderful", "happy"]
negative_words = ["bad", "poor", "terrible", "awful", "sad"]
response_lower = response.lower()
if any(w in response_lower for w in positive_words):
return "positive"
elif any(w in response_lower for w in negative_words):
return "negative"
return "neutral"
result = sentiment_analysis.score(
user_input="How was your day?",
response="It was great!",
)
print(result.value) # "positive"
Async metric function:
from ragas.metrics import discrete_metric
@discrete_metric(name="external_check", allowed_values=["pass", "fail"])
async def external_validation(response: str) -> str:
"""Validate response against an external service."""
# Async operations are supported natively
import httpx
async with httpx.AsyncClient() as client:
resp = await client.post("https://api.example.com/validate", json={"text": response})
return "pass" if resp.json()["valid"] else "fail"
# Use ascore for async metrics
import asyncio
result = asyncio.run(external_validation.ascore(response="Test response"))
print(result.value)
Batch scoring:
inputs = [
{"response": "Hello there!"},
{"response": "Goodbye."},
{"response": "Hi, how are you?"},
]
results = has_greeting.batch_score(inputs)
for r in results:
print(f"{r.value}")
# Output: pass, fail, pass