Skip to content

Memory & Session Reference

This section documents conversation session tracking, message representations, in-memory dialogue storage, and persistent SQLite storage.

ChatMessage & ConversationSession

edgevoiceai.memory.message

Data models for Conversation Messages and User Sessions.

ChatMessage dataclass

Represents a single conversational turn dialogue message.

Source code in edgevoiceai/memory/message.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@dataclass(slots=True)
class ChatMessage:
    """Represents a single conversational turn dialogue message."""
    role: str  # "user" | "assistant" | "system"
    content: str
    timestamp: float = field(default_factory=time.time)
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "role": self.role,
            "content": self.content,
            "timestamp": self.timestamp,
            "metadata": self.metadata,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ChatMessage:
        return cls(
            role=str(data.get("role", "user")),
            content=str(data.get("content", "")),
            timestamp=float(data.get("timestamp", time.time())),
            metadata=dict(data.get("metadata", {})),
        )

ConversationSession dataclass

Represents an ongoing user voice dialogue session with history and profile memory.

Source code in edgevoiceai/memory/message.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass(slots=True)
class ConversationSession:
    """Represents an ongoing user voice dialogue session with history and profile memory."""
    session_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    messages: list[ChatMessage] = field(default_factory=list)
    profile: dict[str, Any] = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

    def add_message(self, role: str, content: str, metadata: dict[str, Any] | None = None) -> ChatMessage:
        msg = ChatMessage(role=role, content=content, metadata=metadata or {})
        self.messages.append(msg)
        self.updated_at = time.time()
        return msg

    def update_profile(self, new_data: dict[str, Any]) -> None:
        """Merges non-null attributes into the user's profile."""
        for k, v in new_data.items():
            if v is not None and v != "":
                self.profile[k] = v
        self.updated_at = time.time()

update_profile 

update_profile(new_data)

Merges non-null attributes into the user's profile.

Source code in edgevoiceai/memory/message.py 
54
55
56
57
58
59
def update_profile(self, new_data: dict[str, Any]) -> None:
    """Merges non-null attributes into the user's profile."""
    for k, v in new_data.items():
        if v is not None and v != "":
            self.profile[k] = v
    self.updated_at = time.time()

Memory Stores

BaseMemoryStore

edgevoiceai.memory.stores.base

Base Interface for Memory and Dialogue Stores.

BaseMemoryStore

Bases: ABC

Abstract Base Class for managing dialogue history and persistent user profiles.

Source code in edgevoiceai/memory/stores/base.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class BaseMemoryStore(ABC):
    """
    Abstract Base Class for managing dialogue history and persistent user profiles.
    """

    @abstractmethod
    async def get_or_create_session(self, session_id: str) -> ConversationSession:
        """Retrieves an existing session or initializes a new one."""
        pass

    @abstractmethod
    async def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        metadata: dict[str, Any] | None = None,
    ) -> ChatMessage:
        """Appends a new turn message to the session dialogue history."""
        pass

    @abstractmethod
    async def get_messages(
        self,
        session_id: str,
        limit: int | None = None,
    ) -> list[ChatMessage]:
        """Returns ordered messages for the given session."""
        pass

    @abstractmethod
    async def update_profile(
        self,
        session_id: str,
        profile_data: dict[str, Any],
    ) -> dict[str, Any]:
        """Merges new profile details into the user's saved state."""
        pass

    @abstractmethod
    async def get_profile(self, session_id: str) -> dict[str, Any]:
        """Returns the current known user profile details."""
        pass

    @abstractmethod
    async def clear_session(self, session_id: str) -> None:
        """Deletes or resets a conversation session."""
        pass

    async def build_context_prompt(
        self,
        session_id: str,
        header: str = "[KNOWN USER CONTEXT]",
    ) -> str:
        """
        Constructs a clean, human-readable context block from known profile details
        for dynamic injection into voice system prompts.
        """
        profile = await self.get_profile(session_id)
        if not profile:
            return ""

        lines = [header]
        for key, val in profile.items():
            if val is not None and val != "":
                # Format key from snake_case to Title Case
                formatted_key = key.replace("_", " ").title()
                if isinstance(val, list):
                    val_str = ", ".join(str(item) for item in val)
                else:
                    val_str = str(val)
                lines.append(f"• {formatted_key}: {val_str}")

        if len(lines) == 1:
            return ""

        lines.append("(Use these known details naturally; do not repeatedly re-ask the user for them)")
        return "\n".join(lines)

add_message abstractmethod async 

add_message(session_id, role, content, metadata=None)

Appends a new turn message to the session dialogue history.

Source code in edgevoiceai/memory/stores/base.py 
22
23
24
25
26
27
28
29
30
31
@abstractmethod
async def add_message(
    self,
    session_id: str,
    role: str,
    content: str,
    metadata: dict[str, Any] | None = None,
) -> ChatMessage:
    """Appends a new turn message to the session dialogue history."""
    pass

build_context_prompt async 

build_context_prompt(session_id, header='[KNOWN USER CONTEXT]')

Constructs a clean, human-readable context block from known profile details for dynamic injection into voice system prompts.

Source code in edgevoiceai/memory/stores/base.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def build_context_prompt(
    self,
    session_id: str,
    header: str = "[KNOWN USER CONTEXT]",
) -> str:
    """
    Constructs a clean, human-readable context block from known profile details
    for dynamic injection into voice system prompts.
    """
    profile = await self.get_profile(session_id)
    if not profile:
        return ""

    lines = [header]
    for key, val in profile.items():
        if val is not None and val != "":
            # Format key from snake_case to Title Case
            formatted_key = key.replace("_", " ").title()
            if isinstance(val, list):
                val_str = ", ".join(str(item) for item in val)
            else:
                val_str = str(val)
            lines.append(f"• {formatted_key}: {val_str}")

    if len(lines) == 1:
        return ""

    lines.append("(Use these known details naturally; do not repeatedly re-ask the user for them)")
    return "\n".join(lines)

clear_session abstractmethod async 

clear_session(session_id)

Deletes or resets a conversation session.

Source code in edgevoiceai/memory/stores/base.py 
56
57
58
59
@abstractmethod
async def clear_session(self, session_id: str) -> None:
    """Deletes or resets a conversation session."""
    pass

get_messages abstractmethod async 

get_messages(session_id, limit=None)

Returns ordered messages for the given session.

Source code in edgevoiceai/memory/stores/base.py 
33
34
35
36
37
38
39
40
@abstractmethod
async def get_messages(
    self,
    session_id: str,
    limit: int | None = None,
) -> list[ChatMessage]:
    """Returns ordered messages for the given session."""
    pass

get_or_create_session abstractmethod async 

get_or_create_session(session_id)

Retrieves an existing session or initializes a new one.

Source code in edgevoiceai/memory/stores/base.py 
17
18
19
20
@abstractmethod
async def get_or_create_session(self, session_id: str) -> ConversationSession:
    """Retrieves an existing session or initializes a new one."""
    pass

get_profile abstractmethod async 

get_profile(session_id)

Returns the current known user profile details.

Source code in edgevoiceai/memory/stores/base.py 
51
52
53
54
@abstractmethod
async def get_profile(self, session_id: str) -> dict[str, Any]:
    """Returns the current known user profile details."""
    pass

update_profile abstractmethod async 

update_profile(session_id, profile_data)

Merges new profile details into the user's saved state.

Source code in edgevoiceai/memory/stores/base.py 
42
43
44
45
46
47
48
49
@abstractmethod
async def update_profile(
    self,
    session_id: str,
    profile_data: dict[str, Any],
) -> dict[str, Any]:
    """Merges new profile details into the user's saved state."""
    pass

InMemoryMemoryStore

edgevoiceai.memory.stores.memory

In-Memory Dialogue and Profile Store. Zero-dependency in-RAM storage for ephemeral voice sessions and fast testing.

InMemoryMemoryStore

Bases: BaseMemoryStore

Thread-safe in-memory store for sessions, messages, and profile attributes.

Source code in edgevoiceai/memory/stores/memory.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class InMemoryMemoryStore(BaseMemoryStore):
    """
    Thread-safe in-memory store for sessions, messages, and profile attributes.
    """

    def __init__(self) -> None:
        self._sessions: dict[str, ConversationSession] = {}
        self._lock = asyncio.Lock()

    async def get_or_create_session(self, session_id: str) -> ConversationSession:
        async with self._lock:
            if session_id not in self._sessions:
                self._sessions[session_id] = ConversationSession(session_id=session_id)
            return self._sessions[session_id]

    async def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        metadata: dict[str, Any] | None = None,
    ) -> ChatMessage:
        session = await self.get_or_create_session(session_id)
        async with self._lock:
            return session.add_message(role=role, content=content, metadata=metadata)

    async def get_messages(
        self,
        session_id: str,
        limit: int | None = None,
    ) -> list[ChatMessage]:
        session = await self.get_or_create_session(session_id)
        async with self._lock:
            if limit is not None and limit > 0:
                return list(session.messages[-limit:])
            return list(session.messages)

    async def update_profile(
        self,
        session_id: str,
        profile_data: dict[str, Any],
    ) -> dict[str, Any]:
        session = await self.get_or_create_session(session_id)
        async with self._lock:
            session.update_profile(profile_data)
            return dict(session.profile)

    async def get_profile(self, session_id: str) -> dict[str, Any]:
        session = await self.get_or_create_session(session_id)
        async with self._lock:
            return dict(session.profile)

    async def clear_session(self, session_id: str) -> None:
        async with self._lock:
            self._sessions.pop(session_id, None)

SQLiteMemoryStore

edgevoiceai.memory.stores.sqlite

SQLite Persistent Dialogue and Profile Store. Lightweight, thread-safe, file-based persistence for edge devices and servers. Uses Python standard library sqlite3 via async thread offloading.

SQLiteMemoryStore

Bases: BaseMemoryStore

Asynchronous SQLite memory store for session history and user profiles.

Source code in edgevoiceai/memory/stores/sqlite.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class SQLiteMemoryStore(BaseMemoryStore):
    """
    Asynchronous SQLite memory store for session history and user profiles.
    """

    def __init__(self, db_path: str = "agent_memory.db") -> None:
        self.db_path = db_path
        self._init_db_sync()

    def _init_db_sync(self) -> None:
        """Initializes database tables if they do not exist."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            # 1. Sessions table
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS voice_sessions (
                    session_id TEXT PRIMARY KEY,
                    created_at REAL,
                    updated_at REAL
                )
            """)
            # 2. Messages table
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS voice_messages (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    session_id TEXT,
                    role TEXT,
                    content TEXT,
                    timestamp REAL,
                    metadata_json TEXT,
                    FOREIGN KEY (session_id) REFERENCES voice_sessions (session_id)
                )
            """)
            # 3. Profiles table
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS voice_profiles (
                    session_id TEXT PRIMARY KEY,
                    profile_json TEXT,
                    updated_at REAL,
                    FOREIGN KEY (session_id) REFERENCES voice_sessions (session_id)
                )
            """)
            conn.commit()

    async def get_or_create_session(self, session_id: str) -> ConversationSession:
        def _get() -> ConversationSession:
            now = time.time()
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT created_at, updated_at FROM voice_sessions WHERE session_id = ?", (session_id,))
                row = cursor.fetchone()
                if row:
                    created_at, updated_at = row[0], row[1]
                else:
                    cursor.execute(
                        "INSERT INTO voice_sessions (session_id, created_at, updated_at) VALUES (?, ?, ?)",
                        (session_id, now, now)
                    )
                    conn.commit()
                    created_at, updated_at = now, now

                # Fetch messages
                cursor.execute(
                    "SELECT role, content, timestamp, metadata_json FROM voice_messages WHERE session_id = ? ORDER BY id ASC",
                    (session_id,)
                )
                msgs = [
                    ChatMessage(
                        role=r[0],
                        content=r[1],
                        timestamp=r[2],
                        metadata=json.loads(r[3]) if r[3] else {}
                    )
                    for r in cursor.fetchall()
                ]

                # Fetch profile
                cursor.execute("SELECT profile_json FROM voice_profiles WHERE session_id = ?", (session_id,))
                p_row = cursor.fetchone()
                profile = json.loads(p_row[0]) if p_row and p_row[0] else {}

                return ConversationSession(
                    session_id=session_id,
                    messages=msgs,
                    profile=profile,
                    created_at=created_at,
                    updated_at=updated_at,
                )

        return await asyncio.to_thread(_get)

    async def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        metadata: dict[str, Any] | None = None,
    ) -> ChatMessage:
        def _add() -> ChatMessage:
            now = time.time()
            meta_json = json.dumps(metadata or {})
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                # Ensure session exists
                cursor.execute(
                    "INSERT OR IGNORE INTO voice_sessions (session_id, created_at, updated_at) VALUES (?, ?, ?)",
                    (session_id, now, now)
                )
                cursor.execute(
                    "INSERT INTO voice_messages (session_id, role, content, timestamp, metadata_json) VALUES (?, ?, ?, ?, ?)",
                    (session_id, role, content, now, meta_json)
                )
                cursor.execute("UPDATE voice_sessions SET updated_at = ? WHERE session_id = ?", (now, session_id))
                conn.commit()
            return ChatMessage(role=role, content=content, timestamp=now, metadata=metadata or {})

        return await asyncio.to_thread(_add)

    async def get_messages(
        self,
        session_id: str,
        limit: int | None = None,
    ) -> list[ChatMessage]:
        def _get() -> list[ChatMessage]:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                query = "SELECT role, content, timestamp, metadata_json FROM voice_messages WHERE session_id = ? ORDER BY id ASC"
                cursor.execute(query, (session_id,))
                rows = cursor.fetchall()
                if limit is not None and limit > 0:
                    rows = rows[-limit:]
                return [
                    ChatMessage(
                        role=r[0],
                        content=r[1],
                        timestamp=r[2],
                        metadata=json.loads(r[3]) if r[3] else {}
                    )
                    for r in rows
                ]

        return await asyncio.to_thread(_get)

    async def update_profile(
        self,
        session_id: str,
        profile_data: dict[str, Any],
    ) -> dict[str, Any]:
        def _update() -> dict[str, Any]:
            now = time.time()
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                # Ensure session exists
                cursor.execute(
                    "INSERT OR IGNORE INTO voice_sessions (session_id, created_at, updated_at) VALUES (?, ?, ?)",
                    (session_id, now, now)
                )
                cursor.execute("SELECT profile_json FROM voice_profiles WHERE session_id = ?", (session_id,))
                row = cursor.fetchone()
                existing = json.loads(row[0]) if row and row[0] else {}

                # Merge non-null data
                for k, v in profile_data.items():
                    if v is not None and v != "":
                        existing[k] = v

                cursor.execute(
                    "INSERT OR REPLACE INTO voice_profiles (session_id, profile_json, updated_at) VALUES (?, ?, ?)",
                    (session_id, json.dumps(existing), now)
                )
                cursor.execute("UPDATE voice_sessions SET updated_at = ? WHERE session_id = ?", (now, session_id))
                conn.commit()
                return existing

        return await asyncio.to_thread(_update)

    async def get_profile(self, session_id: str) -> dict[str, Any]:
        def _get() -> dict[str, Any]:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT profile_json FROM voice_profiles WHERE session_id = ?", (session_id,))
                row = cursor.fetchone()
                return json.loads(row[0]) if row and row[0] else {}

        return await asyncio.to_thread(_get)

    async def clear_session(self, session_id: str) -> None:
        def _clear() -> None:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("DELETE FROM voice_messages WHERE session_id = ?", (session_id,))
                cursor.execute("DELETE FROM voice_profiles WHERE session_id = ?", (session_id,))
                cursor.execute("DELETE FROM voice_sessions WHERE session_id = ?", (session_id,))
                conn.commit()

        await asyncio.to_thread(_clear)