Conversational RAG

Overview

ConversationalRAG adds multi-turn memory around a backend rag_system, persisting each question/answer pair to JSON and using recent turns to enhance follow-up queries (e.g. resolving pronouns like "he"/"it").

RAGConfigConverstion

@dataclass
class RAGConfigConverstion:
    max_history_turns: int = 10
    context_turns: int = 3
    enable_context_compression: bool = False
    include_sources: bool = True
    use_history: bool = True
    auto_save_history: bool = True
    history_save_path: str = "conversations"

[!NOTE] Name is misspelled in source The class is literally named RAGConfigConverstion (missing an "a") in conversation_config.py. This is documented as-is since renaming it would break compatibility with existing imports.

ConversationTurn

@dataclass
class ConversationTurn:
    query: str
    answer: str
    timestamp: datetime
    chunks: Optional[List] = None
    retrieved_chunks: int = 0

    def __str__(self) -> str: ...
    def to_dict(self) -> dict: ...
    @classmethod
    def from_dict(cls, data: dict) -> "ConversationTurn": ...

ConversationHistory

class ConversationHistory:
    def __init__(self, max_turns=..., history_file: Optional[str] = None): ...
    def add_turn(self, query: str, answer: str, chunks: List[str] = None, retrieved_chunks: int = 0): ...
    def get_recent_turns(self, n: int = 3) -> List[ConversationTurn]: ...
    def format_for_prompt(self, n: int = 3, language: str = "ar") -> str: ...
    def clear(self): ...
    def get_file_path(self) -> str: ...
    def __len__(self) -> int: ...

Manages an in-memory deque-like list of turns with live JSON persistence: add_turn both appends to memory and calls the private _save_to_file. history_file=None triggers automatic file naming; passing an explicit path loads existing history via _load_from_file on construction.

ConversationalRAG

Constructor

def __init__(
    self,
    rag_system,
    max_history_turns: int = config.max_history_turns,
    context_turns: int = config.context_turns,
    enable_context_compression: bool = config.enable_context_compression,
    instructions: Optional[str] = None,
    lang: str = "ar",
    history_file: Optional[str] = None,
    auto_save: bool = True,
)

ask(query, include_sources=config.include_sources, use_history=config.use_history) -> str

1. _enhance_query_with_context optionally rewrites the incoming query using recent history (only if use_history=True).
2. _build_conversational_prompt assembles a strict prompt combining conversation history and retrieved context.
3. Calls the backend, formats sources via _format_sources if requested, and returns the answer string.
4. The turn is recorded in ConversationHistory (and saved to disk if auto_save).

Other synchronous methods

Async API

async def aask(self, query: str, use_history: bool = True, include_sources: bool = False, **kwargs) -> str
async def astream(self, query: str, use_history: bool = True)

aask maintains the same history context as ask. astream streams the final answer while still updating history. __aenter__/__aexit__ support async with.

Complete example

from zerda.rag.types.conversational_rag import ConversationalRAG

convo = ConversationalRAG(
    backend,
    lang="en",
    max_history_turns=20,
    context_turns=5,
    history_file="sessions/user_42.json",
)

print(convo.ask("Who directed Inception?"))
print(convo.ask("What year was it released?"))  # uses history to resolve "it"

summary = convo.get_conversation_summary()
print(summary)

Best practices

• Set an explicit history_file per user/session in multi-tenant apps; otherwise auto-naming behavior (undocumented specifics) may collide.
• Keep context_turns well below max_history_turns — only recent turns need to be echoed into every prompt to control token cost.

Common mistakes

• Forgetting auto_save=False disables persistence but not in-memory history — ConversationHistory still keeps turns for the life of the process.
• Assuming enable_context_compression=True guarantees summarization quality; the compression behavior is a config flag whose downstream effect depends on backend support and isn't independently verifiable from this file alone.

Related APIs

• Multi-Hop RAG — for reasoning across multiple retrieval steps within a single query rather than across turns.
• Streaming RAG — lower-level streaming primitive that ConversationalRAG.astream builds on conceptually.