Startseite / Artikel / Praktische Hinweise: Mein RAG-System hat 80 % meiner Daten übersehen. Diese eine Änderung hat das Problem gelöst.

Praktische Hinweise: Mein RAG-System hat 80 % meiner Daten übersehen. Diese eine Änderung hat das Problem gelöst.

Schritt-für-Schritt-Anleitung zu den Praktischen Hinweisen: Mein RAG-System hat 80 % meiner Daten übersehen. Diese eine Änderung hat das Problem gelöst.: Verträge, Überprüfungen sowie Code-Blöcke für Teams, die dieses Muster einsetzen.

5233 Wörter

Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Mein RAG-System hat 80 % meiner Daten übersehen. Diese eine Änderung hat das Problem gelöst. Der Schwerpunkt liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den man ohne Rätseln über die Absicht direkt in ein Repository einfügen kann. In der Übersichtsphase sollten Eingaben, Verantwortliche für die Schritte sowie Abbruchkriterien definiert werden, bevor Code geändert wird. Die Betreiber sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Halten Sie Konfigurationen außerhalb des Anwendungscode. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den die Betreiber überprüfen können, ohne den gesamten Ablauf durchlesen zu müssen.

Was hat sich tatsächlich mit Gemini Embedding 2 geändert?

Beim Bearbeiten der Phase „Was hat sich tatsächlich geändert?“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Protokollieren Sie bei jedem Aufruf die Anfrage-ID, die Modell-ID sowie die Latenzzeit. Ohne diese Aufzeichnungen wirken intermittierende Fehler des Anbieters wie Programmfehler.

Das Problem des einzigen Embedding-Raums

Beim Arbeiten an der Phase „The Single Embedding Space“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten.

Was das Modell tatsächlich unterstützt

Während der Phase „Was macht das Modell tatsächlich?“ sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Cachen Sie stabile Systemanweisungen sowie Tool-Schemata. Das erneute Senden identischer Präambeln ist eine häufige Ursache für Ressourcenverschwendung. Während der Phase „Was macht das Modell tatsächlich?“ sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Legen Sie die Konfiguration außerhalb des Anwendungscode ab. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

Die Architektur – ausführlich erklärt

Die in Phasen erklärte Architektur funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie einen erfolgreichen Fall, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf und den Wiederherstellungsprozess. Versuche, menschliche Kontrollen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zum Abrufen. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

Der Eingabepipeline

Die Phase des Ingestion-Pipelines funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Pipeline-System. Trennen Sie die Chunking-Politik von der Abrufpolitik – Änderungen an einer sollten nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

Das Query-Pipeline

Die Phase des Query-Pipelines funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Veränderung in einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern. Die Phase des Query-Pipelines funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Legen Sie die Konfiguration außerhalb des Anwendungscode ab. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreuer ohne das Durchlesen des gesamten Systems überprüfen können.

Warum diese beiden Pipelines getrennt bleiben müssen

Für die Phase „Warum diese beiden Pipelines?“ sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

Umgebung einrichten

In der Phase des Umgebungsaufbaus sollten die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Es sind kleinere, testbare Einheiten vorzuziehen statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Pipeline-System. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
# config.py
import os
from google import genai
from google.genai import types

GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
EMBEDDING_MODEL = "gemini-embedding-2-preview"
GENERATION_MODEL = "gemini-2.5-pro"
# Output dimensionality options: 128, 256, 512, 768, 1024, 1536, 3072
# 1536 is the recommended default
EMBEDDING_DIMENSIONS = 1536
client = genai.Client(api_key=GEMINI_API_KEY)

Aufbau der Eingabepipeline

Bei der Phase „Erstellung des Ingestion-Pipelines“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken im Indexing unterscheiden. Bei der Phase „Erstellung des Ingestion-Pipelines“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den die Operator ohne Lesen prüfen können.

den gesamten Graphen.

Schritt 1: Der Embedding-Client

Beim Arbeiten am Schritt 1, der Embedding-Phase, sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Messen Sie die Trefferquote anhand einer festgelegten Fragestellung, bevor Sie die Anfragen anpassen. Eine Änderung der Anfragen behebt selten ein schwaches Suchverhalten.

# embedder.py
import time
from pathlib import Path
from google import genai
from google.genai import types
from config import client, EMBEDDING_MODEL, EMBEDDING_DIMENSIONS

def embed_text(text: str, task_type: str = "RETRIEVAL_DOCUMENT") -> list[float]:
    """Embed a plain text chunk."""
    result = client.models.embed_content(
        model=EMBEDDING_MODEL,
        contents=text,
        config=types.EmbedContentConfig(
            task_type=task_type,
            output_dimensionality=EMBEDDING_DIMENSIONS
        )
    )
    return result.embeddings[0].values

def _wait_for_file(uploaded, max_wait: int = 300):
    """Poll until a File API upload is done processing."""
    waited = 0
    poll_interval = 5
    while uploaded.state.name == "PROCESSING" and waited         time.sleep(poll_interval)
        waited += poll_interval
        uploaded = client.files.get(name=uploaded.name)
    if uploaded.state.name != "ACTIVE":
        raise RuntimeError(
            f"File never became ACTIVE. Final state: {uploaded.state.name}"
        )
    return uploaded

def embed_audio(audio_path: str) -> list[float]:
    """
    Embed an audio file natively. No transcription step.
    The model processes the audio signal directly and returns a
    semantic embedding that captures speech content, tone, and
    acoustic features. Max input: 80 seconds per file.
    """
    uploaded = client.files.upload(path=str(audio_path))
    uploaded = _wait_for_file(uploaded, max_wait=120)
    result = client.models.embed_content(
        model=EMBEDDING_MODEL,
        contents=uploaded,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_DOCUMENT",
            output_dimensionality=EMBEDDING_DIMENSIONS
        )
    )
    # Clean up: uploaded files count against your quota
    client.files.delete(name=uploaded.name)
    return result.embeddings[0].values

def embed_video(video_path: str) -> list[float]:
    """
    Embed a video chunk natively. Gemini processes both the
    audio track and visual frames together in one pass.
    This is the key capability: visual demonstrations get captured
    in the embedding alongside what is being said. Max input: 128 seconds.
    """
    uploaded = client.files.upload(path=str(video_path))
    uploaded = _wait_for_file(uploaded, max_wait=300)
    result = client.models.embed_content(
        model=EMBEDDING_MODEL,
        contents=uploaded,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_DOCUMENT",
            output_dimensionality=EMBEDDING_DIMENSIONS
        )
    )
    client.files.delete(name=uploaded.name)
    return result.embeddings[0].values

def embed_with_context(text: str, image_bytes: bytes = None) -> list[float]:
    """
    Embed text and an optional image together in a single call.
    When both are passed, the model returns one vector that
    represents the joint meaning. A query asking about a database
    schema can retrieve a screenshot of that schema.
    """
    contents = [text]
    if image_bytes:
        contents.append(
            types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg")
        )
    result = client.models.embed_content(
        model=EMBEDDING_MODEL,
        contents=contents,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_DOCUMENT",
            output_dimensionality=EMBEDDING_DIMENSIONS
        )
    )
    return result.embeddings[0].values

Schritt 2: Der Media Chunker

Beim Bearbeiten des Schritts 2 „Die Medien“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Messen Sie die Erinnerungskraft anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Abrufverhalten.

# chunker.py
import subprocess
import json
from pathlib import Path
from dataclasses import dataclass
from typing import List

@dataclass
class MediaChunk:
    file_path: str
    start_time: float
    end_time: float
    source_file: str
    modality: str
    chunk_index: int
    total_chunks: int  # Useful for progress reporting

def get_media_duration(file_path: str) -> float:
    """Get exact duration via ffprobe. Works for both audio and video."""
    cmd = [
        "ffprobe", "-v", "quiet",
        "-print_format", "json",
        "-show_streams", str(file_path)
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    data = json.loads(result.stdout)
    # Find the first stream with a duration value
    for stream in data.get("streams", []):
        if "duration" in stream:
            return float(stream["duration"])
    raise ValueError(f"Could not determine duration for: {file_path}")

def _run_ffmpeg_split(input_path: str, output_path: str,
                      start: float, duration: float):
    """Execute a single ffmpeg split operation."""
    cmd = [
        "ffmpeg", "-y",
        "-ss", str(start),
        "-i", str(input_path),
        "-t", str(duration),
        "-c", "copy",           # No re-encoding: much faster, no quality loss
        "-avoid_negative_ts", "make_zero",
        str(output_path)
    ]
    result = subprocess.run(cmd, capture_output=True)
    if result.returncode != 0:
        raise RuntimeError(
            f"ffmpeg failed: {result.stderr.decode()}"
        )

def chunk_video(
    video_path: str,
    chunk_duration: int = 90,
    overlap: int = 10,
    output_dir: str = "./chunks/video"
) -> List[MediaChunk]:
    """
    Split video into overlapping chunks within the 128-second limit.
    Default: 90-second chunks with 10-second overlap.
    Overlap ensures topic transitions are captured in at least one chunk.
    """
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    total_duration = get_media_duration(video_path)
    source_name = Path(video_path).stem
    # Pre-calculate chunk boundaries
    boundaries = []
    start = 0.0
    while start         end = min(start + chunk_duration, total_duration)
        boundaries.append((start, end))
        start += (chunk_duration - overlap)
    chunks = []
    for idx, (start, end) in enumerate(boundaries):
        output_path = f"{output_dir}/{source_name}_{idx:04d}.mp4"
        _run_ffmpeg_split(video_path, output_path, start, end - start)
        chunks.append(MediaChunk(
            file_path=output_path,
            start_time=start,
            end_time=end,
            source_file=str(video_path),
            modality="video",
            chunk_index=idx,
            total_chunks=len(boundaries)
        ))
    return chunks

def chunk_audio(
    audio_path: str,
    chunk_duration: int = 60,
    overlap: int = 5,
    output_dir: str = "./chunks/audio"
) -> List[MediaChunk]:
    """
    Split audio into overlapping chunks within the 80-second limit.
    60 seconds per chunk gives a comfortable buffer under the 80-second cap.
    """
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    total_duration = get_media_duration(audio_path)
    source_name = Path(audio_path).stem
    boundaries = []
    start = 0.0
    while start         end = min(start + chunk_duration, total_duration)
        boundaries.append((start, end))
        start += (chunk_duration - overlap)
    chunks = []
    for idx, (start, end) in enumerate(boundaries):
        output_path = f"{output_dir}/{source_name}_{idx:04d}.mp3"
        _run_ffmpeg_split(audio_path, output_path, start, end - start)
        chunks.append(MediaChunk(
            file_path=output_path,
            start_time=start,
            end_time=end,
            source_file=str(audio_path),
            modality="audio",
            chunk_index=idx,
            total_chunks=len(boundaries)
        ))
    return chunks

Schritt 3: Der Vektor-Speicher

Beim Arbeiten an Schritt 3 „The Vector“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Prozesse ab. Messen Sie die Trefferquote anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchsystem. Beim Arbeiten an Schritt 3 „The Vector“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Legen Sie die Konfiguration außerhalb des Anwendungscode ab. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

# vector_store.py
import chromadb
from chromadb.config import Settings
from pathlib import Path

class MultimodalVectorStore:
    """
    Vector store wrapping ChromaDB for multimodal RAG.
    Stores embeddings + metadata for text, audio, and video chunks.
    """
    def __init__(self, persist_dir: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(
            path=persist_dir,
            settings=Settings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name="multimodal_rag",
            # cosine distance is standard for semantic similarity
            metadata={"hnsw:space": "cosine"}
        )
    def add_text_chunk(
        self,
        chunk_id: str,
        text: str,
        embedding: list[float],
        source_file: str,
        chunk_index: int,
        page: int = None
    ):
        self.collection.add(
            ids=[chunk_id],
            embeddings=[embedding],
            documents=[text],
            metadatas=[{
                "modality": "text",
                "source_file": source_file,
                "chunk_index": chunk_index,
                "page": page or 0,
                "preview": text[:250]
            }]
        )
    def add_media_chunk(
        self,
        chunk_id: str,
        embedding: list[float],
        source_file: str,
        start_time: float,
        end_time: float,
        modality: str,
        chunk_index: int
    ):
        """
        Store a video or audio chunk.
        Note: we store a formatted timestamp string in `documents`
        so ChromaDB has something to display. The actual retrieval
        quality comes entirely from the embedding, not this text.
        """
        ts_start = f"{int(start_time // 60):02d}:{int(start_time % 60):02d}"
        ts_end = f"{int(end_time // 60):02d}:{int(end_time % 60):02d}"
        display = (
            f"[{modality.upper()}] {Path(source_file).name} "
            f"from {ts_start} to {ts_end}"
        )
        self.collection.add(
            ids=[chunk_id],
            embeddings=[embedding],
            documents=[display],
            metadatas=[{
                "modality": modality,
                "source_file": source_file,
                "start_time": start_time,
                "end_time": end_time,
                "timestamp_start": ts_start,
                "timestamp_end": ts_end,
                "chunk_index": chunk_index,
                "preview": display
            }]
        )
    def search(
        self,
        query_embedding: list[float],
        n_results: int = 5,
        modality_filter: str = None
    ) -> list[dict]:
        """
        Retrieve top-k most similar chunks across all modalities.
        Optionally filter to a single modality for targeted search.
        """
        where_clause = {"modality": modality_filter} if modality_filter else None
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            where=where_clause,
            include=["documents", "metadatas", "distances"]
        )
        chunks = []
        for doc, meta, dist in zip(
            results["documents"][0],
            results["metadatas"][0],
            results["distances"][0]
        ):
            chunks.append({
                "content": doc,
                "metadata": meta,
                "modality": meta["modality"],
                # ChromaDB returns cosine distance; convert to similarity score
                "similarity": round(1.0 - dist, 4)
            })
        return sorted(chunks, key=lambda x: x["similarity"], reverse=True)
    def count(self) -> int:
        return self.collection.count()

Schritt 4: Der Ingestion Runner

Die Phase 4, die Eingabestufe, funktioniert am besten, wenn sie als messbare Größe betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein erfolgreiches Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf und den Wiederherstellungsprozess. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zum Abrufen. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

# ingest.py
import os
import hashlib
from pathlib import Path
from chunker import chunk_video, chunk_audio
from embedder import embed_text, embed_audio, embed_video
from vector_store import MultimodalVectorStore

store = MultimodalVectorStore(persist_dir="./chroma_db")

def make_chunk_id(source_path: str, chunk_index: int) -> str:
    """Stable, unique ID for any chunk. Same input always = same ID."""
    raw = f"{os.path.abspath(source_path)}:{chunk_index}"
    return hashlib.sha256(raw.encode()).hexdigest()[:20]

def ingest_text_file(file_path: str):
    with open(file_path, "r", encoding="utf-8") as f:
        text = f.read()
    # Sliding window chunking: 800 chars with 100-char overlap
    chunk_size, overlap = 800, 100
    raw_chunks = []
    start = 0
    while start         end = min(start + chunk_size, len(text))
        raw_chunks.append(text[start:end])
        start += chunk_size - overlap
    for i, chunk_text in enumerate(raw_chunks):
        embedding = embed_text(chunk_text, task_type="RETRIEVAL_DOCUMENT")
        store.add_text_chunk(
            chunk_id=make_chunk_id(file_path, i),
            text=chunk_text,
            embedding=embedding,
            source_file=file_path,
            chunk_index=i
        )
    print(f"    Stored {len(raw_chunks)} text chunks from {Path(file_path).name}")

def ingest_video_file(file_path: str):
    print(f"    Chunking: {Path(file_path).name}")
    chunks = chunk_video(file_path, chunk_duration=90, overlap=10)
    for chunk in chunks:
        print(
            f"    Embedding chunk {chunk.chunk_index + 1}/{chunk.total_chunks} "
            f"({chunk.start_time:.0f}s to {chunk.end_time:.0f}s)"
        )
        try:
            embedding = embed_video(chunk.file_path)
            store.add_media_chunk(
                chunk_id=make_chunk_id(file_path, chunk.chunk_index),
                embedding=embedding,
                source_file=file_path,
                start_time=chunk.start_time,
                end_time=chunk.end_time,
                modality="video",
                chunk_index=chunk.chunk_index
            )
        except Exception as e:
            print(f"    WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
        finally:
            # Always clean up temp files, even on failure
            if os.path.exists(chunk.file_path):
                os.remove(chunk.file_path)
    print(f"    Done. {len(chunks)} video chunks stored.")

def ingest_audio_file(file_path: str):
    print(f"    Chunking: {Path(file_path).name}")
    chunks = chunk_audio(file_path, chunk_duration=60, overlap=5)
    for chunk in chunks:
        try:
            embedding = embed_audio(chunk.file_path)
            store.add_media_chunk(
                chunk_id=make_chunk_id(file_path, chunk.chunk_index),
                embedding=embedding,
                source_file=file_path,
                start_time=chunk.start_time,
                end_time=chunk.end_time,
                modality="audio",
                chunk_index=chunk.chunk_index
            )
        except Exception as e:
            print(f"    WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
        finally:
            if os.path.exists(chunk.file_path):
                os.remove(chunk.file_path)
    print(f"    Done. {len(chunks)} audio chunks stored.")

def ingest_directory(directory: str):
    handlers = {
        ".txt": ingest_text_file,
        ".md": ingest_text_file,
        ".mp4": ingest_video_file,
        ".mov": ingest_video_file,
        ".mp3": ingest_audio_file,
        ".wav": ingest_audio_file,
    }
    all_files = list(Path(directory).rglob("*"))
    media_files = [f for f in all_files if f.suffix.lower() in handlers]
    print(f"Found {len(media_files)} files to ingest\n")
    for file_path in media_files:
        print(f"Processing: {file_path.name}")
        handler = handlers[file_path.suffix.lower()]
        handler(str(file_path))
        print()
    print(f"Ingestion complete. Total chunks indexed: {store.count()}")

if __name__ == "__main__":
    ingest_directory("./knowledge_base")

Aufbau des Abfragen-Pipelines

Die Phase „Building the Query Pipeline“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein optimales Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Gesamtsystem. Trennen Sie die Strategie zur Aufteilung in Teile von der Strategie zur Abrufung. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

# query.py
import os
from pathlib import Path
import google.generativeai as genai
from embedder import embed_text
from vector_store import MultimodalVectorStore
from config import GENERATION_MODEL

store = MultimodalVectorStore(persist_dir="./chroma_db")

def format_context_for_llm(chunks: list[dict]) -> str:
    """
    Format retrieved chunks into a context block for the generative model.
    We include modality, source, and similarity score so the model
    can calibrate its confidence and cite sources accurately.
    """
    parts = []
    for rank, chunk in enumerate(chunks, start=1):
        meta = chunk["metadata"]
        modality = chunk["modality"]
        score = chunk["similarity"]
        if modality == "text":
            parts.append(
                f"[SOURCE {rank} | TEXT | {Path(meta['source_file']).name} "
                f"| chunk {meta['chunk_index']} | similarity {score}]\n"
                f"{chunk['content']}"
            )
        elif modality == "video":
            parts.append(
                f"[SOURCE {rank} | VIDEO | {Path(meta['source_file']).name} "
                f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
                f"| similarity {score}]\n"
                f"Video segment covering this time range."
            )
        elif modality == "audio":
            parts.append(
                f"[SOURCE {rank} | AUDIO | {Path(meta['source_file']).name} "
                f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
                f"| similarity {score}]\n"
                f"Audio segment covering this time range."
            )
    return "\n\n---\n\n".join(parts)

def answer_query(
    query: str,
    n_results: int = 5,
    modality_filter: str = None,
    similarity_threshold: float = 0.6
) -> dict:
    """
    Full RAG pipeline: embed the query, retrieve chunks, generate answer.
    similarity_threshold: chunks below this score are dropped before generation.
    Prevents low-quality matches from polluting the context.
    """
    query_embedding = embed_text(query, task_type="RETRIEVAL_QUERY")
    retrieved = store.search(
        query_embedding=query_embedding,
        n_results=n_results,
        modality_filter=modality_filter
    )
    # Filter out weak matches
    filtered = [c for c in retrieved if c["similarity"] >= similarity_threshold]
    if not filtered:
        return {
            "answer": (
                "No sufficiently relevant content was found in the knowledge base. "
                "The most similar content had a similarity score below the threshold."
            ),
            "sources": retrieved,
            "query": query
        }
    context = format_context_for_llm(filtered)
    system_prompt = """You are a helpful assistant with access to a multimodal
knowledge base that contains text documents, video recordings, and audio files.
When citing a source, reference it by its label (e.g., SOURCE 1, SOURCE 2).
For video and audio sources, always include the timestamp so the user can
navigate to the exact moment in the recording.
If the retrieved context does not contain enough information to answer
confidently, say so clearly rather than guessing."""
    user_message = (
        f"Using only the sources below, answer this question:\n\n"
        f"Question: {query}\n\n"
        f"Sources:\n{context}"
    )
    genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
    model = genai.GenerativeModel(GENERATION_MODEL)
    response = model.generate_content(
        user_message,
        generation_config={"temperature": 0.1}
    )
    return {
        "answer": response.text,
        "sources": filtered,
        "query": query,
        "chunks_retrieved": len(retrieved),
        "chunks_used": len(filtered)
    }

Verbesserung der Abrufgenauigkeit

Die Phase „Verbesserung der Abrufgenauigkeit“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Veränderung in einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern. Die Phase „Verbesserung der Abrufgenauigkeit“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

1. Verwenden Sie jedes Mal die richtige Aufgabentyp

Für den „1 Use the Right“-Schritt sollten vor der Codeänderung die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern – ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

2. Fügen Sie vor dem Einbetten eine Abfragerweiterung hinzu

Zur Phase „2 Add Query Expansion“ sollten die Eingaben, der Verantwortliche für diesen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Es ist besser, kleine, testbare Einheiten statt umfangreicher Skripte zu verwenden. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich hinweisen und nicht auf ein verworrenes Ablaufschema. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken im Indexing unterscheiden.

# query_expander.py
import google.generativeai as genai
import os

genai.configure(api_key=os.getenv("GEMINI_API_KEY"))

def expand_query(raw_query: str) -> str:
    """
    Rewrite a short user query into a more detailed retrieval query.
    Returns the expanded version. Falls back to original on failure.
    """
    model = genai.GenerativeModel("gemini-2.0-flash")
    prompt = (
        "Rewrite the following search query to be more detailed and specific. "
        "Add relevant context, related terminology, and clarify the intent. "
        "Keep it as a single question. Do not add facts not implied by the original.\n\n"
        f"Original query: {raw_query}\n\n"
        "Expanded query:"
    )
    try:
        response = model.generate_content(
            prompt,
            generation_config={"temperature": 0.2, "max_output_tokens": 200}
        )
        return response.text.strip()
    except Exception:
        return raw_query  # Graceful fallback

# Usage in query pipeline:
# expanded = expand_query("API limits engineering review")
# query_embedding = embed_text(expanded, task_type="RETRIEVAL_QUERY")

3. Mehrere Abfragen parallel ausführen

In der Phase „3 Run Multiple Queries“ sollten die Eingaben, der Verantwortliche für diesen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Zitieren Sie die Passagen, die tatsächlich die Antwort begründen. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden. In der Phase „3 Run Multiple Queries“ sollten die Eingaben, der Verantwortliche für diesen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den die Operator ohne das Lesen des gesamten Graphen überprüfen können.

# multi_query.py
import concurrent.futures
from query_expander import expand_query
from embedder import embed_text
from vector_store import MultimodalVectorStore

store = MultimodalVectorStore(persist_dir="./chroma_db")

def generate_query_variants(query: str) -> list[str]:
    """Generate multiple phrasings for a single question."""
    import google.generativeai as genai
    import os
    genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
    model = genai.GenerativeModel("gemini-2.0-flash")
    prompt = (
        f"Generate 3 different ways to search for information about: {query}\n\n"
        "Return exactly 3 queries, one per line, no numbering or bullets."
    )
    response = model.generate_content(prompt)
    variants = [line.strip() for line in response.text.strip().split("\n") if line.strip()]
    return ([query] + variants)[:4]  # Always include original, cap at 4 total

def multi_query_search(query: str, n_per_query: int = 4) -> list[dict]:
    """
    Search with multiple query variants and deduplicate results.
    Returns unique chunks ranked by their best similarity score.
    """
    variants = generate_query_variants(query)
    def search_one(variant: str) -> list[dict]:
        embedding = embed_text(variant, task_type="RETRIEVAL_QUERY")
        return store.search(embedding, n_results=n_per_query)
    # Run all variants in parallel
    all_results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(search_one, v): v for v in variants}
        for future in concurrent.futures.as_completed(futures):
            all_results.extend(future.result())
    # Deduplicate by source file + chunk index, keeping best similarity score
    seen = {}
    for chunk in all_results:
        meta = chunk["metadata"]
        key = f"{meta['source_file']}:{meta['chunk_index']}"
        if key not in seen or chunk["similarity"] > seen[key]["similarity"]:
            seen[key] = chunk
    return sorted(seen.values(), key=lambda x: x["similarity"], reverse=True)

4. Neu bewerten der abgerufenen Fragmente

Während der Phase des Neu bewertens der abgerufenen Fragmente sollten Sie zunächst den Ablaufplan aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch die Fehlerbehebung zusammen. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Messen Sie die Erinnerungsfähigkeit anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen – eine häufige Änderung der Anfragen behebt in der Regel nicht ein schwaches Abrufverhalten.

# reranker.py
from sentence_transformers import CrossEncoder

# This model runs locally, no API cost, fast inference
_reranker = None

def get_reranker():
    global _reranker
    if _reranker is None:
        _reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
    return _reranker

def rerank_chunks(query: str, chunks: list[dict], top_k: int = 5) -> list[dict]:
    """
    Re-rank retrieved chunks using a cross-encoder model.
    Cross-encoders read both query and chunk together, giving much
    more precise relevance scores than embedding cosine similarity.
    Only practical on a small candidate set (10-20 chunks).
    """
    reranker = get_reranker()
    # For video/audio, we use the metadata preview as the text input.
    # For text chunks, we use the actual content.
    pairs = []
    for chunk in chunks:
        if chunk["modality"] == "text":
            doc_text = chunk["content"]
        else:
            meta = chunk["metadata"]
            doc_text = (
                f"{chunk['modality']} recording: {meta['source_file']} "
                f"at {meta.get('timestamp_start', '')} to {meta.get('timestamp_end', '')}"
            )
        pairs.append([query, doc_text])
    scores = reranker.predict(pairs)
    for chunk, score in zip(chunks, scores):
        chunk["rerank_score"] = float(score)
    return sorted(chunks, key=lambda x: x["rerank_score"], reverse=True)[:top_k]

# Usage:
# candidates = store.search(query_embedding, n_results=20)  # Retrieve wide
# final = rerank_chunks(query, candidates, top_k=5)         # Re-rank narro

5. Ein Ähnlichkeitsschwellenwert festlegen und daran halten

Beim Durchlaufen der Phase „5 Set a Similarity“ sollten Sie zunächst den Vertrag aufschreiben: die erforderlichen Eingaben, das Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten.

Die Ergebnisse

Beim Arbeiten in der Phase „Die Ergebnisse“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgsprüfungen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Messen Sie die Erinnerungsfähigkeit anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten. Beim Arbeiten in der Phase „Die Ergebnisse“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Legen Sie die Konfiguration außerhalb des Anwendungscode ab. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

Was die Matryoshka-Dimensionen für Ihre Infrastruktur bedeuten

Die What the Matryoshka Dimensions-Ebene funktioniert am besten, wenn sie als messbare Oberfläche betrachtet wird. Erfassen Sie einen erfolgreichen Fall, einen Fehlerfall sowie die Rollback-Anmerkung, bevor Sie den Umfang erweitern. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Trennen Sie die Chunking-Strategie von der Abrufstrategie – ein Änderung in einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

Was Sie damit bauen können

Die Phase „Was Sie bauen können“ funktioniert am besten, wenn sie als messbarer Rahmen betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein gelungenes Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf. Trennen Sie die Aufteilungspolitik von der Abrufpolitik. Ein Änderungsbedarf bei einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

Praktische Erkenntnisse

Die Phase „Actionable Takeaways“ funktioniert am besten, wenn sie als messbarer Rahmen betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Trennen Sie die Strategie zur Aufteilung in Teile von der Strategie zum Abrufen. Eine Änderung sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern. Die Phase „Actionable Takeaways“ funktioniert am besten, wenn sie als messbarer Rahmen betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreuer ohne das Durchlesen des gesamten Systems prüfen können.

Was fehlt noch

Zur Phase „Was fehlt noch“ sollten die Eingabedaten, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

Lassen Sie uns gemeinsam weiter lernen

In der Phase „Lass uns weiter lernen“ sollten Eingaben, Verantwortliche für die jeweiligen Schritte sowie Abbruchkriterien definiert werden, bevor Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Vorzuziehen sind kleine, testbare Einheiten statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

Operative Checkliste

Die Phase der operativen Checkliste funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo in gemeinsame Umgebungen übergeht.

Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Änderung einer sollte nicht dazu zwingen, die andere neu zu schreiben, wenn sich die Qualitätsmetriken ändern.

Fügen Sie immer dann, wenn das Budget es zulässt, einen Smoke-Test hinzu, der den kritischen Pfad in CI mit Fixtures und nicht mit live genutzten, bezahlten APIs testet.

Lassen Sie die Konfiguration außerhalb des Anwendungscode liegen. Umgebungsdateien, Geheimnisdatenspeicher und Feature-Flags sollten an einem Ort gespeichert werden, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können.

Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Änderung einer sollte nicht dazu zwingen, die andere neu zu schreiben, wenn sich die Qualitätsmetriken ändern.

Vor der Einführung des gesamten Stack sollten Sie die Versionen einfrieren, ein „goldenes“ Transkript des kritischen Pfads erstellen und die Rollback-Schritte überprüfen. Gemeinsam genutzte Umgebungen benötigen Rate-Limits, Überprüfungen der Nutzerzuordnung sowie einen klaren Verantwortlichen für die Rotation von Geheimnissen. Wählen Sie langweilige Zuverlässigkeit statt cleverer, einmaliger Demonstrationen.

Batch-Hinweis für b3567bc23c05: Halten Sie die Anbieter-Schlüssel außerhalb des Repositories, legen Sie eine Obergrenze für Tokens pro Sitzung fest und speichern Sie die Transkripte neben den Evaluierungs-Dateien, damit spätere Modellwechsel vergleichbar bleiben.