# v2 git bundle 7e8a75a40e834b4781ed70e9953e526644dd483a refs/heads/bundle 7e8a75a40e834b4781ed70e9953e526644dd483a HEAD PACK'xLtree 5ab639f7356e9648fcdd2c9c8d48a48857e82538 parent 4e671a0a83278b4c0cfc6932e70a934c55abe303 author mvyshhnyvetska 1774530622 +0100 committer mvyshhnyvetska 1774530622 +0100 Add PyPI publish workflow (Trusted Publisher) Co-Authored-By: Claude Opus 4.6 (1M context) eiF(xvtree e353233f64be8773d5bbc79cfd5c9d175cf637cf author mvyshhnyvetska 1774529400 +0100 committer mvyshhnyvetska 1774529400 +0100 Initial release: breathe-memory v0.1.0 Context optimization and associative memory for LLM applications. Two-phase system: SYNAPSE (pre-generation memory injection) + GraphCompactor (structured context compression). - Interface-based, storage-agnostic, LLM-agnostic - Memory Nexus: PostgreSQL + pgvector reference backend - Zero mandatory dependencies beyond stdlib - 28 tests passing, clean install verified Co-Authored-By: Claude Opus 4.6 (1M context) x$100644 .gitignore1fd2YPk *100644 LICENSEB|0ViiCA100644 README.md%g24P |40000 breatheU.Օ"鮫p-f\x40000 examples ?.x( 40000 memory_nexus^{˦){˄(2Tb100644 pyproject.toml2uA?f6bӅ40000 testsrw/)&\.W[q1x*"40000 .githubR:t^˴cLarsj$ 9x$40000 workflows]bT] mUMͮG)x'100644 publish.yml gҋ~6 =Axname: Publish to PyPI on: release: types: [published] workflow_dispatch: jobs: publish: runs-on: ubuntu-latest permissions: id-token: write steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.11" - name: Install build tools run: pip install build - name: Build package run: python -m build - name: Publish to PyPI uses: pypa/gh-action-pypi-publish@release/v1 xE# Python __pycache__/ *.py[cod] *$py.class *.egg-info/ dist/ build/ *.egg # Virtual environments .venv/ venv/ ENV/ # Environment variables .env *.env .env.* # IDE .idea/ .vscode/ *.swp *.swo *~ # OS .DS_Store Thumbs.db # Testing .pytest_cache/ htmlcov/ .coverage coverage.xml # TDD scan output kenaz-out/ coverage.json {gx Apache License Version 2.0, January 2004 http://www.apache.org/licenses/ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION 1. Definitions. "License" shall mean the terms and conditions for use, reproduction, and distribution as defined by Sections 1 through 9 of this document. "Licensor" shall mean the copyright owner or entity authorized by the copyright owner that is granting the License. "Legal Entity" shall mean the union of the acting entity and all other entities that control, are controlled by, or are under common control with that entity. For the purposes of this definition, "control" means (i) the power, direct or indirect, to cause the direction or management of such entity, whether by contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the outstanding shares, or (iii) beneficial ownership of such entity. "You" (or "Your") shall mean an individual or Legal Entity exercising permissions granted by this License. "Source" form shall mean the preferred form for making modifications, including but not limited to software source code, documentation source, and configuration files. "Object" form shall mean any form resulting from mechanical transformation or translation of a Source form, including but not limited to compiled object code, generated documentation, and conversions to other media types. "Work" shall mean the work of authorship made available under the License, as indicated by a copyright notice that is included in or attached to the work. "Derivative Works" shall mean any work, whether in Source or Object form, that is based on (or derived from) the Work and for which the editorial revisions, annotations, elaborations, or other modifications represent, as a whole, an original work of authorship. "Contribution" shall mean, as submitted to the Licensor for inclusion in the Work by the copyright owner or by an individual or Legal Entity authorized to submit on behalf of the copyright owner. For the purposes of this definition, "submit" means any form of electronic, verbal, or written communication sent to the Licensor or its representatives, including but not limited to communication on electronic mailing lists, source code control systems, and issue tracking systems that are managed by, or on behalf of, the Licensor for the purpose of discussing and improving the Work, but excluding communication that is conspicuously marked or designated in writing by the copyright owner as "Not a Contribution." "Contributor" shall mean Licensor and any Legal Entity on behalf of whom a Contribution has been received by the Licensor and included within the Work. 2. Grant of Copyright License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable copyright license to reproduce, prepare Derivative Works of, publicly display, publicly perform, sublicense, and distribute the Work and such Derivative Works in Source or Object form. 3. Grant of Patent License. Subject to the terms and conditions of this License, each Contributor hereby grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except as stated in this section) patent license to make, have made, use, offer to sell, sell, import, and otherwise transfer the Work. 4. Redistribution. You may reproduce and distribute copies of the Work or Derivative Works thereof in any medium, with or without modifications, and in Source or Object form, provided that You meet the following conditions: (a) You must give any other recipients of the Work or Derivative Works a copy of this License; and (b) You must cause any modified files to carry prominent notices stating that You changed the files; and (c) You must retain, in the Source form of any Derivative Works that You distribute, all copyright, patent, trademark, and attribution notices from the Source form of the Work; and (d) If the Work includes a "NOTICE" text file, you must include a readable copy of the attribution notices contained within such NOTICE file in at least one of the following places: within a NOTICE text file distributed as part of the Derivative Works; within the Source form or documentation, if provided along with the Derivative Works; or, within a display generated by the Derivative Works, if and wherever such third-party notices normally appear. 5. Submission of Contributions. Unless You explicitly state otherwise, any Contribution intentionally submitted for inclusion in the Work by You to the Licensor shall be under the terms and conditions of this License, without any additional terms or conditions. 6. Trademarks. This License does not grant permission to use the trade names, trademarks, service marks, or product names of the Licensor, except as required for reasonable and customary use in describing the origin of the Work. 7. Disclaimer of Warranty. Unless required by applicable law or agreed to in writing, Licensor provides the Work (and each Contributor provides its Contributions) on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 8. Limitation of Liability. In no event and under no legal theory, whether in tort (including negligence), contract, or otherwise, unless required by applicable law (such as deliberate and grossly negligent acts) or agreed to in writing, shall any Contributor be liable to You for damages, including any direct, indirect, special, incidental, or exemplary damages of any character arising as a result of this License or out of the use or inability to use the Work. 9. Accepting Warranty or Additional Liability. While redistributing the Work or Derivative Works thereof, You may choose to offer, and charge a fee for, acceptance of support, warranty, indemnity, or other liability obligations and/or rights consistent with this License. Copyright 2024 Kenaz GmbH Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. .[x9y# breathe-memory **Context optimization and associative memory for LLM applications.** Two-phase system built around how memory actually works — not as lookup, but as association. ``` pip install breathe-memory ``` --- ## What it does LLMs forget. Context windows are finite and expensive. Most solutions either stuff everything in (burns tokens) or summarize (loses structure). **BREATHE** does neither: - **SYNAPSE (inhale)** — before each generation, extracts associative anchors from the user message and injects semantically relevant memories directly into the prompt. The LLM starts thinking with context already loaded. Overhead: 2–20ms. - **GraphCompactor (exhale)** — when context fills up, extracts a structured graph (topics, decisions, open questions, artifacts) instead of a lossy narrative summary. Typically saves 60–80% of tokens while preserving semantic structure. ``` ┌─────────────────────────────────────┐ User message ──▶│ SYNAPSE (inhale) │ │ │ │ 1. Extract anchors (regex, 2ms) │ │ 2. Traverse memory graph (BFS) │ │ 3. Vector search (optional) │ │ 4. Inject │ └──────────────────┬──────────────────┘ │ ▼ LLM with memory context │ ┌──────────────────▼──────────────────┐ │ GraphCompactor (exhale) │ │ (fires when context ~80% full) │ │ │ │ Compressible messages ──▶ LLM call │ │ → Topics, Decisions, Open, │ │ Artifacts, Context, Dropped │ │ │ │ Protected messages ──▶ kept intact │ └─────────────────────────────────────┘ ``` --- ## Quick start ```python import asyncio from breathe import Synapse, GraphCompactor, BreatheConfig from breathe.interfaces import MemoryRepository, LLMClient, RetrievedNode # Implement these two interfaces for your backend class MyMemoryRepo(MemoryRepository): async def get_concepts(self): return {"FastAPI": "uuid-001", "Redis": "uuid-002"} async def graph_bfs(self, start_ids, **kwargs): return [] # implement BFS against your DB async def keyword_search(self, keywords, limit=5): return [] # implement ILIKE against your memories table class MyLLMClient(LLMClient): async def complete(self, prompt, max_tokens=4000, temperature=0.2): # call your LLM API here ... async def main(): config = BreatheConfig() synapse = Synapse(repository=MyMemoryRepo(), config=config) await synapse.initialize() messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "How should I structure my FastAPI endpoints?"}, ] # Inject associative memory before each LLM call messages = await synapse.inject(messages) # When context gets full, compress with GraphCompactor compactor = GraphCompactor(llm_client=MyLLMClient()) result = await compactor.compress(messages) messages = result["compressed_messages"] asyncio.run(main()) ``` --- ## With Memory Nexus (PostgreSQL + pgvector) ```python from breathe import Synapse, BreatheConfig from memory_nexus import PostgresMemoryStore store = PostgresMemoryStore(dsn="postgresql://localhost/mydb") await store.initialize() # Store memories await store.store("FastAPI handles async requests efficiently") await store.store("Redis is ideal for session storage and caching") # Wire into SYNAPSE — store implements VectorSearchClient synapse = Synapse(vector_client=store, config=BreatheConfig()) await synapse.initialize() messages = await synapse.inject(messages) ``` **PostgreSQL schema (default — 384-dim):** ```sql CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE memories ( id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, content TEXT NOT NULL, embedding vector(384), metadata JSONB DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX ON memories USING ivfflat (embedding vector_cosine_ops); ``` **Embedding models:** The default model (`all-MiniLM-L6-v2`, 384-dim, ~90 MB) is good for prototyping. For production, we recommend `intfloat/multilingual-e5-large` (1024-dim, ~1.2 GB) — significantly better retrieval quality, especially for multilingual content. To switch, pass `model_name` and adjust your table's vector dimension: ```python store = PostgresMemoryStore( dsn="postgresql://localhost/mydb", model_name="intfloat/multilingual-e5-large", # 1024-dim, multilingual ) ``` ```sql -- For e5-large, use vector(1024) instead of vector(384) CREATE TABLE memories ( ... embedding vector(1024), ... ); ``` --- ## Language support Built-in: **English**. Custom languages in ~10 lines: ```python import re from breathe import Synapse, BreatheConfig, LanguagePack GERMAN = LanguagePack( code="de", stopwords=frozenset({"der", "die", "das", "und", "ist", ...}), hub_exclusions=frozenset({"system", "speicher"}), temporal_pattern=re.compile(r"\b(gestern|heute|morgen|neulich)\b", re.I), emotional_pattern=re.compile(r"\b(müde|glücklich|traurig|wütend)\b", re.I), labels={"themes": "Themen", "insights": "Erkenntnisse"}, ) config = BreatheConfig(language_packs=[GERMAN], default_language="de") synapse = Synapse(config=config, ...) ``` Language packs control: - **Stopwords** — excluded from relevance scoring - **Hub exclusions** — nodes too generic to be useful for injection (e.g. "system", "memory"). Add your most frequent root concepts here — words that connect to everything are noise in retrieval. The more specific your exclusions, the sharper your injections. - **Temporal and emotional regex patterns** — anchor extraction for time references and emotional signals - **UI section labels** — headers used in the injected `` block --- ## Architecture ### SYNAPSE pipeline (per-request, <200ms) ``` User message │ ▼ AnchorExtractor ├─ Match known concepts (regex, 0.9 confidence) ├─ Temporal patterns (0.7) ├─ Technical patterns (0.5) └─ Emotional signals (0.6) │ ▼ [optional Phase 3 — Apple Silicon only] ModelAnchorExtractor (local LLM via MLX, ~250ms) └─ Fires only when regex finds <5 matched nodes │ ▼ Three traversal strategies (in parallel): 1. Graph BFS ── memory_nodes + memory_edges (recursive CTE) 2. Vector search── any VectorSearchClient (pgvector, Pinecone, etc.) 3. Keyword search── ILIKE on unmatched anchors │ ▼ Relevance filter ├─ Hub exclusion (drop super-generic nodes) ├─ Session dedup (skip already-injected nodes) └─ Keyword overlap scoring (anchor words vs node content) │ ▼ ContextInjector └─ block → prepended to last user message ``` ### GraphCompactor (when context fills up) ``` Old messages (compressible zone) │ ▼ preprocess: compress tool call JSON ▼ LLM extraction call (your LLMClient) │ ▼ SessionGraph: Topics / Decisions / Open / Artifacts / Context / Dropped │ ▼ [SESSION GRAPH] message + protected recent messages ``` --- ## Configuration ```python from breathe import BreatheConfig from breathe.config import ENGLISH config = BreatheConfig( # Language packs (all active simultaneously) language_packs=[ENGLISH], default_language="en", # SYNAPSE tuning min_similarity=0.55, # min vector similarity to accept max_injected_nodes=15, # max nodes per injection enable_model_extractor=True, model_trigger_threshold=5, # model fires when regex finds <5 nodes # Token budgets by conversation mode mode_budgets={ "casual": 1500, "work": 2500, "deep": 4000, "balanced": 2000, }, # GraphCompactor compactor_model="claude-sonnet-4-20250514", compactor_fallback_model="claude-haiku-4-5-20251001", min_tokens_to_compress=300, protected_messages_normal=10, ) ``` --- ## Implementing backends ### MemoryRepository (for graph BFS + keyword search) ```python from breathe.interfaces import MemoryRepository, RetrievedNode class MyRepo(MemoryRepository): async def get_concepts(self) -> dict[str, str]: # Return {concept_text: uuid} from your knowledge graph return {"Redis": "abc-123", "FastAPI": "def-456"} async def graph_bfs(self, start_ids, max_depth=2, **kwargs) -> list[RetrievedNode]: # BFS from start_ids through your concept graph # Recursive CTE on (memory_nodes, memory_edges) works well ... async def keyword_search(self, keywords, limit=5) -> list[RetrievedNode]: # ILIKE search over your memories/documents table ... async def flush_edges(self, edges) -> int: # Optional: persist new session graph edges to long-term storage return 0 ``` ### VectorSearchClient (for semantic search) ```python from breathe.interfaces import VectorSearchClient, RetrievedNode class PineconeClient(VectorSearchClient): async def search(self, query: str, limit: int = 5) -> list[RetrievedNode]: # embed query, search your vector index, return RetrievedNode list ... ``` ### LLMClient (for GraphCompactor) ```python from breathe.interfaces import LLMClient class AnthropicClient(LLMClient): def __init__(self, api_key: str): import anthropic self._client = anthropic.AsyncAnthropic(api_key=api_key) async def complete(self, prompt, max_tokens=4000, temperature=0.2): msg = await self._client.messages.create( model="claude-sonnet-4-20250514", max_tokens=max_tokens, temperature=temperature, messages=[{"role": "user", "content": prompt}], ) return msg.content[0].text class OpenAIClient(LLMClient): async def complete(self, prompt, max_tokens=4000, temperature=0.2): from openai import AsyncOpenAI client = AsyncOpenAI() resp = await client.chat.completions.create( model="gpt-4o", max_tokens=max_tokens, temperature=temperature, messages=[{"role": "user", "content": prompt}], ) return resp.choices[0].message.content ``` --- ## Performance Measured in production on Apple M2 Max: | Component | Latency | Notes | |------------------------|-------------|-------| | Regex extraction | 2ms | always runs | | MLX model extraction | ~250ms | conditional (when regex < 5 matches) | | Graph BFS (PG) | 5–15ms | recursive CTE, depth=2 | | Vector search (pgvector)| 10–30ms | depends on index size | | Keyword search (ILIKE) | 3–10ms | depends on table size | | **Total SYNAPSE** | **2–60ms** | without model | | **Total SYNAPSE** | **~300ms** | with model | | GraphCompactor | 3–8s | one LLM call, happens rarely | GraphCompactor fires infrequently (only at ~80% context fill), so its latency doesn't affect per-request response time. --- ## Memory management BREATHE handles **retrieval and injection** automatically. **Storing memories is your application's responsibility** — you decide what to remember and when. ```python # Your application stores memories explicitly await store.store("User prefers dark mode and concise answers") await store.store("Project uses FastAPI + PostgreSQL + Redis stack") # SYNAPSE retrieves relevant ones automatically before each LLM call messages = await synapse.inject(messages) ``` This is intentional: memory storage policies (what to keep, when to forget, privacy rules) vary wildly between applications. BREATHE gives you the retrieval engine — you control the data. > **Coming soon:** A standalone MCP server wrapping Memory Nexus, so LLMs can store and search memories directly as tool calls. --- ## Optional dependencies ```bash # PostgreSQL + pgvector backend pip install breathe-memory[pg] # Apple Silicon local model extractor (MLX) pip install breathe-memory[mlx] # Anthropic client for GraphCompactor pip install breathe-memory[anthropic] # OpenAI client for GraphCompactor pip install breathe-memory[openai] # Everything pip install breathe-memory[all] ``` Core package has zero dependencies beyond Python stdlib + `typing-extensions`. ### Model extractor (Phase 3) The optional `ModelAnchorExtractor` uses [MLX](https://github.com/ml-explore/mlx) to run a small local LLM for contextual anchor extraction when regex alone isn't enough. **This requires Apple Silicon (M1/M2/M3/M4).** MLX is an Apple-only framework and will not work on Linux or Windows. If MLX is not installed, the model extractor is silently skipped — everything else works normally. The default model is `Qwen3-1.7B` (4-bit, ~1.2 GB RAM). You can swap it for any MLX-compatible model by passing `model_id` to `ModelAnchorExtractor`. If you need cross-platform model extraction, implement your own extractor using any inference backend (ollama, vLLM, API calls) — the interface is a single `extract(message) -> list[Anchor]` method. --- ## Monitoring ```python from breathe import BreatheMetrics stats = BreatheMetrics.get().to_dict() # { # "synapse": { # "total_injections": 142, # "hit_rate": 0.87, # "latency": {"avg_ms": 18.3, "p95_ms": 45.1}, # "top_anchors": [{"text": "FastAPI", "count": 23}, ...] # }, # "compaction": { # "total": 3, # "avg_ratio": 0.71, # "total_saved_tokens": 12400 # } # } ``` Expose via your API: `GET /api/breathe-stats` → `BreatheMetrics.get().to_dict()` --- ## License Apache 2.0 — see [LICENSE](LICENSE). Built by [Kenaz GmbH](https://kenaz.ai) — Custom AI Agents, MCP Servers, Semantic Engineering. fx9100644 __init__.pyz`oTH%*Ξ100644 anchor_extractor.py`!_ D՛)֨100644 config.pyAdCߎT100644 context_injector.py\5Pit%zG100644 graph_compactor.py-hSrpWi100644 interfaces.pyx4gOS o=y40000 lang'^WδO100644 metrics.py\ԲYOY]N>NJ100644 model_extractor.py题30B &e[)q100644 session_graph.pyWL:_MǸ,A3 list[Anchor]: return [a for a in self.anchors if a.anchor_type == "entity"] @property def has_temporal(self) -> bool: return any(a.anchor_type == "temporal" for a in self.anchors) @property def node_ids(self) -> list[str]: """UUIDs of matched memory_nodes.""" return [a.matched_node_id for a in self.anchors if a.matched_node_id] # --- Shared technical pattern (language-agnostic) --- TECH_PATTERN = re.compile( r"(?:" r"[a-zA-Z_]\w+\.(?:py|ts|tsx|js|yaml|json|md|sql|sh)|" # file paths r"(?:def |class |function |const |import |from )\w+|" # code keywords r"(?:localhost:\d+|https?://\S+)|" # URLs r"(?:MCP|API|SSE|JWT|OAuth|CORS|FastAPI|React|PostgreSQL|Redis)\b" # tech terms r")", re.IGNORECASE, ) DATE_PATTERN = re.compile( r"\b(\d{1,2}[./]\d{1,2}[./]\d{2,4}|\d{4}-\d{2}-\d{2}|" r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)\w*\s+\d{4})\b", re.IGNORECASE, ) class AnchorExtractor: """ Extract associative anchors from user message. Phase 1: Regex + known concepts dictionary. The extractor loads known concepts from MemoryRepository at init and matches them against incoming messages. Supports multiple language packs — temporal and emotional patterns are loaded from the configured BreatheConfig. """ def __init__( self, known_concepts: Optional[dict[str, str]] = None, temporal_patterns: Optional[list[re.Pattern]] = None, emotional_patterns: Optional[list[re.Pattern]] = None, ): """ Args: known_concepts: ``{concept_text: node_uuid}`` dict from MemoryRepository. Loaded once at session start via Synapse.initialize(). temporal_patterns: Compiled regex patterns for temporal references. Defaults to English patterns. emotional_patterns: Compiled regex patterns for emotional language. Defaults to English patterns. """ self.known_concepts = known_concepts or {} # Build fast regex from known concepts (sorted by length for greedy match) if self.known_concepts: escaped = [ re.escape(c) for c in sorted(self.known_concepts.keys(), key=len, reverse=True) ] self._concept_pattern: Optional[re.Pattern] = re.compile( r"\b(" + "|".join(escaped) + r")\b", re.IGNORECASE, ) else: self._concept_pattern = None # Default to EN patterns if none supplied if temporal_patterns is None: from .lang.en import TEMPORAL_PATTERN as EN_T temporal_patterns = [EN_T] if emotional_patterns is None: from .lang.en import EMOTIONAL_PATTERN as EN_E emotional_patterns = [EN_E] self._temporal_patterns = temporal_patterns self._emotional_patterns = emotional_patterns def extract( self, message: str, conversation_tail: Optional[list[str]] = None, ) -> AnchorResult: """ Extract anchors from user message + optional conversation tail. Args: message: Current user message text. conversation_tail: Last 2–3 messages for context (strings). Returns: AnchorResult with extracted anchors and detected conversation mode. """ result = AnchorResult(raw_text=message) # Combine message with tail for broader concept matching full_text = message if conversation_tail: full_text = message + " " + " ".join(conversation_tail[-3:]) # 1. Match known concepts (highest confidence — direct graph entry points) if self._concept_pattern: for match in self._concept_pattern.finditer(full_text): concept_text = match.group(0).lower() node_id = next( (uid for key, uid in self.known_concepts.items() if key.lower() == concept_text), None, ) result.anchors.append(Anchor( text=match.group(0), anchor_type="entity", confidence=0.9, source="known_concept", matched_node_id=node_id, )) # 2. Temporal anchors for pattern in (*self._temporal_patterns, DATE_PATTERN): for match in pattern.finditer(message): result.anchors.append(Anchor( text=match.group(0), anchor_type="temporal", confidence=0.7, )) # 3. Technical anchors (message only — tail brings too much noise) for match in TECH_PATTERN.finditer(message): text = match.group(0).strip() if len(text) > 2: result.anchors.append(Anchor( text=text, anchor_type="technical", confidence=0.5, )) # 4. Emotional anchors emotional_count = 0 for pattern in self._emotional_patterns: for match in pattern.finditer(message): emotional_count += 1 result.anchors.append(Anchor( text=match.group(0), anchor_type="emotional", confidence=0.6, )) # Exclamation density as emotional signal if message.count("!") / max(len(message), 1) > 0.02: emotional_count += 1 result.conversation_mode = self._detect_mode(message, result.anchors, emotional_count) result.anchors = self._deduplicate(result.anchors) logger.info( f"Anchors: {len(result.anchors)} extracted " f"({len(result.node_ids)} matched nodes), " f"mode={result.conversation_mode}" ) return result def _detect_mode(self, message: str, anchors: list[Anchor], emotional_count: int) -> str: tech_count = sum(1 for a in anchors if a.anchor_type == "technical") if tech_count >= 3: return "work" if emotional_count >= 2: return "deep" if len(message) < 100 and tech_count == 0: return "casual" return "balanced" @staticmethod def _deduplicate(anchors: list[Anchor]) -> list[Anchor]: """Keep highest-confidence anchor for each unique text (case-insensitive).""" seen: dict[str, Anchor] = {} for anchor in anchors: key = anchor.text.lower() if key not in seen or anchor.confidence > seen[key].confidence: seen[key] = anchor return list(seen.values()) x """ BreatheConfig — unified configuration for BREATHE. All tuneable parameters live here. Language packs plug in as dicts, making BREATHE easy to extend to any language without changing core logic. """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Optional from .lang.en import ( STOPWORDS as EN_STOPWORDS, HUB_EXCLUSIONS as EN_HUB_EXCLUSIONS, LABELS as EN_LABELS, TEMPORAL_PATTERN as EN_TEMPORAL, EMOTIONAL_PATTERN as EN_EMOTIONAL, ) @dataclass class LanguagePack: """ Language configuration for anchor extraction and context injection. To add a new language, create a LanguagePack and pass it to BreatheConfig. Example:: from breathe.config import LanguagePack, BreatheConfig import re my_pack = LanguagePack( code="de", stopwords={"der", "die", "das", "und", "ist", ...}, hub_exclusions={"claude", "speicher"}, temporal_pattern=re.compile(r"\\b(gestern|heute|morgen)\\b", re.I), emotional_pattern=re.compile(r"\\b(traurig|glücklich|wütend)\\b", re.I), labels={"themes": "Themen", "insights": "Erkenntnisse"}, ) config = BreatheConfig(language_packs=[my_pack], default_language="de") """ code: str stopwords: frozenset[str] hub_exclusions: frozenset[str] temporal_pattern: re.Pattern emotional_pattern: re.Pattern labels: dict[str, str] # Pre-built packs ENGLISH = LanguagePack( code="en", stopwords=EN_STOPWORDS, hub_exclusions=EN_HUB_EXCLUSIONS, temporal_pattern=EN_TEMPORAL, emotional_pattern=EN_EMOTIONAL, labels=EN_LABELS, ) @dataclass class BreatheConfig: """ Master configuration for BREATHE. All settings have sensible defaults. The most common customizations are: - ``language_packs``: which languages to support (default: EN) - ``default_language``: primary language for UI strings - ``min_similarity``: threshold for vector search results (0–1) - ``max_injected_nodes``: upper limit on nodes per injection (default 15) - ``enable_model_extractor``: whether to use local MLX model (default True) Token budgets control how much memory is injected per conversation mode. Adjust them based on your model's context window and use case. """ # --- Language --- language_packs: list[LanguagePack] = field( default_factory=lambda: [ENGLISH] ) default_language: str = "en" # --- SYNAPSE --- min_similarity: float = 0.55 """Minimum vector similarity score to accept (below = noise).""" max_injected_nodes: int = 15 """Maximum nodes per injection pass.""" enable_model_extractor: bool = True """Whether to use local MLX model for enhanced anchor extraction (Phase 3).""" model_trigger_threshold: int = 5 """ Model extractor fires when regex finds fewer matched nodes than this. Lower = model runs more often (slower but richer extraction). """ # --- Token budgets by conversation mode --- mode_budgets: dict[str, int] = field( default_factory=lambda: { "casual": 1500, "work": 2500, "deep": 4000, "balanced": 2000, } ) # --- GraphCompactor --- compactor_model: str = "claude-sonnet-4-6" compactor_fallback_model: str = "claude-haiku-4-5-20251001" min_tokens_to_compress: int = 300 protected_messages_normal: int = 10 protected_messages_with_code: int = 5 # --- Metrics --- metrics_history_size: int = 200 """How many events to keep in the rolling metrics window.""" # --- Computed (built from language_packs) --- @property def stopwords(self) -> frozenset[str]: """Union of stopwords from all configured language packs.""" combined: set[str] = set() for pack in self.language_packs: combined |= pack.stopwords return frozenset(combined) @property def hub_exclusions(self) -> frozenset[str]: """Union of hub exclusions from all configured language packs.""" combined: set[str] = set() for pack in self.language_packs: combined |= pack.hub_exclusions return frozenset(combined) @property def labels(self) -> dict[str, str]: """UI labels from the default language pack.""" for pack in self.language_packs: if pack.code == self.default_language: return pack.labels return self.language_packs[0].labels if self.language_packs else {} def get_pack(self, code: str) -> Optional[LanguagePack]: """Return language pack by code, or None if not configured.""" for pack in self.language_packs: if pack.code == code: return pack return None !x""" Context Injector — formats retrieved subgraph into injection text. Takes nodes from SYNAPSE traversal and produces a structured text block for context window injection. The format uses tags so the LLM perceives this as remembered context, not tool output — "I remember" not "I was told." """ from __future__ import annotations import logging from typing import Optional from .interfaces import RetrievedNode logger = logging.getLogger(__name__) # Default token budgets per conversation mode. # Override via BreatheConfig.mode_budgets. DEFAULT_MODE_BUDGETS: dict[str, int] = { "casual": 1500, "work": 2500, "deep": 4000, "balanced": 2000, } class ContextInjector: """ Formats retrieved nodes into an injection text block. Respects per-mode token budgets. Output is structured but natural — the LLM should feel this as memory, not structured data. Labels (e.g. "Themes", "Insights") come from the active language pack. Pass a ``labels`` dict to override defaults. """ def __init__( self, mode_budgets: Optional[dict[str, int]] = None, labels: Optional[dict[str, str]] = None, memory_tag: str = "associative_memory", ): """ Args: mode_budgets: Token budget per conversation mode. Defaults to DEFAULT_MODE_BUDGETS. labels: UI strings for section headers. Keys: ``"themes"``, ``"insights"``. Defaults to English. memory_tag: XML tag wrapping the injection block. """ self._budgets = mode_budgets or DEFAULT_MODE_BUDGETS self._labels = labels or {"themes": "Themes", "insights": "Insights"} self._memory_tag = memory_tag def format_injection( self, nodes: list[RetrievedNode], mode: str = "balanced", anchors_text: Optional[list[str]] = None, ) -> Optional[str]: """ Format nodes into injection text. Args: nodes: Retrieved nodes from SYNAPSE traversal. mode: Conversation mode for budget selection. anchors_text: Original anchor texts (unused currently, reserved for future "Connection to now" section). Returns: Formatted injection string, or None if nothing to inject. """ if not nodes: return None budget_tokens = self._budgets.get(mode, 2000) budget_chars = budget_tokens * 4 # rough char-to-token ratio sections: list[str] = [] # Group nodes by type entities = [n for n in nodes if n.node_type == "entity"] themes = [n for n in nodes if n.node_type == "theme"] insights = [n for n in nodes if n.node_type == "insight"] events = [n for n in nodes if n.node_type == "event"] # Events and entities carry the richest context for node in events[:3]: section = self._format_event(node) if section: sections.append(section) if entities: lines = [] for node in entities[:5]: line = f"**{node.concept}**" if node.summary: line += f" — {node.summary}" if node.relation: line += f" ({node.relation})" lines.append(line) sections.append("\n".join(lines)) if themes: theme_label = self._labels.get("themes", "Themes") theme_lines = [f"{theme_label}: {', '.join(n.concept for n in themes[:4])}"] for node in themes[:2]: if node.summary: theme_lines.append(f"- {node.concept}: {node.summary}") sections.append("\n".join(theme_lines)) if insights: insight_label = self._labels.get("insights", "Insights") insight_lines = [ f"- {node.summary or node.concept}" for node in insights[:3] ] sections.append(f"{insight_label}:\n" + "\n".join(insight_lines)) # Raw memory content (from vector/keyword search) memories_with_content = [n for n in nodes if n.memory_content] for node in memories_with_content[:3]: content = node.memory_content if len(content) > 1500: content = content[:1500] + "..." sections.append(content) if not sections: return None body = "\n\n".join(sections) if len(body) > budget_chars: body = body[:budget_chars] + "\n[...]" injection = f"<{self._memory_tag}>\n{body}\n" logger.info( f"Injection: {len(nodes)} nodes → {len(injection)} chars " f"(budget: {budget_chars} chars, mode: {mode})" ) return injection @staticmethod def _format_event(node: RetrievedNode) -> Optional[str]: parts = [f"### {node.concept}"] if node.summary: parts.append(node.summary) if node.memory_content: content = node.memory_content if len(content) > 1500: content = content[:1500] + "..." parts.append(content) return "\n".join(parts) if len(parts) > 1 else None E_x'%""" GraphCompactor — structured graph extraction for context compression. The exhale of BREATHE. Fires when the context window approaches its limit. Instead of a lossy narrative summary, extracts a structured graph of topics, decisions, open questions, and artifacts. The LLM decides what matters. Design principle: LLM memory is LLM's decision, not a summarizer's interpretation. """ from __future__ import annotations import logging import re import time from datetime import datetime from typing import Optional from .interfaces import LLMClient from .session_graph import SessionGraph from .metrics import BreatheMetrics, CompactionEvent logger = logging.getLogger(__name__) TOOL_USE_PATTERN = re.compile(r"\[tool_use\]\s*(\S+)\[/tool_use\]") TOOL_RESULT_PATTERN = re.compile(r"\[tool_result\](.*?)\[/tool_result\]", re.DOTALL) EXTRACTION_PROMPT = """You are extracting a structured context graph from a conversation. This replaces linear compression. You decide what matters. Be selective, not exhaustive. From the messages below, extract: ## Topics [weight 0.0-1.0] Active topics of conversation. Weight = how central they are right now. Format: - [topic_name] [weight] description | connected_topics ## Decisions What was decided or concluded. One line each. Format: - decision text ## Open Unresolved questions or pending items. Format: - question or pending item ## Artifacts Files, code, configs created or modified. Format: - name: what it is ## Context Emotional state, situational context that affects interpretation. Format: - context note ## Dropped What you chose NOT to preserve, and why (one line). This makes the choice conscious. Format: - what was dropped (why) IMPORTANT: - Tool calls: drop the raw JSON. Keep only "searched X → found Y" or "modified file Z". - Keep the language of the original conversation. - Be concise. This must fit in ~10k tokens. - Weight topics by current relevance, not chronological order. Messages to extract from: """ class GraphCompactor: """ Drop-in replacement for any narrative summarizer. Compresses older conversation history into a structured graph instead of a lossy prose summary. Preserves the semantic structure of what happened. Usage:: compactor = GraphCompactor(llm_client=my_llm_client) result = await compactor.compress(messages) if result["compressed"]: messages = result["compressed_messages"] """ def __init__( self, llm_client: LLMClient, min_tokens_to_compress: int = 300, protected_messages_normal: int = 10, protected_messages_with_code: int = 5, ): """ Args: llm_client: LLM client for graph extraction. Use AnthropicLLMClient or implement LLMClient for any other provider. min_tokens_to_compress: Skip compression if older messages have fewer tokens than this threshold. protected_messages_normal: How many recent assistant turns to keep intact (not compressed). protected_messages_with_code: Protected turns when code/tools detected. """ self._llm = llm_client self.MIN_TOKENS_TO_COMPRESS = min_tokens_to_compress self.PROTECTED_MESSAGES_NORMAL = protected_messages_normal self.PROTECTED_MESSAGES_WITH_CODE = protected_messages_with_code @staticmethod def _count_tokens_rough(text: str) -> int: return len(text) // 4 @staticmethod def _has_tool_calls(messages: list) -> bool: recent = messages[-10:] if len(messages) > 10 else messages return any( "```" in msg.get("content", "") or "[tool_use]" in msg.get("content", "") or "def " in msg.get("content", "") for msg in recent ) def _split_messages(self, messages: list) -> tuple[list, list]: """Split into compressible (older) and protected (recent) zones.""" has_code = self._has_tool_calls(messages) protected_count = ( self.PROTECTED_MESSAGES_WITH_CODE if has_code else self.PROTECTED_MESSAGES_NORMAL ) assistant_indices = [ i for i, msg in enumerate(messages) if msg.get("role") == "assistant" ] if len(assistant_indices) <= protected_count: return [], messages protected_start_idx = assistant_indices[-protected_count] return messages[:protected_start_idx], messages[protected_start_idx:] @staticmethod def _preprocess_tool_calls(messages: list) -> list: """Compress raw tool call JSON before sending to extraction.""" processed = [] for msg in messages: content = msg.get("content", "") content = TOOL_USE_PATTERN.sub(r"[tool: \1]", content) def compress_result(match: re.Match) -> str: result_text = match.group(1).strip() if len(result_text) > 200: return f"[result: {result_text[:200]}...]" return f"[result: {result_text}]" content = TOOL_RESULT_PATTERN.sub(compress_result, content) if content.strip(): processed.append({**msg, "content": content}) return processed @staticmethod def _format_for_extraction(messages: list) -> str: lines = [] for msg in messages: role = msg.get("role", "unknown") content = msg.get("content", "") if len(content) > 3000: content = content[:3000] + "... [truncated]" lines.append(f"[{role}]: {content}") return "\n\n".join(lines) async def compress( self, messages: list, conversation_style: str = "balanced" ) -> dict: """ Extract structured graph from compressible messages. Args: messages: Full conversation messages list. conversation_style: Conversation style hint (unused currently, reserved for future extraction tuning). Returns: Dict with keys: - ``compressed``: bool - ``compressed_messages``: list (use this to replace ``messages``) - ``messages``: same as compressed_messages (alias) - ``metadata``: compression stats - ``session_graph``: SessionGraph instance (if compressed) """ compressible, protected = self._split_messages(messages) if not compressible or len(compressible) < 3: return self._no_op(messages, reason="too_few_messages") original_tokens = sum( self._count_tokens_rough(msg.get("content", "")) for msg in compressible ) if original_tokens < self.MIN_TOKENS_TO_COMPRESS: return self._no_op(messages, reason="below_threshold") compact_start = time.monotonic() logger.info( f"Graph compaction: {len(compressible)} messages, " f"~{original_tokens} tokens → extracting..." ) processed = self._preprocess_tool_calls(compressible) formatted = self._format_for_extraction(processed) prompt = EXTRACTION_PROMPT + formatted extracted_text = await self._llm.complete(prompt) if not extracted_text: logger.error("Graph extraction failed, returning original messages") return self._no_op(messages, reason="extraction_failed") session_graph = SessionGraph.from_structured_text(extracted_text) logger.info(f"Extracted graph: {session_graph}") compressed_tokens = self._count_tokens_rough(extracted_text) saved_tokens = original_tokens - compressed_tokens compression_ratio = ( 1 - (compressed_tokens / original_tokens) if original_tokens > 0 else 0 ) compressed_message = { "role": "system", "content": ( f"[SESSION GRAPH — extracted from {len(compressible)} messages]\n\n" f"{extracted_text}" ), } final_messages = [compressed_message] + protected compact_ms = (time.monotonic() - compact_start) * 1000 BreatheMetrics.get().record_compaction(CompactionEvent( timestamp=time.time(), original_tokens=original_tokens, compressed_tokens=compressed_tokens, saved_tokens=saved_tokens, compression_ratio=compression_ratio, graph_nodes=session_graph.node_count, graph_edges=session_graph.edge_count, strategy=f"graph_{conversation_style}", fallback_used=False, duration_ms=compact_ms, extracted_text=extracted_text[:3000] if extracted_text else "", )) logger.info( f"Graph compaction complete: {original_tokens} → {compressed_tokens} tokens " f"(saved {saved_tokens}, {compression_ratio:.0%}), " f"{session_graph.node_count} nodes, {session_graph.edge_count} edges" ) return { "compressed": True, "messages": final_messages, "compressed_messages": final_messages, "session_graph": session_graph, "metadata": { "original_messages": len(compressible), "original_tokens": original_tokens, "compressed_tokens": compressed_tokens, "saved_tokens": saved_tokens, "compression_ratio": round(compression_ratio, 2), "protected_messages": len(protected), "strategy": f"graph_{conversation_style}", "graph_nodes": session_graph.node_count, "graph_edges": session_graph.edge_count, "compressed_at": datetime.utcnow().isoformat(), }, } @staticmethod def _no_op(messages: list, reason: str) -> dict: return { "compressed": False, "messages": messages, "compressed_messages": messages, "metadata": {"reason": reason}, } xqxP""" Interfaces — abstract contracts for external dependencies. BREATHE is storage-agnostic and LLM-agnostic by design. Implement these interfaces to integrate with any backend. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Optional @dataclass class RetrievedNode: """A node retrieved from graph traversal or memory search.""" node_id: str concept: str node_type: str # entity, theme, insight, event, memory_vector, memory_keyword summary: Optional[str] = None importance: float = 0.5 depth: int = 0 # BFS depth from anchor relation: Optional[str] = None # edge relation that led here memory_content: Optional[str] = None # raw text from memory store class MemoryRepository(ABC): """ Abstract storage backend for BREATHE. Implement this to connect BREATHE to your own database. See ``breathe.backends.postgres`` for the asyncpg/PostgreSQL reference implementation. The interface deliberately stays minimal — you only need what SYNAPSE actually calls. Graph BFS and keyword search are optional; implement the ones you need and raise ``NotImplementedError`` for the rest. """ @abstractmethod async def get_concepts(self) -> dict[str, str]: """ Return all active known concepts as ``{concept_text: node_uuid}``. Called once at initialization to build the concept regex. An empty dict is valid — SYNAPSE falls back to regex-only extraction. """ @abstractmethod async def graph_bfs( self, start_ids: list[str], max_depth: int = 2, min_strength: float = 0.2, limit: int = 20, ) -> list[RetrievedNode]: """ BFS traversal from ``start_ids`` through the concept graph. Args: start_ids: UUIDs of matched memory_nodes. max_depth: Maximum edge hops to follow. min_strength: Minimum edge strength to follow (0–1). limit: Maximum nodes to return. Returns: List of RetrievedNode, sorted by importance descending. """ @abstractmethod async def keyword_search( self, keywords: list[str], limit: int = 5 ) -> list[RetrievedNode]: """ Full-text keyword search over memory content. Called for anchors that didn't match any known concept (no node_id). ILIKE or equivalent is fine — precision matters less than recall here. Args: keywords: Words to search for (case-insensitive). limit: Maximum memories to return. """ async def flush_edges(self, edges: list) -> int: """ Persist new session graph edges to long-term storage. Called at session end by SessionGraph.flush(). Optional — if you don't need cross-session graph persistence, leave this as a no-op. Returns number of edges flushed. """ return 0 class VectorSearchClient(ABC): """ Abstract client for semantic / vector search. Wraps any dense-embedding search backend (pgvector, Pinecone, Weaviate, etc.). BREATHE uses this for Strategy 2 in SYNAPSE traversal — the highest-quality but most expensive retrieval path. """ @abstractmethod async def search(self, query: str, limit: int = 5) -> list[RetrievedNode]: """ Return the most semantically similar memories for ``query``. Args: query: Short anchor phrase (NOT the full user message). limit: Max results. Returns: List of RetrievedNode sorted by similarity descending. Set ``importance`` to the similarity score (0–1). """ class LLMClient(ABC): """ Abstract LLM client for GraphCompactor. GraphCompactor needs a single call: given a long prompt, return text. Implement this to use any LLM (Anthropic, OpenAI, local, etc.). """ @abstractmethod async def complete( self, prompt: str, max_tokens: int = 4000, temperature: float = 0.2, ) -> Optional[str]: """ Generate a completion for ``prompt``. Args: prompt: The full extraction prompt (can be long). max_tokens: Max tokens to generate. temperature: Low values (0.1–0.3) work best for structured extraction. Returns: Generated text, or None on failure. """ cxH100644 __init__.py6{>uA{M 100644 en.py @9=[WKxg"""Language packs for BREATHE anchor extraction and injection.""" from .en import ( STOPWORDS as EN_STOPWORDS, HUB_EXCLUSIONS as EN_HUB_EXCLUSIONS, LABELS as EN_LABELS, TEMPORAL_PATTERN as EN_TEMPORAL, EMOTIONAL_PATTERN as EN_EMOTIONAL, ) __all__ = [ "EN_STOPWORDS", "EN_HUB_EXCLUSIONS", "EN_LABELS", "EN_TEMPORAL", "EN_EMOTIONAL", ] ,dcox"""English language pack for BREATHE.""" import re STOPWORDS: frozenset[str] = frozenset({ # Articles, prepositions, conjunctions "the", "and", "for", "that", "this", "with", "from", "have", "has", "was", "were", "been", "being", "are", "not", "but", "they", "them", "their", "what", "which", "when", "where", "how", "can", "will", "would", "should", "could", "just", "also", "more", "some", "than", "into", "about", "over", "after", "before", "between", "through", "during", "without", "again", "here", "there", "then", "now", "very", "really", "quite", "still", "already", "never", "always", "each", "every", "other", "another", "such", "only", "even", # Common verbs that match everything "work", "make", "use", "get", "set", "run", "add", "see", "say", "need", "want", "know", "think", "look", "come", "give", "take", # Generic nouns "time", "thing", "way", "day", "part", "point", "place", "case", "state", "fact", "line", "end", "start", "step", "type", "kind", }) # Hub node names — too generic to be meaningful in retrieval HUB_EXCLUSIONS: frozenset[str] = frozenset({ "claude", "memory", "atlas", }) # UI labels LABELS: dict[str, str] = { "themes": "Themes", "insights": "Insights", "associative_memory_tag": "associative_memory", } # Temporal patterns (EN) TEMPORAL_PATTERN = re.compile( r"\b(yesterday|today|tomorrow|last week|this week|last month|" r"recently|again|same as|remember when|like before|back then)\b", re.IGNORECASE, ) # Emotional patterns (EN) EMOTIONAL_PATTERN = re.compile( r"\b(tired|hurts|headache|frustrated|angry|annoyed|" r"happy|amazing|awesome|fuck|shit|damn|" r"sad|miss|worried|anxious|" r"love|hug|gentle|warm)\b", re.IGNORECASE, ) #x """ BREATHE Metrics — in-memory analytics for SYNAPSE + GraphCompactor. Singleton collector. Thread-safe via simple dict/list operations. No external dependencies. Expose via your API to power a real-time monitoring dashboard. """ from __future__ import annotations import time import logging from dataclasses import dataclass, field from typing import Optional logger = logging.getLogger(__name__) MAX_HISTORY = 200 @dataclass class AnchorDetail: text: str anchor_type: str # entity, temporal, technical, emotional confidence: float source: str # regex, known_concept, model matched_node_id: Optional[str] = None @dataclass class NodeDetail: concept: str node_type: str summary: Optional[str] = None importance: float = 0.5 depth: int = 0 relation: Optional[str] = None @dataclass class SynapseEvent: timestamp: float anchors_count: int matched_nodes: int injected_nodes: int regex_ms: float model_ms: float = 0.0 total_ms: float = 0.0 mode: str = "balanced" model_triggered: bool = False user_message: str = "" anchors: list[AnchorDetail] = field(default_factory=list) nodes: list[NodeDetail] = field(default_factory=list) injection_text: str = "" @dataclass class CompactionEvent: timestamp: float original_tokens: int compressed_tokens: int saved_tokens: int compression_ratio: float graph_nodes: int graph_edges: int strategy: str = "graph" fallback_used: bool = False duration_ms: float = 0.0 extracted_text: str = "" class BreatheMetrics: """ In-memory metrics collector for BREATHE. Singleton pattern — one instance shared across Synapse and GraphCompactor. Access via ``BreatheMetrics.get()``. Serialize to dict for API exposure via ``to_dict()``. """ _instance: Optional["BreatheMetrics"] = None def __init__(self, history_size: int = MAX_HISTORY): self._history_size = history_size # SYNAPSE self.synapse_events: list[SynapseEvent] = [] self.synapse_total: int = 0 self.synapse_skipped: int = 0 # Compaction self.compaction_events: list[CompactionEvent] = [] self.compaction_total: int = 0 self.compaction_fallbacks: int = 0 # State self.known_concepts_count: int = 0 self.model_triggers: int = 0 self.model_available: bool = False # Graph flush self.flush_count: int = 0 self.flush_edges_total: int = 0 # Anchor frequency self._anchor_counts: dict[str, int] = {} self.started_at: float = time.time() @classmethod def get(cls) -> "BreatheMetrics": if cls._instance is None: cls._instance = cls() return cls._instance @classmethod def reset(cls) -> None: """Reset singleton (useful for testing).""" cls._instance = None def record_synapse(self, event: SynapseEvent) -> None: self.synapse_events.append(event) if len(self.synapse_events) > self._history_size: self.synapse_events = self.synapse_events[-self._history_size:] self.synapse_total += 1 def record_synapse_skip(self) -> None: self.synapse_skipped += 1 def record_anchors(self, anchor_texts: list[str]) -> None: for text in anchor_texts: key = text.lower() self._anchor_counts[key] = self._anchor_counts.get(key, 0) + 1 def record_compaction(self, event: CompactionEvent) -> None: self.compaction_events.append(event) if len(self.compaction_events) > self._history_size: self.compaction_events = self.compaction_events[-self._history_size:] self.compaction_total += 1 if event.fallback_used: self.compaction_fallbacks += 1 def record_flush(self, edges_count: int) -> None: self.flush_count += 1 self.flush_edges_total += edges_count def to_dict(self) -> dict: """Export metrics as JSON-serializable dict.""" now = time.time() uptime_s = now - self.started_at synapse_latencies = [e.total_ms for e in self.synapse_events] regex_latencies = [e.regex_ms for e in self.synapse_events] model_latencies = [e.model_ms for e in self.synapse_events if e.model_triggered] comp_ratios = [e.compression_ratio for e in self.compaction_events] comp_saved = [e.saved_tokens for e in self.compaction_events] total = self.synapse_total + self.synapse_skipped hit_rate = self.synapse_total / total if total > 0 else 0.0 top_anchors = sorted( self._anchor_counts.items(), key=lambda x: x[1], reverse=True )[:15] recent_synapse = [ { "timestamp": e.timestamp, "anchors": e.anchors_count, "nodes": e.injected_nodes, "total_ms": round(e.total_ms, 1), "mode": e.mode, "model_used": e.model_triggered, "user_message": e.user_message, "anchors_detail": [ { "text": a.text, "type": a.anchor_type, "confidence": a.confidence, "source": a.source, "matched": a.matched_node_id is not None, } for a in e.anchors ], "nodes_detail": [ { "concept": n.concept, "type": n.node_type, "summary": n.summary, "importance": n.importance, "depth": n.depth, "relation": n.relation, } for n in e.nodes ], "injection_text": e.injection_text, } for e in self.synapse_events[-20:] ] recent_compactions = [ { "timestamp": e.timestamp, "original_tokens": e.original_tokens, "compressed_tokens": e.compressed_tokens, "ratio": round(e.compression_ratio, 2), "nodes": e.graph_nodes, "edges": e.graph_edges, "fallback": e.fallback_used, "duration_ms": round(e.duration_ms, 1), "extracted_text": e.extracted_text, } for e in self.compaction_events[-10:] ] return { "uptime_seconds": round(uptime_s), "synapse": { "total_injections": self.synapse_total, "total_skipped": self.synapse_skipped, "hit_rate": round(hit_rate, 3), "known_concepts": self.known_concepts_count, "model_available": self.model_available, "model_triggers": self.model_triggers, "latency": { "avg_ms": round(_avg(synapse_latencies), 1), "p50_ms": round(_percentile(synapse_latencies, 0.5), 1), "p95_ms": round(_percentile(synapse_latencies, 0.95), 1), "max_ms": round(max(synapse_latencies), 1) if synapse_latencies else 0, "regex_avg_ms": round(_avg(regex_latencies), 1), "model_avg_ms": round(_avg(model_latencies), 1) if model_latencies else 0, }, "recent": recent_synapse, }, "compaction": { "total": self.compaction_total, "fallbacks": self.compaction_fallbacks, "avg_ratio": round(_avg(comp_ratios), 2) if comp_ratios else 0, "total_saved_tokens": sum(comp_saved), "recent": recent_compactions, }, "graph": { "flush_count": self.flush_count, "flush_edges_total": self.flush_edges_total, "top_anchors": [ {"text": text, "count": count} for text, count in top_anchors ], }, } def _avg(values: list[float]) -> float: return sum(values) / len(values) if values else 0.0 def _percentile(values: list[float], p: float) -> float: if not values: return 0.0 sorted_vals = sorted(values) idx = min(int(len(sorted_vals) * p), len(sorted_vals) - 1) return sorted_vals[idx] պ?x$""" Model-based Anchor Extractor — Phase 3 of BREATHE. Uses a local small language model (Qwen3-1.7B abliterated, MLX 4-bit) for contextual anchor extraction. Runs on Apple Silicon via MLX framework. Designed as an enhancement layer ON TOP of the regex extractor: - Regex is always fast (2ms) and catches known concepts - Model adds contextual understanding (~250ms) when regex is insufficient - Hybrid: regex always runs, model fires when regex finds bool: """Check if MLX is installed without loading the model.""" if self._available is None: try: import mlx_lm # noqa: F401 self._available = True except ImportError: self._available = False logger.debug("mlx_lm not installed — model extractor disabled") return self._available def _ensure_loaded(self) -> bool: if self._model is not None: return True if not self.available: return False try: from mlx_lm import load start = time.monotonic() self._model, self._tokenizer = load(self.model_id) elapsed = (time.monotonic() - start) * 1000 logger.info(f"Model extractor loaded: {self.model_id} ({elapsed:.0f}ms)") return True except Exception as e: logger.warning(f"Failed to load model extractor: {e}") self._available = False return False def extract(self, message: str, max_tokens: int = 100) -> list[Anchor]: """ Extract anchors from message using the local model. Args: message: User message text. max_tokens: Max generation tokens (keep low for speed). Returns: Validated list of Anchor objects. """ if not self._ensure_loaded(): return [] try: from mlx_lm import generate prompt = EXTRACTION_PROMPT.format(message=message[:500]) start = time.monotonic() raw = generate( self._model, self._tokenizer, prompt=prompt, max_tokens=max_tokens, verbose=False, ) elapsed = (time.monotonic() - start) * 1000 anchors = _parse_response(raw) anchors = _validate_against_message(anchors, message) logger.info(f"Model extraction: {len(anchors)} anchors in {elapsed:.0f}ms") return anchors except Exception as e: logger.warning(f"Model extraction failed: {e}") return [] def should_use_model( regex_result: AnchorResult, threshold: int = 5, ) -> bool: """ Decide whether to invoke the model based on regex results. Returns True when regex found fewer matched nodes than ``threshold`` (meaning the message likely contains concepts the regex doesn't know about). Also skips very short messages (greetings, "ok", etc.). """ if len(regex_result.raw_text) < 15: return False return len(regex_result.node_ids) < threshold def _parse_response(raw: str) -> list[Anchor]: anchors: list[Anchor] = [] raw = raw.strip() start_idx = raw.find("{") if start_idx == -1: return anchors depth = 0 end_idx = start_idx for i in range(start_idx, len(raw)): if raw[i] == "{": depth += 1 elif raw[i] == "}": depth -= 1 if depth == 0: end_idx = i + 1 break try: data = json.loads(raw[start_idx:end_idx]) except json.JSONDecodeError: return anchors type_map = {"entities": "entity", "themes": "theme", "emotional": "emotional"} for key, anchor_type in type_map.items(): items = data.get(key, []) if isinstance(items, str) and items: items = [items] if not isinstance(items, list): continue for item in items: text = "" if isinstance(item, str): text = item.strip() elif isinstance(item, dict): text = (item.get("value") or item.get("name") or item.get("text") or "").strip() if text and len(text) > 1: anchors.append(Anchor( text=text, anchor_type=anchor_type, confidence=0.7, source="model", )) return anchors def _validate_against_message(anchors: list[Anchor], message: str) -> list[Anchor]: """Drop anchors not actually present in the message (hallucination guard). Uses stem matching (first 4 chars) to handle morphological variants. """ msg_lower = message.lower() validated = [] for a in anchors: words = a.text.lower().split() found = True for w in words: if w in msg_lower: continue stem = w[:4] if len(w) >= 4 else w if stem in msg_lower: continue found = False break if found: validated.append(a) else: logger.debug(f"Hallucination dropped: '{a.text}' (not in message)") return validated 0@x,""" SessionGraph — in-memory graph for the current session's warm layer. Sits between hot context (full message history) and cold storage (vector DB). Built during compaction (Graph Compactor), traversed by SYNAPSE, flushed to persistent storage at session end. Design principle: the LLM decides what to record. No mechanical fixation. """ from __future__ import annotations import logging from collections import deque from dataclasses import dataclass, field from datetime import datetime from typing import Optional logger = logging.getLogger(__name__) @dataclass class GraphNode: """A node in the session graph — topic, decision, artifact, or state.""" id: str node_type: str # 'topic', 'decision', 'artifact', 'open_question', 'state' label: str # short display name content: str # full content / description weight: float = 0.5 created_at: datetime = field(default_factory=datetime.utcnow) last_activated: datetime = field(default_factory=datetime.utcnow) source_memory_id: Optional[str] = None def activate(self, boost: float = 0.1) -> None: """Strengthen node when referenced again.""" self.weight = min(1.0, self.weight + boost) self.last_activated = datetime.utcnow() def decay(self, factor: float = 0.95, floor: float = 0.1) -> None: """Exponential decay with floor. Nothing fully disappears.""" self.weight = max(floor, self.weight * factor) @dataclass class GraphEdge: """Directed edge between nodes.""" from_id: str to_id: str edge_type: str # 'relates_to', 'led_to', 'part_of', 'temporal_parallel', 'decided' weight: float = 0.5 created_at: datetime = field(default_factory=datetime.utcnow) session_id: Optional[str] = None class SessionGraph: """ In-memory graph for current session. Warm layer between hot (full text) and cold (vector DB). Built during compaction, traversed by SYNAPSE, flushed at session end. Flush to persistent storage is optional — pass a ``MemoryRepository`` to ``flush()`` if you want cross-session graph continuity. """ def __init__(self, session_id: Optional[str] = None): self.session_id = session_id self.nodes: dict[str, GraphNode] = {} self.edges: dict[str, list[GraphEdge]] = {} # from_id → [edges] self._new_edges: list[GraphEdge] = [] self._new_nodes: list[GraphNode] = [] @property def node_count(self) -> int: return len(self.nodes) @property def edge_count(self) -> int: return sum(len(edges) for edges in self.edges.values()) # --- Node operations --- def add_node(self, node: GraphNode) -> GraphNode: """Add a node. If it already exists, activate (strengthen) it.""" if node.id in self.nodes: existing = self.nodes[node.id] existing.activate() if len(node.content) > len(existing.content): existing.content = node.content return existing self.nodes[node.id] = node self._new_nodes.append(node) return node def get_node(self, node_id: str) -> Optional[GraphNode]: return self.nodes.get(node_id) def remove_node(self, node_id: str) -> bool: if node_id not in self.nodes: return False del self.nodes[node_id] self.edges.pop(node_id, None) for from_id in list(self.edges.keys()): self.edges[from_id] = [e for e in self.edges[from_id] if e.to_id != node_id] return True # --- Edge operations --- def add_edge( self, from_id: str, to_id: str, edge_type: str = "relates_to", weight: float = 0.5, ) -> Optional[GraphEdge]: """Connect two nodes. Strengthens existing edge if already present.""" if from_id not in self.nodes or to_id not in self.nodes: logger.warning(f"Cannot create edge: node(s) missing ({from_id} → {to_id})") return None for edge in self.edges.get(from_id, []): if edge.to_id == to_id and edge.edge_type == edge_type: edge.weight = min(1.0, edge.weight + 0.1) return edge edge = GraphEdge( from_id=from_id, to_id=to_id, edge_type=edge_type, weight=weight, session_id=self.session_id, ) self.edges.setdefault(from_id, []).append(edge) self._new_edges.append(edge) return edge def drop_edge(self, from_id: str, to_id: str, edge_type: Optional[str] = None) -> bool: if from_id not in self.edges: return False before = len(self.edges[from_id]) if edge_type: self.edges[from_id] = [ e for e in self.edges[from_id] if not (e.to_id == to_id and e.edge_type == edge_type) ] else: self.edges[from_id] = [e for e in self.edges[from_id] if e.to_id != to_id] return len(self.edges[from_id]) < before # --- Traversal --- def traverse( self, start_ids: list[str], max_depth: int = 3, min_weight: float = 0.3, max_nodes: int = 20, ) -> list[tuple[GraphNode, int]]: """ BFS traversal from start nodes. Returns ``(node, depth)`` pairs sorted by weight descending. In-memory: <1ms for typical session graphs. """ visited: set[str] = set() result: list[tuple[GraphNode, int]] = [] queue: deque[tuple[str, int]] = deque() for start_id in start_ids: if start_id in self.nodes: queue.append((start_id, 0)) visited.add(start_id) while queue and len(result) < max_nodes: node_id, depth = queue.popleft() node = self.nodes.get(node_id) if node: result.append((node, depth)) if depth < max_depth: for edge in self.edges.get(node_id, []): if edge.to_id not in visited and edge.weight >= min_weight: visited.add(edge.to_id) queue.append((edge.to_id, depth + 1)) result.sort(key=lambda x: x[0].weight, reverse=True) return result # --- Serialization --- def to_structured_text(self, max_tokens: int = 10000) -> str: """ Serialize graph to structured text for context window. This is the warm-layer representation — what the LLM sees instead of a lossy summary. Format matches the GraphCompactor extraction prompt so it can be round-tripped through from_structured_text(). """ if not self.nodes: return "" sections: list[str] = [] by_type: dict[str, list[GraphNode]] = {} for node in sorted(self.nodes.values(), key=lambda n: n.weight, reverse=True): by_type.setdefault(node.node_type, []).append(node) if "topic" in by_type: lines = ["## Topics"] for node in by_type["topic"]: connected = [] for edge in self.edges.get(node.id, []): target = self.nodes.get(edge.to_id) if target: connected.append(f"{target.label} ({edge.edge_type})") conn_str = f" | {', '.join(connected)}" if connected else "" lines.append(f"- [{node.label}] [{node.weight:.1f}] {node.content}{conn_str}") sections.append("\n".join(lines)) if "decision" in by_type: lines = ["## Decisions"] for node in by_type["decision"]: lines.append(f"- {node.content}") sections.append("\n".join(lines)) if "open_question" in by_type: lines = ["## Open"] for node in by_type["open_question"]: lines.append(f"- {node.content}") sections.append("\n".join(lines)) if "artifact" in by_type: lines = ["## Artifacts"] for node in by_type["artifact"]: lines.append(f"- {node.label}: {node.content}") sections.append("\n".join(lines)) if "state" in by_type: lines = ["## Context"] for node in by_type["state"]: lines.append(f"- {node.content}") sections.append("\n".join(lines)) text = "\n\n".join(sections) char_budget = max_tokens * 4 if len(text) > char_budget: text = text[:char_budget] + "\n[... graph truncated to fit budget]" return text @classmethod def from_structured_text( cls, text: str, session_id: Optional[str] = None ) -> "SessionGraph": """ Parse structured text back into a SessionGraph. Best-effort parser — used when reloading a previous compaction result from the context window. The graph is the source of truth; the text is a serialization. """ graph = cls(session_id=session_id) if not text.strip(): return graph current_type = "topic" type_map = { "## Topics": "topic", "## Decisions": "decision", "## Open": "open_question", "## Artifacts": "artifact", "## Context": "state", } node_counter = 0 for line in text.split("\n"): line = line.strip() if not line: continue if line in type_map: current_type = type_map[line] continue if line.startswith("- "): content = line[2:].strip() node_counter += 1 node_id = f"restored_{node_counter}" label = content if content.startswith("["): bracket_end = content.find("]") if bracket_end > 0: label = content[1:bracket_end] content = content[bracket_end + 1:].strip() weight = 0.5 if content.startswith("[") and "]" in content: weight_end = content.find("]") try: weight = float(content[1:weight_end]) except ValueError: pass content = content[weight_end + 1:].strip() graph.add_node(GraphNode( id=node_id, node_type=current_type, label=label, content=content or label, weight=weight, )) return graph # --- Persistence --- async def flush(self, repository=None) -> int: """ Persist new edges and nodes to long-term storage. Args: repository: A ``MemoryRepository`` instance. If None, no-op. Returns: Number of items flushed. """ if repository is None or (not self._new_edges and not self._new_nodes): return 0 flushed = await repository.flush_edges(self._new_edges) self._new_edges.clear() self._new_nodes.clear() return flushed def __repr__(self) -> str: return ( f"SessionGraph(nodes={self.node_count}, " f"edges={self.edge_count}, session={self.session_id})" ) 3Eѿx>""" SYNAPSE — pre-generation associative memory injection. The inhale of BREATHE. Fires BEFORE generation, not after tool call. This is middleware, not a tool — because tools are reactive (called), SYNAPSE must be proactive (inject before thinking). Pipeline: [User message] → anchor extraction → graph traversal → relevance filter → injection Total overhead target: <200ms Usage:: synapse = Synapse(repository=my_repo, config=BreatheConfig()) await synapse.initialize() messages = await synapse.inject(messages) # pass enriched messages to your LLM """ from __future__ import annotations import logging import re import time from typing import Optional from .anchor_extractor import AnchorExtractor, AnchorResult, Anchor from .config import BreatheConfig from .context_injector import ContextInjector from .interfaces import MemoryRepository, VectorSearchClient, RetrievedNode from .metrics import BreatheMetrics, SynapseEvent, AnchorDetail, NodeDetail logger = logging.getLogger(__name__) _WORD_RE = re.compile(r"[\w\u0400-\u04FF]{3,}", re.UNICODE) class Synapse: """ Pre-generation memory injection middleware. Intercepts messages before they go to the LLM, extracts associative anchors, traverses the memory graph, and injects relevant context. The LLM starts thinking with memories already present. Hybrid extraction: regex (always, 2ms) + optional local model (~250ms). All external dependencies are optional: - ``repository``: enables graph BFS + keyword search - ``vector_client``: enables semantic vector search - ``enable_model``: enables local MLX model extraction Without any backends, SYNAPSE is a no-op — safe to wire in before backends are ready. """ def __init__( self, repository: Optional[MemoryRepository] = None, vector_client: Optional[VectorSearchClient] = None, config: Optional[BreatheConfig] = None, enable_model: bool = True, ): """ Args: repository: Storage backend for graph BFS + keyword search. If None, SYNAPSE skips graph traversal. vector_client: Semantic search client. If None, skips vector retrieval. config: BreatheConfig with language packs and tuning parameters. Defaults to BreatheConfig() with EN + RU support. enable_model: Whether to enable local MLX model extraction (Phase 3). """ self._repo = repository self._vector = vector_client self._config = config or BreatheConfig() self._enable_model = enable_model self._injector = ContextInjector( mode_budgets=self._config.mode_budgets, labels=self._config.labels, ) self._extractor: Optional[AnchorExtractor] = None self._model_extractor = None self._known_concepts: dict[str, str] = {} self._initialized = False self._session_injected: set[str] = set() self.metrics = BreatheMetrics.get() async def initialize(self) -> None: """ Load known concepts from storage and optionally prepare local model. Called once at session start — safe to call multiple times (idempotent). """ if self._initialized: return try: if self._repo: self._known_concepts = await self._repo.get_concepts() except Exception as e: logger.warning(f"SYNAPSE: failed to load concepts from repo: {e}") self._known_concepts = {} # Build temporal / emotional patterns from all configured language packs temporal_patterns = [ pack.temporal_pattern for pack in self._config.language_packs ] emotional_patterns = [ pack.emotional_pattern for pack in self._config.language_packs ] self._extractor = AnchorExtractor( known_concepts=self._known_concepts, temporal_patterns=temporal_patterns, emotional_patterns=emotional_patterns, ) if self._enable_model: try: from .model_extractor import ModelAnchorExtractor candidate = ModelAnchorExtractor() if candidate.available: self._model_extractor = candidate logger.info("SYNAPSE: model extractor available (lazy load on first use)") except Exception as e: logger.debug(f"SYNAPSE: model extractor unavailable: {e}") self._initialized = True self.metrics.known_concepts_count = len(self._known_concepts) self.metrics.model_available = self._model_extractor is not None logger.info(f"SYNAPSE initialized: {len(self._known_concepts)} known concepts") async def inject(self, messages: list) -> list: """ Main entry point. Inject associative memory into messages. Inserts a ```` block into the last user message. If nothing relevant is found, returns messages unchanged. Args: messages: Conversation messages (system + user/assistant pairs). Returns: Messages with associative memory injected, or unchanged if nothing found. """ if not self._repo and not self._vector: return messages if not self._initialized: await self.initialize() start_time = time.monotonic() try: user_message, tail = self._extract_context(messages) if not user_message: return messages regex_start = time.monotonic() result = self._extractor.extract(user_message) regex_ms = (time.monotonic() - regex_start) * 1000 model_ms = 0.0 model_triggered = False if self._model_extractor: from .model_extractor import should_use_model if should_use_model(result, self._config.model_trigger_threshold): model_triggered = True self.metrics.model_triggers += 1 model_start = time.monotonic() model_anchors = self._model_extractor.extract(user_message) model_ms = (time.monotonic() - model_start) * 1000 if model_anchors: existing_texts = {a.text.lower() for a in result.anchors} added = 0 for anchor in model_anchors: if anchor.text.lower() not in existing_texts: anchor = self._match_to_concepts(anchor) result.anchors.append(anchor) existing_texts.add(anchor.text.lower()) added += 1 logger.info(f"Model added {added} anchors → total {len(result.anchors)}") if not result.anchors: self.metrics.record_synapse_skip() return messages nodes = await self._traverse(result, user_message=user_message) if not nodes: return messages injection_text = self._injector.format_injection( nodes=nodes, mode=result.conversation_mode, anchors_text=[a.text for a in result.anchors], ) if not injection_text: return messages messages = _insert_injection(messages, injection_text) injected_ids = [] for node in nodes: if node.node_id: self._session_injected.add(node.node_id) injected_ids.append(node.node_id[:8]) elapsed_ms = (time.monotonic() - start_time) * 1000 self.metrics.record_synapse(SynapseEvent( timestamp=time.time(), anchors_count=len(result.anchors), matched_nodes=len(result.node_ids), injected_nodes=len(nodes), regex_ms=regex_ms, model_ms=model_ms, total_ms=elapsed_ms, mode=result.conversation_mode, model_triggered=model_triggered, user_message=user_message[:1000], anchors=[ AnchorDetail( text=a.text, anchor_type=a.anchor_type, confidence=a.confidence, source=a.source, matched_node_id=a.matched_node_id, ) for a in result.anchors ], nodes=[ NodeDetail( concept=n.concept, node_type=n.node_type, summary=(n.summary or "")[:500], importance=n.importance, depth=n.depth, relation=n.relation, ) for n in nodes ], injection_text=injection_text[:2000] if injection_text else "", )) self.metrics.record_anchors([a.text for a in result.anchors]) logger.info( f"SYNAPSE: {len(result.anchors)} anchors → " f"{len(nodes)} nodes → injected ({elapsed_ms:.0f}ms)" ) except Exception as e: logger.warning(f"SYNAPSE injection failed: {e}") # Never fail the request — just skip injection return messages def reset_session(self) -> None: """Clear session deduplication cache (call at session start).""" self._session_injected.clear() def _match_to_concepts(self, anchor: Anchor) -> Anchor: text_lower = anchor.text.lower().strip() for concept, node_id in self._known_concepts.items(): concept_lower = concept.lower() if text_lower == concept_lower: anchor.matched_node_id = node_id anchor.source = "model+concept" anchor.confidence = 0.9 break if text_lower in set(concept_lower.split()): anchor.matched_node_id = node_id anchor.source = "model+concept" anchor.confidence = 0.8 break return anchor async def _traverse( self, anchor_result: AnchorResult, user_message: str = "" ) -> list[RetrievedNode]: nodes: list[RetrievedNode] = [] # Strategy 1: Graph BFS from matched node IDs if self._repo and anchor_result.node_ids: try: graph_nodes = await self._repo.graph_bfs(anchor_result.node_ids) nodes.extend(graph_nodes) except Exception as e: logger.warning(f"SYNAPSE graph BFS failed: {e}") # Strategy 2: Vector search (semantic, highest quality) if self._vector and anchor_result.anchors: hub = self._config.hub_exclusions anchor_query = " ".join( a.text for a in anchor_result.anchors if a.text.lower() not in hub ) if anchor_query.strip(): try: vector_nodes = await self._vector.search(anchor_query) filtered = [n for n in vector_nodes if n.importance >= self._config.min_similarity] nodes.extend(filtered) except Exception as e: logger.warning(f"SYNAPSE vector search failed: {e}") # Strategy 3: Keyword search for unmatched anchors if self._repo: unmatched = [ a for a in anchor_result.anchors if not a.matched_node_id and a.anchor_type in ("entity", "temporal", "theme") ] if unmatched: try: kw_nodes = await self._repo.keyword_search([a.text for a in unmatched[:5]]) nodes.extend(kw_nodes) except Exception as e: logger.warning(f"SYNAPSE keyword search failed: {e}") # Deduplicate seen: set[str] = set() unique: list[RetrievedNode] = [] for node in nodes: if node.node_id not in seen: seen.add(node.node_id) unique.append(node) # Session dedup if self._session_injected: unique = [n for n in unique if n.node_id not in self._session_injected] # Hub exclusion hub = self._config.hub_exclusions unique = [n for n in unique if n.concept.lower() not in hub] # Relevance filter anchor_texts = {a.text.lower() for a in anchor_result.anchors} clean_anchors = {t for t in anchor_texts if t not in hub} if clean_anchors: anchor_words: set[str] = set() stopwords = self._config.stopwords for a in clean_anchors: for w in _WORD_RE.findall(a): w_lower = w.lower() if w_lower not in stopwords and len(w_lower) >= 3: anchor_words.add(w_lower) if anchor_words: scored = [] for node in unique: score = _relevance_score(node, anchor_words, stopwords) if score > 0: scored.append((score, node)) scored.sort(key=lambda x: (-x[0], -x[1].importance)) unique = [node for _, node in scored] return unique[: self._config.max_injected_nodes] @staticmethod def _extract_context(messages: list) -> tuple[str, list[str]]: user_message = "" tail: list[str] = [] for msg in reversed(messages): role = msg.get("role", "") content = msg.get("content", "") if isinstance(content, list): content = " ".join( block.get("text", "") for block in content if isinstance(block, dict) and block.get("type") == "text" ) if not isinstance(content, str): continue if role == "user" and not user_message: user_message = content elif role in ("user", "assistant") and content and len(tail) < 3: tail.append(content[:500]) return user_message, tail def _relevance_score( node: RetrievedNode, anchor_words: set[str], stopwords: frozenset[str] ) -> float: """Score node relevance by keyword overlap with extracted anchors.""" node_text = f"{node.concept} {node.summary or ''} {node.relation or ''}" node_words = set(w.lower() for w in _WORD_RE.findall(node_text)) - stopwords if not node_words or not anchor_words: return 0.0 overlap = anchor_words & node_words if not overlap: anchor_stems = {w[:5] for w in anchor_words if len(w) >= 5} node_stems = {w[:5] for w in node_words if len(w) >= 5} stem_overlap = anchor_stems & node_stems if stem_overlap: overlap = stem_overlap return len(overlap) / max(len(anchor_words), 1) def _insert_injection(messages: list, injection_text: str) -> list: """Prepend injection text to the last user message.""" last_user_idx = None for i in range(len(messages) - 1, -1, -1): if messages[i].get("role") == "user": last_user_idx = i break if last_user_idx is None: return messages new_messages = list(messages) original = new_messages[last_user_idx] content = original.get("content", "") if isinstance(content, list): injection_block = {"type": "text", "text": injection_text + "\n\n"} new_messages[last_user_idx] = {**original, "content": [injection_block] + content} else: new_messages[last_user_idx] = {**original, "content": injection_text + "\n\n" + content} return new_messages ֥x{100644 basic_usage.pyGB2Qk΂lX螓`100644 custom_language.py0їF{$T2؁2100644 with_postgres.py;ī:Q"}gF56/x.""" Basic usage — BREATHE with in-memory stub backends. Demonstrates the full pipeline without any external services. Run with: python examples/basic_usage.py """ import asyncio from breathe import ( Synapse, GraphCompactor, BreatheConfig, MemoryRepository, LLMClient, RetrievedNode, ) from breathe.config import ENGLISH # --- Stub implementations --- class StubMemoryRepository(MemoryRepository): """In-memory stub — returns hardcoded concepts and memories.""" async def get_concepts(self) -> dict[str, str]: return { "FastAPI": "uuid-fastapi-001", "PostgreSQL": "uuid-postgres-001", "SYNAPSE": "uuid-synapse-001", } async def graph_bfs(self, start_ids, max_depth=2, min_strength=0.2, limit=20): return [ RetrievedNode( node_id="uuid-related-001", concept="FastAPI async patterns", node_type="entity", summary="FastAPI handles async requests efficiently with async def handlers", importance=0.8, depth=1, ) ] async def keyword_search(self, keywords, limit=5): return [ RetrievedNode( node_id=f"mem-kw-{i}", concept=kw, node_type="memory_keyword", summary=f"Previous discussion about {kw}", importance=0.6, depth=0, memory_content=f"We discussed {kw} in detail last week and decided to use async handlers.", ) for i, kw in enumerate(keywords[:2]) ] class StubLLMClient(LLMClient): """Returns hardcoded graph extraction for demo.""" async def complete(self, prompt, max_tokens=4000, temperature=0.2): return """## Topics - [API Design] [0.9] FastAPI endpoint architecture | PostgreSQL ## Decisions - Use async handlers for all database operations ## Open - Should we add Redis caching for heavy queries? ## Artifacts - api/routes.py: main API router """ # --- Main demo --- async def main(): config = BreatheConfig( language_packs=[ENGLISH], default_language="en", ) repo = StubMemoryRepository() synapse = Synapse(repository=repo, config=config, enable_model=False) await synapse.initialize() messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "How should I structure my FastAPI endpoints?"}, ] print("--- Before injection ---") print(messages[-1]["content"]) print() messages = await synapse.inject(messages) print("--- After SYNAPSE injection ---") print(messages[-1]["content"]) print() # Show metrics stats = synapse.metrics.to_dict() print(f"SYNAPSE stats: {stats['synapse']['total_injections']} injections, " f"hit rate: {stats['synapse']['hit_rate']:.0%}") print() # Demonstrate GraphCompactor history = [] for i in range(15): history.append({"role": "user", "content": f"User message {i}: detailed question about FastAPI, PostgreSQL, and async Python patterns. We discussed caching strategies, database connection pooling, and endpoint design."}) history.append({"role": "assistant", "content": f"Assistant response {i}: Here is a detailed explanation covering async handlers, connection pool configuration, query optimization, and best practices for structuring FastAPI applications with PostgreSQL backends."}) compactor = GraphCompactor(llm_client=StubLLMClient()) result = await compactor.compress(history) print(f"GraphCompactor: compressed={result['compressed']}") if result["compressed"]: meta = result["metadata"] print(f" {meta['original_tokens']} → {meta['compressed_tokens']} tokens " f"({meta['compression_ratio']:.0%} saved)") print(f" Graph: {meta['graph_nodes']} nodes, {meta['graph_edges']} edges") if __name__ == "__main__": asyncio.run(main()) xWx """ Custom language pack — adding German support to BREATHE. Demonstrates how to extend BREATHE to any language in ~20 lines. """ import asyncio import re from breathe import Synapse, BreatheConfig, LanguagePack, ENGLISH, RetrievedNode from breathe.interfaces import MemoryRepository # 1. Define a language pack GERMAN = LanguagePack( code="de", stopwords=frozenset({ "der", "die", "das", "und", "ist", "in", "zu", "von", "mit", "für", "auf", "an", "aus", "bei", "nach", "über", "unter", "nicht", "auch", "noch", "aber", "oder", "wenn", "dann", "ich", "du", "er", "sie", "wir", "ihr", "sie", "es", "sein", "haben", "werden", "können", "müssen", "sollen", }), hub_exclusions=frozenset({"claude", "speicher", "system"}), temporal_pattern=re.compile( r"\b(gestern|heute|morgen|letzte Woche|diese Woche|neulich|wieder|damals)\b", re.IGNORECASE, ), emotional_pattern=re.compile( r"\b(müde|Schmerzen|frustriert|wütend|glücklich|toll|scheiße|traurig|vermisse|liebe)\b", re.IGNORECASE, ), labels={ "themes": "Themen", "insights": "Erkenntnisse", "associative_memory_tag": "assoziatives_gedächtnis", }, ) # 2. Minimal stub repo for demo class MinimalRepo(MemoryRepository): async def get_concepts(self): return {"FastAPI": "uuid-1", "Python": "uuid-2"} async def graph_bfs(self, start_ids, **kwargs): return [] async def keyword_search(self, keywords, limit=5): return [ RetrievedNode( node_id=f"kw-{i}", concept=kw, node_type="memory_keyword", summary=f"Frühere Diskussion über {kw}", importance=0.6, depth=0, memory_content=f"Wir haben {kw} letzte Woche ausführlich besprochen.", ) for i, kw in enumerate(keywords[:2]) ] async def main(): # 3. Use in BreatheConfig — works with EN + DE simultaneously config = BreatheConfig( language_packs=[ENGLISH, GERMAN], default_language="de", ) synapse = Synapse(repository=MinimalRepo(), config=config, enable_model=False) await synapse.initialize() messages = [ {"role": "system", "content": "Du bist ein hilfreicher Assistent."}, {"role": "user", "content": "Wie soll ich meine FastAPI Endpunkte strukturieren?"}, ] enriched = await synapse.inject(messages) print("Enriched message (German):") print(enriched[-1]["content"]) if __name__ == "__main__": asyncio.run(main()) :&x """ Full integration — BREATHE + Memory Nexus with PostgreSQL. Prerequisites: pip install breathe-memory[pg] createdb breathe_demo psql breathe_demo < schema.sql # see memory_nexus/postgres.py for DDL Run with: python examples/with_postgres.py """ import asyncio import os from breathe import Synapse, BreatheConfig from breathe.config import ENGLISH from memory_nexus import PostgresMemoryStore DSN = os.environ.get("DATABASE_URL", "postgresql://localhost/breathe_demo") async def main(): # Initialize Memory Nexus store # Default: all-MiniLM-L6-v2 (384-dim, fast, good for prototyping) # Production: model_name="intfloat/multilingual-e5-large" (1024-dim) # — remember to create your table with vector(1024) instead of vector(384) store = PostgresMemoryStore( dsn=DSN, model_name="sentence-transformers/all-MiniLM-L6-v2", min_similarity=0.55, ) await store.initialize() # Store some memories await store.store( "FastAPI is a modern Python web framework for building APIs with automatic OpenAPI docs.", metadata=None, ) await store.store( "PostgreSQL with pgvector extension enables vector similarity search in SQL.", metadata=None, ) await store.store( "Redis is an in-memory data structure store used for caching and pub/sub.", metadata=None, ) # Initialize SYNAPSE — store acts as both VectorSearchClient and can be # wrapped in a MemoryRepository adapter for graph BFS config = BreatheConfig( language_packs=[ENGLISH], default_language="en", min_similarity=0.55, ) synapse = Synapse( vector_client=store, # Memory Nexus as semantic search backend config=config, enable_model=False, ) await synapse.initialize() messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "What should I use for caching in FastAPI?"}, ] print("Query:", messages[-1]["content"]) print() enriched = await synapse.inject(messages) print("Enriched message:") print(enriched[-1]["content"]) print() # Search memories directly results = await store.search("caching database", limit=3) print(f"Direct search results ({len(results)} found):") for node in results: print(f" [{node.importance:.2f}] {node.summary[:80]}...") await store.close() if __name__ == "__main__": asyncio.run(main()) + xh100644 __init__.pyv y!秷C100644 models.py-,ZE9co100644 postgres.pybK#w] P9u;xI""" Memory Nexus — hybrid semantic memory for LLM applications. Provides persistent memory storage with semantic search via pgvector. Integrates with BREATHE as a VectorSearchClient backend. Architecture: MemoryStore — abstract storage interface PostgresStore — asyncpg + pgvector implementation (optional dep) HybridSearch — dense + keyword search Quick start:: from memory_nexus import PostgresMemoryStore from breathe import Synapse store = PostgresMemoryStore(dsn="postgresql://...") await store.initialize() synapse = Synapse(vector_client=store) """ from .models import Memory, MemoryMetadata from .store import MemoryStore __version__ = "0.1.0" __all__ = [ "Memory", "MemoryMetadata", "MemoryStore", ] # Optional: PostgreSQL backend (requires asyncpg + pgvector) try: from .postgres import PostgresMemoryStore __all__.append("PostgresMemoryStore") except ImportError: pass IpxP"""Data models for Memory Nexus.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional @dataclass class MemoryMetadata: """Structured metadata attached to a memory.""" tags: list[str] = field(default_factory=list) source: str = "" importance: float = 0.5 # 0–1 extra: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict: return { "tags": self.tags, "source": self.source, "importance": self.importance, **self.extra, } @classmethod def from_dict(cls, data: dict) -> "MemoryMetadata": return cls( tags=data.get("tags", []), source=data.get("source", ""), importance=data.get("importance", 0.5), extra={k: v for k, v in data.items() if k not in ("tags", "source", "importance")}, ) @dataclass class Memory: """A single memory entry.""" id: str content: str metadata: MemoryMetadata = field(default_factory=MemoryMetadata) created_at: datetime = field(default_factory=datetime.utcnow) similarity: Optional[float] = None # set during search results def to_dict(self) -> dict: return { "id": self.id, "content": self.content, "metadata": self.metadata.to_dict(), "created_at": self.created_at.isoformat(), "similarity": self.similarity, } @classmethod def from_dict(cls, data: dict) -> "Memory": meta = data.get("metadata", {}) if isinstance(meta, str): import json try: meta = json.loads(meta) except Exception: meta = {} return cls( id=data["id"], content=data["content"], metadata=MemoryMetadata.from_dict(meta), created_at=datetime.fromisoformat(data["created_at"]) if "created_at" in data else datetime.utcnow(), similarity=data.get("similarity"), ) x.x~$""" PostgreSQL + pgvector implementation of MemoryStore. Requires: pip install breathe-memory[pg] Database setup (default — 384-dim model): CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE memories ( id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, content TEXT NOT NULL, embedding vector(384), metadata JSONB DEFAULT '{}', created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE INDEX ON memories USING ivfflat (embedding vector_cosine_ops); CREATE INDEX ON memories (created_at DESC); Embedding models: Default: sentence-transformers/all-MiniLM-L6-v2 (384-dim, ~90 MB, English). Good for prototyping and quick starts. Production: intfloat/multilingual-e5-large (1024-dim, ~1.2 GB, multilingual). Significantly better retrieval quality, especially for non-English text. To use: pass model_name="intfloat/multilingual-e5-large" and create your table with ``embedding vector(1024)`` instead of ``vector(384)``. The vector dimension is determined by the model — make sure your table column matches the model output dimension. """ from __future__ import annotations import json import logging import uuid from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Optional from .models import Memory, MemoryMetadata from .store import MemoryStore from breathe.interfaces import RetrievedNode logger = logging.getLogger(__name__) _executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="breathe-embed") class PostgresMemoryStore(MemoryStore): """ asyncpg + pgvector memory store with sentence-transformer embeddings. Usage:: store = PostgresMemoryStore( dsn="postgresql://user:pass@localhost/mydb", model_name="sentence-transformers/all-MiniLM-L6-v2", ) await store.initialize() mem = await store.store("Claude is an AI assistant made by Anthropic") results = await store.search("AI assistant", limit=3) """ def __init__( self, dsn: str, model_name: str = "sentence-transformers/all-MiniLM-L6-v2", min_similarity: float = 0.55, ): """ Args: dsn: PostgreSQL connection string. model_name: Sentence-transformer model for embeddings. Default: all-MiniLM-L6-v2 (384-dim, fast, good for prototyping). Production: intfloat/multilingual-e5-large (1024-dim, multilingual). Make sure your ``embedding vector(N)`` column matches the model dimension. min_similarity: Minimum cosine similarity to return from search. """ self._dsn = dsn self._model_name = model_name self._min_similarity = min_similarity self._pool = None self._model = None async def initialize(self) -> None: """Create connection pool and load embedding model.""" try: import asyncpg except ImportError as e: raise ImportError( "asyncpg is required for PostgresMemoryStore. " "Install with: pip install breathe-memory[pg]" ) from e self._pool = await asyncpg.create_pool(self._dsn, min_size=1, max_size=5) # Register pgvector codec async with self._pool.acquire() as conn: await conn.execute("CREATE EXTENSION IF NOT EXISTS vector") await conn.set_type_codec( "vector", encoder=lambda v: "[" + ",".join(map(str, v)) + "]", decoder=lambda v: list(map(float, v.strip("[]").split(","))), schema="pg_catalog", format="text", ) logger.info(f"PostgresMemoryStore initialized (model: {self._model_name})") def _get_model(self): """Lazy-load embedding model.""" if self._model is None: try: from sentence_transformers import SentenceTransformer self._model = SentenceTransformer(self._model_name) except ImportError as e: raise ImportError( "sentence-transformers is required. " "Install with: pip install breathe-memory[pg]" ) from e return self._model async def _embed(self, text: str) -> list[float]: """Encode text to embedding vector (runs in thread pool).""" import asyncio loop = asyncio.get_running_loop() model = self._get_model() vector = await loop.run_in_executor( _executor, lambda: model.encode(text).tolist() ) return vector async def store( self, content: str, metadata: Optional[MemoryMetadata] = None, ) -> Memory: if not self._pool: raise RuntimeError("Call initialize() first") meta = metadata or MemoryMetadata() embedding = await self._embed(content) memory_id = str(uuid.uuid4()) async with self._pool.acquire() as conn: row = await conn.fetchrow( """ INSERT INTO memories (id, content, embedding, metadata, created_at) VALUES ($1, $2, $3::vector, $4, NOW()) RETURNING id, created_at """, memory_id, content, embedding, json.dumps(meta.to_dict()), ) return Memory( id=row["id"], content=content, metadata=meta, created_at=row["created_at"], ) async def get(self, memory_id: str) -> Optional[Memory]: if not self._pool: raise RuntimeError("Call initialize() first") async with self._pool.acquire() as conn: row = await conn.fetchrow( "SELECT id, content, metadata, created_at FROM memories WHERE id = $1", memory_id, ) if not row: return None meta = json.loads(row["metadata"]) if isinstance(row["metadata"], str) else row["metadata"] return Memory( id=row["id"], content=row["content"], metadata=MemoryMetadata.from_dict(meta or {}), created_at=row["created_at"], ) async def delete(self, memory_id: str) -> bool: if not self._pool: raise RuntimeError("Call initialize() first") async with self._pool.acquire() as conn: result = await conn.execute( "DELETE FROM memories WHERE id = $1", memory_id ) return result.split()[-1] == "1" async def search(self, query: str, limit: int = 5) -> list[RetrievedNode]: if not self._pool: raise RuntimeError("Call initialize() first") embedding = await self._embed(query) async with self._pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, content, metadata, 1 - (embedding <=> $1::vector) AS similarity FROM memories WHERE 1 - (embedding <=> $1::vector) >= $2 ORDER BY embedding <=> $1::vector LIMIT $3 """, embedding, self._min_similarity, limit, ) nodes = [] for row in rows: content = row["content"] or "" meta = row["metadata"] or {} if isinstance(meta, str): try: meta = json.loads(meta) except Exception: meta = {} similarity = float(row["similarity"]) tags = meta.get("tags", []) concept = tags[0] if tags else content[:50] nodes.append(RetrievedNode( node_id=row["id"], concept=concept, node_type="memory_vector", summary=content[:500] if len(content) > 500 else content, importance=similarity, depth=0, memory_content=content, )) return nodes async def get_recent(self, limit: int = 20) -> list[Memory]: if not self._pool: raise RuntimeError("Call initialize() first") async with self._pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, content, metadata, created_at FROM memories ORDER BY created_at DESC LIMIT $1 """, limit, ) memories = [] for row in rows: meta = row["metadata"] or {} if isinstance(meta, str): try: meta = json.loads(meta) except Exception: meta = {} memories.append(Memory( id=row["id"], content=row["content"], metadata=MemoryMetadata.from_dict(meta), created_at=row["created_at"], )) return memories async def close(self) -> None: if self._pool: await self._pool.close() self._pool = None q$wϴx"""Abstract MemoryStore interface.""" from __future__ import annotations from abc import ABC, abstractmethod from typing import Optional from .models import Memory, MemoryMetadata from breathe.interfaces import RetrievedNode, VectorSearchClient class MemoryStore(VectorSearchClient, ABC): """ Abstract memory store — combines VectorSearchClient with CRUD operations. Implements VectorSearchClient so it can be passed directly to Synapse:: store = PostgresMemoryStore(dsn="postgresql://...") synapse = Synapse(vector_client=store) Implement this to connect Memory Nexus to any backend. See ``PostgresMemoryStore`` for the asyncpg + pgvector reference implementation. """ @abstractmethod async def initialize(self) -> None: """Initialize the store (create tables, load models, etc.).""" @abstractmethod async def store( self, content: str, metadata: Optional[MemoryMetadata] = None, ) -> Memory: """ Store a memory and return it with generated ID. Args: content: The text content to store. metadata: Optional metadata (tags, source, importance). """ @abstractmethod async def get(self, memory_id: str) -> Optional[Memory]: """Retrieve a memory by ID.""" @abstractmethod async def delete(self, memory_id: str) -> bool: """Delete a memory by ID. Returns True if deleted.""" @abstractmethod async def search(self, query: str, limit: int = 5) -> list[RetrievedNode]: """ Semantic search. Implements VectorSearchClient. Args: query: Search query text. limit: Maximum results. Returns: List of RetrievedNode sorted by similarity descending. """ @abstractmethod async def get_recent(self, limit: int = 20) -> list[Memory]: """Return most recently stored memories.""" async def close(self) -> None: """Clean up resources (close connections, etc.).""" ){xJ[build-system] requires = ["setuptools>=68", "wheel"] build-backend = "setuptools.build_meta" [project] name = "breathe-memory" version = "0.1.0" description = "Context optimization and associative memory for LLM applications" readme = "README.md" license = { text = "Apache-2.0" } authors = [{ name = "Kenaz GmbH", email = "hello@kenaz.ai" }] keywords = ["llm", "context", "memory", "rag", "compression", "ai"] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Topic :: Scientific/Engineering :: Artificial Intelligence", ] requires-python = ">=3.11" dependencies = [ "typing-extensions>=4.0", ] [project.optional-dependencies] # PostgreSQL backend for Memory Nexus pg = [ "asyncpg>=0.29", "sentence-transformers>=3.0", "numpy>=1.26", ] # Apple Silicon local model extractor (MLX) mlx = [ "mlx>=0.16", "mlx-lm>=0.16", ] # Anthropic LLM client for GraphCompactor anthropic = [ "anthropic>=0.40", ] # OpenAI LLM client for GraphCompactor openai = [ "openai>=1.0", ] # Everything all = [ "breathe-memory[pg,anthropic]", ] # Development dev = [ "pytest>=8.0", "pytest-asyncio>=0.23", "httpx>=0.27", "ruff>=0.4", "mypy>=1.10", ] [project.urls] Homepage = "https://github.com/tkenaz/breathe-memory" Documentation = "https://github.com/tkenaz/breathe-memory#readme" Issues = "https://github.com/tkenaz/breathe-memory/issues" [tool.setuptools.packages.find] include = ["breathe*", "memory_nexus*"] [tool.pytest.ini_options] asyncio_mode = "auto" testpaths = ["tests"] [tool.ruff] line-length = 100 target-version = "py311" [tool.ruff.lint] select = ["E", "F", "I", "UP"] ignore = ["E501"] [tool.mypy] python_version = "3.11" strict = false ignore_missing_imports = true 's x?100644 __init__.py⛲CK)wZS100644 test_anchor_extractor.pyJ8M-_湃u8x|*ã100644 test_context_injector.pyB;X ̙GUlMk5100644 test_session_graph.py%X9}+Bs&3gM*0xx, """Tests for AnchorExtractor.""" import pytest from breathe.anchor_extractor import AnchorExtractor, AnchorResult def make_extractor(concepts=None): return AnchorExtractor(known_concepts=concepts or {}) def test_known_concept_match(): extractor = make_extractor({"FastAPI": "uuid-001", "Redis": "uuid-002"}) result = extractor.extract("How do I add caching to FastAPI?") entity_texts = [a.text.lower() for a in result.entity_anchors] assert "fastapi" in entity_texts def test_known_concept_sets_node_id(): extractor = make_extractor({"FastAPI": "uuid-001"}) result = extractor.extract("I am working on FastAPI") matched = [a for a in result.anchors if a.matched_node_id] assert len(matched) >= 1 assert matched[0].matched_node_id == "uuid-001" def test_temporal_en(): extractor = make_extractor() result = extractor.extract("remember when we fixed the bug yesterday?") temporal = [a for a in result.anchors if a.anchor_type == "temporal"] assert len(temporal) >= 1 def test_technical_anchor(): extractor = make_extractor() result = extractor.extract("check the routes.py file for the API handler") tech = [a for a in result.anchors if a.anchor_type == "technical"] assert any("routes.py" in a.text for a in tech) def test_deduplication(): extractor = make_extractor({"FastAPI": "uuid-001"}) # FastAPI appears twice — should deduplicate result = extractor.extract("FastAPI is great, I love FastAPI") fastapi_anchors = [a for a in result.anchors if "fastapi" in a.text.lower()] assert len(fastapi_anchors) == 1 def test_conversation_mode_work(): extractor = make_extractor() result = extractor.extract("class MyModel: def __init__(): import asyncpg from FastAPI") assert result.conversation_mode == "work" def test_conversation_mode_casual(): extractor = make_extractor() result = extractor.extract("hey") assert result.conversation_mode == "casual" def test_empty_message(): extractor = make_extractor() result = extractor.extract("") assert isinstance(result, AnchorResult) assert len(result.anchors) == 0 def test_node_ids_property(): extractor = make_extractor({"Redis": "uuid-redis"}) result = extractor.extract("I need Redis for caching") assert "uuid-redis" in result.node_ids 趥xV """Tests for ContextInjector.""" import pytest from breathe.context_injector import ContextInjector from breathe.interfaces import RetrievedNode def make_node(concept, node_type="entity", summary=None, memory_content=None, importance=0.7): return RetrievedNode( node_id=f"id-{concept}", concept=concept, node_type=node_type, summary=summary, importance=importance, depth=0, memory_content=memory_content, ) def test_returns_none_for_empty_nodes(): injector = ContextInjector() assert injector.format_injection([]) is None def test_wraps_in_memory_tag(): injector = ContextInjector() nodes = [make_node("Redis", summary="In-memory data store")] result = injector.format_injection(nodes) assert "" in result assert "" in result def test_custom_memory_tag(): injector = ContextInjector(memory_tag="my_memory") nodes = [make_node("Redis", summary="cache")] result = injector.format_injection(nodes) assert "" in result def test_entity_concept_in_output(): injector = ContextInjector() nodes = [make_node("PostgreSQL", summary="Relational database with pgvector")] result = injector.format_injection(nodes) assert "PostgreSQL" in result def test_memory_content_included(): injector = ContextInjector() nodes = [make_node("Redis", node_type="memory_keyword", memory_content="Redis was discussed last week for session storage")] result = injector.format_injection(nodes) assert "session storage" in result def test_respects_budget(): injector = ContextInjector(mode_budgets={"balanced": 10}) # tiny budget nodes = [make_node("Redis", summary="x" * 1000)] result = injector.format_injection(nodes) # Should be truncated to ~40 chars + tag overhead assert len(result) < 200 def test_custom_labels(): injector = ContextInjector(labels={"themes": "Themen", "insights": "Erkenntnisse"}) nodes = [ make_node("performance", node_type="theme"), make_node("always use indexes", node_type="insight", summary="Always add DB indexes for hot query paths"), ] result = injector.format_injection(nodes) assert "Themen" in result or "Erkenntnisse" in result def test_event_nodes_formatted_with_header(): injector = ContextInjector() nodes = [make_node("Deployment outage", node_type="event", summary="Production went down for 2 hours")] result = injector.format_injection(nodes) assert "### Deployment outage" in result IrwxG """Tests for SessionGraph.""" import pytest from breathe.session_graph import SessionGraph, GraphNode, GraphEdge def make_graph(): g = SessionGraph(session_id="test-session") a = g.add_node(GraphNode(id="a", node_type="topic", label="API", content="REST API design")) b = g.add_node(GraphNode(id="b", node_type="decision", label="Use FastAPI", content="Chose FastAPI for async support")) c = g.add_node(GraphNode(id="c", node_type="artifact", label="main.py", content="Application entry point")) g.add_edge("a", "b", edge_type="led_to", weight=0.8) g.add_edge("b", "c", edge_type="part_of", weight=0.7) return g def test_add_node(): g = SessionGraph() node = g.add_node(GraphNode(id="x", node_type="topic", label="Test", content="test content")) assert g.node_count == 1 assert g.get_node("x") is not None def test_add_node_activates_existing(): g = SessionGraph() g.add_node(GraphNode(id="x", node_type="topic", label="Test", content="content", weight=0.5)) g.add_node(GraphNode(id="x", node_type="topic", label="Test", content="content")) assert g.node_count == 1 # no duplicate assert g.get_node("x").weight > 0.5 # activated def test_add_edge(): g = make_graph() assert g.edge_count == 2 def test_add_edge_missing_node(): g = SessionGraph() g.add_node(GraphNode(id="a", node_type="topic", label="A", content="content A")) result = g.add_edge("a", "nonexistent") assert result is None def test_traverse_bfs(): g = make_graph() results = g.traverse(["a"], max_depth=2) node_ids = [n.id for n, _ in results] assert "a" in node_ids assert "b" in node_ids # depth 1 assert "c" in node_ids # depth 2 def test_traverse_max_depth(): g = make_graph() results = g.traverse(["a"], max_depth=1) node_ids = [n.id for n, _ in results] assert "a" in node_ids assert "b" in node_ids assert "c" not in node_ids # too deep def test_traverse_sorted_by_weight(): g = SessionGraph() g.add_node(GraphNode(id="low", node_type="topic", label="low", content="", weight=0.3)) g.add_node(GraphNode(id="high", node_type="topic", label="high", content="", weight=0.9)) results = g.traverse(["low", "high"]) assert results[0][0].id == "high" def test_remove_node(): g = make_graph() g.remove_node("b") assert g.node_count == 2 assert g.get_node("b") is None def test_serialization_roundtrip(): g = make_graph() text = g.to_structured_text() assert "## Topics" in text assert "## Decisions" in text restored = SessionGraph.from_structured_text(text) assert restored.node_count > 0 def test_empty_graph_serialization(): g = SessionGraph() assert g.to_structured_text() == "" def test_repr(): g = make_graph() r = repr(g) assert "nodes=3" in r assert "edges=2" in r +YQHM¢'0/