本文使用的技术栈为python+qdrant(docker 镜像）+ollama(使用嵌入模型 bge-m3)+deepseek(兼容 openai 接口)。如果连大模型都想用本地部署的，只需要用ollama再下载个deepseek大模型，将配置改成本地即可。
预先启动向量数据库及嵌入模型服务。
$ docker run -d -ti -p 6333:6333 qdrant/qdrant # 启动向量数据库
$ ollama pull bge-m3 # 启动本地嵌入模型 bge-m3，注意，嵌入模型的好坏影响匹配结果
以下是整个应用的整体流程。
1. init_collection：初始化向量数据库（qdrant），生成知识库对应的 collection；
2. embed（作为 ingest 的子步骤）：通过嵌入模型（bge-m3）将知识库的内容生成对应的嵌入向量（embed）；
3. ingest：将嵌入向量及其对应的知识库内容一一匹配地插入向量数据库（qdrant）；
4. search：将客户端搜索的内容（question）进行向量化（embed）之后传到向量数据库（qdrant）进行匹配，获取匹配分数高的数据（context），这里可以根据需要稍微获取多一些匹配的数据（limit），减少嵌入模型（bge-m3）的误差，但也不能太大，一般 3-5 即可，否则把不相关的内容也塞给大模型，反而干扰回答；
5. chat：将向量数据库中匹配到的知识库内容（context）和客户端搜索的内容（question）一起作为查询内容传给大模型（deepseek）进行处理，这就是传说中的 RAG（Retrieval-Augmented Generation，检索增强生成）了；
6. main：将大模型（deepseek）返回的数据返回给客户端；
代码如下，由于兼容openai的接口，根据实际情况修改自己的base_url、api_key和model即可。
import os
from openai import OpenAI
import requests
# ─── 配置 ─────────────────────────────────────────────────
OLLAMA_URL = "http://localhost:11434"
OLLAMA_EMBED_MODEL = "bge-m3" # 嵌入模型，这个对中文支持比较好，其它支持不好的模型可能很难获取到预期效果
QDRANT_URL = "http://localhost:6333"
COLLECTION = "product_kb" # 类似于关系型数据库的“table”概念
client = OpenAI(
base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com/v1"),
api_key=os.environ.get("LLM_API_KEY", "sk-xxx"),
)
LLM_MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")
# ─── 嵌入模型 ─────────────────────────────────────────────
def embed(text: str) -> list:
res = requests.post(f"{OLLAMA_URL}/api/embed", json={
"model": OLLAMA_EMBED_MODEL,
"input": text,
})
return res.json()["embeddings"][0]
# ─── Qdrant ───────────────────────────────────────────────
def qdrant(method: str, path: str, body: dict = None):
res = requests.request(method, f"{QDRANT_URL}{path}", json=body)
return res.json()
# ─── 初始化 Collection ────────────────────────────────────
def init_collection():
qdrant("DELETE", f"/collections/{COLLECTION}")
qdrant("PUT", f"/collections/{COLLECTION}", {
"vectors": {"size": 1024, "distance": "Cosine"},
})
print("Collection 创建完成")
# ─── 导入知识库 ───────────────────────────────────────────
def ingest(docs):
points = []
for doc in docs:
print(f"嵌入中：{doc['text'][:20]}...")
points.append({
"id": doc["id"],
"vector": embed(doc["text"]),
"payload": {"text": doc["text"]},
})
qdrant("PUT", f"/collections/{COLLECTION}/points", {"points": points})
print("知识库导入完成")
# ─── 检索 ─────────────────────────────────────────────────
def search(question: str, limit: int = 2) -> list:
vector = embed(question)
result = qdrant("POST", f"/collections/{COLLECTION}/points/search", {
"vector": vector,
"limit": limit,
"with_payload": True,
})
hits = result["result"]
for h in hits:
print(f"分数: {h['score']:.4f} | {h['payload']['text'][:30]}")
return [h["payload"]["text"] for h in hits]
# ─── LLM 调用 ─────────────────────────────────────────────
def chat(question: str, context: str) -> str:
res = client.chat.completions.create(
model=LLM_MODEL,
messages=[
{"role": "system", "content": "你是知识库助手，根据提供的知识库内容自由回答。"},
{"role": "user", "content": f"知识库：\n{context}\n\n问题：{question}"},
],
)
print("\n知识库：")
print(context)
return res.choices[0].message.content
# ─── 主流程 ───────────────────────────────────────────────
def ask(question: str) -> str:
docs = search(question)
context = "\n".join(docs)
return chat(question, context)
if __name__ == "__main__":
knowledge = [
{"id": 1, "text": "退款流程：在订单页面点击申请退款，填写原因，3个工作日内处理完成。"},
{"id": 2, "text": "发货时间：下单后24小时内发货，节假日顺延。"},
{"id": 3, "text": "保修政策：产品自购买日起享有一年免费保修服务。"},
]
my_question = "我想退款，怎么操作？"
init_collection()
ingest(knowledge)
print(f"\n问题：{my_question}")
print(f"\n回答：\n{ask(my_question)}")