【从0搭建AI智能体·2】给智能体装上「永久记忆」：RAG 向量数据库实战（附完整可运行代码）

📚 本文是《从 0 搭建你的 AI 智能体》专栏第 2 篇。
上一篇：《AgentKey：密钥管理与基础调用》

标签：RAG 向量数据库 Embedding 大模型 Python 智能体 LLM

📌 前言：为什么截断和摘要都救不了你的智能体

在上一篇《AgentKey 智能体开发完全指南》里，我们聊了三种对话记忆管理法：滑动窗口、Token 截断、摘要压缩。它们看着够用，但一上生产就露馅——信息是「线性丢失」的：

• 滑动窗口：聊到第 100 轮，第 1 轮用户说的「我对花生过敏」早被挤出窗口了。之后 AI 推荐了花生酱饼干，出了事算谁的？
• 摘要压缩：反复摘要像反复压缩 JPG，越压越糊，关键数字、专有名词一次次被「抹平」；
• 二者本质都是**「按时间顺序」取舍**——离得近的留下，离得远的丢掉。可现实是，最该被记住的信息，往往发生在很久以前。

我们真正要的是：用户现在问什么，就精准捞出历史里跟这个问题最相关的那几段，哪怕它发生在 500 轮之前。这就是 RAG（Retrieval-Augmented Generation，检索增强生成）——智能体「无限记忆」的终极方案，也是当下企业知识库、智能客服的事实标准架构。

💡 本文适合谁：能跑通大模型基础调用、想给 Agent / 知识库加长期记忆的开发者。全文代码可直接跑通，文末附一个完整整合脚本，复制即用。阅读约 15 分钟。

目录

1. RAG 到底是什么：用「开卷考试」秒懂
2. 一张图看懂 RAG 完整链路
3. 核心三要素：Embedding、向量库、相似度
4. 环境准备与依赖安装
5. 第一步：把文本变成向量（Embedding 实战）
6. 第二步：用 ChromaDB 存储与检索
7. 第三步：拼装完整 RAG 问答链路
8. 第四步：加上「相似度阈值」，拒绝答非所问
9. 把 RAG 用作对话「长期记忆」（含多用户隔离）
10. 检索质量优化：分块、Top-K、重排（附代码）
11. 常见坑与生产建议
12. FAQ：读者最常问的 8 个问题
13. 完整可运行 Demo
14. 总结与延伸

---

① RAG 到底是什么：用「开卷考试」秒懂

把大模型想象成一个知识渊博但记性不好、还容易吹牛的考生。

• 闭卷考试（裸调模型）：全凭脑子里的记忆答题，遇到不会的就编一个（这就是「幻觉」）；
• 开卷考试（RAG）：答题前先翻书，找到相关那几页，照着资料组织答案。不会的直接说「书上没有」。

RAG 就是给大模型开卷。大模型负责「理解和表达」，向量数据库负责「记忆和检索」，各司其职。这样即使知识库有几百万条，每次也只把最相关的几条「翻」给模型看。

对比维度	塞进上下文（长 Prompt）	RAG 检索
知识规模	受上下文长度限制（几万 Token 封顶）	理论无上限（百万级）
单次成本	每次都重发全部，越聊越贵	只发相关片段，成本恒定
更新知识	要重构整个 Prompt	往库里加/删一条即可
精准度	无关信息干扰，容易带偏	按语义相关性精准召回
可追溯	答案来源不明	能定位到具体来源片段

---

② 一张图看懂 RAG 完整链路

RAG 分离线建库和在线问答两个阶段，一张图讲清：

【离线：建立知识库】
   原始文档
      │  切块（Chunking）
      ▼
   文本块 ──Embedding──► 向量  ──存入──►  📦 向量数据库
                                          (向量 + 原文 + 元数据)

【在线：一次问答】
   用户提问
      │  Embedding
      ▼
   问题向量 ──相似度检索──►  📦 向量数据库
                              │  返回 Top-K 相关片段
                              ▼
   [相关片段] + [用户问题]  ──拼进 Prompt──►  🤖 大模型
                                                │
                                                ▼
                                          有据可查的回答

记住这张图，后面所有代码都是在实现它的某一环。

---

③ 核心三要素：Embedding、向量库、相似度

• Embedding（嵌入向量）：把一段文本转成一串数字（如 1536 维浮点数组）。语义越接近的文本，向量在空间里距离越近。「猫」和「小猫」很近，「猫」和「金融」很远。
• 向量数据库：专门存这些向量、并能高速「找出最相近的 K 个」的数据库。ChromaDB、FAISS、Milvus、Qdrant、Pinecone 都是。
• 相似度 / 距离：衡量两个向量有多接近。常用余弦相似度（越接近 1 越相关）；ChromaDB 默认返回的是距离（越接近 0 越相关）——这个方向别搞反了，第 ⑧ 节会用到。

一句话串起来：知识切块 → 转向量存库 → 提问也转向量 → 库里找最近的几块 → 交给大模型作答。

---

④ 环境准备与依赖安装

我们用 ChromaDB（轻量、本地零配置，最适合入门和中小项目）：

python -m venv rag_env
source rag_env/bin/activate      # Windows: rag_env\Scripts\activate

pip install chromadb openai python-dotenv

依赖	作用
chromadb	本地向量数据库，负责存储与检索
openai	调用 Embedding + Chat 接口（兼容 OpenAI 协议的服务都可用）
python-dotenv	加载 .env 里的密钥

.env 配置（沿用上一篇的安全规范，密钥绝不硬编码）：

AGENT_KEY=sk-your-actual-key-here
API_BASE_URL=https://api.example.com/v1

💡 本文用 OpenAI 兼容 SDK 演示。换别家服务时，只需改 base_url 和模型名，逻辑完全一致。

---

⑤ 第一步：把文本变成向量（Embedding 实战）

先封装「文本转向量」函数，这是 RAG 的地基：

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI(api_key=os.getenv("AGENT_KEY"), base_url=os.getenv("API_BASE_URL"))
EMBED_MODEL = "text-embedding-3-small"  # 换成你的服务支持的向量模型


def embed(texts):
    """把字符串或字符串列表转成向量列表。支持批量以省钱省时。"""
    if isinstance(texts, str):
        texts = [texts]
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return [item.embedding for item in resp.data]


# 快速验证：语义相近 → 相似度更高
if __name__ == "__main__":
    import math

    def cosine(a, b):
        dot = sum(x * y for x, y in zip(a, b))
        na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
        return dot / (na * nb)

    vecs = embed(["我养了一只猫", "我家有只小猫咪", "今天股市大跌"])
    print("向量维度:", len(vecs[0]))
    print("猫 vs 小猫:", round(cosine(vecs[0], vecs[1]), 3))   # 接近 1
    print("猫 vs 股市:", round(cosine(vecs[0], vecs[2]), 3))   # 明显更低

跑一下，「猫 vs 小猫」的相似度会明显高于「猫 vs 股市」——这就是语义检索的底层魔法。

⚠️ 避坑：Embedding 接口支持批量输入（一次传整个列表），千万别在 for 循环里一条条调用——既慢又极易触发限流。上面的 embed 已支持批量。

---

⑥ 第二步：用 ChromaDB 存储与检索

把向量存进 ChromaDB，并实现「按问题召回最相关的 K 条」。注意这里我给每条数据挂了元数据（metadata），这是生产 RAG 的关键，后面多用户隔离要靠它：

import chromadb
import hashlib

# 持久化到本地磁盘（生产可换成 Server/HttpClient 模式）
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(name="my_knowledge")


def doc_id(text):
    """内容稳定哈希做 ID。⚠️ 别用内置 hash()——它每次进程启动会加随机盐，
    重启后同一内容得到不同 ID，去重就失效了。md5 才是跨进程稳定的。"""
    return hashlib.md5(text.encode("utf-8")).hexdigest()[:16]


def add_documents(docs, source="manual"):
    """写入向量库。docs: List[str]"""
    vectors = embed(docs)
    collection.add(
        # 用内容哈希做 ID，天然去重、重复写入不会产生垃圾副本
        ids=[f"doc_{doc_id(d)}" for d in docs],
        embeddings=vectors,
        documents=docs,                                  # 原文一并存下
        metadatas=[{"source": source} for _ in docs],    # 挂元数据，便于过滤
    )


def search(query, top_k=3):
    """检索与 query 最相关的 top_k 条，返回 [(原文, 距离), ...]"""
    query_vec = embed(query)[0]
    res = collection.query(query_embeddings=[query_vec], n_results=top_k)
    docs = res["documents"][0]
    dists = res["distances"][0]   # 距离越小越相关
    return list(zip(docs, dists))


if __name__ == "__main__":
    add_documents([
        "公司年假政策：入职满一年享 5 天带薪年假，每满一年增加 1 天，上限 15 天。",
        "报销流程：发票需在消费后 30 天内提交，超过 500 元需部门经理审批。",
        "远程办公规定：每周最多远程 2 天，需提前一天在系统申请。",
        "食堂供应时间：午餐 11:30-13:00，晚餐 17:30-19:00，周末不供餐。",
    ], source="hr_policy")

    for doc, dist in search("我入职三年了，能休几天假？"):
        print(f"[距离{dist:.3f}] {doc}")

即使问句里没出现「年假」的完整表述，系统也能精准召回那条政策——它匹配的是语义，不是关键词。

🎯 两个生产级细节：

1. ID 用内容哈希（md5）而非 doc_0/doc_1——后者重复写入会互相覆盖，前者天然幂等去重。注意必须用 hashlib 这类稳定哈希，内置 hash() 每次进程重启结果都不同（随机盐），会让去重悄悄失效；
2. 一定要存 metadata——哪怕现在用不上，后面按用户、按时间、按来源过滤全靠它。

---

⑦ 第三步：拼装完整 RAG 问答链路

检索到片段后，塞进 Prompt 交给大模型作答，临门一脚：

CHAT_MODEL = "standard-agent-v1"  # 换成你的对话模型


def rag_answer(question, top_k=3):
    hits = search(question, top_k=top_k)
    context_text = "\n".join(f"- {doc}" for doc, _ in hits)

    system_prompt = (
        "你是企业知识库助手。请严格依据下面的【参考资料】回答问题。"
        "若资料中没有相关信息，直接回答「资料中未提及」，绝不编造。\n\n"
        f"【参考资料】\n{context_text}"
    )
    resp = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
        temperature=0.2,  # 事实问答，温度调低更严谨
    )
    return resp.choices[0].message.content

🎯 抑制幻觉的核心：System Prompt 里那句「没有就回答未提及、绝不编造」至关重要。这正是 RAG 相比裸调模型最大的价值——答案有据可查。

---

⑧ 第四步：加上「相似度阈值」，拒绝答非所问

上面的 search 有个隐患：它永远返回 Top-K 条，哪怕全都不相关。用户问「公司养不养宠物」，库里根本没有，但检索照样会把「年假政策」当成最接近的塞给模型——这就是噪声污染，是 RAG 答非所问的头号原因。

解决办法：设一个距离阈值，太远的直接扔掉：

def search_with_threshold(query, top_k=3, max_distance=1.0):
    """只返回距离小于阈值的结果，过滤掉不相关的噪声"""
    query_vec = embed(query)[0]
    res = collection.query(query_embeddings=[query_vec], n_results=top_k)
    filtered = [
        (doc, dist)
        for doc, dist in zip(res["documents"][0], res["distances"][0])
        if dist < max_distance   # 距离越小越相关，超过阈值视为无关
    ]
    return filtered


def rag_answer_safe(question, top_k=3, max_distance=1.0):
    hits = search_with_threshold(question, top_k, max_distance)
    if not hits:   # 一条相关的都没有，别硬答
        return "抱歉，知识库中没有找到相关信息。"
    context_text = "\n".join(f"- {doc}" for doc, _ in hits)
    # ...（后续与 rag_answer 相同，略）

⚠️ 阈值怎么定：不同 Embedding 模型的距离尺度不一样，max_distance 没有万能值。做法是先跑一批真实问答，打印距离，观察「相关」和「不相关」的分界线，再定阈值。这一步偷不得懒，它直接决定线上准确率。

---

⑨ 把 RAG 用作对话「长期记忆」（含多用户隔离）

回到系列主线：怎么用 RAG 给聊天机器人做长期记忆？思路是每轮对话后写进向量库；下轮提问前先检索相关历史。

但生产环境必须解决两个真问题：① 多个用户的记忆不能串（隐私事故）；② 记忆要带时间戳。这两点靠 metadata 过滤实现：

mem = chroma_client.get_or_create_collection(name="chat_memory")


def remember(user_id, user_msg, assistant_msg, ts):
    """存一轮对话。ts 为时间戳，由调用方传入（如 int(time.time())）"""
    text = f"用户说：{user_msg}\n我回复：{assistant_msg}"
    mem.add(
        # ID 带 user_id + 时间戳 + 内容哈希（复用上文的 doc_id），全局唯一、永不覆盖
        ids=[f"{user_id}_{ts}_{doc_id(text)}"],
        embeddings=embed(text),
        documents=[text],
        metadatas=[{"user_id": user_id, "ts": ts}],   # 关键：隔离与排序靠它
    )


def recall(user_id, query, top_k=3):
    """只召回【当前用户】的相关历史，杜绝串号"""
    if mem.count() == 0:
        return []
    res = mem.query(
        query_embeddings=embed(query),
        n_results=top_k,
        where={"user_id": user_id},   # ⭐ metadata 过滤，实现多租户隔离
    )
    return res["documents"][0]


def chat_with_memory(user_id, user_input, ts):
    related = recall(user_id, user_input)          # 只捞相关历史，不塞全部
    memory_text = "\n".join(related) if related else "（暂无相关历史）"

    system_prompt = (
        "你是有长期记忆的助手。下面是与当前问题相关的历史片段，"
        f"请结合它们理解上下文：\n{memory_text}"
    )
    resp = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input},
        ],
    )
    reply = resp.choices[0].message.content
    remember(user_id, user_input, reply, ts)       # 本轮存进记忆库
    return reply

这样一来，无论聊多久，上下文长度都恒定（只带 Top-K 条相关历史），既省钱又不撞长度上限，还能记住 500 轮前的关键信息，且每个用户的记忆互相隔离。这正是滑动窗口和摘要都做不到的。

🚨 上一版代码的坑（特意留给你看）：如果用一个全局自增计数器当 ID（turn_1、turn_2……），程序一重启计数器归零，新对话的 turn_1 会覆盖掉上次的第一条记忆；而且没有 where 过滤，A 用户能召回到 B 用户的隐私。上面的写法用 user_id + ts + hash 组合 ID + metadata 过滤，两个坑一并堵死。

---

⑩ 检索质量优化：分块、Top-K、重排（附代码）

RAG 效果好不好，七成取决于「检索」。三个最关键的调优点：

优化项	说明	实践建议
文本分块（Chunking）	长文档要切成小块再存	每块 200~500 字，块间留 10~20% 重叠
Top-K 数量	召回几条给模型	太少会漏，太多引噪声、增成本；3~5 条起步
重排（Rerank）	对召回结果二次精排	先粗召回 20 条 → Rerank 精选 5 条，准确率显著提升

分块代码（带重叠的滑动切分）：

def chunk_text(text, size=400, overlap=50):
    """把长文本切成带重叠的块，避免关键句被切断"""
    chunks, start = [], 0
    while start < len(text):
        chunks.append(text[start:start + size])
        start += size - overlap   # 回退 overlap 保证上下文连续
    return chunks

重排思路（两阶段检索，工业级 RAG 标配）：

def search_with_rerank(query, recall_k=20, final_k=5):
    """
    两阶段：先向量粗召回 recall_k 条（快、召回全），
    再用 Rerank 模型对这 recall_k 条精排，取 final_k 条（准）。
    Rerank 模型（如 bge-reranker、cohere-rerank）直接对
    (query, doc) 打分，比纯向量相似度更精细。
    """
    candidates = [doc for doc, _ in search(query, top_k=recall_k)]
    # 伪代码：scores = rerank_model.score(query, candidates)
    # ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    # return [doc for doc, _ in ranked[:final_k]]
    return candidates[:final_k]

💡 重叠为什么重要：一句关键信息若正好落在两块的切割线上，没有重叠就会被拆散、两块都检索不全。留点重叠，语义才完整。中文按字符切分够用；追求精细可按句号/段落等语义边界切。

---

⑪ 常见坑与生产建议

1. Embedding 模型必须前后一致：存库和检索必须用同一个模型。混用不同模型，向量空间对不上，检索结果全是垃圾——这是新手最隐蔽的 bug。

2. ID 用内容哈希或业务主键：别用自增序号，重复写入会互相覆盖。

3. 一定设相似度阈值（第 ⑧ 节）：否则库里没答案时也会硬塞噪声，导致答非所问。

4. 给 Embedding 加缓存：每次写入/检索都调接口，高频场景成本可观。相同文本的向量可缓存复用。

5. 向量库选型对照表：

   场景	推荐	理由
   原型 / 中小项目	ChromaDB	本文所用，零运维、本地跑
   极致性能、纯本地	FAISS	Meta 出品，检索极快
   大规模、需分布式	Milvus / Qdrant	生产级、水平扩展
   不想自己运维	Pinecone / 云托管	全托管，按量付费

6. 敏感数据合规：把用户对话存进向量库前，确认符合隐私政策，必要时脱敏，并提供「删除我的记忆」入口（用 mem.delete(where={"user_id": ...})）。

---

⑫ FAQ：读者最常问的 8 个问题

Q1：RAG 和微调（Fine-tuning）怎么选？
RAG 适合「知识频繁更新、要可追溯」的场景（如企业文档）；微调适合「改变模型说话风格/固定格式」。多数业务先上 RAG，成本低见效快。

Q2：Embedding 用哪个模型？
中文场景可选 bge-large-zh、m3e，或各云厂商的向量模型；英文/多语可用 text-embedding-3-small/large。关键是存检索一致。

Q3：检索不准怎么办？
按优先级排查：① 分块是否合理 → ② 是否加了重排 → ③ Embedding 模型是否匹配语种 → ④ 阈值是否过松。

Q4：知识库要更新/删除怎么办？
ChromaDB 支持 collection.update() 和 collection.delete(ids=...)，按 ID 或 where 条件增删改，无需重建整库。

Q5：一次检索要花多少钱/多久？
主要成本是 1 次 Embedding 调用（把问题转向量）+ 1 次 Chat 调用。向量检索本身在本地毫秒级、几乎免费。

Q6：能同时接入多个知识库吗？
能。建多个 collection，或用 metadata 的 source 字段区分，检索时用 where 过滤。

Q7：Top-K 设多少合适？
先从 3~5 起步。答案分散在多处就调大，噪声多就调小，配合重排效果最好。

Q8：ChromaDB 数据存在哪、会丢吗？
用 PersistentClient(path=...) 会持久化到磁盘，重启不丢。记得把该目录纳入备份。

---

⑬ 完整可运行 Demo

把前面所有环节整合成一个能直接跑的脚本（配好 .env 即可）：

import os
import hashlib
from dotenv import load_dotenv
from openai import OpenAI
import chromadb

load_dotenv()
client = OpenAI(api_key=os.getenv("AGENT_KEY"), base_url=os.getenv("API_BASE_URL"))
EMBED_MODEL, CHAT_MODEL = "text-embedding-3-small", "standard-agent-v1"

db = chromadb.PersistentClient(path="./chroma_db")
col = db.get_or_create_collection(name="demo_kb")


def embed(texts):
    if isinstance(texts, str):
        texts = [texts]
    return [d.embedding for d in client.embeddings.create(model=EMBED_MODEL, input=texts).data]


def build_kb(docs):
    # md5 稳定哈希做 ID：重复运行脚本不会产生重复条目
    ids = [f"d_{hashlib.md5(d.encode('utf-8')).hexdigest()[:16]}" for d in docs]
    col.add(ids=ids, embeddings=embed(docs), documents=docs)


def ask(question, top_k=3, max_distance=1.0):
    res = col.query(query_embeddings=[embed(question)[0]], n_results=top_k)
    hits = [d for d, dist in zip(res["documents"][0], res["distances"][0]) if dist < max_distance]
    if not hits:
        return "知识库中没有相关信息。"
    context = "\n".join(f"- {h}" for h in hits)
    resp = client.chat.completions.create(
        model=CHAT_MODEL, temperature=0.2,
        messages=[
            {"role": "system", "content": f"仅依据以下资料回答，没有就说未提及：\n{context}"},
            {"role": "user", "content": question},
        ],
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    build_kb([
        "公司年假：满一年 5 天，每满一年加 1 天，上限 15 天。",
        "报销：发票 30 天内提交，超 500 元需经理审批。",
        "远程办公：每周最多 2 天，提前一天申请。",
    ])
    print(ask("入职三年能休几天年假？"))
    print(ask("公司允许养宠物吗？"))   # 库里没有 → 触发阈值，回答未提及

---

⑭ 总结与延伸

至此，智能体的「记忆管理」补齐成了完整三部曲：

方案	适用场景	关键词
滑动窗口 / Token 截断	短对话、成本敏感	简单、按时间取舍
摘要压缩	中等长度、要点留存	省长度、有损
RAG 向量检索	长程记忆、大规模知识库	按语义取舍、无限扩展、可追溯

RAG 不只是「记忆」方案，更是企业知识库问答、智能客服、文档助手的标准架构。掌握它，你就能让大模型「说人话」的同时「有据可查」。

下一篇讲智能体工具调用（Function Calling）——让 Agent 不只会聊天，还能真正调 API、查库、执行代码去「干活」。这是从「聊天机器人」进化到「智能体」的关键一跳。

🔜 下一篇预告：《Function Calling 工具调用实战》——让 Agent 不只会聊天，还能真正动手干活。
👍 如果本文帮到你，点赞 / 收藏 / 关注，追更不迷路。评论区聊聊你在 RAG 上踩过的坑。