Related Series
- Building an AI VTuber Agent, Part 1: Overview
- Building an AI VTuber Agent, Part 2: LLM and Persona
- Building an AI VTuber Agent, Part 3: Memory (this post)
- Building an AI VTuber Agent, Part 4: STT
- Building an AI VTuber Agent, Part 5: TTS
- Building an AI VTuber Agent, Part 6: Live2D
- Building an AI VTuber Agent, Part 7: Stream Integration
- Building an AI VTuber Agent, Part 8: Production
Introduction
In Part 2, we gave Haru a real personality and emotions — Jinja2 dynamic prompts, an emotion transition engine, a character state machine. Haru now shifts emotions based on conversational context, builds rapport over time, and tracks topics.
But there's one gaping hole left. At the end of Part 2, we had this note:
messages = [system, *history[-20:], user_msg]
Only the last 20 messages make it into the context. Anything said before that 21st message simply vanishes from Haru's world. A heartfelt conversation today becomes a story Haru has never heard when tomorrow rolls around.
"Haru, do you remember that game I said I liked last time?"
For Haru to answer that question, something far beyond history[-20:] is needed. Part 3 is about building exactly that — a way for an AI to genuinely remember.
Mapping Human Memory Models to AI Memory
Psychology and neuroscience classify human memory in several ways. Three subtypes of long-term memory map directly onto AI memory system design.
Episodic Memory
Autobiographical memory of when, where, and what happened.
"Three weeks ago on Tuesday, Minjun said he was exhausted from working late"
"Last week, Minjun got really excited talking about games"
"When he connected late last night, he seemed tired"
In AI systems, episodic memory is implemented by storing individual conversation turns as vectors. When something similar comes up later — like "that time we talked about games" — relevant episodes are retrieved via vector similarity search.
Semantic Memory
General facts and knowledge independent of any specific episode — essentially a distillation of many episodes into a summary.
"Minjun likes FPS games"
"Minjun is a backend developer"
"Minjun doesn't drink coffee"
"Minjun usually starts conversations after 10 PM"
In AI systems, semantic memory is implemented as a user profile + important facts store. An LLM generates these by summarizing multiple episodes, or they are extracted automatically when specific keywords are detected.
Procedural Memory
Memory of how to respond — patterns of behavior.
"When Minjun asks a technical question, he prefers explanations paired with example code"
"When Minjun says he's tired, he prefers quiet companionship over active consolation"
This is the hardest type to implement. The plan is to build it incrementally using reinforcement learning after Part 7 (user interaction). This part focuses on episodic and semantic memory.
Memory Architecture Diagram
┌─────────────────────────────────────────────────────────────────┐
│ Memory System │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Short-term Memory (Working Memory) │ │
│ │ Current session conversation history (last 10–20 turns)│ │
│ │ CharacterState (emotion, rapport, recent topics) │ │
│ └────────────────────────┬─────────────────────────────────┘ │
│ │ Session end / important event │
│ ┌────────────────────────▼─────────────────────────────────┐ │
│ │ Long-term Memory │ │
│ │ │ │
│ │ ┌──────────────────┐ ┌────────────────────────────┐ │ │
│ │ │ Episodic Memory │ │ Semantic Memory │ │ │
│ │ │ Qdrant Vector │ │ SQLite Facts DB │ │ │
│ │ │ │ │ │ │ │
│ │ │ - Conversation │ │ - User facts │ │ │
│ │ │ text │ │ - Preferences / traits │ │ │
│ │ │ - Emotion │ │ - Emotional history patterns │ │ │
│ │ │ metadata │ │ - Confidence scores │ │ │
│ │ │ - Importance │ │ │ │ │
│ │ │ score │ │ │ │ │
│ │ │ - Timestamp │ │ │ │ │
│ │ └──────────────────┘ └────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ RAG Pipeline (Retrieval-Augmented Generation) │ │
│ │ New user message → Embedding → Qdrant search │ │
│ │ → Relevant episodes + semantic facts → Inject into │ │
│ │ system prompt │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
How Vector Search Works
Vector similarity search is the core technology behind the memory system. Without understanding how it works, it's hard to see why we use Qdrant or why embeddings are necessary.
What Embeddings Are
An embedding converts text into a high-dimensional numeric vector. Text with similar meaning ends up close together in the vector space.
"I love FPS games so much" → [0.12, -0.34, 0.89, ..., 0.23] (768-dim)
"Battle royale games are so fun" → [0.15, -0.31, 0.85, ..., 0.19] (768-dim)
"The weather is really nice today" → [-0.45, 0.72, -0.12, ..., 0.67] (768-dim)
The vectors for the first two sentences are close to each other (cosine similarity ≈ 0.94), while the third is far from both (≈ 0.11). When you encode the query "search for something about liking games," the FPS sentence ranks at the top.
Cosine Similarity
cosine_similarity(A, B) = (A · B) / (|A| × |B|)
Range: -1 (completely opposite) ~ 0 (unrelated) ~ 1 (identical)
Vector databases perform this computation across hundreds of millions of vectors in milliseconds using ANN (Approximate Nearest Neighbor) algorithms. Qdrant uses the HNSW (Hierarchical Navigable Small World) graph.
Why Not Keyword Search
Keyword search: "FPS game" → finds only sentences containing the phrase "FPS game"
Vector search: "FPS game" → also finds "battle royale", "shooter", "first-person", "Overwatch"
A query like "Haru, do you remember when we talked about games?" will surface "that conversation where Minjun got really excited" even if the stored text never uses the word "game." Keyword matching can never make that kind of connection.
Installing and Configuring Qdrant
Local Mode (File-based, No Server Required)
The Qdrant Python client can operate purely from local files without a running server — the simplest way to get started in a development environment.
uv add qdrant-client sentence-transformers
from qdrant_client import QdrantClient
# Local file-based (no server needed)
client = QdrantClient(path="./data/qdrant")
# Or in-memory (for testing)
client = QdrantClient(":memory:")
Running Qdrant as a Docker Service (Production)
# Add to docker-compose.yml
services:
qdrant:
image: qdrant/qdrant:latest
ports:
- "6333:6333"
- "6334:6334" # gRPC
volumes:
- ./data/qdrant:/qdrant/storage
environment:
QDRANT__SERVICE__GRPC_PORT: "6334"
# Connect in server mode
client = QdrantClient(host="localhost", port=6333)
Collection Schema Design
Qdrant stores vectors and metadata (payload) together. We create two collections — one for episodic memory and one for semantic memory.
# src/vtuber/memory/store.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PayloadSchemaType,
)
EMBEDDING_DIM = 768 # based on jhgan/ko-sroberta-multitask
EPISODE_COLLECTION = "episodes"
SEMANTIC_COLLECTION = "semantic_facts"
def init_collections(client: QdrantClient) -> None:
"""Create required collections if they don't already exist."""
existing = {c.name for c in client.get_collections().collections}
if EPISODE_COLLECTION not in existing:
client.create_collection(
collection_name=EPISODE_COLLECTION,
vectors_config=VectorParams(
size=EMBEDDING_DIM,
distance=Distance.COSINE,
),
)
# Payload indexes for faster filtering
client.create_payload_index(
collection_name=EPISODE_COLLECTION,
field_name="user_id",
field_schema=PayloadSchemaType.KEYWORD,
)
client.create_payload_index(
collection_name=EPISODE_COLLECTION,
field_name="importance",
field_schema=PayloadSchemaType.FLOAT,
)
if SEMANTIC_COLLECTION not in existing:
client.create_collection(
collection_name=SEMANTIC_COLLECTION,
vectors_config=VectorParams(
size=EMBEDDING_DIM,
distance=Distance.COSINE,
),
)
client.create_payload_index(
collection_name=SEMANTIC_COLLECTION,
field_name="user_id",
field_schema=PayloadSchemaType.KEYWORD,
)
Embedding Model Selection
Choosing an embedding model that handles Korean well is critical. Here's a comparison of the options.
| Model | Dimensions | Korean | Speed | Cost | Notes |
|---|---|---|---|---|---|
all-MiniLM-L6-v2 | 384 | △ Fair | ⚡⚡⚡ Fast | Free local | Optimized for English |
paraphrase-multilingual-mpnet-base-v2 | 768 | ✅ Good | ⚡⚡ Moderate | Free local | Multilingual support |
jhgan/ko-sroberta-multitask | 768 | ✅✅ Excellent | ⚡⚡ Moderate | Free local | Korean-specialized |
nomic-embed-text (Ollama) | 768 | ✅ Good | ⚡⚡ Moderate | Free local | Easy Ollama integration |
text-embedding-3-small (OpenAI) | 1536 | ✅✅ Excellent | ⚡ Network | Paid | Highest quality |
This series uses: jhgan/ko-sroberta-multitask (default) + Ollama nomic-embed-text (for integration convenience)
We build an abstraction that supports both models so they can be swapped based on the environment.
# src/vtuber/memory/embedder.py
from abc import ABC, abstractmethod
class EmbeddingBackend(ABC):
@abstractmethod
async def embed(self, text: str) -> list[float]: ...
@abstractmethod
async def embed_batch(self, texts: list[str]) -> list[list[float]]: ...
class SentenceTransformerEmbedder(EmbeddingBackend):
"""Local sentence-transformers model-based embedding"""
def __init__(self, model_name: str = "jhgan/ko-sroberta-multitask"):
from sentence_transformers import SentenceTransformer
self._model = SentenceTransformer(model_name)
async def embed(self, text: str) -> list[float]:
return self._model.encode(text, normalize_embeddings=True).tolist()
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
return self._model.encode(texts, normalize_embeddings=True).tolist()
class OllamaEmbedder(EmbeddingBackend):
"""Ollama server-based embedding (nomic-embed-text, etc.)"""
def __init__(
self,
base_url: str = "http://localhost:11434",
model: str = "nomic-embed-text",
):
import httpx
self._client = httpx.AsyncClient(timeout=30.0)
self._base_url = base_url
self._model = model
async def embed(self, text: str) -> list[float]:
resp = await self._client.post(
f"{self._base_url}/api/embeddings",
json={"model": self._model, "prompt": text},
)
return resp.json()["embedding"]
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
import asyncio
return await asyncio.gather(*[self.embed(t) for t in texts])
What to Remember — Importance Scoring
What happens if you store every conversation? Search results fill up with noise. Lightweight exchanges like "lol", "yeah", "got it" bury the memories that actually matter.
That's where an Importance Score comes in — a 0.0–1.0 value used to store only high-scoring memories or to forget low-scoring ones first.
Rule-Based Fast Filter (First Pass)
# src/vtuber/memory/importance.py
import re
from dataclasses import dataclass
# Signals that raise importance
HIGH_IMPORTANCE_PATTERNS = [
r"좋아(해|요|하다)", # preferences
r"싫어(해|요|하다)", # dislikes
r"(생일|기념일|결혼|졸업)", # important dates
r"(이름은|불러줘|라고 해)", # identity information
r"(항상|절대|절대로|반드시)", # strong habits/rules
r"(처음|마지막|유일)", # unique experiences
r"(고마워|감사|사랑|보고싶)", # emotional intensity
r"(무서워|힘들어|외로워|슬퍼)",# emotional vulnerability
]
# Signals that lower importance
LOW_IMPORTANCE_PATTERNS = [
r"^(ㅋ+|ㅎ+|ㅠ+|ㅜ+)$", # emoticons only
r"^(응|어|네|아|오)\.?$", # single-syllable reactions
r"^(맞아|그렇구나|그래)\.?$", # simple acknowledgments
]
@dataclass
class ImportanceResult:
score: float # 0.0 ~ 1.0
is_worth_storing: bool
signals: list[str] # signals that contributed to the score
def quick_importance_filter(
user_msg: str,
bot_msg: str,
emotion_intensity: float = 0.5,
) -> ImportanceResult:
combined = user_msg + " " + bot_msg
score = 0.3 # base score
signals = []
# Low-importance patterns → immediately return a low score
for pattern in LOW_IMPORTANCE_PATTERNS:
if re.search(pattern, user_msg.strip()):
return ImportanceResult(score=0.05, is_worth_storing=False, signals=["low_signal"])
# Apply high-importance patterns
for pattern in HIGH_IMPORTANCE_PATTERNS:
if re.search(pattern, combined):
score += 0.15
signals.append(pattern)
# Factor in emotional intensity
score += emotion_intensity * 0.2
# Message length (longer messages are more likely to be important)
if len(user_msg) > 50:
score += 0.1
signals.append("long_message")
score = min(1.0, score)
return ImportanceResult(
score=score,
is_worth_storing=score >= 0.3,
signals=signals,
)
LLM-Based Importance Scoring (Second Pass, Optional)
For ambiguous conversations that pass the rule-based filter, we delegate scoring to an LLM.
async def llm_importance_score(
user_msg: str,
bot_msg: str,
llm_backend, # LLMBackend
) -> float:
"""Uses an LLM to evaluate how worth storing a conversation is for long-term memory."""
from ..llm.base import ChatMessage
prompt = f"""다음 대화가 AI 비서가 장기적으로 기억해야 할 가치가 있는지 평가하라.
대화:
사용자: {user_msg}
AI: {bot_msg}
기억 가치가 높은 대화:
- 사용자의 선호도, 이름, 직업, 가족 정보 포함
- 강한 감정적 반응
- 중요한 날짜나 사건 언급
기억 가치가 낮은 대화:
- 단순한 인사나 확인
- 반복적인 일상 대화
- 감탄사나 단음절 반응
0.0 (기억 불필요) ~ 1.0 (반드시 기억) 사이 숫자 하나만 출력:"""
result = await llm_backend.chat(
[ChatMessage(role="user", content=prompt)],
max_tokens=10,
temperature=0.1,
)
try:
return float(result.strip())
except ValueError:
return 0.4 # default value if parsing fails
Implementing Episodic Memory
Memory Entry Schema
# src/vtuber/memory/schema.py
from dataclasses import dataclass, field
from datetime import datetime
import uuid
@dataclass
class EpisodeMemory:
"""Episodic memory — memory of a specific conversation turn"""
id: str = field(default_factory=lambda: str(uuid.uuid4()))
user_id: str = "default"
user_message: str = ""
bot_message: str = ""
emotion: str = "calm"
emotion_intensity: float = 0.5
topics: list[str] = field(default_factory=list)
importance: float = 0.5
timestamp: float = field(default_factory=lambda: datetime.now().timestamp())
decay_factor: float = 1.0 # forgetting factor (1.0=vivid, 0.0=nearly forgotten)
def to_searchable_text(self) -> str:
"""Generate representative text for embedding"""
return f"{self.user_message} {self.bot_message}"
def to_payload(self) -> dict:
"""Serialize to Qdrant payload"""
return {
"id": self.id,
"user_id": self.user_id,
"user_message": self.user_message,
"bot_message": self.bot_message,
"emotion": self.emotion,
"emotion_intensity": self.emotion_intensity,
"topics": self.topics,
"importance": self.importance,
"timestamp": self.timestamp,
"decay_factor": self.decay_factor,
}
@dataclass
class SemanticFact:
"""Semantic memory — extracted facts about the user"""
id: str = field(default_factory=lambda: str(uuid.uuid4()))
user_id: str = "default"
fact: str = "" # "Minjun likes FPS games"
category: str = "general" # preference / personal / emotion / habit
confidence: float = 0.8
source_episode_id: str = ""
confirmed_count: int = 1
timestamp: float = field(default_factory=lambda: datetime.now().timestamp())
def to_payload(self) -> dict:
return {
"id": self.id,
"user_id": self.user_id,
"fact": self.fact,
"category": self.category,
"confidence": self.confidence,
"source_episode_id": self.source_episode_id,
"confirmed_count": self.confirmed_count,
"timestamp": self.timestamp,
}
Storing and Retrieving Episodes
# src/vtuber/memory/episodic.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
PointStruct, Filter, FieldCondition, MatchValue, Range,
)
from .schema import EpisodeMemory
from .embedder import EmbeddingBackend
from .store import EPISODE_COLLECTION
import uuid
class EpisodicMemoryStore:
def __init__(self, client: QdrantClient, embedder: EmbeddingBackend):
self._client = client
self._embedder = embedder
async def store(self, memory: EpisodeMemory) -> str:
"""Store an episodic memory in the vector DB"""
text = memory.to_searchable_text()
embedding = await self._embedder.embed(text)
self._client.upsert(
collection_name=EPISODE_COLLECTION,
points=[
PointStruct(
id=str(uuid.uuid4()),
vector=embedding,
payload=memory.to_payload(),
)
],
)
return memory.id
async def search(
self,
query: str,
user_id: str,
top_k: int = 5,
min_importance: float = 0.0,
min_decay: float = 0.1,
) -> list[EpisodeMemory]:
"""Retrieve episodic memories similar to the query"""
query_vec = await self._embedder.embed(query)
filters = [
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="importance", range=Range(gte=min_importance)),
FieldCondition(key="decay_factor", range=Range(gte=min_decay)),
]
results = self._client.search(
collection_name=EPISODE_COLLECTION,
query_vector=query_vec,
query_filter=Filter(must=filters),
limit=top_k,
with_payload=True,
)
memories = []
for hit in results:
p = hit.payload
memories.append(EpisodeMemory(
id=p["id"],
user_id=p["user_id"],
user_message=p["user_message"],
bot_message=p["bot_message"],
emotion=p["emotion"],
emotion_intensity=p["emotion_intensity"],
topics=p["topics"],
importance=p["importance"],
timestamp=p["timestamp"],
decay_factor=p["decay_factor"],
))
return memories
async def reinforce_memory(self, episode_id: str, boost: float = 0.2) -> None:
"""Boost the decay_factor of a retrieved memory (memory reinforcement)"""
results, _ = self._client.scroll(
collection_name=EPISODE_COLLECTION,
scroll_filter=Filter(
must=[FieldCondition(key="id", match=MatchValue(value=episode_id))]
),
limit=1,
with_payload=True,
with_vectors=False,
)
for point in results:
new_decay = min(1.0, point.payload["decay_factor"] + boost)
self._client.set_payload(
collection_name=EPISODE_COLLECTION,
payload={"decay_factor": new_decay},
points=[point.id],
)
Semantic Memory — Fact Extraction Engine
This engine summarizes episodic memories to extract semantic facts like "Minjun likes FPS games."
# src/vtuber/memory/semantic.py
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, Filter, FieldCondition, MatchValue
from .schema import SemanticFact
from .embedder import EmbeddingBackend
from .store import SEMANTIC_COLLECTION
from ..llm.base import LLMBackend, ChatMessage
import uuid, json, re
FACT_EXTRACTION_PROMPT = """다음 대화에서 사용자에 관해 장기적으로 기억할 만한 사실을 JSON으로 추출하라.
대화:
사용자: {user_msg}
AI: {bot_msg}
추출 기준:
- 사용자의 이름, 나이, 직업, 거주지 등 개인 정보
- 음식, 게임, 음악, 영화 등 취향/선호도
- 중요한 일정, 계획, 목표
- 반복되는 감정 패턴이나 습관
팩트가 없으면 빈 배열을 반환하라.
출력 형식 (JSON만):
[
{{"fact": "사실 내용", "category": "preference|personal|emotion|habit|goal", "confidence": 0.0~1.0}}
]"""
class SemanticMemoryStore:
def __init__(
self,
client: QdrantClient,
embedder: EmbeddingBackend,
llm: LLMBackend,
):
self._client = client
self._embedder = embedder
self._llm = llm
async def extract_and_store(
self,
user_id: str,
user_msg: str,
bot_msg: str,
episode_id: str,
) -> list[SemanticFact]:
"""Extract semantic facts from a conversation and store them."""
raw = await self._llm.chat(
[ChatMessage(
role="user",
content=FACT_EXTRACTION_PROMPT.format(
user_msg=user_msg, bot_msg=bot_msg
),
)],
max_tokens=256,
temperature=0.1,
)
json_match = re.search(r'\[.*\]', raw, re.DOTALL)
if not json_match:
return []
try:
items = json.loads(json_match.group())
except json.JSONDecodeError:
return []
stored = []
for item in items:
fact_text = item.get("fact", "").strip()
if not fact_text:
continue
# If a similar fact already exists, increment confirmed_count instead
existing = await self.search(fact_text, user_id, top_k=1, threshold=0.92)
if existing:
await self._increment_confirmation(existing[0])
stored.append(existing[0])
continue
fact = SemanticFact(
user_id=user_id,
fact=fact_text,
category=item.get("category", "general"),
confidence=float(item.get("confidence", 0.8)),
source_episode_id=episode_id,
)
embedding = await self._embedder.embed(fact_text)
self._client.upsert(
collection_name=SEMANTIC_COLLECTION,
points=[PointStruct(
id=str(uuid.uuid4()),
vector=embedding,
payload=fact.to_payload(),
)],
)
stored.append(fact)
return stored
async def search(
self,
query: str,
user_id: str,
top_k: int = 5,
threshold: float = 0.7,
) -> list[SemanticFact]:
"""Search for semantic facts relevant to the query."""
query_vec = await self._embedder.embed(query)
results = self._client.search(
collection_name=SEMANTIC_COLLECTION,
query_vector=query_vec,
query_filter=Filter(
must=[FieldCondition(key="user_id", match=MatchValue(value=user_id))]
),
limit=top_k,
score_threshold=threshold,
with_payload=True,
)
facts = []
for hit in results:
p = hit.payload
facts.append(SemanticFact(**{k: p[k] for k in SemanticFact.__dataclass_fields__}))
return facts
async def _increment_confirmation(self, fact: SemanticFact) -> None:
"""Boost confidence when the same fact is reconfirmed."""
fact.confirmed_count += 1
fact.confidence = min(1.0, fact.confidence + 0.05)
self._client.set_payload(
collection_name=SEMANTIC_COLLECTION,
payload={
"confirmed_count": fact.confirmed_count,
"confidence": fact.confidence,
},
points=[fact.id],
)
Forgetting Mechanism
For memory to feel real, forgetting needs to feel natural too. We apply the Ebbinghaus Forgetting Curve to the AI's memory system.
The Forgetting Curve Formula
retention(t) = e^(-t / stability)
t = elapsed time since last access (days)
stability = memory stability (higher importance = higher stability)
retention = remaining memory strength (0.0 ~ 1.0)
Memories with high importance that were accessed recently have a decay_factor close to 1.0; old, low-importance memories approach 0.0.
# src/vtuber/memory/forgetting.py
import math
from datetime import datetime
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
from .store import EPISODE_COLLECTION
def ebbinghaus_retention(
elapsed_days: float,
importance: float,
confirmed_count: int = 1,
) -> float:
"""Compute remaining memory strength using the Ebbinghaus forgetting curve.
stability scales with importance and repetition count.
"""
stability = 1.0 + (importance * 10) + (confirmed_count * 2)
return math.exp(-elapsed_days / stability)
class ForgettingEngine:
"""Runs periodically to update the decay_factor of each memory."""
def __init__(self, client: QdrantClient):
self._client = client
def run_decay_cycle(self, user_id: str) -> dict[str, int]:
"""Update decay_factor across all episodic memories.
Recommended cadence: once per day or at session start.
Returns: {"updated": n, "pruned": n}
"""
now = datetime.now().timestamp()
stats = {"updated": 0, "pruned": 0}
offset = None
while True:
results, next_offset = self._client.scroll(
collection_name=EPISODE_COLLECTION,
scroll_filter=Filter(
must=[FieldCondition(
key="user_id", match=MatchValue(value=user_id)
)]
),
limit=100,
offset=offset,
with_payload=True,
with_vectors=False,
)
if not results:
break
updates, prune_ids = [], []
for point in results:
p = point.payload
elapsed_days = (now - p["timestamp"]) / 86400
new_decay = ebbinghaus_retention(
elapsed_days=elapsed_days,
importance=p["importance"],
)
if new_decay < 0.05:
# Nearly forgotten — mark for deletion
prune_ids.append(point.id)
else:
updates.append((point.id, new_decay))
for point_id, decay in updates:
self._client.set_payload(
collection_name=EPISODE_COLLECTION,
payload={"decay_factor": decay},
points=[point_id],
)
stats["updated"] += len(updates)
if prune_ids:
self._client.delete(
collection_name=EPISODE_COLLECTION,
points_selector=prune_ids,
)
stats["pruned"] += len(prune_ids)
offset = next_offset
if offset is None:
break
return stats
Important Memories Don't Fade
There is one critical exception in the forgetting mechanism. Memories that the user explicitly brings up, or that are actually retrieved during a search, gain a higher stability. This mirrors how human memory actually works — the more you recall something, the longer it sticks. EpisodicMemoryStore.reinforce_memory() handles this.
Here are some example stability values based on importance and repetition count:
importance=0.3, confirmed=1 → stability=4.0 → 37% retained after 4 days
importance=0.7, confirmed=1 → stability=8.0 → 37% retained after 8 days
importance=0.9, confirmed=5 → stability=20.0 → 37% retained after 20 days
importance=0.9, confirmed=20 → stability=50.0 → 37% retained after 50 days
If "Minjun likes FPS games" has been confirmed across multiple conversations, the AI will still remember it months later.
Emotion Memory and Emotion History Tracking
For Haru to understand how a user's emotional state changes over time, emotion history must be tracked separately. She needs to remember "that day you said you were struggling" and compare it against the current situation.
# src/vtuber/memory/emotion_history.py
import sqlite3
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
@dataclass
class EmotionRecord:
user_id: str
emotion: str
intensity: float
trigger: str # What topic/situation triggered the emotion
timestamp: float
class EmotionHistoryDB:
"""SQLite-backed emotion history store"""
def __init__(self, db_path: str = "./data/emotion_history.db"):
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self) -> None:
self._conn.execute("""
CREATE TABLE IF NOT EXISTS emotion_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
emotion TEXT NOT NULL,
intensity REAL NOT NULL,
trigger TEXT DEFAULT '',
timestamp REAL NOT NULL
)
""")
self._conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_ts "
"ON emotion_records(user_id, timestamp)"
)
self._conn.commit()
def record(self, rec: EmotionRecord) -> None:
self._conn.execute(
"INSERT INTO emotion_records "
"(user_id, emotion, intensity, trigger, timestamp) "
"VALUES (?, ?, ?, ?, ?)",
(rec.user_id, rec.emotion, rec.intensity, rec.trigger, rec.timestamp),
)
self._conn.commit()
def get_recent_emotions(
self, user_id: str, days: int = 7
) -> list[EmotionRecord]:
since = datetime.now().timestamp() - days * 86400
rows = self._conn.execute(
"SELECT user_id, emotion, intensity, trigger, timestamp "
"FROM emotion_records WHERE user_id=? AND timestamp>=? "
"ORDER BY timestamp DESC LIMIT 50",
(user_id, since),
).fetchall()
return [EmotionRecord(*row) for row in rows]
def get_emotion_summary(self, user_id: str, days: int = 7) -> dict:
"""Summarize emotion patterns over the past N days"""
records = self.get_recent_emotions(user_id, days)
if not records:
return {"dominant": "unknown", "avg_intensity": 0.5, "low_days": []}
emotion_counts: dict[str, int] = {}
for r in records:
emotion_counts[r.emotion] = emotion_counts.get(r.emotion, 0) + 1
dominant = max(emotion_counts, key=emotion_counts.get)
avg_intensity = sum(r.intensity for r in records) / len(records)
# Days with low intensity (hard days)
low_days = sorted({
datetime.fromtimestamp(r.timestamp).strftime("%m/%d")
for r in records
if r.intensity < 0.3
and r.emotion in ("nervous", "tired", "concerned")
})
return {
"dominant": dominant,
"avg_intensity": round(avg_intensity, 2),
"low_days": low_days,
"counts": emotion_counts,
}
RAG Pipeline — Injecting Memories into the Prompt
Now we wire all the components together into a pipeline that automatically retrieves relevant memories and injects them into the system prompt whenever a user message arrives.
MemoryManager — Unified Memory Interface
# src/vtuber/memory/manager.py
from dataclasses import dataclass
from .episodic import EpisodicMemoryStore
from .semantic import SemanticMemoryStore
from .schema import EpisodeMemory, SemanticFact
from .importance import quick_importance_filter
from .emotion_history import EmotionHistoryDB, EmotionRecord
from .forgetting import ForgettingEngine
from ..character.state import CharacterState
from datetime import datetime
@dataclass
class MemoryContext:
"""Memory context assembled via RAG"""
relevant_episodes: list[EpisodeMemory]
relevant_facts: list[SemanticFact]
emotion_summary: dict
def to_prompt_section(self) -> str:
"""Serialize to text for injection into the system prompt"""
parts = []
if self.relevant_facts:
facts_text = "\n".join(
f"- {f.fact} (confidence: {f.confidence:.0%})"
for f in self.relevant_facts
)
parts.append(f"[Facts remembered about the user]\n{facts_text}")
if self.relevant_episodes:
episodes_text = "\n".join(
f"- ({datetime.fromtimestamp(e.timestamp).strftime('%m/%d')}) "
f"'{e.user_message[:40]}' → emotion at the time: {e.emotion}"
for e in self.relevant_episodes[:3] # up to 3
)
parts.append(f"[Relevant past conversations]\n{episodes_text}")
if self.emotion_summary.get("low_days"):
days = ", ".join(self.emotion_summary["low_days"])
parts.append(f"[Emotion history] Recent hard days: {days}")
return "\n\n".join(parts) if parts else ""
class MemoryManager:
"""Unified interface for managing episodic, semantic, and emotion memory"""
def __init__(
self,
episodic: EpisodicMemoryStore,
semantic: SemanticMemoryStore,
emotion_db: EmotionHistoryDB,
forgetting: ForgettingEngine,
llm, # LLMBackend
):
self._episodic = episodic
self._semantic = semantic
self._emotion_db = emotion_db
self._forgetting = forgetting
self._llm = llm
async def retrieve_context(
self,
user_id: str,
query: str,
top_k: int = 5,
) -> MemoryContext:
"""Retrieve memory context relevant to the user's message"""
import asyncio
episodes, facts = await asyncio.gather(
self._episodic.search(query, user_id, top_k=top_k),
self._semantic.search(query, user_id, top_k=top_k),
)
emotion_summary = self._emotion_db.get_emotion_summary(user_id, days=7)
# Reinforce retrieved episodes to slow decay
for ep in episodes:
await self._episodic.reinforce_memory(ep.id)
return MemoryContext(
relevant_episodes=episodes,
relevant_facts=facts,
emotion_summary=emotion_summary,
)
async def store_interaction(
self,
state: CharacterState,
user_msg: str,
bot_msg: str,
) -> None:
"""Store a conversation turn in memory and extract facts"""
imp_result = quick_importance_filter(
user_msg, bot_msg, state.emotion_intensity
)
episode = EpisodeMemory(
user_id=f"haru_{state.user_name}",
user_message=user_msg,
bot_message=bot_msg,
emotion=state.emotion.value,
emotion_intensity=state.emotion_intensity,
topics=state.recent_topics,
importance=imp_result.score,
)
if imp_result.is_worth_storing:
episode_id = await self._episodic.store(episode)
# Extract semantic memories only for importance >= 0.6
if imp_result.score >= 0.6:
await self._semantic.extract_and_store(
user_id=episode.user_id,
user_msg=user_msg,
bot_msg=bot_msg,
episode_id=episode_id,
)
# Always record emotion history
self._emotion_db.record(EmotionRecord(
user_id=episode.user_id,
emotion=state.emotion.value,
intensity=state.emotion_intensity,
trigger=", ".join(state.recent_topics),
timestamp=episode.timestamp,
))
Integrating with Part 2 — Updating chat.py
Now wire the memory system into the chat endpoint built in Part 2. There are two key changes:
- Before processing a request: retrieve relevant memories and inject them into the system prompt
- After processing a request: store the conversation in memory (asynchronously)
# src/vtuber/api/chat.py (업데이트 핵심 부분)
@router.post("/stream")
async def chat_stream(
request: ChatRequest,
llm: LLMBackend,
prompt_builder: PromptBuilder,
emotion_engine: EmotionTransitionEngine,
topic_tracker: TopicTracker,
memory_manager: MemoryManager, # ← 추가
):
state, history = get_or_create_session(request.session_id, request.user_name)
# ── 1. 메모리 컨텍스트 검색 ────────────────────────────────────
user_id = f"haru_{request.user_name}"
memory_ctx = await memory_manager.retrieve_context(
user_id=user_id,
query=request.message,
top_k=5,
)
memory_section = memory_ctx.to_prompt_section()
# ── 2. 시스템 프롬프트 생성 (메모리 주입) ──────────────────────
system_prompt = prompt_builder.build_system_prompt(state)
if memory_section:
system_prompt = system_prompt + "\n\n" + memory_section
messages = [
ChatMessage(role="system", content=system_prompt),
*history[-10:], # 단기 기억: 최근 10턴
ChatMessage(role="user", content=request.message),
]
async def generate():
full_response = []
async for chunk in llm.chat_stream(messages):
full_response.append(chunk.content)
if chunk.content:
yield f"data: {json.dumps({'type': 'token', 'content': chunk.content}, ensure_ascii=False)}\n\n"
if chunk.is_final:
response_text = "".join(full_response)
clean_text, detected_emotion = (
emotion_engine.parse_emotion_from_response(response_text)
)
updated_state = emotion_engine.update_state(
state, detected_emotion, request.message
)
updated_state = topic_tracker.update_topics(
updated_state, request.message, clean_text
)
_sessions[request.session_id] = (updated_state, history)
history.append(ChatMessage(role="user", content=request.message))
history.append(ChatMessage(role="assistant", content=clean_text))
# ── 3. 메모리 저장 (비동기, 응답 레이턴시에 영향 없음) ──
import asyncio
asyncio.create_task(
memory_manager.store_interaction(
state=updated_state,
user_msg=request.message,
bot_msg=clean_text,
)
)
meta = json.dumps({
"type": "done",
"emotion": updated_state.emotion.value,
"intimacy_level": updated_state.intimacy_level,
"memories_used": len(memory_ctx.relevant_episodes),
}, ensure_ascii=False)
yield f"data: {meta}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
asyncio.create_task() fully decouples memory storage from the response path. The save happens in the background after the user has already received the full response, so it has zero impact on latency.
Updated Project Structure
The structure as it grows from Part 2 to Part 3:
vtuber-assistant/
├── pyproject.toml
├── src/
│ └── vtuber/
│ ├── main.py ← MemoryManager 의존성 추가
│ ├── config.py
│ ├── character/ ← 2편에서 구현
│ │ ├── state.py
│ │ ├── engine.py
│ │ ├── tracker.py
│ │ └── prompt.py
│ ├── llm/ ← 2편에서 구현
│ │ ├── base.py
│ │ ├── ollama.py
│ │ └── claude.py
│ ├── memory/ ← 3편에서 추가
│ │ ├── store.py # Qdrant collection 초기화
│ │ ├── schema.py # EpisodeMemory, SemanticFact
│ │ ├── embedder.py # 임베딩 백엔드 추상화
│ │ ├── episodic.py # EpisodicMemoryStore
│ │ ├── semantic.py # SemanticMemoryStore + 팩트 추출
│ │ ├── importance.py # 중요도 평가
│ │ ├── forgetting.py # 망각 엔진
│ │ ├── emotion_history.py# SQLite 감정 이력
│ │ └── manager.py # MemoryManager (통합)
│ └── api/
│ └── chat.py ← 메모리 연동 업데이트
├── characters/ ← 2편에서 구현
│ └── haru/
│ ├── profile.yaml
│ └── system_prompt.j2
└── data/
├── qdrant/ ← 벡터 DB 파일
└── emotion_history.db ← SQLite 감정 이력
Additional Dependencies
# pyproject.toml dependencies에 추가
"qdrant-client>=1.12.0",
"sentence-transformers>=3.3.0",
Performance Considerations and Trade-offs
Embedding Latency
Embedding generation can be slower than you might expect:
sentence-transformers (CPU): ~50ms per text
sentence-transformers (GPU): ~5ms per text
Ollama nomic-embed-text: ~30ms per text
OpenAI API: ~100~300ms (네트워크)
retrieve_context is called before response generation, so it directly contributes to latency. Consider three optimization strategies:
Strategy 1: Skip search for short messages
# 의미 있는 쿼리에만 벡터 검색 실행
SKIP_SEARCH_PATTERNS = re.compile(r"^(ㅋ+|ㅎ+|응|어|네|맞아|그래)\.?$")
if SKIP_SEARCH_PATTERNS.match(request.message.strip()):
memory_ctx = MemoryContext([], [], {})
else:
memory_ctx = await memory_manager.retrieve_context(...)
Strategy 2: Run embedding lookup and state prep in parallel
# 임베딩 검색 + 시스템 프롬프트 빌드 병렬 처리
import asyncio
memory_ctx, system_base = await asyncio.gather(
memory_manager.retrieve_context(user_id, request.message),
asyncio.to_thread(prompt_builder.build_system_prompt, state),
)
Strategy 3: Cache embedding results
Apply lru_cache for repeated identical queries (e.g., the same message sent within a short window). Keep the cache TTL short.
Context Length Management
Memory injection can make the system prompt much longer than intended:
권장 시스템 프롬프트 길이: 1,000 ~ 2,000 토큰
- 캐릭터 기본 프롬프트: ~500 토큰
- 메모리 컨텍스트: ~300 토큰
- 현재 상태 주입: ~200 토큰
That's why to_prompt_section() caps episodes at 3 and summarizes each to 40 characters. Keeping the token budget tight lets the LLM make better use of the injected memories.
Verifying the System in Action
Use the following scenario to confirm the memory system is working:
# 세션 1 (첫 대화)
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"session_id":"test","message":"나 롤 게임 엄청 좋아해요!","user_name":"민준"}'
# → EpisodeMemory 저장: emotion=excited, importance=0.75
# → SemanticFact 추출: "민준은 롤(League of Legends) 게임을 좋아한다"
# ... 30번의 대화 후 (세션 여러 번 재시작) ...
# 세션 N (나중 대화)
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"session_id":"new","message":"요즘 뭐하면서 놀아요?","user_name":"민준"}'
# 시스템 프롬프트에 자동 주입:
# [사용자에 대해 기억하는 사실]
# - 민준은 롤(League of Legends) 게임을 좋아한다 (확신도: 80%)
#
# → 하루 응답: "[EMOTION:playful] 롤이요, 아직도 하세요?
# 저번에 너무 좋아한다고 했잖아요~"
Real memory that reaches back beyond history[-20:] is now in place.
Wrap-up
Here's a summary of everything built in this installment:
- Mapping human memory models to AI: episodic memory (Qdrant vector DB) + semantic memory (SQLite fact extraction) + emotional memory (emotion history tracking)
- Embedding abstraction: a swappable
EmbeddingBackendsupporting SentenceTransformer and Ollama - Importance filtering: rule-based fast filter + LLM-based fine-grained scoring, so only meaningful memories get stored
- Ebbinghaus forgetting curve: old memories fade naturally; retrieved memories get reinforced
- RAG pipeline: user message → vector search → relevant memories → automatic system prompt injection
- Async storage: memory is saved in the background with no impact on response latency
Haru now genuinely remembers. A conversation from today persists a week later, and the system accumulates knowledge of the user's preferences over time. By around the 30th conversation, Haru will be the one to ask, "Still playing League?"
Part 4 goes beyond text chat and builds a system where Haru speaks and listens. That means real-time STT with faster-whisper, voice activity detection, and an interruptible pipeline tying it all together.
Coming Up in Part 4
Part 4: STT — Teaching the AI to Listen
The moment you started talking, Haru responded. When you interrupted her mid-sentence, she stopped and listened again.
Topics covered:
- VAD (Voice Activity Detection): real-time detection that distinguishes speech from silence
- faster-whisper pipeline: implementing streaming, real-time STT
- Interrupt mechanism: the full handling flow when a user cuts in while the AI is speaking
- Echo cancellation: keeping the AI's own voice out of the microphone without headphones
- WebSocket audio stream: a real-time pipeline from browser microphone → server → STT