Knowledge Kernel

At the heart of CogMem is the Knowledge Kernel—a persistent, deterministic system responsible for how memory is formed, structured, and maintained. The kernel is not an agent that generates text; it governs the knowledge lifecycle. The kernel operates according to the active MemoryStrategy resolved by the Adaptive Cognitive Plane. While the kernel itself is deterministic, the strategy that governs its behavior can evolve over time based on observed outcomes.

What the Kernel Does

The Knowledge Kernel sits beneath agents and tools, governing how information becomes knowledge and how that knowledge evolves:

Admission Control

Deciding what information is worth remembering

Transformation

Promoting raw data into structured representations (Claims)

Conflict Tracking

Identifying and retaining competing claims

Maintenance

Merging, decaying, refreshing, and reinforcing memories

from kos.kernel import KnowledgeKernel
from kos.core.acp.strategy_resolver import StrategyResolver

resolver = StrategyResolver(strategy_store=strategy_store)
strategy = await resolver.resolve(tenant_id="my-tenant")

kernel = KnowledgeKernel(
    object_store=object_store,
    outbox_store=outbox_store,
    claim_store=claim_store,
    strategy=strategy,  # Strategy-aware kernel
)

# Process incoming information through the kernel
await kernel.ingest(item)
await kernel.extract_claims(passage_ids)
await kernel.detect_conflicts(entity_id)

Kernel Event Model

The kernel logs all decisions as events for debugging, visualization, and evaluation:

Event-Driven

All kernel decisions emit events to the outbox

Idempotent

Operations can be safely retried on failure

Provenance

Every decision is logged for traceability

Inspectable

Event stream can be queried and visualized

Kernel Event Types

class KernelEventType(str, Enum):
    # Ingestion events
    ITEM_INGESTED = "item_ingested"
    PASSAGE_EXTRACTED = "passage_extracted"
    
    # Knowledge formation events
    ENTITY_LINKED = "entity_linked"
    CLAIM_PROPOSED = "claim_proposed"
    CLAIM_ACCEPTED = "claim_accepted"
    CLAIM_CONFLICT_DETECTED = "claim_conflict_detected"
    
    # Artifact events
    ARTIFACT_GENERATED = "artifact_generated"

Agent Subsystem

Agents interact through the kernel but do not own memory themselves. The kernel coordinates specialized agents:

ChunkAgent

Splits items into passages. Chunking parameters are governed by the active MemoryStrategy’s document_policy.

Property	Value
Consumes	`ITEM_UPSERTED`
Emits	`PASSAGES_CREATED`
Writes to	ObjectStore (passages)
Strategy-aware	`document_policy.chunking_mode`, `chunk_size`, `overlap`

from kos.agents.ingest.chunk_agent import ChunkAgent

# ChunkAgent reads document_policy from the resolved strategy
agent = ChunkAgent(
    object_store=object_store,
    outbox_store=outbox_store,
    strategy_resolver=strategy_resolver,
)

EmbedAgent

Generates embeddings for passages.

Property	Value
Consumes	`PASSAGES_CREATED`
Emits	`VECTORS_CREATED`
Writes to	VectorSearchProvider

from kos.agents.enrich.embed_agent import EmbedAgent

agent = EmbedAgent(
    object_store=object_store,
    outbox_store=outbox_store,
    vector_search=vector_search,
    embedder=embedder,
    batch_size=32,
)

IndexTextAgent

Indexes passages in text search.

Property	Value
Consumes	`PASSAGES_CREATED`
Emits	`TEXT_INDEXED`
Writes to	TextSearchProvider

from kos.agents.index.index_text_agent import IndexTextAgent

agent = IndexTextAgent(
    object_store=object_store,
    outbox_store=outbox_store,
    text_search=text_search,
)

EntityExtractAgent

Extracts named entities from passages.

Property	Value
Consumes	`PASSAGES_CREATED`
Emits	`ENTITIES_EXTRACTED`
Writes to	ObjectStore (entities), GraphSearchProvider

from kos.agents.extract.entity_extract_agent import EntityExtractAgent

agent = EntityExtractAgent(
    object_store=object_store,
    outbox_store=outbox_store,
    graph_search=graph_search,
    llm_gateway=llm_gateway,  # Optional
    use_llm=True,
)

ClaimExtractAgent

Extracts structured claims from passages.

Property	Value
Consumes	`PASSAGES_CREATED`
Emits	`CLAIM_PROPOSED`
Writes to	ObjectStore (claims)

from kos.agents.extract.claim_extract_agent import ClaimExtractAgent

agent = ClaimExtractAgent(
    object_store=object_store,
    outbox_store=outbox_store,
    llm_gateway=llm_gateway,
)

Knowledge Formation Pipeline

The Knowledge Kernel orchestrates the transformation of raw data into structured knowledge:

                    ITEM_INGESTED
                          │
                          ▼
               ┌─────────────────────┐
               │   Knowledge Kernel  │
               │  (admission control)│
               └─────────────────────┘
                          │
                          ▼
                  PASSAGE_EXTRACTED
                          │
     ┌────────────────────┼────────────────────┐
     ▼                    ▼                    ▼
┌──────────┐       ┌──────────────┐     ┌──────────────┐
│  Embed   │       │EntityExtract │     │ ClaimExtract │
└──────────┘       └──────────────┘     └──────────────┘
     │                    │                    │
     ▼                    ▼                    ▼
VECTORS_CREATED    ENTITY_LINKED        CLAIM_PROPOSED
                                               │
                                               ▼
                                    ┌─────────────────────┐
                                    │   Conflict Detection│
                                    └─────────────────────┘
                                               │
                              ┌────────────────┴────────────────┐
                              ▼                                 ▼
                      CLAIM_ACCEPTED              CLAIM_CONFLICT_DETECTED

Running Workers

# Start worker to process all events
kos run-worker

# With options
kos run-worker --poll-interval 1.0 --batch-size 10

Creating Custom Agents

from kos.agents.base import BaseAgent
from kos.core.events.event_types import EventType
from kos.core.events.envelope import EventEnvelope

class MyCustomAgent(BaseAgent):
    agent_id = "my_custom_agent"
    consumes_events = [EventType.PASSAGES_CREATED]
    
    async def process_event(self, event: EventEnvelope) -> list[EventEnvelope]:
        # Process the event
        passage_ids = event.payload.get("passage_ids", [])
        
        # Do work...
        
        # Log action for provenance
        await self.log_action(
            tenant_id=event.tenant_id,
            user_id=event.payload.get("user_id"),
            action_type="my_action",
            inputs=passage_ids,
            outputs=["output-1"],
        )
        
        # Optionally emit new events
        return []

Next Steps

Adaptive Cognitive Plane

How the system evolves its own strategies

Knowledge Workflows

See how knowledge becomes artifacts

API Reference

Explore the API

Start Here

Platform

Architecture

Knowledge Kernel

Knowledge Kernel

What the Kernel Does

Admission Control

Transformation

Conflict Tracking

Maintenance

Kernel Event Model

Event-Driven

Idempotent

Provenance

Inspectable

Kernel Event Types

Agent Subsystem

ChunkAgent

EmbedAgent

IndexTextAgent

EntityExtractAgent

ClaimExtractAgent

Knowledge Formation Pipeline

Running Workers

Creating Custom Agents

Next Steps

Adaptive Cognitive Plane

Knowledge Workflows

API Reference

​Knowledge Kernel

​What the Kernel Does

Admission Control

Transformation

Conflict Tracking

Maintenance

​Kernel Event Model

Event-Driven

Idempotent

Provenance

Inspectable

​Kernel Event Types

​Agent Subsystem

​ChunkAgent

​EmbedAgent

​IndexTextAgent

​EntityExtractAgent

​ClaimExtractAgent

​Knowledge Formation Pipeline

​Running Workers

​Creating Custom Agents

​Next Steps

Adaptive Cognitive Plane

Knowledge Workflows

API Reference

Knowledge Kernel

What the Kernel Does

Kernel Event Model

Kernel Event Types

Agent Subsystem

ChunkAgent

EmbedAgent

IndexTextAgent

EntityExtractAgent

ClaimExtractAgent

Knowledge Formation Pipeline

Running Workers

Creating Custom Agents

Next Steps