AI Agent Memory Management: Building Multi-Turn Conversation Memory Systems

AI与大数据

Summary

  • AI Agent memory systems are divided into 3 layers: working memory (context window), short-term memory (session-level), and long-term memory (cross-session persistence)
  • Conversation context compression is essential for long conversation scenarios; LLM summary compression can compress 100 turns of conversation into under 2K tokens
  • Three storage modes for long-term memory: vector retrieval memory, knowledge graph memory, and structured key-value memory
  • RAG approach for memory retrieval: embed historical conversations into a vector database and recall key segments by relevance
  • This article provides a complete solution from memory architecture design to production deployment, including Redis+Milvus persistence implementation

Table of Contents

3-Layer Architecture of AI Agent Memory Systems

┌──────────────────────────────────────────────────────────────┐ │ AI Agent 3-Layer Memory Architecture │ │ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 1: Working Memory │ │ │ │ Conversation history within context window │ │ │ │ Capacity: 4K-128K tokens │ │ │ │ Latency: 0ms (already in memory) │ │ │ │ Lifecycle: Single request │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ Overflow compression │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 2: Short-Term Memory │ │ │ │ Complete conversation state for current session │ │ │ │ Storage: Redis / In-memory cache │ │ │ │ Capacity: Unlimited │ │ │ │ Lifecycle: Single session (expires after browser close)│ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ Key information extraction │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 3: Long-Term Memory │ │ │ │ Cross-session user preferences, knowledge, history │ │ │ │ Storage: Milvus vector DB + PostgreSQL structured │ │ │ │ Capacity: Unlimited │ │ │ │ Lifecycle: Permanent │ │ │ └──────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

3-Layer Memory Comparison

Dimension Working Memory Short-Term Memory Long-Term Memory
Storage LLM context window Redis/Memory Milvus+PostgreSQL
Capacity 4K-128K tokens Unlimited Unlimited
Latency 0ms 1-5ms 10-30ms
Lifecycle Single request Single session Permanent
Retrieval Direct access Key-Value Vector+Structured
Typical Content Current conversation Session history User preferences/knowledge

Working Memory: Context Window Management

Token Budget Allocation

┌──────────────────────────────────────────────────────────┐ │ Token Budget Allocation for 8K Context Window│ │ │ │ System Prompt: 500 tokens (6%) │ │ Long-term memory recall: 1000 tokens (12%) │ │ Conversation history: 5000 tokens (62%) │ │ Tool call results: 1000 tokens (12%) │ │ Reserved output space: 500 tokens (6%) │ │ ───────────────────────────────── │ │ Total: 8000 tokens │ └──────────────────────────────────────────────────────────┘

Context Window Manager

`python from dataclasses import dataclass, field from typing import Optional

@dataclass class Message: role: str content: str token_count: int = 0 metadata: dict = field(default_factory=dict)

class ContextWindowManager: def init( self, max_tokens: int = 8192, system_prompt_tokens: int = 500, reserved_output_tokens: int = 500, memory_recall_tokens: int = 1000, ): self.max_tokens = max_tokens self.system_prompt_tokens = system_prompt_tokens self.reserved_output_tokens = reserved_output_tokens self.memory_recall_tokens = memory_recall_tokens self.available_for_history = ( max_tokens - system_prompt_tokens - reserved_output_tokens - memory_recall_tokens )

def build_context(
    self,
    system_prompt: str,
    memory_facts: list[str],
    conversation_history: list[Message],
    tool_results: list[str] = None,
) -> list[dict]:
    messages = [{"role": "system", "content": system_prompt}]

    if memory_facts:
        memory_text = "User-related memory:\n" + "\n".join(f"- {f}" for f in memory_facts)
        messages.append({"role": "system", "content": memory_text})

    budget = self.available_for_history
    selected_history = []
    for msg in reversed(conversation_history):
        if budget <= 0:
            break
        if msg.token_count <= budget:
            selected_history.insert(0, msg)
            budget -= msg.token_count
        else:
            break

    for msg in selected_history:
        messages.append({"role": msg.role, "content": msg.content})

    if tool_results:
        for result in tool_results:
            messages.append({"role": "tool", "content": result})

    return messages

`

Short-Term Memory: Session-Level State Persistence

Redis Session Storage

`python import json import redis from datetime import timedelta

class SessionMemory: def init(self, redis_url: str = "redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.session_ttl = timedelta(hours=24)

def _key(self, session_id: str) -> str:
    return f"session:{session_id}"

async def save_message(self, session_id: str, role: str, content: str):
    key = self._key(session_id)
    message = {"role": role, "content": content, "timestamp": int(time.time())}
    await self.redis.rpush(key, json.dumps(message, ensure_ascii=False))
    await self.redis.expire(key, self.session_ttl)

async def get_history(self, session_id: str, limit: int = 50) -> list[dict]:
    key = self._key(session_id)
    raw_messages = await self.redis.lrange(key, -limit, -1)
    return [json.loads(m) for m in raw_messages]

async def get_session_summary(self, session_id: str) -> Optional[str]:
    key = f"session_summary:{session_id}"
    summary = await self.redis.get(key)
    return summary.decode() if summary else None

async def save_session_summary(self, session_id: str, summary: str):
    key = f"session_summary:{session_id}"
    await self.redis.set(key, summary, ex=self.session_ttl)

async def clear_session(self, session_id: str):
    await self.redis.delete(self._key(session_id))
    await self.redis.delete(f"session_summary:{session_id}")

`

Long-Term Memory: Cross-Session Knowledge Accumulation

Three Long-Term Memory Modes

Mode Storage Retrieval Method Use Case
Vector Retrieval Memory Milvus Semantic similarity Open-ended knowledge recall
Knowledge Graph Memory Neo4j Relationship traversal Structured relationship reasoning
Structured Key-Value Memory PostgreSQL Exact match User preferences/configuration

Vector Retrieval Memory Implementation

`python from pymilvus import MilvusClient, DataType from openai import OpenAI import hashlib

class VectorLongTermMemory: def init(self, milvus_uri: str, embedding_model: str = "text-embedding-3-small"): self.client = MilvusClient(uri=milvus_uri) self.embedding_client = OpenAI() self.embedding_model = embedding_model self._ensure_collection()

def _ensure_collection(self):
    if self.client.has_collection("agent_memory"):
        return
    schema = self.client.create_schema(auto_id=True, enable_dynamic_field=True)
    schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
    schema.add_field(field_name="user_id", datatype=DataType.VARCHAR, max_length=128)
    schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535)
    schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=1536)
    schema.add_field(field_name="category", datatype=DataType.VARCHAR, max_length=64)
    schema.add_field(field_name="timestamp", datatype=DataType.INT64)

    index_params = self.client.prepare_index_params()
    index_params.add_index(field_name="embedding", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 200})

    self.client.create_collection("agent_memory", schema=schema, index_params=index_params)

def _embed(self, text: str) -> list[float]:
    response = self.embedding_client.embeddings.create(input=text, model=self.embedding_model)
    return response.data[0].embedding

async def store(self, user_id: str, content: str, category: str = "general"):
    embedding = self._embed(content)
    self.client.insert("agent_memory", [{
        "user_id": user_id,
        "content": content,
        "embedding": embedding,
        "category": category,
        "timestamp": int(time.time()),
    }])

async def recall(self, user_id: str, query: str, top_k: int = 5) -> list[str]:
    query_embedding = self._embed(query)
    results = self.client.search(
        "agent_memory",
        data=[query_embedding],
        filter=f'user_id == "{user_id}"',
        limit=top_k,
        output_fields=["content", "category"],
        search_params={"metric_type": "COSINE", "params": {"ef": 64}},
    )
    return [r["entity"]["content"] for r in results[0]]

`

Structured Key-Value Memory Implementation

`python import asyncpg

class StructuredLongTermMemory: def init(self, db_url: str): self.db_url = db_url self.pool = None

async def init(self):
    self.pool = await asyncpg.create_pool(self.db_url, min_size=2, max_size=10)
    async with self.pool.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS user_memory (
                id SERIAL PRIMARY KEY,
                user_id VARCHAR(128) NOT NULL,
                memory_key VARCHAR(256) NOT NULL,
                memory_value TEXT NOT NULL,
                category VARCHAR(64) DEFAULT 'general',
                updated_at TIMESTAMP DEFAULT NOW(),
                UNIQUE(user_id, memory_key)
            )
        """)

async def set_memory(self, user_id: str, key: str, value: str, category: str = "general"):
    async with self.pool.acquire() as conn:
        await conn.execute("""
            INSERT INTO user_memory (user_id, memory_key, memory_value, category)
            VALUES (, , , )
            ON CONFLICT (user_id, memory_key)
            DO UPDATE SET memory_value = , category = , updated_at = NOW()
        """, user_id, key, value, category)

async def get_memory(self, user_id: str, key: str) -> Optional[str]:
    async with self.pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT memory_value FROM user_memory WHERE user_id =  AND memory_key = ",
            user_id, key
        )
        return row["memory_value"] if row else None

async def get_all_memories(self, user_id: str, category: str = None) -> list[dict]:
    async with self.pool.acquire() as conn:
        if category:
            rows = await conn.fetch(
                "SELECT memory_key, memory_value, category FROM user_memory WHERE user_id =  AND category = ",
                user_id, category
            )
        else:
            rows = await conn.fetch(
                "SELECT memory_key, memory_value, category FROM user_memory WHERE user_id = ",
                user_id
            )
        return [{"key": r["memory_key"], "value": r["memory_value"], "category": r["category"]} for r in rows]

`

Conversation Compression: Essential for Long Conversation Scenarios

Three Compression Strategies

Strategy Compression Ratio Information Loss Latency Use Case
Sliding window truncation High Large 0ms Simple conversations
LLM summary compression Medium Small 200-500ms Long conversations
Structured extraction Low Minimal 100-300ms Task-oriented conversations

LLM Summary Compression

`python class ConversationCompressor: def init(self, llm_client, max_summary_tokens: int = 512): self.llm = llm_client self.max_summary_tokens = max_summary_tokens

async def compress(self, messages: list[dict], keep_recent: int = 4) -> list[dict]:
    if len(messages) <= keep_recent + 2:
        return messages

    to_compress = messages[:-keep_recent]
    recent = messages[-keep_recent:]

    conversation_text = "\n".join(
        f"{m['role']}: {m['content'][:200]}" for m in to_compress
    )

    response = self.llm.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[{
            "role": "system",
            "content": "Compress the following conversation history into a concise summary, preserving key facts, decisions, and user preferences. Output only the summary without explanation."
        }, {
            "role": "user",
            "content": conversation_text
        }],
        max_tokens=self.max_summary_tokens,
        temperature=0.0,
    )

    summary = response.choices[0].message.content
    return [
        {"role": "system", "content": f"Previous conversation summary: {summary}"},
        *recent,
    ]

`

Structured Information Extraction

`python class StructuredMemoryExtractor: def init(self, llm_client): self.llm = llm_client

async def extract(self, user_message: str, assistant_message: str) -> list[dict]:
    response = self.llm.chat.completions.create(
        model="Qwen/Qwen2.5-7B-Instruct",
        messages=[{
            "role": "system",
            "content": """Extract structured information from the conversation that should be remembered long-term.

Output a JSON array where each element contains: key (memory key name), value (memory value), category (classification: preference/fact/decision/task)

Example output: [{"key":"preferred_language","value":"Python","category":"preference"}]

If there is no information worth remembering, output an empty array []""" }, { "role": "user", "content": f"User: {user_message}\nAssistant: {assistant_message}" }], max_tokens=256, temperature=0.0, response_format={"type": "json_object"}, )

    try:
        data = json.loads(response.choices[0].message.content)
        return data.get("memories", data.get("items", []))
    except json.JSONDecodeError:
        return []

`

Production Deployment: Redis+Milvus Memory Service

Unified Memory Service Interface

`python class AgentMemoryService: def init(self, redis_url: str, milvus_uri: str, pg_url: str, llm_client): self.session_memory = SessionMemory(redis_url) self.vector_memory = VectorLongTermMemory(milvus_uri) self.structured_memory = StructuredLongTermMemory(pg_url) self.compressor = ConversationCompressor(llm_client) self.extractor = StructuredMemoryExtractor(llm_client)

async def on_user_message(self, session_id: str, user_id: str, message: str) -> list[str]:
    vector_facts = await self.vector_memory.recall(user_id, message, top_k=3)
    structured = await self.structured_memory.get_all_memories(user_id)
    preference_facts = [f"{m['key']}: {m['value']}" for m in structured if m["category"] == "preference"]
    return vector_facts + preference_facts

async def on_assistant_message(self, session_id: str, user_id: str, user_msg: str, assistant_msg: str):
    await self.session_memory.save_message(session_id, "user", user_msg)
    await self.session_memory.save_message(session_id, "assistant", assistant_msg)

    memories = await self.extractor.extract(user_msg, assistant_msg)
    for mem in memories:
        await self.structured_memory.set_memory(user_id, mem["key"], mem["value"], mem["category"])

    await self.vector_memory.store(user_id, f"User said: {user_msg}", category="conversation")

async def build_prompt(self, session_id: str, user_id: str, current_message: str) -> list[dict]:
    memory_facts = await self.on_user_message(session_id, user_id, current_message)
    history = await self.session_memory.get_history(session_id, limit=50)
    messages = self.session_memory.build_context(
        system_prompt="You are an AI assistant with memory that can remember user preferences and conversation history.",
        memory_facts=memory_facts,
        conversation_history=history,
    )
    messages.append({"role": "user", "content": current_message})
    return messages

`

K8s Deployment

yaml apiVersion: apps/v1 kind: Deployment metadata: name: agent-memory-service namespace: ai-agent spec: replicas: 2 selector: matchLabels: app: agent-memory-service template: spec: containers: - name: memory-service image: myregistry/agent-memory-service:v1.0 ports: - containerPort: 8080 resources: requests: cpu: "1" memory: 512Mi limits: cpu: "2" memory: 1Gi env: - name: REDIS_URL value: "redis://redis:6379" - name: MILVUS_URI value: "http://milvus-svc:19530" - name: PG_URL value: "postgresql://postgres:password@postgres:5432/agent_memory"

Memory System Performance Benchmarks

Operation Latency (P50) Latency (P99) QPS
Save message (Redis) 1.2ms 3ms 8000
Get history (Redis) 2ms 5ms 6000
Vector store (Milvus) 15ms 30ms 500
Vector recall (Milvus) 8ms 15ms 1000
Structured store (PG) 3ms 8ms 3000
Structured query (PG) 2ms 5ms 5000
Conversation compression (LLM) 350ms 800ms 50

Summary and Further Reading

The AI Agent memory system is the core infrastructure of Agent intelligence. The 3-layer architecture (working memory -> short-term memory -> long-term memory) solves three major problems: limited context windows, lost session state, and forgotten cross-session knowledge. Conversation compression and structured extraction are key technologies for long conversation scenarios.

Key Development Takeaways:

  1. 3-layer memory architecture: Working memory (context window) -> Short-term memory (Redis) -> Long-term memory (Milvus+PG)
  2. Token budget allocation: System 6% + Memory recall 12% + Conversation history 62% + Tools 12% + Output 6%
  3. Prefer LLM summary for conversation compression; use structured extraction for task-oriented conversations
  4. Three long-term memory modes: Vector retrieval (open-ended), Knowledge graph (relational), Structured key-value (exact match)
  5. Production deployment with the Redis+Milvus+PostgreSQL trio

Related Reading:

Authoritative References:

Try these browser-local tools — no sign-up required →

#AI Agent记忆管理#多轮对话记忆#长期记忆架构#Agent状态持久化#对话上下文压缩#2026