---

一、前言：我的Agent，好像得了“阿尔茨海默症”

兄弟们，你们有没有过这种崩溃时刻？

花了三天三夜，写了一个号称“智能对话”的AI Agent。第一次对话，它还能清晰记住我是谁、我要做什么、甚至我上一句随口提的“我想给客户A做一个季度销售分析”。结果聊到第三轮，我问它“那客户A的分析报告里，我们重点要突出什么数据？”

它愣了三秒，然后回了一句：“抱歉，我没有收到过关于‘客户A’的相关信息，请您重新说明。”

我：？？？

我看着后台的日志，陷入了沉思。我花那么多钱接的大模型，怎么转头就把我忘了？这哪里是智能助手，这分明是得了阿尔茨海默症的金鱼，记忆只有7个Token。

作为一个后端开发，我不能忍。于是我决定，亲手给我的Agent装个“大脑外挂”——搞定上下文管理，让它再也不会在多轮对话里“失忆”。这篇文章，就是我踩过的坑、趟过的雷，以及最终跑通的完整方案，全是干货，还有可直接跑的代码。

---

二、罪魁祸首：为什么你的Agent会“失忆”？

在动手之前，我们得先搞明白，Agent为什么会忘事？

很多新手（包括曾经的我）以为，大模型是个无限内存的U盘，我说过的每一句话它都记得。但事实是，大模型的“记忆”，完全依赖于你每次发请求时，塞进去的messages列表。

举个例子：

• 第一轮请求：messages = [{"role": "user", "content": "我是你老板，你要记住我，我叫老王"}]
• 模型回复：{"role": "assistant", "content": "好的老王老板，我记住了！"}
• 第二轮请求：如果你只发messages = [{"role": "user", "content": "我是谁？"}]，模型根本不知道你是老王，自然会答错。

所以，最基础的“记忆”方案，就是把所有历史对话，都拼到messages里，一起发给模型。

但这又带来了新问题：Token限制。

大模型的上下文窗口是有限的，比如GPT-3.5是4k/16k，GPT-4是8k/32k，哪怕是国产的大模型，也有token上限。每一个汉字、每一个标点，甚至空格，都在消耗Token。对话多了，历史消息越长，Token就越容易超标，然后就会触发：

• 报错：context window is full
• 模型自动截断，直接忘掉前面的对话
• 成本飙升：Token越多，API调用越贵

所以，我们的目标不是简单地“把对话全塞进去”，而是在Token限制内，高效地管理对话上下文，让Agent既能记住关键信息，又不爆内存、不超预算。

而实现这个目标，我们需要三件套：

1. Token计数：精准监控你的上下文，别超了还不知道
2. 上下文窗口管理：把没用的信息丢掉，留下有用的
3. 记忆链（Memory Chain）：把对话里的关键信息抽出来，存成结构化数据，让Agent永远不忘

接下来，我们一个一个来搞定。

---

三、第一步：先搞懂Token，别再当“盲盒玩家”

很多人对Token的理解就是“大概是个字数单位”，但差远了。不同语言的Token消耗不一样，中文1个Token大概对应1-2个汉字，英文1个Token大概对应0.75个单词。而且，大模型的输入和输出，都要算Token。

所以，第一步，我们必须给Agent加上Token计数器，实时监控上下文的Token消耗，这是所有上下文管理的基础。

3.1 用OpenAI官方工具精准计数

如果你用的是OpenAI系列模型，官方提供了tiktoken库，可以精准计算Token数量。

import tiktoken

# 选择对应模型的编码器
def count_tokens(text: str, model: str = "gpt-3.5-turbo-16k") -> int:
    """计算文本的token数量"""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        # 如果模型不在内置列表里，用cl100k_base（大部分OpenAI模型都用这个）
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

# 测试一下
if __name__ == "__main__":
    test_text = "Token、上下文管理、记忆链：我是如何让AI Agent在多轮对话里不“失忆”的？"
    print(f"文本：{test_text}")
    print(f"Token数：{count_tokens(test_text)}")

运行一下，你会发现，一句话的Token数和字数完全不一样。

3.2 非OpenAI模型怎么办？

如果你用的是国产大模型，比如文心一言、通义千问，没有官方的Token计数器怎么办？

没关系，我们可以用一个通用的估算方法，或者直接调用模型的API自带的Token统计接口。这里给一个通用的估算函数，误差不大，足够我们做上下文管理了：

def estimate_tokens(text: str) -> int:
    """通用Token估算，中文按1.5字/Token，英文按0.75词/Token"""
    # 中文部分按字数估算
    chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
    # 英文部分按单词数估算
    english_words = len([word for word in text.split() if word.isascii()])
    return int(chinese_chars / 1.5 + english_words / 0.75)

# 测试
print(estimate_tokens("我是老王，我想让你帮我分析客户A的季度销售数据"))

有了Token计数，我们就可以给上下文设置一个“安全水位线”，比如模型上限是16k，我们就设14k为安全线，超过了就开始裁剪。

---

四、第二步：上下文窗口管理，把没用的对话“丢进垃圾桶”

最粗暴的上下文管理，就是“直接截断”，把最早的对话删掉，只留最新的N轮对话。但这种方法有个问题：如果关键信息在很早的对话里，直接截断就丢了。

所以，我们需要一个更智能的窗口管理策略，这里给大家实现两种常用方案：

4.1 方案一：滑动窗口（Sliding Window）——简单粗暴但有效

滑动窗口的逻辑很简单：设置一个最大对话轮数，比如10轮，当对话超过10轮时，就把最早的1轮删掉，保持窗口内永远只有最新的10轮对话。

from typing import List, Dict

class SlidingWindowManager:
    def __init__(self, max_rounds: int = 10, max_tokens: int = 14000, model: str = "gpt-3.5-turbo-16k"):
        self.max_rounds = max_rounds
        self.max_tokens = max_tokens
        self.model = model
        self.messages: List[Dict] = []
    
    def add_message(self, role: str, content: str):
        """添加新消息，同时管理窗口"""
        self.messages.append({"role": role, "content": content})
        # 先按轮数裁剪
        if len(self.messages) > self.max_rounds * 2:  # 每轮对话包含user和assistant两条
            # 删掉最早的一轮（两条消息）
            self.messages = self.messages[2:]
        # 再按Token裁剪，确保不超过上限
        self._trim_by_tokens()
    
    def _trim_by_tokens(self):
        """按Token数裁剪消息，从最早的开始删"""
        while True:
            # 计算当前所有消息的总Token数
            total_tokens = sum(
                count_tokens(msg["content"], self.model) 
                for msg in self.messages
            )
            if total_tokens <= self.max_tokens or len(self.messages) <= 2:
                # 至少保留系统提示和最后一轮对话
                break
            # 删掉最早的非系统消息
            if self.messages[0]["role"] == "system":
                # 系统提示不能删，删下一条
                del self.messages[1]
            else:
                del self.messages[0]
    
    def get_messages(self) -> List[Dict]:
        """获取最终要发给模型的消息列表"""
        return self.messages

# 使用示例
if __name__ == "__main__":
    manager = SlidingWindowManager(max_rounds=5, max_tokens=1000)
    # 添加系统提示
    manager.add_message("system", "你是一个专业的销售分析助手，要记住用户的所有关键信息")
    # 模拟多轮对话
    manager.add_message("user", "我是老王，客户A的季度销售额是100万")
    manager.add_message("assistant", "好的，我记住了，客户A的季度销售额是100万")
    manager.add_message("user", "客户B的销售额是200万")
    manager.add_message("assistant", "收到，客户B销售额200万")
    manager.add_message("user", "帮我对比一下两个客户的销售数据")
    print("当前消息列表：", manager.get_messages())

滑动窗口的优点是实现简单、性能高，缺点是会丢失早期的对话信息。适合对话轮数不多、不需要长期记忆的场景。

4.2 方案二：基于Token的动态裁剪——精准控制内存

滑动窗口是按轮数裁剪，而动态裁剪是直接按Token数裁剪，更精准。我们刚才的代码里已经加了_trim_by_tokens方法，就是干这个的。

这里有个小技巧：系统提示词（system prompt）尽量放在最前面，并且不要轻易裁剪，因为它决定了Agent的角色和行为逻辑，丢了就彻底跑偏了。

---

五、第三步：记忆链（Memory Chain）——给Agent装个“长期记忆”

滑动窗口只能解决短期对话不爆Token的问题，但还是会丢早期的关键信息。比如用户第一轮说“我是老王，我要分析客户A的数据”，对话了20轮之后，滑动窗口把第一轮删掉了，模型还是会忘。

这时候，我们就需要记忆链（Memory Chain），把对话里的关键信息，抽出来存成结构化数据，比如用户是谁、用户的需求是什么、客户信息是什么，不管对话多少轮，这些关键信息永远存在。

5.1 记忆链的核心：结构化记忆存储

我们先定义一个结构化的记忆模型，用来存对话里的关键信息：

from dataclasses import dataclass, asdict
from typing import Optional, List, Dict
import json

@dataclass
class UserInfo:
    """用户基础信息"""
    name: Optional[str] = None  # 用户名
    role: Optional[str] = None  # 用户角色（比如老板、销售）
    preferences: List[str] = None  # 用户偏好

@dataclass
class ClientInfo:
    """客户信息"""
    client_id: str
    name: Optional[str] = None
    sales_data: Optional[float] = None
    key_requirements: List[str] = None

@dataclass
class ConversationMemory:
    """对话记忆"""
    user_info: UserInfo = None
    clients: Dict[str, ClientInfo] = None  # key是client_id
    current_task: Optional[str] = None  # 当前用户的核心任务
    key_entities: List[str] = None  # 对话中提到的关键实体

    def __post_init__(self):
        if self.user_info is None:
            self.user_info = UserInfo()
        if self.clients is None:
            self.clients = {}
        if self.key_entities is None:
            self.key_entities = []

    def to_dict(self):
        """转成字典，方便发给模型"""
        return asdict(self)
    
    def to_json(self):
        """转成JSON字符串"""
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)

5.2 关键：用大模型做记忆抽取

我们需要在每一轮对话之后，调用一次模型，把对话里的关键信息，更新到我们的ConversationMemory里。

这里的核心是写一个结构化输出的Prompt，让模型按我们定义的格式，输出更新后的记忆数据。

def extract_memory_from_conversation(
    messages: List[Dict], 
    current_memory: ConversationMemory,
    llm_call_func  # 这里传入你的大模型调用函数
) -> ConversationMemory:
    """从对话中抽取关键信息，更新记忆"""
    # 构建Prompt，要求模型按JSON格式输出
    prompt = f"""
    你是一个对话记忆抽取助手，需要根据用户和助手的对话历史，更新用户的记忆信息。
    请严格按照给定的JSON格式输出，不要添加任何额外解释，只返回JSON。

    当前已有的记忆信息：
    {current_memory.to_json()}

    最新的对话历史：
    {json.dumps(messages[-4:], ensure_ascii=False, indent=2)}  # 取最新的几轮对话，减少Token消耗

    请根据对话，更新以下信息：
    1. user_info: 用户的姓名、角色、偏好
    2. clients: 对话中提到的客户信息（名称、销售额、需求）
    3. current_task: 用户当前的核心任务
    4. key_entities: 对话中提到的关键实体（比如客户名、产品名）

    输出格式（只返回JSON，不要其他内容）：
    {{
        "user_info": {{
            "name": "用户姓名，没有则保持原有值",
            "role": "用户角色，没有则保持原有值",
            "preferences": ["用户偏好，列表形式"]
        }},
        "clients": {{
            "client_id_1": {{
                "name": "客户名称",
                "sales_data": 1000000,
                "key_requirements": ["客户需求"]
            }}
        }},
        "current_task": "当前核心任务，没有则保持原有值",
        "key_entities": ["关键实体列表"]
    }}
    """

    # 调用大模型，获取结构化输出
    response = llm_call_func(prompt)
    # 解析JSON
    try:
        updated_memory_dict = json.loads(response)
        # 用新的信息更新原有记忆
        updated_memory = ConversationMemory(
            user_info=UserInfo(**updated_memory_dict["user_info"]),
            clients={k: ClientInfo(client_id=k, **v) for k, v in updated_memory_dict["clients"].items()},
            current_task=updated_memory_dict["current_task"],
            key_entities=updated_memory_dict["key_entities"]
        )
        return updated_memory
    except Exception as e:
        print(f"记忆抽取失败，使用原有记忆：{e}")
        return current_memory

这里有个关键点：每次只取最新的几轮对话（比如最后4轮）来做记忆抽取，这样可以大幅减少Token消耗，而且早期的关键信息已经被存在记忆里了，不需要重复抽取。

5.3 把记忆链注入到上下文里

有了结构化的记忆之后，我们就可以把它拼到每次发给模型的消息列表里，这样不管对话多少轮，模型都能看到用户的关键信息了。

我们改造一下之前的SlidingWindowManager，加上记忆链的支持：

class MemoryEnhancedManager(SlidingWindowManager):
    def __init__(self, max_rounds: int = 10, max_tokens: int = 14000, model: str = "gpt-3.5-turbo-16k"):
        super().__init__(max_rounds, max_tokens, model)
        self.memory = ConversationMemory()
        # 记忆在消息列表中的位置，默认在系统提示之后
        self.memory_index = 1

    def add_message(self, role: str, content: str, llm_call_func=None):
        """添加消息，同时更新记忆（如果提供了llm_call_func）"""
        super().add_message(role, content)
        # 如果提供了LLM调用函数，就更新记忆
        if llm_call_func and role == "assistant":
            # 对话完成后，更新记忆
            self.memory = extract_memory_from_conversation(
                self.messages, self.memory, llm_call_func
            )
            # 把记忆拼接到消息列表里
            self._inject_memory_to_messages()

    def _inject_memory_to_messages(self):
        """把结构化记忆注入到消息列表中"""
        # 先把原来的记忆消息删掉（如果有的话）
        if len(self.messages) > self.memory_index and self.messages[self.memory_index]["role"] == "system" and "对话记忆信息" in self.messages[self.memory_index]["content"]:
            del self.messages[self.memory_index]
        # 生成记忆消息
        memory_content = f"""
        【对话记忆信息】
        以下是本次对话的关键信息，请你务必记住并在回答中使用：
        {self.memory.to_json()}
        """
        # 插入到系统提示之后
        self.messages.insert(self.memory_index, {"role": "system", "content": memory_content})
        # 再做一次Token裁剪，防止超了
        self._trim_by_tokens()

# 使用示例
if __name__ == "__main__":
    # 模拟一个简单的LLM调用函数（实际项目中换成你的模型调用）
    def mock_llm_call(prompt):
        # 这里只是模拟，实际需要调用OpenAI或其他大模型的API
        return """
        {
            "user_info": {"name": "老王", "role": "销售经理", "preferences": ["数据对比用柱状图", "结果要简洁明了"]},
            "clients": {
                "clientA": {"name": "客户A", "sales_data": 1000000, "key_requirements": ["季度销售分析", "同比增长率"]},
                "clientB": {"name": "客户B", "sales_data": 2000000, "key_requirements": ["季度销售分析", "客户画像分析"]}
            },
            "current_task": "对比客户A和客户B的季度销售数据，输出分析报告",
            "key_entities": ["客户A", "客户B", "季度销售额", "销售分析"]
        }
        """
    
    manager = MemoryEnhancedManager(max_rounds=5, max_tokens=10000)
    # 添加系统提示
    manager.add_message("system", "你是一个专业的销售分析助手，要基于对话记忆信息回答用户问题")
    # 模拟对话
    manager.add_message("user", "我是老王，客户A的季度销售额是100万，客户B是200万")
    manager.add_message("assistant", "好的，我记住了客户A和B的销售额")
    # 对话完成后，更新记忆
    manager.add_message("user", "帮我对比一下两个客户的销售数据")
    manager.add_message("assistant", "好的，我来帮你分析", llm_call_func=mock_llm_call)

    print("最终消息列表：")
    for msg in manager.get_messages():
        print(f"{msg['role']}: {msg['content'][:100]}...")

这样，不管滑动窗口怎么裁剪早期的对话，【对话记忆信息】里的关键数据永远都在，模型再也不会忘了你是谁、客户是谁了。

---

六、进阶优化：让记忆链更稳定、更高效

到这里，我们的基础方案已经跑通了，但在实际业务中，还有几个坑要填，不然你的Agent还是会“间歇性失忆”。

6.1 优化1：结构化输出失败？用JSON Schema兜底

有时候，大模型会“抽风”，输出的JSON格式不对，导致记忆抽取失败。这时候，我们可以用JSON Schema来约束模型的输出，比如用pydantic来定义模型，再用LangChain的with_structured_output功能。

from pydantic import BaseModel, Field
from typing import List, Dict, Optional

# 用pydantic定义记忆的Schema
class UserInfoSchema(BaseModel):
    name: Optional[str] = Field(None, description="用户姓名，没有则不修改")
    role: Optional[str] = Field(None, description="用户角色，比如销售经理")
    preferences: List[str] = Field(default_factory=list, description="用户的偏好设置")

class ClientInfoSchema(BaseModel):
    name: Optional[str] = Field(None, description="客户名称")
    sales_data: Optional[float] = Field(None, description="客户销售额")
    key_requirements: List[str] = Field(default_factory=list, description="客户的关键需求")

class ConversationMemorySchema(BaseModel):
    user_info: UserInfoSchema
    clients: Dict[str, ClientInfoSchema] = Field(default_factory=dict)
    current_task: Optional[str] = Field(None, description="用户当前的核心任务")
    key_entities: List[str] = Field(default_factory=list, description="对话中提到的关键实体")

# 在LangChain中使用结构化输出
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0)
structured_llm = llm.with_structured_output(ConversationMemorySchema)

def extract_memory_with_schema(messages: List[Dict], current_memory: ConversationMemory) -> ConversationMemorySchema:
    prompt = f"""
    你是对话记忆抽取助手，请根据对话更新记忆信息，严格按照要求的格式输出。
    当前记忆：{current_memory.to_json()}
    最新对话：{json.dumps(messages[-4:], ensure_ascii=False)}
    """
    response = structured_llm.invoke([
        SystemMessage(content=prompt)
    ])
    return response

用了pydantic的JSON Schema约束，模型输出的JSON格式基本不会出错，记忆抽取的稳定性大大提升。

6.2 优化2：记忆合并，避免信息冲突

当对话多了，不同轮次抽取的记忆可能会冲突，比如用户第一次说“我是老王”，第二次又说“我是小王”，这时候模型不知道该留哪个。

我们可以给记忆加一个更新策略：

• 优先保留最新的信息，覆盖旧信息
• 列表类型的信息（比如用户偏好、关键实体），不覆盖，而是合并去重
• 数值类型的信息（比如销售额），如果有新值，直接覆盖旧值

def merge_memory(old_memory: ConversationMemory, new_memory: ConversationMemorySchema) -> ConversationMemory:
    """合并新旧记忆，解决信息冲突"""
    # 用户信息：用新值覆盖旧值，None则保留旧值
    if new_memory.user_info.name is not None:
        old_memory.user_info.name = new_memory.user_info.name
    if new_memory.user_info.role is not None:
        old_memory.user_info.role = new_memory.user_info.role
    # 偏好列表：合并去重
    old_memory.user_info.preferences = list(set(old_memory.user_info.preferences + new_memory.user_info.preferences))

    # 客户信息：更新已有客户，新增新客户
    for client_id, new_client in new_memory.clients.items():
        if client_id in old_memory.clients:
            old_client = old_memory.clients[client_id]
            if new_client.name is not None:
                old_client.name = new_client.name
            if new_client.sales_data is not None:
                old_client.sales_data = new_client.sales_data
            old_client.key_requirements = list(set(old_client.key_requirements + new_client.key_requirements))
        else:
            old_memory.clients[client_id] = ClientInfo(
                client_id=client_id,
                name=new_client.name,
                sales_data=new_client.sales_data,
                key_requirements=new_client.key_requirements
            )
    
    # 当前任务：优先用新值
    if new_memory.current_task is not None:
        old_memory.current_task = new_memory.current_task
    # 关键实体：合并去重
    old_memory.key_entities = list(set(old_memory.key_entities + new_memory.key_entities))

    return old_memory

6.3 优化3：给记忆加个“索引”，用RAG快速召回

当对话轮数非常多（比如几百轮），哪怕是结构化记忆，塞到上下文里也会占不少Token。这时候，我们可以把历史对话向量存储起来，用RAG（检索增强生成）技术，根据当前用户的问题，召回相关的历史对话片段，动态注入到上下文里。

# 伪代码，展示核心逻辑
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 初始化向量数据库
embeddings = OpenAIEmbeddings()
vector_store = FAISS.from_texts([], embedding=embeddings)

def add_to_vector_store(role: str, content: str, turn_id: int):
    """把每轮对话存到向量数据库"""
    doc = f"轮次{turn_id} {role}: {content}"
    vector_store.add_texts([doc], metadatas=[{"turn_id": turn_id}])

def retrieve_relevant_memory(query: str, top_k: int = 3) -> List[str]:
    """根据用户当前的问题，召回相关的历史对话"""
    docs = vector_store.similarity_search(query, k=top_k)
    return [doc.page_content for doc in docs]

# 在生成上下文时，把召回的对话拼进去
relevant_dialogs = retrieve_relevant_memory("帮我对比客户A和B的销售数据")
context = f"""
【相关历史对话】
{chr(10).join(relevant_dialogs)}
【对话记忆信息】
{memory.to_json()}
"""

RAG的好处是，只把和当前问题相关的历史对话召回，不用把所有历史对话都塞进去，Token消耗大大降低，而且模型也能拿到需要的上下文。

---

七、踩坑实录：这些问题我替你踩过了

最后，给大家分享几个我在实现过程中踩过的坑，帮你少走弯路：

1. 坑1：Token计数不准，导致超了上下文窗口

   • 解决：不要用通用的字数估算，一定要用对应模型的官方Token计数器，比如OpenAI的tiktoken，不然很容易算少了，导致API报错。

2. 坑2：记忆抽取Prompt写得不好，模型乱输出

   • 解决：结构化输出的Prompt一定要写清楚格式，加上例子，并且用JSON Schema约束，不要让模型自由发挥。

3. 坑3：记忆更新太频繁，导致Token消耗过高

   • 解决：不要每一轮对话都做记忆抽取，可以设置一个阈值，比如每3轮对话抽取一次，或者对话轮数超过10轮再抽取，减少API调用次数。

4. 坑4：把所有记忆都塞到上下文里，Token还是不够用

   • 解决：用“分层记忆”的思路，关键信息存在结构化记忆里，次要的历史对话用RAG召回，不要全塞进去。

---

八、总结：终于，我的Agent不再失忆了

经过这一套组合拳：Token计数 + 滑动窗口管理 + 结构化记忆链 + RAG召回，我的Agent终于治好了“阿尔茨海默症”。不管对话多少轮，它都能记住我是老王、记住客户A和B的销售额，甚至还记得我之前提过“对比数据要用柱状图”。

其实，上下文管理的核心，从来都不是“把所有对话都存下来”，而是在有限的Token里，把模型最需要的信息，以最高效的方式喂给它。滑动窗口解决短期对话不爆内存，结构化记忆解决关键信息不丢失，RAG解决海量历史对话的召回问题，三者结合，才能让Agent真正“有脑子”。

希望这篇文章，能帮你少踩我踩过的坑，写出不“失忆”的AI Agent。如果你有更好的方案，也欢迎在评论区交流~

---