LlamaIndex CondenseQuestion engine for chat-based RAG

The CondenseQuestion engine transforms multi-turn conversations into standalone queries before retrieval, making it the practical choice for chat-based RAG. Unlike simpler engines that stuff raw chat history into the prompt, CondenseQuestion uses an LLM to rewrite follow-up questions with full context, then runs a single retrieval pass. This tutorial builds a working implementation from document ingestion through streaming responses.

Prerequisites

You need Python 3.10+ and an OpenAI-compatible API key. Install the core dependencies:

pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
    llama-index-vector-stores-chroma chromadb pypdf

Set your API key and base URL (if using a gateway):

export OPENAI_API_KEY="sk-..."
# Optional: export OPENAI_BASE_URL="https://api.n4n.ai/v1"

The examples assume a directory ./data containing PDFs or text files you want to query.

Ingest documents into a vector index

LlamaIndex separates ingestion from querying. First, load documents, chunk them, embed, and persist to Chroma.

# ingest.py
import os
from pathlib import Path
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    Settings,
)
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# Configure global settings
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
Settings.chunk_overlap = 50

# Persistent Chroma client
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("rag_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Load and index
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context, show_progress=True
)
print(f"Indexed {len(documents)} documents into Chroma")

Run it:

python ingest.py

Expected output:

Indexed 4 documents into Chroma

The index now persists in ./chroma_db. Re-running ingest.py adds to the existing collection; delete the directory to start fresh.

Build the CondenseQuestion query engine

The engine has three pieces: a retriever, a question condenser, and a response synthesizer. Wire them together explicitly so you can swap components later.

# query_engine.py
from llama_index.core import (
    VectorStoreIndex,
    StorageContext,
    Settings,
    get_response_synthesizer,
)
from llama_index.core.query_engine import CondenseQuestionChatEngine
from llama_index.core.chat_engine import CondenseQuestionChatEngine as CQCE
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# Settings (must match ingestion)
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# Load existing index
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_collection("rag_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_vector_store(vector_store, storage_context=storage_context)

# Retriever: top-k with similarity threshold
retriever = index.as_retriever(similarity_top_k=4, similarity_cutoff=0.7)

# Condense prompt: rewrites follow-ups into standalone questions
from llama_index.core.prompts import PromptTemplate

condense_prompt = PromptTemplate(
    "Given the conversation history and a follow-up question, "
    "rephrase the follow-up as a standalone question.\n\n"
    "Chat History:\n{chat_history}\n\n"
    "Follow-up Question: {question}\n\n"
    "Standalone Question:"
)

# Response synthesizer: compact mode fits more context
response_synthesizer = get_response_synthesizer(
    response_mode="compact",
    verbose=True,
)

# Memory buffer keeps last N turns for condensing
memory = ChatMemoryBuffer.from_defaults(token_limit=3000)

# Assemble the chat engine
chat_engine = CondenseQuestionChatEngine.from_defaults(
    retriever=retriever,
    response_synthesizer=response_synthesizer,
    condense_question_prompt=condense_prompt,
    memory=memory,
    verbose=True,
)

print("Chat engine ready. Type 'exit' to quit.")

Run a multi-turn conversation

Add a simple REPL to see the engine in action. The condenser rewrites each follow-up; the retriever runs once per turn; the synthesizer streams the answer.

# chat.py
from query_engine import chat_engine

def main():
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in {"exit", "quit"}:
            break
        
        # Streaming response
        response = chat_engine.stream_chat(user_input)
        print("Assistant: ", end="", flush=True)
        for token in response.response_gen:
            print(token, end="", flush=True)
        print()  # newline after stream
        
        # Show condensed question for debugging
        if hasattr(response, "condensed_question"):
            print(f"\n[Condensed: {response.condensed_question}]")

if __name__ == "__main__":
    main()

Run it:

python chat.py

Sample session with a corpus about Kubernetes:

You: What is a Kubernetes Deployment?
Assistant: A Kubernetes Deployment manages a set of identical Pods...
[Condensed: What is a Kubernetes Deployment?]

You: How do I roll back a failed one?
Assistant: Use `kubectl rollout undo deployment/<name>` to revert...
[Condensed: How do I roll back a failed Kubernetes Deployment?]

You: What about canary deployments?
Assistant: For canary deployments, you typically create a second Deployment...
[Condensed: What about canary deployments in Kubernetes?]

Notice the condensed questions absorb context (“in Kubernetes”) without you repeating it.

Inspect retrieval and synthesis internals

Verbose mode prints the retrieved nodes and the synthesized prompt. Enable it temporarily to debug relevance issues.

# debug_retrieval.py
from query_engine import chat_engine, retriever

# Direct retriever inspection
nodes = retriever.retrieve("How do I roll back a failed Deployment?")
for i, node in enumerate(nodes):
    print(f"Node {i}: score={node.score:.3f}")
    print(node.text[:200])
    print("---")

# Full engine trace
response = chat_engine.chat("How do I roll back a failed Deployment?")
print("\nFinal response:", response.response)

Output shows which chunks actually fed the answer:

Node 0: score=0.842
A Deployment's rollout history is stored in ReplicaSets. To roll back...
---
Node 1: score=0.791
kubectl rollout undo deployment/myapp --to-revision=2 ...
---
Node 2: score=0.715
If a Deployment gets stuck, check `kubectl rollout status`...
---

If scores cluster below your cutoff, adjust similarity_cutoff or increase top_k.

Customize the condense prompt for domain behavior

The default condense prompt works for general chat. For technical domains, add instructions that preserve terminology.

# domain_condense.py
from llama_index.core.prompts import PromptTemplate

k8s_condense_prompt = PromptTemplate(
    "You are a Kubernetes expert. Rewrite the follow-up question "
    "as a standalone question that preserves all technical terms, "
    "resource kinds, and CLI flags.\n\n"
    "Chat History:\n{chat_history}\n\n"
    "Follow-up Question: {question}\n\n"
    "Standalone Question:"
)

# Pass to engine constructor:
# chat_engine = CondenseQuestionChatEngine.from_defaults(
#     ..., condense_question_prompt=k8s_condense_prompt, ...
# )

Test the difference:

Follow-up: "What about the dry-run flag?"
Generic condensed: "What is the dry-run flag?"
Domain condensed: "What does the --dry-run flag do in kubectl apply?"

The domain version keeps the kubectl apply context that the generic version drops.

Handle streaming with source citations

Production chat surfaces sources. The response object includes source_nodes; stream the answer first, then append citations.

# chat_with_sources.py
from query_engine import chat_engine

def main():
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in {"exit", "quit"}:
            break
        
        streaming_response = chat_engine.stream_chat(user_input)
        print("Assistant: ", end="", flush=True)
        full_response = ""
        for token in streaming_response.response_gen:
            print(token, end="", flush=True)
            full_response += token
        print()
        
        # Citations after stream completes
        if streaming_response.source_nodes:
            print("\nSources:")
            for i, node in enumerate(streaming_response.source_nodes):
                meta = node.metadata
                src = meta.get("file_name", meta.get("source", f"doc_{i}"))
                page = meta.get("page_label", "?")
                print(f"  [{i+1}] {src} (page {page}) score={node.score:.3f}")

if __name__ == "__main__":
    main()

Sample output:

Assistant: Use `kubectl rollout undo deployment/myapp` to revert...

Sources:
  [1] k8s_deployments.pdf (page 12) score=0.842
  [2] kubectl_cheatsheet.md (page 1) score=0.791

Persist conversation state

ChatMemoryBuffer lives in memory. For multi-session persistence, serialize the memory to JSON and reload.

# persist_memory.py
import json
from query_engine import chat_engine, memory

def save_memory(path: str):
    data = {
        "chat_history": [
            {"role": msg.role.value, "content": msg.content}
            for msg in memory.get()
        ]
    }
    with open(path, "w") as f:
        json.dump(data, f, indent=2)

def load_memory(path: str):
    with open(path) as f:
        data = json.load(f)
    memory.reset()
    for msg in data["chat_history"]:
        memory.put(type("Msg", (), {"role": msg["role"], "content": msg["content"]})())

# Usage
load_memory("session.json")
# ... chat ...
save_memory("session.json")

Note: CondenseQuestionChatEngine re-condenses the full history on each turn, so long conversations grow the condenser prompt. For very long sessions, consider summarizing history periodically or switching to a ContextChatEngine with a fixed context window.

Swap the LLM for a local model

The engine is LLM-agnostic. Replace the OpenAI LLM with any LlamaIndex-compatible local model (Ollama, vLLM, llama.cpp).

# local_llm.py
from llama_index.llms.ollama import Ollama

Settings.llm = Ollama(model="llama3.1:8b", request_timeout=120.0)
# Rest of engine construction unchanged

The condenser and synthesizer both use Settings.llm by default. If you want different models for condensing vs. answering, pass llm and condense_llm explicitly to from_defaults.

Evaluate condenser quality

Bad condensing breaks retrieval. Write a small eval that checks whether the condensed question retrieves the same top document as the gold standalone question.

# eval_condense.py
from query_engine import chat_engine, retriever

test_cases = [
    {
        "history": ["What is a Kubernetes Service?"],
        "followup": "How does ClusterIP differ from NodePort?",
        "gold": "How does ClusterIP Service differ from NodePort Service in Kubernetes?",
    },
    # Add more cases
]

for tc in test_cases:
    # Simulate history
    for h in tc["history"]:
        chat_engine.chat(h)
    
    # Get condensed question
    response = chat_engine.chat(tc["followup"])
    condensed = response.condensed_question
    
    # Retrieve with condensed vs gold
    condensed_nodes = retriever.retrieve(condensed)
    gold_nodes = retriever.retrieve(tc["gold"])
    
    top_condensed = condensed_nodes[0].node_id if condensed_nodes else None
    top_gold = gold_nodes[0].node_id if gold_nodes else None
    
    match = "✓" if top_condensed == top_gold else "✗"
    print(f"{match} Condensed: {condensed}")
    print(f"   Gold:      {tc['gold']}")
    print(f"   Top match: {top_condensed == top_gold}")

Run periodically when you tweak the condense prompt.

When to choose CondenseQuestion over other engines

CondenseQuestion hits the sweet spot for typical chat RAG: one retrieval, context-aware rewriting, predictable latency.

Common failure modes and fixes

Condenser hallucinates terms not in history
Lower condenser temperature (Settings.llm.temperature = 0.0) or add “Do not introduce new information” to the condense prompt.

Retrieval misses after condensing
Check similarity_cutoff. A condensed question may be shorter and less specific than the original follow-up. Increase top_k or lower the cutoff.

Memory grows unbounded
ChatMemoryBuffer.token_limit truncates by token count, not turn count. Set it to roughly 50% of your condenser prompt’s token budget.

Streaming cuts off mid-sentence
Ensure the synthesizer’s response_mode="compact" (default) or "tree_summarize" for longer answers. "refine" mode doesn’t stream cleanly.

Next steps

• Add a reranker (SentenceTransformerRerank) after retrieval to improve precision
• Implement hybrid search (vector + BM25) via ChromaVectorStore metadata filters
• Wrap the engine in a FastAPI endpoint with session management for production deployment
• Log condensed questions and retrieval scores to monitor condenser drift over time

The CondenseQuestion engine gives you a maintainable, debuggable chat RAG loop. Start with the explicit component wiring shown here — retriever, condenser, synthesizer, memory — and swap pieces as requirements evolve.