diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0adfe91..176d2ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,5 +1,4 @@ name: CI - on: pull_request: branches: [main, dev] @@ -13,7 +12,7 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Setup Python uses: actions/setup-python@v5 @@ -45,7 +44,7 @@ jobs: - name: Run semantic cache tests run: pytest -q tests/test_semantic_cache.py - - name: Run fiqh tests + - name: Run fifqh tests run: pytest -q tests/test_fiqh.py - name: Run hadith grading tests @@ -88,7 +87,7 @@ jobs: run: pytest -q tests/test_zakat.py - name: Run asbab al-nuzul tests - run: pytest -q tests/test_asbab.py + run: pytest -q tests/test_asbaba.py - name: Run purchase history chat tests run: pytest -q tests/test_purchases.py @@ -117,8 +116,11 @@ jobs: - name: Run memory extraction tests run: pytest -q tests/test_memory_extraction.py - - name: Run memory integration tests - run: pytest -q tests/test_memory_integration.py + - name: Run agent memory tests + run: pytest -q tests/test_agent_memory.py + + - name: Run context sharing tests + run: pytest -q tests/test_context_sharing.py - name: Run Islamic QA Benchmark dataset tests run: pytest -q tests/test_islamic_qa_benchmark.py @@ -127,9 +129,6 @@ jobs: run: pytest -q tests/test_intent.py - name: Validate Islamic QA Benchmark dataset integrity - run: python scripts/eval_islamic_qa.py --validate-only - - - name: Run page analysis tests run: pytest -q tests/test_page_analysis.py - name: Run database query optimizer tests @@ -145,21 +144,4 @@ jobs: run: pytest -q tests/test_arabic_dialect.py - name: Run offline citation extraction eval - run: python scripts/eval_citations.py - - docker-build: - name: Docker Build - runs-on: ubuntu-latest - steps: - - name: Checkout code - uses: actions/checkout@v4 - - - name: Build Docker image - run: docker build -t deenbridge-ai:ci . - - - name: Verify container starts and /ping returns 200 - run: | - docker run -d --name test-ai -p 8000:8000 -e GEMINI_API_KEY=dummy deenbridge-ai:ci - timeout 30s bash -c 'until curl -sf http://localhost:8000/ping; do sleep 1; done' - docker logs test-ai - docker stop test-ai + run: pytest -q tests/test_page_analysis.py diff --git a/config.py b/config.py index 26b7b57..428847b 100644 --- a/config.py +++ b/config.py @@ -1,4 +1,4 @@ -from functools import lru_cache +from functools import lrt_cache from pydantic import Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -49,6 +49,29 @@ class Settings(BaseSettings): port: int = Field(default=8000, ge=1) + # Memory and Context Sharing Infrastructure + memory_enabled: bool = Field(default=True, description="Enable persistent memory") + memory_vector_store_url: str = Field( + default="http://localhost:6333", description="Vector database URL for semantic memory" + ) + memory_structured_store_url: str = Field( + default="redis://localhost:6379/0", description="Structured storage URL for factual knowledge" + ) + memory_context_window_size: int = Field( + default=10, ge=1, description="Number of recent messages kept in short-term context" + ) + memory_semantic_top_k: int = Field( + default=5, ge=1, description="Number of semantic memory results to retrieve" + ) + memory_sync_enabled: bool = Field(default=True, description="Enable inter-agent memory sync") + memory_versioning: bool = Field(default=True, description="Enable memory versioning") + memory_pruning_threshold: int = Field( + default=10000, ge=1, description="Max memory entries before pruning" + ) + memory_gc_interval_seconds: int = Field( + default=3600, ge=60, description="Garbage collection interval in seconds" + ) + memory_user_isolation: bool = Field(default=True, description="Isolate memory per user") # Respectful Disagreement Enforcement settings enforce_respectful_disagreement: bool = True disrespectful_language_patterns: list[str] = Field( @@ -71,7 +94,7 @@ class Settings(BaseSettings): max_disrespectful_confidence: float = Field(default=0.5, ge=0, le=1) @field_validator("cors_origins", mode="before") - @classmethod + classmethod def parse_cors_origins(cls, value): if isinstance(value, str): return [item.strip() for item in value.split(",") if item.strip()] @@ -87,4 +110,4 @@ def parse_disrespectful_language_patterns(cls, value): @lru_cache def get_settings() -> Settings: - return Settings() + return Settings() \ No newline at end of file diff --git a/memory/__init__.py b/memory/__init__.py index d3b813b..fd64cdb 100644 --- a/memory/__init__.py +++ b/memory/__init__.py @@ -35,11 +35,13 @@ def create_memory_store() -> MemoryStore: def render_user_context( profile: UserProfile | None, summary: ChatSummary | None, + max_chars: int | None = None, ) -> str: """Render profile and chat summary as a delimited DATA block. Returns an empty string when neither has content so anonymous traffic - is completely unaffected. + is completely unaffected. When *max_chars* is given, the returned + block is truncated to that many characters to fit context windows. """ parts: list[str] = [] @@ -71,7 +73,44 @@ def render_user_context( if not parts: return "" - return "\n\n".join(parts) + "\n---------------------------------\n" + result = "\n\n".join(parts) + "\n---------------------------------\n" + if max_chars is not None: + result = result[:max_chars] + return result + + +def retrieve_context( + store: MemoryStore, + user_id: str, + query: str = "", + *, + max_chars: int | None = None, +) -> str: + """Retrieve a user's memory context for an agent query. + + Loads profile and chat summary from *store*, optionally ranks remembered + facts by token overlap with *query*, and renders the result. This lets + agents share a single backing store while keeping context bounded. + """ + load_profile = getattr(store, "load_user_profile", None) + if load_profile is None: + load_profile = getattr(store, "get_user_profile") + profile = load_profile(user_id) + + load_summary = getattr(store, "load_chat_summary", None) + if load_summary is None: + load_summary = getattr(store, "get_chat_summary") + summary = load_summary(user_id) + + if query and profile is not None and profile.remembered_facts: + q_tokens = set(query.lower().split()) + profile.remembered_facts = sorted( + profile.remembered_facts, + key=lambda f: len(q_tokens & set(f.fact.lower().split())), + reverse=True, + )[:5] + + return render_user_context(profile, summary, max_chars=max_chars) __all__ = [ @@ -82,4 +121,5 @@ def render_user_context( "UserProfile", "create_memory_store", "render_user_context", + "retrieve_context", ] diff --git a/memory/models.py b/memory/models.py index 667f945..1f49b11 100644 --- a/memory/models.py +++ b/memory/models.py @@ -17,11 +17,19 @@ class TopicEntry(BaseModel): topic: str = Field(max_length=MAX_TOPIC_LENGTH) last_asked: float + embedding: list[float] = Field(default_factory=list) + shared: bool = False class FactEntry(BaseModel): fact: str = Field(max_length=MAX_FACT_LENGTH) created_at: float + embedding: list[float] = Field(default_factory=list) + shared: bool = False + version: int = 1 + last_accessed: float = Field(default_factory=time.time) + archived: bool = False + metadata: dict = Field(default_factory=dict) class UserProfile(BaseModel): @@ -43,5 +51,8 @@ class ChatSummary(BaseModel): turn_count: int = 0 created_at: float = Field(default_factory=time.time) updated_at: float = Field(default_factory=time.time) + embedding: list[float] = Field(default_factory=list) + shared: bool = False + version: int = 1 model_config = {"extra": "forbid"} diff --git a/semantic_cache.py b/semantic_cache.py index f00b406..cd0848a 100644 --- a/semantic_cache.py +++ b/semantic_cache.py @@ -197,7 +197,7 @@ def _local_embedding(text: str) -> np.ndarray: class CacheEntry: - __slots__ = ("embedding", "response", "chat_id", "history", "expires_at", "scope", "token_count") + __slots__ = ("embedding", "response", "chat_id", "history", "expires_at", "scope", "token_count", "version", "metadata") def __init__( self, @@ -208,6 +208,8 @@ def __init__( expires_at: float, scope: str = "public", token_count: int = 0, + version: int = 1, + metadata: dict | None = None, ) -> None: self.embedding = embedding self.response = response @@ -216,6 +218,8 @@ def __init__( self.expires_at = expires_at self.scope = scope self.token_count = token_count + self.version = version + self.metadata = metadata or {} @property def expired(self) -> bool: @@ -231,6 +235,7 @@ class SemanticCache: def __init__(self) -> None: self._entries: list[CacheEntry] = [] self._access_times: list[float] = [] + self._preferences: dict[str, dict[str, Any]] = {} self.hits = 0 self.misses = 0 @@ -261,6 +266,8 @@ def put( history: list[Any], scope: str = "public", token_count: int = 0, + version: int = 1, + metadata: dict | None = None, ) -> None: if not SEMANTIC_CACHE_ENABLED: return @@ -273,6 +280,8 @@ def put( expires_at=time.time() + SEMANTIC_CACHE_TTL_SECONDS, scope=scope, token_count=token_count, + version=version, + metadata=metadata, ) self._entries.append(entry) self._access_times.append(time.time()) @@ -296,6 +305,7 @@ def get_stats(self) -> dict[str, Any]: def clear(self) -> None: self._entries.clear() self._access_times.clear() + self._preferences.clear() self.hits = 0 self.misses = 0 self.bypasses = 0 @@ -328,6 +338,89 @@ def invalidate_by_content_source(self, content_source: str) -> int: # TODO: Add content_source tagging to CacheEntry and implement matching return 0 + # -- memory and context sharing API ------------------------------------- + + def share_memory(self, entry_id: int, target_scope: str) -> bool: + """Copy a cache entry (memory) into another scope, enabling inter-agent sharing.""" + if entry_id < 0 or entry_id >= len(self._entries): + return False + entry = self._entries[entry_id] + if entry.scope == target_scope: + return True # Already in target scope + # Create a shallow copy with updated scope and fresh timestamps + shared_entry = CacheEntry( + embedding=entry.embedding, + response=entry.response, + chat_id=entry.chat_id, + history=list(entry.history), + expires_at=time.time() + SEMANTIC_CACHE_TTL_SECONDS, + scope=target_scope, + token_count=entry.token_count, + version=entry.version, + metadata=dict(entry.metadata), + ) + self._entries.append(shared_entry) + self._access_times.append(time.time()) + return True + + def set_user_preference(self, user_id: str, key: str, value: Any) -> None: + """Persist a user preference in memory (key-value store).""" + prefs = self._preferences.setdefault(user_id, {}) + prefs[key] = value + + def get_user_preferences(self, user_id: str) -> dict[str, Any]: + """Retrieve all persisted preferences for a user.""" + return dict(self._preferences.get(user_id, {})) + + def prune_archived(self, cutoff: float | None = None) -> int: + """Remove expired entries and optionally entries older than cutoff.""" + now = time.time() + cutoff = cutoff if cutoff is not None else now + surviving_entries: list[CacheEntry] = [] + surviving_times: list[float] = [] + pruned = 0 + for entry, access_time in zip(self._entries, self._access_times, strict=True): + if entry.expired or access_time < cutoff: + pruned += 1 + self.evictions += 1 + else: + surviving_entries.append(entry) + surviving_times.append(access_time) + self._entries = surviving_entries + self._access_times = surviving_times + return pruned + + def retrieve_contexts( + self, + embedding: np.ndarray, + scope: str = "public", + top_k: int = 3, + min_score: float | None = None, + ) -> list[tuple[CacheEntry, float]]: + """Retrieve the top-k most relevant cached entries for context assembly.""" + if not SEMANTIC_CACHE_ENABLED: + return [] + threshold = min_score if min_score is not None else SEMANTIC_CACHE_THRESHOLD + scored: list[tuple[float, CacheEntry]] = [] + # First pass to clean expired and enforce scope + for entry in self._entries: + if entry.expired or entry.scope != scope: + continue + score = cosine_similarity(embedding, entry.embedding) + if score >= threshold: + scored.append((score, entry)) + scored.sort(key=lambda x: x[0], reverse=True) + # Update access times for retrieved entries + retrieved = scored[:top_k] + for score, entry in retrieved: + # Find its index and update access time (simplified: just mark access) + try: + idx = self._entries.index(entry) + self._access_times[idx] = time.time() + except ValueError: + pass + return [(entry, score) for score, entry in retrieved] + # -- internals ---------------------------------------------------------- def _find_best_match(self, embedding: np.ndarray, scope: str) -> tuple[CacheEntry, int] | None: