synap 0.1.0

Cognitive memory architecture for LLM agents

Synap

Cognitive memory architecture for LLM agents.

Synap manages three types of memory — semantic, procedural, and episodic — backed by a shared typed property graph. It resolves the fundamental memory-vs-attention contradiction in transformer-based models: more context degrades reasoning quality. Instead of stuffing everything into the prompt, Synap uses structurally selective retrieval (similarity search finds entry points, then graph traversal returns connected subgraphs instead of flat ranked lists) and output-side enforcement (procedures become output schemas, not instructions).

How is this different?

Most agent memory systems (Mem0, Letta, Zep, LangMem) treat memory as a retrieval problem — store text, find similar text, put it in the prompt. Synap takes a different position:

• Structural enforcement, not instructions. Procedural memory produces output schemas where field ordering is the reasoning procedure. The model must generate evidence before conclusions — enforced by the schema, not by telling it to "think step by step."
• Graph traversal, not flat retrieval. Semantic memory returns connected subgraphs where relationships are explicit. A query about "lumbar fusion requirements" traverses requires and includes edges, not just the top-K similar chunks.
• Self-amending procedures. When the same failure pattern repeats, the consolidation engine generates a new schema field and registers an amended procedure version. The system structurally prevents the mistake from recurring.
• Precision over convenience. Synap is a library, not a managed service. You own the agent loop, the LLM client, and the embedding provider. Memory operations are explicit and auditable.

Installation

pip install synap

# With Kùzu for persistent graph storage (recommended)
pip install synap[kuzu]

# With uv
uv add synap --extra kuzu

Providers

Synap needs two providers you implement — one for embeddings, one for LLM text generation. Here's a minimal example using OpenAI:

import openai

class OpenAIEmbedder:
    def __init__(self, client: openai.AsyncOpenAI, model: str = "text-embedding-3-small"):
        self.client = client
        self.model = model

    async def embed(self, text: str) -> list[float]:
        response = await self.client.embeddings.create(input=text, model=self.model)
        return response.data[0].embedding

    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
        response = await self.client.embeddings.create(input=texts, model=self.model)
        return [item.embedding for item in response.data]


class OpenAILLM:
    def __init__(self, client: openai.AsyncOpenAI, model: str = "gpt-4o"):
        self.client = client
        self.model = model

    async def generate(self, prompt: str, output_schema: dict | None = None) -> str:
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content

Any class matching the EmbeddingProvider and LLMProvider protocols works — no inheritance required. See docs/architecture.md for details.

Quick Start

from synap import (
    CognitiveMemory, CapacityHints, Procedure, EpisodeOutcome,
    SemanticMemory, MemoryGraph,
)

# Create the graph and domain adapter
graph = MemoryGraph()
domain = SemanticMemory(graph=graph, embedding_provider=your_embedder)

# You provide the embedding and LLM providers
memory = CognitiveMemory(
    domain=domain,
    embedding_provider=your_embedder,
    llm_provider=your_llm,
    graph=graph,
    capacity=CapacityHints(max_context_tokens=8192),
)

# Register a procedure — field ordering IS the enforcement
await memory.procedural.register(Procedure(
    task_type="diagnose_bug",
    description="Diagnose a bug from error logs and code context",
    schema={
        "error_classification": {"type": "string"},
        "root_cause": {"type": "string"},
        "fix_proposal": {"type": "string"},
    },
    field_ordering=["error_classification", "root_cause", "fix_proposal"],
    prerequisite_fields={"fix_proposal": ["error_classification", "root_cause"]},
))

# Seed knowledge
await domain.store("Stripe webhook payloads vary by event type; always validate shape")

# Prepare context for an LLM call
ctx = await memory.prepare_call(
    task_description="Diagnose TypeError in payment webhook handler"
)
# ctx.output_schema → enforces: classify error → find root cause → THEN propose fix
# ctx.domain_context → relevant facts from the domain adapter
# ctx.warnings → "Last time you misdiagnosed a similar TypeError..."

# Record what happened (including tool calls if any)
from synap import ToolCall

await memory.record_outcome(
    task_description="Diagnose TypeError in payment webhook handler",
    input_data={"error": "Cannot read property 'amount' of undefined"},
    output={"error_classification": "null reference", "root_cause": "...", "fix_proposal": "..."},
    outcome=EpisodeOutcome.SUCCESS,
    task_type="diagnose_bug",
    tool_calls=[
        ToolCall(
            query="find webhook handler source",
            server="code-search",
            tool_name="search_files",
            parameters={"pattern": "handleWebhook"},
            result_summary="Found src/webhooks/stripe.ts:45",
            success=True,
        ),
    ],
)

Domain Adapters

Synap's semantic layer is pluggable via the SemanticDomain protocol. Every project brings its own knowledge types — contradictions and forces for geopolitical analysis, clinical policies for healthcare, code patterns for dev tools.

from synap.protocols import SemanticDomain
from synap.types import DomainResult, MemoryNode

class MyDomain:
    """Implements SemanticDomain — retrieves and absorbs domain knowledge."""

    async def retrieve(self, task_description, task_type=None, metadata=None):
        # Return domain knowledge relevant to this task
        return [DomainResult(content="...", relevance=0.9, source_id="...")]

    async def absorb(self, insights, source_episodes, metadata=None):
        # Store consolidated insights in your domain's schema
        return "domain_node_id"

SemanticMemory is the built-in generic implementation — text nodes with embeddings and graph traversal. Use it to get started, replace it when your domain needs custom types.

Persistence

By default, the graph lives in memory. Pass a storage backend for persistence:

from synap.backends.kuzu import KuzuBackend
from synap.persistent_graph import PersistentGraph

backend = KuzuBackend("./agent_memory", embedding_dim=768)
graph = PersistentGraph(backend=backend)
domain = SemanticMemory(graph=graph, embedding_provider=your_embedder)

memory = CognitiveMemory(
    domain=domain,
    embedding_provider=your_embedder,
    llm_provider=your_llm,
    graph=graph,
)

For multi-process deployments (web servers, worker pools), use the Postgres backend:

pip install synap[postgres]

import asyncpg
from synap.backends.postgres import PostgresBackend
from synap.persistent_graph import PersistentGraph

pool = await asyncpg.create_pool("postgresql://localhost:5432/mydb")
backend = PostgresBackend(pool, embedding_dim=768)
await backend.init()  # Creates tables (idempotent)
graph = PersistentGraph(backend=backend)

Backend	Graph traversal	Vector search	Persistence	Concurrency
In-memory (default)	Python BFS	Python cosine	None	Single process
KuzuBackend	Native Cypher	Native array_cosine_similarity	File-based	Single process
SQLiteBackend	Python BFS	Python cosine	File-based	Single process
PostgresBackend	Recursive CTE	pgvector <=>	Server-based	Multi-process safe

Documentation

• Architecture & Concepts — How the three memory subsystems work and why
• API Reference — Complete interface documentation
• Bootstrap Guide — Cold start: seeding memory from existing data
• Examples — Geopolitical analysis, healthcare, coding agents

How It Works

Semantic memory is pluggable via the SemanticDomain protocol. The built-in SemanticMemory stores facts as a knowledge graph with retrieval via graph traversal. Projects with domain-specific types (contradictions, policies, etc.) implement the protocol directly.

Procedural memory maps task types to output schemas where field ordering is the procedure. The model must generate intermediate reasoning before conclusions. Enforced structurally, not instructionally.

Episodic memory records agent experiences as cue→content→outcome subgraphs. Failed episodes are boosted during retrieval (more learning signal). Over time, repeated patterns consolidate into domain knowledge or procedural amendments. Episodes can include structured tool call tracking — which MCP server, tool, parameters, and result — enabling consolidation to detect tool usage patterns (wrong tool selection, parameter malformation) and generate procedural amendments.

All three operate on a shared typed property graph. Edges cross partitions — this is how consolidation links episodic experiences to domain facts without a separate join mechanism.

Async-First

All public APIs are async. Synap is designed for integration with async frameworks (FastAPI, Sanic, etc.):

# All operations are awaitable
ctx = await memory.prepare_call("task description")
episode_id = await memory.record_outcome(...)
results = await memory.consolidate()
stats = await memory.stats()

Storage backends stay synchronous (embedded DBs don't benefit from async). PersistentGraph bridges with asyncio.to_thread.

Contributing

Contributions are welcome! See CONTRIBUTING.md for development setup and guidelines.

License

MIT — see LICENSE for details.