Ragas for Streaming RAG Responses: Evaluating Partial Outputs

Why Streaming Breaks Your Evaluation Pipeline

Every RAG demo evaluates a finished answer. The retriever fetches chunks, the generator produces a complete response, and Ragas scores faithfulness, context precision, and answer relevancy against that finished text. It's clean. It's also not how production chat interfaces work anymore.

Users expect tokens to appear as they're generated. Your frontend renders a streaming response character by character, and by the time the last token lands, the user has already read most of the answer and possibly acted on it. If the answer starts hallucinating in the first sentence, the user has already been misled before your evaluation pipeline even has a complete string to score.

This creates a real gap. Ragas, LangSmith, and most RAG evaluation tooling assume you have the full generated text before you compute anything. Faithfulness checks need the complete claim set. Answer relevancy needs the complete answer to compare against the query. Context recall needs the whole response to check ground-truth overlap. None of these were designed with partial, growing strings in mind.

But teams shipping streaming chat products still need answers to real questions: Is this response drifting off-topic halfway through? Is it about to hallucinate a claim the retrieved context doesn't support? Should we cut the stream short and regenerate? Waiting for the full response to evaluate it defeats the purpose of streaming in the first place — you'd be adding evaluation latency back onto a UX optimization built specifically to remove latency.

This article walks through practical patterns for applying Ragas metrics to streaming RAG outputs: what breaks, what still works, and how to build an evaluation layer that gives you signal on partial generations without pretending streaming text is static text.

Before adapting anything, it helps to understand exactly why streaming and batch-style evaluation don't compose naturally.

Ragas metrics like Faithfulness work by decomposing the answer into atomic claims, then checking each claim against the retrieved context using an LLM judge. That decomposition step assumes sentence boundaries exist and that the claims are complete thoughts. Feed it "The mitochondria is the pow" and the claim extractor either fails silently or extracts a claim that isn't actually what the model meant to say.

AnswerRelevancy works by generating synthetic questions from the answer and comparing their embeddings to the original query. A half-formed answer produces synthetic questions that don't reflect the eventual, complete intent — so your relevancy score on a partial string is not a leading indicator of the final relevancy score, it's just noise.

ContextRecall and ContextPrecision are less sensitive to this because they primarily operate on the retrieved contexts rather than the generated text, so these two metrics are actually your best friends in a streaming setup — more on that below.

The practical takeaway: don't try to force claim-based metrics to run token-by-token. Instead, redesign your evaluation checkpoints around what's actually stable at each point in the stream — the retrieved context is stable from token one, and the generated text becomes progressively more stable as sentences complete.

Chunking the Stream Into Evaluable Units

The fix is to stop treating "the response" as one string that eventually completes, and start treating it as a sequence of evaluable units. The natural unit boundary is the sentence, not the token.

Here's a simple sentence-boundary buffer that sits between your token stream and your evaluation calls:

import re

class SentenceBuffer:
    """Accumulates streamed tokens and yields complete sentences
    as soon as they can be reliably evaluated."""

    SENTENCE_END = re.compile(r"(?<=[.!?])\s+")

    def __init__(self):
        self.buffer = ""
        self.emitted = []

    def push(self, token: str):
        self.buffer += token
        parts = self.SENTENCE_END.split(self.buffer)

        # Everything except the last fragment is a complete sentence
        complete, self.buffer = parts[:-1], parts[-1]

        new_sentences = []
        for sentence in complete:
            sentence = sentence.strip()
            if sentence:
                self.emitted.append(sentence)
                new_sentences.append(sentence)
        return new_sentences

    def full_text_so_far(self):
        return " ".join(self.emitted + ([self.buffer] if self.buffer else []))

This gives you a clean signal: every time push() returns a non-empty list, you have one or more complete sentences you can safely hand to an evaluator. You're not evaluating "The mitochondria is the pow" — you're evaluating "The mitochondria is the powerhouse of the cell." once it's actually complete.

The tradeoff is latency versus granularity. Evaluating after every sentence gives you fast feedback but noisier per-unit scores (a single sentence has less context than a full answer). Evaluating after every 2-3 sentences smooths out the noise at the cost of a slightly longer detection window. Most production setups land on a 2-sentence or 200-character window, whichever comes first.

Faithfulness is the metric you care about most during streaming, because it's the one that catches hallucination before the user reads the whole thing. Ragas's standard Faithfulness metric is built to run once, on a complete answer, against the full retrieved context set. You can still use the same underlying claim-decomposition logic — you just run it incrementally, on each sentence chunk, against the same fixed context.

from ragas.metrics import Faithfulness
from ragas.dataset_schema import SingleTurnSample
from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI

evaluator_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini"))
faithfulness_metric = Faithfulness(llm=evaluator_llm)

async def score_partial_faithfulness(question, retrieved_contexts, partial_answer):
    sample = SingleTurnSample(
        user_input=question,
        response=partial_answer,
        retrieved_contexts=retrieved_contexts,
    )
    return await faithfulness_metric.single_turn_ascore(sample)

The key design decision: keep retrieved_contexts fixed for the whole stream, since retrieval already happened before generation started. Only partial_answer grows. This means each incremental faithfulness check is comparing an increasingly complete claim set against a stable, known-good context — exactly the setup the metric was designed for, just run more often and on smaller inputs.

In practice, run this on the accumulated text (all sentences emitted so far), not just the newest sentence in isolation. A single sentence like "It was founded in 1998." is unfalsifiable without the surrounding sentence that names the company. Scoring the running total, not the delta, keeps the claim extractor working with enough context to actually parse claims correctly.

Putting the buffer and the incremental faithfulness scorer together, here's what an evaluation loop looks like wired into a token stream:

import asyncio

async def evaluate_stream(token_generator, question, retrieved_contexts, threshold=0.7):
    buffer = SentenceBuffer()
    alerts = []

    async for token in token_generator:
        new_sentences = buffer.push(token)
        if not new_sentences:
            continue

        running_answer = buffer.full_text_so_far()
        score = await score_partial_faithfulness(
            question, retrieved_contexts, running_answer
        )

        if score < threshold:
            alerts.append({
                "checkpoint_text": running_answer,
                "faithfulness": score,
            })
            # Signal upstream: consider truncating or flagging this response
            yield {"type": "faithfulness_warning", "score": score}

        yield {"type": "token", "content": token}

    yield {"type": "done", "alerts": alerts}

This isn't free — you're adding an LLM judge call every time a sentence boundary is crossed, which means for a 10-sentence answer you might make 5-10 extra evaluator calls instead of one. That's real added cost and real added latency between sentences. For high-stakes domains (medical, legal, financial advice bots), that tradeoff is usually worth it: catching a hallucination at sentence 3 and cutting the stream is far better than letting the user read all 10 sentences and correcting after the fact. For low-stakes chat, you'll want to sample — evaluate every third sentence, or only evaluate when the retrieved context has low similarity to the query in the first place, as a way to save cost on the cases least likely to hallucinate.

Context-Based Metrics: Your Cheap, Stable Signal

Here's the part that's easy to miss: two of Ragas's most useful metrics don't actually need the generated answer to be complete at all, because they operate primarily on the retrieved contexts and the query.

ContextPrecision measures whether the retrieved chunks are relevant to the question — this is knowable the instant retrieval finishes, before generation even starts. If context precision is low, you already know the generation is working with weak material, and you can flag the response as higher-risk before a single token streams out.

ContextRelevance-style checks (comparing retrieved chunks to the query directly) are similarly available pre-generation. Run these as a pre-flight check:

from ragas.metrics import LLMContextPrecisionWithoutReference

context_precision = LLMContextPrecisionWithoutReference(llm=evaluator_llm)

async def preflight_check(question, retrieved_contexts):
    sample = SingleTurnSample(
        user_input=question,
        retrieved_contexts=retrieved_contexts,
    )
    score = await context_precision.single_turn_ascore(sample)
    return score

If this pre-flight score comes back low, you have a decision to make before streaming even begins: re-retrieve with a rewritten query, fall back to a broader search, or set a lower faithfulness threshold for this particular response because you already know the grounding is thin. This is the cheapest evaluation signal in the whole pipeline because it costs one LLM call and happens before generation, adding zero perceived latency to the stream itself.

Handling the Final Reconciliation Pass

Incremental checks give you real-time signal, but they shouldn't be the only evaluation that counts. Once the stream finishes, run the full standard Ragas suite — Faithfulness, AnswerRelevancy, ContextRecall — on the complete response. This final pass is your ground truth for logging, dashboards, and regression tracking across model or prompt changes.

from ragas import evaluate
from ragas.metrics import Faithfulness, AnswerRelevancy, ContextRecall
from datasets import Dataset

def final_evaluation(question, contexts, full_answer, ground_truth=None):
    data = {
        "question": [question],
        "contexts": [contexts],
        "answer": [full_answer],
    }
    metrics = [Faithfulness(), AnswerRelevancy()]

    if ground_truth:
        data["ground_truth"] = [ground_truth]
        metrics.append(ContextRecall())

    dataset = Dataset.from_dict(data)
    return evaluate(dataset, metrics=metrics)

Treat the incremental scores as an early-warning system and the final pass as the record of truth. In practice, teams log both: the incremental scores get used for runtime decisions (truncate, regenerate, flag for review), and the final pass gets used for offline analytics (which prompt version has better faithfulness on average, which retriever config reduces hallucination rate). Conflating the two — trying to make the incremental average equal the final score — isn't necessary and isn't the point. They answer different questions.

Deciding What to Do When a Partial Score Fails

Detecting a low faithfulness score mid-stream is only useful if you have a defined action to take. There are three common patterns, roughly in order of how aggressive they are:

• Flag and continue: Let the stream finish, but attach a visible or logged warning. Lowest risk of breaking UX, useful for lower-stakes applications where a false positive (flagging a fine answer) is cheap.
• Regenerate from the checkpoint: Stop the stream at the sentence boundary that failed, discard it, and re-prompt the model with an instruction to stick closer to the retrieved context, then resume streaming from that point. This costs a regeneration but avoids showing the user a broken answer.
• Hard stop with fallback message: Kill the stream entirely and replace it with a "I don't have enough information to answer this confidently" message. Reserved for genuinely high-stakes domains where a partial hallucinated answer is worse than no answer.

Which pattern you pick should depend on your false-positive tolerance. LLM judges scoring short sentence fragments will have a higher variance than judges scoring complete answers, simply because there's less text to reason over. Run your threshold decision through a validation set of known-good and known-bad partial answers before shipping it — don't just eyeball a threshold of 0.7 and assume it transfers from your batch evaluation setup.

Cost and Latency Tradeoffs in Production

The honest tradeoff here is that streaming evaluation multiplies your LLM judge calls. If you evaluate every sentence, a 10-sentence answer that used to cost one Ragas evaluation call now costs up to ten. At scale, that's a meaningful line item, and it also adds real latency between sentences if your evaluator call is synchronous with the stream.

A few practical mitigations:

1. Use a cheaper, faster model as the judge for incremental checks, and reserve your stronger judge model for the final reconciliation pass. The incremental checks are directional signals, not final scores — they don't need frontier-model precision.
2. Batch sentence windows instead of scoring every single sentence. Evaluating every 2-3 sentences cuts judge calls by half or more while still catching drift early enough to act on it.
3. Run the evaluator asynchronously, off the critical rendering path. The user keeps seeing tokens stream normally; the evaluation happens in parallel and only interrupts the stream if a threshold is actually breached. This avoids adding judge-call latency to every sentence boundary.
4. Use context precision as a gate, not just a metric. If pre-flight context precision is already low, you can skip expensive incremental faithfulness checks altogether and just apply a conservative fallback strategy from the start.

None of this eliminates the added cost of evaluating streams — it's real, and teams should budget for it as an infrastructure cost, not treat it as free instrumentation. But the alternative — no evaluation until the response is fully rendered and already read by the user — undermines the reason you're evaluating faithfulness in a live product in the first place.

Wiring This Into an SSE or WebSocket Endpoint

Everything above works as a standalone loop, but production streaming almost always goes over Server-Sent Events or a WebSocket, not a plain async generator you control end to end. The evaluation layer needs to sit between your LLM provider's stream and whatever you send down the wire to the browser, without becoming a bottleneck for either side.

A common pattern is to run the token stream and the evaluator as two loosely coupled tasks that share the buffer, so a slow judge call never blocks a fast token from reaching the client:

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def sse_stream(question, retrieved_contexts, llm_token_stream):
    buffer = SentenceBuffer()
    pending_checks = asyncio.Queue()

    async def run_checks():
        while True:
            running_answer = await pending_checks.get()
            if running_answer is None:
                break
            score = await score_partial_faithfulness(
                question, retrieved_contexts, running_answer
            )
            if score < 0.7:
                print(f"[faithfulness-alert] score={score:.2f} text={running_answer[-120:]}")

    checker_task = asyncio.create_task(run_checks())

    async for token in llm_token_stream:
        yield f"data: {token}\n\n"
        new_sentences = buffer.push(token)
        if new_sentences:
            pending_checks.put_nowait(buffer.full_text_so_far())

    await pending_checks.put(None)
    await checker_task
    yield "data: [DONE]\n\n"

@app.get("/chat/stream")
async def chat_stream(question: str):
    retrieved_contexts = retrieve(question)  # your existing retriever call
    llm_tokens = generate_tokens(question, retrieved_contexts)
    return StreamingResponse(sse_stream(question, retrieved_contexts, llm_tokens), media_type="text/event-stream")

Notice the token is yielded to the client immediately, before the faithfulness check for that checkpoint even runs. The queue-and-worker pattern decouples "get tokens to the user fast" from "evaluate faithfulness thoroughly," which is exactly the separation you want. If you need the evaluator to be able to interrupt the stream (the hard-stop pattern from earlier), swap the print statement for a shared flag the token loop checks before each yield — just be aware that adds a synchronization point back into the hot path, so measure the added latency before shipping it to all traffic.

This same shape works over a WebSocket instead of SSE; the only change is that you're pushing JSON frames instead of data: lines, and you likely want a dedicated frame type for faithfulness warnings so the frontend can render them distinctly from normal content tokens (a small red underline on the flagged sentence, for instance, rather than a modal that interrupts reading).

Testing Your Streaming Evaluator Before Production

Before this goes anywhere near real traffic, build a small regression harness that replays known streams through your evaluator and checks that it fires (or doesn't fire) where expected. This is cheap to build and saves you from tuning thresholds against production incidents instead of a controlled test set.

import asyncio

async def fake_token_stream(text, chunk_size=3):
    for i in range(0, len(text), chunk_size):
        yield text[i:i + chunk_size]
        await asyncio.sleep(0)  # yield control, simulate streaming

test_cases = [
    {
        "question": "When was the company founded?",
        "contexts": ["The company was founded in 1998 in Palo Alto by two graduate students."],
        "answer": "The company was founded in 1998 in Palo Alto. It later expanded to Europe in 2003, "
                  "a fact not present in the provided context.",
        "expect_alert": True,
    },
    {
        "question": "When was the company founded?",
        "contexts": ["The company was founded in 1998 in Palo Alto by two graduate students."],
        "answer": "The company was founded in 1998 in Palo Alto by two graduate students.",
        "expect_alert": False,
    },
]

async def run_regression_suite():
    for case in test_cases:
        alerts = []
        async for event in evaluate_stream(
            fake_token_stream(case["answer"]), case["question"], case["contexts"]
        ):
            if event["type"] == "faithfulness_warning":
                alerts.append(event)
        fired = len(alerts) > 0
        status = "PASS" if fired == case["expect_alert"] else "FAIL"
        print(f"{status}: expected_alert={case['expect_alert']} got={fired}")

Keep this suite growing as you find real failure cases in production logs — every incident where the evaluator missed a hallucination, or flagged a perfectly fine answer, becomes a new test case. Over a few months this becomes your most valuable asset for tuning thresholds, because you're validating changes against a fixed, known set of behaviors instead of re-litigating the same threshold argument every time someone notices a false positive in the dashboard.

Common Pitfalls to Avoid

A few mistakes show up repeatedly when teams first wire up streaming evaluation:

• Scoring token deltas instead of running totals. A single new sentence in isolation often lacks the context to be judged fairly; always score the accumulated text.
• Reusing batch thresholds without re-validating them. A 0.7 faithfulness threshold tuned on complete answers does not necessarily hold for 2-sentence fragments, which tend to score noisier.
• Ignoring the pre-flight context signal. Teams often build elaborate incremental faithfulness pipelines while skipping the free, pre-generation context precision check that would have flagged half their problem cases before generation even started.
• Forgetting the final reconciliation pass. Incremental scores are for runtime decisions; without a final full-answer evaluation, you lose the ability to compare prompt versions or retriever changes over time in a stable, comparable way.
• Not logging checkpoint text alongside scores. When a faithfulness alert fires, you need the exact sentence-level text that triggered it to debug later — logging only the numeric score makes post-mortems nearly impossible.

Wrapping Up

Streaming changes what "evaluating a RAG response" means. You're no longer scoring a finished artifact — you're scoring a process while it's still unfolding, with an incomplete answer against a fixed, already-known context. The practical adaptation is straightforward once you see it clearly: chunk the stream into sentence-level units, run cheap context-based checks before generation even starts, run incremental faithfulness checks on the running total rather than the newest fragment, and reserve the full Ragas metric suite for a final reconciliation pass once the stream completes.

None of this requires abandoning Ragas or building a separate evaluation framework from scratch — it requires rethinking when you call the same metrics and what you feed them. Get that right, and you get real-time hallucination detection on top of a streaming UX, instead of having to choose between the two.

If you want to go deeper on wiring Ragas metrics into production RAG pipelines — including streaming setups, custom metric design, and CI-based regression testing for retrieval quality — our Ragas Tutorial course on teachyou.ai walks through all of it hands-on, from first metric to a full evaluation harness you can drop into your own stack.