Ragas — Other Metrics, Agents & Custom¶

from ragas.metrics import DiscreteMetric, NumericMetric, RankingMetric

plain = DiscreteMetric(
    name="plain_language",
    prompt="Does the response avoid jargon?\nQuestion: {user_input}\nResponse: {response}\nAnswer pass or fail.",
    allowed_values=["pass", "fail"],
)
clarity = NumericMetric(name="clarity", prompt="Rate clarity from 0 to 10.\nResponse: {response}", allowed_values=(0.0, 10.0))
order = RankingMetric(name="rank", prompt="Rank the candidates from best to worst.\nCandidates: {candidates}", allowed_values=3)

result = await plain.ascore(llm=llm, user_input="How do I reset my password?", response="Use the reset link on the login page.")
result.value          # "pass" or "fail"  (for DiscreteMetric a value outside allowed_values fails validation and is retried)
result.reason         # the judge's explanation
result.traces         # {"input": rendered prompt, "output": {"reason": ..., "value": ...}}
plain.get_variables() # ['user_input', 'response']

from ragas.prompt import Prompt

prompt = Prompt(
    "Does the response avoid jargon?\nResponse: {response}\nAnswer with value pass or fail and a reason.",
    examples=[
        ({"response": "Reset your password from the login page."}, {"value": "pass", "reason": "plain words"}),
        ({"response": "Invoke the credential rotation workflow."}, {"value": "fail", "reason": "jargon"}),
    ],
)
plain = DiscreteMetric(name="plain_language", prompt=prompt, allowed_values=["pass", "fail"])
plain.save("plain_language.json")                         # version the prompt with your code
plain = DiscreteMetric.load("plain_language.json")

from ragas.metrics import discrete_metric, numeric_metric


@discrete_metric(name="has_citation", allowed_values=["yes", "no"])
def has_citation(response: str) -> str:
    return "yes" if "[1]" in response else "no"


@numeric_metric(name="brevity", allowed_values=(0.0, 1.0))
def brevity(response: str, max_words: int = 50) -> float:
    return min(1.0, max_words / max(len(response.split()), 1))


has_citation.score(response="See [1].")      # MetricResult(value=yes)
await brevity.ascore(response="a b c")        # MetricResult(value=1.0)
has_citation(response="x")                    # plain call returns the raw function result: "no"

from ragas.metrics.collections import DomainSpecificRubrics, InstanceSpecificRubrics

rubrics = {
    "score1_description": "Gives wrong refund terms.",
    "score2_description": "Right window, wrong conditions.",
    "score3_description": "Right terms, missing the exception for sale items.",
    "score4_description": "Right terms with one minor omission.",
    "score5_description": "Complete and correct terms.",
}
metric = DomainSpecificRubrics(llm=llm, rubrics=rubrics, with_reference=True)
result = await metric.ascore(user_input=q, response=a, reference=ref)    # value is the raw 1-5 integer as float, reason is the feedback
per_sample = await InstanceSpecificRubrics(llm=llm).ascore(rubrics=rubrics, user_input=q, response=a)

from ragas.messages import AIMessage, HumanMessage, ToolCall, ToolMessage
from ragas.metrics.collections import AgentGoalAccuracy, ToolCallAccuracy, ToolCallF1

conversation = [
    HumanMessage(content="Book a table for 2 at 7pm"),
    AIMessage(content="Checking", tool_calls=[ToolCall(name="check_availability", args={"party": 2, "time": "19:00"})]),
    ToolMessage(content="available"),
    AIMessage(content="Booking", tool_calls=[ToolCall(name="book_table", args={"party": 2, "time": "19:00"})]),
    ToolMessage(content="confirmed #42"),
    AIMessage(content="Your table is booked, confirmation #42."),
]
expected = [ToolCall(name="check_availability", args={"party": 2, "time": "19:00"}),
            ToolCall(name="book_table", args={"party": 2, "time": "19:00"})]

(await ToolCallAccuracy().ascore(user_input=conversation, reference_tool_calls=expected)).value                 # 1.0
(await ToolCallAccuracy().ascore(user_input=conversation, reference_tool_calls=expected[::-1])).value           # 0.0, order differs
(await ToolCallAccuracy(strict_order=False).ascore(user_input=conversation, reference_tool_calls=expected[::-1])).value   # 1.0
(await ToolCallF1().ascore(user_input=conversation, reference_tool_calls=expected[:1])).value                   # 0.67, one extra call
(await AgentGoalAccuracy(llm=llm).ascore(user_input=conversation, reference="Table for 2 booked at 19:00")).value

from pydantic import BaseModel
from ragas.llms import InstructorBaseRagasLLM
from ragas.metrics import MetricResult
from ragas.metrics.collections import BaseMetric


class Verdict(BaseModel):
    polite: bool
    reason: str


class Politeness(BaseMetric):                                   # LLM-judged
    def __init__(self, llm: InstructorBaseRagasLLM, name: str = "politeness", **kwargs):
        self.llm = llm
        super().__init__(name=name, **kwargs)

    async def ascore(self, response: str) -> MetricResult:
        verdict = await self.llm.agenerate(f"Is this reply polite? Reply: {response}", Verdict)
        return MetricResult(value=1.0 if verdict.polite else 0.0, reason=verdict.reason)


class WithinWordLimit(BaseMetric):                              # pure Python, no llm attribute
    def __init__(self, limit: int = 120, name: str = "within_word_limit", **kwargs):
        self.limit = limit
        super().__init__(name=name, **kwargs)

    async def ascore(self, response: str) -> MetricResult:
        return MetricResult(value=float(len(response.split()) <= self.limit))

import asyncio
import json

from openai import AsyncOpenAI
from ragas.llms import llm_factory
from sklearn.metrics import cohen_kappa_score, confusion_matrix

# evals/judge_labels.jsonl rows: {"user_input": ..., "response": ..., "human": "pass" | "fail"}
rows = [json.loads(line) for line in open("evals/judge_labels.jsonl")]


async def judge_all(metric, rows, concurrency=8):
    llm = llm_factory("gpt-4o-mini", client=AsyncOpenAI(), temperature=0)     # client created inside the loop
    sem = asyncio.Semaphore(concurrency)

    async def one(row):
        async with sem:
            result = await metric.ascore(llm=llm, user_input=row["user_input"], response=row["response"])
            return result.value

    return await asyncio.gather(*[one(r) for r in rows])


predicted = asyncio.run(judge_all(plain, rows))               # `plain` is the DiscreteMetric defined above
human = [r["human"] for r in rows]
print(cohen_kappa_score(human, predicted))                    # also: plain.get_correlation(human, predicted)
print(confusion_matrix(human, predicted, labels=["pass", "fail"]))