Memory Providers¶

from cognitia.memory import InMemoryMemoryProvider

memory = InMemoryMemoryProvider()

# Store a fact
await memory.upsert_fact("user_1", "name", "Alice")

# Retrieve facts
facts = await memory.get_facts("user_1")
print(facts)  # {"name": "Alice"}

# Store a message
await memory.save_message("user_1", "topic_1", "user", "Hello!")

# Get messages
messages = await memory.get_messages("user_1", "topic_1", limit=10)

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from cognitia.memory import PostgresMemoryProvider

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
session_factory = async_sessionmaker(engine, expire_on_commit=False)

memory = PostgresMemoryProvider(session_factory)

from cognitia.memory import SQLiteMemoryProvider

memory = SQLiteMemoryProvider(db_path="./agent.db")
# Tables are created automatically on first use

# Development
memory = InMemoryMemoryProvider()

# Production
memory = PostgresMemoryProvider(session_factory)

# Your code uses protocols, not concrete classes:
async def save_user_fact(store: FactStore, user_id: str):
    await store.upsert_fact(user_id, "onboarded", "true")

from cognitia.memory.summarizer import TemplateSummaryGenerator
from cognitia.memory.types import MemoryMessage

summarizer = TemplateSummaryGenerator(max_messages=20, max_message_chars=200)

messages = [
    MemoryMessage(role="user", content="What's the weather?"),
    MemoryMessage(role="assistant", content="It's sunny today."),
]
summary = summarizer.summarize(messages)
# "- [user]: What's the weather?\n- [assistant]: It's sunny today."

from cognitia.memory.llm_summarizer import LlmSummaryGenerator

async def my_llm_call(prompt: str, text: str) -> str:
    # Your LLM integration here
    return await call_claude(prompt + "\n\n" + text)

summarizer = LlmSummaryGenerator(llm_call=my_llm_call)

# Sync (delegates to template fallback):
summary = summarizer.summarize(messages)

# Async (calls LLM, falls back on error):
summary = await summarizer.asummarize(messages)

from cognitia.memory.episodic_types import Episode
from cognitia.memory.episodic import InMemoryEpisodicMemory

memory = InMemoryEpisodicMemory()

# Store an episode
episode = Episode(
    id="ep-001",
    summary="User asked to deploy app to staging. Used sandbox tool.",
    key_decisions=("chose blue-green deployment", "skipped canary"),
    tools_used=("sandbox", "web"),
    outcome="success",
    session_id="sess-42",
    tags=("deployment", "staging"),
)
await memory.store(episode)

# Recall by keyword search
results = await memory.recall("deployment staging", top_k=3)

# Recall recent episodes
recent = await memory.recall_recent(n=5)

# Recall by tag
deployments = await memory.recall_by_tag("deployment")

from cognitia.memory.episodic_sqlite import SqliteEpisodicMemory

memory = SqliteEpisodicMemory(db_path="./episodes.db")
# Tables and FTS5 index are created automatically on first use

await memory.store(episode)
results = await memory.recall("deploy staging")  # Uses FTS5 full-text search

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from cognitia.memory.episodic_postgres import PostgresEpisodicMemory

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
session_factory = async_sessionmaker(engine, expire_on_commit=False)

memory = PostgresEpisodicMemory(session_factory)
results = await memory.recall("deploy staging")  # Uses tsvector full-text search

from cognitia.memory.procedural_types import Procedure, ProcedureStep
from cognitia.memory.procedural import InMemoryProceduralMemory

memory = InMemoryProceduralMemory()

# Define a learned procedure
deploy_proc = Procedure(
    id="proc-001",
    name="Deploy to staging",
    description="Full deployment pipeline for staging environment",
    trigger="deploy application to staging",
    steps=(
        ProcedureStep(
            tool_name="sandbox",
            args_template={"command": "run tests"},
            expected_outcome="All tests pass",
        ),
        ProcedureStep(
            tool_name="sandbox",
            args_template={"command": "build docker image"},
            expected_outcome="Image built successfully",
        ),
        ProcedureStep(
            tool_name="sandbox",
            args_template={"command": "deploy to staging"},
            expected_outcome="Deployment complete",
        ),
    ),
    tags=("deployment", "staging"),
)
await memory.store(deploy_proc)

# Suggest procedures for a task
suggestions = await memory.suggest("deploy app to staging", top_k=3)
# Returns procedures ranked by relevance + success rate

# Record outcome to improve future suggestions
await memory.record_outcome("proc-001", success=True)

# Check procedure stats
proc = await memory.get("proc-001")
print(f"Success rate: {proc.success_rate:.0%}")  # "Success rate: 100%"

from cognitia.memory.procedural_sqlite import SqliteProceduralMemory

memory = SqliteProceduralMemory(db_path="./procedures.db")
# Tables and FTS5 index created automatically

await memory.store(deploy_proc)
suggestions = await memory.suggest("deploy staging")

from cognitia.memory.procedural_postgres import PostgresProceduralMemory

memory = PostgresProceduralMemory(session_factory)
# Requires the `procedures` table — schema available as POSTGRES_PROCEDURAL_SCHEMA

class FactExtractor(Protocol):
    async def extract(self, episodes: list[Episode]) -> list[str]: ...

from cognitia.memory.episodic import InMemoryEpisodicMemory
from cognitia.memory.episodic_types import Episode
from cognitia.memory.consolidation import ConsolidationPipeline

episodic = InMemoryEpisodicMemory()

# Store several episodes over time...
await episodic.store(Episode(
    id="ep-1", summary="Deployed to staging using sandbox",
    tools_used=("sandbox",), outcome="success", tags=("deployment",),
))
await episodic.store(Episode(
    id="ep-2", summary="Deployed to production using sandbox",
    tools_used=("sandbox",), outcome="success", tags=("deployment",),
))
await episodic.store(Episode(
    id="ep-3", summary="Ran data migration using sandbox",
    tools_used=("sandbox",), outcome="success", tags=("deployment",),
))

# Run consolidation
pipeline = ConsolidationPipeline(episodic=episodic)
result = await pipeline.consolidate(min_episodes=3, max_episodes=50)

print(result.new_facts)
# ("Tool 'sandbox' is frequently used in successful tasks",
#  "Tasks tagged 'deployment' succeed 3/3 times")
print(result.episodes_processed)  # 3

pipeline = ConsolidationPipeline(
    episodic=episodic,
    fact_store=my_fact_store,  # any object with async add_fact(user_id, fact)
)
result = await pipeline.consolidate()
# New facts are automatically stored in the fact store

from cognitia.memory.consolidation import FactExtractor

class LlmFactExtractor:
    """Extract facts using an LLM for richer semantic understanding."""

    async def extract(self, episodes: list[Episode]) -> list[str]:
        summaries = "\n".join(ep.summary for ep in episodes)
        # Call your LLM to extract patterns
        response = await call_llm(
            f"Extract recurring patterns from these episodes:\n{summaries}"
        )
        return response.split("\n")

pipeline = ConsolidationPipeline(
    episodic=episodic,
    extractor=LlmFactExtractor(),
)