模块7: 记忆与上下文管理

每个LLM都有一个有限的上下文窗口 — 它在单次请求中能处理的最大token数。这包括系统提示、所有对话历史、工具定义和模型的响应。当您的对话超过此限制时，您必须决定保留什么和丢弃什么。

上下文窗口已经大幅增长（从早期GPT-3.5的4K token到Claude的200K和Gemini的1M+），但它们仍然是有限的。更重要的是，更长的上下文花费更多的钱并增加延迟。即使模型支持200K token，如果只有5K是相关的，每次请求都发送200K token是浪费的。

因此，记忆管理涉及两个方面：保持在限制范围内，以及保持效率。一个设计良好的记忆系统让智能体可以访问所有需要的信息，同时将token使用量降到最低。

最简单的记忆策略是滑动窗口：保留最近的N条消息，丢弃更早的所有内容。这类似于人类专注于当前对话同时遗忘早期细节的方式。

关键的设计决策是在哪里截断。简单的做法是丢弃最旧的消息，但这可能会移除系统提示或重要的早期指令。更好的做法是始终保留系统提示和第一条用户消息，然后对中间的所有内容应用窗口。

class SlidingWindowMemory:
    """Keep the system prompt + last N message pairs."""

    def __init__(self, max_pairs: int = 20):
        self.system_prompt = ""
        self.messages: list[dict] = []
        self.max_pairs = max_pairs  # each pair = user + assistant

    def set_system(self, prompt: str):
        self.system_prompt = prompt

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # Keep max_pairs * 2 messages (user+assistant pairs)
        max_msgs = self.max_pairs * 2
        if len(self.messages) > max_msgs:
            self.messages = self.messages[-max_msgs:]

    def get_messages(self) -> list[dict]:
        """Return messages formatted for the API."""
        result = []
        if self.system_prompt:
            result.append({"role": "system", "content": self.system_prompt})
        result.extend(self.messages)
        return result

    @property
    def token_estimate(self) -> int:
        """Rough token count (4 chars per token)."""
        total = len(self.system_prompt)
        total += sum(len(m["content"]) for m in self.messages)
        return total // 4

基于Token的窗口

import tiktoken

class TokenWindowMemory:
    """Keep messages that fit within a token budget."""

    def __init__(self, max_tokens: int = 8000, model: str = "gpt-4o"):
        self.max_tokens = max_tokens
        self.encoder = tiktoken.encoding_for_model(model)
        self.messages: list[dict] = []

    def _count_tokens(self, messages: list[dict]) -> int:
        return sum(len(self.encoder.encode(m["content"])) for m in messages)

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # Trim from the front until we're within budget
        while (len(self.messages) > 2 and
               self._count_tokens(self.messages) > self.max_tokens):
            self.messages.pop(1)  # keep index 0 (first message)

滑动窗口会永久丢失信息。摘要策略将较旧的消息压缩成简洁的摘要，保留关键事实同时减少token数量。您使用LLM本身来生成这些摘要。

典型的做法是：当对话超过阈值时，取最旧的一批消息，进行摘要，用摘要替换它们，然后继续。这创建了一个分层记忆，其中最近的消息是逐字记录的，而较旧的消息是压缩的。

import anthropic

client = anthropic.Anthropic()

class SummarisedMemory:
    """Memory that summarises old messages to stay within budget."""

    def __init__(self, max_tokens: int = 6000, summary_threshold: int = 8000):
        self.messages: list[dict] = []
        self.summary: str = ""
        self.max_tokens = max_tokens
        self.summary_threshold = summary_threshold

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if self._estimate_tokens() > self.summary_threshold:
            self._compress()

    def _estimate_tokens(self) -> int:
        total = len(self.summary)
        total += sum(len(m["content"]) for m in self.messages)
        return total // 4

    def _compress(self):
        """Summarise the oldest half of messages."""
        mid = len(self.messages) // 2
        old_messages = self.messages[:mid]
        self.messages = self.messages[mid:]

        # Use the LLM to create a summary
        conversation = "\n".join(
            f'{m["role"]}: {m["content"]}' for m in old_messages
        )
        prompt = f"""Summarise this conversation, preserving:
- Key decisions and facts
- User preferences mentioned
- Any pending tasks or commitments

Previous summary: {self.summary or 'None'}

Conversation to summarise:
{conversation}"""

        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}]
        )
        self.summary = response.content[0].text

    def get_messages(self) -> list[dict]:
        msgs = []
        if self.summary:
            msgs.append({"role": "system",
                         "content": f"Conversation summary so far:\n{self.summary}"})
        msgs.extend(self.messages)
        return msgs

对于需要与用户在数天、数周或数月内交互的智能体，您需要跨会话持久化的长期记忆。最有效的方法使用向量嵌入来语义化地存储和检索相关的过往交互。

您不是重放整个历史记录，而是将每轮对话嵌入并存储在向量数据库中。当用户发送新消息时，您搜索最相关的过往交互并将其作为上下文包含在内。这使智能体看似拥有无限记忆，同时使用最少的token。

import chromadb
from datetime import datetime

class LongTermMemory:
    """Semantic long-term memory using vector embeddings."""

    def __init__(self, user_id: str):
        self.client = chromadb.PersistentClient(path="./memory_db")
        self.collection = self.client.get_or_create_collection(
            name=f"memory_{user_id}",
            metadata={"hnsw:space": "cosine"}
        )
        self.user_id = user_id

    def store(self, content: str, metadata: dict = None):
        """Store a memory with timestamp."""
        meta = {"timestamp": datetime.now().isoformat(),
                "user_id": self.user_id}
        if metadata:
            meta.update(metadata)
        self.collection.add(
            documents=[content],
            metadatas=[meta],
            ids=[f"mem_{datetime.now().timestamp()}"]
        )

    def recall(self, query: str, n_results: int = 5) -> list[str]:
        """Retrieve relevant memories for a query."""
        if self.collection.count() == 0:
            return []
        results = self.collection.query(
            query_texts=[query],
            n_results=min(n_results, self.collection.count())
        )
        return results["documents"][0]

    def get_context_string(self, query: str) -> str:
        """Format memories as context for the LLM."""
        memories = self.recall(query)
        if not memories:
            return ""
        formatted = "\n".join(f"- {m}" for m in memories)
        return f"Relevant memories from past conversations:\n{formatted}"

不同类型的记忆数据需要不同的存储后端。在实践中，生产级智能体会同时使用多个存储系统。

使用Redis的用户档案存储

import redis
import json

class UserProfileMemory:
    """Fast key-value store for user preferences and facts."""

    def __init__(self, user_id: str):
        self.redis = redis.Redis(host="localhost", port=6379, db=0)
        self.user_id = user_id
        self.key = f"agent:user:{user_id}:profile"

    def set_fact(self, key: str, value: str):
        """Store a user fact. e.g., set_fact('timezone', 'UTC+1')"""
        self.redis.hset(self.key, key, value)

    def get_fact(self, key: str) -> str | None:
        val = self.redis.hget(self.key, key)
        return val.decode() if val else None

    def get_all_facts(self) -> dict:
        data = self.redis.hgetall(self.key)
        return {k.decode(): v.decode() for k, v in data.items()}

    def get_context_string(self) -> str:
        facts = self.get_all_facts()
        if not facts:
            return ""
        lines = [f"- {k}: {v}" for k, v in facts.items()]
        return "Known user preferences:\n" + "\n".join(lines)

生产级的记忆系统通常分层使用多种策略。以下是一个完整示例，结合了滑动窗口（近期）、摘要（中期）和向量检索（长期）。

class AgentMemory:
    """Layered memory: recent window + summary + long-term retrieval."""

    def __init__(self, user_id: str, system_prompt: str):
        self.system_prompt = system_prompt
        self.window = SlidingWindowMemory(max_pairs=10)
        self.summarised = SummarisedMemory()
        self.long_term = LongTermMemory(user_id)
        self.profile = UserProfileMemory(user_id)

    def add_turn(self, user_msg: str, assistant_msg: str):
        # Store in short-term window
        self.window.add("user", user_msg)
        self.window.add("assistant", assistant_msg)
        # Store in long-term
        self.long_term.store(f"User: {user_msg}\nAssistant: {assistant_msg}")

    def build_messages(self, current_query: str) -> list[dict]:
        # Layer 1: System prompt + user profile
        profile_ctx = self.profile.get_context_string()
        system = self.system_prompt
        if profile_ctx:
            system += f"\n\n{profile_ctx}"

        # Layer 2: Relevant long-term memories
        ltm_ctx = self.long_term.get_context_string(current_query)
        if ltm_ctx:
            system += f"\n\n{ltm_ctx}"

        # Layer 3: Conversation summary (if any)
        if self.summarised.summary:
            system += f"\n\nConversation summary:\n{self.summarised.summary}"

        # Layer 4: Recent messages (verbatim)
        messages = [{"role": "system", "content": system}]
        messages.extend(self.window.messages)
        messages.append({"role": "user", "content": current_query})
        return messages