pgmem docs

This walks the whole arc: wire providers, run a turn loop, form durable memory from a tool call, promote it into the semantic graph, then start a second session and watch the agent recall what happened before. The steps build up one runnable script — Run it shows how to assemble and run it against your own database.

Prerequisites

pgmem installed and the quickstart working.
You've skimmed How a turn flows and Events vs state.

Walk through it

Bring providers

A real agent needs two seams: an Embedder (async embed(texts), a fixed dimension of 1536) for the semantic graph, and an LLMClient (async generate(prompt, *, response_model)) for pgmem's internal work (extraction, compaction summaries, saga briefs). pgmem is provider-agnostic — install a provider extra (pgmem[openai], pgmem[anthropic], pgmem[ollama]) and use its adapter, or implement the two protocols against any model you like.

import asyncio
import os

from pgmem import Authority, Evidence, PgMem, PromotionWorker, SubjectRef

DSN = os.environ["DATABASE_URL"]   # e.g. postgresql://user:pass@host:5432/db

async def main():
    mem = await PgMem.create(DSN, embedder=embedder, llm=llm)

Run a turn

Record the user turn, compile context, call your model with the messages, then record the reply. Only compile_context is on the response path.

    session = await mem.session.create(group_id="acme", user_id="customer-123")
    await session.add_turn(
        role="user",
        content="I'm evaluating the enterprise plan for a July rollout.",
    )

    context = await session.compile_context(query="enterprise plan rollout")
    reply = await generate_reply(context.messages)   # your own model call
    await session.add_turn(role="assistant", content=reply)

Form a claim from a tool event

Record the event as an evidence-backed claim; leave current state to be fetched live.

    claim, _ = await session.observe(
        subject=SubjectRef(kind="customer", identifier="customer-123"),
        predicate="billing.plan",
        value={"cycle": "annual", "price": 1_200},
        authority=Authority.application,
        evidence=[Evidence(kind="tool_event", reference="billing.plan_selected", authority=Authority.application)],
    )

Compact off the turn, then end

After replying, fold the buffer down without blocking, then end the session when the conversation is over. Ending seals state and queues derived projection refresh.

    await session.compact_async()
    await session.end()

Refresh derived projections

Promotion runs in a worker, off the turn. Drain the queue once (in production this runs as a background loop with worker.run()):

    worker = PromotionWorker(mem)
    await worker.run_once()

The claim remains authoritative while the worker refreshes its Tier 4 projections:

    print(await mem.driver.pool.fetchval(
        "SELECT count(*) FROM pgmem.memory_claim_graph_edges WHERE claim_uuid = $1", claim.uuid
    ))

Continue in a new session

Weeks later, a fresh session for the same user. It starts with a clean execution tree, but compile_context retrieves the durable memory — the agent continues naturally.

    later = await mem.session.create(group_id="acme", user_id="customer-123")
    context = await later.compile_context(query="Where did we land on billing?")
    # context.recalled_claims / recalled_semantic carry relevant prior memory.
    await later.end()
    await mem.close()

asyncio.run(main())

Run it

Put the steps above into one file — say first_agent.py — point pgmem at your database, and run it:

export DATABASE_URL=postgresql://user:pass@host:5432/db
python first_agent.py

It walks a session through every tier, promotes the decision into the graph, and the second session recalls it. The graph steps (promotion and search) need an embedder and llm wired — bring your own or a pgmem[...] provider adapter; the pgmem code stays identical either way.

What you built

An agent that forms attributed, durable memory from its interactions and recalls it in later conversations — with retrieval on the turn and all the heavy work (compaction, promotion) off it.

Decide what to remember — route memory with a policy
Use live system-of-record state — don't store what you can fetch
Tune retrieval — strategies, depths, budget
LiveKit / voice agents — the stateless-worker pattern

Your first memory-backed agent