The Complete Memory API

import os
import asyncio
from neo4j_agent_memory import MemoryClient, MemorySettings
from neo4j_agent_memory.config import Neo4jConfig, EmbeddingConfig

settings = MemorySettings(
    neo4j=Neo4jConfig(
        uri=os.environ["NEO4J_URI"],
        username=os.environ["NEO4J_USERNAME"],
        password=os.environ["NEO4J_PASSWORD"]
    ),
    embedding=EmbeddingConfig(
        api_key=os.environ["OPENAI_API_KEY"]
    )
)

async with MemoryClient(settings) as memory:
    # List all existing sessions
    sessions = await memory.short_term.list_sessions()
    for s in sessions:
        print(s.session_id)

    # Remove a session and all its messages when the conversation ends
    await memory.short_term.clear_session("user_123")

async with MemoryClient(settings) as memory:
    # Store messages — the session is created automatically on first use
    await memory.short_term.add_message(
        session_id="user_123",
        role="user",
        content="Review Jessica Norris account for a credit limit increase"
    )
    await memory.short_term.add_message(
        session_id="user_123",
        role="assistant",
        content="I will retrieve Jessica's full profile now."
    )

    # Retrieve the full conversation and its messages
    conversation = await memory.short_term.get_conversation("user_123")

    # Semantic search over message history
    results = await memory.short_term.search_messages(
        query="credit limit increase",
        session_id="user_123",
        limit=10
    )

    # Auto-generate a plain-English summary of the conversation
    summary = await memory.short_term.get_conversation_summary("user_123")
    print(summary)

async with MemoryClient(settings) as memory:
    # Store a typed entity using the POLE+O classification
    await memory.long_term.add_entity(
        name="Jessica Norris",
        entity_type="PERSON",
        subtype="CUSTOMER",
        description="High-value customer, flagged for compliance review April 2025",
        properties={"risk_score": 0.415}
    )

    # Store a temporal fact linking two entities
    await memory.long_term.add_fact(
        subject="Jessica Norris",
        predicate="manages",
        object="Acme Corp account",
        valid_from="2024-01-01",
        valid_until="2025-03-31"
    )

    # Store a user or agent preference
    await memory.long_term.add_preference(
        category="communication",
        preference="Prefers concise responses",
        context="Confirmed during onboarding"
    )

    # Semantic search across all entity types
    entities = await memory.long_term.search_entities(
        query="Jessica Norris accounts",
        limit=10
    )

    # Retrieve preferences matching a query
    prefs = await memory.long_term.search_preferences(query="communication")

    # Retrieve the full neighborhood subgraph around an entity (2-hop by default)
    subgraph = await memory.long_term.get_entity_graph(
        entity_id="jessica-norris",
        depth=2
    )

async with MemoryClient(settings) as memory:
    context = await memory.get_context(
        query="What do I know about Jessica Norris?",
        session_id="user_123",
        include_short_term=True,
        include_long_term=True,
        include_reasoning=True,
        max_items=10
    )

    # context is a formatted string ready to inject into an LLM prompt
    print(context)

async with MemoryClient(settings) as memory:
    # 1. Start a new trace — creates the ReasoningTrace node
    trace = await memory.reasoning.start_trace(
        task="Evaluate credit limit for Jessica Norris",
        session_id="user_123"
    )

    # 2. Record a reasoning step
    step = await memory.reasoning.add_step(
        trace_id=trace.id,
        thought="Retrieving customer entity from long-term memory",
        action="search_entities"
    )

    # 3. Record a tool call within the step
    await memory.reasoning.record_tool_call(
        step_id=step.id,
        tool_name="search_entities",
        arguments={"query": "Jessica Norris", "limit": 5},
        result={"entities": ["Jessica Norris (EntityPerson)"]},
        status="success"
    )

    # 4. Complete the trace with an outcome
    await memory.reasoning.complete_trace(
        trace.id,
        outcome="Approved — risk score within threshold",
        success=True
    )

async with MemoryClient(settings) as memory:
    # Find past traces for semantically similar tasks
    similar = await memory.reasoning.get_similar_traces(
        task="What do you know about me?",
        limit=3
    )

    # List all traces, with optional filtering
    traces = await memory.reasoning.list_traces()

    # Retrieve pre-aggregated tool usage statistics
    stats = await memory.reasoning.get_tool_stats()
    for tool_name, count in stats.items():
        print(f"{tool_name}: {count} calls")

    # Retrieve the complete causal chain for one trace
    provenance = await memory.reasoning.get_trace_provenance(trace.id)