Chain-of-Thought RAG: Schrittweises Schlussfolgern für bessere Antworten

TL;DR

Der Chain-of-Thought (CoT) zwingt das LLM dazu, sein Denken zu erläutern, bevor es antwortet. Im RAG verbessert diese Technik die Qualität komplexer Antworten um 35–45%, reduziert Halluzinationen und ermöglicht die Nachverfolgbarkeit des Dokumentensyntheseprozesses. Dieser Leitfaden behandelt die verschiedenen Varianten von CoT, deren Implementierung und optimale Anwendungsfälle.

Was ist Chain-of-Thought?

Das Problem direkter Antworten

Ohne CoT generiert ein LLM eine sofortige Antwort, was bei komplexen Fragen zu Fehlern führen kann:

DEVELOPERpython
# ❌ Direkte Antwort (problematisch bei komplexen Fragen)
prompt = """
Documents: [3 articles sur les politiques de retour]
Question: Un client a acheté un produit personnalisé il y a 20 jours,
peut-il le retourner ?
"""

# Das LLM kann Denkschritte überspringen und sich irren

Die Lösung: Chain-of-Thought

CoT fordert das LLM auf, Schritt für Schritt zu denken:

DEVELOPERpython
# ✅ Mit Chain-of-Thought
prompt = """
Documents: [3 articles sur les politiques de retour]
Question: Un client a acheté un produit personnalisé il y a 20 jours,
peut-il le retourner ?

Raisonne étape par étape avant de répondre:
1. Identifie les règles pertinentes dans les documents
2. Vérifie chaque condition applicable
3. Tire une conclusion basée sur l'analyse

Ton raisonnement:
"""

# Antwort des LLM:
# "1. Dokument 1 nennt eine Rückgabefrist von 30 Tagen.
#  2. Dokument 2 präzisiert, dass personalisierte Produkte ausgeschlossen sind.
#  3. Obwohl die Frist (20 Tage < 30 Tage) eingehalten wird,
#     greift der Ausschluss personalisierter Produkte.
#  Fazit: Nein, die Rückgabe ist nicht möglich, da personalisierte
#  Produkte von der Rückgaberichtlinie ausgeschlossen sind."

Varianten des Chain-of-Thought

1. Zero-Shot CoT

Die einfachste Form: "Denke Schritt für Schritt" hinzufügen:

DEVELOPERpython
ZERO_SHOT_COT_PROMPT = """
Documents:
{context}

Question: {query}

Réfléchis étape par étape, puis donne ta réponse finale.
"""

Vorteile: Einfach zu implementieren Grenzen: Weniger strukturiert, variable Qualität

2. Few-Shot CoT

Beispiele für Denkprozesse liefern:

DEVELOPERpython
FEW_SHOT_COT_PROMPT = """
Voici comment analyser une question avec les documents:

## Exemple 1
Question: Le produit X est-il compatible avec Windows 11?
Documents: "Le produit X fonctionne sur Windows 10 et macOS 12+"
Raisonnement:
- Le document mentionne Windows 10 et macOS 12+
- Windows 11 n'est pas explicitement mentionné
- Cependant, Windows 11 est rétrocompatible avec Windows 10
- MAIS je ne dois pas supposer la compatibilité sans confirmation
Réponse: La compatibilité Windows 11 n'est pas confirmée dans la
documentation. Le produit fonctionne sur Windows 10. Je recommande
de contacter le support pour confirmation.

## Exemple 2
Question: Puis-je annuler ma commande après expédition?
Documents: "Les annulations sont possibles avant expédition. Après
expédition, utilisez notre processus de retour standard."
Raisonnement:
- Le document distingue avant/après expédition
- Avant expédition: annulation possible
- Après expédition: pas d'annulation, mais retour possible
Réponse: Une fois la commande expédiée, l'annulation n'est plus
possible. Vous pouvez cependant effectuer un retour selon notre
politique standard une fois le colis reçu.

---

Maintenant, analyse cette question:

Documents:
{context}

Question: {query}

Raisonnement:
"""

3. Self-Consistency CoT

Mehrere Denkpfade generieren und den Konsens ermitteln:

DEVELOPERpython
import asyncio
from collections import Counter

class SelfConsistencyCoT:
    def __init__(self, llm_client, num_paths=5):
        self.llm = llm_client
        self.num_paths = num_paths

    async def generate_with_consistency(
        self,
        context: str,
        query: str
    ) -> dict:
        """
        Generiert mehrere Argumentationsketten und
        gibt die Mehrheitsantwort zurück.
        """

        # N Denkpfade parallel generieren
        tasks = [
            self._generate_single_path(context, query)
            for _ in range(self.num_paths)
        ]
        results = await asyncio.gather(*tasks)

        # Endantworten extrahieren
        answers = [r["answer"] for r in results]

        # Konsens finden
        answer_counts = Counter(answers)
        consensus_answer, count = answer_counts.most_common(1)[0]
        confidence = count / self.num_paths

        return {
            "answer": consensus_answer,
            "confidence": confidence,
            "reasoning_paths": results,
            "agreement": f"{count}/{self.num_paths}"
        }

    async def _generate_single_path(
        self,
        context: str,
        query: str
    ) -> dict:
        prompt = f"""
        Documents: {context}

        Question: {query}

        Raisonne étape par étape, puis donne ta réponse finale.
        Format ta réponse comme:
        RAISONNEMENT: [ton analyse]
        RÉPONSE: [ta conclusion]
        """

        response = await self.llm.generate(
            prompt,
            temperature=0.7  # Höhere Temperatur für Vielfalt
        )

        return self._parse_response(response)

4. Tree-of-Thought (ToT)

Mehrere Denkzweige erkunden:

DEVELOPERpython
class TreeOfThought:
    """
    Erkundet mehrere Denkzweige
    und wählt den besten aus.
    """

    def __init__(self, llm_client, max_depth=3, branching_factor=3):
        self.llm = llm_client
        self.max_depth = max_depth
        self.branching_factor = branching_factor

    async def solve(self, context: str, query: str) -> dict:
        """Löst ein Problem mit Tree-of-Thought."""

        # Wurzel: Anfangszustand
        root = ThoughtNode(
            state=f"Question: {query}\nContext: {context}",
            parent=None
        )

        # Baum erkunden
        best_leaf = await self._explore(root, depth=0)

        return {
            "answer": best_leaf.conclusion,
            "reasoning_path": best_leaf.get_path(),
            "alternatives_explored": self._count_nodes(root)
        }

    async def _explore(self, node: ThoughtNode, depth: int) -> ThoughtNode:
        if depth >= self.max_depth:
            # Bewerten und abschließen
            node.conclusion = await self._generate_conclusion(node)
            node.score = await self._evaluate(node)
            return node

        # Zweige generieren (mögliche Denkschritte)
        branches = await self._generate_branches(node)

        # Vielversprechende Zweige bewerten und filtern
        scored_branches = []
        for branch in branches:
            branch.score = await self._evaluate(branch)
            scored_branches.append(branch)

        # Die besten Zweige behalten
        top_branches = sorted(
            scored_branches,
            key=lambda x: x.score,
            reverse=True
        )[:self.branching_factor]

        # Rekursiv erkunden
        best_leaf = None
        for branch in top_branches:
            leaf = await self._explore(branch, depth + 1)
            if best_leaf is None or leaf.score > best_leaf.score:
                best_leaf = leaf

        return best_leaf

    async def _generate_branches(self, node: ThoughtNode) -> list:
        """Generiert die nächsten möglichen Denkschritte."""
        prompt = f"""
        État actuel du raisonnement:
        {node.state}

        Génère 3 prochaines étapes de raisonnement différentes.
        Format:
        ÉTAPE 1: [description]
        ÉTAPE 2: [description]
        ÉTAPE 3: [description]
        """

        response = await self.llm.generate(prompt, temperature=0.8)
        return self._parse_branches(response, node)

Praktische Implementierung für RAG

Vollständiges CoT-Template

DEVELOPERpython
RAG_COT_PROMPT = """
Tu es un assistant qui analyse les documents pour répondre aux questions.

## Documents disponibles
{context}

## Question
{query}

## Processus d'analyse (à suivre obligatoirement)

### Étape 1: Identification des informations pertinentes
Parcours chaque document et identifie les passages qui concernent
la question. Cite les passages exacts.

### Étape 2: Analyse des informations
Pour chaque passage pertinent:
- Que dit-il exactement?
- Est-ce une réponse directe ou partielle?
- Y a-t-il des conditions ou exceptions?

### Étape 3: Vérification de cohérence
- Les documents se contredisent-ils?
- Y a-t-il des informations manquantes?
- Quelles sont mes certitudes et incertitudes?

### Étape 4: Synthèse et réponse
Formule une réponse claire basée sur l'analyse ci-dessus.
Cite les sources.

---

## Ton analyse

### Étape 1: Informations pertinentes
"""

def build_cot_prompt(context: str, query: str) -> str:
    return RAG_COT_PROMPT.format(context=context, query=query)

Parsing der CoT-Antwort

DEVELOPERpython
import re

class CoTResponseParser:
    """Parst eine strukturierte Chain-of-Thought-Antwort."""

    def parse(self, response: str) -> dict:
        sections = {
            "relevant_info": self._extract_section(response, "Étape 1", "Étape 2"),
            "analysis": self._extract_section(response, "Étape 2", "Étape 3"),
            "consistency_check": self._extract_section(response, "Étape 3", "Étape 4"),
            "final_answer": self._extract_section(response, "Étape 4", None)
        }

        return {
            "reasoning": sections,
            "answer": self._extract_final_answer(sections["final_answer"]),
            "confidence": self._assess_confidence(sections),
            "sources_cited": self._extract_sources(response)
        }

    def _extract_section(
        self,
        text: str,
        start_marker: str,
        end_marker: str
    ) -> str:
        pattern = f"{start_marker}[^#]*?(?={end_marker}|$)" if end_marker else f"{start_marker}.*"
        match = re.search(pattern, text, re.DOTALL)
        return match.group(0) if match else ""

    def _assess_confidence(self, sections: dict) -> float:
        """Bewertet die Konfidenz basierend auf dem Denkprozess."""
        confidence = 1.0

        # Reduzieren, wenn Unsicherheiten erwähnt werden
        uncertainty_phrases = [
            "pas certain", "incertain", "manque", "contradictoire",
            "pas mentionné", "pas clair", "ambigu"
        ]

        full_text = " ".join(sections.values()).lower()
        for phrase in uncertainty_phrases:
            if phrase in full_text:
                confidence -= 0.15

        return max(0.2, min(1.0, confidence))

Validierung des Denkprozesses

DEVELOPERpython
class ReasoningValidator:
    """Validiert die Qualität des CoT-Denkprozesses."""

    def __init__(self, llm_client):
        self.llm = llm_client

    async def validate(
        self,
        context: str,
        query: str,
        reasoning: str,
        answer: str
    ) -> dict:
        """
        Prüft, ob der Denkprozess valide ist und
        ob die Schlussfolgerung logisch daraus folgt.
        """

        validation_prompt = f"""
        Évalue la qualité de ce raisonnement:

        QUESTION: {query}
        DOCUMENTS: {context}
        RAISONNEMENT: {reasoning}
        CONCLUSION: {answer}

        Vérifie:
        1. Le raisonnement utilise-t-il les documents fournis?
        2. La conclusion découle-t-elle logiquement du raisonnement?
        3. Y a-t-il des sauts logiques ou des suppositions non justifiées?
        4. La réponse est-elle fidèle aux documents (pas d'hallucination)?

        Réponds au format:
        VALIDE: [OUI/NON]
        SCORE: [1-10]
        PROBLÈMES: [liste si applicable]
        """

        response = await self.llm.generate(
            validation_prompt,
            temperature=0.1  # Deterministische Validierung
        )

        return self._parse_validation(response)

Optimierungen für RAG

1. Selektiver CoT

CoT nur für komplexe Fragen verwenden:

DEVELOPERpython
class SelectiveCoT:
    """Nutzt CoT nur wenn nötig."""

    def __init__(self, llm_client, complexity_threshold=0.6):
        self.llm = llm_client
        self.threshold = complexity_threshold

    async def answer(self, context: str, query: str) -> dict:
        # Komplexität der Frage bewerten
        complexity = await self._assess_complexity(query, context)

        if complexity < self.threshold:
            # Einfache Frage: direkte Antwort
            return await self._direct_answer(context, query)
        else:
            # Komplexe Frage: Chain-of-Thought
            return await self._cot_answer(context, query)

    async def _assess_complexity(self, query: str, context: str) -> float:
        """Bewertet die Komplexität von 0 bis 1."""
        complexity_indicators = {
            "multi_step": any(w in query.lower() for w in ["und", "dann", "anschließend", "auch"]),
            "conditional": any(w in query.lower() for w in ["wenn", "falls", "sofern", "bedingung"]),
            "comparison": any(w in query.lower() for w in ["vergleichen", "unterschied", "versus", "oder"]),
            "multi_doc": len(context.split("Document")) > 2,
            "long_query": len(query.split()) > 15
        }

        return sum(complexity_indicators.values()) / len(complexity_indicators)

2. CoT mit Inline-Zitaten

Das LLM zwingen, während des Denkens zu zitieren:

DEVELOPERpython
COT_WITH_CITATIONS_PROMPT = """
Analyse les documents et réponds en citant tes sources à chaque étape.

## Documents
{context}

## Question
{query}

## Analyse avec citations

### Étape 1: Faits pertinents
- Fait 1: "[citation exacte]" [Source: Doc X]
- Fait 2: "[citation exacte]" [Source: Doc Y]

### Étape 2: Raisonnement
En combinant le fait 1 [Doc X] et le fait 2 [Doc Y], on peut déduire que...

### Étape 3: Conclusion
Basé sur [Doc X] et [Doc Y]: [réponse finale]
"""

3. Parallelisierter CoT

CoT durch Parallelisierung beschleunigen:

DEVELOPERpython
class ParallelCoT:
    """Parallelisiert unabhängige Denkschritte."""

    async def analyze_documents(
        self,
        documents: list,
        query: str
    ) -> dict:
        # Schritt 1: Jedes Dokument parallel analysieren
        analysis_tasks = [
            self._analyze_single_document(doc, query)
            for doc in documents
        ]
        doc_analyses = await asyncio.gather(*analysis_tasks)

        # Schritt 2: Analysen zusammenführen
        synthesis = await self._synthesize(doc_analyses, query)

        # Schritt 3: Antwort formulieren
        answer = await self._formulate_answer(synthesis, query)

        return {
            "document_analyses": doc_analyses,
            "synthesis": synthesis,
            "answer": answer
        }

    async def _analyze_single_document(
        self,
        document: str,
        query: str
    ) -> dict:
        prompt = f"""
        Document: {document}
        Question: {query}

        Analyse ce document par rapport à la question:
        1. Informations pertinentes trouvées: [liste]
        2. Répond-il à la question: [oui/partiellement/non]
        3. Informations clés: [résumé]
        """

        return await self.llm.generate(prompt)

Metriken und Bewertung

CoT-spezifische Metriken

DEVELOPERpython
class CoTMetrics:
    """Metriken zur Bewertung der Chain-of-Thought-Qualität."""

    def evaluate(self, cot_response: dict) -> dict:
        return {
            "reasoning_depth": self._measure_depth(cot_response),
            "source_grounding": self._measure_grounding(cot_response),
            "logical_coherence": self._measure_coherence(cot_response),
            "conclusion_validity": self._measure_validity(cot_response)
        }

    def _measure_depth(self, response: dict) -> float:
        """Misst die Tiefe des Denkprozesses (Anzahl der Schritte)."""
        reasoning = response.get("reasoning", {})
        steps = [v for v in reasoning.values() if v.strip()]
        return min(len(steps) / 4, 1.0)  # 4 Schritte = maximale Bewertung

    def _measure_grounding(self, response: dict) -> float:
        """Misst die Verankerung in den Quellen."""
        citations = response.get("sources_cited", [])
        # Mehr Zitate = bessere Verankerung
        return min(len(citations) / 3, 1.0)

    def _measure_coherence(self, response: dict) -> float:
        """Misst die logische Kohärenz des Denkprozesses."""
        # Vereinfachte Implementierung - in Produktion ein Modell verwenden
        reasoning_text = str(response.get("reasoning", ""))

        # Kohärenzindikatoren
        coherence_markers = ["also", "folglich", "somit", "denn", "weil"]
        marker_count = sum(1 for m in coherence_markers if m in reasoning_text.lower())

        return min(marker_count / 3, 1.0)

Optimale Anwendungsfälle für CoT

Wann CoT im RAG einsetzen

Integration mit Ailog

Ailog unterstützt Chain-of-Thought nativ:

DEVELOPERpython
from ailog import AilogClient

client = AilogClient(api_key="your-key")

response = client.chat(
    channel_id="support-widget",
    message="Puis-je combiner la réduction membre et la promo en cours?",
    reasoning_mode="chain_of_thought",  # Aktiviert CoT
    cot_settings={
        "show_reasoning": True,  # Denkprozess dem Benutzer anzeigen
        "validate_reasoning": True,  # Logik validieren
        "max_steps": 4
    }
)

print(response.reasoning)  # Denkschritte
print(response.answer)     # Endantwort
print(response.confidence) # Konfidenzwert

Fazit

Chain-of-Thought verbessert komplexe RAG-Antworten erheblich. Die wichtigsten Punkte:

1. CoT selektiv einsetzen für komplexe Fragen
2. Few-Shot ist zuverlässiger als Zero-Shot
3. Self-Consistency erhöht die Zuverlässigkeit
4. Den Denkprozess validieren um logische Fehler zu vermeiden
5. Während des Denkens zitieren für die Nachverfolgbarkeit

Weiterführende Ressourcen

• Einführung in RAG - Grundlagen
• LLM-Generierung für RAG - Übergeordneter Leitfaden
• Prompt Engineering RAG - Prompts optimieren
• Strukturierte Outputs RAG - Ausgabeformate

Lust auf fortgeschrittenes Schlussfolgern ohne Komplexität? Testen Sie Ailog - Chain-of-Thought integriert, automatische Validierung, garantierte Konfidenz.