memory_agent

Level 3 · Memory — persistent context across turns using DocumentStore.

Documents written to a store persist between agent turns, and a ContextInjectionNode loads them into the prompt so the model can reason over them. This example writes facts, verifies they survive across simulated turns (including a write mid-conversation), then runs an agent backed by that store.

• Lines: ~44
• Prerequisites: none
• Key concepts: DocumentStore, InMemoryDocumentStore, memory persistence

Run it

Installed via pip? Copy this example into your project with the CLI (no checkout needed):

pip install "symfonic-core[cli]"
symfonic examples add memory_agent
python -m memory_agent

Or, from a source checkout (run from the repo root):

python -m examples.memory_agent

Expected output:

Turn 1 — found 1 doc(s): The project uses DI.
Turn 2 — found 1 doc(s): Phase 5 adds examples.
Agent: Memory recalled.

Full code

"""Memory agent -- what persists across turns, and who can see it.

InMemory only, zero external network calls.
Usage: python -m examples.memory_agent

Public surface used, and nothing else::

    from symfonic import Agent
    from symfonic.agent import FrameworkTenantScope
    from symfonic.capabilities.memory import (
        GraphBackedHms, MemoryAdminService, as_memory_scope, memory_capabilities,
    )
    from symfonic.capabilities.prompting import PromptingCapability
    from symfonic.memory.backends import InMemoryGraphBackend

The previous version wrote three documents into a store, searched it twice,
printed "Turn 1 -- found 1 doc(s)", then built a runtime and ran an agent
whose answer was a canned "Memory recalled.". Nothing connected the store to
the agent. Both halves worked and the sentence they added up to was not
true.

``basic_agent`` shows an agent that remembers in about ten lines. This one
shows the other half a deployment needs: the **operator's** view.
``MemoryAdminService`` reads the same store the turns recall from, so a
dashboard and an agent cannot disagree about what is remembered -- and a
question nobody asked stays absent rather than being invented.
"""

from __future__ import annotations

import asyncio

from symfonic import Agent
from symfonic.agent import FrameworkTenantScope
from symfonic.capabilities.memory import (
    GraphBackedHms,
    MemoryAdminService,
    as_memory_scope,
    memory_capabilities,
)
from symfonic.capabilities.prompting import PromptingCapability
from symfonic.core.testing import MockModelProvider
from symfonic.memory.backends import InMemoryGraphBackend


def agent_for(store: GraphBackedHms, scope: object, answer: str) -> Agent:
    """One agent per turn, bound to one scope.

    Both capabilities are load-bearing. ``memory_capabilities`` records and
    recalls; ``PromptingCapability`` renders what was recalled. Without the
    second the agent stores perfectly and the model reads none of it -- and
    every store-side assertion still passes.
    """
    return Agent(
        MockModelProvider(response=answer),
        instructions="You are the project assistant. Be brief.",
        capabilities=[
            *memory_capabilities(store, scope, limit=5),
            PromptingCapability(sources=[]),
        ],
    )


async def main() -> None:
    store = GraphBackedHms(InMemoryGraphBackend())
    scope = FrameworkTenantScope.root("org", "project-42")
    admin = MemoryAdminService(store)

    await agent_for(store, scope, "Noted.").run("The project uses dependency injection.")
    print(f"Turn 1 stored : {len(await admin.records(as_memory_scope(scope)))} record(s)")

    await agent_for(store, scope, "Noted.").run("All storage is async.")
    print(f"Turn 2 stored : {len(await admin.records(as_memory_scope(scope)))} record(s)")

    third = agent_for(store, scope, "Dependency injection, and async storage.")
    result = await third.run("What do you remember about the project?")
    print("Turn 3 asked  : What do you remember about the project?")
    print(f"Turn 3 answer : {result.text}")

    print()
    print("What the operator sees, from the same store:")
    for record in await admin.records(as_memory_scope(scope)):
        print(f"  [{record.layer}] {str(record.text)[:58]}")

    other = FrameworkTenantScope.root("org", "project-99")
    print()
    print(f"Another tenant: {len(await admin.records(as_memory_scope(other)))} record(s)")
    print("The store is shared; the scope is the boundary, and it is checked")
    print("on every read and every write.")


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

Step by step

1. Populate the store

doc_store = InMemoryDocumentStore()
await doc_store.write(Document(id="fact-1", content="The project uses DI."))
await doc_store.write(Document(id="fact-2", content="All storage is async."))

Each Document has a stable id and a content body. Writing with the same id again would update in place. This store stands in for any persistence backend — swap InMemoryDocumentStore for a Postgres- or Mongo-backed one and the rest of the code is unchanged.

2. Register the store

deps = BaseAgentDeps()
deps.register(ModelProvider, MockModelProvider(response="Memory recalled."))
deps.register(DocumentStore, doc_store)

The store is injected by the DocumentStore protocol. The graph's context nodes resolve it from deps at run time.

3. Confirm persistence across turns

found = await doc_store.search("DI")
print(f"Turn 1 — found {len(found)} doc(s): {found[0].content}")

await doc_store.write(Document(id="fact-3", content="Phase 5 adds examples."))
found2 = await doc_store.search("Phase")
print(f"Turn 2 — found {len(found2)} doc(s): {found2[0].content}")

search(query) returns documents matching the query. The key idea: the write in "Turn 2" is visible to subsequent reads — state accumulates across turns rather than resetting. This is what makes the agent remember within a session.

4. Run the agent over its memory

graph = AgentGraph()
runtime = AgentRuntime(graph=graph, deps=deps)
result = await runtime.run("What do you remember?")
print(f"Agent: {result['final_response']}")

The runtime's context-injection node reads from the same DocumentStore and splices matching documents into the prompt before the model runs — so the model's answer is grounded in what was stored. (With MockModelProvider the reply is canned; with a real provider the documents shape the response.)

DocumentStore vs the 5-layer HMS

This example uses the raw DocumentStore — the simplest persistence primitive. The high-level SymfonicAgent builds on top of it with the full Hierarchical Memory System (Semantic, Episodic, Working, Procedural, Prospective layers), which adds consolidation, spreading activation, and automatic hydration. Start here to understand the primitive; graduate to the HMS for production memory.

What to try next

• Round-trip memory tools (memory_read / memory_write / memory_search) → see agent_with_memory in the Examples Index
• The full memory model → Memory (5-Pentad)

See also

• Memory Concepts
• Conversation Persistence