Context Modular

from fennec_community.context import ContextEngine, ContextEngineConfig

# Simplest usage — all defaults applied
engine = ContextEngine()
engine.index(chunks=my_chunks, embed_fn=my_embed_function)

result = engine.run("What is machine learning?")
print(result.context)       # Context ready to feed into LLM
print(result.get_stats())   # Detailed pipeline statistics

# Advanced usage — full customization
from fennec_community.context import (
    ContextEngine, ContextEngineConfig,
    ComposerConfig, RetrieverConfig, CacheConfig
)

config = ContextEngineConfig(
    composer  = ComposerConfig(max_context_length=4000, template="english"),
    retriever = RetrieverConfig(top_k=30, vector_weight=0.7),
    cache     = CacheConfig(enabled=True, ttl_seconds=600),
)

engine = ContextEngine(config=config)
engine.index(chunks=my_chunks, embed_fn=my_embed_fn)

result = engine.run(
    query            = "Compare BERT and GPT architectures",
    metadata_filters = {"category": "nlp"},
    max_chunks       = 10,
)

ContextEngine(
    config:           Optional[ContextEngineConfig]      = None,
    query_analyzer:   Optional[QueryAnalyzerStrategy]    = None,
    retriever:        Optional[HybridRetriever]          = None,
    chunk_filter:     Optional[ChunkFilterStrategy]      = None,
    ranker:           Optional[ChunkRankingStrategy]     = None,
    composer:         Optional[ContextComposerStrategy]  = None,
    guard:            Optional[ContextGuardStrategy]     = None,
    budget_allocator: Optional[BudgetAllocationStrategy] = None,
    length_counter:   Optional[LengthCounter]            = None,
)

engine.index(
    chunks:               List[Any],
    embed_fn:             Optional[Callable[[str], np.ndarray]] = None,
    vector_store_adapter: Optional[Any]                          = None,
    source_quality_map:   Optional[Dict[str, SourceQuality]]    = None,
) -> ContextEngine

# Method chaining example
result = (
    ContextEngine(config=my_config)
    .index(chunks=docs, embed_fn=embedder)
    .run("How does the BM25 algorithm work?")
)

engine.run(
    query:            str,
    metadata_filters: Optional[Dict[str, Any]]  = None,
    max_chunks:       Optional[int]              = None,
    override_budget:  Optional[int]              = None,
    use_history:      bool                       = True,
    record_turn:      bool                       = True,
) -> ContextResult

result = engine.run(
    query            = "How do LSTM networks work?",
    metadata_filters = {"domain": "deep_learning"},
    override_budget  = 3000,
)

# Use the output
llm_prompt = f"Context:\n{result.context}\n\nQuestion: {result.query_analysis.original_query}"

# Inspect pipeline statistics
stats = result.get_stats()
print(f"Retrieved  : {stats['total_retrieved']} chunks")
print(f"Included   : {stats['total_included']} chunks")
print(f"Latency    : {stats['pipeline_ms']} ms")
print(f"Guard OK   : {stats['guard_passed']}")

engine.run_multi(
    queries:       List[str],
    global_budget: Optional[int] = None,
    use_weighted:  bool          = True,
    **kwargs,
) -> str

combined_context = engine.run_multi(
    queries = [
        "What is the Transformer architecture?",
        "How does the Attention Mechanism work?",
        "What is the difference between BERT and GPT?",
    ],
    global_budget = 6000,
    use_weighted  = True,   # More complex queries receive larger budget slices
)

engine.build(
    query:      str,
    chunks:     List[Tuple],   # [(chunk, score), ...]
    max_chunks: Optional[int] = None,
    max_length: Optional[int] = None,
) -> str

engine.compress_context(context: str, target_length: int) -> str

engine.get_context_stats(context: str) -> Dict[str, Any]

engine.get_performance_report() -> Dict[str, Any]

engine.set_conversation(history: ConversationHistory) -> None

engine.get_history() -> Optional[ConversationHistory]

engine.clear_cache() -> None

@dataclass
class ContextEngineConfig:
    query_analyzer:  QueryAnalyzerConfig = QueryAnalyzerConfig()
    retriever:       RetrieverConfig     = RetrieverConfig()
    filter_cfg:      FilterConfig        = FilterConfig()
    ranking:         RankingConfig       = RankingConfig()
    composer:        ComposerConfig      = ComposerConfig()
    guard:           GuardConfig         = GuardConfig()
    cache:           CacheConfig         = CacheConfig()
    language:        str  = "auto"       # "auto" | "ar" | "en"
    enable_adaptive: bool = True         # Enable adaptive learning
    multi_turn:      bool = True         # Enable multi-turn conversation support

@dataclass
class ContextResult:
    context:            str                  # Context string ready for LLM
    query_analysis:     QueryAnalysis        # Full query analysis result
    chunks_used:        List[ScoredChunk]    # Chunks actually included in context
    total_retrieved:    int                  # Chunks retrieved before filtering
    total_after_filter: int                  # Chunks remaining after filtering
    total_after_rank:   int                  # Chunks remaining after ranking
    total_included:     int                  # Chunks included in the final context
    context_length:     int                  # Context length in characters
    pipeline_ms:        float                # Total end-to-end latency
    guard_passed:       bool                 # Whether ContextGuard validation passed
    guard_warnings:     List[str]            # Guard warning messages (may be empty)
    metadata:           Dict[str, Any]       # Extra metadata (e.g. {"from_cache": True})

result.get_stats() -> Dict[str, Any]

@dataclass
class QueryAnalysis:
    original_query:  str
    normalized:      str             # Cleaned text (diacritics removed, whitespace normalised)
    intent:          QueryIntent     # FACTUAL | REASONING | CODE | COMPARATIVE | ...
    complexity:      QueryComplexity # LOW | MEDIUM | HIGH
    keywords:        List[str]       # Extracted keywords, ranked by relevance
    entities:        List[str]       # Named entities detected in the query
    is_arabic:       bool
    language:        str             # "ar" | "en"
    question_type:   str             # "wh-question" | "yes/no" | "open"
    context_budget:  int             # Recommended context size in characters

analysis.to_dict() -> Dict[str, Any]

@dataclass
class ScoredChunk:
    chunk:            Any            # Original chunk object (any type)
    vector_score:     float          # Cosine similarity from vector search
    keyword_score:    float          # BM25 score (normalised 0-1)
    metadata_score:   float          # Metadata filter match score
    source_quality:   SourceQuality  # HIGH | MEDIUM | LOW | UNKNOWN
    recency_score:    float          # Recency signal (0-1)
    composite_score:  float          # Final weighted ranking score
    retrieval_method: RetrievalMethod
    retrieval_rank:   int            # Original rank before re-ranking

chunk.to_dict() -> Dict[str, Any]

retriever.retrieve(
    query:   str,
    top_k:   Optional[int]           = None,
    filters: Optional[Dict[str, Any]] = None,
) -> List[ScoredChunk]

rrf_score(chunk) = Σ  1 / (k + rank(chunk, list_i))     where k = 60

vector_retriever.index_chunks(
    chunks:   List[Any],
    embed_fn: Callable[[str], np.ndarray],
) -> None

keyword_retriever.index_chunks(chunks: List[Any]) -> None

chunk_filter.filter(chunks: List[ScoredChunk]) -> List[ScoredChunk]

chunk_filter.filter_with_embeddings(
    chunks:     List[ScoredChunk],
    embeddings: List[Optional[np.ndarray]],
) -> List[ScoredChunk]

ranker.rank(
    chunks:   List[ScoredChunk],
    analysis: QueryAnalysis,
    max_n:    Optional[int] = None,
) -> List[ScoredChunk]

composer.compose(
    chunks:   List[ScoredChunk],
    analysis: QueryAnalysis,
    budget:   int,
) -> str

composer.get_stats(context: str) -> Dict[str, Any]

guard.validate(
    context:  str,
    query:    str,
    analysis: QueryAnalysis,
) -> Tuple[bool, List[str]]

# Standalone usage example
guard = ContextGuard(config)
passed, warnings = guard.validate(context, query, analysis)

if not passed:
    for w in warnings:
        logger.warning("ContextGuard: %s", w)
    # fallback: retrieve more chunks or reduce strictness

learning.suggest_budget(analysis: QueryAnalysis) -> int

learning.get_performance_report() -> Dict[str, Any]

cache.get(key: str) -> Optional[str]

cache.put(key: str, context: str) -> None

cache.make_key(
    query:   str,
    filters: Optional[Dict] = None,
) -> str

cache.stats -> Dict[str, Any]

print(cache.stats)
# {
#   'size': 45, 'max_size': 256,
#   'hits': 120, 'misses': 30,
#   'hit_rate': '80.0%', 'ttl_sec': 300
# }

cache.clear() -> None

EqualBudgetAllocation().allocate(
    total_budget: int,
    n_queries:    int,
) -> List[int]

WeightedBudgetAllocation(weights=[0.5, 0.3, 0.2]).allocate(
    total_budget: int,
    n_queries:    int,
) -> List[int]

ComplexityBasedAllocation().allocate(
    total_budget: int,
    n_queries:    int,
    analyses:     Optional[List[QueryAnalysis]] = None,
) -> List[int]

history.add_turn(
    query:    str,
    answer:   str        = "",
    context:  str        = "",
    analysis: Optional[QueryAnalysis] = None,
) -> ConversationTurn

history.get_recent_context(n: int = 3) -> str

Q: First question
A: First answer

Q: Second question
A: Second answer

history.get_enriched_query(
    current_query: str,
    n:             int = 2,
) -> str

# Turn 1
engine.run("What is the Transformer model?")

# Turn 2 — "it" is ambiguous without history
engine.run("How does it handle long sequences?")
# Internally enriched to:
# "[Previous conversation:
#   Q: What is the Transformer model?
#   A: ...]
#  Current question: How does it handle long sequences?"

from fennec_community.context import char_counter

char_counter(text: str) -> int

from fennec_community.context import make_token_counter

token_counter = make_token_counter(tokenizer)

from transformers import AutoTokenizer
from fennec_community.context import make_token_counter, ContextEngine

tokenizer     = AutoTokenizer.from_pretrained("bert-base-uncased")
token_counter = make_token_counter(tokenizer)

engine = ContextEngine(length_counter=token_counter)
# All budget values now interpreted as token counts

from fennec_community.context import build_query_analyzer

analyzer = build_query_analyzer(
    config:   ContextEngineConfig,
    strategy: Optional[QueryAnalyzerStrategy] = None,
) -> QueryAnalyzerStrategy

from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Optional, Dict
import numpy as np
from fennec_community.context import (
    ContextEngine,
    ContextEngineConfig,
    ContextManager,
    QueryAnalyzerConfig,
    RetrieverConfig,
    FilterConfig,
    RankingConfig,
    ComposerConfig,
    GuardConfig,
    CacheConfig,
    SourceQuality,
    ScoredChunk,
    char_counter,
)
from fennec_community.chunks import DocumentChunk

_WORD_EMBEDDINGS: Dict[str, np.ndarray] = {}
def mock_embed(text: str) -> np.ndarray:
    """
    simulations for embedding model for test
    """
    import re
    words = re.findall(r"\w+", text.lower())
    vec   = np.zeros(64)
    for w in words:
        if w not in _WORD_EMBEDDINGS:
            np.random.seed(hash(w) % (2**31))
            _WORD_EMBEDDINGS[w] = np.random.randn(64)
        vec += _WORD_EMBEDDINGS[w]
    norm = np.linalg.norm(vec)
    return (vec / norm) if norm > 0 else vec


KNOWLEDGE_BASE = [
    DocumentChunk(
        doc_id="ml_001",
        text=(
            "التعلم الآلي هو فرع من فروع الذكاء الاصطناعي يركز على بناء أنظمة تتعلم من البيانات. "
            "تستخدم هذه الأنظمة الخوارزميات لتحليل البيانات وتعلم الأنماط واتخاذ القرارات. "
            "من أبرز تطبيقاته: التعرف على الصور، ومعالجة اللغة الطبيعية، وأنظمة التوصية."
        ),
        metadata={"category": "ml", "lang": "ar", "date": "2024","source":"كتاب التعلم الآلي" },
    ),
    DocumentChunk(
        doc_id="dl_002",
        text=(
            "التعلم العميق هو مجال فرعي من التعلم الآلي يعتمد على الشبكات العصبية الاصطناعية "
            "ذات الطبقات المتعددة. تستطيع هذه الشبكات تعلم تمثيلات هرمية للبيانات. "
            "تُستخدم في رؤية الكمبيوتر، ومعالجة الكلام، والنصوص."
        ),
        metadata={"category": "dl", "lang": "ar", "date": "2024","source":"كتاب التعلم العميق" },
    ),
    DocumentChunk(
        doc_id="nlp_003",
        text=(
            "معالجة اللغة الطبيعية (NLP) هي مجال في الذكاء الاصطناعي يُعنى بفهم اللغة البشرية. "
            "يشمل مهام مثل: تصنيف النصوص، واستخراج المعلومات، والترجمة الآلية، "
            "وتوليد النصوص. نماذج مثل BERT وGPT غيّرت هذا المجال جذرياً."
        ),
        metadata={"category": "nlp", "lang": "ar", "date": "2024","source":"كتاب معالجة اللغة الطبيعية" },
    ),
    DocumentChunk(
        doc_id="rag_004",
        text=(
            "Retrieval-Augmented Generation (RAG) هي تقنية تجمع بين قواعد المعرفة الخارجية "
            "ونماذج اللغة الكبيرة. تُحسِّن دقة الإجابات وتقلل الـ hallucination. "
            "تعتمد على: استرجاع المعلومات ذات الصلة، وتضمينها في سياق الـ LLM."
        ),
        metadata={"category": "rag", "lang": "ar", "date": "2024"},
    ),
    DocumentChunk(
        doc_id="transformer_005",
        text=(
            "معمارية Transformer ثورت مجال معالجة اللغة الطبيعية منذ ورقة Attention is All You Need. "
            "تعتمد على آلية self-attention التي تُحدد العلاقات بين الكلمات في النص. "
            "أصبحت أساساً لنماذج مثل BERT وGPT وT5."
        ),
        metadata={"category": "transformer", "lang": "ar", "date": "2017"},
    ),
    DocumentChunk(
        doc_id="vector_006",
        text=(
            "قواعد البيانات المتجهية (Vector Databases) تُخزّن embeddings وتتيح البحث بالتشابه الدلالي. "
            "من أبرزها: Pinecone, Chroma, Weaviate, FAISS. "
            "تُستخدم في أنظمة RAG لاسترجاع المعلومات ذات الصلة بالاستعلام."
        ),
        metadata={"category": "infrastructure", "lang": "ar", "date": "2024"},
    ),
    DocumentChunk(
        doc_id="duplicate_007",  # chunk مكرر عمداً للاختبار
        text=(
            "التعلم الآلي هو فرع من فروع الذكاء الاصطناعي يركز على بناء أنظمة تتعلم من البيانات. "
            "تستخدم هذه الأنظمة الخوارزميات لتحليل البيانات وتعلم الأنماط واتخاذ القرارات."
        ),
        metadata={"category": "ml", "lang": "ar"},
    ),

    DocumentChunk(
        doc_id="llm_009",
        text=(
            "نماذج اللغة الكبيرة (LLMs) مثل GPT-4 وClaude وGemini تُنتج نصاً يشبه الكتابة البشرية. "
            "تُدرَّب على مليارات الكلمات وتستطيع الإجابة على الأسئلة، والكتابة، والبرمجة. "
            "التحديات الرئيسية تشمل: الـ hallucination، والتحيز، والتكلفة الحسابية."
        ),
        metadata={"category": "llm", "lang": "ar", "date": "2024"},
    ),
    DocumentChunk(
        doc_id="fine_tuning_010",
        text=(
            "Fine-tuning هي عملية تدريب نموذج مُدرَّب مسبقاً على بيانات خاصة بمجال معين. "
            "تُحسِّن الأداء في مهام محددة دون الحاجة إلى التدريب من الصفر. "
            "LoRA وQLoRA هما تقنيتان شائعتان لـ fine-tuning فعّال من حيث الموارد."
        ),
        metadata={"category": "training", "lang": "ar", "date": "2024"},
    ),
]

def example_basic():
    print("\n" + "="*60)
    print("📌 example 1 : basic using")
    print("="*60)

    engine = ContextEngine()  # buila a context engine with default config


    # initializing index cuhnks for retrieval 
    engine.index(chunks=KNOWLEDGE_BASE, embed_fn=mock_embed)

    # run the engine with a query
    result = engine.run("ما هو التعلم الآلي؟")

    print(f"\n🔍 الاستعلام: ما هو التعلم الآلي؟")
    print(f"📊 النية: {result.query_analysis.intent.value}")
    print(f"📊 التعقيد: {result.query_analysis.complexity.value}")
    print(f"🔑 الكلمات المفتاحية: {result.query_analysis.keywords[:5]}")
    print(f"\n📄 السياق المُركَّب:\n{result.context}")
    print(f"\n📈 الإحصائيات:")
    stats = result.get_stats()
    for k, v in stats.items():
        print(f"   {k}: {v}")

def example_custom_config():
    print("\n" + "="*60)
    print("📌 example 2:custom config using")
    print("="*60)

    config = ContextEngineConfig(
        query_analyzer = QueryAnalyzerConfig(
            budget_low    = 600,
            budget_medium = 1500,
            budget_high   = 3500,
        ),
        retriever = RetrieverConfig(
            top_k          = 15,
            vector_weight  = 0.65,
            keyword_weight = 0.30,
            metadata_weight= 0.05,
        ),
        filter_cfg = FilterConfig(
            dedup_method   = "hash",
            min_chunk_length = 30,
        ),
        ranking = RankingConfig(
            weight_vector_score   = 0.45,
            weight_keyword_score  = 0.30,
            weight_source_quality = 0.15,
            weight_recency        = 0.10,
            max_chunks_to_rank    = 5,
        ),
        composer = ComposerConfig(
            max_context_length = 2500,
            template           = "arabic",
            include_scores     = True,
            include_metadata   = False,
            group_by_source    = False,
        ),
        guard = GuardConfig(
            enabled              = True,
            min_context_length   = 50,
            check_source_coverage= True,
        ),
        cache = CacheConfig(
            enabled     = True,
            max_size    = 128,
            ttl_seconds = 600,
        ),
    )

    engine = ContextEngine(config=config)

    # specific for source quality
    source_quality = {
        "كتاب التعلم الآلي":   SourceQuality.HIGH,
        "كتاب التعلم العميق":  SourceQuality.HIGH,
        "مجلة NLP":            SourceQuality.MEDIUM,
        "بحث RAG":             SourceQuality.HIGH,
        "مصدر مكرر":           SourceQuality.LOW,
    }

    engine.index(
        chunks             = KNOWLEDGE_BASE,
        embed_fn           = mock_embed,
        source_quality_map = source_quality,
    )

    
    result = engine.run(
        "قارن بين التعلم الآلي والتعلم العميق، وما هي أبرز الفروقات؟",
        metadata_filters={"lang": "ar"},
    )

    print(f"\n🔍 الاستعلام: قارن بين التعلم الآلي والتعلم العميق")
    print(f"📊 النية: {result.query_analysis.intent.value}")
    print(f"📊 التعقيد: {result.query_analysis.complexity.value}")
    print(f"💰 الميزانية: {result.query_analysis.context_budget} حرف")
    print(f"🛡️ Guard: {'✓ نجح' if result.guard_passed else '✗ فشل'}")
    print(f"\n📄 السياق:\n{result.context}")

def example_multi_turn():
    print("\n" + "="*60)
    print("📌 example 3 :multi conversion turn using")
    print("="*60)

    engine = ContextEngine(
        config=ContextEngineConfig(multi_turn=True)
    )
    engine.index(chunks=KNOWLEDGE_BASE, embed_fn=mock_embed)

    conversation = [
        "ما هو التعلم الآلي؟",
        "وما علاقته بالتعلم العميق؟",
        "كيف يُستخدم في RAG؟",
    ]

    for i, query in enumerate(conversation):
        result = engine.run(query, use_history=True, record_turn=True)

        # simulations for llm response
        fake_answer = f"[إجابة LLM للسؤال {i+1}]"

        # record the answer in conversion history
        if engine.get_history() and engine.get_history().turns:
            engine.get_history().turns[-1].answer = fake_answer

        print(f"\n💬 السؤال {i+1}: {query}")
        print(f"   📊 الاستعلام المُثرى: {result.query_analysis.normalized[:80]}...")
        print(f"   📄 طول السياق: {result.context_length} حرف")
        print(f"   ⏱️ الوقت: {result.pipeline_ms:.1f}ms")

    # statistics for the conversation history
    history = engine.get_history()
    print(f"\n📚 تاريخ المحادثة: {len(history.turns)} دورة")

def example_cache_performance():
    print("\n" + "="*60)
    print("📌 example 4 : using cache performance")
    print("="*60)

    engine = ContextEngine(
        config=ContextEngineConfig(
            cache=CacheConfig(enabled=True, max_size=50, ttl_seconds=60)
        )
    )
    engine.index(chunks=KNOWLEDGE_BASE, embed_fn=mock_embed)

    query = "ما هو التعلم الآلي؟"

    # first query (cache miss)
    t0 = time.perf_counter()
    r1 = engine.run(query)
    t1 = (time.perf_counter() - t0) * 1000

    # same query again (cache hit)
    t0 = time.perf_counter()
    r2 = engine.run(query)
    t2 = (time.perf_counter() - t0) * 1000

    print(f"\n📊 الاستعلام الأول (cache miss): {t1:.1f}ms")
    print(f"📊 الاستعلام الثاني (cache hit):  {t2:.1f}ms")
    print(f"⚡ تسريع: {t1/t2:.1f}x" if t2 > 0 else "")

    # statistics for the cache
    perf = engine.get_performance_report()
    print(f"\n📈 تقرير الأداء:")
    for k, v in perf.items():
        if isinstance(v, dict):
            print(f"   {k}:")
            for kk, vv in v.items():
                print(f"      {kk}: {vv}")
        else:
            print(f"   {k}: {v}")

composite_score =
    vector_score   × 0.40    (semantic similarity)
  + keyword_boost  × 0.25    (query keyword matches)
  + source_quality × 0.20    (source trustworthiness)
  + recency_score  × 0.10    (information freshness)
  + position_score × 0.05    (chunk position in document)

 1. Cache Lookup          Is this query already cached?
 2. History Enrichment    Prepend prior conversation context to query
 3. Query Analysis        Detect intent, complexity, keywords, budget
 4. Budget Suggestion     Adaptive learning adjusts budget from history
 5. Retrieval             Vector + BM25 + Metadata  →  RRF Fusion
 6. Filtering             Noise removal + length filter + deduplication
 7. Ranking               Composite score re-ranking
 8. Composition           Format chunks → truncate to budget
 9. Guard Validation      Empty? Too short? Source present? Relevant?
10. Cache Store           Store result for future identical queries
11. History Record        Save this turn to conversation history
12. Adaptive Record       Log metrics for continuous self-tuning