从零实现Chromadb与ChatGPT插件集成：FastChat和NextChat的最简方案

从零实现Chromadb与ChatGPT插件集成：FastChat和NextChat的最简方案

摘要：把本地知识塞进大模型，又不希望把钱包塞爆？本文用“FastChat + NextChat + Chromadb”三件套，演示一条最简路径：30 分钟搭好可离线的 AI 辅助开发环境，顺带把 ChatGPT 插件也接进来。全程踩坑记录，直接可抄。

1. 背景与痛点：为什么又造轮子？

过去一年，我们团队把 GPT 接进内部 DevOps 流程，结果遇到两个老大难：

1. 私域文档检索慢：把 2 G 的 API 手册切成 512 token 扔给 OpenAI Embedding，再存回 Postgres，一次相似搜索 2~4 s，开发体验直接劝退。
2. 插件链路长：FastChat 官方只给 OpenAI 原生接口，ChatGPT Retrieval Plugin 又默认 Pinecone/SQL，想换库就得改 schema、改鉴权，牵一发动全身。

目标很明确：本地向量库 + 轻量级 LLM 网关 + 前端 NextChat，一键启动，延迟 <300 ms，单卡也能跑。

2. 技术选型：为什么敲定 Chromadb？

结论：Chromadb 对 Python 最友好，嵌入脚本里即可启动，调试阶段不用写 Dockerfile，省时间。

3. 核心实现：30 分钟跑通

3.1 整体架构

┌-------------┐ gRPC/REST ┌-----------┐ │ NextChat │<----------------->│ FastChat │ └-----┬-------┘ └-----┬-----┘ │OpenAI-compatible │/v1/chat/completions ▼ ▼ ┌-------------------------------┐ ┌------------------┘ │ Chromadb（本地，持久化） │ │ └-------------------------------┘ │ ▲（插件回调） │ └------------------------------┘

• Chromadb 既当向量库，也当插件的“知识中心”。
• FastChat 通过自定义model_worker把检索结果注入 system prompt，前端零感知。

3.2 分步部署指南

1. 准备环境

   python >=3.9，推荐 3.11 git clone https://github.com/lm-sys/FastChat.git git clone https://github.com/Yidadaa/ChatGPT-Next-Web.git # 下文简称 NextChat

2. 安装依赖

   pip install "chromadb[http]" fastchat openai tiktoken

3. 启动 Chromadb（带持久化）

   mkdir -p data/chroma chroma run --path data/chroma --port 8000

   默认 REST 端口 8000，后续插件用http://localhost:8000即可。

4. 创建知识库并灌数据

   把docs/下的 markdown 批量切成 chunk，脚本如下：

   # ingest.py import chromadb, tiktoken, glob, re from chromadb.utils import embedding_functions client = chromadb.HttpClient(host="localhost", port=8000) emb_fn = embedding_functions.OpenAIEmbeddingFunction( api_key="sk-xxxxx", model_name="text-embedding-ada-002") collection = client.get_or_create_collection( name="dev_docs", embedding_function=emb_fn) enc = tiktoken.get_encoding("cl100k_base") max_tokens = 800 overlap = 100 for file in glob.glob("docs/**/*.md"): with open(file, encoding="utf-8") as f: text = f.read() # 简易切分 tokens_list = [] start = 0 while start < len(enc.encode(text)): end = start + max_tokens chunk = enc.decode(enc.encode(text)[start:end]) tokens_list.append(chunk) start += max_tokens - overlap # 写入 ids = [f"{file}_{i}" for i in range(len(tokens_list))] metAS = [{"source": file} for _ in ids] collection.add(documents=tokens_list, ids=ids, metadatas=metAS)

   运行python ingest.py，2 G 文档约 3 分钟完成。

5. 启动 FastChat 本地 worker

   FastChat 的model_worker支持--embeddings参数，但我们要的是“检索 + 生成”一体化，所以自定义一个rag_worker.py：

   # rag_worker.py import os, json, chromadb, openai from fastchat.serve.inference import generate_stream from fastchat.serve.base_worker import BaseModelWorker class RAGWorker(BaseModelWorker): def __init__(self): super().__init__("rag-assistant") self.chroma = chromadb.HttpClient(host="localhost", port=8000) self.collection = self.chroma.get_collection("dev_docs") self.openai = openai.AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY")) async def generate_stream_gate(self, params): prompt = params["prompt"] # 1. 检索 top5 res = self.collection.query(query_texts=[prompt], n_results=5) contexts = "\n\n".join(res["documents"][0]) # 2. 拼装 system prompt sys_msg = ("You are a helpful assistant. Answer using the following " "context:\n" + contexts) messages = [{"role": "system", "content": sys_msg}, {"role": "user", "content": prompt}] # 3. 调用 OpenAI async for chunk in self.openai.chat.completions.create( model="gpt-3.5-turbo", messages=messages, stream=True): if x := chunk.choices[0].delta.content: yield {"text": x} # 标准入口 if __name__ == "__main__": import uvicorn, fastchat.serve.base_worker as bw worker = RAGWorker() bw.run_worker(worker, host="0.0.0.0", port=21005)

   启动：

   export OPENAI_API_KEY=sk-xxxxx python rag_worker.py

6. 启动 FastChat Controller + OpenAI-API Server

   # 窗口 1 python -m fastchat.serve.controller --host 0.0.0.0 # 窗口 2（把刚写的 rag_worker 注册到 controller） python -m fastchat.serve.model_worker --model-path "dummy" --model-names rag-assistant --worker-address http://localhost:21005 --controller-address http://localhost:21001 # 窗口 3（对外暴露 /v1/chat/completions） python -m fastchat.serve.openai_api_server --host 0.0.0.0 --port 8001 --controller-address http://localhost:21001

7. 前端 NextChat 指向本地网关

   在 NextChat 根目录复制.env.example为.env.local，只改一行：

   BASE_URL=http://localhost:8001/v1

   npm run dev后，打开浏览器即可聊天，后台链路完全本地。

4. 性能优化：让延迟 <300 ms

4.1 索引构建策略

• 提前计算 Embedding：上面脚本已把 Ada-002 结果持久化；若后期换模型，只需重建 collection，不用改代码。
• 分段大小：800 token 是 Ada-002 的“甜点”，再大边际收益递减。
• Metadata 过滤：给每个文档加version、lang字段，查询时带where={"lang": "zh"}，可把候选集缩小 70%，延迟从 180 ms 降到 50 ms。

4.2 查询优化技巧

1. 客户端缓存：NextChat 自带“历史消息”缓存，可复用；对相同问题直接命中，不再走向量库。
2. 批量检索：Chromadb 0.4 支持一次传 100 条query_texts，平均 RTT 不变，吞吐线性提升。
3. 多进程 worker：FastChat 的openai_api_server可用gunicorn -k uvicorn.worker.UvicornWorker --workers 4，把并发 QPS 从 30 提到 120。

5. 生产环境注意事项

5.1 并发处理

• Chromadb 默认线程池 32，若峰值 QPS>200，启动加chroma run --workers 64。
• FastChat 的 controller 会心跳检测 worker，超时 30 s 即剔除；高并发时把--heartbeat-interval 15调小，可更快摘除故障节点。

5.2 错误恢复机制

• 向量库宕机：在rag_worker.py里捕获chromadb.NetworkError，降级为“无上下文”模式，至少保证对话不中断。
• OpenAI 限流：官方库抛RateLimitError，用tenacity重试 3 次，指数退避，仍失败则返回“服务繁忙”提示，避免前端空等。

6. 总结与可继续折腾的方向

整套方案把“向量库 + LLM 网关 + 前端”拆成三个可独立替换的积木：

• 想换模型？把rag_worker.py里的gpt-3.5-turbo改成自研 7B，只需改两行。
• 想支持图片？把 Chromadb 换成支持多模态的chromadb-beta分支，Embedding 用 CLIP，检索逻辑不变。
• 想做版本回退？给 metadata 加commit_hash，查询带where={"commit": "abc123"}，秒级切换。

下一步，我们准备把 Chromadb 的delta备份接入 CI，每次 MR 自动增量更新，实现“代码合并即知识库更新”。如果你也踩过相似坑，欢迎留言交流。

图：本地三件套数据流——所有组件均可单机 Docker 化，也可拆到 K8s。

写完收工。整套脚本已放到团队内部模板库，新成员git clone && docker-compose up5 分钟就能在本地拥有一个带私域知识的 ChatGPT，调试接口、查文档、生成单测，一条龙。如果你也厌倦了“打开浏览器→搜 Confluence→翻三页找参数”的低效循环，不妨动手试试，祝折腾愉快。