Streaming RAG — `streaming_rag` Module — Public API Reference

from fennec_community.rag.types.streaming_rag import (
    StreamingRAG,
    StreamConfig,
    SyncStreamableLLM,
    AsyncStreamableLLM,
    EventType,
)

from fennec_community.rag.types.streaming_rag import StreamConfig

StreamConfig(
    chunk_size: int = 10,
    sim_delay: float = 0.04,
    max_context_docs: int = 5,
    max_doc_chars: int = 500,
    emit_metadata: bool = True,
    retrieval_timeout: float = 10.0,
    generation_timeout: float = 60.0,
    prompt_template: str = <built-in>,
)

You are a precise RAG assistant.

Strict rules:
- Use ONLY the provided context.
- Do NOT guess.
- If missing info, say: 'no information to answer '.
- The answer MUST be in the SAME language as the question.
- NEVER switch language.

Context:
{context}

Question: {query}

Final Answer:

from fennec_community.rag.types.streaming_rag import StreamConfig

config = StreamConfig(
    chunk_size=5,
    max_context_docs=8,
    max_doc_chars=800,
    retrieval_timeout=15.0,
    generation_timeout=90.0,
    prompt_template=(
        "Context:\n{context}\n\n"
        "Question: {query}\n\n"
        "Answer concisely:"
    ),
)

from fennec_community.rag.types.streaming_rag import StreamingRAG

StreamingRAG(
    rag_system: Any,
    llm: Any = None,
    config: Optional[StreamConfig] = None,
    *,
    chunk_size: Optional[int] = None,
    on_chunk: Optional[Callable[[str], None]] = None,
    on_event: Optional[Callable[[StreamEvent], None]] = None,
)

from fennec_community.rag.types.streaming_rag import StreamingRAG, StreamConfig

def save_chunk(chunk: str):
    db.append_token(chunk)

config = StreamConfig(max_context_docs=6, retrieval_timeout=8.0)

rag = StreamingRAG(
    rag_system=my_rag,
    llm=my_llm,
    config=config,
    on_chunk=save_chunk,
)

rag.stream(
    query: str,
    context: Optional[Dict] = None,
) -> Iterator[str]

print("Answer: ", end="")
for chunk in rag.stream("What is retrieval-augmented generation?"):
    print(chunk, end="", flush=True)
print()  # newline after stream

rag.stream_events(
    query: str,
    context: Optional[Dict] = None,
) -> Iterator[StreamEvent]

for event in rag.stream_events("Explain neural networks"):
    if event.is_chunk:
        print(event.data, end="", flush=True)
    elif event.type == EventType.RETRIEVAL_DONE:
        print(f"\n[Retrieved {event.data['num_docs']} docs in {event.latency:.2f}s]")
    elif event.type == EventType.ERROR:
        print(f"\n[Error at {event.data['stage']}: {event.data['msg']}]")
        break

rag.query(
    query: str,
    context: Optional[Dict] = None,
) -> Dict[str, Any]

result = rag.query("What are the main benefits of RAG?")
print(result["answer"])

# Inspect the event trace
for ev in result["events"]:
    if ev.type == EventType.GENERATION_DONE:
        print(f"Total latency: {ev.data['total_latency']:.2f}s")

async for chunk in rag.astream(
    query: str,
    context: Optional[Dict] = None,
):
    print(chunk, end="", flush=True)

import asyncio

async def main():
    print("Answer: ", end="")
    async for chunk in rag.astream("How does vector similarity search work?"):
        print(chunk, end="", flush=True)
    print()

asyncio.run(main())

async for event in rag.astream_events(
    query: str,
    context: Optional[Dict] = None,
) -> AsyncIterator[StreamEvent]:

import asyncio
from fennec_community.rag.types.streaming_rag import EventType

async def rich_consumer():
    async for event in rag.astream_events("Describe transformer architecture"):
        match event.type:
            case EventType.RETRIEVAL_START:
                print(f"🔍 Retrieving... (t={event.latency:.3f}s)")
            case EventType.RETRIEVAL_DONE:
                print(f"📄 Got {event.data['num_docs']} docs (t={event.latency:.3f}s)")
            case EventType.GENERATION_START:
                print(f"✍️  Generating {event.data['prompt_chars']} char prompt...")
            case EventType.CHUNK:
                print(event.data, end="", flush=True)
            case EventType.GENERATION_DONE:
                print(f"\n✅ Done in {event.data['total_latency']:.2f}s")
            case EventType.ERROR:
                print(f"\n❌ Error at [{event.data['stage']}]: {event.data['msg']}")

asyncio.run(rich_consumer())

rag.stats -> Dict[str, Any]

import json

# After several queries:
print(json.dumps(rag.stats, indent=2))
# {
#   "total_streams": 12,
#   "total_chunks": 847,
#   "total_errors": 1,
#   "avg_stream_latency_s": 2.341
# }

repr(rag)

print(repr(rag))
# StreamingRAG(llm=async, chunk_size=10)

from fennec_community.rag.types.streaming_rag import StreamEvent

event.is_chunk -> bool

for event in rag.stream_events("What is RAG?"):
    if event.is_chunk:
        print(event.data, end="")

str(event)  # → str

# Ultra-compact consumer — works because str(non-chunk event) == ""
answer = "".join(str(ev) for ev in rag.stream_events("Summarise the context"))
print(answer)

from fennec_community.rag.types.streaming_rag import EventType

async for event in rag.astream_events(query):
    match event.type:
        case EventType.RETRIEVAL_DONE:
            print(f"Found {event.data['num_docs']} documents")
        case EventType.CHUNK:
            print(event.data, end="")
        case EventType.ERROR:
            print(f"Error: {event.data['msg']}")

from fennec_community.rag.types.streaming_rag import SyncStreamableLLM, AsyncStreamableLLM

class SyncStreamableLLM(Protocol):
    def stream(self, prompt: str) -> Iterator[str]: ...

from fennec_community.rag.types.streaming_rag import SyncStreamableLLM
from typing import Iterator

class MyOpenAILLM:  # implicitly satisfies SyncStreamableLLM
    def stream(self, prompt: str) -> Iterator[str]:
        for chunk in openai_client.chat.completions.create(
            model="gpt-4o", messages=[{"role":"user","content":prompt}], stream=True
        ):
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

assert isinstance(MyOpenAILLM(), SyncStreamableLLM)  # → True

class AsyncStreamableLLM(Protocol):
    async def astream(self, prompt: str) -> AsyncIterator[str]: ...

from fennec_community.rag.types.streaming_rag import AsyncStreamableLLM
from typing import AsyncIterator

class MyAsyncLLM:  # implicitly satisfies AsyncStreamableLLM
    async def astream(self, prompt: str) -> AsyncIterator[str]:
        async for chunk in await async_openai_client.chat.completions.create(
            model="gpt-4o", messages=[{"role":"user","content":prompt}], stream=True
        ):
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

assert isinstance(MyAsyncLLM(), AsyncStreamableLLM)  # → True

┌─────────────────────────────────────────────────────────────────┐
│              Generation Strategy Decision Tree                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  llm is not None                                                │
│  AND isinstance(llm, AsyncStreamableLLM)?                       │
│         │                                                       │
│         ├── YES → Strategy 1: Native async stream               │
│         │         Calls llm.astream(prompt)                     │
│         │         ✅ Best latency, true token-level streaming   │
│         │                                                       │
│         └── NO ──→ isinstance(llm, SyncStreamableLLM)?          │
│                          │                                      │
│                          ├── YES → Strategy 2: Sync stream      │
│                          │         Runs llm.stream(prompt)      │
│                          │         in a thread via executor     │
│                          │         Bridges sync → async safely  │
│                          │         ✅ Token-level, thread-safe  │
│                          │                                      │
│                          └── NO ──→ Strategy 3: Simulation      │
│                                    Calls rag_system.generate()  │
│                                    Splits answer into word      │
│                                    chunks of size=chunk_size    │
│                                    with sim_delay between each  │
│                                    ⚠️ Pseudo-streaming only     │
└─────────────────────────────────────────────────────────────────┘

Time →
│
├─ [t=0.0000] RETRIEVAL_START  {"query": "..."}
│
│  [retrieval executes — up to retrieval_timeout seconds]
│
├─ [t=0.Xsec]  RETRIEVAL_DONE  {"num_docs": N, "docs": [...]}
│
│  [context built, prompt formatted]
│
├─ [t=0.Xsec]  GENERATION_START {"prompt_chars": N}
│
│  [LLM streaming begins]
│
├─ [t=0.Xsec]  CHUNK  "First few words "
├─ [t=0.Xsec]  CHUNK  "of the answer "
├─ [t=0.Xsec]  CHUNK  "..."
│   ...
├─ [t=0.Xsec]  CHUNK  "last chunk."
│
└─ [t=0.Xsec]  GENERATION_DONE {"total_latency": X.XXXX}


─── On retrieval error or timeout: ───────────────────────────────
│
├─ RETRIEVAL_START
└─ ERROR  {"msg": "Retrieval timed out", "stage": "retrieval"}
   [stream ends — no further events]


─── On generation error or timeout: ──────────────────────────────
│
├─ RETRIEVAL_START
├─ RETRIEVAL_DONE
├─ GENERATION_START
├─ CHUNK  (zero or more chunks may have been emitted before error)
└─ ERROR  {"msg": "Generation timed out", "stage": "generation"}
   [stream ends — no further events]

import asyncio
import json
from fennec_community.rag.types.streaming_rag import StreamingRAG, StreamConfig, EventType
from fennec_community.rag.types.streaming_rag import SyncStreamableLLM, AsyncStreamableLLM

# ═══════════════════════════════════════════════════════════════════
# 1. Implement the AsyncStreamableLLM protocol (Strategy 1)
# ═══════════════════════════════════════════════════════════════════
class MyAsyncLLM:
    async def astream(self, prompt: str):
        # Replace with a real async LLM call (OpenAI, Anthropic, etc.)
        words = f"Answer based on context: {prompt[:30]}...".split()
        for word in words:
            yield word + " "
            await asyncio.sleep(0.02)

# ═══════════════════════════════════════════════════════════════════
# 2. Configure the streamer
# ═══════════════════════════════════════════════════════════════════
config = StreamConfig(
    max_context_docs=5,
    max_doc_chars=600,
    retrieval_timeout=10.0,
    generation_timeout=60.0,
)

# ═══════════════════════════════════════════════════════════════════
# 3. Callbacks (optional)
# ═══════════════════════════════════════════════════════════════════
log = []

def on_chunk(chunk: str):
    log.append(chunk)          # side-effect: accumulate tokens elsewhere

def on_event(event):
    if event.type == EventType.ERROR:
        print(f"[CALLBACK] ❌ Error: {event.data}")

# ═══════════════════════════════════════════════════════════════════
# 4. Instantiate
# ═══════════════════════════════════════════════════════════════════
rag = StreamingRAG(
    rag_system=my_rag_system,
    llm=MyAsyncLLM(),
    config=config,
    on_chunk=on_chunk,
    on_event=on_event,
)
print(repr(rag))
# StreamingRAG(llm=async, chunk_size=10)

# ═══════════════════════════════════════════════════════════════════
# 5. Synchronous plain text streaming
# ═══════════════════════════════════════════════════════════════════
print("Answer: ", end="")
for chunk in rag.stream("What is retrieval-augmented generation?"):
    print(chunk, end="", flush=True)
print()

# ═══════════════════════════════════════════════════════════════════
# 6. Synchronous rich event streaming
# ═══════════════════════════════════════════════════════════════════
for event in rag.stream_events("Explain transformer attention"):
    if event.type == EventType.RETRIEVAL_DONE:
        print(f"\n[Docs: {event.data['num_docs']}  Latency: {event.latency:.3f}s]")
    elif event.type == EventType.CHUNK:
        print(event.data, end="", flush=True)
    elif event.type == EventType.GENERATION_DONE:
        print(f"\n[Done in {event.data['total_latency']:.2f}s]")
    elif event.type == EventType.ERROR:
        print(f"\n[Error at {event.data['stage']}: {event.data['msg']}]")
        break

# ═══════════════════════════════════════════════════════════════════
# 7. Non-streaming (collect full answer)
# ═══════════════════════════════════════════════════════════════════
result = rag.query("Summarise the key points about BERT")
print(result["answer"])
print(f"Events captured: {len(result['events'])}")

# ═══════════════════════════════════════════════════════════════════
# 8. Async plain text streaming
# ═══════════════════════════════════════════════════════════════════
async def async_text():
    print("Async: ", end="")
    async for chunk in rag.astream("How does vector search work?"):
        print(chunk, end="", flush=True)
    print()

asyncio.run(async_text())

# ═══════════════════════════════════════════════════════════════════
# 9. Async rich event streaming
# ═══════════════════════════════════════════════════════════════════
async def async_events():
    async for event in rag.astream_events("Describe the RAG pipeline"):
        match event.type:
            case EventType.RETRIEVAL_START:
                print(f"🔍 Retrieving (t={event.latency:.3f}s)")
            case EventType.RETRIEVAL_DONE:
                print(f"📄 Retrieved {event.data['num_docs']} docs")
            case EventType.GENERATION_START:
                print(f"✍️  Generating ({event.data['prompt_chars']} chars)")
            case EventType.CHUNK:
                print(event.data, end="", flush=True)
            case EventType.GENERATION_DONE:
                print(f"\n✅ Total: {event.data['total_latency']:.2f}s")
            case EventType.ERROR:
                print(f"\n❌ [{event.data['stage']}] {event.data['msg']}")

asyncio.run(async_events())

# ═══════════════════════════════════════════════════════════════════
# 10. Runtime statistics
# ═══════════════════════════════════════════════════════════════════
print(json.dumps(rag.stats, indent=2))
# {
#   "total_streams": 4,
#   "total_chunks": 312,
#   "total_errors": 0,
#   "avg_stream_latency_s": 1.847
# }

# ═══════════════════════════════════════════════════════════════════
# 11. Protocol compliance check
# ═══════════════════════════════════════════════════════════════════
from fennec_community.rag.types.streaming_rag import AsyncStreamableLLM, SyncStreamableLLM

print(isinstance(MyAsyncLLM(), AsyncStreamableLLM))  # → True

class MySyncLLM:
    def stream(self, prompt: str):
        for word in "Hello world from sync LLM".split():
            yield word + " "

print(isinstance(MySyncLLM(), SyncStreamableLLM))  # → True

# Switch to sync LLM
rag2 = StreamingRAG(
    rag_system=my_rag_system,
    llm=MySyncLLM(),
    chunk_size=5,            # override via constructor shortcut
)
print(repr(rag2))
# StreamingRAG(llm=sync, chunk_size=5)

# ═══════════════════════════════════════════════════════════════════
# 12. Fallback (no LLM — word-level simulation)
# ═══════════════════════════════════════════════════════════════════
rag3 = StreamingRAG(
    rag_system=my_rag_system,   # must expose .generate()
    # no llm provided — Strategy 3 (simulation) activated
    config=StreamConfig(chunk_size=3, sim_delay=0.05),
)
print(repr(rag3))
# StreamingRAG(llm=sim, chunk_size=3)

for chunk in rag3.stream("Quick question for simulation mode"):
    print(chunk, end="", flush=True)
print()

from fennec_community.llm import GeminiInterface
from fennec_community.document_loaders import TextLoader 
from fennec_community.vector_database import FAISSVectorDatabase
from fennec_community.chunks import ArabicTextChunker
from fennec_community.context import ContextManager
from fennec_community.embeddings import OllamaEmbedder
from fennec_community.rag.core import RAGSystem 
from fennec_community.rag.types.streaming_rag import StreamingRAG, StreamConfig, EventType

loader_1 = TextLoader("./data_kn/faq.txt").load()
chunker = ArabicTextChunker(chunk_size=100, overlap=20)
embedder = OllamaEmbedder()
vector_db = FAISSVectorDatabase(embedder=embedder)
llm = GeminiInterface(api_key=llm_api)
context_manager = ContextManager()
rag_system = RAGSystem(llm=llm, vector_db=vector_db,chunker=chunker, context_manager=context_manager)

rag_system.add_documents(loader_1)
config = StreamConfig(
    chunk_size=3,
    sim_delay=0.05,
    retrieval_timeout=15.0,
    generation_timeout=90.0,
    max_context_docs=5,
    max_doc_chars=800,
)

def on_chunk_side_effect(token: str):
    # مثال: إرسال عبر WebSocket
    # websocket.send(token)
    pass  # ← لا طباعة هنا

def on_event_side_effect(event):
    if event.type == EventType.ERROR:
        print("خطأ في %s: %s", event.data["stage"], event.data["msg"])


rag = StreamingRAG(
    rag_system=rag_system,
    llm=llm,
    config=config,
    on_chunk=on_chunk_side_effect,   # جانب خفي فقط
    on_event=on_event_side_effect,
)
for token in rag.stream("ماهي طرق الدفع المتاحة؟"):
    print(token, end="", flush=True)   # ← مكان الطباعة الوحيد
print()