Route memory searches with the hybrid provider
Use HybridMemoryProvider to route message, entity and preference retrieval within a Neo4j-backed MemoryClient. Despite its AgentCore integration package name, this class does not connect to a second AWS AgentCore Memory service or synchronize data between independent stores.
1. Prepare the selected environment
Use Python 3.10+ and a POSIX shell. The recipe uses BoltSettings to select Neo4j explicitly and disables entity extraction so that message persistence can be checked independently. Use a dedicated AuraDB instance with vector-index support and no incompatible existing vectors. Follow the Aura connection setup and copy its connection values into the exports below. These recipes use the published Python 0.7.0 package; provider access and database execution must be verified in your environment.
Create a local folder and virtual environment for the examples on this page. The commands reuse an existing environment without changing its files:
mkdir -p ~/agent-memory-tutorials
cd ~/agent-memory-tutorials
if [ -e .venv ]; then
printf '%s\n' 'Using the existing virtual environment.'
else
python3 -m venv .venv
fi
source .venv/bin/activateExpected: ~/agent-memory-tutorials is your working directory and its virtual environment is active. Install the published SDK with the command below.
Each complete code block labelled Save as names a file to create in this folder using your editor. Copy the entire block, including imports and the entry point. Expand each helper disclosure and use Copy code to copy its full source. Keep all files together so their imports resolve.
When continuing from another tutorial or guide, retain the existing environment, configuration, session files, and .tutorial-state/. Reuse unchanged helper files; compare an existing file before replacing it, and finish any pending cleanup or recovery before changing the code that owns its state.
python -m pip install 'neo4j-agent-memory[openai]==0.7.0'
export NEO4J_URI='neo4j+s://<instance-id>.databases.neo4j.io'
export NEO4J_USERNAME='neo4j'
export NEO4J_PASSWORD='replace-with-your-Aura-password'
export NEO4J_DATABASE='neo4j'
export OPENAI_API_KEY='replace-with-your-provider-key'The OpenAI-backed recipes require access to text-embedding-3-small; it produces 1536-dimensional vectors in this configuration. Configure the selected chat model separately where required. Credentials are read from the environment, never printed by the programs.
2. Read the complete recipe
Create the subdirectory below, then save both complete files under the displayed names. Run commands from agent-memory-tutorials/; Python resolves common from the script’s integrations/ directory.
mkdir -p integrationsThe script imports this shared helper from the same directory. It supplies explicit database settings, closes clients through the calling context manager, and fails when expected records are absent. Task traces record the observable outcome of the call; they do not expose hidden model reasoning.
Complete integrations/common.py
integrations/common.py"""Shared Aura connection through Bolt and readback for integration recipes."""
import os
def settings(embedding=None):
from neo4j_agent_memory import BoltSettings
from neo4j_agent_memory.llm import from_provider
return BoltSettings(
neo4j={
"uri": os.environ["NEO4J_URI"],
"username": os.environ["NEO4J_USERNAME"],
"password": os.environ["NEO4J_PASSWORD"],
"database": os.getenv("NEO4J_DATABASE", "neo4j"),
},
embedding=embedding or from_provider("openai/text-embedding-3-small", kind="embedding"),
extraction={"extractor_type": "none"},
)
async def verify_messages(client, session_id, expected):
conversation = await client.short_term.get_conversation(session_id)
contents = [message.content for message in conversation.messages]
if not all(text in contents for text in expected):
raise RuntimeError(f"Message readback failed for {session_id}")
print(f"Verified stored messages; session={session_id}")
async def recorded_turn(client, session_id, prompt, respond):
"""Record a task outcome; this does not claim to capture hidden model reasoning."""
trace = await client.reasoning.start_trace(session_id=session_id, task=prompt)
try:
await client.short_term.add_message(session_id, "user", prompt, extract_entities=False)
context = await client.get_context(prompt, session_id=session_id)
reply = str(await respond(context))
if not reply.strip():
raise RuntimeError("The framework returned no text")
await client.short_term.add_message(session_id, "assistant", reply, extract_entities=False)
await verify_messages(client, session_id, [prompt, reply])
except Exception as exc:
await client.reasoning.complete_trace(
trace.id, success=False, outcome=f"Run failed: {type(exc).__name__}"
)
raise
await client.reasoning.complete_trace(
trace.id, success=True, outcome="Reply stored and read back"
)
stored = await client.reasoning.get_trace_with_steps(trace.id)
if stored is None or stored.success is not True:
raise RuntimeError(f"Trace readback failed for {trace.id}")
print(f"Verified task trace: {trace.id}")
return replyComplete integrations/hybrid_recipe.py
integrations/hybrid_recipe.py"""Route an explicit message search inside the same Neo4j-backed client."""
import asyncio
from uuid import uuid4
from common import settings
async def exercise(client):
from neo4j_agent_memory.integrations.agentcore import HybridMemoryProvider
provider = HybridMemoryProvider(
client,
namespace="docs",
routing_strategy="explicit",
extract_entities=False,
sync_entities=False,
)
session_id = f"docs-hybrid-{uuid4().hex[:8]}"
content = "The fictional project code is cedar-lantern-47."
stored = await provider.store_memory(session_id, content, memory_type="message")
history = await provider.get_session_memories(session_id)
if not any(item.id == stored.id and item.content == content for item in history):
raise RuntimeError(f"Provider readback failed for {session_id}")
# Message search is not scoped by session on Bolt, so no session_id is passed.
result = await provider.search_memory(
"project code",
memory_types=["message"],
include_entities=False,
include_preferences=False,
include_relationships=False,
threshold=0.0,
)
if not result.memories:
raise RuntimeError("Message search returned nothing; check the vector index and dimensions")
print(f"Verified stored message; session={session_id}")
print(f"Route searched: {result.filters_applied['memory_types_searched']}")
print(f"Search returned {len(result.memories)} candidate(s) from the whole database")
return result
async def main():
from neo4j_agent_memory import MemoryClient
async with MemoryClient(settings()) as client:
await exercise(client)
if __name__ == "__main__":
asyncio.run(main())3. Run and verify persistence
python integrations/hybrid_recipe.pyExpected: Verified stored message, with a session ID, then the searched route and a candidate count. The script verifies persistence by reading the session back with get_session_memories and matching the returned message ID; semantic search rank is not used as the persistence check. It then runs a message search with a zero threshold and stops if it returns nothing, because the provider logs a failed message search, such as a missing vector index or a dimension mismatch, and returns an empty result instead of raising. Route searched reports the kinds the explicit strategy was given; it is not evidence that the search succeeded.
|
On Bolt, |
get_session_memories(session_id) returns a session’s stored messages and get_entity_relationships(entity_name) returns the Bolt relationships around one entity; the recipe uses only the first. On Bolt in 0.7.0, the provider’s clear_session(session_id) matches a sessionId message property that Bolt messages do not carry, so it returns 0 and deletes nothing. client.short_term.clear_session(session_id) removes the session’s conversation and messages. Reasoning traces, steps and tool calls remain; see clear_session for the query that deletes them by session_id.
Troubleshooting and cleanup
Stop when the program raises an exception; a printed model response alone does not verify persistence. Check the configured database, provider access and vector dimensions before retrying. The program prints the session or trace IDs needed for inspection and closes the client. Retained example records remain in the dedicated database; follow the Aura tutorial’s cleanup procedure for that dedicated instance only when you no longer need them. Reusing a session groups records; it does not authorize a user to read them.
Adapt the integration
The supported search kinds are message, entity and preference. store_memory accepts message, preference or fact; it does not accept the old SHORT_TERM/LONG_TERM/EPISODIC examples or create a separate Episode store.
explicit uses the supplied search kinds; all selects all three. auto applies keyword patterns to choose kinds, not a learned classifier. short_term_first searches messages then falls back to entities/preferences only when the first result is empty; long_term_first reverses that order. These are choices inside the same provider, not independent backend selection. Evaluate routing using your own queries and inspect filters_applied rather than claiming it always selects the best source.
extract_entities is a constructor option; it is not a store_memory keyword. sync_entities does not implement cross-store synchronization. When extraction is enabled, wait for the backend’s supported readiness condition before relying on derived entities. Namespaces and user IDs on this adapter are not a universal access-control boundary; entities and preferences do not inherit a session’s authorization automatically.
For relationship enrichment, enable include_relationships only when the underlying Bolt graph surface and expected entities exist; failures or empty results are not evidence of a successful enrichment. See Shared memory consistency and the Strands integration for the larger application context.
See Backend capabilities and scoping before adding hosted or multi-user behavior. Neo4j Agent Memory is a Neo4j Labs project with community support.