ContextBrain

ContextBrain is the “Memory” of the ContextUnity ecosystem. It stores and retrieves knowledge using PostgreSQL + pgvector, providing hybrid semantic search, episodic memory, and taxonomy management.

Vector Storage

PostgreSQL + pgvector for multi-dimensional embeddings supporting semantic similarity search.

Hybrid Search

Combines vector similarity with full-text search and cross-encoder reranking.

Memory System

Semantic, episodic, and entity memory types for AI agent context management.

Taxonomy

ltree-based hierarchical classification with AI-powered categorization.

Architecture

src/contextbrain/
├── service/                       # gRPC service (modular)
│   ├── server.py                  # Server setup
│   ├── brain_service.py           # Main service class
│   ├── commerce_service.py        # Commerce operations
│   ├── embedders.py               # Embedding providers
│   └── handlers/                  # Domain-specific handlers
│       ├── knowledge.py           # Knowledge management
│       ├── memory.py              # Episodic memory
│       ├── taxonomy.py            # Taxonomy operations
│       ├── commerce.py            # Commerce handlers
│       └── news.py                # News engine handlers
│
├── storage/
│   ├── postgres/                  # PostgreSQL + pgvector (primary)
│   │   ├── store/                 # Modular store (mixin pattern)
│   │   │   ├── base.py            # Base connection handling
│   │   │   ├── search.py          # Vector search operations
│   │   │   ├── graph.py           # Graph CRUD operations
│   │   │   ├── episodes.py        # Episodic memory
│   │   │   └── taxonomy.py        # Taxonomy operations
│   │   ├── news.py                # News post storage
│   │   └── schema.py              # Database schema
│   └── duckdb_store.py            # Testing backend
│
├── payloads.py                    # Pydantic validation models
├── ingestion/
│   └── rag/                       # RAG pipeline, processors
└── core/                          # Config, registry, interfaces

gRPC API

Method	Description
Search	Streaming knowledge search
QueryMemory	Hybrid search (vector + text) for knowledge retrieval
Upsert	Store knowledge with embeddings
GraphSearch	Search the Knowledge Graph
CreateKGRelation	Create Knowledge Graph relations
AddEpisode	Add conversation turn to episodic memory
GetRecentEpisodes	Retrieve recent episodes for a session
UpsertFact	Store entity facts (user preferences, etc.)
GetUserFacts	Retrieve user entity facts
UpsertTaxonomy	Sync taxonomy entries
GetTaxonomy	Export taxonomy for a domain
LogTrace	Record an execution trace
GetTraces	Retrieve traces for analysis
UpsertNewsItem	Store news facts
GetNewsItems	Retrieve news by criteria
UpsertNewsPost	Store generated posts
CheckNewsPostExists	Check for duplicate news posts

Quick Start

# As gRPC client
from contextcore import ContextUnit, create_channel_sync
from contextcore import brain_pb2_grpc, context_unit_pb2

channel = create_channel_sync("localhost:50051")
stub = brain_pb2_grpc.BrainServiceStub(channel)

# All RPCs use ContextUnit as the envelope
unit = ContextUnit(
    payload={
        "tenant_id": "my_app",
        "query": "How does PostgreSQL work?",
        "top_k": 5,
    },
    provenance=["client:search"],
)

# QueryMemory returns a stream of ContextUnit
for result_pb in stub.QueryMemory(unit.to_protobuf(context_unit_pb2)):
    result = ContextUnit.from_protobuf(result_pb)
    print(result.payload)

Links

• Knowledge Store — vector storage and semantic search
• Memory System — episodic and entity memory
• Taxonomy & Ontology — hierarchical classification
• RAG Pipeline — retrieval-augmented generation
• Configuration — environment variables
• GitHub Repository