Bridging Dimensions: How AI Agents Remember 3D Worlds Through Text Conversations

The Challenge: When 3D Meets Text

• Information loss during encoding: Converting dense 3D data into text-storable formats often strips away nuanced spatial relationships

• Session isolation: Each new conversation context window starts fresh, losing previous environmental awareness

• Cross-modal translation gaps: The way an agent "sees" space doesn't naturally align with how it "talks" about space

Understanding Cross-Modal Memory Architecture

1. Spatial Encoding Layer

import torch
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass
class SpatialNode:
    """Represents a spatial entity in the scene graph"""
    object_id: str
    object_type: str
    position: np.ndarray  # 3D coordinates
    dimensions: np.ndarray  # bounding box
    semantic_features: torch.Tensor  # embedding vector
    
@dataclass
class SpatialEdge:
    """Represents spatial relationships between objects"""
    source_id: str
    target_id: str
    relationship_type: str  # "on_top", "adjacent", "inside", etc.
    distance: float
    direction: np.ndarray  # unit vector

class SemanticSceneGraph:
    def __init__(self):
        self.nodes: Dict[str, SpatialNode] = {}
        self.edges: List[SpatialEdge] = []
        
    def add_object(self, node: SpatialNode):
        self.nodes[node.object_id] = node
        
    def add_relationship(self, edge: SpatialEdge):
        self.edges.append(edge)
        
    def to_textual_embedding(self) -> str:
        """Convert scene graph to text-compatible representation"""
        descriptions = []
        for node in self.nodes.values():
            desc = f"{node.object_type} at position ({node.position[0]:.2f}, {node.position[1]:.2f}, {node.position[2]:.2f})"
            descriptions.append(desc)
        return " | ".join(descriptions)
    
    def get_spatial_context(self, object_id: str, radius: float = 2.0) -> List[SpatialNode]:
        """Retrieve objects within a radius of target object"""
        target = self.nodes.get(object_id)
        if not target:
            return []
        
        nearby = []
        for node in self.nodes.values():
            if node.object_id != object_id:
                dist = np.linalg.norm(node.position - target.position)
                if dist <= radius:
                    nearby.append(node)
        return nearby

2. Persistent Memory Store

from datetime import datetime
import json
import sqlite3
from typing import Any

class PersistentMemoryStore:
    def __init__(self, db_path: str = "spatial_memory.db"):
        self.db_path = db_path
        self._init_database()
        
    def _init_database(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS spatial_memories (
                memory_id TEXT PRIMARY KEY,
                session_id TEXT,
                timestamp DATETIME,
                scene_graph_json TEXT,
                text_description TEXT,
                embedding_blob BLOB,
                environment_id TEXT
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS spatial_queries (
                query_id TEXT PRIMARY KEY,
                memory_id TEXT,
                query_text TEXT,
                response_text TEXT,
                timestamp DATETIME,
                FOREIGN KEY (memory_id) REFERENCES spatial_memories(memory_id)
            )
        ''')
        
        conn.commit()
        conn.close()
        
    def store_memory(self, memory_id: str, session_id: str, 
                     scene_graph: SemanticSceneGraph, 
                     embedding: np.ndarray,
                     environment_id: str):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO spatial_memories 
            (memory_id, session_id, timestamp, scene_graph_json, 
             text_description, embedding_blob, environment_id)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            memory_id,
            session_id,
            datetime.now().isoformat(),
            json.dumps(self._serialize_scene_graph(scene_graph)),
            scene_graph.to_textual_embedding(),
            embedding.tobytes(),
            environment_id
        ))
        
        conn.commit()
        conn.close()
        
    def _serialize_scene_graph(self, graph: SemanticSceneGraph) -> dict:
        return {
            "nodes": [
                {
                    "id": n.object_id,
                    "type": n.object_type,
                    "position": n.position.tolist(),
                    "dimensions": n.dimensions.tolist()
                }
                for n in graph.nodes.values()
            ],
            "edges": [
                {
                    "source": e.source_id,
                    "target": e.target_id,
                    "relation": e.relationship_type,
                    "distance": e.distance
                }
                for e in graph.edges
            ]
        }
    
    def retrieve_relevant_memory(self, query_embedding: np.ndarray, 
                                  environment_id: str,
                                  threshold: float = 0.7) -> Optional[dict]:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT memory_id, scene_graph_json, text_description, embedding_blob
            FROM spatial_memories
            WHERE environment_id = ?
            ORDER BY timestamp DESC
        ''', (environment_id,))
        
        best_match = None
        best_score = threshold
        
        for row in cursor.fetchall():
            stored_embedding = np.frombuffer(row[3], dtype=np.float32)
            score = self._cosine_similarity(query_embedding, stored_embedding)
            
            if score > best_score:
                best_score = score
                best_match = {
                    "memory_id": row[0],
                    "scene_graph": json.loads(row[1]),
                    "description": row[2],
                    "similarity": score
                }
        
        conn.close()
        return best_match
    
    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

3. Cross-Modal Retrieval System

from openai import OpenAI
import hashlib

class CrossModalRetrieval:
    def __init__(self, memory_store: PersistentMemoryStore, 
                 embedding_model: str = "text-embedding-3-small"):
        self.memory_store = memory_store
        self.client = OpenAI()
        self.embedding_model = embedding_model
        
    def process_query(self, user_query: str, environment_id: str) -> str:
        # Generate embedding for the query
        query_embedding = self._get_embedding(user_query)
        
        # Retrieve relevant spatial memory
        memory = self.memory_store.retrieve_relevant_memory(
            query_embedding, environment_id
        )
        
        if not memory:
            return "I don't have any spatial memory of that environment. Please provide a scan first."
        
        # Reconstruct spatial context
        scene_graph = self._reconstruct_scene_graph(memory["scene_graph"])
        
        # Generate response using spatial context
        response = self._generate_spatial_response(
            user_query, scene_graph, memory["description"]
        )
        
        return response
    
    def _get_embedding(self, text: str) -> np.ndarray:
        response = self.client.embeddings.create(
            model=self.embedding_model,
            input=text
        )
        return np.array(response.data[0].embedding, dtype=np.float32)
    
    def _reconstruct_scene_graph(self, graph_dict: dict) -> SemanticSceneGraph:
        graph = SemanticSceneGraph()
        
        for node_data in graph_dict["nodes"]:
            node = SpatialNode(
                object_id=node_data["id"],
                object_type=node_data["type"],
                position=np.array(node_data["position"]),
                dimensions=np.array(node_data["dimensions"]),
                semantic_features=torch.zeros(768)  # placeholder
            )
            graph.add_object(node)
            
        for edge_data in graph_dict["edges"]:
            edge = SpatialEdge(
                source_id=edge_data["source"],
                target_id=edge_data["target"],
                relationship_type=edge_data["relation"],
                distance=edge_data["distance"],
                direction=np.zeros(3)
            )
            graph.edges.append(edge)
            
        return graph
    
    def _generate_spatial_response(self, query: str, 
                                    scene_graph: SemanticSceneGraph,
                                    context: str) -> str:
        # Build spatial prompt
        system_prompt = """You are a spatial reasoning AI with persistent memory of 3D environments.
Answer questions about the environment using the provided spatial context.
Be precise about locations, distances, and spatial relationships."""
        
        spatial_context = f"""
Current spatial memory:
{context}

Available objects: {list(scene_graph.nodes.keys())}
Spatial relationships: {len(scene_graph.edges)} recorded
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": spatial_context},
                {"role": "user", "content": query}
            ]
        )
        
        return response.choices[0].message.content

Practical Implementation Strategies

Strategy 1: Hierarchical Spatial Abstraction

• Level 1: Room/zone boundaries

• Level 2: Major objects and furniture

• Level 3: Fine details and small items

Strategy 2: Incremental Memory Updates

class IncrementalMemoryUpdater:
    def __init__(self, memory_store: PersistentMemoryStore):
        self.memory_store = memory_store
        self.change_threshold = 0.15  # 15% change triggers new memory
        
    def update_environment(self, environment_id: str, 
                          new_scan: SemanticSceneGraph,
                          session_id: str):
        # Compare with existing memory
        existing = self._get_latest_memory(environment_id)
        
        if not existing:
            self._create_new_memory(environment_id, new_scan, session_id)
            return "new_environment"
        
        change_score = self._calculate_change_score(
            existing, new_scan
        )
        
        if change_score > self.change_threshold:
            self._create_revised_memory(
                environment_id, new_scan, session_id, existing
            )
            return "significant_change"
        else:
            self._merge_minor_changes(existing, new_scan)
            return "minor_update"
            
    def _calculate_change_score(self, existing: dict, 
                                 new_scan: SemanticSceneGraph) -> float:
        existing_objects = set(existing["scene_graph"]["nodes"].keys())
        new_objects = set(new_scan.nodes.keys())
        
        added = new_objects - existing_objects
        removed = existing_objects - new_objects
        
        total_objects = len(existing_objects.union(new_objects))
        if total_objects == 0:
            return 0.0
            
        return (len(added) + len(removed)) / total_objects

Strategy 3: Contextual Compression

def compress_spatial_memory(scene_graph: SemanticSceneGraph, 
                           max_tokens: int = 500) -> str:
    """
    Compress scene graph into token-efficient description
    while preserving spatial reasoning capability
    """
    # Group objects by type and proximity
    clusters = _cluster_nearby_objects(scene_graph, radius=1.5)
    
    compressed = []
    for cluster in clusters:
        objects = cluster["objects"]
        if len(objects) == 1:
            obj = objects[0]
            compressed.append(
                f"{obj.object_type}[{obj.object_id}]@{obj.position.round(1).tolist()}"
            )
        else:
            types = [o.object_type for o in objects]
            center = np.mean([o.position for o in objects], axis=0)
            compressed.append(
                f"{len(objects)}x{set(types)}@cluster:{center.round(1).tolist()}"
            )
    
    return "; ".join(compressed)

def _cluster_nearby_objects(scene_graph: SemanticSceneGraph, 
                           radius: float) -> List[dict]:
    from sklearn.cluster import DBSCAN
    
    if not scene_graph.nodes:
        return []
        
    positions = np.array([n.position for n in scene_graph.nodes.values()])
    ids = list(scene_graph.nodes.keys())
    
    clustering = DBSCAN(eps=radius, min_samples=1).fit(positions)
    
    clusters = {}
    for idx, label in enumerate(clustering.labels_):
        if label not in clusters:
            clusters[label] = []
        clusters[label].append(scene_graph.nodes[ids[idx]])
    
    return [{"objects": objs} for objs in clusters.values()]

Best Practices for Production Systems

1. Implement Memory Versioning: Track changes to spatial memories over time. This enables rollback and temporal queries ("What was in the corner last month?").

1. Use Confidence Scoring: Not all spatial observations are equally reliable. Store confidence scores with each memory and use them during retrieval.

1. Handle Ambiguity Gracefully: Spatial queries can be ambiguous. When a user asks "near the table," your agent should either ask for clarification or provide the most likely interpretation with a confidence estimate.

1. Optimize for Query Patterns: Analyze how users query spatial information and optimize your indexing strategy accordingly. If users frequently ask about object relationships, pre-compute and index those relationships.

1. Implement Memory Decay: Not all memories should persist indefinitely. Implement a decay function based on access frequency and environmental change rates.

Real-World Applications

• Warehouse Robotics: Robots that remember inventory layouts across shifts and can answer questions about stock locations via chat interfaces

• Smart Home Assistants: AI that maintains a mental model of your home's layout and can guide visitors or service technicians

• AR/VR Experiences: Virtual agents that remember spatial details from previous sessions, creating continuity in immersive experiences

• Search and Rescue: Drones that scan disaster areas and provide human operators with persistent spatial memory for planning

Conclusion