Unified Query Across Vector, Graph, and Relational Stores: One Interface for Agent Memory

Agent memory is rarely homogeneous. A single agent might need to recall a dense vector embedding for semantic similarity, traverse a knowledge graph for relational context, and query a relational table for structured facts. Running separate clients for each store creates code duplication, inconsistent access patterns, and a maintenance burden. This article walks through a concrete architecture for unifying query across Postgres, Qdrant, and Neo4j under a single interface — the approach used in the RiNET stack.

The Problem with Three Clients

A typical agent memory stack might include:

• Relational: PostgreSQL for structured data (user profiles, session logs, transaction records).
• Vector: Qdrant for dense embeddings from BGE-M3 or similar models.
• Graph: Neo4j for entity relationships, provenance, and traversal queries.

Without unification, the agent code looks like this:

# Fragmented query pattern
pg_result = pg_client.execute("SELECT * FROM memories WHERE user_id = %s", (uid,))
qdrant_result = qdrant_client.search(collection="memories", query_vector=vec, limit=5)
neo4j_result = neo4j_client.run("MATCH (m:Memory)-[:RELATES_TO]->(e) WHERE m.id = $id RETURN e", id=mid)

This forces the agent to understand three different query languages, handle three result formats, and coordinate merging logic. It’s brittle and hard to extend.

The Unified Interface: AbstractQuery

The goal is a single AbstractQuery object that can be dispatched to the appropriate store(s) based on the query type. The interface should support:

• Vector similarity (vector_query)
• Graph traversal (graph_query)
• Relational filter (relational_query)
• Hybrid (any combination of the above)

Here’s a minimal interface in Python:

from dataclasses import dataclass, field
from typing import Any, Optional

@dataclass
class Query:
    vector: Optional[list[float]] = None
    graph_pattern: Optional[str] = None
    relational_filter: Optional[dict] = None
    limit: int = 10
    metadata: dict = field(default_factory=dict)

@dataclass
class QueryResult:
    source: str  # 'vector', 'graph', 'relational', 'hybrid'
    payload: list[dict]
    metadata: dict = field(default_factory=dict)

Routing Logic: The QueryDispatcher

The dispatcher inspects the Query and decides which store(s) to hit. For simple cases, only one store is needed. For hybrid queries, it may fan out to multiple stores and merge results.

class QueryDispatcher:
    def __init__(self, stores: dict[str, Any]):
        self.stores = stores

    def dispatch(self, query: Query) -> QueryResult:
        if query.vector and query.relational_filter:
            return self._hybrid_vector_relational(query)
        elif query.graph_pattern:
            return self._graph_only(query)
        elif query.vector:
            return self._vector_only(query)
        elif query.relational_filter:
            return self._relational_only(query)
        else:
            raise ValueError("No query parameters provided")

    def _vector_only(self, query: Query) -> QueryResult:
        results = self.stores['qdrant'].search(
            collection="memories",
            query_vector=query.vector,
            limit=query.limit
        )
        return QueryResult(source='vector', payload=results)

    def _graph_only(self, query: Query) -> QueryResult:
        # graph_pattern is a Cypher query fragment
        cypher = f"MATCH {query.graph_pattern} RETURN * LIMIT {query.limit}"
        results = self.stores['neo4j'].run(cypher).data()
        return QueryResult(source='graph', payload=results)

    def _relational_only(self, query: Query) -> QueryResult:
        # Build SQL from filter dict
        where_clause = " AND ".join(f"{k} = %s" for k in query.relational_filter)
        sql = f"SELECT * FROM memories WHERE {where_clause} LIMIT %s"
        params = list(query.relational_filter.values()) + [query.limit]
        results = self.stores['postgres'].execute(sql, params)
        return QueryResult(source='relational', payload=results)

    def _hybrid_vector_relational(self, query: Query) -> QueryResult:
        # Two-phase: first filter relational, then re-rank by vector
        # Or use a pre-filter approach
        # This is a simplified version
        rel_results = self._relational_only(query)
        # Get IDs from relational results
        ids = [row['id'] for row in rel_results.payload]
        if not ids:
            return QueryResult(source='hybrid', payload=[])
        # Vector search filtered by IDs (if store supports filtering)
        vector_results = self.stores['qdrant'].search(
            collection="memories",
            query_vector=query.vector,
            limit=query.limit,
            filter={"must": [{"key": "id", "match": {"any": ids}}]}
        )
        return QueryResult(source='hybrid', payload=vector_results)

Data Model Alignment

Each store must share a common identifier for cross-store references. In the RiNET stack, every memory entry gets a UUID that is stored as:

• Primary key in Postgres (memories.id)
• Payload field in Qdrant (payload.id)
• Property on Neo4j nodes (memory.id)

This allows the dispatcher to merge results by ID after parallel queries.

Handling Schema Drift

Stores evolve independently. The unified interface should not expose raw schema. Instead, each store registers a schema adapter that maps internal fields to a canonical set:

@dataclass
class SchemaAdapter:
    store_name: str
    field_mappings: dict[str, str]  # canonical -> store-specific

    def to_canonical(self, record: dict) -> dict:
        return {canon: record.get(store_key) for canon, store_key in self.field_mappings.items()}

When returning results, the dispatcher runs each record through the adapter, so the agent always sees the same keys.

Error Handling and Fallbacks

One store may be down. The dispatcher should implement circuit breakers and fallback strategies:

import asyncio
from functools import wraps

def fallback_on_error(fallback_store: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(self, query, *args, **kwargs):
            try:
                return await func(self, query, *args, **kwargs)
            except Exception:
                # Fallback to another store (e.g., relational for vector)
                if fallback_store == 'relational':
                    # Convert vector query to keyword search
                    return await self._relational_keyword_fallback(query)
                raise
        return wrapper
    return decorator

Performance Considerations

Unified query does not mean single query. The dispatcher may issue multiple queries in parallel using asyncio.gather:

async def dispatch_async(self, query: Query) -> QueryResult:
    tasks = []
    if query.vector:
        tasks.append(self._vector_only_async(query))
    if query.graph_pattern:
        tasks.append(self._graph_only_async(query))
    if query.relational_filter:
        tasks.append(self._relational_only_async(query))
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Merge results by ID
    merged = self._merge_results(results)
    return QueryResult(source='hybrid', payload=merged)

Real-World Tradeoffs

• Consistency: Cross-store transactions are hard. The RiNET stack uses an event-driven approach: write to Postgres first, then emit events to update Qdrant and Neo4j asynchronously. This means eventual consistency, which is acceptable for agent memory.
• Query complexity: Graph traversal combined with vector similarity is expensive. A common pattern is to use the graph to narrow down candidate nodes, then vector-search within that subset.
• Maintenance: Adding a new store requires implementing the adapter and registering it in the dispatcher. The interface remains stable.

Conclusion

A unified query interface for vector, graph, and relational stores is not about hiding differences — it’s about managing them systematically. By defining a common query object, a dispatcher, schema adapters, and error handling, you can give your agent a single entry point for memory retrieval, regardless of the underlying store. The code examples above are production patterns from the RiNET stack, running on three Hetzner servers with WireGuard mesh, Qwen on vLLM, BGE-M3 embeddings, and nightly LoRA updates. They work.