AI Agent多輪記憶實戰：構建長期記憶與上下文壓縮系統

AI与大数据

摘要

  • AI Agent的記憶系統分為3層：工作記憶（上下文視窗）、短期記憶（會話級）、長期記憶（跨會話持久化）
  • 對話上下文壓縮是長對話場景的剛需，LLM摘要壓縮可將100輪對話壓縮到2K tokens內
  • 長期記憶的3種儲存模式：向量檢索記憶、知識圖譜記憶、結構化鍵值記憶
  • 記憶檢索的RAG方案：將歷史對話嵌入向量資料庫，按相關性召回關鍵片段
  • 本文提供從記憶架構設計到生產部署的完整方案，含Redis+Milvus持久化實作

目錄

AI Agent記憶系統的3層架構

┌──────────────────────────────────────────────────────────────┐ │ AI Agent 3層記憶架構 │ │ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 1: 工作記憶 (Working Memory) │ │ │ │ 上下文視窗內的對話歷史 │ │ │ │ 容量: 4K-128K tokens │ │ │ │ 延遲: 0ms (已在記憶體) │ │ │ │ 生命週期: 單次請求 │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ 溢出壓縮 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 2: 短期記憶 (Short-Term Memory) │ │ │ │ 當前會話的完整對話狀態 │ │ │ │ 儲存: Redis / 記憶體快取 │ │ │ │ 容量: 無限制 │ │ │ │ 生命週期: 單次會話 (關閉瀏覽器後過期) │ │ │ └──────────────────────────────────────────────────────┘ │ │ ↓ 關鍵資訊提取 │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Layer 3: 長期記憶 (Long-Term Memory) │ │ │ │ 跨會話的使用者偏好、知識、歷史 │ │ │ │ 儲存: Milvus向量庫 + PostgreSQL結構化 │ │ │ │ 容量: 無限制 │ │ │ │ 生命週期: 永久 │ │ │ └──────────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

3層記憶對比

維度 工作記憶 短期記憶 長期記憶
儲存 LLM上下文視窗 Redis/記憶體 Milvus+PostgreSQL
容量 4K-128K tokens 無限制 無限制
延遲 0ms 1-5ms 10-30ms
生命週期 單次請求 單次會話 永久
檢索方式 直接存取 Key-Value 向量+結構化
典型內容 當前對話 會話歷史 使用者偏好/知識

工作記憶：上下文視窗管理

Token預算分配

┌──────────────────────────────────────────────────────────┐ │ 8K上下文視窗的Token預算分配 │ │ │ │ System Prompt: 500 tokens (6%) │ │ 長期記憶召回: 1000 tokens (12%) │ │ 對話歷史: 5000 tokens (62%) │ │ 工具呼叫結果: 1000 tokens (12%) │ │ 預留輸出空間: 500 tokens (6%) │ │ ───────────────────────────────── │ │ 合計: 8000 tokens │ └──────────────────────────────────────────────────────────┘

上下文視窗管理器

`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 = "使用者相關記憶：\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

`

短期記憶：會話級狀態持久化

Redis會話儲存

`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}")

`

長期記憶：跨會話知識積累

3種長期記憶模式

模式 儲存 檢索方式 適用場景
向量檢索記憶 Milvus 語意相似度 開放式知識回憶
知識圖譜記憶 Neo4j 關係遍歷 結構化關係推理
結構化鍵值記憶 PostgreSQL 精確匹配 使用者偏好/設定

向量檢索記憶實作

`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]]

`

結構化鍵值記憶實作

`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]

`

對話壓縮：長對話場景的剛需

3種壓縮策略

策略 壓縮比 資訊損失 延遲 適用場景
滑動視窗截斷 高 大 0ms 簡單對話
LLM摘要壓縮 中 小 200-500ms 長對話
結構化提取 低 極小 100-300ms 任務型對話

LLM摘要壓縮

`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": "將以下對話歷史壓縮為簡潔摘要，保留關鍵事實、決策和使用者偏好。只輸出摘要內容，不要解釋。"
        }, {
            "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"之前的對話摘要：{summary}"},
        *recent,
    ]

`

結構化資訊提取

`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": """從對話中提取需要長期記住的結構化資訊。

輸出JSON陣列，每個元素包含：key(記憶鍵名)、value(記憶值)、category(分類：preference/fact/decision/task)

範例輸出： [{"key":"preferred_language","value":"Python","category":"preference"}]

如果沒有需要記住的資訊，輸出空陣列 []""" }, { "role": "user", "content": f"使用者: {user_message}\n助手: {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 []

`

生產部署：Redis+Milvus記憶服務

記憶服務統一介面

`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_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="你是一個有記憶的AI助手，能記住使用者的偏好和歷史對話。",
        memory_facts=memory_facts,
        conversation_history=history,
    )
    messages.append({"role": "user", "content": current_message})
    return messages

`

K8s部署

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"

記憶系統效能基準

操作 延遲(P50) 延遲(P99) QPS
儲存訊息(Redis) 1.2ms 3ms 8000
取得歷史(Redis) 2ms 5ms 6000
向量儲存(Milvus) 15ms 30ms 500
向量召回(Milvus) 8ms 15ms 1000
結構化儲存(PG) 3ms 8ms 3000
結構化查詢(PG) 2ms 5ms 5000
對話壓縮(LLM) 350ms 800ms 50

總結與延伸閱讀

AI Agent記憶系統是Agent智慧的核心基礎設施。3層架構（工作記憶→短期記憶→長期記憶）解決了上下文視窗有限、會話狀態遺失、跨會話知識遺忘三大問題。對話壓縮和結構化提取是長對話場景的關鍵技術。

開發要點回顧：

  1. 3層記憶架構：工作記憶(上下文視窗)→短期記憶(Redis)→長期記憶(Milvus+PG)
  2. Token預算分配：System 6% + 記憶召回 12% + 對話歷史 62% + 工具 12% + 輸出 6%
  3. 對話壓縮優先使用LLM摘要，任務型對話使用結構化提取
  4. 長期記憶3種模式：向量檢索(開放式)、知識圖譜(關係型)、結構化鍵值(精確匹配)
  5. 生產部署Redis+Milvus+PostgreSQL三件套

相關閱讀：

權威參考：

本站提供瀏覽器本地工具，免註冊即可試用 →

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