前四篇我们依次做了：AI 问答、话术库、轻量知识库、多片段召回与命中统计。

靠关键词命中——这有个硬伤：用户换个说法就匹配不到了。比如知识里写「会员等级」，用户问「你们有几种会员」「VIP 怎么分档」，关键词一个都对不上，只能退回大模型瞎编。

RAG 向量版要解决的正是「语义层面找得准」：把知识切块、转成向量存进向量库，提问也转向量，用语义相似度召回最相关片段，再交给大模型作答。

一、整体链路

知识库文档 (KnowledgeDoc)   └─ 切块 → DocumentChunk（文本存 SQLite，便于展示/统计）   └─ Embedding API → 向量 → Chroma 持久化（collection="knowledge"）用户提问   └─ Embedding API → 向量   └─ Chroma 向量召回 Top-K 相似分块   └─ 拼接为【参考资料】交给大模型作答（命中分块 hit_count++）

回落优先级：标准话术(script) → RAG 向量召回(rag) → 轻量知识库(knowledge) → 大模型(ai)。

二、技术选型（已为你定好）

• 向量库：Chroma 本地持久化
  。以本地目录（chroma_store/）存向量，无需起独立服务，和现有 SQLite 项目最契合，部署最简单。
• Embedding：国产大模型 API（OpenAI 兼容）
  。以通义 text-embedding-v3 为例，复用 openai SDK 即可；也可切豆包等任意兼容接口。

只需在环境变量配置 EMBEDDING_API_KEY，没配时系统自动跳过 RAG 层、回落到原有逻辑，不影响运行。

三、配置 Embedding（settings.py）

import os# 向量知识库（RAG）配置：Embedding 由国产大模型 API 生成EMBEDDING_CONFIG = {    "api_key": os.environ.get("EMBEDDING_API_KEY", ""),    "base_url": os.environ.get("EMBEDDING_BASE_URL",                               "https://dashscope.aliyuncs.com/compatible-mode/v1"),    "model": os.environ.get("EMBEDDING_MODEL", "text-embedding-v3"),}

四、ai_client 增加向量化函数

在 aitest/ai_client.py 末尾追加（复用 OpenAI 客户端，与对话共用协议）：

def embed_texts(texts):    """批量获取文本向量，返回 list[list[float]]。"""    cfg = _embed_config()    if not cfg["api_key"]:        raise RuntimeError("未配置 EMBEDDING_API_KEY，无法生成向量")    client = OpenAI(api_key=cfg["api_key"], base_url=cfg["base_url"])    resp = client.embeddings.create(model=cfg["model"], input=texts)    return [item.embedding for item in resp.data]

五、新增模型（models.py）

知识库文档切块后生成向量；文本存 SQLite 方便展示与命中统计，向量交给 Chroma。

class KnowledgeDoc(models.Model):    title = models.CharField(max_length=255, verbose_name="文档标题")    content = models.TextField(verbose_name="文档正文")    source = models.CharField(max_length=255, blank=True, verbose_name="来源")    created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")class DocumentChunk(models.Model):    doc = models.ForeignKey(KnowledgeDoc, on_delete=models.CASCADE, related_name="chunks")    chunk_index = models.PositiveIntegerField(verbose_name="块序号")    content = models.TextField(verbose_name="文本片段")    hit_count = models.PositiveIntegerField(default=0, verbose_name="命中次数")

别忘了：python manage.py makemigrations aitest && python manage.py migrate

六、RAG 核心（aitest/rag.py）

四个关键函数：切块 → 建索引 → 召回 → 命中统计，外加 rag_available() 判断能力是否就绪。

def _split_text(text, size=300, overlap=50):    """优先按段落，超长段落再按长度滑动窗口切分。"""    text = re.sub(r"\s+", " ", text or "").strip()    if not text:        return []    paras = [p.strip() for p in re.split(r"\n+", text) if p.strip()]    chunks, buf = [], ""    for p in paras:        if len(p) > size:            if buf:                chunks.append(buf); buf = ""            step = max(1, size - overlap)            for i in range(0, len(p), step):                c = p[i:i + size].strip()                if c:                    chunks.append(c)        else:            if buf and len(buf) + len(p) + 1 > size:                chunks.append(buf); buf = ""            buf = (buf + " " + p).strip() if buf else p    if buf:        chunks.append(buf)    return chunksdef index_document(doc):    """切块、向量化、写入 Chroma 与 SQLite；重复调用会重建索引。"""    chunks = _split_text(doc.content)    DocumentChunk.objects.filter(doc=doc).delete()    try:        col = get_collection()        old = col.get(where={"doc_id": doc.id})        if old and old.get("ids"):            col.delete(ids=old["ids"])    except Exception:        pass    if not chunks:        return 0    vectors = embed_texts(chunks)    ids = []; docs = []; embs = []; metas = []    for i, (c, v) in enumerate(zip(chunks, vectors)):        DocumentChunk.objects.create(doc=doc, chunk_index=i, content=c)        ids.append(f"{doc.id}_{i}"); docs.append(c); embs.append(v)        metas.append({"doc_id": doc.id, "chunk_index": i, "title": doc.title})    get_collection().upsert(ids=ids, documents=docs, embeddings=embs, metadatas=metas)    return len(chunks)def search(query, top_k=3):    """向量召回：返回 DocumentChunk 列表（按相似度从高到低）。"""    col = get_collection()    if col.count() == 0:        return []    vec = embed_texts([query])[0]    res = col.query(query_embeddings=[vec], n_results=min(top_k, col.count()))    out = []    for cid in res["ids"][0]:        doc_id, _, idx = cid.rpartition("_")        c = DocumentChunk.objects.filter(doc_id=int(doc_id), chunk_index=int(idx)).first()        if c:            out.append(c)    return out

七、views.ask 四级回落

把 RAG 召回插到「标准话术」之后、「轻量知识库」之前：

# 1) 标准话术优先（零成本、秒回）script_hits = [h for h in match_scripts(question) if h.mode == Script.MODE_SCRIPT]if script_hits:    return JsonResponse({..., "source": "script"})# 2) RAG 向量召回if rag_available():    chunks = rag_search(question, top_k=3)    if chunks:        record_chunk_hits(chunks)        knowledge = "\n\n".join(            f"【资料 {i+1}】（来源：{c.doc.title}）\n{c.content}"            for i, c in enumerate(chunks))        answer = ai_ask(question, knowledge=knowledge)        return JsonResponse({..., "source": "rag"})# 3) 回落：轻量知识库  4) 兜底：大模型（source="ai"）

八、后台自动建索引 + 一键种子

admin.py：保存文档时自动切块建向量，删除时同步清理；再提供一键灌入示例文档的命令：

python manage.py seed_knowledge

九、前端显示「来源」

聊天界面在 AI 回答下方加一个来源标签，让运营一眼看清答案出处：

来源：向量知识库(RAG) · deepseek-chat

四种来源对应 source 字段：script / rag / knowledge / ai。

十、怎么跑起来（启用 RAG）

# 1) 配置 Embedding 密钥（以通义为例）$env:EMBEDDING_API_KEY = "你的密钥"# 2) 建表 + 灌入示例知识库python manage.py makemigrations aitest && python manage.py migratepython manage.py seed_knowledge# 3) 启动python manage.py runserver 127.0.0.1:8000

配好密钥后，用「你们有几种会员」「VIP 怎么分档」这类换种说法的问题去问——轻量关键词方案会失灵的地方，RAG 能靠语义相似度把正确片段捞出来。