Graph Overlay for Agent Memory

SochDB provides a lightweight graph layer on top of its KV storage for modeling agent memory relationships. This is NOT a full graph database - it's optimized for typical agent memory patterns.

Per-SDK surface differs

The graph layer is exposed differently in each SDK:

• Rust ships a real GraphOverlay type (sochdb::graph) generic over the connection, plus a TemporalGraphOverlay (sochdb::temporal_graph).
• Node.js exposes graph methods directly on EmbeddedDatabase (re-exported as Database): addNode / addEdge / traverse.
• Go exposes graph methods on the IPC client and the gRPC client (AddNode / AddEdge / Traverse); the embedded engine is behind the sochdb_embedded build tag.
• Python exposes graph methods directly on Database (add_node / add_edge / traverse / get_neighbors / find_path) plus temporal edges (add_temporal_edge). There is no importable GraphOverlay class in the Python SDK — the standalone "graph overlay" demos live in the sochdb-python-examples repo, not in the installed package.

SDK versions: Core engine 2.0.3, Python SDK 0.5.9, Node.js SDK 0.5.3, Go SDK 0.4.5.

Overview

The Graph Overlay enables:

• Entity relationships: user to conversation to message
• Causal chains: action1 to action2 to action3
• Reference graphs: document referenced by citation referenced by quote

Storage Model

Data Type	Key Pattern	Value
Nodes	_graph/{ns}/nodes/{id}	{type, properties}
Edges	_graph/{ns}/edges/{from}/{type}/{to}	{properties}
Reverse Index	_graph/{ns}/index/{type}/{to}/{from}	from_id

This enables O(1) node/edge operations and O(degree) traversals.

Quick Start

Python

In the Python SDK (pip install sochdb, v0.5.9) the graph operations are methods on Database. The first argument is always the graph namespace, and node/edge IDs are plain strings.

from sochdb import Database

db = Database.open("./agent_memory")
ns = "agent_001"

# Create nodes: add_node(namespace, node_id, node_type, properties?)
db.add_node(ns, "user_1", "User", {"name": "Alice"})
db.add_node(ns, "conv_1", "Conversation", {"title": "Planning Session"})
db.add_node(ns, "msg_1", "Message", {"content": "Let's start planning"})

# Create edges: add_edge(namespace, from_id, edge_type, to_id, properties?)
db.add_edge(ns, "user_1", "STARTED", "conv_1")
db.add_edge(ns, "conv_1", "CONTAINS", "msg_1")
db.add_edge(ns, "user_1", "SENT", "msg_1")

# Traverse: returns (nodes, edges) as lists of dicts
nodes, edges = db.traverse(ns, "user_1", max_depth=2, order="bfs")
for node in nodes:
    print(f"Node: {node['id']} ({node['node_type']})")

# Shortest path: find_path(from, to, max_depth=10, namespace="default")
result = db.find_path("user_1", "msg_1", max_depth=10, namespace=ns)
if result:
    print(result["path"])  # e.g. ["user_1", "conv_1", "msg_1"]

note

Property values passed to add_node / add_edge are serialized to JSON; use string keys and JSON-serializable values.

Node.js / TypeScript

In the Node.js SDK (npm install @sochdb/sochdb, v0.5.3) the graph operations are methods on EmbeddedDatabase (re-exported as Database). The first argument is the namespace. EmbeddedDatabase.open() is synchronous (no await).

import { Database } from '@sochdb/sochdb';

const db = Database.open('./agent_memory'); // synchronous
const ns = 'agent_001';

// addNode(namespace, nodeId, nodeType, properties?)
await db.addNode(ns, 'user_1', 'User', { name: 'Alice' });
await db.addNode(ns, 'conv_1', 'Conversation', { title: 'Planning' });
await db.addNode(ns, 'msg_1', 'Message', { content: "Let's start" });

// addEdge(namespace, fromId, edgeType, toId, properties?)
await db.addEdge(ns, 'user_1', 'STARTED', 'conv_1');
await db.addEdge(ns, 'conv_1', 'CONTAINS', 'msg_1');
await db.addEdge(ns, 'user_1', 'SENT', 'msg_1');

// traverse(namespace, startNode, maxDepth=10, order='bfs'|'dfs') => { nodes, edges }
const { nodes, edges } = await db.traverse(ns, 'user_1', 2, 'bfs');
console.log(nodes, edges);

note

The Node SDK does not export a GraphOverlay class or an EdgeDirection enum, and there are no bfs / shortestPath helpers — use traverse and walk the returned { nodes, edges }. The same addNode / addEdge / traverse methods are also available on the gRPC SochDBClient.

Go

In the Go SDK (go get github.com/sochdb/sochdb-go, v0.4.5) graph methods live on the connection clients. The SDK is remote-first by default; the example below uses the IPC client. The first argument is the namespace.

import "github.com/sochdb/sochdb-go"

client, _ := sochdb.ConnectToDatabase("./agent_memory")
defer client.Close()
ns := "agent_001"

// AddNode(namespace, nodeID, nodeType, properties)
client.AddNode(ns, "user_1", "User", map[string]interface{}{"name": "Alice"})
client.AddNode(ns, "conv_1", "Conversation", map[string]interface{}{"title": "Planning"})
client.AddNode(ns, "msg_1", "Message", map[string]interface{}{"content": "Let's start"})

// AddEdge(namespace, fromID, edgeType, toID, properties)
client.AddEdge(ns, "user_1", "STARTED", "conv_1", nil)
client.AddEdge(ns, "conv_1", "CONTAINS", "msg_1", nil)
client.AddEdge(ns, "user_1", "SENT", "msg_1", nil)

// Traverse(namespace, startNode, maxDepth, order) => *TraverseResult{Nodes, Edges}
result, _ := client.Traverse(ns, "user_1", 2, "bfs")
fmt.Println(result.Nodes, result.Edges)

note

The Go SDK has no NewGraphOverlay constructor and no BFS / ShortestPath helpers; use Traverse and inspect the returned Nodes / Edges. The same operations are available on the gRPC client as AddGraphNode / AddGraphEdge / TraverseGraph. The embedded in-process engine requires building with -tags sochdb_embedded.

Rust

In the Rust SDK (cargo add sochdb, crate sochdb v2.0.3) the graph layer is a real GraphOverlay<C> type, generic over any connection that implements ConnectionTrait. Node IDs are RecordId values, and properties are HashMap<String, SochValue>.

use sochdb::Connection;
use sochdb::graph::GraphOverlay;
use sochdb_core::{RecordId, SochValue};
use std::collections::HashMap;

let conn = Connection::open("./agent_memory")?;
let graph = GraphOverlay::new(conn, "agent_001");

// Create a node
let mut props = HashMap::new();
props.insert("name".to_string(), SochValue::Text("Alice".to_string()));
let user = RecordId::new("user", 1);
graph.add_node(&user, "User", Some(props))?;

// Create an edge
let conv = RecordId::from_string("conv", "abc");
graph.add_edge(&user, "STARTED", &conv, None)?;

// Breadth-first traversal: bfs(start, max_depth, edge_types?, node_types?)
let reachable = graph.bfs(&user, 2, None, None)?;

// Shortest path: shortest_path(from, to, max_depth, edge_types?) -> Option<Vec<RecordId>>
let path = graph.shortest_path(&user, &conv, 10, None)?;

Node Operations

These operations are exposed on the Rust GraphOverlay. The other SDKs expose a subset (add_node / add_edge / traverse and, in Python, delete_node / get_neighbors / find_path) as methods on the connection.

Operation	Description	Complexity
add_node(id, type, props)	Create or update node	O(1)
get_node(id)	Retrieve node by ID	O(1)
update_node(id, props, type?)	Update properties or type	O(1)
delete_node(id, cascade)	Delete node (optionally with edges)	O(degree)
node_exists(id)	Check if node exists	O(1)

Edge Operations

Operation	Description	Complexity
add_edge(from, type, to, props)	Create directed edge	O(1)
get_edge(from, type, to)	Get specific edge	O(1)
get_edges(from, type?)	Get outgoing edges	O(degree)
get_incoming_edges(to, type?)	Get incoming edges	O(degree)
delete_edge(from, type, to)	Delete edge	O(1)

Traversal Operations

Where these live

bfs, dfs, shortest_path, get_neighbors, get_nodes_by_type, and get_subgraph are first-class methods on the Rust GraphOverlay. In Python, use traverse(...) (which returns (nodes, edges)) plus the get_neighbors(...) and find_path(...) helpers on Database. In Node.js and Go, only traverse / Traverse is provided — walk the returned nodes/edges yourself.

BFS / DFS (Rust)

// Find all reachable nodes within 3 hops
let nodes = graph.bfs(&user, 3, None, None)?;

// Filter by edge types
let nodes = graph.bfs(&user, 3, Some(&["SENT", "CONTAINS"]), None)?;

// Filter by node types
let nodes = graph.bfs(&user, 3, None, Some(&["Message"]))?;

// Depth-first traversal
let nodes = graph.dfs(&user, 5, None, None)?;

Traverse (Python)

# Breadth-first traversal returns (nodes, edges)
nodes, edges = db.traverse("agent_001", "user_1", max_depth=3, order="bfs")

# Depth-first traversal
nodes, edges = db.traverse("agent_001", "user_1", max_depth=5, order="dfs")

Shortest Path

Python

# find_path(from, to, max_depth=10, namespace="default")
result = db.find_path("user_1", "msg_10", max_depth=10, namespace="agent_001")
# result is {"path": [...], "edges": [...]} or None if unreachable

Rust

// shortest_path(from, to, max_depth, edge_types?) -> Option<Vec<RecordId>>
let path = graph.shortest_path(&user, &target, 10, None)?;
// Some(vec![...]) or None if unreachable

Query Operations

Get Neighbors

Python

# get_neighbors(node_id, direction="outgoing", edge_type=None, namespace="default")
out = db.get_neighbors("user_1", direction="outgoing", namespace="agent_001")
incoming = db.get_neighbors("msg_1", direction="incoming", namespace="agent_001")
both = db.get_neighbors("conv_1", direction="both", namespace="agent_001")
filtered = db.get_neighbors("user_1", edge_type="STARTED", namespace="agent_001")
# Each returns a dict like {"neighbors": [...]}

Rust

use sochdb::graph::EdgeDirection;

// get_neighbors(node, direction, edge_type?) -> Vec<Neighbor>
let neighbors = graph.get_neighbors(&user, EdgeDirection::Outgoing, None)?;

Get Nodes by Type (Rust)

// Get all User nodes (scans, use sparingly)
let users = graph.get_nodes_by_type("User", 100)?;

Get Subgraph (Rust)

// Extract a subgraph around a node
let subgraph = graph.get_subgraph(&user, 2)?;
println!("Nodes: {}", subgraph.nodes.len());
println!("Edges: {}", subgraph.edges.len());

Temporal Edges

For time-travel queries ("what did the system know at time T?"), edges can carry a validity interval.

• Rust: TemporalGraphOverlay (sochdb::temporal_graph) with add_edge_at, invalidate_edge_at, get_edges_at, get_edges_in_window, neighbors_at, and subgraph_at.
• Python: db.add_temporal_edge(...), db.query_temporal_graph(...), and db.end_temporal_edge(...) on Database.

import time

now = int(time.time() * 1000)
one_hour = 60 * 60 * 1000

# valid_from / valid_until are Unix epoch milliseconds (0 = no expiry)
db.add_temporal_edge(
    namespace="smart_home",
    from_id="door_front",
    edge_type="STATE",
    to_id="open",
    valid_from=now - one_hour,
    valid_until=now,
    properties={"sensor": "motion_1"},
)

# modes: "CURRENT" | "POINT_IN_TIME" | "RANGE"
edges = db.query_temporal_graph(
    namespace="smart_home",
    node_id="door_front",
    mode="POINT_IN_TIME",
    timestamp=now - 30 * 60 * 1000,
)

Agent Memory Patterns

The examples below use the Python Database graph API. The first argument is the namespace ("agent_001" here).

Conversation History

ns = "agent_001"
# Model a conversation thread
db.add_node(ns, "conv_1", "Conversation", {"title": "Support Chat"})
db.add_node(ns, "msg_1", "Message", {"role": "user", "content": "Help!"})
db.add_node(ns, "msg_2", "Message", {"role": "assistant", "content": "I can help"})

db.add_edge(ns, "conv_1", "CONTAINS", "msg_1")
db.add_edge(ns, "conv_1", "CONTAINS", "msg_2")
db.add_edge(ns, "msg_1", "FOLLOWED_BY", "msg_2")

# Walk the conversation
nodes, edges = db.traverse(ns, "conv_1", max_depth=2)

Tool Call Chains

ns = "agent_001"
# Model tool execution sequences
db.add_node(ns, "action_1", "ToolCall", {"tool": "search", "query": "docs"})
db.add_node(ns, "action_2", "ToolCall", {"tool": "read_file", "path": "README.md"})
db.add_node(ns, "action_3", "ToolCall", {"tool": "summarize", "input": "..."})

db.add_edge(ns, "action_1", "CAUSED", "action_2")
db.add_edge(ns, "action_2", "CAUSED", "action_3")

# Walk the causal chain
nodes, edges = db.traverse(ns, "action_1", max_depth=10, order="dfs")

Knowledge References

ns = "agent_001"
# Model document references
db.add_node(ns, "doc_1", "Document", {"title": "API Guide"})
db.add_node(ns, "chunk_1", "Chunk", {"text": "Authentication uses..."})
db.add_node(ns, "chunk_2", "Chunk", {"text": "Rate limits are..."})

db.add_edge(ns, "doc_1", "CONTAINS", "chunk_1")
db.add_edge(ns, "doc_1", "CONTAINS", "chunk_2")
db.add_edge(ns, "chunk_2", "REFERENCES", "chunk_1")

# Find chunks reachable from the document
nodes, edges = db.traverse(ns, "doc_1", max_depth=1)

Performance Characteristics

Operation	Time Complexity	Notes
Add/Get Node	O(1)	Direct KV lookup
Add/Get Edge	O(1)	Direct KV lookup
Outgoing Edges	O(degree)	Prefix scan
Incoming Edges	O(degree)	Reverse index lookup
BFS/DFS	O(V + E)	For reachable subgraph
Shortest Path	O(V + E)	BFS-based

Best Practices

1. Use meaningful edge types: SENT, CONTAINS, REFERENCES are clearer than generic RELATES_TO
2. Namespace by agent: Use separate namespaces for each agent's memory
3. Limit traversal depth: Set reasonable max_depth to avoid runaway queries
4. Use cascade delete carefully: In Rust, delete_node(id, true) removes all connected edges
5. Filter early: In Rust traversals, use edge_types and node_types to reduce work

See Also

• Context Query - Token-aware retrieval
• Python SDK - Full Python API