|
本页对应课程实战代码（`ai58/0902-bigmodel/code/3-rag-embeddings` 与 `ai58/0916-rag/13.向量工程和RAG系统`，两目录课上代码一致，后者多预习讲义/GraphRAG 文档/PPT）。知识链路：
```text
认知（LLM局限、RAG vs 微调）
→ 文档工程（PDF解析、数据转文本、滑窗分块+overlap）
→ 关键词检索（NLTK/jieba 预处理 → ES 倒排索引 → 局限演示）
→ 向量工程（Embedding 原理、API/本地BGE 双路线、相似度、Chroma、ANN选型）
→ 检索进阶（两阶段召回 + CrossEncoder 重排、混合检索）
→ 编排（模板注入、RAG_Bot、调试与批处理）
→ 评估与优化（对比测试、MTEB、三维度优化）
→ 扩展（多模态/增量/多语言、GraphRAG）
```
**通用原则：任何数据进 RAG 前先转文本**——图片→OCR/多模态描述；表格→文字描述或导入数据库走 nl2sql；目前纯文本搜索质量最好。
|
|
RAG_Bot（依赖注入设计：检索函数/LLM/模板皆可替换）：
```python
class RAG_Bot:
def __init__(self, search_func, llm_func, prompt_template, n_results=2):
self.search = search_func; self.llm = llm_func
self.template = prompt_template; self.n_results = n_results
def chat(self, user_query):
results = self.search(user_query, self.n_results)
if not results:
return "抱歉，没有找到相关信息来回答您的问题。"
prompt = build_prompt(self.template, info=results, query=user_query)
return self.llm(prompt)
```
把 `search_func` 换成 ES 关键词检索或 Chroma 向量检索，Bot 一行不改——**换引擎只换检索函数**。
|
|
升级版 AdvancedRAG_Bot：
- 检索源换成向量库（Chroma 封装类）
- 增加 `batch_chat` 批量问答（跑测试集）
- 增加 `debug_search`：先看检索回了什么，再谈生成——**RAG 排障第一步永远是看召回**
VectorDatabase 教学封装：
```python
import chromadb
class VectorDatabase:
def __init__(self, name, embedding_fn):
self.client = chromadb.Client(Settings(allow_reset=True))
self.collection = self.client.get_or_create_collection(name=name)
self.embedding_fn = embedding_fn
def add_documents(self, documents):
self.collection.add(
embeddings=self.embedding_fn(documents), documents=documents,
ids=[f"doc_{i}" for i in range(len(documents))])
def search(self, query, top_k=3):
res = self.collection.query(query_embeddings=self.embedding_fn([query]),
n_results=top_k)
return res['documents'][0]
```
|
|
pdfminer 提取 + 按空行重组段落（`min_line_length` 过滤页眉/目录噪声，连字符修复跨行单词）：
```python
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainer
def extract_text_from_pdf(filename, page_numbers=None, min_line_length=1):
paragraphs, buffer, full_text = [], '', ''
for i, page_layout in enumerate(extract_pages(filename)):
if page_numbers is not None and i not in page_numbers:
continue
for element in page_layout:
if isinstance(element, LTTextContainer):
full_text += element.get_text() + '\n'
for text in full_text.split('\n'):
if len(text) >= min_line_length:
buffer += (' ' + text) if not text.endswith('-') else text.strip('-')
elif buffer:
paragraphs.append(buffer); buffer = ''
if buffer: paragraphs.append(buffer)
return paragraphs
paragraphs = extract_text_from_pdf("llama2.pdf", min_line_length=10)
```
|
|
**句子级滑窗分块（chunk + overlap）**——比固定字符切割更保语义：
```python
from nltk.tokenize import sent_tokenize
def split_text(paragraphs, chunk_size=300, overlap_size=100, min_line_length=10):
sentences = [s.strip() for p in paragraphs for s in sent_tokenize(p)]
chunks, i = [], 0
while i < len(sentences):
chunk = sentences[i]; overlap = ''
prev = i - 1
while prev >= 0 and len(sentences[prev]) + len(overlap) <= overlap_size:
overlap = sentences[prev] + ' ' + overlap # 向前取重叠句
prev -= 1
chunk = overlap + chunk
nxt = i + 1
while nxt < len(sentences) and len(sentences[nxt]) + len(chunk) <= chunk_size:
chunk = chunk + ' ' + sentences[nxt] # 向后填充
nxt += 1
chunks.append(chunk); i = nxt
return chunks
```
要点：overlap 是"句级"而非字符级，保证块边界都落在完整句子处。
|
|
- **图片**：pdfminer 遍历 `LTFigure/LTImage` 递归抽取，PIL 解析失败则回退存原始字节（.bin 兜底）
- **表格**：用文本元素 bbox 坐标聚类启发式检测（Y 容差 10 分行、X 容差 30 判列对齐），导出 CSV/Excel
- 落库时带 metadata 溯源：`metadatas=[{"source": "llama2.pdf"} for _ in chunks]`
|
|
路线一：API（火山方舟示例，OpenAI 兼容）：
```python
client = OpenAI(api_key=api_key, base_url="https://ark.cn-beijing.volces.com/api/v3")
def get_embeddings(texts, model="doubao-embedding-large-text-250515", dimensions=None):
if model == "doubao-embedding-large-text-250515":
dimensions = None # 坑：该模型不支持 dimensions 参数
if dimensions:
data = client.embeddings.create(input=texts, model=model,
dimensions=dimensions).data
else:
data = client.embeddings.create(input=texts, model=model).data
return [x.embedding for x in data]
```
路线二：本地开源（BGE 中文，免 API 依赖）：
```python
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
vecs = model.encode(["文本1", "文本2"], normalize_embeddings=True)
```
|
|
本地中文模型谱系（embedding 演进 + 加载要点）：
| 模型 | 定位 | 关键用法 |
|------|------|---------|
| bert-chinese / roberta-wwm | 基础编码 | `last_hidden_state[:,0,:]` 取 [CLS] 句向量 |
| text2vec-base-chinese | 句向量 | AutoModel + 平均池化 |
| bge-base/large-zh-v1.5 | 检索专用 | `encode_queries()` 带检索指令；cls pooling + F.normalize |
| bge-reranker-large | 重排 | `compute_score(['query','passage'])`，分数无固定范围 |
bge 自定义组装：
```python
from sentence_transformers import SentenceTransformer, models
transformer = models.Transformer("/wks/models/bge-base-zh-v1.5")
pooling = models.Pooling(word_embedding_dimension=transformer.get_word_embedding_dimension(),
pooling_mode_cls_token=True)
model = SentenceTransformer(modules=[transformer, pooling])
```
环境管理：`HF_HOME / TRANSFORMERS_CACHE / HUGGINGFACE_HUB_CACHE / MODELSCOPE_CACHE` 指到大磁盘；模型目录标准件 config.json / pytorch_model.bin / vocab.txt / tokenizer.json。
|
|
相似度与跨语言实测：
```python
from numpy import dot
from numpy.linalg import norm
def cos_sim(a, b): return dot(a, b) / (norm(a) * norm(b))
def euclidean(a, b):
from numpy.linalg import norm as N
return N(a - b)
```
实验：英文 query "global conflicts" 对 5 条中文新闻算余弦相似度排序——语义排序正确，证明中文 embedding 模型具备跨语言语义能力（跨语言检索可行但仍是常见弱点，需实测验证）。
|
|
中文版 to_keywords（jieba 搜索引擎粒度 + 中文停用词）：
```python
import jieba
from nltk.corpus import stopwords
def to_keywords_cn(input_string):
word_tokens = jieba.cut_for_search(input_string) # 搜索引擎粒度
stop_words = set(stopwords.words('chinese'))
return ' '.join(w for w in word_tokens if w not in stop_words)
```
断句也按中文标点：`re.split(r'(?<=[。！？；])', paragraph)`。英文版用 `nltk.word_tokenize` + PorterStemmer 取词根。
|
|
ES 灌库（双字段设计：keywords 供检索、text 留原文）：
```python
from elasticsearch import Elasticsearch, helpers
es = Elasticsearch(hosts=['http://127.0.0.1:9200'])
if es.indices.exists(index="index"): es.indices.delete(index="index")
es.indices.create(index="index")
actions = [{"_index": "index", "_id": f"doc_{i}",
"_source": {"keywords": to_keywords(p), "text": p}}
for i, p in enumerate(paragraphs)]
helpers.bulk(es, actions)
def search(query_string, top_n=3):
res = es.search(index="index",
query={"match": {"keywords": query_string}}, size=top_n)
return [hit["_source"]["text"] for hit in res["hits"]["hits"]]
```
局限演示（必做实验）：检索 "chat version" 与 "conversational variant" 结果完全不同——词汇鸿沟是引入向量检索的最佳动机。
|
|
两阶段检索（`scripts/rank.py` 核心）：向量召回 top5 → CrossEncoder 精排：
```python
from sentence_transformers import CrossEncoder
user_query = "how safe is llama 2"
cands = semantic_search(user_query, 5) # chroma 召回
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2', max_length=512)
scores = reranker.predict([(user_query, d) for d in cands])
ranked = sorted(zip(scores, cands), key=lambda x: x[0], reverse=True)
```
范式：双塔召回（快、粗、可海选）→ 交叉编码重排（慢、准、只排小集合）。BM25 也可做轻量重排（tpf BM25Reranker，10 万条 1.27s/查询）。
|
|
"向量工程"五组件（预习讲义框架）：文档处理 / 向量化 / 存储 / 检索 / 生成——每层独立可替换：
- 存储：Chroma(百万级) / Weaviate / Milvus(亿级) / Qdrant / 商业 Pinecone
- 索引：IVF / HNSW / LSH
- 框架：LangChain / LlamaIndex / Haystack（手写版便于理解，框架版便于复用）
- 部署：Docker / K8s / FastAPI / Streamlit 演示层
|
|
优化三维度清单：
| 维度 | 手段 |
|------|------|
| 检索质量 | 分块粒度调优 / 混合检索 / 重排序 / 查询改写 |
| 生成质量 | 提示词 / 上下文管理 / 输出格式约束 |
| 系统性能 | 缓存 / 并发 / 监控 |
|
|
课程实测踩坑（specstory 调试记录留档）：
1. ES 服务不可用时 `search` 未定义 → `if 'search' not in globals()` 守卫 + 依赖检查 + 降级抛错
2. doubao-embedding-large-text 不支持 dimensions 参数，需强制 None
3. `chroma_client.reset()` 仅演示用，生产勿重置
4. key 与服务地址不硬编码进 notebook（做占位符）
5. 检索无结果 → 直接"无法回答"，防幻觉约束同时写进提示词
|
MTEB
Sentence-Transformers
ElasticSearch