From 8c84fa394d6bd3f4e0df4c282a9ef25bc596bbff Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:16:14 +0200 Subject: [PATCH 01/13] Add shared embedding regeneration service --- src/embedding_regeneration.py | 46 +++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 src/embedding_regeneration.py diff --git a/src/embedding_regeneration.py b/src/embedding_regeneration.py new file mode 100644 index 0000000..746c9b9 --- /dev/null +++ b/src/embedding_regeneration.py @@ -0,0 +1,46 @@ +"""Shared embedding regeneration workflow for REST and MCP surfaces.""" +from __future__ import annotations + +import uuid +from typing import Any, Dict + +from .db.attribution import get_memory_embedding_target, update_memory_embedding +from .embedder import create_embedding + + +def regenerate_memory_embedding( + memory_id: uuid.UUID, + *, + force: bool = False, +) -> Dict[str, Any]: + """Regenerate one memory embedding without changing memory identity or timestamps. + + By default an existing embedding is left untouched. ``force=True`` is intended + for embedding provider/model migrations where all stored vectors must be rebuilt. + """ + target = get_memory_embedding_target(memory_id) + if target is None: + raise KeyError(str(memory_id)) + + if target["has_embedding"] and not force: + return { + "id": str(memory_id), + "status": "unchanged", + "reason": "embedding_exists", + } + + embedding = create_embedding(target["content"]) + updated = update_memory_embedding(memory_id, embedding, force=force) + if not updated: + # A concurrent writer may have filled a NULL embedding after our read. + return { + "id": str(memory_id), + "status": "unchanged", + "reason": "embedding_exists", + } + + return { + "id": str(memory_id), + "status": "regenerated", + "force": force, + } From 115025074d75788a953a199411e66bb6b5cc852a Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:16:36 +0200 Subject: [PATCH 02/13] Support conditional embedding regeneration --- src/db/attribution.py | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/src/db/attribution.py b/src/db/attribution.py index b13afd2..e261d18 100644 --- a/src/db/attribution.py +++ b/src/db/attribution.py @@ -164,6 +164,42 @@ def get_memory_by_id(memory_id: uuid.UUID) -> Optional[Dict[str, Any]]: return _decode_memory(row) if row else None +def get_memory_embedding_target(memory_id: uuid.UUID) -> Optional[Dict[str, Any]]: + """Return content and whether a memory already has an embedding.""" + with get_db_cursor() as cursor: + cursor.execute( + """ + SELECT content, embedding IS NOT NULL AS has_embedding + FROM memory + WHERE id = %s + """, + (memory_id,), + ) + row = cursor.fetchone() + return dict(row) if row else None + + +def update_memory_embedding( + memory_id: uuid.UUID, + embedding: List[float], + *, + force: bool = False, +) -> bool: + """Set a memory embedding, overwriting an existing vector only when forced.""" + with get_db_cursor() as cursor: + cursor.execute( + """ + UPDATE memory + SET embedding = %s + WHERE id = %s + AND (%s OR embedding IS NULL) + RETURNING id + """, + (embedding, memory_id, force), + ) + return cursor.fetchone() is not None + + def get_recent_memories( limit: int = 50, offset: int = 0, From 0e04e7c5a1e3afa803bc7662f45e1773f23be70e Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:17:16 +0200 Subject: [PATCH 03/13] Chunk oversized embedding requests --- src/embedder/__init__.py | 486 ++++++++++++++++++--------------------- 1 file changed, 218 insertions(+), 268 deletions(-) diff --git a/src/embedder/__init__.py b/src/embedder/__init__.py index d4e6b1d..2dcd296 100644 --- a/src/embedder/__init__.py +++ b/src/embedder/__init__.py @@ -1,103 +1,81 @@ """ Multi-provider embedder for Open Brain. -Supports: -- openrouter: OpenRouter API (free tier available) -- openai: OpenAI API -- ollama: Local Ollama -- custom: Any OpenAI-compatible API - -Configure in settings.yaml: -```yaml -embedder: - provider: openrouter # openrouter, openai, ollama, custom - model: nomic-embed-text # or openrouter model - - # OpenRouter (default - free tier available) - openrouter_api_key: ${OPENROUTER_API_KEY} - - # OpenAI (optional) - openai_api_key: ${OPENAI_API_KEY} - - # Ollama (optional) - ollama_base_url: http://localhost:11434 - - # Custom API (optional) - custom_base_url: ${CUSTOM_API_URL} - custom_api_key: ${CUSTOM_API_KEY} - - dimensions: 768 -``` +Supports OpenRouter, OpenAI, Ollama, and custom OpenAI-compatible providers. """ +from __future__ import annotations + +import math import os -from typing import ClassVar, List, Optional from abc import ABC, abstractmethod +from typing import ClassVar, List, Optional import requests import yaml +DEFAULT_MAX_CHARS = 4000 +DEFAULT_CHUNK_OVERLAP = 200 + + class EmbedderConfig: """Configuration for embedder with multiple provider support.""" - - _instance: Optional['EmbedderConfig'] = None - + + _instance: Optional["EmbedderConfig"] = None + def __init__(self, config_path: str = None): if config_path is None: config_path = os.path.join( - os.path.dirname(__file__), - '..', '..', 'config', 'settings.yaml' + os.path.dirname(__file__), "..", "..", "config", "settings.yaml" ) - - # Load config - with open(config_path, 'r') as f: - config = yaml.safe_load(f) - - embedder_cfg = config.get('embedder', {}) - - # Provider settings - self.provider = embedder_cfg.get('provider', 'openrouter') - - # Model - self.model = embedder_cfg.get('model', 'nomic-embed-text') - - # Dimensions - self.dimensions = embedder_cfg.get('dimensions', 768) - - # OpenRouter (default) + + with open(config_path, "r") as file: + config = yaml.safe_load(file) + + embedder_cfg = config.get("embedder", {}) + self.provider = embedder_cfg.get("provider", "openrouter") + self.model = embedder_cfg.get("model", "nomic-embed-text") + self.dimensions = embedder_cfg.get("dimensions", 768) + self.max_chars = int( + os.environ.get( + "EMBEDDING_MAX_CHARS", + embedder_cfg.get("max_chars", DEFAULT_MAX_CHARS), + ) + ) + self.chunk_overlap = int( + os.environ.get( + "EMBEDDING_CHUNK_OVERLAP", + embedder_cfg.get("chunk_overlap", DEFAULT_CHUNK_OVERLAP), + ) + ) + if self.max_chars <= 0: + raise ValueError("embedder.max_chars must be greater than zero") + if self.chunk_overlap < 0 or self.chunk_overlap >= self.max_chars: + raise ValueError("embedder.chunk_overlap must be >= 0 and < max_chars") + self.openrouter_api_key = os.environ.get( - 'OPENROUTER_API_KEY', - embedder_cfg.get('openrouter_api_key', '') + "OPENROUTER_API_KEY", embedder_cfg.get("openrouter_api_key", "") ) - - # OpenAI self.openai_api_key = os.environ.get( - 'OPENAI_API_KEY', - embedder_cfg.get('openai_api_key', '') + "OPENAI_API_KEY", embedder_cfg.get("openai_api_key", "") ) - - # Ollama self.ollama_base_url = embedder_cfg.get( - 'ollama_base_url', - os.environ.get('OLLAMA_BASE_URL', 'http://localhost:11434') + "ollama_base_url", + os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434"), ) - - # Custom (any OpenAI-compatible API) self.custom_base_url = os.environ.get( - 'CUSTOM_API_URL', - embedder_cfg.get('custom_base_url', '') + "CUSTOM_API_URL", embedder_cfg.get("custom_base_url", "") ) self.custom_api_key = os.environ.get( - 'CUSTOM_API_KEY', - embedder_cfg.get('custom_api_key', '') + "CUSTOM_API_KEY", embedder_cfg.get("custom_api_key", "") ) - + @classmethod - def get_instance(cls, config_path: str = None) -> 'EmbedderConfig': + def get_instance(cls, config_path: str = None) -> "EmbedderConfig": if cls._instance is None: cls._instance = cls(config_path) return cls._instance - + @classmethod def reset(cls): """Reset singleton (useful for testing).""" @@ -106,107 +84,85 @@ def reset(cls): class BaseEmbedder(ABC): """Abstract base class for embedders.""" - + @abstractmethod def embed(self, text: str) -> List[float]: """Generate embedding for a single text.""" - pass - + raise NotImplementedError + @abstractmethod def embed_batch(self, texts: List[str]) -> List[List[float]]: """Generate embeddings for multiple texts.""" - pass - + raise NotImplementedError + @abstractmethod def is_available(self) -> bool: """Check if the provider is available.""" - pass - + raise NotImplementedError + @property @abstractmethod def dimensions(self) -> int: """Embedding dimensions.""" - pass + raise NotImplementedError class OpenRouterEmbedder(BaseEmbedder): - """OpenRouter API embedder (default, free tier available).""" - - # Default free models on OpenRouter + """OpenRouter API embedder.""" + DEFAULT_MODELS: ClassVar[dict[str, int]] = { - 'text-embedding-3-small': 1536, - 'text-embedding-ada-002': 1536, + "text-embedding-3-small": 1536, + "text-embedding-ada-002": 1536, } - + def __init__(self, config: EmbedderConfig): self.config = config self.api_key = config.openrouter_api_key - self.model = config.model or 'text-embedding-3-small' - self.base_url = 'https://openrouter.ai/api/v1' - - # Get dimensions for model - self._dimensions = config.dimensions - if self.model in self.DEFAULT_MODELS: - self._dimensions = self.DEFAULT_MODELS[self.model] - + self.model = config.model or "text-embedding-3-small" + self.base_url = "https://openrouter.ai/api/v1" + self._dimensions = self.DEFAULT_MODELS.get(self.model, config.dimensions) + @property def dimensions(self) -> int: return self._dimensions - - def embed(self, text: str) -> List[float]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', - 'HTTP-Referer': 'https://openbrain.local', # Required by OpenRouter - 'X-Title': 'Open Brain', + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + "HTTP-Referer": "https://openbrain.local", + "X-Title": "Open Brain", } - + + def embed(self, text: str) -> List[float]: response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': text, - }, - headers=headers, - timeout=30 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": text}, + headers=self._headers(), + timeout=30, ) response.raise_for_status() - - result = response.json() - return result['data'][0]['embedding'] - + return response.json()["data"][0]["embedding"] + def embed_batch(self, texts: List[str]) -> List[List[float]]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', - 'HTTP-Referer': 'https://openbrain.local', - 'X-Title': 'Open Brain', - } - response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': texts, - }, - headers=headers, - timeout=60 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": texts}, + headers=self._headers(), + timeout=60, ) response.raise_for_status() - - result = response.json() - # Sort by index to maintain order - embeddings = sorted(result['data'], key=lambda x: x['index']) - return [e['embedding'] for e in embeddings] - + embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) + return [item["embedding"] for item in embeddings] + def is_available(self) -> bool: if not self.api_key: return False try: response = requests.get( - f'{self.base_url}/models', - headers={'Authorization': f'Bearer {self.api_key}'}, - timeout=10 + f"{self.base_url}/models", + headers={"Authorization": f"Bearer {self.api_key}"}, + timeout=10, ) return response.status_code == 200 except Exception: @@ -215,79 +171,59 @@ def is_available(self) -> bool: class OpenAIEmbedder(BaseEmbedder): """OpenAI API embedder.""" - + DEFAULT_MODELS: ClassVar[dict[str, int]] = { - 'text-embedding-3-small': 1536, - 'text-embedding-3-large': 3072, - 'text-embedding-ada-002': 1536, + "text-embedding-3-small": 1536, + "text-embedding-3-large": 3072, + "text-embedding-ada-002": 1536, } - + def __init__(self, config: EmbedderConfig): self.config = config self.api_key = config.openai_api_key - self.model = config.model or 'text-embedding-3-small' - self.base_url = os.environ.get( - 'OPENAI_BASE_URL', - 'https://api.openai.com/v1' - ) - - self._dimensions = config.dimensions - if self.model in self.DEFAULT_MODELS: - self._dimensions = self.DEFAULT_MODELS[self.model] - + self.model = config.model or "text-embedding-3-small" + self.base_url = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1") + self._dimensions = self.DEFAULT_MODELS.get(self.model, config.dimensions) + @property def dimensions(self) -> int: return self._dimensions - - def embed(self, text: str) -> List[float]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", } - + + def embed(self, text: str) -> List[float]: response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': text, - }, - headers=headers, - timeout=30 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": text}, + headers=self._headers(), + timeout=30, ) response.raise_for_status() - - result = response.json() - return result['data'][0]['embedding'] - + return response.json()["data"][0]["embedding"] + def embed_batch(self, texts: List[str]) -> List[List[float]]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', - } - response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': texts, - }, - headers=headers, - timeout=60 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": texts}, + headers=self._headers(), + timeout=60, ) response.raise_for_status() - - result = response.json() - embeddings = sorted(result['data'], key=lambda x: x['index']) - return [e['embedding'] for e in embeddings] - + embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) + return [item["embedding"] for item in embeddings] + def is_available(self) -> bool: if not self.api_key: return False try: response = requests.get( - f'{self.base_url}/models', - headers={'Authorization': f'Bearer {self.api_key}'}, - timeout=10 + f"{self.base_url}/models", + headers={"Authorization": f"Bearer {self.api_key}"}, + timeout=10, ) return response.status_code == 200 except Exception: @@ -296,47 +232,39 @@ def is_available(self) -> bool: class OllamaEmbedder(BaseEmbedder): """Local Ollama embedder.""" - + def __init__(self, config: EmbedderConfig): self.config = config - self.model = config.model or 'nomic-embed-text' + self.model = config.model or "nomic-embed-text" self.base_url = config.ollama_base_url self.dims = config.dimensions - + @property def dimensions(self) -> int: return self.dims - + def embed(self, text: str) -> List[float]: response = requests.post( - f'{self.base_url}/api/embeddings', - json={ - 'model': self.model, - 'prompt': text - }, - timeout=30 + f"{self.base_url}/api/embeddings", + json={"model": self.model, "prompt": text}, + timeout=30, ) response.raise_for_status() - - result = response.json() - return result.get('embedding', []) - + return response.json().get("embedding", []) + def embed_batch(self, texts: List[str]) -> List[List[float]]: embeddings = [] - for text in texts: try: - embedding = self.embed(text) - embeddings.append(embedding) - except Exception as e: - print(f"Error embedding text: {e}") + embeddings.append(self.embed(text)) + except Exception as exc: + print(f"Error embedding text: {exc}") embeddings.append([0.0] * self.dimensions) - return embeddings - + def is_available(self) -> bool: try: - response = requests.get(f'{self.base_url}/api/tags', timeout=5) + response = requests.get(f"{self.base_url}/api/tags", timeout=5) return response.status_code == 200 except Exception: return False @@ -344,67 +272,53 @@ def is_available(self) -> bool: class CustomEmbedder(BaseEmbedder): """Custom OpenAI-compatible API embedder.""" - + def __init__(self, config: EmbedderConfig): self.config = config - self.model = config.model or 'text-embedding-ada-002' + self.model = config.model or "text-embedding-ada-002" self.base_url = config.custom_base_url self.api_key = config.custom_api_key self.dims = config.dimensions - + @property def dimensions(self) -> int: return self.dims - - def embed(self, text: str) -> List[float]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", } - + + def embed(self, text: str) -> List[float]: response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': text, - }, - headers=headers, - timeout=30 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": text}, + headers=self._headers(), + timeout=30, ) response.raise_for_status() - - result = response.json() - return result['data'][0]['embedding'] - + return response.json()["data"][0]["embedding"] + def embed_batch(self, texts: List[str]) -> List[List[float]]: - headers = { - 'Authorization': f'Bearer {self.api_key}', - 'Content-Type': 'application/json', - } - response = requests.post( - f'{self.base_url}/embeddings', - json={ - 'model': self.model, - 'input': texts, - }, - headers=headers, - timeout=60 + f"{self.base_url}/embeddings", + json={"model": self.model, "input": texts}, + headers=self._headers(), + timeout=60, ) response.raise_for_status() - - result = response.json() - embeddings = sorted(result['data'], key=lambda x: x['index']) - return [e['embedding'] for e in embeddings] - + embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) + return [item["embedding"] for item in embeddings] + def is_available(self) -> bool: if not self.base_url or not self.api_key: return False try: response = requests.get( - f'{self.base_url}/models', - headers={'Authorization': f'Bearer {self.api_key}'}, - timeout=10 + f"{self.base_url}/models", + headers={"Authorization": f"Bearer {self.api_key}"}, + timeout=10, ) return response.status_code == 200 except Exception: @@ -413,34 +327,29 @@ def is_available(self) -> bool: class EmbedderFactory: """Factory for creating embedders based on configuration.""" - + @staticmethod def create(config: EmbedderConfig = None) -> BaseEmbedder: - """Create an embedder based on config.""" if config is None: config = EmbedderConfig.get_instance() - + provider = config.provider.lower() - - if provider == 'openrouter': + if provider == "openrouter": return OpenRouterEmbedder(config) - elif provider == 'openai': + if provider == "openai": return OpenAIEmbedder(config) - elif provider == 'ollama': + if provider == "ollama": return OllamaEmbedder(config) - elif provider == 'custom': + if provider == "custom": return CustomEmbedder(config) - else: - # Default to OpenRouter - return OpenRouterEmbedder(config) + return OpenRouterEmbedder(config) -# Global embedder instance _embedder: Optional[BaseEmbedder] = None def reset_embedder(): - """Reset the global embedder instance (e.g., after config change).""" + """Reset the global embedder instance (e.g. after config change).""" global _embedder _embedder = None @@ -448,26 +357,67 @@ def reset_embedder(): def get_embedder(config_path: str = None) -> BaseEmbedder: """Get the global embedder instance.""" global _embedder - if _embedder is None: if config_path: EmbedderConfig.reset() config = EmbedderConfig.get_instance(config_path) _embedder = EmbedderFactory.create(config) - return _embedder +def _chunk_text(text: str, max_chars: int, overlap: int) -> List[str]: + """Split text into bounded overlapping character chunks.""" + if len(text) <= max_chars: + return [text] + + chunks: List[str] = [] + start = 0 + while start < len(text): + end = min(start + max_chars, len(text)) + chunks.append(text[start:end]) + if end == len(text): + break + start = end - overlap + return chunks + + +def _pool_embeddings(embeddings: List[List[float]]) -> List[float]: + """Mean-pool chunk embeddings and L2-normalize the resulting vector.""" + if not embeddings or not embeddings[0]: + raise ValueError("embedding provider returned an empty vector") + + dimensions = len(embeddings[0]) + if any(len(embedding) != dimensions for embedding in embeddings): + raise ValueError("embedding provider returned inconsistent dimensions") + + pooled = [ + sum(embedding[index] for embedding in embeddings) / len(embeddings) + for index in range(dimensions) + ] + norm = math.sqrt(sum(value * value for value in pooled)) + if norm == 0: + raise ValueError("embedding provider returned only zero vectors") + return [value / norm for value in pooled] + + def create_embedding(text: str) -> List[float]: - """Convenience function to create an embedding.""" - return get_embedder().embed(text) + """Create one embedding, chunking oversized input before provider calls.""" + embedder = get_embedder() + config = getattr(embedder, "config", EmbedderConfig.get_instance()) + max_chars = getattr(config, "max_chars", DEFAULT_MAX_CHARS) + overlap = getattr(config, "chunk_overlap", DEFAULT_CHUNK_OVERLAP) + chunks = _chunk_text(text, max_chars, overlap) + if len(chunks) == 1: + return embedder.embed(chunks[0]) + + # Use individual calls so any provider failure propagates. Callers that allow + # degraded storage can then persist NULL and explicitly regenerate later. + return _pool_embeddings([embedder.embed(chunk) for chunk in chunks]) def create_embeddings(texts: List[str]) -> List[List[float]]: - """Convenience function to create multiple embeddings.""" - return get_embedder().embed_batch(texts) + """Create multiple embeddings with the same chunking guarantees.""" + return [create_embedding(text) for text in texts] -# For backward compatibility -OllamaEmbedder = OllamaEmbedder OllamaConfig = EmbedderConfig From 0390d7d3d20532059c5ff5194dbc1de628731ca5 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:17:42 +0200 Subject: [PATCH 04/13] Expose REST embedding regeneration --- src/api/main.py | 50 +++++++++++++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/src/api/main.py b/src/api/main.py index 79dc760..388f0df 100644 --- a/src/api/main.py +++ b/src/api/main.py @@ -4,6 +4,7 @@ """ import logging import os +import uuid from contextlib import asynccontextmanager from datetime import datetime from typing import List, Optional @@ -30,6 +31,7 @@ ) from ..db.queries import get_memory_stats from ..embedder import create_embedding +from ..embedding_regeneration import regenerate_memory_embedding from ..extractors.entities import extract_entities from ..extractors.tagger import get_tagger from ..version import get_version @@ -105,26 +107,23 @@ class SearchRequest(BaseModel): captured_by: Optional[List[str]] = None +class EmbeddingRegenerationRequest(BaseModel): + force: bool = False + + @app.get("/") async def root(): - """Root endpoint.""" - return { - "name": "Open Brain API", - "version": VERSION, - "docs": "/docs", - } + return {"name": "Open Brain API", "version": VERSION, "docs": "/docs"} @app.get("/health") @app.get("/health/live") async def health(): - """Liveness probe: the process and event loop are responsive.""" return {"status": "healthy"} @app.get("/health/ready") async def readiness(): - """Readiness probe: startup and database dependencies are available.""" startup_error = getattr(app.state, "startup_error", None) if startup_error: return JSONResponse( @@ -147,13 +146,11 @@ async def get_memories( source: Optional[str] = None, captured_by: Optional[str] = None, ): - """Get memories with optional transport and authoring-agent filters.""" return get_recent_memories(limit, offset, source, captured_by) @app.post("/memories", response_model=dict) async def create_memory(memory: MemoryCreate): - """Create a new memory.""" content = memory.content source = memory.source captured_by = memory.captured_by @@ -162,7 +159,6 @@ async def create_memory(memory: MemoryCreate): metadata = memory.metadata entities = extract_entities(content) - tagger = get_tagger() tag_sources = tagger.tag(content, entities, source, user_tags) tags = list(tag_sources.keys()) @@ -195,24 +191,41 @@ async def create_memory(memory: MemoryCreate): @app.get("/memories/{memory_id}", response_model=MemoryResponse) async def get_memory(memory_id: str): - """Get a specific memory by ID.""" - import uuid - try: parsed_id = uuid.UUID(memory_id) except ValueError as exc: raise HTTPException(status_code=400, detail=f"Invalid UUID: {memory_id}") from exc memory = get_memory_by_id(parsed_id) - if not memory: raise HTTPException(status_code=404, detail="Memory not found") - return memory +@app.post("/memories/{memory_id}/regenerate-embedding", response_model=dict) +async def regenerate_embedding_endpoint( + memory_id: str, + request: EmbeddingRegenerationRequest, +): + """Regenerate a failed embedding or force replacement during model migration.""" + try: + parsed_id = uuid.UUID(memory_id) + except ValueError as exc: + raise HTTPException(status_code=400, detail=f"Invalid UUID: {memory_id}") from exc + + try: + return regenerate_memory_embedding(parsed_id, force=request.force) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Memory not found") from exc + except Exception as exc: + logger.warning("embedding regeneration failed: %s", type(exc).__name__) + raise HTTPException( + status_code=503, + detail="Embedding provider unavailable", + ) from exc + + @app.post("/memories/search", response_model=List[MemoryResponse]) async def search_memories_endpoint(search: SearchRequest): - """Search memories by semantic content and structured filters.""" embedding = None if search.query: try: @@ -240,18 +253,15 @@ async def search_memories_endpoint(search: SearchRequest): @app.get("/stats") async def get_stats(): - """Get memory statistics.""" return get_memory_stats() @app.get("/trends") async def get_trends(weeks: int = Query(4, ge=1, le=12)): - """Get trending topics.""" analyzer = TrendAnalyzer() return {"trends": analyzer.get_top_trending(weeks)} @app.get("/report/weekly") async def get_weekly_report(days: int = Query(7, ge=1, le=30)): - """Generate weekly report.""" return {"report": generate_weekly_report(days)} From 1389ecbb5e51995c4ed7007d0efe4029dd268ade Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:18:17 +0200 Subject: [PATCH 05/13] Expose MCP embedding regeneration --- src/main.py | 49 +++++++++++++++++++++++++++++++++++-------------- 1 file changed, 35 insertions(+), 14 deletions(-) diff --git a/src/main.py b/src/main.py index ee38ad2..631ca9b 100644 --- a/src/main.py +++ b/src/main.py @@ -4,6 +4,7 @@ """ import os import sys +import uuid from typing import Any, Dict, List import yaml @@ -20,6 +21,7 @@ get_memory_stats, ) from .embedder import create_embedding +from .embedding_regeneration import regenerate_memory_embedding from .extractors.entities import extract_entities from .extractors.tagger import auto_tag from .analytics.weekly_report import generate_weekly_report @@ -28,10 +30,9 @@ def load_config() -> Dict: """Load configuration from settings.yaml.""" config_path = os.path.join( - os.path.dirname(__file__), - '..', 'config', 'settings.yaml' + os.path.dirname(__file__), "..", "config", "settings.yaml" ) - with open(config_path, 'r') as file: + with open(config_path, "r") as file: return yaml.safe_load(file) @@ -89,6 +90,21 @@ async def list_tools() -> List[Tool]: "required": ["content"], }, ), + Tool( + name="memory_regenerate_embedding", + description=( + "Regenerate a missing embedding. Set force=true to replace an " + "existing embedding after changing provider or model." + ), + inputSchema={ + "type": "object", + "properties": { + "memory_id": {"type": "string", "description": "UUID of the memory"}, + "force": {"type": "boolean", "description": "Replace an existing embedding", "default": False}, + }, + "required": ["memory_id"], + }, + ), Tool( name="memory_get_related", description="Get memories related to a specific memory by ID.", @@ -150,6 +166,8 @@ async def call_tool(name: str, arguments: Any) -> List[TextContent]: return await handle_memory_search(arguments) if name == "memory_store": return await handle_memory_store(arguments) + if name == "memory_regenerate_embedding": + return await handle_memory_regenerate_embedding(arguments) if name == "memory_get_related": return await handle_memory_get_related(arguments) if name == "memory_get_entity": @@ -166,7 +184,6 @@ async def call_tool(name: str, arguments: Any) -> List[TextContent]: async def handle_memory_search(args: Dict) -> List[TextContent]: - """Handle memory_search tool.""" query = args.get("query", "") limit = args.get("limit", 5) sources = args.get("sources") @@ -196,7 +213,6 @@ async def handle_memory_search(args: Dict) -> List[TextContent]: async def handle_memory_store(args: Dict) -> List[TextContent]: - """Handle memory_store tool.""" content = args["content"] source = args.get("source", "mcp") captured_by = args.get("captured_by") @@ -236,12 +252,23 @@ async def handle_memory_store(args: Dict) -> List[TextContent]: )] +async def handle_memory_regenerate_embedding(args: Dict) -> List[TextContent]: + memory_id = args["memory_id"] + try: + parsed_id = uuid.UUID(memory_id) + except ValueError: + return [TextContent(type="text", text=f"Invalid memory ID: {memory_id}")] + + try: + result = regenerate_memory_embedding(parsed_id, force=args.get("force", False)) + except KeyError: + return [TextContent(type="text", text=f"Memory not found: {memory_id}")] + return [TextContent(type="text", text=str(result))] + + async def handle_memory_get_related(args: Dict) -> List[TextContent]: - """Handle memory_get_related tool.""" memory_id = args["memory_id"] limit = args.get("limit", 5) - - import uuid try: parsed_id = uuid.UUID(memory_id) except ValueError: @@ -251,7 +278,6 @@ async def handle_memory_get_related(args: Dict) -> List[TextContent]: async def handle_memory_get_entity(args: Dict) -> List[TextContent]: - """Handle memory_get_entity tool.""" results = get_memories_by_entity( args["entity_type"], args["entity_name"], @@ -261,23 +287,19 @@ async def handle_memory_get_entity(args: Dict) -> List[TextContent]: async def handle_memory_today(args: Dict) -> List[TextContent]: - """Handle memory_today tool.""" results = get_today_memories(args.get("limit", 10)) return [TextContent(type="text", text=format_memory_list(results))] async def handle_memory_stats(args: Dict) -> List[TextContent]: - """Handle memory_stats tool.""" return [TextContent(type="text", text=str(get_memory_stats()))] async def handle_weekly_report(args: Dict) -> List[TextContent]: - """Handle memory_weekly_report tool.""" return [TextContent(type="text", text=generate_weekly_report(args.get("days", 7)))] def format_memory_list(memories: List[Dict]) -> str: - """Format a list of memories for display.""" if not memories: return "No memories found." @@ -295,7 +317,6 @@ def format_memory_list(memories: List[Dict]) -> str: async def main(): - """Main entry point.""" init_server() async with stdio_server() as (read_stream, write_stream): await app.run( From 8f7ef686d84df502130c830834b5733f2e98b994 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:18:42 +0200 Subject: [PATCH 06/13] Test embedding failure and model migration resilience --- tests/test_embedding_resilience.py | 172 +++++++++++++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 tests/test_embedding_resilience.py diff --git a/tests/test_embedding_resilience.py b/tests/test_embedding_resilience.py new file mode 100644 index 0000000..f9aa491 --- /dev/null +++ b/tests/test_embedding_resilience.py @@ -0,0 +1,172 @@ +"""Regression tests for oversized and failed embedding recovery.""" +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import Mock, patch +import asyncio +import uuid + +import pytest +import requests + + +class _FakeEmbedder: + def __init__(self, *, max_chars=8, overlap=2, fail_on_call=None): + self.config = SimpleNamespace( + max_chars=max_chars, + chunk_overlap=overlap, + ) + self.calls = [] + self.fail_on_call = fail_on_call + + def embed(self, text): + self.calls.append(text) + if self.fail_on_call == len(self.calls): + raise requests.ConnectionError("provider down") + return [3.0, 4.0] + + +def _cursor(*, row=None): + cursor = Mock() + cursor.fetchone.return_value = row + + @contextmanager + def manager(): + yield cursor + + return cursor, manager + + +def test_long_embedding_is_chunked_and_pooled(): + from src import embedder + + fake = _FakeEmbedder(max_chars=8, overlap=2) + with patch.object(embedder, "_embedder", fake): + result = embedder.create_embedding("abcdefghijklmnopqr") + + assert len(fake.calls) == 3 + assert all(len(chunk) <= 8 for chunk in fake.calls) + assert fake.calls[0][-2:] == fake.calls[1][:2] + assert result == pytest.approx([0.6, 0.8]) + + +def test_chunk_provider_failure_propagates_instead_of_silently_degrading(): + from src import embedder + + fake = _FakeEmbedder(max_chars=8, overlap=2, fail_on_call=2) + with patch.object(embedder, "_embedder", fake): + with pytest.raises(requests.ConnectionError): + embedder.create_embedding("abcdefghijklmnopqr") + + assert len(fake.calls) == 2 + + +def test_regeneration_retries_memory_after_provider_failure(): + from src import embedding_regeneration as regeneration + + memory_id = uuid.uuid4() + with ( + patch.object( + regeneration, + "get_memory_embedding_target", + return_value={"content": "recover me", "has_embedding": False}, + ), + patch.object(regeneration, "create_embedding", return_value=[0.1, 0.2]) as create, + patch.object(regeneration, "update_memory_embedding", return_value=True) as update, + ): + result = regeneration.regenerate_memory_embedding(memory_id) + + create.assert_called_once_with("recover me") + update.assert_called_once_with(memory_id, [0.1, 0.2], force=False) + assert result["status"] == "regenerated" + + +def test_model_migration_requires_force_before_overwriting_embedding(): + from src import embedding_regeneration as regeneration + + memory_id = uuid.uuid4() + target = {"content": "migrate me", "has_embedding": True} + + with ( + patch.object(regeneration, "get_memory_embedding_target", return_value=target), + patch.object(regeneration, "create_embedding") as create, + patch.object(regeneration, "update_memory_embedding") as update, + ): + result = regeneration.regenerate_memory_embedding(memory_id) + + assert result == { + "id": str(memory_id), + "status": "unchanged", + "reason": "embedding_exists", + } + create.assert_not_called() + update.assert_not_called() + + with ( + patch.object(regeneration, "get_memory_embedding_target", return_value=target), + patch.object(regeneration, "create_embedding", return_value=[0.9, 0.8]) as create, + patch.object(regeneration, "update_memory_embedding", return_value=True) as update, + ): + result = regeneration.regenerate_memory_embedding(memory_id, force=True) + + create.assert_called_once_with("migrate me") + update.assert_called_once_with(memory_id, [0.9, 0.8], force=True) + assert result == { + "id": str(memory_id), + "status": "regenerated", + "force": True, + } + + +def test_provider_failure_does_not_replace_existing_embedding(): + from src import embedding_regeneration as regeneration + + memory_id = uuid.uuid4() + with ( + patch.object( + regeneration, + "get_memory_embedding_target", + return_value={"content": "retry later", "has_embedding": False}, + ), + patch.object( + regeneration, + "create_embedding", + side_effect=requests.ConnectionError("provider down"), + ), + patch.object(regeneration, "update_memory_embedding") as update, + ): + with pytest.raises(requests.ConnectionError): + regeneration.regenerate_memory_embedding(memory_id) + + update.assert_not_called() + + +def test_database_update_is_null_only_unless_forced(): + from src.db import attribution + + memory_id = uuid.uuid4() + cursor, manager = _cursor(row={"id": memory_id}) + with patch.object(attribution, "get_db_cursor", manager): + assert attribution.update_memory_embedding( + memory_id, [0.1, 0.2], force=False + ) + + statement, params = cursor.execute.call_args.args + assert "embedding IS NULL" in statement + assert params == ([0.1, 0.2], memory_id, False) + + +def test_rest_and_mcp_expose_regeneration_controls(): + from src.api.main import EmbeddingRegenerationRequest, app + from src import main + + assert EmbeddingRegenerationRequest(force=True).force is True + assert any( + route.path == "/memories/{memory_id}/regenerate-embedding" + for route in app.routes + ) + + tools = asyncio.run(main.list_tools()) + schemas = {tool.name: tool.inputSchema for tool in tools} + schema = schemas["memory_regenerate_embedding"] + assert schema["required"] == ["memory_id"] + assert schema["properties"]["force"]["default"] is False From 9ef83fa98268698e369d9ba547639e808fa03def Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:18:55 +0200 Subject: [PATCH 07/13] Configure embedding chunk limits --- config/settings.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/config/settings.yaml b/config/settings.yaml index 6a726a4..83dd254 100644 --- a/config/settings.yaml +++ b/config/settings.yaml @@ -9,6 +9,8 @@ embedder: provider: ollama model: nomic-embed-text dimensions: 768 + max_chars: 4000 + chunk_overlap: 200 ollama_base_url: http://ollama:11434 mcp: From 2e2d51dadfad14aa9dcce736dbf1c09ef6398600 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:20:21 +0200 Subject: [PATCH 08/13] Keep REST regeneration diff targeted --- src/api/main.py | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/src/api/main.py b/src/api/main.py index 388f0df..31a54c0 100644 --- a/src/api/main.py +++ b/src/api/main.py @@ -4,7 +4,6 @@ """ import logging import os -import uuid from contextlib import asynccontextmanager from datetime import datetime from typing import List, Optional @@ -113,17 +112,24 @@ class EmbeddingRegenerationRequest(BaseModel): @app.get("/") async def root(): - return {"name": "Open Brain API", "version": VERSION, "docs": "/docs"} + """Root endpoint.""" + return { + "name": "Open Brain API", + "version": VERSION, + "docs": "/docs", + } @app.get("/health") @app.get("/health/live") async def health(): + """Liveness probe: the process and event loop are responsive.""" return {"status": "healthy"} @app.get("/health/ready") async def readiness(): + """Readiness probe: startup and database dependencies are available.""" startup_error = getattr(app.state, "startup_error", None) if startup_error: return JSONResponse( @@ -146,11 +152,13 @@ async def get_memories( source: Optional[str] = None, captured_by: Optional[str] = None, ): + """Get memories with optional transport and authoring-agent filters.""" return get_recent_memories(limit, offset, source, captured_by) @app.post("/memories", response_model=dict) async def create_memory(memory: MemoryCreate): + """Create a new memory.""" content = memory.content source = memory.source captured_by = memory.captured_by @@ -159,6 +167,7 @@ async def create_memory(memory: MemoryCreate): metadata = memory.metadata entities = extract_entities(content) + tagger = get_tagger() tag_sources = tagger.tag(content, entities, source, user_tags) tags = list(tag_sources.keys()) @@ -191,13 +200,18 @@ async def create_memory(memory: MemoryCreate): @app.get("/memories/{memory_id}", response_model=MemoryResponse) async def get_memory(memory_id: str): + """Get a specific memory by ID.""" + import uuid + try: parsed_id = uuid.UUID(memory_id) except ValueError as exc: raise HTTPException(status_code=400, detail=f"Invalid UUID: {memory_id}") from exc memory = get_memory_by_id(parsed_id) + if not memory: raise HTTPException(status_code=404, detail="Memory not found") + return memory @@ -207,6 +221,8 @@ async def regenerate_embedding_endpoint( request: EmbeddingRegenerationRequest, ): """Regenerate a failed embedding or force replacement during model migration.""" + import uuid + try: parsed_id = uuid.UUID(memory_id) except ValueError as exc: @@ -226,6 +242,7 @@ async def regenerate_embedding_endpoint( @app.post("/memories/search", response_model=List[MemoryResponse]) async def search_memories_endpoint(search: SearchRequest): + """Search memories by semantic content and structured filters.""" embedding = None if search.query: try: @@ -253,15 +270,18 @@ async def search_memories_endpoint(search: SearchRequest): @app.get("/stats") async def get_stats(): + """Get memory statistics.""" return get_memory_stats() @app.get("/trends") async def get_trends(weeks: int = Query(4, ge=1, le=12)): + """Get trending topics.""" analyzer = TrendAnalyzer() return {"trends": analyzer.get_top_trending(weeks)} @app.get("/report/weekly") async def get_weekly_report(days: int = Query(7, ge=1, le=30)): + """Generate weekly report.""" return {"report": generate_weekly_report(days)} From 94ae4b4c499b8bba701b04171abf9b6c5a3ae0bf Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:20:57 +0200 Subject: [PATCH 09/13] Keep MCP regeneration diff targeted --- src/main.py | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src/main.py b/src/main.py index 631ca9b..77f102e 100644 --- a/src/main.py +++ b/src/main.py @@ -4,7 +4,6 @@ """ import os import sys -import uuid from typing import Any, Dict, List import yaml @@ -30,9 +29,10 @@ def load_config() -> Dict: """Load configuration from settings.yaml.""" config_path = os.path.join( - os.path.dirname(__file__), "..", "config", "settings.yaml" + os.path.dirname(__file__), + '..', 'config', 'settings.yaml' ) - with open(config_path, "r") as file: + with open(config_path, 'r') as file: return yaml.safe_load(file) @@ -184,6 +184,7 @@ async def call_tool(name: str, arguments: Any) -> List[TextContent]: async def handle_memory_search(args: Dict) -> List[TextContent]: + """Handle memory_search tool.""" query = args.get("query", "") limit = args.get("limit", 5) sources = args.get("sources") @@ -213,6 +214,7 @@ async def handle_memory_search(args: Dict) -> List[TextContent]: async def handle_memory_store(args: Dict) -> List[TextContent]: + """Handle memory_store tool.""" content = args["content"] source = args.get("source", "mcp") captured_by = args.get("captured_by") @@ -253,6 +255,9 @@ async def handle_memory_store(args: Dict) -> List[TextContent]: async def handle_memory_regenerate_embedding(args: Dict) -> List[TextContent]: + """Handle memory_regenerate_embedding tool.""" + import uuid + memory_id = args["memory_id"] try: parsed_id = uuid.UUID(memory_id) @@ -267,8 +272,11 @@ async def handle_memory_regenerate_embedding(args: Dict) -> List[TextContent]: async def handle_memory_get_related(args: Dict) -> List[TextContent]: + """Handle memory_get_related tool.""" memory_id = args["memory_id"] limit = args.get("limit", 5) + + import uuid try: parsed_id = uuid.UUID(memory_id) except ValueError: @@ -278,6 +286,7 @@ async def handle_memory_get_related(args: Dict) -> List[TextContent]: async def handle_memory_get_entity(args: Dict) -> List[TextContent]: + """Handle memory_get_entity tool.""" results = get_memories_by_entity( args["entity_type"], args["entity_name"], @@ -287,19 +296,23 @@ async def handle_memory_get_entity(args: Dict) -> List[TextContent]: async def handle_memory_today(args: Dict) -> List[TextContent]: + """Handle memory_today tool.""" results = get_today_memories(args.get("limit", 10)) return [TextContent(type="text", text=format_memory_list(results))] async def handle_memory_stats(args: Dict) -> List[TextContent]: + """Handle memory_stats tool.""" return [TextContent(type="text", text=str(get_memory_stats()))] async def handle_weekly_report(args: Dict) -> List[TextContent]: + """Handle memory_weekly_report tool.""" return [TextContent(type="text", text=generate_weekly_report(args.get("days", 7)))] def format_memory_list(memories: List[Dict]) -> str: + """Format a list of memories for display.""" if not memories: return "No memories found." @@ -317,6 +330,7 @@ def format_memory_list(memories: List[Dict]) -> str: async def main(): + """Main entry point.""" init_server() async with stdio_server() as (read_stream, write_stream): await app.run( From cd2e1a43de606fb92e44fae7c8e08b41b28ffa7a Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:21:40 +0200 Subject: [PATCH 10/13] Keep chunking change focused --- src/embedder/__init__.py | 460 ++++++++++++++++++++++++--------------- 1 file changed, 289 insertions(+), 171 deletions(-) diff --git a/src/embedder/__init__.py b/src/embedder/__init__.py index 2dcd296..2eb6cfb 100644 --- a/src/embedder/__init__.py +++ b/src/embedder/__init__.py @@ -1,14 +1,38 @@ """ Multi-provider embedder for Open Brain. -Supports OpenRouter, OpenAI, Ollama, and custom OpenAI-compatible providers. +Supports: +- openrouter: OpenRouter API (free tier available) +- openai: OpenAI API +- ollama: Local Ollama +- custom: Any OpenAI-compatible API + +Configure in settings.yaml: +```yaml +embedder: + provider: openrouter # openrouter, openai, ollama, custom + model: nomic-embed-text # or openrouter model + + # OpenRouter (default - free tier available) + openrouter_api_key: ${OPENROUTER_API_KEY} + + # OpenAI (optional) + openai_api_key: ${OPENAI_API_KEY} + + # Ollama (optional) + ollama_base_url: http://localhost:11434 + + # Custom API (optional) + custom_base_url: ${CUSTOM_API_URL} + custom_api_key: ${CUSTOM_API_KEY} + + dimensions: 768 +``` """ -from __future__ import annotations - import math import os -from abc import ABC, abstractmethod from typing import ClassVar, List, Optional +from abc import ABC, abstractmethod import requests import yaml @@ -20,62 +44,79 @@ class EmbedderConfig: """Configuration for embedder with multiple provider support.""" - - _instance: Optional["EmbedderConfig"] = None - + + _instance: Optional['EmbedderConfig'] = None + def __init__(self, config_path: str = None): if config_path is None: config_path = os.path.join( - os.path.dirname(__file__), "..", "..", "config", "settings.yaml" - ) - - with open(config_path, "r") as file: - config = yaml.safe_load(file) - - embedder_cfg = config.get("embedder", {}) - self.provider = embedder_cfg.get("provider", "openrouter") - self.model = embedder_cfg.get("model", "nomic-embed-text") - self.dimensions = embedder_cfg.get("dimensions", 768) - self.max_chars = int( - os.environ.get( - "EMBEDDING_MAX_CHARS", - embedder_cfg.get("max_chars", DEFAULT_MAX_CHARS), + os.path.dirname(__file__), + '..', '..', 'config', 'settings.yaml' ) - ) - self.chunk_overlap = int( - os.environ.get( - "EMBEDDING_CHUNK_OVERLAP", - embedder_cfg.get("chunk_overlap", DEFAULT_CHUNK_OVERLAP), - ) - ) + + # Load config + with open(config_path, 'r') as f: + config = yaml.safe_load(f) + + embedder_cfg = config.get('embedder', {}) + + # Provider settings + self.provider = embedder_cfg.get('provider', 'openrouter') + + # Model + self.model = embedder_cfg.get('model', 'nomic-embed-text') + + # Dimensions + self.dimensions = embedder_cfg.get('dimensions', 768) + + # Chunking guardrail for providers with smaller context windows + self.max_chars = int(os.environ.get( + 'EMBEDDING_MAX_CHARS', + embedder_cfg.get('max_chars', DEFAULT_MAX_CHARS) + )) + self.chunk_overlap = int(os.environ.get( + 'EMBEDDING_CHUNK_OVERLAP', + embedder_cfg.get('chunk_overlap', DEFAULT_CHUNK_OVERLAP) + )) if self.max_chars <= 0: raise ValueError("embedder.max_chars must be greater than zero") if self.chunk_overlap < 0 or self.chunk_overlap >= self.max_chars: raise ValueError("embedder.chunk_overlap must be >= 0 and < max_chars") - + + # OpenRouter (default) self.openrouter_api_key = os.environ.get( - "OPENROUTER_API_KEY", embedder_cfg.get("openrouter_api_key", "") + 'OPENROUTER_API_KEY', + embedder_cfg.get('openrouter_api_key', '') ) + + # OpenAI self.openai_api_key = os.environ.get( - "OPENAI_API_KEY", embedder_cfg.get("openai_api_key", "") + 'OPENAI_API_KEY', + embedder_cfg.get('openai_api_key', '') ) + + # Ollama self.ollama_base_url = embedder_cfg.get( - "ollama_base_url", - os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434"), + 'ollama_base_url', + os.environ.get('OLLAMA_BASE_URL', 'http://localhost:11434') ) + + # Custom (any OpenAI-compatible API) self.custom_base_url = os.environ.get( - "CUSTOM_API_URL", embedder_cfg.get("custom_base_url", "") + 'CUSTOM_API_URL', + embedder_cfg.get('custom_base_url', '') ) self.custom_api_key = os.environ.get( - "CUSTOM_API_KEY", embedder_cfg.get("custom_api_key", "") + 'CUSTOM_API_KEY', + embedder_cfg.get('custom_api_key', '') ) - + @classmethod - def get_instance(cls, config_path: str = None) -> "EmbedderConfig": + def get_instance(cls, config_path: str = None) -> 'EmbedderConfig': if cls._instance is None: cls._instance = cls(config_path) return cls._instance - + @classmethod def reset(cls): """Reset singleton (useful for testing).""" @@ -84,85 +125,107 @@ def reset(cls): class BaseEmbedder(ABC): """Abstract base class for embedders.""" - + @abstractmethod def embed(self, text: str) -> List[float]: """Generate embedding for a single text.""" - raise NotImplementedError - + pass + @abstractmethod def embed_batch(self, texts: List[str]) -> List[List[float]]: """Generate embeddings for multiple texts.""" - raise NotImplementedError - + pass + @abstractmethod def is_available(self) -> bool: """Check if the provider is available.""" - raise NotImplementedError - + pass + @property @abstractmethod def dimensions(self) -> int: """Embedding dimensions.""" - raise NotImplementedError + pass class OpenRouterEmbedder(BaseEmbedder): - """OpenRouter API embedder.""" - + """OpenRouter API embedder (default, free tier available).""" + + # Default free models on OpenRouter DEFAULT_MODELS: ClassVar[dict[str, int]] = { - "text-embedding-3-small": 1536, - "text-embedding-ada-002": 1536, + 'text-embedding-3-small': 1536, + 'text-embedding-ada-002': 1536, } - + def __init__(self, config: EmbedderConfig): self.config = config self.api_key = config.openrouter_api_key - self.model = config.model or "text-embedding-3-small" - self.base_url = "https://openrouter.ai/api/v1" - self._dimensions = self.DEFAULT_MODELS.get(self.model, config.dimensions) - + self.model = config.model or 'text-embedding-3-small' + self.base_url = 'https://openrouter.ai/api/v1' + + # Get dimensions for model + self._dimensions = config.dimensions + if self.model in self.DEFAULT_MODELS: + self._dimensions = self.DEFAULT_MODELS[self.model] + @property def dimensions(self) -> int: return self._dimensions - - def _headers(self) -> dict[str, str]: - return { - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json", - "HTTP-Referer": "https://openbrain.local", - "X-Title": "Open Brain", - } - + def embed(self, text: str) -> List[float]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + 'HTTP-Referer': 'https://openbrain.local', # Required by OpenRouter + 'X-Title': 'Open Brain', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": text}, - headers=self._headers(), - timeout=30, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': text, + }, + headers=headers, + timeout=30 ) response.raise_for_status() - return response.json()["data"][0]["embedding"] - + + result = response.json() + return result['data'][0]['embedding'] + def embed_batch(self, texts: List[str]) -> List[List[float]]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + 'HTTP-Referer': 'https://openbrain.local', + 'X-Title': 'Open Brain', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": texts}, - headers=self._headers(), - timeout=60, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': texts, + }, + headers=headers, + timeout=60 ) response.raise_for_status() - embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) - return [item["embedding"] for item in embeddings] - + + result = response.json() + # Sort by index to maintain order + embeddings = sorted(result['data'], key=lambda x: x['index']) + return [e['embedding'] for e in embeddings] + def is_available(self) -> bool: if not self.api_key: return False try: response = requests.get( - f"{self.base_url}/models", - headers={"Authorization": f"Bearer {self.api_key}"}, - timeout=10, + f'{self.base_url}/models', + headers={'Authorization': f'Bearer {self.api_key}'}, + timeout=10 ) return response.status_code == 200 except Exception: @@ -171,59 +234,79 @@ def is_available(self) -> bool: class OpenAIEmbedder(BaseEmbedder): """OpenAI API embedder.""" - + DEFAULT_MODELS: ClassVar[dict[str, int]] = { - "text-embedding-3-small": 1536, - "text-embedding-3-large": 3072, - "text-embedding-ada-002": 1536, + 'text-embedding-3-small': 1536, + 'text-embedding-3-large': 3072, + 'text-embedding-ada-002': 1536, } - + def __init__(self, config: EmbedderConfig): self.config = config self.api_key = config.openai_api_key - self.model = config.model or "text-embedding-3-small" - self.base_url = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1") - self._dimensions = self.DEFAULT_MODELS.get(self.model, config.dimensions) - + self.model = config.model or 'text-embedding-3-small' + self.base_url = os.environ.get( + 'OPENAI_BASE_URL', + 'https://api.openai.com/v1' + ) + + self._dimensions = config.dimensions + if self.model in self.DEFAULT_MODELS: + self._dimensions = self.DEFAULT_MODELS[self.model] + @property def dimensions(self) -> int: return self._dimensions - - def _headers(self) -> dict[str, str]: - return { - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json", - } - + def embed(self, text: str) -> List[float]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": text}, - headers=self._headers(), - timeout=30, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': text, + }, + headers=headers, + timeout=30 ) response.raise_for_status() - return response.json()["data"][0]["embedding"] - + + result = response.json() + return result['data'][0]['embedding'] + def embed_batch(self, texts: List[str]) -> List[List[float]]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": texts}, - headers=self._headers(), - timeout=60, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': texts, + }, + headers=headers, + timeout=60 ) response.raise_for_status() - embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) - return [item["embedding"] for item in embeddings] - + + result = response.json() + embeddings = sorted(result['data'], key=lambda x: x['index']) + return [e['embedding'] for e in embeddings] + def is_available(self) -> bool: if not self.api_key: return False try: response = requests.get( - f"{self.base_url}/models", - headers={"Authorization": f"Bearer {self.api_key}"}, - timeout=10, + f'{self.base_url}/models', + headers={'Authorization': f'Bearer {self.api_key}'}, + timeout=10 ) return response.status_code == 200 except Exception: @@ -232,39 +315,47 @@ def is_available(self) -> bool: class OllamaEmbedder(BaseEmbedder): """Local Ollama embedder.""" - + def __init__(self, config: EmbedderConfig): self.config = config - self.model = config.model or "nomic-embed-text" + self.model = config.model or 'nomic-embed-text' self.base_url = config.ollama_base_url self.dims = config.dimensions - + @property def dimensions(self) -> int: return self.dims - + def embed(self, text: str) -> List[float]: response = requests.post( - f"{self.base_url}/api/embeddings", - json={"model": self.model, "prompt": text}, - timeout=30, + f'{self.base_url}/api/embeddings', + json={ + 'model': self.model, + 'prompt': text + }, + timeout=30 ) response.raise_for_status() - return response.json().get("embedding", []) - + + result = response.json() + return result.get('embedding', []) + def embed_batch(self, texts: List[str]) -> List[List[float]]: embeddings = [] + for text in texts: try: - embeddings.append(self.embed(text)) - except Exception as exc: - print(f"Error embedding text: {exc}") + embedding = self.embed(text) + embeddings.append(embedding) + except Exception as e: + print(f"Error embedding text: {e}") embeddings.append([0.0] * self.dimensions) + return embeddings - + def is_available(self) -> bool: try: - response = requests.get(f"{self.base_url}/api/tags", timeout=5) + response = requests.get(f'{self.base_url}/api/tags', timeout=5) return response.status_code == 200 except Exception: return False @@ -272,53 +363,67 @@ def is_available(self) -> bool: class CustomEmbedder(BaseEmbedder): """Custom OpenAI-compatible API embedder.""" - + def __init__(self, config: EmbedderConfig): self.config = config - self.model = config.model or "text-embedding-ada-002" + self.model = config.model or 'text-embedding-ada-002' self.base_url = config.custom_base_url self.api_key = config.custom_api_key self.dims = config.dimensions - + @property def dimensions(self) -> int: return self.dims - - def _headers(self) -> dict[str, str]: - return { - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json", - } - + def embed(self, text: str) -> List[float]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": text}, - headers=self._headers(), - timeout=30, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': text, + }, + headers=headers, + timeout=30 ) response.raise_for_status() - return response.json()["data"][0]["embedding"] - + + result = response.json() + return result['data'][0]['embedding'] + def embed_batch(self, texts: List[str]) -> List[List[float]]: + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Content-Type': 'application/json', + } + response = requests.post( - f"{self.base_url}/embeddings", - json={"model": self.model, "input": texts}, - headers=self._headers(), - timeout=60, + f'{self.base_url}/embeddings', + json={ + 'model': self.model, + 'input': texts, + }, + headers=headers, + timeout=60 ) response.raise_for_status() - embeddings = sorted(response.json()["data"], key=lambda item: item["index"]) - return [item["embedding"] for item in embeddings] - + + result = response.json() + embeddings = sorted(result['data'], key=lambda x: x['index']) + return [e['embedding'] for e in embeddings] + def is_available(self) -> bool: if not self.base_url or not self.api_key: return False try: response = requests.get( - f"{self.base_url}/models", - headers={"Authorization": f"Bearer {self.api_key}"}, - timeout=10, + f'{self.base_url}/models', + headers={'Authorization': f'Bearer {self.api_key}'}, + timeout=10 ) return response.status_code == 200 except Exception: @@ -327,29 +432,34 @@ def is_available(self) -> bool: class EmbedderFactory: """Factory for creating embedders based on configuration.""" - + @staticmethod def create(config: EmbedderConfig = None) -> BaseEmbedder: + """Create an embedder based on config.""" if config is None: config = EmbedderConfig.get_instance() - + provider = config.provider.lower() - if provider == "openrouter": + + if provider == 'openrouter': return OpenRouterEmbedder(config) - if provider == "openai": + elif provider == 'openai': return OpenAIEmbedder(config) - if provider == "ollama": + elif provider == 'ollama': return OllamaEmbedder(config) - if provider == "custom": + elif provider == 'custom': return CustomEmbedder(config) - return OpenRouterEmbedder(config) + else: + # Default to OpenRouter + return OpenRouterEmbedder(config) +# Global embedder instance _embedder: Optional[BaseEmbedder] = None def reset_embedder(): - """Reset the global embedder instance (e.g. after config change).""" + """Reset the global embedder instance (e.g., after config change).""" global _embedder _embedder = None @@ -357,20 +467,22 @@ def reset_embedder(): def get_embedder(config_path: str = None) -> BaseEmbedder: """Get the global embedder instance.""" global _embedder + if _embedder is None: if config_path: EmbedderConfig.reset() config = EmbedderConfig.get_instance(config_path) _embedder = EmbedderFactory.create(config) + return _embedder def _chunk_text(text: str, max_chars: int, overlap: int) -> List[str]: - """Split text into bounded overlapping character chunks.""" + """Split text into bounded overlapping chunks.""" if len(text) <= max_chars: return [text] - chunks: List[str] = [] + chunks = [] start = 0 while start < len(text): end = min(start + max_chars, len(text)) @@ -382,7 +494,7 @@ def _chunk_text(text: str, max_chars: int, overlap: int) -> List[str]: def _pool_embeddings(embeddings: List[List[float]]) -> List[float]: - """Mean-pool chunk embeddings and L2-normalize the resulting vector.""" + """Mean-pool chunk vectors and L2-normalize the result.""" if not embeddings or not embeddings[0]: raise ValueError("embedding provider returned an empty vector") @@ -401,23 +513,29 @@ def _pool_embeddings(embeddings: List[List[float]]) -> List[float]: def create_embedding(text: str) -> List[float]: - """Create one embedding, chunking oversized input before provider calls.""" + """Convenience function to create an embedding with long-input chunking.""" embedder = get_embedder() - config = getattr(embedder, "config", EmbedderConfig.get_instance()) - max_chars = getattr(config, "max_chars", DEFAULT_MAX_CHARS) - overlap = getattr(config, "chunk_overlap", DEFAULT_CHUNK_OVERLAP) + config = getattr(embedder, 'config', None) + if config is None: + config = EmbedderConfig.get_instance() + max_chars = getattr(config, 'max_chars', DEFAULT_MAX_CHARS) + overlap = getattr(config, 'chunk_overlap', DEFAULT_CHUNK_OVERLAP) chunks = _chunk_text(text, max_chars, overlap) + if len(chunks) == 1: return embedder.embed(chunks[0]) - # Use individual calls so any provider failure propagates. Callers that allow - # degraded storage can then persist NULL and explicitly regenerate later. - return _pool_embeddings([embedder.embed(chunk) for chunk in chunks]) + # Embed chunks individually so provider failures are never hidden by a + # provider-specific batch fallback (notably Ollama's zero-vector fallback). + embeddings = [embedder.embed(chunk) for chunk in chunks] + return _pool_embeddings(embeddings) def create_embeddings(texts: List[str]) -> List[List[float]]: - """Create multiple embeddings with the same chunking guarantees.""" + """Convenience function to create multiple embeddings.""" return [create_embedding(text) for text in texts] +# For backward compatibility +OllamaEmbedder = OllamaEmbedder OllamaConfig = EmbedderConfig From 1c067fb4d454d3e02314de763c5368dc630b01c8 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:25:10 +0200 Subject: [PATCH 11/13] Fix existing correctness lint diagnostics --- src/extractors/entities.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/extractors/entities.py b/src/extractors/entities.py index 1fc7435..116756b 100644 --- a/src/extractors/entities.py +++ b/src/extractors/entities.py @@ -42,7 +42,7 @@ def _ensure_nltk_data(): # Network or disk failure is non-fatal; the chunker/tokenizer # will raise a clearer error at the call site if the data # is truly missing. - pass + continue _ensure_nltk_data() From 0fa0f463b8a294f3584e9ea499f9776e478fb078 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:26:09 +0200 Subject: [PATCH 12/13] Fix existing E2E lint diagnostics --- tests/e2e/test_api.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/e2e/test_api.py b/tests/e2e/test_api.py index ee33d9c..ab1ef55 100644 --- a/tests/e2e/test_api.py +++ b/tests/e2e/test_api.py @@ -257,7 +257,7 @@ def test_continuity(): s, _ = req("POST", f"/v1/sessions/{S.session_id}/close", { "summary": "E2E test session — verified identity resolution, event recording, and continuity.", }, expect=(200, 201, 204)) - assert s in (200, 201, 204), f"session close failed" + assert s in (200, 201, 204), "session close failed" def test_context(): @@ -336,7 +336,7 @@ def _copy_markdown_to_container(container_md: str) -> None: ) result = subprocess.run( ["docker", "cp", str(host_md), f"openbrain-api:{container_md}"], - capture_output=True, text=True, timeout=30, + capture_output=True, text=True, timeout=30, check=False, ) if result.returncode != 0: raise FileNotFoundError( @@ -494,4 +494,4 @@ def _parse_argv() -> None: # print FAIL but don't always raise). The summary prints both. test_summary() failures += S.failed - sys.exit(0 if failures == 0 else 1) + sys.exit(0 if failures == 0 else 1) \ No newline at end of file From 096e71f88319a286a4008366c0b0177efad483f3 Mon Sep 17 00:00:00 2001 From: "BenClawBot (AlphaOM3G1)" Date: Thu, 3 Sep 2026 10:29:59 +0200 Subject: [PATCH 13/13] Make embedding dimension migration fresh-db safe --- .../migrations/015_embedding_dim_change.sql | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/db/migrations/015_embedding_dim_change.sql b/src/db/migrations/015_embedding_dim_change.sql index 7ceaf9b..6b74ab5 100644 --- a/src/db/migrations/015_embedding_dim_change.sql +++ b/src/db/migrations/015_embedding_dim_change.sql @@ -6,14 +6,17 @@ -- local Ollama service with nomic-embed-text (768 dimensions). The embedding -- column must match so semantic search can store and compare vectors. -- --- This migration is safe on a fresh database that has not yet stored real --- embeddings. On an installation with previously-stored 1536-dim vectors the --- column would need to be dropped and re-created (loss of prior embeddings); --- that has been intentionally avoided here. +-- The legacy memory table is installed from schema.sql rather than the v2 +-- migration ledger, so a fresh migration-only database may not have it yet. +-- Existing installations still receive the dimension change and index rebuild. +DO $$ +BEGIN + IF to_regclass('public.memory') IS NOT NULL THEN + ALTER TABLE memory ALTER COLUMN embedding TYPE vector(768); -ALTER TABLE memory ALTER COLUMN embedding TYPE vector(768); - --- The HNSW index on embedding needs to be rebuilt to match the new dimension. -DROP INDEX IF EXISTS idx_memory_embedding; -CREATE INDEX IF NOT EXISTS idx_memory_embedding - ON memory USING hnsw (embedding vector_cosine_ops); + -- The HNSW index on embedding needs to be rebuilt to match the new dimension. + DROP INDEX IF EXISTS idx_memory_embedding; + CREATE INDEX IF NOT EXISTS idx_memory_embedding + ON memory USING hnsw (embedding vector_cosine_ops); + END IF; +END $$;