Attach memory to an agent

What this recipe does

Registers a RunEphemeralMemoryConnector under a memory ref and writes/reads a value scoped to a single run, so an agent can keep scratch state across turns without a database.

When to use

• An agent needs a notepad that survives across multiple turns inside one run (e.g. partial plans, running totals, tool call history).
• You want a fast in-process connector for tests and examples.
• You're prototyping a workflow and don't want to spin up Postgres or Chroma yet — you will swap in a real connector later.

When NOT to use

• The data must survive the run — use ThreadMemoryConnector or a persistent backend (memory-pg, memory-chroma, memory-es).
• Multiple agents across processes need shared state — the ephemeral connector is in-process only.

Recipe

"""05 — Attach memory to an agent so it remembers across runs on the same thread.

What this shows
---------------
A :class:`ThreadMemoryConnector` registered against
:class:`InMemoryConnectorRegistry` and wrapped in a
:class:`MemoryConnectorResolver`. The agent declares the memory via
``memory_refs`` in its :class:`AgentNodeData` and its
:class:`AgentConfig`. The orchestrator injects the resolver into the
runner automatically at dispatch time, so the agent's ``run`` method
reads thread-scoped memory *before* the prompt is assembled and writes
its output back after the call.

The example drives the same ``thread_id`` twice: the second run sees
what the first one wrote. That's the real loop agents use for
conversation state.

Run
---
    uv run python examples/05_memory.py
"""

from __future__ import annotations

# Allow python examples/NN_name.py to find the sibling examples/_common.py helper.
import sys as _sys
from pathlib import Path as _Path

_sys.path.insert(0, str(_Path(__file__).resolve().parents[1]))

import asyncio
import sys

from examples._common import print_run_summary, running_service
from examples._contracts import Answer, Question
from zeroth.contracts.graph import (
    AgentNode,
    AgentNodeData,
    DisplayMetadata,
    ExecutionSettings,
    Graph,
)
from zeroth.integrations.memory import (
    ConnectorManifest,
    InMemoryConnectorRegistry,
    MemoryConnectorResolver,
    ThreadMemoryConnector,
)
from zeroth.integrations.memory.governed.models import MemoryScope
from zeroth.runtime.agents import (
    AgentConfig,
    AgentRunner,
    DeterministicProviderAdapter,
    ProviderResponse,
)

MEMORY_REF = "memory://conversation"


def build_graph() -> Graph:
    return Graph(
        graph_id="memory-demo",
        name="Memory demo",
        version=1,
        entry_step="assistant",
        execution_settings=ExecutionSettings(max_total_steps=5),
        nodes=[
            AgentNode(
                node_id="assistant",
                graph_version_ref="memory-demo@1",
                display=DisplayMetadata(title="Assistant with memory"),
                input_contract_ref="contract://question",
                output_contract_ref="contract://answer",
                # WS-C: the node loads and stores memory, so under capability
                # enforcement it must declare MEMORY_READ and MEMORY_WRITE.
                capability_bindings=["memory_read", "memory_write"],
                agent=AgentNodeData(
                    instruction=(
                        "Answer the user briefly. When you've answered before on this "
                        "thread, acknowledge it."
                    ),
                    model_provider="openai/gpt-4o-mini",
                    # Declaring the memory ref on the node tells the
                    # orchestrator to resolve it per run and hand the
                    # binding to the agent runner.
                    memory_refs=[MEMORY_REF],
                    state_persistence={"mode": "thread"},
                    thread_participation="full",
                ),
            ),
        ],
        edges=[],
    )


def build_memory_resolver(service) -> MemoryConnectorResolver:
    """Register an in-process thread-scoped connector and return a resolver."""
    registry = InMemoryConnectorRegistry()
    registry.register(
        MEMORY_REF,
        ConnectorManifest(
            connector_type="thread",
            scope=MemoryScope.THREAD,
        ),
        ThreadMemoryConnector(),
    )
    return MemoryConnectorResolver(
        registry=registry,
        thread_repository=service.thread_repository,
    )


async def main() -> int:
    # Two turns on the same thread: first answer, then acknowledgement.
    provider = DeterministicProviderAdapter(
        responses=[
            ProviderResponse(content={"answer": "The capital of France is Paris."}),
            ProviderResponse(
                content={"answer": "As I said earlier, the capital of France is Paris."}
            ),
        ]
    )
    runner = AgentRunner(
        AgentConfig(
            name="assistant",
            description="Assistant that reads conversation memory.",
            instruction="Answer with memory.",
            model_name="openai/gpt-4o-mini",
            input_model=Question,
            output_model=Answer,
            memory_refs=[MEMORY_REF],
        ),
        provider,
    )

    async with running_service(
        build_graph(),
        contracts={
            "contract://question": Question,
            "contract://answer": Answer,
        },
        agent_runners={"assistant": runner},
    ) as demo:
        # Wire the memory registry into the orchestrator. The orchestrator
        # injects the resolver into any runner whose ``memory_resolver`` is
        # still None (so tests that pre-wire their own resolvers are
        # respected). See RuntimeOrchestrator._dispatch_agent.
        demo.service.orchestrator.memory_resolver = build_memory_resolver(demo.service)

        thread_id = "conversation-1"
        for turn in ("first", "second"):
            run = await demo.service.orchestrator.run_graph(
                demo.service.graph,
                {"question": "What is the capital of France?"},
                deployment_ref=demo.deployment_ref,
                thread_id=thread_id,
            )
            print_run_summary(run, label=f"{turn}-turn")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

How it works

InMemoryConnectorRegistry.register stores a (manifest, connector) tuple under a memory ref. At run time the orchestrator resolves the ref to the live connector and wraps it with ScopedMemoryConnector so MemoryScope.RUN writes automatically target the current run_id. The ephemeral connector keeps everything in a process-local dict — zero infrastructure, zero I/O.

See also

• Usage Guide: memory
• Concept: memory
• Concept: agents