AI Agent记忆三种实现方式：滑动窗口、摘要与向量存储

我们在 Post #4 中构建的 Agent 有一个大问题——脚本一结束，它就会忘记一切。下次再运行，它从零开始，既不记得之前的对话，也不保留任何事实。

这作为快速实验还可以。但对于任何真正可用的产品而言，这就是致命缺陷。记忆正是让助手随着时间推移逐渐"认识"你的关键。

三种记忆

给 Agent 添加记忆有三种不同的方式，它们各自服务于不同的目的。

短期记忆就是我们一直在用的消息列表。它快速、直接，但只持续当前会话，且受限于 context window。当对话变得过长而无法容纳时，你必须做出选择：丢弃旧消息、压缩它们，或者将它们归档到某处。

长期记忆是模型外部的持久存储——事实、过往对话，任何希望跨会话保留的内容都以 embedding 形式存储在向量数据库中，并在相关时检索出来。这正是 Post #3 中我们对文档做的事，只不过现在应用于记忆本身。

情景记忆是关于过往会话中发生了什么事的结构化记录——不是原始事实，而是事件和上下文。可以把它想象成一本日记："周一，用户问了 X，我们经历了 Y，然后在 Z 处卡住了。"以会话或日期为键，可以按时间顺序方便地查阅。

生产系统通常将三者结合使用。下面的练习中，我们会逐一构建。

当 context 变得过长时

处理不断增长的 context 最简单的方法是滑动窗口——只保留最近 N 条消息，丢弃更早的所有内容。实现简单但也很粗暴：信息一旦滑出窗口就会戛然消失。

更好的方法是 summarisation（摘要）。当历史记录过长时，让模型将其压缩成几句话，保留这个摘要，然后重新开始。会丢失一些细节，但保留了对话的主线。

pip install ollama chromadb

ChromaDB 有两种模式。我们之前一直在用的是内存模式：

# 内存模式 —— 重启后丢失（Post #3 和 Post #4 中使用的方式）
client = chromadb.Client()

# 持久模式 —— 重启后保留（记忆所需的模式）
client = chromadb.PersistentClient(path="./memory_store")

持久客户端只是写入磁盘上的一个文件夹。不需要任何额外的基础设施。

练习 1 —— 滑动窗口

只保留 context 中最近 N 条消息：

import ollama

def chat_with_sliding_window(max_history: int = 6):
    history = []
    system = "You are a helpful assistant."

    print(f"Sliding window chat (last {max_history} messages kept). Type 'quit' to exit.\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() == "quit":
            break

        history.append({"role": "user", "content": user_input})
        windowed = history[-max_history:]

        response = ollama.chat(
            model="llama3.2",
            messages=[{"role": "system", "content": system}] + windowed
        )

        reply = response.message.content
        history.append({"role": "assistant", "content": reply})

        print(f"AI: {reply}")
        print(f"[Sending {len(windowed)} of {len(history)} total messages]\n")

chat_with_sliding_window()

在前几轮对话中告诉它你的名字，然后进行 8 轮以上无关的闲聊，再问"我叫什么名字？"——它会忘记。这就是窗口切断了旧 context 的结果。简单且易于实现，但你能感受到那种硬切。

练习 2 —— 对话摘要

不丢弃旧消息，而是压缩它们：

import ollama

def summarise_history(history: list[dict]) -> str:
    conversation = "\n".join(
        f"{m['role'].upper()}: {m['content']}" for m in history
    )
    response = ollama.chat(
        model="llama3.2",
        options={"temperature": 0},
        messages=[{
            "role": "user",
            "content": f"""Summarise this conversation in 3–4 sentences.
Capture key facts, decisions, and current context.
Write as a neutral summary, not as a participant.

Conversation:
{conversation}

Summary:"""
        }]
    )
    return response.message.content

def chat_with_summarisation(compress_after: int = 8):
    history = []
    summary = ""
    system = "You are a helpful assistant."

    print(f"Summarisation chat (compresses after {compress_after} messages). Type 'quit' to exit.\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() == "quit":
            break

        history.append({"role": "user", "content": user_input})

        if len(history) > compress_after:
            print("\n[Compressing history...]\n")
            summary = summarise_history(history[:-2])
            history = history[-2:]
            print(f"[Summary: {summary[:100]}...]\n")

        messages = [{"role": "system", "content": system}]
        if summary:
            messages.append({
                "role": "system",
                "content": f"Summary of earlier conversation:\n{summary}"
            })
        messages.extend(history)

        response = ollama.chat(model="llama3.2", messages=messages)
        reply = response.message.content
        history.append({"role": "assistant", "content": reply})

        print(f"AI: {reply}")
        print(f"[History: {len(history)} msgs | Summary: {'yes' if summary else 'no'}]\n")

chat_with_summarisation()

运行和练习 1 相同的长时间对话，然后比较。摘要能更好地保留主线——但注意看，压缩过程中具体细节有时会丢失。这就是权衡。

练习 3 —— 长期向量记忆

现在来构建跨脚本重启仍然存在的记忆：

import ollama
import chromadb
import uuid
import datetime

client = chromadb.PersistentClient(path="./memory_store")
memory = client.get_or_create_collection("long_term_memory")

def store_memory(content: str, source: str = "conversation"):
    emb = ollama.embeddings(model="nomic-embed-text", prompt=content).embedding
    memory.add(
        ids=[str(uuid.uuid4())],
        embeddings=[emb],
        documents=[content],
        metadatas=[{"source": source, "timestamp": datetime.datetime.now().isoformat()}]
    )
    print(f"  [Stored: {content[:60]}]")

def retrieve_memories(query: str, top_k: int = 3) -> list[str]:
    if memory.count() == 0:
        return []
    emb = ollama.embeddings(model="nomic-embed-text", prompt=query).embedding
    results = memory.query(
        query_embeddings=[emb],
        n_results=min(top_k, memory.count())
    )
    return results["documents"][0]

def chat_with_memory():
    system = "You are a helpful assistant with access to memories from past conversations."

    print(f"Memory chat. Existing memories: {memory.count()}")
    print("Commands: 'remember: <fact>' | 'recall: <topic>' | 'quit'\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() == "quit":
            break

        if user_input.lower().startswith("remember:"):
            fact = user_input[9:].strip()
            store_memory(fact, source="manual")
            print("AI: Got it, I'll remember that.\n")
            continue

        if user_input.lower().startswith("recall:"):
            topic = user_input[7:].strip()
            memories = retrieve_memories(topic)
            if memories:
                print("AI: Here's what I remember:")
                for m in memories:
                    print(f"  - {m}")
            else:
                print("AI: Nothing stored about that yet.")
            print()
            continue

        relevant = retrieve_memories(user_input)

        messages = [{"role": "system", "content": system}]
        if relevant:
            memory_text = "\n".join(f"- {m}" for m in relevant)
            messages.append({
                "role": "system",
                "content": f"Relevant memories from past conversations:\n{memory_text}"
            })
        messages.append({"role": "user", "content": user_input})

        response = ollama.chat(model="llama3.2", messages=messages)
        reply = response.message.content

        store_memory(f"User said: {user_input}")

        print(f"AI: {reply}")
        print(f"[{len(relevant)} memories retrieved | {memory.count()} total]\n")

chat_with_memory()

告诉它几个事实，退出，再运行一次。它记得。运行后检查 ./memory_store/ 文件夹——这只是磁盘上的文件。每次运行记忆都会增长，而检索是语义化的，所以即使措辞不完全匹配，也能找到相关记忆。

练习 4 —— 具备记忆工具的 Agent

最后，让我们给 Post #4 中的 ReAct Agent 添加存储和检索记忆的能力作为工具：

import ollama
import chromadb
import uuid
import datetime

client = chromadb.PersistentClient(path="./agent_memory")
memory = client.get_or_create_collection("agent_memory")

def store_memory(content: str) -> str:
    emb = ollama.embeddings(model="nomic-embed-text", prompt=content).embedding
    memory.add(
        ids=[str(uuid.uuid4())],
        embeddings=[emb],
        documents=[content],
        metadatas=[{"timestamp": datetime.datetime.now().isoformat()}]
    )
    return "Memory stored."

def recall_memory(query: str) -> str:
    if memory.count() == 0:
        return "No memories stored yet."
    emb = ollama.embeddings(model="nomic-embed-text", prompt=query).embedding
    results = memory.query(query_embeddings=[emb], n_results=min(3, memory.count()))
    docs = results["documents"][0]
    return "\n".join(f"- {d}" for d in docs) if docs else "Nothing relevant found."

def calculate(expression: str) -> str:
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

TOOLS = {"recall_memory": recall_memory, "store_memory": store_memory, "calculate": calculate}

tools = [
    {
        "type": "function",
        "function": {
            "name": "recall_memory",
            "description": "Search past memories for relevant information from previous conversations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "What to search for in memory"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "store_memory",
            "description": "Save an important fact to memory for future conversations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "content": {"type": "string", "description": "The fact to remember"}
                },
                "required": ["content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Perform arithmetic calculations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Python math expression"}
                },
                "required": ["expression"]
            }
        }
    }
]

def run_agent(task: str, max_iterations: int = 8):
    print(f"\nTask: {task}")
    print("=" * 60)
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant with memory tools. "
                "Use recall_memory to check if you have seen relevant information before. "
                "Use store_memory to save important facts for future conversations. "
                "Use calculate for math. Answer clearly when done."
            )
        },
        {"role": "user", "content": task}
    ]
    for i in range(max_iterations):
        response = ollama.chat(model="qwen2.5", messages=messages, tools=tools)
        if response.message.tool_calls:
            messages.append(response.message)
            for tool_call in response.message.tool_calls:
                name = tool_call.function.name
                args = tool_call.function.arguments
                print(f"  → {name}({args})")
                result = TOOLS[name](**args)
                print(f"  ← {str(result)[:120]}")
                messages.append({"role": "tool", "content": str(result)})
        else:
            print(f"\nAnswer: {response.message.content}")
            return
    print("[Max iterations reached]")

# First run
run_agent("My name is Alex and I prefer Python over JavaScript. Remember this.")

# Quit and restart the script, then run this:
run_agent("What do you know about my programming preferences?")

第一次运行后退出，重启脚本，再运行第二个任务。Agent 应该能回忆起在上一会话中存储的内容。注意观察它在开始时是否主动使用 recall_memory——这才是具备记忆能力的 Agent 应有的行为。

记忆将一个无状态工具变成了一个真正"认识你"的东西。滑动窗口最简单，但也最受限。摘要适合处理长对话。持久化向量记忆则是跨多会话场景的必备方案。

实际上，大多数真实系统会结合使用这三种方式——在 context 中保留近期消息，从向量存储中检索长期事实，以及用会话摘要来处理情景上下文。我们现在已经分别构建了每个组件，这意味着你可以根据需要自由组合。

在 Post #6 中，我们会换一个方向——不再让一个 Agent 独自完成所有事情，而是构建多个协同工作的专业 Agent。下次见。🚀