目前国内还是很缺AI人才的，希望更多人能真正加入到AI行业，共同促进行业进步。想要系统学习AI知识的朋友可以看看我的教程http://blog.csdn.net/jiangjunshow，教程通俗易懂，风趣幽默，从深度学习基础原理到各领域实战应用都有讲解。

前言

各位小伙伴，前面咱们知道了：Chroma是本地开发的“轻量小能手”，零运维、开箱即用；而Agent的“短期记忆”，本质就是把用户对话、偏好、历史操作等信息，实时存入向量库，需要时快速检索出来——就像我们记最近发生的事一样，不用记太久，但要随用随取。

今天咱们就手把手实战：用Chroma搭建本地向量库，给Agent加一个“短期记忆模块”，让它能记住用户的口味偏好、对话关键信息，下次互动时不用重复问！

全程代码复制就能跑，核心实现3个功能：

1. 把用户对话、偏好存入Chroma（记忆存储）；
2. Agent回答时，自动检索相关记忆（记忆召回）；
3. 新对话更新记忆，旧记忆自动保留（记忆更新）。

---

一、核心逻辑：Agent的短期记忆怎么工作？

先搞懂原理，再写代码更清晰：

• 记忆内容：用户的问题、Agent的回答、用户偏好（比如“少糖去冰”“喜欢青提味”）、对话关键信息（比如“配送地址XX路”）；
• 存储方式：把每段对话/偏好转成向量，存入Chroma本地库；
• 检索逻辑：用户新提问时，先从Chroma检索“和当前问题相关的历史记忆”，再把记忆+新问题一起喂给LLM，让Agent基于记忆回答；
• 短期特性：记忆存在本地文件（./chroma_memory），关闭程序也不丢，下次启动自动加载，适合“单次会话/短期互动”（比如一次奶茶点单、一场咨询）。

简单说：Chroma = Agent的“短期记事本”，存最近的关键信息，用的时候快速翻到！

---

二、实战步骤：3步搭建Agent短期记忆

准备工作

1. 安装依赖（之前装过的可以跳过）：

pip install chromadb sentence-transformers openai python-dotenv

2. 环境说明：

• 用OpenAI的GPT-3.5/4o做LLM（也可以换国产模型，比如通义千问、智谱清言，代码稍作修改即可）；
• Chroma默认存本地./chroma_memory文件夹，不用额外配置Docker、数据库；
• 嵌入模型用sentence-transformers的all-MiniLM-L6-v2（轻量、本地运行快）。

步骤1：初始化Chroma记忆库（记忆“记事本”）

先写一个基础类，封装Chroma的“增删改查”，方便后续调用：

import chromadb
from sentence_transformers import SentenceTransformer
from dotenv import load_dotenv
import os
from openai import OpenAI

# 加载环境变量（存OpenAI API Key）
load_dotenv()
client_openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"), timeout=30)

class AgentShortTermMemory:
    def __init__(self, memory_name="agent_memory"):
        # 1. 加载嵌入模型（把文字转向量）
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        # 2. 初始化Chroma，本地持久化存储（关闭程序不丢数据）
        self.chroma_client = chromadb.PersistentClient(path="./chroma_memory")
        # 3. 创建/获取记忆集合（相当于“记事本”的本子）
        self.memory_collection = self.chroma_client.get_or_create_collection(
            name=memory_name,
            metadata={"description": "Agent的短期记忆库，存储用户对话、偏好、关键信息"}
        )
        print(f"✅ 短期记忆库初始化成功！存储路径：./chroma_memory")

    def add_memory(self, content: str, metadata: dict = None):
        """
        新增记忆（比如用户说的话、偏好、关键信息）
        :param content: 记忆内容（比如“用户喜欢少糖去冰的奶茶”）
        :param metadata: 记忆元数据（比如{"type": "preference", "user_id": "123"}）
        """
        # 生成向量（文字→向量）
        embedding = self.embedding_model.encode(content).tolist()
        # 生成唯一ID（用时间戳，避免重复）
        memory_id = f"mem_{os.urandom(8).hex()}"  # 随机8位16进制字符串
        # 存入Chroma
        self.memory_collection.add(
            ids=[memory_id],
            documents=[content],
            embeddings=[embedding],
            metadatas=[metadata] if metadata else None
        )
        print(f"📝 新增记忆：{content}")
        return memory_id

    def search_memory(self, query: str, top_k: int = 3):
        """
        检索相关记忆（比如用户新提问时，找相关的历史信息）
        :param query: 检索关键词（比如用户新问题“推荐一款奶茶”）
        :param top_k: 返回最相关的top_k条记忆
        :return: 相关记忆列表
        """
        # 生成查询向量
        query_embedding = self.embedding_model.encode(query).tolist()
        # 相似度检索
        results = self.memory_collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k,
            include_metadata=True
        )
        # 整理结果（只保留有价值的信息）
        relevant_memories = []
        for doc, meta, dist in zip(
            results["documents"][0], results["metadatas"][0], results["distances"][0]
        ):
            # 相似度阈值：低于0.5的记忆认为不相关，过滤掉
            if dist < 0.5:
                relevant_memories.append({
                    "content": doc,
                    "metadata": meta,
                    "similarity": 1 - dist  # 转换为相似度（越高越相关）
                })
        return relevant_memories

    def clear_memory(self):
        """清空所有记忆（慎用！）"""
        self.chroma_client.delete_collection(name=self.memory_collection.name)
        print("🗑️  所有记忆已清空！")

步骤2：给Agent加记忆功能（核心逻辑）

让Agent在回答时，先检索记忆库，再结合记忆+新问题生成答案：

class MemoryAgent:
    def __init__(self):
        # 初始化短期记忆库
        self.short_term_memory = AgentShortTermMemory()

    def get_agent_response(self, user_query: str, user_id: str = "default_user"):
        """
        Agent核心响应逻辑：检索记忆→结合记忆回答→保存新记忆
        :param user_query: 用户新问题
        :param user_id: 用户ID（区分不同用户的记忆）
        :return: Agent回答
        """
        # 1. 检索相关记忆（比如用户之前说过的偏好）
        print("\n🔍 正在检索相关记忆...")
        relevant_memories = self.short_term_memory.search_memory(query=user_query)
        
        # 2. 整理记忆（转成自然语言，喂给LLM）
        if relevant_memories:
            memory_text = "以下是相关的历史记忆：\n"
            for idx, mem in enumerate(relevant_memories, 1):
                memory_text += f"{idx}. {mem['content']}（相似度：{mem['similarity']:.2f}）\n"
            print(f"📖 检索到相关记忆：\n{memory_text}")
        else:
            memory_text = "暂无相关历史记忆。"
            print("📖 未检索到相关记忆")

        # 3. 构建Prompt：结合记忆+新问题
        prompt = f"""
        你是一个奶茶店智能Agent，负责回答用户的奶茶点单、推荐问题。
        请基于以下历史记忆和用户当前提问，生成友好、准确的回答：
        {memory_text}
        用户当前提问：{user_query}
        回答要求：
        1. 如果有相关记忆，一定要用上（比如用户之前说过喜欢少糖，就推荐少糖款）；
        2. 没有相关记忆，就正常回答，同时可以询问用户偏好（比如甜度、冰量）；
        3. 语言口语化，不要太正式。
        """

        # 4. 调用LLM生成回答
        response = client_openai.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.4
        )
        agent_answer = response.choices[0].message.content.strip()

        # 5. 保存新记忆（用户提问+Agent回答，作为下次的历史记忆）
        new_memory_content = f"用户提问：{user_query} | Agent回答：{agent_answer}"
        self.short_term_memory.add_memory(
            content=new_memory_content,
            metadata={"user_id": user_id, "type": "conversation", "timestamp": os.time()}
        )

        return agent_answer

步骤3：测试Agent的短期记忆（实际互动）

运行代码，模拟用户和Agent的多轮对话，看Agent是否能记住偏好：

if __name__ == "__main__":
    # 初始化带记忆的Agent
    agent = MemoryAgent()
    print("\n🤖 奶茶店智能Agent已上线！（输入'退出'结束对话）")

    # 多轮对话
    while True:
        user_input = input("\n👤 你：")
        if user_input.strip() == "退出":
            print("🤖 Agent：再见！下次见～")
            break
        # 获取Agent回答
        agent_answer = agent.get_agent_response(user_query=user_input)
        print(f"🤖 Agent：{agent_answer}")

---

三、运行效果：Agent真的能记住！

咱们模拟3轮对话，看Agent的记忆效果：

第一轮：用户说偏好

🤖 奶茶店智能Agent已上线！（输入'退出'结束对话）

👤 你：我喜欢少糖去冰的奶茶，而且喜欢水果味的
🔍 正在检索相关记忆...
📖 未检索到相关记忆
📝 新增记忆：用户提问：我喜欢少糖去冰的奶茶，而且喜欢水果味的 | Agent回答：好嘞！记下你的偏好啦～ 少糖去冰+水果味，那必须给你推荐我们家的青提茉莉呀！青提果味超浓郁，搭配茉莉茶香，无奶盖更清爽，完全符合你的口味，售价18元，要不要试试呀？
🤖 Agent：好嘞！记下你的偏好啦～ 少糖去冰+水果味，那必须给你推荐我们家的青提茉莉呀！青提果味超浓郁，搭配茉莉茶香，无奶盖更清爽，完全符合你的口味，售价18元，要不要试试呀？

第二轮：用户让推荐（Agent应该记住偏好）

👤 你：再给我推荐一款类似的
🔍 正在检索相关记忆...
📖 检索到相关记忆：
以下是相关的历史记忆：
1. 用户提问：我喜欢少糖去冰的奶茶，而且喜欢水果味的 | Agent回答：好嘞！记下你的偏好啦～ 少糖去冰+水果味，那必须给你推荐我们家的青提茉莉呀！青提果味超浓郁，搭配茉莉茶香，无奶盖更清爽，完全符合你的口味，售价18元，要不要试试呀？（相似度：0.89）

📝 新增记忆：用户提问：再给我推荐一款类似的 | Agent回答：明白！基于你喜欢的少糖去冰+水果味，给你推荐西瓜啵啵～ 新鲜西瓜果肉搭配脆波波，少糖去冰喝起来冰爽解渴，果味超足，和青提茉莉一样清爽，售价19元，这个也很适合你哦！
🤖 Agent：明白！基于你喜欢的少糖去冰+水果味，给你推荐西瓜啵啵～ 新鲜西瓜果肉搭配脆波波，少糖去冰喝起来冰爽解渴，果味超足，和青提茉莉一样清爽，售价19元，这个也很适合你哦！

第三轮：用户问价格（Agent应该记住之前推荐的款式）

👤 你：刚才推荐的两款多少钱呀？
🔍 正在检索相关记忆...
📖 检索到相关记忆：
以下是相关的历史记忆：
1. 用户提问：再给我推荐一款类似的 | Agent回答：明白！基于你喜欢的少糖去冰+水果味，给你推荐西瓜啵啵～ 新鲜西瓜果肉搭配脆波波，少糖去冰喝起来冰爽解渴，果味超足，和青提茉莉一样清爽，售价19元，这个也很适合你哦！（相似度：0.85）
2. 用户提问：我喜欢少糖去冰的奶茶，而且喜欢水果味的 | Agent回答：好嘞！记下你的偏好啦～ 少糖去冰+水果味，那必须给你推荐我们家的青提茉莉呀！青提果味超浓郁，搭配茉莉茶香，无奶盖更清爽，完全符合你的口味，售价18元，要不要试试呀？（相似度：0.78）

📝 新增记忆：用户提问：刚才推荐的两款多少钱呀？ | Agent回答：刚才给你推荐的两款都是少糖去冰的水果味奶茶哦～ 青提茉莉售价18元，西瓜啵啵售价19元，价格都很划算，要不要一起点呀？
🤖 Agent：刚才给你推荐的两款都是少糖去冰的水果味奶茶哦～ 青提茉莉售价18元，西瓜啵啵售价19元，价格都很划算，要不要一起点呀？

效果总结：

• Agent记住了用户的核心偏好（少糖去冰、水果味）；
• 后续推荐、回答价格时，自动用上了历史记忆；
• 记忆存在本地./chroma_memory，下次启动Agent，依然能检索到这些记忆！

---

四、关键优化：让记忆更实用

1. 记忆过滤（避免冗余）

如果用户重复说同样的话（比如反复说“少糖去冰”），可以在add_memory时先检索，避免重复存储：

def add_memory(self, content: str, metadata: dict = None):
    # 先检索是否已有相似记忆（相似度>0.9）
    similar_memories = self.search_memory(query=content, top_k=1)
    if similar_memories and similar_memories[0]["similarity"] > 0.9:
        print(f"⚠️  已有相似记忆，无需重复存储：{similar_memories[0]['content']}")
        return None
    # 没有相似记忆，再存入
    # ... 之前的存入逻辑 ...

2. 记忆过期（真正“短期”）

短期记忆不需要永久保存，可以加过期时间（比如24小时），检索时过滤过期记忆：

def search_memory(self, query: str, top_k: int = 3, expire_hours: int = 24):
    query_embedding = self.embedding_model.encode(query).tolist()
    results = self.memory_collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include_metadata=True
    )
    relevant_memories = []
    current_time = os.time()
    for doc, meta, dist in zip(
        results["documents"][0], results["metadatas"][0], results["distances"][0]
    ):
        # 过滤过期记忆（如果有timestamp字段）
        if meta and "timestamp" in meta:
            memory_time = meta["timestamp"]
            if (current_time - memory_time) > expire_hours * 3600:
                print(f"⏳ 过滤过期记忆：{doc}")
                continue
        if dist < 0.5:
            relevant_memories.append({...})  # 之前的整理逻辑
    return relevant_memories

3. 区分用户（多用户支持）

如果多个用户使用Agent，可以在检索时按user_id过滤，避免记忆混淆：

def search_memory(self, query: str, user_id: str, top_k: int = 3):
    query_embedding = self.embedding_model.encode(query).tolist()
    # 按user_id过滤（metadata筛选）
    results = self.memory_collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include_metadata=True,
        where={"user_id": user_id}  # 只检索当前用户的记忆
    )
    # ... 后续整理逻辑 ...

---

五、常见问题：避坑指南

1. 运行报错：OpenAI API Key未配置？

• 解决方案：创建.env文件，写入OPENAI_API_KEY="你的API Key"，放在代码同级目录；
• 替代方案：换国产模型，比如通义千问，修改LLM调用部分代码（参考通义千问Python SDK）。

2. 记忆检索不到？

• 检查相似度阈值：代码中设置的是dist<0.5（相似度>0.5），如果阈值太高（比如dist<0.3），可能检索不到，可适当调低；
• 检查嵌入模型：确保嵌入模型和检索时用的是同一个（都是all-MiniLM-L6-v2），维度一致（384维）。

3. 记忆太多，检索变慢？

• 清理过期记忆：用前面的“记忆过期”功能，自动删除老记忆；
• 限制记忆数量：在add_memory时，检查集合大小，超过100条就删除最旧的（用timestamp排序）。

---

六、总结：Chroma实现短期记忆的核心价值

• 零成本落地：不用部署复杂数据库，pip安装+几行代码，就能给Agent加记忆；
• 实时响应：本地向量库检索毫秒级，不影响Agent回答速度；
• 灵活扩展：支持过滤、过期、多用户，满足不同场景需求；
• 无缝衔接：后续可以升级到Pinecone（云端）或Milvus（大规模），记忆数据迁移方便。

对Agent开发来说，短期记忆是“入门级核心功能”——有了记忆，Agent才不是“一次性对话机器人”，而是能持续理解用户、优化互动的智能助手。

---

目前国内还是很缺AI人才的，希望更多人能真正加入到AI行业，共同促进行业进步。想要系统学习AI知识的朋友可以看看我的教程http://blog.csdn.net/jiangjunshow，教程通俗易懂，风趣幽默，从深度学习基础原理到各领域实战应用都有讲解。