RAG是神么
大模型就像一个 “学生”，面对 “答题”（生成答案）任务，两种技术对应两种完全不同的 “备考 + 答题” 模式：
| 核心概念 | 类比场景 | 核心逻辑 |
|---|---|---|
| 大模型预训练 | 学生上学时的基础学习 | 掌握通用知识（语文、数学、常识），但缺乏针对性知识点 |
| 微调（Fine-tuning） | 闭卷考试的“考前死记硬背” | 把私有/最新数据“背进脑子里”，考试时不查资料，全靠记忆答题 |
| RAG（检索增强生成） | 开卷考试的“边查资料边答题” | 先检索私有/最新文档片段，再结合基础知识点现场组织答案 |
二、为什么需要 RAG？—— 闭卷考试的 “三大痛点”
大模型只靠预训练（或仅微调），就像学生只靠知识记忆参加闭卷考试，必然遇到三个问题，这正是 RAG 要解决的核心：
- 知识盲区 = 闭卷考试考了 “课外题”
这就像CDA三级有部分超纲题一样，学生（大模型）只背了课本或是大纲（预训练数据）里的内容，但考试突然考 “公司内部规章制度”“班级专属活动安排”—— 这些大纲和课本里没有的 “课外题”，学生完全不会答。
对应大模型：对企业内部文档、个人私有数据，业务说明资料等 “课外知识” 一无所知。 - 知识过期 = 闭卷考试考了 “新教材内容”
就像CDA三级一个难度也是它变化很快，跟着时代发展，会不断更新大纲，学生（大模型）背的是 3 年前的旧课本（预训练数据有截断日期，比如 2023 年），但考试考的是 2025 年的新政策、新科技（如transformer ，大模型等）—— 旧知识根本覆盖不到。
对应大模型：预训练数据之后的 “新鲜事”（如最新产品参数、突发新闻），模型完全不知道。 - 幻觉编故事 = 闭卷考试 “不会就瞎编”
学生（大模型）遇到不会的题，又不能查资料，为了 “交卷” 可能编一个听起来很真的答案（比如 “这个知识点我记得是 XXX”），但实际完全错误。
对应大模型：面对知识盲区或过期内容时，会生成逻辑通顺但与事实不符的 “伪答案”，即 “幻觉”。
而 RAG 的核心思路，就是把 “闭卷考试” 改成 “开卷考试”——不让学生硬背所有知识点，而是允许他考试时先查 “权威参考资料”（私有 / 最新文档），再结合自己的基础认知答题：既补了 “课外知识” 和 “新知识”，又因为答案有原文支撑，大幅减少 “瞎编”（幻觉）。
三、详细拆解：RAG（开卷）vs 微调（闭卷）
- RAG：开卷考试（边查资料边答题）
答题流程：
拿到题目（用户提问）→ 2. 翻参考资料（检索引擎从 “知识库” 中找到与问题相关的文档片段，比如公司内部手册、2025 年新规原文）→ 3. 结合自己的基础知识点（预训练数据）→ 4. 基于 “资料片段 + 基础认知” 生成答案。
类比场景：
老师让学生答 “2025 年某城市的垃圾分类新规”—— 学生没背过，但开卷考试允许查政府官网原文，于是先找到新规文档，再结合自己对 “垃圾分类” 的基础理解，整理出答案。
核心优势（解决闭卷痛点）：
知识不过期：资料库（参考资料）随时更新（比如新增 2025 年新规），无需学生重新 “背”（模型无需重新训练）；
无知识盲区：资料库可以放任意私有内容（如公司内部文档、图片说明，数据库），查得到就答得对；
幻觉率极低：答案基于 “原文片段” 生成，相当于 “引用资料答题”，不会瞎编。
缺点：
答题稍慢：多了 “查资料” 步骤（检索过程），比闭卷答题（直接调用模型）响应时间略长；
依赖资料质量：如果资料库没有相关内容，还是答不出（相当于开卷考试没找到对应参考资料）。 - 微调：闭卷考试（考前死记硬背）
答题流程：
考前把新知识点（私有 / 最新数据，如公司手册、2025 年新规）“硬背” 下来（模型通过微调，把数据融入预训练参数）→ 2. 考试时不查资料，直接靠 “记忆”（模型参数）生成答案。
类比场景：
老师让学生答 “2025 年某城市的垃圾分类新规”—— 学生考前把新规原文死记硬背下来，考试时不查资料，直接默写 + 整理答案。
核心优势：
答题快：没有 “查资料” 步骤，模型直接调用参数生成答案，响应速度比 RAG 快；
答案更连贯：知识点 “内化” 在模型里，生成的答案比 “拼接资料片段” 更流畅自然。
缺点（对应闭卷痛点）：
知识容量有限：学生（模型）的 “记忆力”（参数量）有限，背太多知识点会记混（比如把 A 公司手册和 B 公司手册搞混，导致幻觉）；
更新成本高：新规变了（数据更新），学生要重新背（模型要重新微调），耗时耗力（需要大量计算资源）；
知识盲区难覆盖：如果要背的知识点太多（比如 1000 本内部手册），学生根本背不完（模型微调会 “过拟合”，反而效果变差）。
小结对比
| 对比维度 | RAG（开卷） | 微调（闭卷） |
|---|---|---|
| 核心场景 | 私有/最新数据多、更新频繁（公司文档、新闻、知识库） | 数据量小、更新少、需高流畅度（专属话术、特定领域小数据集） |
| 知识更新 | 简单（直接换资料库） | 复杂（重新微调） |
| 幻觉风险 | 低（基于原文） | 中高（基于记忆，易记混） |
| 响应速度 | 稍慢（多检索步骤） | 快（直接生成） |
| 成本 | 低（无需大量算力） | 高（需 GPU，耗算力） |
# 1. 安装依赖（已安装可跳过）
!pip install langchain openai faiss-cpu tiktoken
!pip uninstall -y langchain langchain-core langchain-community
!pip install langchain langchain-community -U -i https://pypi.tuna.tsinghua.edu.cn/simple
!pip show langchain langchain-community langchain-core langchain*
!pip list show langchain, langchain-classic, langchain-community, langchain-ollama, langchain-qdrant, langchain-text-splitters, langgraph, langgraph-checkpoint, langgraph-prebuilt
langchain基础操作
1.1 提示模板 Prompt Templates
prompt templates 提供了一种预定义，动态注入，模型无关和参数化的提示词生成方式，以便在不同的语言模型之间重用模板
#基础template
from langchain_core.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template("介绍一些{country}的历史悠久的{object}")
prompt1 = prompt_template.format(country="西班牙",object="美食")
print(prompt1)
#chat
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate([
("system","you are a helpful AI bot .你的名字是{name}."),
("human","你好"),
("ai","你好，有什么可以帮助你的？"),
("human","{user_input}"),
])
prompt_value = template.invoke(
{
"name": "haleyprince",
"user_input": "你叫什么名字?"
}
)
print(prompt_value)
template
1.2 模型接口
模型接口有两种形式，一种是专用的，例如ollama，openai等，i啊有一种是二级接口，例如通义千问
!pip install -U langchain-ollama
from langchain_community.llms import Ollama
from langchain_ollama import ChatOllama
OLLAMA_BASE_URL = "http://192.168.52.88:11434"
MODEL_NAME = "qwen:0.5b"
llm = ChatOllama(
#model="qwen2.5:7b-instruct-q4_K_M",
model=MODEL_NAME,
base_url=OLLAMA_BASE_URL, # ← 指定远程地址
temperature=0.6
)
# 适用chatollaa
from langchain_ollama import ChatOllama
from langchain_core.messages import AIMessage
# llm = ChatOllama(
# model="qwen2.5:0.5b",
# temperature=0.8,
# )
messages = [
("system","你是一名专业的翻译家，可以将用户的英文翻译为中文。"),
("human","This is a framework for developing applications powered by large language models (LLMs)."),
]
aimsg = llm.invoke(messages) #invoke() 是同步阻塞的：代码执行到这一行时，会暂停等待 Ollama 返回结果，直到拿到回答后才继续执行下一行
print(aimsg)
#流式
stream = llm.stream(messages)
answer = next(stream)
for chunk in stream:
answer = answer + chunk
print(answer.content)
# 通译千问
import dashscope
API_KEY = "你的apikey"
from langchain_community.chat_models import ChatTongyi
tongyi_chat = ChatTongyi(
#model="qwen2.5:7b-instruct-q4_K_M",
model="qwen-turbo", #模型名是线上的名称与本地不一样
api_key=API_KEY,
temperature=0.6
)
messages = [
("system","你是一名专业的翻译家，可以将用户的英文翻译为中文。"),
("human","This is a framework for developing applications powered by large language models (LLMs)."),
]
aimsg = tongyi_chat.invoke(messages) #invoke() 是同步阻塞的：代码执行到这一行时，会暂停等待 Ollama 返回结果，直到拿到回答后才继续执行下一行
print(aimsg.content)
1.3 聊天记录
#定义聊天记录函数
from langchain_core.chat_history import BaseChatMessageHistory ,InMemoryChatMessageHistory #获取和存储聊天记录
from langchain_core.runnables.history import RunnableWithMessageHistory
store = {}
def get_session_history(session_id: str)->BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryChatMessageHistory()
return store[session_id]
with_message_history = RunnableWithMessageHistory(tongyi_chat,get_session_history)
#必须设置sessionid
config = {"configurable": {"session_id": "abc"}}
#另一种传入tuple信息的方式
from langchain_core.messages import SystemMessage
from langchain_core.messages import HumanMessage
#每次用户的提问用invoke注入，并告诉是哪个id
response = with_message_history.invoke(
[SystemMessage(content="you are a helpful AI bot ."),
HumanMessage(content="您好，我叫一辉")],
config =config,
)
response.content
response = with_message_history.invoke(
[HumanMessage(content="我叫什么")],
config=config,
)
response.content
store
1.4 chains
# 使用LCEL
from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
prompt_template = PromptTemplate.from_template("介绍一些{country}的历史悠久的{object}")
llm = ChatOllama(
#model="qwen2.5:7b-instruct-q4_K_M",
model=MODEL_NAME,
base_url=OLLAMA_BASE_URL, # ← 指定远程地址
temperature=0.6
)
chain = prompt_template | llm #LCEL 连接langchain的功能
result = chain.invoke({"country": "中国", "object":'名胜古迹'})
print(result.content)
2 LangChain在RAG中的基础使用
2.1 文档加载
先要将各种形式的内容资转为文本
文档格式 → 文本提取工具
| 场景 | 推荐工具 / 方案 | 一句话说明 | 开源地址 |
|---|---|---|---|
| PDF → Markdown | marker | 版面还原好，公式、表格一并转 MD | https://github.com/VikParuchuri/marker |
| PDF → 纯文本 | pdfplumber | 轻量、按页精准提取，支持表格坐标 | https://github.com/jsvine/pdfplumber |
| Word (.doc/.docx) | python-docx | 无需装 Office，直接读段落/表格 | https://github.com/python-openxml/python-docx |
| Excel (.xlsx) | pandas + Awesome-LLM-Tabular | 先读成 DataFrame，再让 LLM 做汇总/问答 | pandas 官方 + https://github.com/johnnyhwu/Awesome-LLM-Tabular |
| 图片 (JPG/PNG/PPT) | PaddleOCR | 离线多语言 OCR，一行命令出文本；也可调用 GPT-4 Vision | https://github.com/PaddlePaddle/PaddleOCR |
| YouTube 视频转文字 | youtube-dl + 自动字幕 | 下载视频/字幕，一步到位 | https://github.com/yt-dl-org/youtube-dl |
| B 站视频转文字 | bili2text | 输入 BV 号，直接输出全文 | https://github.com/lanbinshijie/bili2text |
##Load文档
from langchain_community.document_loaders import TextLoader
file_path = "./doc.txt"
loader=TextLoader(file_path,encoding="utf-8")
docs = loader.load()
docs
2.2 文本分割
将文本切分为chunks
| 方法 | 一句话原理 | 优点 | 缺点 | 代码示例 |
|---|---|---|---|---|
| 固定长度 | 按字符数硬切 | 最快、零依赖 | 易断句、断词 | [text[i:i+512] for i in range(0,len(text),512)] |
| 句子级 | 按标点切→句列表 | 语义完整 | 标点不规范就崩 | nltk.sent_tokenize(text) |
| 递归分块 ⭐ | 按优先级分隔符逐级尝试，直到满足长度 | 长度可控，又能保留段落/句子 | 比固定长度慢一丢丢 | RecursiveCharacterTextSplitter(...).split_text(text) |
| 语义分块 ⭐ | 先算句向量，再按相似度聚合并切 | 块内语义最连贯 | 需 GPU/模型，耗时 | SemanticChunker(encoder, max_chunk_tokens=256).chunks(text) |
#1. 固定长度切块（最轻量）
def fixed_chunks(text, size=512):
return [text[i:i+size] for i in range(0, len(text), size)]
print(fixed_chunks("一二三四五六七八九十"*100, 20))
#2按分隔符切块（csv、日志最常见）
def delim_chunks(text, sep="\n"):
return text.split(sep)
print(delim_chunks("a|b|c", sep="|"))
import nltk
import os
# 添加路径到NLTK路径列表的开头（优先级最高）
nltk.data.path.insert(0, '/kaggle/kaggle/data/nltk_data')
# 或者清空现有路径，只使用指定路径
nltk.data.path.clear()
nltk.data.path.append('/kaggle/kaggle/data/nltk_data')
# 验证配置
try:
from nltk.tokenize import sent_tokenize
print("✓ NLTK配置成功")
except LookupError as e:
print(f"✗ NLTK配置失败: {e}")
import nltk, ssl, os
def sent_chunks(text):
return nltk.sent_tokenize(text, language="english")
print(sent_chunks("This is the first sentence. This is the second sentence! Is this the third sentence? Yes, it is indeed."))
#4. 递归字符切块（LangChain 默认，长度+分隔符双保险）
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20,
separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)
print(splitter.split_text("一二三四五六七八九十。"*10))
from langchain_text_splitters import RecursiveCharacterTextSplitter
#5 使用LangChain的语义分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=20,
chunk_overlap=10,
separators=["\n\n", "\n", " ", ""]
)
text = "骑车时需要眼、手、脚协调配合，同时保持平衡，这对前庭系统和本体感觉都是很好的训练。掌握平衡车或自行车能极大提升孩子的自信心。"
chunks = text_splitter.split_text(text)
print(chunks)
快速 demo → 1 或 2
通用 RAG → 4
要求“一句不散” → 3
高端语义边界 → 5
5 种方法已覆盖 99% 场景，按需取用即可。
"""
Args:
file_path: 文件路径
size: 块大小，默认512
overlap: 重叠大小，默认50
Returns:
分块后的文本列表
"""
file_path = "./doc.txt"
# 检查文件是否存在
if not os.path.exists(file_path):
raise FileNotFoundError(f"文件不存在: {file_path}")
# 读取文件内容
try:
with open(file_path, 'r', encoding='utf-8') as file:
text = file.read()
except UnicodeDecodeError:
# 如果UTF-8解码失败，尝试其他编码
with open(file_path, 'r', encoding='gbk') as file:
text = file.read()
# 创建分割器
splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20,
separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)
# 分割文本
chunks = splitter.split_text(text)
print(chunks)
##语义分块（Sentence-BERT 版）
import os
import re
def semantic_chunking_simple(file_path, max_chunk_size=512):
"""
简化版语义分块，避免numpy兼容性问题
"""
# 读取文件
if not os.path.exists(file_path):
raise FileNotFoundError(f"文件不存在: {file_path}")
try:
with open(file_path, 'r', encoding='utf-8') as file:
text = file.read()
except UnicodeDecodeError:
with open(file_path, 'r', encoding='gbk') as file:
text = file.read()
# 按语义边界分割（句子级别）
sentences = re.split(r'([。！？!?])', text)
sentences = [sentences[i] + (sentences[i+1] if i+1 < len(sentences) else '')
for i in range(0, len(sentences)-1, 2)]
sentences = [s.strip() for s in sentences if s.strip()]
# 智能合并句子到块
chunks = []
current_chunk = ""
for sentence in sentences:
# 如果加上当前句子会超过最大大小，则开始新块
if len(current_chunk + sentence) > max_chunk_size and current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence
else:
current_chunk += sentence
# 如果当前块已经很大，也开启新块
if len(current_chunk) > max_chunk_size:
# 找到合适的位置分割
chunks.append(current_chunk.strip())
current_chunk = ""
# 添加最后一个块
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
# 执行分块
try:
file_path = "./doc.txt"
chunks = semantic_chunking_simple(file_path, max_chunk_size=512)
print(f"分块完成，共生成 {len(chunks)} 个块")
print("\n分块结果:")
for i, chunk in enumerate(chunks):
print(f"块 {i+1}: {chunk[:100]}{'...' if len(chunk) > 100 else ''}")
print("---")
except Exception as e:
print(f"错误: {e}")
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
embedding = DashScopeEmbeddings(
model="text-embedding-v2",
dashscope_api_key=API_KEY
)
text = "阿里dashscope平台上可用的模型https://help.aliyun.com/zh/model-studio/getting-started/models"
query_result = embedding.embed_query(text)
query_result
#隐藏层节点数
len(query_result)
向量数据库
量数据库是专为存储、管理和高效检索高维向量数据而设计的新型数据库系统，广泛应用于生成式AI、语义搜索、推荐系统、图像识别和RAG（检索增强生成）等场景。与传统关系型数据库依赖精确匹配不同，向量数据库通过计算向量之间的相似度（如余弦相似度、欧氏距离）来实现“近似最近邻”（ANN, Approximate Nearest Neighbor）搜索，在海量非结构化数据中快速定位最相关的结果。
随着大模型技术的爆发，向量数据库已成为支撑AI应用的核心基础设施之一。其核心优势在于能够将文本、图像、音频等非结构化数据通过Embedding模型转化为向量，并在高维空间中进行语义级检索，从而实现“语义理解”而非关键词匹配
市场格局与定位
2025 年是向量数据库从“概念验证”向“企业级部署”迈进的关键一年。市场主要分为 托管 SaaS 和 自托管开源 两大阵营。
pinecone
定位：AI 原生云数据库的标杆。提供全托管服务，无需维护集群，适合企业快速落地。优势在于高可用、
自动扩容和极低的运维门槛，但劣势是成本较高，且扩展性受限于厂商套餐
Milvus
定位：当前最成熟、功能最全的开源向量数据库。由 Zilliz 维护，支持多种索引算法（IVF、HNSW、DISKANN 等），
在企业级生产环境中表现最稳健。优势是生态丰富、社区活跃；劣势是部署和调优相对复杂，学习曲线较陡
weaviate
定位：一款将向量搜索与图数据库特性相结合的产品。它不仅存储向量，还支持结构化数据（Schema）和 GraphQL 接口，内置模块化的Text2Vec模块，适合需要复杂关系查询的应用
底层算法
向量数据库的核心竞争力之一在于其底层的近似最近邻（ANN）索引算法，这些算法牺牲少量精度以换取数量级的性能提升。常见算法包括：
HNSW (Hierarchical Navigable Small World)
原理：构建多层图结构，高层用于快速跳转，底层精确搜索。
特点：召回率高、延迟低，是当前最主流的ANN算法之一，被 Milvus、Qdrant、Weaviate 广泛采用
IVF (Inverted File Index)
原理：先聚类（如K-Means），查询时只搜索最近簇内的向量。
特点：适合大数据集，但对分布不均的数据敏感，常与PQ（乘积量化）结合使用
FLAT (暴力搜索)
原理：计算查询向量与所有向量的距离。
特点：精度最高，但时间复杂度O(n)，仅适用于百万级以下数据
PQ / SQ / BQ (向量量化技术)
原理：将高维向量压缩编码，减少存储和计算开销。
应用：Qdrant 和 Weaviate 支持标量/二进制量化，显著提升吞吐量
主流产品对比
| 数据库 | 类型 | 核心优势 | 适用场景 | 局限性 |
|---|---|---|---|---|
| Pinecone | 商业 SaaS | 全托管、自动扩缩容、低延迟、混合搜索 | 实时推荐、企业级 RAG | 闭源、定制性差、长期成本高 |
| Milvus | 开源 + 云服务 | 十亿级向量、存算分离、分布式、GPU 加速 | 大规模语义搜索、AI 中台 | 需 K8s，部署运维门槛高 |
| Chroma | 开源嵌入式 | 轻量、Python/JS 原生 API、快速原型 | 本地实验、MVP 验证 | 单机有限，无法横向扩展 |
| PgVector | PG 扩展 | SQL 无缝集成、事务一致、混合查询 | 中小规模、已有 PG 业务 | 亿级后性能衰减，无原生分布式 |
| Elasticsearch | 搜索扩展 | 全文+向量混合、成熟监控 | 日志语义检索、多模态搜索 | ANN 性能弱于专用库，资源开销大 |
| Redis Stack | 内存扩展 | 亚毫秒延迟、实时更新、标签过滤 | 实时推荐、会话缓存 | 内存贵、持久化/备份需额外设计 |
选型建议：
若追求极致易用性和稳定性，且预算充足，可选择 Pinecone 或 Weaviate Cloud。
若需要大规模部署和企业级控制力，Milvus 是首选，尤其适合已有K8s环境的企业。
若用于快速验证或小型项目，Chroma 和 FAISS 是理想起点。
若已使用 PostgreSQL / Elasticsearch / Redis，可通过插件形式平滑引入向量能力，降低迁移成本。
chroma数据库简介
Chroma 是专为 AI 开发者设计的轻量级向量数据库，像操作列表一样简单，特别适合 Python 环境（如 LangChain、LlamaIndex）和小规模数据实验
Chroma 天然支持向量（Embedding）。你可以直接存放 文本、图像 或 音频 的向量表示，省去手动计算相似度的麻烦
默认是 本地模式（文件持久化）。不像 Milvus 需要部署服务，直接在你的机器上运行，像普通 Python 库一样使用
不仅能查相似，还能按 类别、标签 等属性过滤，功能媲美 SQL 数据库的 WHERE 条件 ,就像like一样的
使用示例 ：
Chroma 的 API 设计得非常优雅，通常只需要三步：创建 → 写入 → 查询。
安装数据库
docker-compose 安装
version: "3.8"
services:
chromadb:
image: chromadb/chroma:latest
container_name: chromadb
ports:
- "8000:8000"
volumes:
- /mnt/e/data/chroma_data:/chroma/chroma # 宿主机路径按需修改
environment:
- IS_PERSISTENT=TRUE
- ANONYMIZED_TELEMETRY=TRUE
运行启动 docker-compose up
连接测试
```python
import chromadb
client = chromadb.HttpClient(host='192.168.52.88', port=8000)
try:
client.get_version()
print("连接成功")
except Exception as e:
print(f"连接失败: {e}")
2. 创建数据库实例
创建 Client 和 Collection
Chroma 将 Collection 理解为一张表。这里我们创建一个名为 my_collection 的集合。
import chromadb
# 创建本地客户端（默认保存在 .chromadb 目录）
client = chromadb.Client()
# 创建一个集合（类似于 SQL 的表）
collection = client.create_collection(name="my_collection")
#使用http调用客户端
import chromadb
# 改为 HTTP 客户端，默认端口 8000
client = chromadb.HttpClient(host="192.168.52.88", port=8000)
# 复用或新建集合
collection = client.get_or_create_collection(name="my_collection")
3. 写入数据
添加文档（Document）
每一条记录包含 ids（唯一标识）、documents（文本内容）和 metadatas（元数据，可选）。
# 添加数据到集合
collection.add(
documents=["这是关于菠萝的文档", "这是关于橙子的文档"],
ids=["id1", "id2"], # 唯一标识符
metadatas=[{"category": "水果"}, {"category": "水果"}] # 可选的过滤字段
)
4. 进行向量查询
语义搜索（Semantic Search）
输入一段查询文本，Chroma 会自动将其转化为向量（Embedding），并返回语义上最相似的文档。
# 查询 "菠萝" 相关的文档
results = collection.query(
query_texts=["菠萝是什么"], # Chroma 会自动调用嵌入模型
n_results=2, # 返回最近邻的数量
where={"category": "水果"} # 可选的元数据过滤
)
print(results)
# 结果结构: {'ids': [...], 'documents': [...], 'distances': [...]}
向量数据库Qdrant
1. 安装
① 装 Qdrant（Docker 最快）
docker run -d --name qdrant -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant
docker-compose方式
version: "3.8"
services:
qdrant:
image: qdrant/qdrant:v1.14.1 # 2025-12 官方稳定版
container_name: qdrant
restart: unless-stopped
ports:
- "6333:6333" # REST & Web UI
- "6334:6334" # gRPC
environment:
# 如需认证，取消下行注释并自行更换密钥
# - QDRANT__SERVICE__API_KEY=your_very_strong_key
- QDRANT__CLUSTER__ENABLED=false # 单机模式
- QDRANT__LOG_LEVEL=INFO
- QDRANT__STORAGE__MAX_OPEN_FILES=32768
- QDRANT__PERFORMANCE__MAX_SEARCH_THREADS=4
ulimits:
nofile:
soft: 32768
hard: 32768
deploy:
resources:
limits:
cpus: '4'
memory: 8G
② 装 Python 客户端
docker启动后
webui http://localhost:6333/dashboard
image.png
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
embeddings = DashScopeEmbeddings(
model="text-embedding-v2",
dashscope_api_key=API_KEY
)
text = "阿里dashscope平台上可用的模型https://help.aliyun.com/zh/model-studio/getting-started/models"
query_result = embedding.embed_query(text)
#创建集合collection
from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance,VectorParams
#client=QdrantClietn(":memory:")
client = QdrantClient(host="192.168.52.88",port=6333)
client.create_collection(
collection_name="demo_collection1",
vectors_config=VectorParams(size=1536,distance=Distance.COSINE),
)
vector_store=QdrantVectorStore(
client=client,
collection_name="demo_collection1",
embedding=embeddings,
)
image.png
#add documents
from langchain_core.documents import Document
doc1 = Document(page_content="我是文档一")
doc2 = Document(page_content="我是文档二")
doc3 = Document(page_content="我等下要被删除")
documents = [doc1,doc2,doc3]
vector_store.add_documents(documents=documents)
image.png
#Delete documents
vector_store.delete(ids=['06dec5b0b0fc41bdb8383a4d78d41dbe'])
#search
results = vector_store.similarity_search(query="文档一",k=1) #返回几个，默认4
results
#uss as retriever
retriever = vector_store.as_retriever(search_kwargs={"k": 1})
retriever.invoke("文档一")
## MMR (maximal marginal reievance) 搜索
# 配置 MMR 检索器
retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={
"k": 1, # 返回前 k 个结果
"fetch_k": 1, # 先从向量库中获取 20 个最相似的文档，再从中选出 k 个
"lambda_mult": 0.7 # MMR 的平衡参数，值越大越注重相关性，越小越注重多样性
}
)
# 执行查询
query = "一"
docs = retriever.invoke(query)
for doc in docs:
print(doc.page_content)
RAG pipeline示例案例
import os
from dotenv import load_dotenv
load_dotenv()
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader # 可替换为 PyPDFLoader、DirectoryLoader 等
from langchain_community.vectorstores import Qdrant
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.llms import Ollama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from qdrant_client import QdrantClient
# === 4. 构建 Qdrant 向量库（新版兼容写法）===
from qdrant_client import QdrantClient
from qdrant_client.http import models as rest
# 手动管理集合（避免 force_recreate 问题）
client = QdrantClient(host="192.168.52.88", port=6333)
collection_name = "doc1"
if client.collection_exists(collection_name):
client.delete_collection(collection_name)
client.create_collection(
collection_name=collection_name,
vectors_config=rest.VectorParams(size=1536, distance=rest.Distance.COSINE)
)
# ✅ 关键：from_documents 只传 url（字符串），不传 client！
vectorstore = Qdrant.from_documents(
documents=chunks,
embedding=embeddings,
url="http://192.168.52.88:6333", # 带协议！
collection_name=collection_name
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# === 5. 初始化 Ollama LLM ===
ollama_base_url = "http://192.168.52.88:11434"
llm = Ollama(
base_url=ollama_base_url,
model="qwen:0.5b", # 可替换为 qwen:14b、qwen:32b 等
temperature=0.3,
timeout=300, # 防止长文本生成超时
)
# === 6. Prompt 模板 ===
prompt_template = """
你是一个专业、准确的 AI 助手。请根据以下提供的上下文信息回答用户的问题。
- 如果上下文中有明确答案，请直接引用或总结。
- 如果上下文中没有相关信息，请回答：“根据现有资料无法回答该问题。”
- 不要编造信息。
上下文：
{context}
问题：
{question}
回答：
"""
prompt = ChatPromptTemplate.from_template(prompt_template)
# === 7. 辅助函数：格式化检索结果 ===
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# === 8. RAG Chain（标准方式）===
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# === 9. 【新增】显式 Generate 函数（便于调试或自定义）===
# def generate_with_context(question: str, k: int = 3) -> str:
# """
# 手动检索 + 调用 Ollama 生成答案
# """
# # 1. 直接使用 vectorstore 检索（支持 k）
# docs = vectorstore.similarity_search(question, k=k)
# context = "\n\n".join(doc.page_content for doc in docs)
# # 2. 构造 prompt
# formatted_prompt = prompt_template.format(context=context, question=question)
# # 3. 调用大模型
# response = llm.invoke(formatted_prompt)
# return response.strip()
def generate_with_context(question: str, k: int = 3) -> str:
"""
手动检索 + 调用 Ollama 生成答案
"""
# 1. 动态创建一个 k 值可变的 retriever
dynamic_retriever = vectorstore.as_retriever(search_kwargs={"k": k})
docs = dynamic_retriever.invoke(question)
context = format_docs(docs)
# 2. 构造 prompt
formatted_prompt = prompt_template.format(context=context, question=question)
# 3. 调用 Ollama generate
response = llm.invoke(formatted_prompt)
return response.strip()
# === 10. 测试 ===
if __name__ == "__main__":
question = "请简要说明文档中的核心内容？"
print("【方式1：RAG Chain】")
answer1 = rag_chain.invoke(question)
print("回答：\n", answer1)
print("\n" + "="*50 + "\n")
print("【方式2：显式 Generate】")
answer2 = generate_with_context(question)
print("回答：\n", answer2)
