Context Recall in Machine Learning

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

# Prepare evaluation dataset
eval_data = {
    "question": [
        "What is the capital of France?",
        "When was Python created?"
    ],
    "answer": [
        "The capital of France is Paris, which is also the largest city.",
        "Python was created by Guido van Rossum and released in 1991."
    ],
    "contexts": [
        ["Paris is the capital and most populous city of France."],
        ["Python is a high-level programming language.",
         "Guido van Rossum began working on Python in the late 1980s."]
    ],
    "ground_truth": [
        "Paris is the capital of France and its largest city.",
        "Python was created by Guido van Rossum and first released in 1991."
    ]
}

dataset = Dataset.from_dict(eval_data)
result = evaluate(dataset, metrics=[context_recall])
print(f"Context Recall: {result['context_recall']:.4f}")
# Per-sample scores
for i, score in enumerate(result.scores):
    print(f"  Sample {i}: {score['context_recall']:.4f}")

import openai
import json
from typing import List, Tuple

client = openai.OpenAI()

def decompose_claims(ground_truth: str) -> List[str]:
    """Break ground-truth answer into atomic claims."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": (
                "Decompose the following text into independent, "
                "atomic factual claims. Each claim should be a "
                "single, self-contained statement that can be "
                "verified independently. Return a JSON array of strings."
            )},
            {"role": "user", "content": ground_truth}
        ],
        response_format={"type": "json_object"},
        temperature=0.0
    )
    result = json.loads(response.choices[0].message.content)
    return result.get("claims", [])


def check_attribution(
    claim: str,
    contexts: List[str]
) -> Tuple[bool, str]:
    """Check if a claim is attributable to retrieved contexts."""
    context_text = "\n---\n".join(
        f"Context {i+1}: {c}" for i, c in enumerate(contexts)
    )
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": (
                "Determine if the given claim can be attributed to "
                "the provided context passages. A claim is attributed "
                "if the context contains information that supports or "
                "implies the claim. Respond with JSON: "
                '{"attributed": true/false, "reasoning": "..."}'
            )},
            {"role": "user", "content": (
                f"Claim: {claim}\n\nContexts:\n{context_text}"
            )}
        ],
        response_format={"type": "json_object"},
        temperature=0.0
    )
    result = json.loads(response.choices[0].message.content)
    return result["attributed"], result["reasoning"]


def compute_context_recall(
    ground_truth: str,
    contexts: List[str]
) -> dict:
    """Compute context recall with full claim-level details."""
    claims = decompose_claims(ground_truth)
    if not claims:
        return {"score": 0.0, "claims": [], "error": "No claims extracted"}

    results = []
    attributed_count = 0
    for claim in claims:
        is_attributed, reasoning = check_attribution(claim, contexts)
        results.append({
            "claim": claim,
            "attributed": is_attributed,
            "reasoning": reasoning
        })
        if is_attributed:
            attributed_count += 1

    return {
        "score": attributed_count / len(claims),
        "attributed_count": attributed_count,
        "total_claims": len(claims),
        "claims": results
    }

# Usage
result = compute_context_recall(
    ground_truth="Paris is the capital of France. It has a population of over 2 million.",
    contexts=["Paris is the capital city of France, located on the Seine river."]
)
print(f"Score: {result['score']:.2f}")
for c in result['claims']:
    status = 'YES' if c['attributed'] else 'NO'
    print(f"  [{status}] {c['claim']}")

import asyncio
import openai
import json
from typing import List, Dict
from dataclasses import dataclass, field


@dataclass
class EvalSample:
    query: str
    ground_truth: str
    contexts: List[str]
    score: float = 0.0
    claim_details: List[dict] = field(default_factory=list)


async def evaluate_sample(client, sample, semaphore):
    async with semaphore:
        # Decompose ground truth into claims
        resp = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content":
                    "Extract atomic factual claims. "
                    "Return JSON: {\"claims\": [\"...\"]}"},
                {"role": "user", "content": sample.ground_truth}
            ],
            response_format={"type": "json_object"},
            temperature=0.0
        )
        claims = json.loads(
            resp.choices[0].message.content
        ).get("claims", [])
        if not claims:
            return sample

        # Check attribution for each claim in parallel
        ctx = "\n".join(
            f"[{i+1}] {c}" for i, c in enumerate(sample.contexts)
        )
        tasks = []
        for claim in claims:
            tasks.append(client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {"role": "system", "content":
                        "Is this claim supported by the contexts? "
                        'JSON: {"attributed": bool}'},
                    {"role": "user", "content":
                        f"Claim: {claim}\n\n{ctx}"}
                ],
                response_format={"type": "json_object"},
                temperature=0.0
            ))
        verdicts = await asyncio.gather(*tasks)

        attributed = 0
        for claim, v in zip(claims, verdicts):
            result = json.loads(v.choices[0].message.content)
            is_attr = result.get("attributed", False)
            sample.claim_details.append({
                "claim": claim, "attributed": is_attr
            })
            if is_attr:
                attributed += 1
        sample.score = attributed / len(claims)
        return sample


async def batch_evaluate(
    samples: List[EvalSample], max_concurrent: int = 10
) -> Dict:
    client = openai.AsyncOpenAI()
    sem = asyncio.Semaphore(max_concurrent)
    results = await asyncio.gather(*[
        evaluate_sample(client, s, sem) for s in samples
    ])
    scores = [s.score for s in results]
    return {
        "mean_context_recall": sum(scores) / len(scores),
        "num_samples": len(scores),
        "samples": results
    }

# Usage
samples = [
    EvalSample(
        query="What is BERT?",
        ground_truth="BERT is a transformer model by Google.",
        contexts=["BERT is a pre-trained transformer model."]
    ),
]
results = asyncio.run(batch_evaluate(samples))
print(f"Mean Context Recall: {results['mean_context_recall']:.3f}")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from typing import List


class ContextRecallEvaluator:
    def __init__(self, model_name: str = "gpt-4o"):
        self.llm = ChatOpenAI(model=model_name, temperature=0)

    def _decompose(self, text: str) -> List[str]:
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract atomic factual claims. "
             "Return JSON: {{\"claims\": [\"...\"]}}"),
            ("user", "{text}")
        ])
        chain = prompt | self.llm | JsonOutputParser()
        result = chain.invoke({"text": text})
        return result.get("claims", [])

    def _attribute(self, claim: str, context: str) -> bool:
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Is this claim supported by the context? "
             "Return JSON: {{\"attributed\": true/false}}"),
            ("user", "Claim: {claim}\nContext: {context}")
        ])
        chain = prompt | self.llm | JsonOutputParser()
        result = chain.invoke({
            "claim": claim, "context": context
        })
        return result.get("attributed", False)

    def evaluate(
        self, ground_truth: str, contexts: List[str]
    ) -> dict:
        claims = self._decompose(ground_truth)
        combined = "\n\n".join(contexts)
        verdicts = [
            {"claim": c, "attributed": self._attribute(c, combined)}
            for c in claims
        ]
        attr = sum(1 for v in verdicts if v["attributed"])
        return {
            "score": attr / len(claims) if claims else 0.0,
            "claims": verdicts
        }

# Usage
evaluator = ContextRecallEvaluator()
result = evaluator.evaluate(
    ground_truth="TensorFlow was developed by Google Brain "
                 "and released in 2015.",
    contexts=[
        "TensorFlow is an open-source ML framework by Google.",
        "It was first released on November 9, 2015."
    ]
)
print(f"Context Recall: {result['score']:.2f}")

import json
import sys
from pathlib import Path
from ragas import evaluate
from ragas.metrics import context_recall
from datasets import Dataset


def load_eval_dataset(path: str) -> Dataset:
    """Load evaluation dataset from JSON file."""
    with open(path) as f:
        data = json.load(f)
    return Dataset.from_dict(data)


def run_context_recall_check(
    dataset_path: str,
    threshold: float = 0.75,
    output_path: str = "eval_results.json"
) -> bool:
    """Run context recall evaluation and check against threshold."""
    dataset = load_eval_dataset(dataset_path)
    result = evaluate(dataset, metrics=[context_recall])

    mean_score = result["context_recall"]
    per_sample = [
        {"query": q, "score": s["context_recall"]}
        for q, s in zip(dataset["question"], result.scores)
    ]

    # Find failing samples
    failures = [s for s in per_sample if s["score"] < threshold]

    report = {
        "metric": "context_recall",
        "mean_score": round(mean_score, 4),
        "threshold": threshold,
        "passed": mean_score >= threshold,
        "num_samples": len(per_sample),
        "num_failures": len(failures),
        "failures": failures[:10]  # Top 10 worst
    }

    Path(output_path).write_text(json.dumps(report, indent=2))
    print(f"Context Recall: {mean_score:.4f} (threshold: {threshold})")
    print(f"Status: {'PASS' if report['passed'] else 'FAIL'}")

    if failures:
        print(f"\n{len(failures)} samples below threshold:")
        for f in failures[:5]:
            print(f"  {f['query'][:60]}... → {f['score']:.3f}")

    return report["passed"]


if __name__ == "__main__":
    dataset_path = sys.argv[1] if len(sys.argv) > 1 \
        else "eval_dataset.json"
    threshold = float(sys.argv[2]) if len(sys.argv) > 2 else 0.75

    passed = run_context_recall_check(dataset_path, threshold)
    sys.exit(0 if passed else 1)  # Non-zero exit fails CI