Score a single response

The scorer is a single function. Give it a question, an answer, a list of contexts, and optionally a ground truth. It returns an EvalScore. RAGAS itself is a heavy import that pulls in datasets, numpy, and transitively torch, so the score function imports RAGAS lazily and falls back to a lightweight heuristic when RAGAS is unavailable.

ragas_utils.py
python
def score_answer(question, answer, contexts, ground_truth=None):
    """Score one answer. Tries RAGAS first, falls back to lightweight heuristic."""
    try:
        from ragas import evaluate
        from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
        from datasets import Dataset

        data = {"question": [question], "answer": [answer], "contexts": [contexts]}
        metrics = [faithfulness, answer_relevancy, context_precision]
        if ground_truth:
            data["ground_truth"] = [ground_truth]
            metrics.append(context_recall)

        ds = Dataset.from_dict(data)
        result = evaluate(ds, metrics=metrics)
        row = result.to_pandas().iloc[0].to_dict()
        return EvalScore(
            faithfulness=float(row.get("faithfulness", 0.0) or 0.0),
            answer_relevance=float(row.get("answer_relevancy", 0.0) or 0.0),
            context_precision=float(row.get("context_precision", 0.0) or 0.0),
            context_recall=float(row["context_recall"]) if ground_truth else None,
        )
    except Exception as e:
        logger.info(f"RAGAS unavailable or failed ({e}); using lightweight scorer.")
        return _lightweight_scores(question, answer, contexts, ground_truth)

Lazy import keeps module cost low and lets the scorer degrade gracefully when RAGAS is not installed or fails to run, which is common in local dev without provider keys.

rag_graph.py
python
async def _evaluate_node(state: GraphState) -> GraphState:
    """Score the answer with RAGAS (lazy import) if requested."""
    if not state.get("with_evaluation"):
        return {"eval_score": None}

    from ragas_utils import score_answer

    score = score_answer(
        question=state["question"],
        answer=state.get("answer", ""),
        contexts=state.get("contexts", []) or [],
        ground_truth=None,
    )
    return {"eval_score": score.model_dump()}

Evaluation is opt-in via the with_evaluation flag. Scoring adds a few seconds per request because RAGAS itself calls an LLM. Turn it on for eval runs, off for user-facing latency-critical traffic.

terminal
bash
curl -X POST http://localhost:8000/advanced-rag/ask \
  -H "Content-Type: application/json" \
  -d '{
    "question": "What embedding model does the pipeline use?",
    "rewrite_n": 3,
    "anonymize": false,
    "with_evaluation": true
  }' | jq '.eval_score'

Turn on with_evaluation and you get an eval_score object attached to every answer with faithfulness, answer_relevance, and context_precision. Context_recall stays null unless you provide a ground truth.