DDR-003 — Arquitectura basada en protocolos para Engrama v1.0¶

Versión: 0.2.0 | Fecha: 2026-04-12 | Estado: Propuesto Sustituye: DDR-003 v0.1.0 (búsqueda híbrida solo Neo4j)

Contexto¶

El análisis DAFO identificó siete funcionalidades que Engrama necesita para competir. Implementarlas de forma independiente genera retrabajo: la búsqueda híbrida (#1) cableada directamente a Neo4j entra en conflicto con la abstracción de base de datos (#2); el razonamiento temporal (#3) afecta a cada ruta de consulta; la seguridad (#4) y el multi-ámbito (#6) son transversales a todas las operaciones de almacenamiento.

Este DDR diseña una arquitectura unificada basada en protocolos donde las siete funcionalidades componen de forma limpia. La idea clave: #2 (abstracción de base de datos) es la base, no #1 (búsqueda híbrida). Todas las demás funcionalidades se construyen sobre los protocolos de almacenamiento. Definimos esos protocolos primero, y luego cada funcionalidad se convierte en una capa que se conecta sin refactorizar lo anterior.

Las siete funcionalidades y sus dependencias¶

#5 Benchmarks ← necesita #1, #3
#6 Multi-ámbito ← necesita #2, #4
#4 Seguridad   ← necesita #2, #3
#1 Búsqueda híbrida ← necesita #2, #7
#3 Temporal   ← necesita #2
#7 Agnóstico de LLM ← independiente
#2 Abstracción de BD ← BASE

Orden de implementación: #2 + #7 → #1 + #3 → #4 → #6 → #5

Decisión¶

Reemplazar el acoplamiento directo de Engrama con Neo4j por tres interfaces de protocolo. Cada skill, adaptador y herramienta MCP habla con estos protocolos — nunca con una base de datos específica. Neo4j se convierte en la primera (y predeterminada) implementación.

Parte 1 — Protocolos de almacenamiento (#2)¶

1.1 Protocolo GraphStore¶

Cubre: CRUD de nodos, relaciones, búsqueda de texto completo, consultas de patrones.

from typing import Protocol, Any, runtime_checkable
from datetime import datetime

@runtime_checkable
class GraphStore(Protocol):
    """Abstract interface for graph storage backends."""

    # --- Node operations ---
    async def merge_node(
        self, label: str, key_field: str, key_value: str,
        properties: dict[str, Any],
        embedding: list[float] | None = None,
    ) -> dict[str, Any]:
        """Create or update a node. Always MERGE semantics."""
        ...

    async def get_node(
        self, label: str, key_field: str, key_value: str
    ) -> dict[str, Any] | None:
        """Retrieve a single node by its unique key."""
        ...

    async def delete_node(
        self, label: str, key_field: str, key_value: str,
        soft: bool = True,
    ) -> bool:
        """Delete or archive a node. soft=True sets status='archived'."""
        ...

    # --- Relationship operations ---
    async def merge_relation(
        self, from_label: str, from_key: str, from_value: str,
        rel_type: str,
        to_label: str, to_key: str, to_value: str,
    ) -> dict[str, Any]:
        """Create a relationship (idempotent)."""
        ...

    # --- Query operations ---
    async def get_neighbours(
        self, label: str, key_field: str, key_value: str,
        hops: int = 1, limit: int = 50,
    ) -> list[dict[str, Any]]:
        """Traverse N hops from a node."""
        ...

    async def fulltext_search(
        self, query: str, limit: int = 10,
    ) -> list[dict[str, Any]]:
        """Keyword search across all text properties.
        Returns: [{node, score, label, key}]"""
        ...

    async def run_cypher(
        self, query: str, params: dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """Execute a raw query (backend-specific).
        For reflect patterns that need full query power.
        Backends that don't support Cypher raise NotImplementedError."""
        ...

    # --- Schema operations ---
    async def init_schema(self, schema: "SchemaDefinition") -> None:
        """Apply constraints, indexes, and seed data."""
        ...

    async def health_check(self) -> dict[str, Any]:
        """Return backend status and version info."""
        ...

    # --- Temporal operations (#3) ---
    async def get_node_history(
        self, label: str, key_field: str, key_value: str,
    ) -> list[dict[str, Any]]:
        """Return the temporal history of a node's property changes.
        Each entry: {properties, valid_from, valid_to, ingested_at}"""
        ...

    async def decay_scores(
        self, max_age_days: int = 90, decay_rate: float = 0.01,
    ) -> int:
        """Apply confidence decay to stale nodes.
        Returns count of nodes affected."""
        ...

    # --- Scope operations (#6) ---
    async def set_scope(
        self, scope: "MemoryScope",
    ) -> None:
        """Set the active scope for all subsequent operations.
        Scope filters are applied automatically to every query."""
        ...

1.2 Protocolo VectorStore¶

Cubre: almacenamiento de embeddings, búsqueda por similitud. Puede ser el mismo backend que GraphStore (Neo4j) o uno separado (ChromaDB, pgvector).

@runtime_checkable
class VectorStore(Protocol):
    """Abstract interface for vector similarity search."""

    dimensions: int

    async def store_vectors(
        self, items: list[tuple[str, list[float]]],
    ) -> int:
        """Store embeddings for nodes. items: [(node_id, embedding)].
        Returns count stored."""
        ...

    async def search_vectors(
        self, query_embedding: list[float],
        limit: int = 10,
        scope: "MemoryScope | None" = None,
    ) -> list[dict[str, Any]]:
        """k-ANN similarity search.
        Returns: [{node_id, score, label, key}]"""
        ...

    async def delete_vectors(
        self, node_ids: list[str],
    ) -> int:
        """Remove embeddings for deleted/archived nodes."""
        ...

    async def count(self) -> int:
        """Total vectors stored."""
        ...

1.3 Por qué dos protocolos, no uno¶

Algunos backends implementan ambos (Neo4j tiene grafo nativo + vectores nativos). Otros no (NetworkX no tiene índice vectorial; ChromaDB no tiene grafo). Mantenerlos separados permite combinar backends:

Combinación	GraphStore	VectorStore	Caso de uso
Solo Neo4j	Neo4jGraphStore	Neo4jVectorStore	Por defecto, más simple
Neo4j + Chroma	Neo4jGraphStore	ChromaVectorStore	Mejor rendimiento vectorial
Kuzu + Chroma	KuzuGraphStore	ChromaVectorStore	Embebido, sin Docker
NetworkX + None	NetworkXGraphStore	NullVectorStore	Prototipado sin dependencias
PG+AGE + pgvector	PgGraphStore	PgVectorStore	Postgres único

1.4 Implementación Neo4j (primer backend)¶

El adaptador de Neo4j implementa ambos protocolos. Envuelve exactamente las mismas consultas Cypher que existen hoy en server.py — cero cambios de lógica, solo extracción detrás de la interfaz.

class Neo4jBackend:
    """Implements both GraphStore and VectorStore using Neo4j."""

    def __init__(self, driver: AsyncDriver, config: dict):
        self.driver = driver
        self.config = config
        self._scope: MemoryScope | None = None

    # --- GraphStore ---
    async def merge_node(self, label, key_field, key_value,
                         properties, embedding=None):
        # Same MERGE Cypher as current engine.py
        # + stores embedding property if provided
        ...

    async def fulltext_search(self, query, limit=10):
        # Same db.index.fulltext.queryNodes('memory_search', ...)
        ...

    # --- VectorStore ---
    async def search_vectors(self, query_embedding, limit=10, scope=None):
        # CALL db.index.vector.queryNodes('memory_vectors', $k, $emb)
        ...

    async def store_vectors(self, items):
        # SET n.embedding = $embedding (already on the same node)
        ...

1.5 Configuración¶

# Storage backend
GRAPH_BACKEND=neo4j          # neo4j | kuzu | networkx | postgres
VECTOR_BACKEND=neo4j         # neo4j | chroma | pgvector | none

# Neo4j (when GRAPH_BACKEND=neo4j or VECTOR_BACKEND=neo4j)
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=changeme

# ChromaDB (when VECTOR_BACKEND=chroma)
CHROMA_PATH=./chroma_data    # local persistent directory
CHROMA_COLLECTION=engrama

# Kuzu (when GRAPH_BACKEND=kuzu)
KUZU_PATH=./kuzu_data

La factoría lee .env y devuelve las implementaciones correctas:

def create_stores(config: dict) -> tuple[GraphStore, VectorStore]:
    graph = _create_graph_store(config)
    vector = _create_vector_store(config, graph)
    return graph, vector

Cuando VECTOR_BACKEND coincide con GRAPH_BACKEND (p. ej., ambos neo4j), la factoría devuelve el mismo objeto para los dos — sin conexiones desperdiciadas.

Parte 2 — Proveedor de embeddings (#7)¶

Desacoplado de cualquier proveedor de LLM. Tres implementaciones en el lanzamiento.

@runtime_checkable
class EmbeddingProvider(Protocol):
    dimensions: int
    async def embed(self, text: str) -> list[float]: ...
    async def embed_batch(self, texts: list[str]) -> list[list[float]]: ...
    async def health_check(self) -> bool: ...

Proveedor	Modelo	Dims	Local	Coste
`OllamaProvider`	nomic-embed-text	768	sí	gratuito
`OllamaProvider`	nomic-embed-text-v2-moe	768	sí	gratuito, multilingüe
`OpenAIProvider`	text-embedding-3-small	1536	no	$0,02/1M tokens
`SentenceTransformerProvider`	all-MiniLM-L6-v2	384	sí	gratuito, sin Ollama
`NullProvider`	—	0	—	—

EMBEDDING_PROVIDER=ollama           # ollama | openai | sentence_transformer | none
EMBEDDING_MODEL=nomic-embed-text
EMBEDDING_DIMENSIONS=768
OLLAMA_URL=http://localhost:11434

Representación textual para embedding¶

Cada nodo se embebe a partir de la concatenación de sus propiedades de texto:

def node_to_text(label: str, props: dict) -> str:
    parts = [f"{label}:"]
    parts.append(props.get("name") or props.get("title", ""))
    for field in ("description", "notes", "rationale",
                  "solution", "context", "body"):
        if value := props.get(field):
            parts.append(value)
    return " ".join(parts)

Parte 3 — Motor de búsqueda híbrida (#1)¶

El motor de búsqueda reside en core/search.py. Solo habla con los protocolos — cero código específico de base de datos.

Algoritmo¶

query ──► EmbeddingProvider.embed(query)
               │
     ┌─────────┴──────────┐
     ▼                    ▼
VectorStore          GraphStore
.search_vectors()    .fulltext_search()
     │                    │
     └─────────┬──────────┘
               ▼
         merge by node_id
         normalize scores to [0,1]
               │
               ▼
    final = α·v_score + (1-α)·f_score + β·graph_boost
               │
               ▼
         optional: GraphStore.get_neighbours()
         for top-K results (1-hop expansion)
               │
               ▼
         ranked results

Puntuación¶

@dataclass
class HybridConfig:
    alpha: float = 0.6       # vector weight
    graph_beta: float = 0.15 # graph boost weight
    boost_cap: float = 0.3   # max graph boost per node
    vector_k: int = 20       # candidates from vector search
    fulltext_k: int = 20     # candidates from fulltext search

class HybridSearchEngine:
    def __init__(self, graph: GraphStore, vector: VectorStore,
                 embedder: EmbeddingProvider, config: HybridConfig):
        self.graph = graph
        self.vector = vector
        self.embedder = embedder
        self.config = config

    async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
        # 1. Embed query
        query_vec = await self.embedder.embed(query)

        # 2. Parallel search (or sequential if same backend)
        v_results = await self.vector.search_vectors(
            query_vec, limit=self.config.vector_k
        ) if query_vec else []
        f_results = await self.graph.fulltext_search(
            query, limit=self.config.fulltext_k
        )

        # 3. Merge by node identity
        merged = self._merge(v_results, f_results)

        # 4. Score
        for r in merged:
            r.final_score = (
                self.config.alpha * r.vector_score
                + (1 - self.config.alpha) * r.fulltext_score
                + self.config.graph_beta * r.graph_boost
            )

        # 5. Rank and return
        merged.sort(key=lambda r: r.final_score, reverse=True)
        return merged[:limit]

Degradación elegante¶

Escenario	Comportamiento
`EMBEDDING_PROVIDER=none`	α forzado a 0.0, solo texto completo
`VECTOR_BACKEND=none`	Igual que el anterior
Ollama no está arrancado	Fallback a texto completo + aviso
El nodo no tiene embedding	Aparece solo en resultados de texto completo
Sin índice fulltext (Kuzu)	α forzado a 1.0, solo vectorial

Cero cambios disruptivos. El comportamiento actual (solo texto completo en Neo4j) es la configuración por defecto.

Parte 4 — Razonamiento temporal (#3)¶

Modelo bi-temporal¶

Cada nodo lleva cuatro marcas temporales:

Campo	Significado	Establecido por
`created_at`	Cuándo se creó el nodo por primera vez	engine (ya existe)
`updated_at`	Última fecha de modificación	engine (ya existe)
`valid_from`	Cuándo el hecho se hizo verdadero en el mundo real	usuario/agente (nuevo)
`valid_to`	Cuándo el hecho dejó de ser verdadero	usuario/agente (nuevo)

valid_from y valid_to permiten consultas temporales: "¿Qué tecnologías usaba el proyecto en enero?" — filtrar por valid_from <= date AND (valid_to IS NULL OR valid_to >= date).

Decaimiento de confianza¶

Los nodos obtienen una propiedad confidence (float, 0.0–1.0, por defecto 1.0). El decaimiento se ejecuta en GraphStore.decay_scores():

confidence = initial × exp(-decay_rate × days_since_updated)

El skill reflect ya usa puntuaciones de confianza en los nodos Insight. Extender esto a todos los nodos permite que los resultados de búsqueda prioricen el conocimiento reciente y activamente mantenido sobre las entradas obsoletas.

TTL y ciclo de vida¶

El ForgetSkill existente ya soporta forget_by_ttl() con borrado suave (archivar) y borrado definitivo (purgar). El razonamiento temporal lo extiende con:

Auto-decaimiento programado (CLI: engrama decay --rate 0.01)
Detección de conflictos: cuando engrama_remember actualiza un nodo cuyo valid_to ya está establecido, marcarlo para revisión en lugar de sobreescribir silenciosamente

Adiciones al esquema¶

# Added to all nodes via GraphStore.merge_node()
temporal_fields = {
    "valid_from": "datetime | None",   # when fact became true
    "valid_to": "datetime | None",     # when fact stopped being true
    "confidence": "float",             # 0.0–1.0, decays over time
}

No se necesitan nuevos índices — valid_from y valid_to se filtran en las consultas, no se buscan. El índice de rango existente sobre updated_at cubre el cálculo de decaimiento.

Parte 5 — Seguridad de la memoria (#4)¶

Modelo de amenazas¶

El Top 10 de OWASP para Aplicaciones Agénticas (dic. 2025) clasifica el envenenamiento de memoria como ASI06. El ataque MINJA logra >95% de tasa de inyección con éxito. Para Engrama — dirigido a profesionales de seguridad — esto es crítico a nivel reputacional.

Capas de defensa¶

Capa 1 — Saneamiento de entrada (en el engine, por encima de los stores):

class Sanitiser:
    """Validates and cleans all inputs before they reach storage."""

    def sanitise_properties(self, props: dict) -> dict:
        """Strip injection attempts from property values."""
        ...

    def validate_label(self, label: str) -> str:
        """Whitelist-check against schema labels."""
        ...

    def validate_relation(self, rel_type: str) -> str:
        """Whitelist-check against schema relation types."""
        ...

Capa 2 — Rastreo de procedencia (metadatos en cada escritura):

provenance_fields = {
    "source": "str",         # "mcp" | "sdk" | "cli" | "sync"
    "source_agent": "str",   # which agent wrote this
    "source_session": "str", # session identifier
    "trust_level": "float",  # 0.0–1.0, based on source
}

Capa 3 — Recuperación consciente de la confianza (en el motor de búsqueda):

Los resultados de búsqueda se ponderan por nivel de confianza. Los nodos escritos por fuentes verificadas (sincronización del vault, CLI) obtienen mayor confianza que los procedentes de conversaciones con agentes. La fórmula de puntuación híbrida se extiende:

final = α·vector + (1-α)·fulltext + β·graph_boost + γ·trust_level

Donde γ = 0.1 por defecto. Esto significa que un nodo de baja confianza necesita mayor relevancia semántica/por palabras clave para posicionarse por encima de un nodo de alta confianza.

Capa 4 — Aislamiento por ámbito (véase Parte 6).

Parte 6 — Memoria multi-ámbito (#6)¶

Modelo de ámbito¶

@dataclass
class MemoryScope:
    user_id: str | None = None      # whose memory
    agent_id: str | None = None     # which agent
    session_id: str | None = None   # which conversation
    org_id: str | None = None       # which organisation

Cuando se establece un ámbito vía GraphStore.set_scope(), cada consulta filtra automáticamente por los campos de ámbito. Los nodos creados dentro de un ámbito llevan esos campos como propiedades.

Jerarquía de ámbitos¶

org_id (más amplio)
  └── user_id
        └── agent_id
              └── session_id (más estrecho)

Una consulta con user_id="alice" ve: - Todos los nodos con user_id="alice" (su memoria personal) - Todos los nodos con org_id="acme" y sin user_id (memoria compartida de la organización) - Todos los nodos sin campos de ámbito (memoria global/pública)

Implementación¶

Para Neo4j, los ámbitos se convierten en filtros de propiedades en cada cláusula MATCH. El método GraphStore.set_scope() almacena el ámbito activo, y merge_node() / fulltext_search() / search_vectors() lo aplican automáticamente.

Para la v1, Engrama sigue siendo mono-usuario (los campos de ámbito existen pero por defecto son None). El soporte multi-usuario es un cambio de configuración, no un cambio de código.

Parte 7 — Benchmarks (#5)¶

Benchmarks objetivo¶

Benchmark	Qué mide	Líder (2026)
LOCOMO	Memoria de conversación larga (1.986 preguntas)	MemMachine 91,7%
LongMemEval	Evaluación de memoria a largo plazo (500 preguntas)	Mem0 93,0%

Qué necesitamos antes de hacer benchmarking¶

Búsqueda híbrida (#1) — LOCOMO pone a prueba fuertemente la recuperación semántica
Razonamiento temporal (#3) — LongMemEval prueba preguntas temporales
Un arnés de benchmark que cargue datos de test, ejecute consultas y puntúe los resultados

Objetivos realistas¶

Con búsqueda híbrida (grafo+vectorial+texto completo), Engrama debería apuntar a: - LOCOMO: 70–80% (competitivo, no líder) - LongMemEval: 75–85% (el impulso del grafo ayuda en preguntas temporales)

Incluso puntuaciones modestas publicadas con transparencia establecen credibilidad. El término de impulso del grafo es la ventaja estructural de Engrama — ningún competidor usa la topología del grafo como señal de ranking.

Estructura de directorios revisada¶

engrama/
├── core/
│   ├── protocols.py       # GraphStore, VectorStore, EmbeddingProvider
│   ├── search.py          # HybridSearchEngine (protocol-based)
│   ├── security.py        # Sanitiser, provenance, trust
│   ├── scope.py           # MemoryScope dataclass + filtering
│   ├── temporal.py        # Decay, bi-temporal queries
│   ├── engine.py          # Orchestrator (uses protocols)
│   ├── client.py          # (deprecated, kept for backward compat)
│   └── schema.py          # SchemaDefinition, node dataclasses
│
├── backends/
│   ├── __init__.py        # create_stores() factory
│   ├── neo4j/
│   │   ├── graph.py       # Neo4jGraphStore
│   │   ├── vector.py      # Neo4jVectorStore
│   │   └── backend.py     # Neo4jBackend (unified, implements both)
│   ├── kuzu/              # future
│   ├── networkx/          # future
│   └── null.py            # NullGraphStore, NullVectorStore
│
├── embeddings/
│   ├── __init__.py        # create_provider() factory
│   ├── ollama.py          # OllamaProvider
│   ├── openai.py          # OpenAIProvider
│   ├── sentence_transformer.py
│   └── null.py            # NullProvider
│
├── skills/                # sin cambios — usan protocolos vía engine
├── adapters/              # sin cambios — usan protocolos vía engine
└── ...

Migración desde el código actual¶

La refactorización extrae, no reescribe:

Actual	Se convierte en	Cambio
Lógica de merge en `core/engine.py`	`backends/neo4j/graph.py`	Extraer
Búsqueda fulltext en `core/engine.py`	`backends/neo4j/graph.py`	Extraer
Orquestación en `core/engine.py`	`core/engine.py` (ahora usa protocolos)	Delgado
`adapters/mcp/server.py`	Mismo archivo, usa engine	Mínimo
`skills/*.py`	Mismos archivos, usan engine	Ninguno

Las herramientas MCP y los skills no cambian en absoluto — llaman a métodos engine.*, y el engine delega en los protocolos. El único código que se mueve es el Cypher específico de Neo4j, de engine.py a backends/neo4j/.

Fases de implementación¶

Fase A — Protocolos + extracción de Neo4j (base)¶

Estimado: 4–6h | Sin funcionalidades nuevas, sin regresiones

Crear core/protocols.py con GraphStore, VectorStore, EmbeddingProvider
Crear backends/neo4j/backend.py — extraer Cypher existente de engine.py
Crear backends/null.py — NullGraphStore, NullVectorStore
Crear embeddings/null.py — NullProvider
Crear backends/__init__.py — factoría que lee .env
Refactorizar core/engine.py para aceptar protocolos vía constructor
Actualizar el ciclo de vida del servidor MCP para usar la factoría
Ejecutar los 100 tests existentes — deben pasar sin cambios

Definición de hecho: Todos los tests existentes pasan. Las herramientas MCP funcionan de forma idéntica. Cero cambios visibles para el usuario.

Fase B — Proveedores de embedding (#7)¶

Estimado: 2–3h | Habilita la búsqueda vectorial

Crear embeddings/ollama.py — OllamaProvider
Crear embeddings/openai.py — OpenAIProvider (opcional)
Crear embeddings/sentence_transformer.py (opcional)
Crear embeddings/__init__.py — factoría
Añadir variables .env: EMBEDDING_PROVIDER, EMBEDDING_MODEL, etc.
Tests: mockear API de Ollama, verificar embed/embed_batch

Fase C — Almacenamiento vectorial + búsqueda híbrida (#1)¶

Estimado: 4–5h | La gran funcionalidad

Añadir creación de índice vectorial al init del esquema de Neo4j
Implementar Neo4jVectorStore.store_vectors() y search_vectors()
Modificar engine.merge_node() para embeber + almacenar en una sola llamada
Crear core/search.py — HybridSearchEngine
Actualizar la herramienta MCP engrama_search para usar el motor híbrido
Actualizar engrama_remember para embeber al escribir
CLI: engrama reindex — re-embeber todos los nodos por lotes
Tests: puntuación híbrida, degradación elegante, fallback a texto completo

Fase D — Razonamiento temporal (#3)¶

Estimado: 3–4h

Añadir valid_from, valid_to, confidence a merge_node
Implementar decay_scores() en el backend Neo4j
CLI: engrama decay --rate 0.01 --max-age 90
Modificar recall/search para incorporar confianza en la puntuación
Detección de conflictos en remember (marcar cuando valid_to está establecido)
Tests: cálculo de decaimiento, filtrado temporal

Fase E — Endurecimiento de seguridad (#4)¶

Estimado: 3–4h

Crear core/security.py — clase Sanitiser
Añadir campos de procedencia a merge_node
Añadir trust_level a la fórmula de puntuación
Validación de entrada en todas las entradas de herramientas MCP
Tests: intentos de inyección, rastreo de procedencia

Fase F — Multi-ámbito (#6)¶

Estimado: 2–3h

Crear core/scope.py — dataclass MemoryScope
Añadir campos de ámbito a merge_node
Añadir filtrado por ámbito a todos los métodos de consulta
MCP: parámetros opcionales de ámbito en las herramientas
Tests: aislamiento de ámbito, resolución de jerarquía

Fase G — Benchmarks (#5)¶

Estimado: 3–4h

Arnés de benchmark: cargar datos LOCOMO/LongMemEval
Ejecutar consultas a través del motor de búsqueda híbrida
Puntuar y publicar resultados en docs/benchmarks/
Iterar sobre los parámetros α, β, γ en función de los resultados

Total estimado: 22–29 horas en todas las fases

Referencia de .env (completa)¶

# === Storage backends ===
GRAPH_BACKEND=neo4j
VECTOR_BACKEND=neo4j

# === Neo4j ===
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=CHANGE_ME_BEFORE_FIRST_RUN

# === Embeddings ===
EMBEDDING_PROVIDER=none
EMBEDDING_MODEL=nomic-embed-text
EMBEDDING_DIMENSIONS=768
OLLAMA_URL=http://localhost:11434

# === Hybrid search ===
HYBRID_ALPHA=0.6
HYBRID_GRAPH_BETA=0.15
HYBRID_TRUST_GAMMA=0.1

# === Obsidian ===
VAULT_PATH=

# === Temporal ===
DECAY_RATE=0.01
DECAY_MAX_AGE_DAYS=90

# === Scope (v1: single user, leave empty) ===
DEFAULT_USER_ID=
DEFAULT_ORG_ID=

Consecuencias¶

Positivas¶

Construir una vez, extender para siempre. Los nuevos backends (Kuzu, PostgreSQL+AGE) implementan los protocolos sin tocar skills, adaptadores ni herramientas MCP.
Sin retrabajo. Búsqueda híbrida, temporal, seguridad y multi-ámbito componen todos sobre la misma capa de protocolos. Cada fase añade, nunca refactoriza.
Riesgo de Neo4j mitigado. Si las licencias de Neo4j cambian o se necesita una alternativa más ligera, se intercambia el backend — el resto de Engrama no se ve afectado.
Cero cambios disruptivos. La configuración por defecto reproduce exactamente el comportamiento actual. Cada nueva funcionalidad se activa opcionalmente vía .env.
Testeable de forma aislada. Cada implementación de protocolo se puede probar de forma independiente. NullStore permite tests unitarios puros sin ninguna base de datos.

Negativas¶

Impuesto de abstracción. Una capa extra de indirección entre skills y almacenamiento. Mitigado: los protocolos son delgados (~200 líneas en total), y la implementación de Neo4j envuelve exactamente el mismo Cypher que tenemos hoy.
La Fase A no produce funcionalidades nuevas. La extracción es invisible para los usuarios. Inversión necesaria, pero no ofrece valor inmediato al usuario.
No todos los backends serán iguales. NetworkX no puede ejecutar Cypher; Kuzu tiene una sintaxis de consulta diferente. El método run_cypher() es específico de cada backend y puede lanzar NotImplementedError. Los patrones de reflect que dependen de Cypher complejo necesitarán traducciones por backend — o solo funcionarán en backends compatibles con Cypher.
Lock-in del modelo de embedding dentro de un grafo. Cambiar el modelo de embedding requiere re-indexar todos los nodos. Mitigado por engrama reindex.

Referencias¶

DDR-001: Sistema de clasificación facetada
DDR-002: Sincronización bidireccional y portabilidad del vault
Índices vectoriales de Neo4j: https://neo4j.com/docs/cypher-manual/5/indexes/semantic-indexes/vector-indexes/
nomic-embed-text: https://ollama.com/library/nomic-embed-text
OWASP Agentic AI Top 10: https://genaisecurityproject.com
Mem0 LOCOMO benchmark: https://mem0.ai/research
Arquitectura temporal de Zep: https://arxiv.org/abs/2501.13956