orchestrata_ultimate.memory.embedding

Reference › orchestrata_ultimate › memory › embedding

EmbeddingService turns text into vectors and ranks candidates by similarity, so memory search is semantic and not an exact match. The MemoryEngine uses it for stores and searches, and the tools.search module imports it too. The first implementation has no external dependency: compute_embedding lower-cases the text, extracts words, counts their frequency, projects each word into one of dimension slots (default 384) with hash(word) % dimension, and normalises the vector to unit length. Empty or word-less text gives a zero vector. cosine_similarity returns 0.0 for vectors of different length or zero norm. search(query_embedding, candidates, limit, min_similarity) scores (key, vector) pairs, keeps those at or above min_similarity (default 0.7) and returns the best limit as (key, score) pairs. The sentence-model alternative is deferred, as the memory narrative records.

At a glance

Class diagram

Dependency diagram

API

Classes

EmbeddingService

Decorators: traces(spec='010', story='US-004', reqs=['REQ-010-015', 'REQ-010-039']) · Line 16

Embedding generation and similarity search.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/memory/embedding.py

"""Embedding service for semantic memory search.

SPEC-010 FR-009/FR-010: Embedding generation and similarity search.
Uses simple word-frequency vectors for P1 (no external dependencies).
sentence-transformers integration deferred to P2.
"""

import math
import re
from collections import Counter

from orchestrata_ultimate.core.base.traces import traces


@traces(spec="010", story="US-004", reqs=["REQ-010-015", "REQ-010-039"])
class EmbeddingService:
    """Embedding generation and similarity search.

    P1 implementation uses TF-IDF-style word vectors for simplicity
    and zero external dependencies. Produces deterministic, reproducible
    results suitable for testing without GPU or model downloads.
    """

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-042"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-054"])
    def __init__(self, dimension: int = 384):
        self._dimension = dimension
        self._vocabulary: dict[str, int] = {}
        self._idf: dict[str, float] = {}
        self._doc_count = 0

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-043"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-055"])
    def _tokenize(self, text: str) -> list[str]:
        """Tokenize text into lowercase words."""
        return re.findall(r"\b[a-zA-Z][a-zA-Z0-9_]*\b", text.lower())

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-044"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-056"])
    def compute_embedding(self, text: str) -> list[float]:
        """Compute a fixed-dimension embedding vector for text.

        Uses a hash-based projection of word frequencies into a
        fixed-dimension space. This is a simple but effective approach
        for P1 that requires no external models.
        """
        if not text:
            return [0.0] * self._dimension
        tokens = self._tokenize(text)
        if not tokens:
            return [0.0] * self._dimension
        freq = Counter(tokens)
        vector = [0.0] * self._dimension
        for word, count in freq.items():
            idx = hash(word) % self._dimension
            vector[idx] += count
        norm = math.sqrt(sum(v * v for v in vector))
        if norm > 0:
            vector = [v / norm for v in vector]
        return vector

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-045"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-057"])
    def cosine_similarity(self, vec_a: list[float], vec_b: list[float]) -> float:
        """Compute cosine similarity between two vectors."""
        if len(vec_a) != len(vec_b):
            return 0.0
        dot = sum(a * b for a, b in zip(vec_a, vec_b))
        norm_a = math.sqrt(sum(a * a for a in vec_a))
        norm_b = math.sqrt(sum(b * b for b in vec_b))
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return dot / (norm_a * norm_b)

    @traces(spec="010", story="US-010-001", reqs=["REQ-010-046"])
    @traces(spec="017", story="US-010-001", reqs=["REQ-017-058"])
    def search(
        self,
        query_embedding: list[float],
        candidates: list[tuple[str, list[float]]],
        limit: int = 10,
        min_similarity: float = 0.7,
    ) -> list[tuple[str, float]]:
        """Search candidates by cosine similarity to query.

        Args:
            query_embedding: Query vector.
            candidates: List of (key, embedding) tuples.
            limit: Maximum results to return.
            min_similarity: Minimum similarity threshold.

        Returns:
            List of (key, similarity) tuples, sorted by similarity descending.
        """
        scored = []
        for key, emb in candidates:
            sim = self.cosine_similarity(query_embedding, emb)
            if sim >= min_similarity:
                scored.append((key, sim))
        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:limit]