Fennec Memory Module

┌──────────────────────────────────────────────────────────────────┐
│                          WRITE PATH                              │
│                                                                  │
│  user_input + assistant_output                                   │
│        │                                                         │
│        ▼                                                         │
│  [SensitiveDataMasker]  ← optional PII redaction                │
│        │                                                         │
│        ▼                                                         │
│  [MemoryEncryptor]      ← optional Fernet encryption            │
│        │                                                         │
│        ├──────────────────────────────────────────────────────┐  │
│        ▼                                                      │  │
│  [ShortTermMemory]      ← in-memory sliding window            │  │
│        │                                                      │  │
│        ├─── background thread ──► [SemanticMemory]            │  │
│        │                          FAISS / NumPy vector index  │  │
│        │                                                      │  │
│        └─── if importance ≥ threshold ──► [LongTermMemory]    │  │
│                                           SQLite persistence  │  │
│                                                               │  │
│  [UserProfileManager]   ← updated with topics & importance   ◄─┘ │
└──────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│                          READ PATH                               │
│                                                                  │
│  query string                                                    │
│        │                                                         │
│        ├──► [SemanticMemory.search]     cosine similarity top-K │
│        ├──► [LongTermMemory.get_top_*]  importance-ranked       │
│        └──► [ShortTermMemory.get_all]   recent window           │
│                                                                  │
│  ───── merge + deduplicate ─────                                 │
│        │                                                         │
│        ▼                                                         │
│  [MemorySelector]   ← composite score: relevance × recency × importance
│        │                                                         │
│        ▼                                                         │
│  [WorkingMemory]    ← token-budgeted active context             │
│        │                                                         │
│        ▼                                                         │
│  [ContextBuilder]   ← assembles: profile + memories + docs + history
│        │                                                         │
│        ▼                                                         │
│  BuiltContext.full_text  → inject into LLM prompt               │
└──────────────────────────────────────────────────────────────────┘

from memory.core import MemoryType

@dataclass
class MemoryEntry:
    content: Any                    # The stored data (str, dict, or any serialisable type)
    timestamp: float                # Unix creation time
    memory_type: MemoryType
    importance: float = 0.5         # 0.0–1.0; controls promotion, ordering, eviction
    access_count: int = 0           # Incremented on every read
    tags: List[str] = []
    metadata: Dict[str, Any] = {}
    embedding: Optional[List[float]] = None  # Set by SemanticMemory
    last_access: float = time.time()
    decay_factor: float = 1.0       # Multiplied into importance on decay cycles
    original_importance: Optional[float] = None  # Preserved from creation for audit/reset

entry.access()                          # Increments access_count, updates last_access,
                                        # and applies a small importance boost ∝ (1 − importance).
entry.apply_decay(decay_rate: float = 0.1)   # Manually apply one decay cycle in place.
entry.add_tag(tag: str)                 # Append a tag if not already present.
entry.has_tag(tag: str) -> bool         # Check for tag membership.
entry.to_dict(include_embedding: bool = False) -> dict  # Serialise to plain dict.
entry.to_json() -> str                  # Serialise to JSON string.

composite = w_relevance × cosine_sim
          + w_recency   × exp(−λ × hours_old)
          + w_importance × entry.effective_importance

pip install fennec-memory
pip install sentence-transformers  # embeddings
pip install faiss-cpu              # fast vector search (optional, falls back to NumPy)
pip install cryptography           # Fernet encryption (optional)

from fennec_memory.memory import AIMemoryManager, MemoryConfig

# 1. Configure
config = MemoryConfig(
    persistence_path="./my_memory",
    importance_threshold=0.6,
    max_tokens=4000,
)

# 2. Initialise
mgr = AIMemoryManager(config=config)

# 3. Save an interaction
mgr.save_interaction(
    user_id="alice",
    tenant_id="acme",
    user_input="How do Python decorators work?",
    assistant_output="Decorators are higher-order functions that wrap another function.",
    importance=0.8,
    topics=["python"],
)

# 4. Build context for the next query
ctx = mgr.build_context(
    user_id="alice",
    tenant_id="acme",
    query="Show me a decorator example.",
)

# 5. Inject into your LLM
prompt = ctx.full_text + "\n\nUser: Show me a decorator example.\nAssistant:"

def get_response(mgr, user_id, tenant_id, query):
    # Build memory-enriched context
    ctx = mgr.build_context(user_id, tenant_id, query)
    
    # Call your LLM
    response = my_llm.generate(ctx.full_text + "\n\n" + query)
    
    # Store the interaction back
    mgr.save_interaction(
        user_id=user_id,
        tenant_id=tenant_id,
        user_input=query,
        assistant_output=response,
        importance=0.7,
    )
    return response

from fennec_memory.memory import AIMemoryManager

AIMemoryManager(
    config: Optional[MemoryConfig] = None,
    llm: Optional[LLMProtocol] = None,
    *,
    encryption_key: Optional[str] = None,
    enable_privacy_masking: bool = True,
    max_workers: int = 4,
)

def save_interaction(
    self,
    user_id: str,
    tenant_id: str,
    user_input: str,
    assistant_output: str,
    *,
    importance: float = 0.5,
    tags: Optional[List[str]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    topics: Optional[List[str]] = None,
) -> MemoryEntry

entry = mgr.save_interaction(
    user_id="alice",
    tenant_id="acme",
    user_input="What is RAG?",
    assistant_output="RAG is Retrieval-Augmented Generation ...",
    importance=0.9,
    tags=["AI", "RAG"],
    topics=["AI", "LLM"],
    metadata={"session_id": "session_001"},
)
print(entry.id)  # e.g. "3f2a1b9c7e4d5a6f"

def build_context(
    self,
    user_id: str,
    tenant_id: str,
    query: str,
    *,
    documents: Optional[List[Document]] = None,
    top_k_semantic: int = 8,
    top_k_ltm: int = 10,
    include_profile: bool = True,
) -> BuiltContext

from fennec_memory.memory import Document

ctx = mgr.build_context(
    user_id="alice",
    tenant_id="acme",
    query="How do I implement a caching layer?",
    documents=[
        Document(page_content="Redis supports LRU eviction ...", metadata={"source":"redis_docs"}, doc_id="doc_001")
    ],
    top_k_semantic=5,
)

# Use ctx.full_text as the LLM context block
print(ctx.full_text)
print(f"Memories used: {ctx.memory_count}, truncated: {ctx.truncated}")

def build_prompt(
    self,
    user_id: str,
    tenant_id: str,
    query: str,
    *,
    documents: Optional[List[Document]] = None,
    system_instruction: str = "You are a helpful, knowledgeable assistant.",
) -> str

prompt = mgr.build_prompt(
    user_id="alice",
    tenant_id="acme",
    query="Explain async/await in Python.",
    system_instruction="You are an expert Python engineer.",
)
response = my_llm.generate(prompt)

def run_maintenance(self, tenant_id: str) -> Dict[str, Any]

import asyncio

async def maintenance_loop(mgr, tenant_id, interval_seconds=3600):
    while True:
        report = await mgr.arun_maintenance(tenant_id)
        print(f"Maintenance: {report}")
        await asyncio.sleep(interval_seconds)

mgr.grant(user_id: str, tenant_id: str, role: str = "writer") -> None
mgr.revoke(user_id: str, tenant_id: str, role: str) -> None

mgr.grant("alice", "acme", role="owner")
mgr.grant("bob",   "acme", role="writer")
mgr.grant("guest", "acme", role="reader")

def stats(self, tenant_id: str) -> Dict[str, Any]

s = mgr.stats("acme")
# s["stm"]     → ShortTermMemory stats
# s["working"] → WorkingMemory stats
# s["ltm"]     → LongTermMemory stats
# s["semantic"]→ SemanticMemory stats
# s["profile_manager"] → aggregate user profile stats
# s["encryption_active"] → bool
# s["privacy_masking_active"] → bool

def clear_tenant(self, tenant_id: str) -> None

entry = await mgr.asave_interaction(user_id, tenant_id, user_input, assistant_output, **kwargs)
ctx   = await mgr.abuild_context(user_id, tenant_id, query, **kwargs)
report = await mgr.arun_maintenance(tenant_id)

from fennec_memory.memory import MemoryConfig

config = MemoryConfig(
    max_short_term=1000,
    max_long_term=1000,
    max_working=1000,
    importance_threshold=0.7,
    similarity_threshold=0.85,
    retrieval_limit=10,
    embedding_model="all-MiniLM-L6-v2",
    embedding_batch_size=32,
    embedding_cache_size=1000,
    normalize_text=False,
    preserve_case=True,
    enable_persistence=True,
    persistence_path="./memory_storage",
    auto_save_interval=300,
    enable_decay=True,
    decay_rate=0.1,
    min_importance=0.1,
    enable_consolidation=True,
    consolidation_interval=3600,
    max_tokens=2000,
    window_size=5,
    log_level="INFO",
    enable_stats=True,
)

config = MemoryConfig.from_env()

d = config.to_dict()

from fennec_memory.memory import ConversationBufferMemory

mem = ConversationBufferMemory(
    return_messages=False,   # True → returns List[dict], False → returns text string
    input_key="input",
    output_key="output",
    memory_key="history",
)

mem.save_context({"input": "Hello"}, {"output": "Hi there!"})
variables = mem.load_memory_variables({})
# variables["history"] → "user: Hello\n assistant: Hi there!"

from fennec_memory.memory import ConversationBufferWindowMemory

mem = ConversationBufferWindowMemory(k=5)  # remembers last 5 turns
info = mem.get_window_info()
# {"current_size": 3, "max_size": 5, "available_slots": 2, "is_full": False}

from fennec_memory.memory import ConversationSummaryMemory

mem = ConversationSummaryMemory(
    llm=my_llm,
    max_token_limit=2000,
)

mem.save_context({"input": "..."}, {"output": "..."})  # triggers summarisation when needed
summary = mem.get_summary()       # retrieve the current running summary
mem.force_summarize()             # trigger summarisation immediately

from fennec_memory.memory import ConversationEntityMemory

mem = ConversationEntityMemory(
    llm=my_llm,         # used for entity extraction; falls back to stanza NLP
    memory_key="entity_info",
    lang="en",          # stanza language code (used in fallback)
)

mem.save_context(
    {"input": "I was talking to Dr. Smith at OpenAI."},
    {"output": "That's interesting."},
)

entities = mem.list_entities()           # ["Dr. Smith", "OpenAI"]
info = mem.get_entity_info("OpenAI")     # {"name": "OpenAI", "contexts_count": 1, ...}

class BaseMemory(ABC):
    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None: ...
    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: ...
    def clear(self) -> None: ...
    def get_memory_stats(self) -> Dict[str, Any]: ...

    # Async variants (via asyncio.to_thread)
    async def asave_context(self, inputs, outputs): ...
    async def aload_memory_variables(self, inputs): ...

from fennec_memory.memory import SemanticMemory

sem = SemanticMemory(
    model_name="all-MiniLM-L6-v2",
    cache_size=1000,
    tenant_id="acme",
)

results = sem.search("Python async patterns", k=5, min_score=0.3)
for entry, score in results:
    print(f"Score: {score:.3f}  Content: {str(entry.content)[:60]}")

from fennec_memory.memory import LongTermMemory

ltm = LongTermMemory(
    db_path="./memory_storage/acme_ltm.db",
    tenant_id="acme",
)

from fennec_memory.memory import MemorySelector, SelectionConfig

config = SelectionConfig(
    relevance_weight=0.50,
    recency_weight=0.20,
    importance_weight=0.30,
    recency_half_life_hours=24.0,
    min_composite_score=0.10,
    top_k=10,
    token_budget=3000,
)
selector = MemorySelector(config=config)

explanation = selector.explain(
    query="Python async patterns",
    candidates=stm.get_all(),
)
for row in explanation:
    print(f"id={row['id']}  composite={row['composite']:.3f}  "
          f"relevance={row['relevance']:.3f}  recency={row['recency']:.3f}")

from fennec_memory.memory import MemoryCompressor

compressor = MemoryCompressor(
    llm=my_llm,
    merge_similarity_threshold=0.92,
    tenant_id="acme",
)

result = compressor.compress_batch(
    candidates,
    remove_noise=True,
    merge=True,
    summarise=True,
    summarise_threshold=5,
)
print(f"Reduced {len(candidates)} → {len(result['entries'])} entries")
print(f"Noise removed: {result['noise_removed']}, Merges: {result['merges']}")

from fennec_memory.memory import ForgettingMechanism, ForgettingConfig

forgetter = ForgettingMechanism(ForgettingConfig(
    base_decay_rate=0.15,           # importance lost per day at zero accesses
    recency_boost_per_access=0.05,  # each access multiplies effective half-life
    max_recency_boost=2.0,          # cap on recency boost multiplier
    min_importance=0.05,            # minimum floor before removal
    deletion_threshold=0.05,        # entries at or below this are "forgotten"
    high_frequency_threshold=5,     # ≥5 accesses → "well-rehearsed"
    apply_decay_on_read=False,      # if True, decay runs on every load
))

alive, forgotten = forgetter.apply(entries)
print(f"{len(alive)} entries retained, {len(forgotten)} forgotten")

# Preview without mutating
report = forgetter.report(entries)
print(f"At risk: {report['at_risk_count']} of {report['total']}")

from fennec_memory.memory import ContextBuilder, Document

builder = ContextBuilder(
    total_token_budget=4000,
    profile_budget_fraction=0.05,
    memory_budget_fraction=0.30,
    document_budget_fraction=0.45,
    history_budget_fraction=0.20,
)

def build(
    self,
    *,
    documents: Optional[List[Document]] = None,
    memories: Optional[List[MemoryEntry]] = None,
    history: Optional[List[Dict[str, str]]] = None,
    user_profile_text: Optional[str] = None,
    query: Optional[str] = None,
) -> BuiltContext

def build_prompt(
    self,
    query: str,
    *,
    documents=None,
    memories=None,
    history=None,
    user_profile_text=None,
    system_instruction="You are a helpful, knowledgeable assistant.",
) -> str

@dataclass
class Document:
    page_content: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    doc_id: Optional[str] = None

from fennec_memory.memory import UserProfileManager

manager = UserProfileManager(
    persist_dir="./memory_storage/profiles",
    auto_save=True,
)

profile = manager.get_or_create("alice")
manager.set_preference("alice", "verbosity", "detailed")
manager.set_preference("alice", "preferred_language", "en")

# Inject into prompt
profile_text = profile.to_context_string()

from fennec_memory.memory import MemoryConfig

config = MemoryConfig(
    # Memory layer capacities
    max_short_term=1000,
    max_long_term=1000,
    max_working=1000,
    max_episodic=1000,
    max_semantic=1000,
    max_procedral=1000,

    # Retrieval
    importance_threshold=0.7,    # LTM promotion cutoff
    similarity_threshold=0.85,   # duplicate detection
    retrieval_limit=10,          # default top_k

    # Embeddings
    embedding_model="all-MiniLM-L6-v2",
    embedding_batch_size=32,
    embedding_cache_size=1000,

    # Text processing
    normalize_text=False,
    preserve_case=True,
    remove_duplicates=True,

    # Persistence
    enable_persistence=True,
    persistence_path="./memory_storage",
    auto_save_interval=300,

    # LangChain-style keys
    return_messages=False,
    input_key="input",
    output_key="output",
    memory_key="history",
    max_token_limit=2000,
    window_size=5,

    # Decay
    enable_decay=True,
    decay_rate=0.1,        # 10% per day at zero accesses
    min_importance=0.1,

    # Consolidation
    enable_consolidation=True,
    consolidation_interval=3600,

    # LLM token budget
    max_tokens=2000,

    # Logging
    log_level="INFO",
    enable_stats=True,
)

from fennec_memory.memory import SelectionConfig

selection_config = SelectionConfig(
    relevance_weight=0.50,         # cosine similarity fraction
    recency_weight=0.20,           # recency exponential decay fraction
    importance_weight=0.30,        # effective_importance fraction
    recency_half_life_hours=24.0,  # score halves every 24 hours
    min_composite_score=0.10,      # minimum score to be included
    top_k=10,                      # maximum entries to select
    token_budget=3000,             # maximum tokens across all selected entries
)

from fennec_memory.memory import ForgettingConfig

forgetting_config = ForgettingConfig(
    base_decay_rate=0.15,
    recency_boost_per_access=0.05,
    max_recency_boost=2.0,
    min_importance=0.05,
    deletion_threshold=0.05,
    high_frequency_threshold=5,
    apply_decay_on_read=False,
)

export MEMORY_MAX_SHORT_TERM=500
export MEMORY_MAX_LONG_TERM=5000
export MEMORY_EMBEDDING_MODEL="all-MiniLM-L6-v2"
export MEMORY_PERSISTENCE_PATH="/var/data/memory"
export MEMORY_ENABLE_PERSISTENCE="true"

config = MemoryConfig.from_env()

from fennec_memory.memory import SensitiveDataMasker

masker = SensitiveDataMasker(
    custom_patterns=[
        ("INTERNAL_ID", r"INT-\d{6}"),  # add your own
    ],
    placeholder_fmt="[{label}]",
)

cleaned, report = masker.mask("Contact me at ceo@company.com, card: 4111111111111111")
# cleaned → "Contact me at [EMAIL], card: [CREDIT_CARD]"
# report  → {"EMAIL": 1, "CREDIT_CARD": 1}

# Check for sensitive data without masking
has_pii = masker.has_sensitive_data("Contact me at ceo@company.com")
# → True

# Recursively mask nested dicts or lists
cleaned_content, report = masker.mask_entry_content({"input": "my SSN is 123-45-6789"})

from fennec_memory.memory import MemoryEncryptor

enc = MemoryEncryptor(secret_key="my-production-secret")

token = enc.encrypt({"input": "sensitive data", "output": "sensitive answer"})
data  = enc.decrypt(token)

print(enc.is_real_encryption)   # True if cryptography package is installed
print(enc.key_fingerprint)      # SHA-256 fingerprint of the key (first 16 hex chars)

from fennec_memory.memory import AccessController, Permission

ac = AccessController()

# Built-in role assignment
ac.assign_role("alice", "acme", "owner")
ac.assign_role("bob",   "acme", "writer")
ac.assign_role("guest", "acme", "reader")

# Custom role
ac.define_role("analyst", {Permission.READ})
ac.assign_role("carol", "acme", "analyst")

# Check and enforce
if ac.has_permission("bob", "acme", Permission.WRITE):
    # proceed
    pass

ac.require("guest", "acme", Permission.DELETE)  # raises PermissionError

# Inspect
perms = ac.list_permissions("alice", "acme")  # {Permission.READ, Permission.WRITE, Permission.DELETE}

# Revoke
ac.revoke_role("bob", "acme", "writer")

from fennec_memory.memory import ShortTermMemory

stm = ShortTermMemory(window_size=20, tenant_id="acme")

from fennec_memory.memory import WorkingMemory

wm = WorkingMemory(capacity=10, token_budget=3000, tenant_id="acme")

s = mgr.stats("acme")

{
  "tenant_id": "acme",
  "stm": {
    "type": "ShortTermMemory",
    "window_size": 20,
    "current_size": 14,
    "is_full": false,
    "tenant_id": "acme"
  },
  "working": {
    "type": "WorkingMemory",
    "capacity": 10,
    "token_budget": 4000,
    "entries": 6,
    "estimated_tokens_used": 850,
    "tenant_id": "acme"
  },
  "ltm": {
    "type": "LongTermMemory",
    "db_path": "./memory_storage/acme_ltm.db",
    "tenant_id": "acme",
    "total_entries": 342
  },
  "semantic": {
    "type": "SemanticMemory",
    "backend": "faiss",
    "model": "all-MiniLM-L6-v2",
    "entries": 356,
    "tenant_id": "acme"
  },
  "profile_manager": {
    "user_count": 3,
    "total_interactions": 1204,
    "avg_interactions_per_user": 401.3,
    "top_global_topics": [["python", 340], ["AI", 210]]
  },
  "encryption_active": true,
  "privacy_masking_active": true
}

report = mgr.run_maintenance("acme")
# {
#   "tenant_id": "acme",
#   "timestamp": 1714000000.0,
#   "stm_evicted": 2,
#   "ltm_decay_updated": 120,
#   "ltm_deleted": 5,
#   "working_forgotten": 1,
#   "ltm_compression": {"merges": 3, "noise_removed": 7}
# }

report = mgr._forgetter.report(bundle.stm.get_all())
print(f"At-risk: {report['at_risk_count']} / {report['total']}")
print(f"At-risk IDs: {report['at_risk_ids']}")

explanation = mgr._selector.explain(
    query="Python caching strategies",
    candidates=bundle.stm.get_all() + bundle.ltm.get_top_by_importance(20),
)
for row in explanation:
    print(f"{row['id'][:10]}  composite={row['composite']:.3f}  "
          f"rel={row['relevance']:.3f}  rec={row['recency']:.3f}  "
          f"imp={row['importance']:.3f}  |  {row['content'][:60]}")

mgr._profile_manager.aggregate_stats()
# {
#   "user_count": 15,
#   "total_interactions": 5230,
#   "avg_interactions_per_user": 348.7,
#   "top_global_topics": [["python", 1200], ...]
# }

# Correct in async context
entry = await mgr.asave_interaction(...)

# Correct in sync context
entry = asyncio.run(mgr.asave_interaction(...))

# Wrong — returns a coroutine object, not the result
entry = mgr.asave_interaction(...)  # ← missing await

if entries:
    alive, forgotten = forgetter.apply(entries)
else:
    alive, forgotten = [], []

from memory import AIMemoryManager, MemoryConfig

config = MemoryConfig(
    persistence_path="/var/data/memory",
    importance_threshold=0.6,
    max_tokens=4000,
    enable_persistence=True,
)

mgr = AIMemoryManager(
    config=config,
    llm=my_llm,
    enable_privacy_masking=True,
    encryption_key=os.environ["MEMORY_ENCRYPTION_KEY"],
)

# Tenant A — full team roles
mgr.grant("admin_user", "tenant_a", role="admin")
mgr.grant("agent_1",    "tenant_a", role="writer")
mgr.grant("readonly",   "tenant_a", role="reader")

# Tenant B — separate namespace, no data overlap
mgr.grant("agent_2", "tenant_b", role="owner")

# Tenant A and B memories are fully isolated
mgr.save_interaction("agent_1", "tenant_a", "secret A", "answer A", importance=0.9)
mgr.save_interaction("agent_2", "tenant_b", "secret B", "answer B", importance=0.9)

ctx_a = mgr.build_context("agent_1", "tenant_a", "secret B")
# ctx_a.full_text will NOT contain "secret B" — tenant_b data is invisible

# High-quality responses: store with elevated importance
mgr.save_interaction(..., importance=0.9)

# Routine exchanges: store with lower importance
mgr.save_interaction(..., importance=0.4)

# Profile-based boost: users whose memories are consistently useful
# get a `memory_retention_boost` applied automatically
mgr._profile_manager.get_or_create("alice").update_retention_boost(0.15)

from memory import Document

def answer_with_memory_and_rag(mgr, retriever, user_id, tenant_id, query):
    # 1. Retrieve documents from your vector / BM25 store
    raw_docs = retriever.search(query, k=5)
    documents = [
        Document(content=d.text, source=d.url, score=d.score)
        for d in raw_docs
    ]

    # 2. Build memory-enriched context
    ctx = mgr.build_context(
        user_id=user_id,
        tenant_id=tenant_id,
        query=query,
        documents=documents,
        top_k_semantic=5,
    )

    # 3. Generate response
    response = my_llm.generate(ctx.full_text + f"\n\nUser: {query}\nAssistant:")

    # 4. Persist the interaction
    mgr.save_interaction(user_id, tenant_id, query, response, importance=0.7)

    return response

mgr._profile_manager.set_preference("alice", "verbosity", "concise")
mgr._profile_manager.set_preference("alice", "preferred_language", "fr")
mgr._profile_manager.set_preference("alice", "domain", "finance")

profile = mgr._profile_manager.get_or_create("alice")
print(profile.to_context_string())
# User language: fr
# Response style: concise / neutral
# Interests: finance, python, ...
# domain: finance

from fastapi import FastAPI
from contextlib import asynccontextmanager
from fennec_memory.memory import AIMemoryManager, MemoryConfig

memory_manager: AIMemoryManager = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global memory_manager
    config = MemoryConfig(persistence_path="/var/data/memory", max_tokens=4000)
    memory_manager = AIMemoryManager(config=config, llm=my_llm)
    yield
    # Cleanup if needed

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(user_id: str, tenant_id: str, message: str):
    ctx = await memory_manager.abuild_context(user_id, tenant_id, message)
    response = await my_async_llm.generate(ctx.full_text + "\n\n" + message)
    await memory_manager.asave_interaction(
        user_id, tenant_id, message, response, importance=0.7
    )
    return {"response": response}

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_memory.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    ConversationEntityMemory,
)

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)

rag = rag_system  
def conversational_rag_query(
    query: str,
    memory,
    memory_key: str = "history",
) -> str:
    mem_vars  = memory.load_memory_variables({"input": query})
    history   = mem_vars.get(memory_key, "")
    retrieved = rag.retrieve(query, top_k=2)
    context   = rag.context_manager.build(query, retrieved)
    if history:
        full_prompt = (
            f"سياق المحادثة السابقة:\n{str(history)[:400]}\n\n"
            f"معلومات مسترجعة:\n{context}\n\n"
            f"السؤال الحالي: {query}\nالإجابة:"
        )
    else:
        full_prompt = f"معلومات مسترجعة:\n{context}\n\nالسؤال: {query}\nالإجابة:"
    answer = rag.llm.generate(full_prompt)
    memory.save_context({"input": query}, {"output": answer})
    return answer

print("\n  [5a] ConversationBufferMemory ")
buffer_mem = ConversationBufferMemory(
    return_messages=False, input_key="input", output_key="output", memory_key="history"
)
conversation = [
    "ما هي طرق الدفع المتاحه ",
    "ما هي اعدادهم",
]
print("\n  💬  conversion with RAG (Buffer Memory):")
for turn, q in enumerate(conversation, 1):
    answer = conversational_rag_query(q, buffer_mem)
    print(f"  [{turn}] 👤 {q}")
    print(f"       🤖 {answer[:80]}...")
    print()
print(f"  📝 Buffer memory: {len(buffer_mem.chat_memory)}  saved message")
# ── 5b: Window Memory ──────────────── #
print("\n  [5b] ConversationBufferWindowMemory — نافذة K=2")
window_mem = ConversationBufferWindowMemory(k=2)
long_conversation = [
    "ما هو  طرق التواصل مع فريق الدعم ",
    "اعطني مثال عليها",
]
for q in long_conversation:
    conversational_rag_query(q, window_mem)
print(f"  📝 Window memory (k=2): {len(window_mem.chat_memory)} رسائل محفوظة (آخر 2)")
if window_mem.chat_memory:
    print(f"  last  quesion: {window_mem.chat_memory[-1].get('input', '')}")