RAGAS-Evaluierung für Open-WebUI-RAG-Pipelines: Drei Integrationsmuster

Open WebUI bietet keine native RAGAS-Integration. Der empfohlene Weg – auch laut Community-Diskussion – führt über das Pipeline- und Funktionssystem von Open WebUI als Brücke. Dieser Beitrag zeigt drei Muster, die sich in der Praxis bewährt haben:

1. Standalone-Evaluierungsskript – Batch-Evaluierung einer Open-WebUI-RAG-Pipeline gegen ein Testdatensatz
2. Open-WebUI-Filter-Funktion – Retrieval- und Generierungsdaten in Echtzeit erfassen
3. Langfuse-Integration – RAGAS-Scores als Traces für kontinuierliches Monitoring

---

Voraussetzungen

pip install ragas langchain-openai langchain-community httpx
# Für Ollama als Evaluator-LLM:
pip install langchain-ollama

DSGVO-Hinweis: Für eine vollständig lokale Evaluierung Ollama als Evaluator-LLM verwenden statt OpenAI. RAGAS unterstützt jeden LangChain-kompatiblen LLM.

---

Teil 1: Standalone-Evaluierungsskript

Dieses Skript fragt Ihre Open-WebUI-Instanz über deren OpenAI-kompatible API ab, erfasst den RAG-Kontext und führt die RAGAS-Metriken aus.

1.1 Open WebUI abfragen und RAG-Kontext erfassen

# eval_openwebui_rag.py
import httpx
import json
from dataclasses import dataclass

@dataclass
class RAGResult:
    question: str
    answer: str
    contexts: list[str]
    reference: str  # ground truth, if available

OPENWEBUI_BASE = "http://localhost:3000"  # your Open WebUI URL
API_KEY = "sk-..."  # Open WebUI API key (Admin Panel → Settings)
MODEL = "llama3.1:latest"  # your RAG-enabled model

def query_openwebui(question: str, collection_name: str = None) -> dict:
    """
    Query Open WebUI's chat completion endpoint.
    If you have documents uploaded as a knowledge base,
    reference them via the model's RAG configuration.
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": MODEL,
        "messages": [{"role": "user", "content": question}],
        "stream": False,
    }

    # If using Open WebUI's knowledge collections, the RAG context
    # is injected server-side based on model configuration.
    # You can also prefix with #collection_name in the message.
    if collection_name:
        payload["messages"][0]["content"] = f"#{collection_name} {question}"

    resp = httpx.post(
        f"{OPENWEBUI_BASE}/api/chat/completions",
        json=payload,
        headers=headers,
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()


def extract_rag_context(question: str, collection_name: str = None) -> RAGResult:
    """
    Open WebUI injects RAG context into the user message before
    sending to the LLM. To capture it, we use the /api/chat/completed
    endpoint or parse the augmented prompt.

    Alternative approach: Use the retrieval API directly.
    """
    # Direct retrieval query to get contexts
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }

    # Step 1: Get retrieved documents via RAG retrieval endpoint
    retrieval_resp = httpx.post(
        f"{OPENWEBUI_BASE}/api/v1/retrieval/query",
        json={
            "collection_name": collection_name or "",
            "query": question,
            "k": 5,  # top-k chunks
        },
        headers=headers,
        timeout=60,
    )

    contexts = []
    if retrieval_resp.status_code == 200:
        docs = retrieval_resp.json().get("documents", [[]])
        contexts = docs[0] if docs else []

    # Step 2: Get the generated answer
    chat_resp = query_openwebui(question, collection_name)
    answer = chat_resp["choices"][0]["message"]["content"]

    return RAGResult(
        question=question,
        answer=answer,
        contexts=contexts,
        reference="",  # fill from your test dataset
    )

1.2 RAGAS-Evaluierung ausführen

# run_ragas_eval.py
from ragas import evaluate
from ragas.dataset_schema import SingleTurnSample, EvaluationDataset
from ragas.metrics import (
    LLMContextRecall,
    Faithfulness,
    FactualCorrectness,
    ResponseRelevancy,
    ContextPrecision,
)
from ragas.llms import LangchainLLMWrapper
from langchain_ollama import ChatOllama
# Or for OpenAI: from langchain_openai import ChatOpenAI

from eval_openwebui_rag import extract_rag_context

# --- Konfiguration ---
# Lokales LLM als Evaluator (DSGVO-konform)
evaluator_llm = LangchainLLMWrapper(
    ChatOllama(model="llama3.1:70b", base_url="http://localhost:11434")
)
# Alternativ OpenAI:
# evaluator_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini"))

# --- Testdatensatz ---
test_cases = [
    {
        "question": "What is the maximum context length for RAG in Open WebUI?",
        "reference": "The default context length is 2048 tokens, but can be increased to 8192+ in Admin Panel > Models > Advanced Parameters.",
    },
    {
        "question": "How does hybrid search work in Open WebUI?",
        "reference": "Hybrid search combines BM25 keyword search with vector similarity search, using CrossEncoder for re-ranking results.",
    },
    # Hier eigene domänenspezifische Testfälle ergänzen...
]

# --- RAG-Ergebnisse erfassen ---
samples = []
for tc in test_cases:
    result = extract_rag_context(
        question=tc["question"],
        collection_name="your-knowledge-base",  # anpassen
    )

    samples.append(
        SingleTurnSample(
            user_input=tc["question"],
            response=result.answer,
            retrieved_contexts=result.contexts,
            reference=tc["reference"],
        )
    )

evaluation_dataset = EvaluationDataset(samples=samples)

# --- Evaluierung ---
metrics = [
    Faithfulness(),        # Ist die Antwort im Kontext gedeckt?
    ResponseRelevancy(),   # Beantwortet die Antwort die Frage?
    LLMContextRecall(),    # Hat das Retrieval die richtigen Dokumente gefunden?
    ContextPrecision(),    # Sind die Chunks relevant (kein Rauschen)?
    FactualCorrectness(),  # Stimmt die Antwort mit dem Ground Truth überein?
]

result = evaluate(
    dataset=evaluation_dataset,
    metrics=metrics,
    llm=evaluator_llm,
)

print(result)
# {'faithfulness': 0.85, 'answer_relevancy': 0.91, ...}

# Export zur weiteren Analyse
df = result.to_pandas()
df.to_csv("ragas_eval_results.csv", index=False)
print(df[["user_input", "faithfulness", "response_relevancy"]].to_string())

---

Teil 2: Open-WebUI-Filter-Funktion (Echtzeit-Logging)

Diese Filter-Funktion erfasst jede RAG-Interaktion in Echtzeit und speichert sie für spätere RAGAS-Evaluierung.

2.1 Als Open-WebUI-Funktion installieren

Unter Admin Panel → Functions → Add Function einfügen:

"""
title: RAG Evaluation Logger
description: Captures RAG context and responses for RAGAS evaluation.
             Logs to local JSON and optionally to Langfuse.
author: stagl.systems
version: 0.1.0
"""

import json
import time
import os
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, Field
from typing import Optional


class Filter:
    class Valves(BaseModel):
        log_dir: str = Field(
            default="/app/backend/data/ragas_logs",
            description="Directory to store RAG evaluation logs",
        )
        langfuse_enabled: bool = Field(
            default=False,
            description="Send traces to Langfuse",
        )
        langfuse_host: str = Field(
            default="http://langfuse:3000",
            description="Langfuse server URL",
        )
        langfuse_public_key: str = Field(
            default="",
            description="Langfuse public key",
        )
        langfuse_secret_key: str = Field(
            default="",
            description="Langfuse secret key",
        )

    def __init__(self):
        self.valves = self.Valves()
        self._current_contexts = {}

    async def inlet(self, body: dict, __user__: dict = None) -> dict:
        """
        Capture the user's original question before RAG augmentation.
        """
        messages = body.get("messages", [])
        if messages:
            last_msg = messages[-1].get("content", "")
            session_key = f"{__user__.get('id', 'anon')}_{int(time.time())}"
            body["__ragas_session_key"] = session_key
            body["__ragas_original_question"] = last_msg

            if "[context]" in last_msg or "retrieved" in last_msg.lower():
                self._current_contexts[session_key] = {
                    "augmented_prompt": last_msg,
                    "timestamp": datetime.utcnow().isoformat(),
                }

        return body

    async def outlet(self, body: dict, __user__: dict = None) -> dict:
        """
        After LLM responds, log the full RAG interaction
        for offline RAGAS evaluation.
        """
        messages = body.get("messages", [])
        if len(messages) < 2:
            return body

        session_key = body.get("__ragas_session_key", "")
        original_question = body.get("__ragas_original_question", "")
        response = messages[-1].get("content", "")

        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": __user__.get("id", "anonymous") if __user__ else "anonymous",
            "user_input": original_question,
            "response": response,
            "model": body.get("model", "unknown"),
            "augmented_context": self._current_contexts.get(session_key, {}),
        }

        # In Datei schreiben
        try:
            log_dir = Path(self.valves.log_dir)
            log_dir.mkdir(parents=True, exist_ok=True)
            log_file = log_dir / f"rag_eval_{datetime.utcnow().strftime('%Y%m%d')}.jsonl"
            with open(log_file, "a") as f:
                f.write(json.dumps(record) + "\n")
        except Exception as e:
            print(f"[RAG Eval Logger] File write error: {e}")

        # Optional: an Langfuse senden
        if self.valves.langfuse_enabled and self.valves.langfuse_secret_key:
            try:
                await self._log_to_langfuse(record)
            except Exception as e:
                print(f"[RAG Eval Logger] Langfuse error: {e}")

        self._current_contexts.pop(session_key, None)

        return body

    async def _log_to_langfuse(self, record: dict):
        """Trace an Langfuse senden."""
        try:
            from langfuse import Langfuse

            langfuse = Langfuse(
                public_key=self.valves.langfuse_public_key,
                secret_key=self.valves.langfuse_secret_key,
                host=self.valves.langfuse_host,
            )

            trace = langfuse.trace(
                name="rag-evaluation",
                input=record["user_input"],
                output=record["response"],
                metadata={
                    "model": record["model"],
                    "source": "openwebui-ragas-filter",
                },
            )

            langfuse.flush()
        except ImportError:
            print("[RAG Eval Logger] langfuse package not installed")

2.2 Geloggte Daten mit RAGAS batch-evaluieren

# evaluate_logs.py
import json
from pathlib import Path
from ragas import evaluate
from ragas.dataset_schema import SingleTurnSample, EvaluationDataset
from ragas.metrics import Faithfulness, ResponseRelevancy
from ragas.llms import LangchainLLMWrapper
from langchain_ollama import ChatOllama

LOG_DIR = Path("/app/backend/data/ragas_logs")
evaluator_llm = LangchainLLMWrapper(
    ChatOllama(model="llama3.1:70b", base_url="http://localhost:11434")
)

# Geloggte Interaktionen laden
samples = []
for log_file in sorted(LOG_DIR.glob("rag_eval_*.jsonl")):
    with open(log_file) as f:
        for line in f:
            record = json.loads(line)
            ctx = record.get("augmented_context", {})
            retrieved = ctx.get("augmented_prompt", "")

            samples.append(
                SingleTurnSample(
                    user_input=record["user_input"],
                    response=record["response"],
                    retrieved_contexts=[retrieved] if retrieved else [],
                    # Kein Ground Truth in Produktionslogs
                    # -> nur referenzfreie Metriken verwenden
                )
            )

if not samples:
    print("No log data found.")
    exit()

dataset = EvaluationDataset(samples=samples)

# Referenzfreie Metriken (kein Ground Truth nötig)
result = evaluate(
    dataset=dataset,
    metrics=[Faithfulness(), ResponseRelevancy()],
    llm=evaluator_llm,
)

print(result)
df = result.to_pandas()
df.to_csv("production_ragas_eval.csv", index=False)

# Niedrig bewertete Antworten markieren
low_faith = df[df["faithfulness"] < 0.7]
if not low_faith.empty:
    print(f"\n{len(low_faith)} Antworten mit niedriger Faithfulness:")
    print(low_faith[["user_input", "faithfulness"]].to_string())

---

Teil 3: Langfuse + RAGAS Dashboard

Da Langfuse ohnehin im Stack läuft, können RAGAS-Scores direkt als Langfuse-Scores auf Traces geschrieben werden:

# langfuse_ragas_scores.py
from langfuse import Langfuse
from ragas import evaluate
from ragas.dataset_schema import SingleTurnSample, EvaluationDataset
from ragas.metrics import Faithfulness, ResponseRelevancy
from ragas.llms import LangchainLLMWrapper
from langchain_ollama import ChatOllama

langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="http://your-langfuse:3000",  # self-hosted
)

evaluator_llm = LangchainLLMWrapper(
    ChatOllama(model="llama3.1:70b", base_url="http://localhost:11434")
)

# Aktuelle Traces aus Langfuse abrufen
traces = langfuse.fetch_traces(
    name="rag-evaluation",
    limit=50,
    order_by="timestamp",
)

samples = []
trace_ids = []
for trace in traces.data:
    samples.append(
        SingleTurnSample(
            user_input=trace.input or "",
            response=trace.output or "",
            retrieved_contexts=[],  # erweitern, wenn Kontext geloggt wird
        )
    )
    trace_ids.append(trace.id)

if samples:
    dataset = EvaluationDataset(samples=samples)
    result = evaluate(
        dataset=dataset,
        metrics=[Faithfulness(), ResponseRelevancy()],
        llm=evaluator_llm,
    )

    df = result.to_pandas()

    # Scores zurück in Langfuse-Traces schreiben
    for i, trace_id in enumerate(trace_ids):
        for metric in ["faithfulness", "response_relevancy"]:
            if metric in df.columns:
                langfuse.score(
                    trace_id=trace_id,
                    name=metric,
                    value=float(df.iloc[i][metric]),
                )

    langfuse.flush()
    print(f"RAGAS-Scores für {len(trace_ids)} Traces in Langfuse geschrieben.")

---

Teil 4: Synthetische Testdaten generieren

Wer noch keinen beschrifteten Testdatensatz hat, kann ihn direkt aus den eigenen Dokumenten erzeugen:

# generate_test_data.py
from ragas.testset import TestsetGenerator
from ragas.llms import LangchainLLMWrapper
from langchain_ollama import ChatOllama
from langchain_community.document_loaders import DirectoryLoader

# Wissensbasis-Dokumente laden
loader = DirectoryLoader(
    "/path/to/your/documents",
    glob="**/*.md",  # Dateitypen anpassen
)
documents = loader.load()

generator_llm = LangchainLLMWrapper(
    ChatOllama(model="llama3.1:70b", base_url="http://localhost:11434")
)

generator = TestsetGenerator(llm=generator_llm)

# Synthetische QA-Paare generieren
testset = generator.generate_with_langchain_docs(
    documents=documents,
    testset_size=50,  # Anzahl Testfälle
)

df = testset.to_pandas()
df.to_csv("synthetic_testset.csv", index=False)
print(f"{len(df)} Testfälle generiert")
print(df[["user_input", "reference"]].head())

---

Empfohlener Workflow

Schnellstart-Checkliste:

1. Filter-Funktion in Open WebUI deployen (Teil 2)
2. Langfuse-Verbindung in den Valve-Einstellungen konfigurieren
3. Synthetischen Testdatensatz aus eigenen Dokumenten generieren (Teil 4)
4. Batch-Evaluierungsskript als Cron-Job einrichten (Teil 1/2)
5. Scores in Langfuse für Dashboarding pushen (Teil 3)
6. Iterieren: Chunking, Top-K und Re-Ranking anhand der RAGAS-Scores tunen

Schlüsselmetriken im Überblick

Metrik	Was sie misst	Ground Truth nötig?
Faithfulness	Ist die Antwort im abgerufenen Kontext gedeckt?	Nein
Response Relevancy	Beantwortet die Antwort die Frage?	Nein
Context Precision	Sind die abgerufenen Chunks relevant?	Ja
Context Recall	Hat das Retrieval alle nötigen Infos gefunden?	Ja
Factual Correctness	Stimmt die Antwort mit dem Ground Truth überein?	Ja

Empfehlung: Mit Faithfulness und Response Relevancy starten (kein Ground Truth nötig), dann in einen beschrifteten Testdatensatz investieren für die vollständige Metrik-Suite.

Fazit

RAGAS und Open WebUI lassen sich über drei Integrationsmuster verbinden, die aufeinander aufbauen: Das Standalone-Skript für den ersten Qualitäts-Check, die Filter-Funktion für kontinuierliches Produktions-Logging, und die Langfuse-Integration für langfristiges Dashboard-Monitoring. Der gesamte Stack – Ollama als Evaluator-LLM, JSONL-Logging auf lokaler Disk, Langfuse self-hosted – läuft ohne externe Abhängigkeiten und bleibt damit vollständig DSGVO-konform.

---

Martin Stagl ist Systems Engineer und Data Scientist in Wien. Er schreibt auf stagl.systems über Enterprise-KI-Infrastruktur und DSGVO-konforme LLM-Deployments im DACH-Raum.