Introduction:
We talk about "AI Memory" constantly. We say our agents "remember" us. But what does that actually look like?
Most of the time, AI memory is just a black box—a hidden vector index or a JSON file buried in a database.
Today, I decided to change that. I built a real-time introspection tool I call the **"Glass Brain"**. It allows me to see my AI's memories forming, connecting, and evolving in real-time.
The Problem: The Invisible Mind:
In Day 8, we integrated **Neo4j** (a Graph Database) with **Mem0**. This gave our agent the ability to store structured data.
Instead of just remembering *"Alice likes Python"* as a vague text chunk, it stored it as:
`(Alice) -[:LIKES]-> (Python)`
This is powerful, but invisible. If the AI starts hallucinating relationships (e.g., believing `(Python) -[:IS_A]-> (Snake)` instead of a programming language), I wouldn't know until the agent failed. I needed a way to debug the AI's mind visually.
The Solution: Streamlit + Agraph :
I chose Streamlit because it's the fastest way to build data apps in Python. Combined with the `streamlit-agraph` library, we can render interactive, physics-based network graphs directly in the browser.
Prerequisites & Setup :
Before we build the visualizer, we need our databases running. Here is the `docker-compose.yml` to spin up **Neo4j** (Graph Store) and **Qdrant** (Vector Store).
1- docker-compose.yml :
services:
vector-db:
image: qdrant/qdrant
ports:
- "6333:6333"
neo4j:
image: neo4j:latest
ports:
- "7474:7474"
- "7687:7687"
environment:
NEO4J_AUTH: neo4j/password
Run it with :
docker-compose up -d
2. Install Python Libraries:
pip install streamlit streamlit-agraph neo4j mem0ai python-dotenv
Here Full Code With Streamlit Page :
import streamlit as st
import os
from dotenv import load_dotenv
from neo4j import GraphDatabase
from streamlit_agraph import agraph, Node, Edge, Config
from mem0 import Memory
# --- 1. Environmental Setup ---
load_dotenv()
# Configuration Variables
NEO_URI = os.getenv("NEO_CONNECTION_URI")
NEO_USER = os.getenv("NEO_USERNAME")
NEO_PASSWORD = os.getenv("NEO_PASSWORD")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
# Mem0 Configuration
mem_config = {
"version": "v1.1",
"embedder": {
"provider": "openai",
"config": {
"api_key": OPENAI_API_KEY,
"model": "text-embedding-3-small"
}
},
"llm": {
"provider": "openai",
"config": {
"api_key": OPENAI_API_KEY,
"model": "gpt-4o-mini"
}
},
"graph_store": {
"provider": "neo4j",
"config": {
"url": NEO_URI,
"username": NEO_USER,
"password": NEO_PASSWORD
}
},
"vector_store": {
"provider": "qdrant",
"config": {
"host": "localhost",
"port": 6333,
}
}
}
# --- 2. Streamlit UI Setup ---
st.set_page_config(layout="wide", page_title="Day 9: The Glass Brain")
# Custom Title & Header
st.title("The Glass Brain")
# --- 3. Sidebar: Controls ---
with st.sidebar:
st.header("Add Memory")
user_input = st.text_area("Tell your agent something:", height=100)
if st.button("Add to Brain", type="primary"):
if user_input:
with st.spinner("Processing memory..."):
try:
m = Memory.from_config(mem_config)
# Using a fixed user_id for demo purposes
m.add(user_input, user_id="user_day9")
st.success("Memory Added!")
st.rerun() # Refresh graph to show new nodes
except Exception as e:
st.error(f"Error adding memory: {e}")
st.markdown("---")
with st.expander("Danger Zone"):
st.warning("This will delete ALL data in the graph!")
if st.button("Clear Graph Memory"):
with st.spinner("Wiping memory..."):
try:
driver = GraphDatabase.driver(NEO_URI, auth=(NEO_USER, NEO_PASSWORD))
with driver.session() as session:
session.run("MATCH (n) DETACH DELETE n")
st.success("Brain wiped successfully!")
st.rerun()
except Exception as e:
st.error(f"Error wiping memory: {e}")
# --- 4. Main Area: Visualization ---
def get_graph_data():
"""Fetches nodes and relationships from Neo4j."""
try:
driver = GraphDatabase.driver(NEO_URI, auth=(NEO_USER, NEO_PASSWORD))
driver.verify_connectivity()
nodes = []
edges = []
with driver.session() as session:
# Fetch all nodes and relationships
result = session.run("MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 100")
node_ids = set()
for record in result:
source = record["n"]
target = record["m"]
rel = record["r"]
src_id = str(source.element_id)
tgt_id = str(target.element_id)
# Extract labels for visualization
src_label = source.get("name") or source.get("id") or "Node"
tgt_label = target.get("name") or target.get("id") or "Node"
if src_id not in node_ids:
nodes.append(Node(id=src_id, label=src_label, size=20, color="#FF6F61")) # Coral Red
node_ids.add(src_id)
if tgt_id not in node_ids:
nodes.append(Node(id=tgt_id, label=tgt_label, size=20, color="#6B5B95")) # Rich Purple
node_ids.add(tgt_id)
edges.append(Edge(source=src_id, target=tgt_id, label=rel.type, color="#888888"))
driver.close()
return nodes, edges
except Exception as e:
# Fail gracefully if Neo4j is down
st.error(f"⚠️ Connection Error: {e}")
return [], []
# Render the Graph
try:
with st.spinner("Loading Neural Network..."):
nodes, edges = get_graph_data()
if not nodes:
st.info("The graph is empty. Add a memory/fact in the sidebar to start!")
else:
# Agraph Configuration
config = Config(
width="100%",
height=600,
directed=True,
nodeHighlightBehavior=True,
highlightColor="#F7A7A6",
collapsible=False
)
st.caption(f"Visualizing {len(nodes)} nodes and {len(edges)} connections.")
return_value = agraph(nodes=nodes, edges=edges, config=config)
except Exception as e:
st.error(f"Error rendering visualization: {e}")
How It Works:
1. **Mem0 Integration**: When you click "Add to Brain", Mem0 processes your text, extracts entities, and saves them to Neo4j.
2. **Neo4j Query**: The `get_graph_data()` function runs `MATCH (n)-[r]->(m)` to get every connection in the database.
3. **Visualization**: `streamlit-agraph` takes these nodes and renders them as a physics simulation. You can drag them around!
Why This Matters:
Visualizing memory unlocks **Explainability**.
If my agent recommends a specific tool to a user, I can look at the graph and see exactly *why*. I can see the path `(User) -> (Project) -> (Requires) -> (Tool)`.
It turns the "Black Box" of AI into a "Glass Box".