11e - 高级 RAG 技术

RRF_score(d) = Σ 1 / (k + rank_i(d))

from qdrant_client import QdrantClient, models
from fastembed import TextEmbedding, SparseTextEmbedding
 
# 初始化客户端和模型
client = QdrantClient(url="http://localhost:6333")
dense_model = TextEmbedding("BAAI/bge-small-en-v1.5")
sparse_model = SparseTextEmbedding("Qdrant/bm25")
 
# 创建支持混合搜索的集合
client.create_collection(
    collection_name="hybrid_docs",
    vectors_config={
        "dense": models.VectorParams(
            size=384, distance=models.Distance.COSINE
        )
    },
    sparse_vectors_config={
        "sparse": models.SparseVectorParams(
            modifier=models.Modifier.IDF  # 启用 BM25 IDF 加权
        )
    },
)
 
# 索引文档（同时生成稠密和稀疏向量）
documents = ["RAG 系统的核心是检索增强生成...", "向量数据库存储嵌入向量..."]
dense_vectors = list(dense_model.embed(documents))
sparse_vectors = list(sparse_model.embed(documents))
 
for i, doc in enumerate(documents):
    client.upsert(
        collection_name="hybrid_docs",
        points=[models.PointStruct(
            id=i,
            vector={
                "dense": dense_vectors[i].tolist(),
                "sparse": models.SparseVector(
                    indices=sparse_vectors[i].indices.tolist(),
                    values=sparse_vectors[i].values.tolist(),
                )
            },
            payload={"text": doc}
        )]
    )
 
# 混合搜索查询（RRF 融合）
query = "什么是 RAG？"
query_dense = list(dense_model.embed([query]))[0]
query_sparse = list(sparse_model.embed([query]))[0]
 
results = client.query_points(
    collection_name="hybrid_docs",
    prefetch=[
        models.Prefetch(
            query=query_dense.tolist(), using="dense", limit=20
        ),
        models.Prefetch(
            query=models.SparseVector(
                indices=query_sparse.indices.tolist(),
                values=query_sparse.values.tolist(),
            ),
            using="sparse", limit=20
        ),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # RRF 融合
    limit=10,
)

import weaviate from 'weaviate-client';
 
const client = await weaviate.connectToLocal();
 
// 创建集合（自动支持混合搜索）
const collection = await client.collections.create({
  name: 'HybridDocs',
  vectorizers: [
    weaviate.configure.vectorizer.text2VecOpenAI({
      name: 'default',
      model: 'text-embedding-3-small',
    }),
  ],
});
 
// 混合搜索：alpha 控制语义 vs 关键词权重
// alpha=0.75 表示 75% 语义 + 25% 关键词
const results = await collection.query.hybrid('RAG 系统架构', {
  alpha: 0.75,        // 0=纯关键词, 1=纯语义
  limit: 10,
  fusionType: 'RelativeScore',  // 或 'Ranked'（RRF）
  returnProperties: ['text', 'title'],
});
 
for (const item of results.objects) {
  console.log(item.properties.title, item.metadata?.score);
}

你是一个 RAG 系统架构师。请根据以下需求设计混合搜索策略：

**数据特征：**
- 文档类型：[技术文档/法律合同/客服对话/产品目录]
- 是否包含专有名词或编码：[是/否]
- 查询语言：[中文/英文/多语言]

**请输出：**
1. 推荐的稀疏检索方案（BM25 变体选择）
2. 推荐的稠密检索方案（嵌入模型选择）
3. 融合策略（RRF/加权/学习排序）及参数建议
4. alpha 权重建议及调优方向
5. 预期的召回率提升范围

查询 → 初始检索（Top-K=50~100）→ 重排序（Top-N=5~10）→ LLM 生成

import cohere
 
co = cohere.Client("YOUR_COHERE_API_KEY")
 
# 假设已从向量搜索获取初始结果
query = "如何优化 RAG 系统的检索精度？"
initial_results = [
    "RAG 系统通过检索外部知识来增强 LLM 的生成能力...",
    "向量数据库的索引策略直接影响检索速度...",
    "重排序模型可以将检索精度提升 10-30%...",
    "分块策略的选择对 RAG 质量至关重要...",
    "嵌入模型的微调可以显著提升领域内检索效果...",
]
 
# 重排序
rerank_response = co.rerank(
    model="rerank-v3.5",
    query=query,
    documents=initial_results,
    top_n=3,  # 只保留 Top-3
    return_documents=True,
)
 
for result in rerank_response.results:
    print(f"排名: {result.index}, 相关性: {result.relevance_score:.4f}")
    print(f"文档: {result.document.text[:100]}...")
    print()

from sentence_transformers import CrossEncoder
 
# 加载开源重排序模型
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
 
query = "RAG 系统如何处理多语言文档？"
documents = [
    "多语言 RAG 需要选择支持多语言的嵌入模型...",
    "ChromaDB 是一个轻量级向量数据库...",
    "跨语言检索可以使用 multilingual-e5-large 模型...",
]
 
# 计算查询-文档对的相关性分数
pairs = [[query, doc] for doc in documents]
scores = reranker.predict(pairs)
 
# 按分数排序
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
for doc, score in ranked:
    print(f"[{score:.4f}] {doc[:80]}...")

from langchain_community.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import Qdrant
from langchain_openai import OpenAIEmbeddings
 
# 基础检索器
vectorstore = Qdrant.from_existing_collection(
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name="my_docs",
    url="http://localhost:6333",
)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
 
# 添加 Cohere 重排序
compressor = CohereRerank(
    model="rerank-v3.5",
    top_n=5,
    cohere_api_key="YOUR_KEY",
)
 
# 组合：先检索 20 条，再重排序取 Top-5
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever,
)
 
results = compression_retriever.invoke("如何设计 Agent 记忆系统？")

你是一个搜索质量优化专家。请帮我设计重排序策略：

**当前系统状况：**
- 初始检索方式：[向量搜索/混合搜索/BM25]
- 初始检索 Top-K：[数量]
- 平均查询延迟要求：[毫秒]
- 部署环境：[云端 API/自托管 GPU/CPU-only]

**请推荐：**
1. 最适合的重排序模型及理由
2. Top-K → Top-N 的参数建议
3. 预期的精度提升和延迟增加
4. 是否需要级联重排序（粗排→精排）
5. 成本估算（按月查询量 [数量] 计算）

from openai import OpenAI
import numpy as np
 
client = OpenAI()
 
def hyde_search(query: str, vectorstore, k: int = 5):
    """HyDE: 先生成假设文档，再用假设文档的嵌入检索"""
    
    # 第一步：让 LLM 生成假设性回答
    hyde_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": "请直接回答以下问题，写一段详细的技术文档段落。"
                       "不需要说明这是假设性的，直接写内容。"
        }, {
            "role": "user",
            "content": query
        }],
        temperature=0.7,
        max_tokens=300,
    )
    hypothetical_doc = hyde_response.choices[0].message.content
    
    # 第二步：用假设文档的嵌入去检索真实文档
    # 假设文档的嵌入与真实文档在向量空间中更接近
    results = vectorstore.similarity_search(hypothetical_doc, k=k)
    
    return results, hypothetical_doc
 
# 使用示例
query = "微服务架构下如何实现分布式事务？"
results, hyde_doc = hyde_search(query, my_vectorstore)
print(f"HyDE 生成的假设文档:\n{hyde_doc[:200]}...")
print(f"\n检索到 {len(results)} 个相关文档")

from openai import OpenAI
 
client = OpenAI()
 
def decompose_query(query: str) -> list[str]:
    """将复杂查询分解为多个子查询"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": (
                "你是一个查询分解专家。将用户的复杂问题分解为 2-4 个独立的子问题，"
                "每个子问题可以独立检索。输出 JSON 数组格式。"
            )
        }, {
            "role": "user",
            "content": query
        }],
        response_format={"type": "json_object"},
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return result.get("sub_queries", [])
 
def multi_query_search(query: str, vectorstore, k_per_query: int = 3):
    """分解查询 → 分别检索 → 去重合并"""
    sub_queries = decompose_query(query)
    print(f"分解为 {len(sub_queries)} 个子查询: {sub_queries}")
    
    all_docs = []
    seen_ids = set()
    for sub_q in sub_queries:
        docs = vectorstore.similarity_search(sub_q, k=k_per_query)
        for doc in docs:
            doc_id = hash(doc.page_content)
            if doc_id not in seen_ids:
                seen_ids.add(doc_id)
                all_docs.append(doc)
    
    return all_docs
 
# 使用示例
query = "对比 RAG 和微调的优缺点，以及各自适合什么场景？"
results = multi_query_search(query, my_vectorstore)

from openai import OpenAI
 
client = OpenAI()
 
def generate_hypothetical_questions(chunk: str, n: int = 3) -> list[str]:
    """为文档块预生成假设性问题（离线阶段）"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": (
                f"阅读以下文档段落，生成 {n} 个用户可能会问的问题。"
                "这些问题应该能通过该段落回答。输出 JSON 数组。"
            )
        }, {
            "role": "user",
            "content": chunk
        }],
        response_format={"type": "json_object"},
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return result.get("questions", [])
 
# 离线预处理：为每个文档块生成假设问题并索引
for chunk in document_chunks:
    questions = generate_hypothetical_questions(chunk.text)
    for q in questions:
        # 将问题的嵌入存入向量库，关联到原始文档块
        vectorstore.add_texts(
            texts=[q],
            metadatas=[{"source_chunk_id": chunk.id, "type": "hype_question"}]
        )
 
# 运行时：用户查询直接与预计算的问题嵌入匹配（零额外延迟）
results = vectorstore.similarity_search(user_query, k=5)

你是一个 RAG 查询优化专家。请分析以下查询并推荐最佳的查询转换策略：

**原始查询：** [用户查询]
**知识库特征：** [技术文档/FAQ/学术论文/产品手册]
**当前检索效果：** [好/一般/差，简述问题]

**请输出：**
1. 查询类型判断（简单事实/复合问题/模糊意图/对比分析）
2. 推荐的转换策略（HyDE/查询改写/查询分解/Step-Back/不需要转换）
3. 转换后的查询示例
4. 预期的检索效果改善
5. 额外的延迟和成本估算

from llama_parse import LlamaParse
from llama_index.core import VectorStoreIndex
 
# 解析 PDF（自动提取文本、表格、图像）
parser = LlamaParse(
    api_key="YOUR_LLAMA_CLOUD_KEY",
    result_type="markdown",           # 输出 Markdown 格式
    use_vendor_multimodal_model=True,  # 使用多模态模型解析图表
    vendor_multimodal_model="openai-gpt-4o",
    language="ch_sim",                 # 中文支持
)
 
documents = parser.load_data("./technical_report.pdf")
 
# 构建索引（文本 + 图表描述统一索引）
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
 
response = query_engine.query("报告中的架构图展示了哪些组件？")

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
from llama_index.multi_modal_llms.openai import OpenAIMultiModal
from llama_index.core.schema import ImageDocument, TextNode
 
# 加载文本和图像
text_docs = SimpleDirectoryReader("./docs/text/").load_data()
image_docs = SimpleDirectoryReader(
    "./docs/images/", 
    file_extractor={".png": "image", ".jpg": "image"}
).load_data()
 
# 方案 A：用 GPT-4o 将图像转为文本描述后索引
mm_llm = OpenAIMultiModal(model="gpt-4o", max_new_tokens=500)
 
image_text_nodes = []
for img_doc in image_docs:
    # 生成图像描述
    description = mm_llm.complete(
        prompt="详细描述这张图片的内容，包括所有文字、数据和关系。",
        image_documents=[img_doc],
    )
    node = TextNode(
        text=str(description),
        metadata={
            "source": img_doc.metadata.get("file_name"),
            "type": "image_description",
            "original_image_path": img_doc.metadata.get("file_path"),
        }
    )
    image_text_nodes.append(node)
 
# 合并文本和图像描述，统一索引
all_nodes = [node for doc in text_docs for node in doc.to_nodes()] + image_text_nodes
index = VectorStoreIndex(nodes=all_nodes)

# ColPali：直接对文档页面截图进行检索，无需 OCR
from colpali_engine.models import ColPali, ColPaliProcessor
import torch
from PIL import Image
 
# 加载模型
model = ColPali.from_pretrained(
    "vidore/colpali-v1.3", torch_dtype=torch.bfloat16
).eval().cuda()
processor = ColPaliProcessor.from_pretrained("vidore/colpali-v1.3")
 
# 索引：将文档页面截图编码为向量
page_images = [Image.open(f"page_{i}.png") for i in range(10)]
with torch.no_grad():
    batch = processor.process_images(page_images).to("cuda")
    page_embeddings = model(**batch)
 
# 查询：将文本查询编码后与页面向量匹配
query = "2024年Q3的营收数据"
with torch.no_grad():
    query_batch = processor.process_queries([query]).to("cuda")
    query_embedding = model(**query_batch)
 
# 计算相似度
scores = processor.score_multi_vector(query_embedding, page_embeddings)
top_pages = scores[0].argsort(descending=True)[:3]
print(f"最相关的页面: {top_pages.tolist()}")

你是一个多模态 RAG 系统设计师。请根据以下需求设计方案：

**数据源：**
- 文档类型：[PDF报告/PPT演示/技术手册/产品图册]
- 包含的模态：[纯文本/文本+表格/文本+图表/文本+图片+表格]
- 文档数量：[数量] 份，平均 [页数] 页
- 语言：[中文/英文/多语言]

**请输出：**
1. 推荐的文档解析方案（LlamaParse/Unstructured/自定义）
2. 多模态处理策略（统一嵌入/文本化/原生多模态）
3. 各模态的嵌入和索引方案
4. 检索和生成阶段的模型选择
5. 预估成本（按文档量计算解析和查询成本）

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain_core.messages import HumanMessage
from typing import TypedDict, Literal
import json
 
# 定义状态
class AgentState(TypedDict):
    query: str
    retrieved_docs: list[str]
    answer: str
    needs_retrieval: bool
    retry_count: int
    search_queries: list[str]
 
llm = ChatOpenAI(model="gpt-4o-mini")
vectorstore = Qdrant.from_existing_collection(
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name="knowledge_base",
    url="http://localhost:6333",
)
 
# 节点 1：路由决策——是否需要检索
def route_query(state: AgentState) -> AgentState:
    response = llm.invoke([HumanMessage(content=f"""
    分析以下查询，判断是否需要从知识库检索信息。
    如果是常识性问题或简单计算，不需要检索。
    如果涉及特定领域知识、最新信息或具体数据，需要检索。
    
    查询: {state["query"]}
    
    输出 JSON: {{"needs_retrieval": true/false, "reason": "..."}}
    """)])
    result = json.loads(response.content)
    state["needs_retrieval"] = result["needs_retrieval"]
    return state
 
# 节点 2：智能检索
def retrieve(state: AgentState) -> AgentState:
    query = state["search_queries"][-1] if state["search_queries"] else state["query"]
    docs = vectorstore.similarity_search(query, k=5)
    state["retrieved_docs"] = [doc.page_content for doc in docs]
    return state
 
# 节点 3：生成回答
def generate(state: AgentState) -> AgentState:
    context = "\n\n".join(state.get("retrieved_docs", []))
    response = llm.invoke([HumanMessage(content=f"""
    基于以下上下文回答问题。如果上下文不足以回答，明确说明。
    
    上下文: {context}
    问题: {state["query"]}
    """)])
    state["answer"] = response.content
    return state
 
# 节点 4：质量检查——是否需要重试
def check_quality(state: AgentState) -> AgentState:
    response = llm.invoke([HumanMessage(content=f"""
    评估以下回答的质量。检查：
    1. 是否完整回答了问题
    2. 是否有"信息不足"的表述
    3. 是否需要更多检索
    
    问题: {state["query"]}
    回答: {state["answer"]}
    
    输出 JSON: {{"quality": "good"/"needs_retry", "improved_query": "..."}}
    """)])
    result = json.loads(response.content)
    if result["quality"] == "needs_retry" and state["retry_count"] < 2:
        state["retry_count"] += 1
        state["search_queries"].append(result.get("improved_query", state["query"]))
        state["needs_retrieval"] = True
    else:
        state["needs_retrieval"] = False
    return state
 
# 构建图
def should_retrieve(state: AgentState) -> Literal["retrieve", "generate"]:
    return "retrieve" if state["needs_retrieval"] else "generate"
 
def should_retry(state: AgentState) -> Literal["retrieve", "end"]:
    return "retrieve" if state["needs_retrieval"] else "end"
 
graph = StateGraph(AgentState)
graph.add_node("route", route_query)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_node("check", check_quality)
 
graph.set_entry_point("route")
graph.add_conditional_edges("route", should_retrieve)
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "check")
graph.add_conditional_edges("check", should_retry, {"retrieve": "retrieve", "end": END})
 
app = graph.compile()
 
# 运行
result = app.invoke({
    "query": "我们的 RAG 系统在处理中文技术文档时的最佳分块策略是什么？",
    "retrieved_docs": [],
    "answer": "",
    "needs_retrieval": False,
    "retry_count": 0,
    "search_queries": [],
})
print(result["answer"])

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core import VectorStoreIndex, SQLDatabase
from llama_index.core.query_engine import NLSQLTableQueryEngine
from llama_index.llms.openai import OpenAI
 
llm = OpenAI(model="gpt-4o-mini")
 
# 工具 1：向量知识库检索
vector_index = VectorStoreIndex.from_documents(documents)
vector_tool = QueryEngineTool(
    query_engine=vector_index.as_query_engine(),
    metadata=ToolMetadata(
        name="knowledge_base",
        description="搜索技术文档知识库，适合查找概念解释、最佳实践、架构设计等信息",
    ),
)
 
# 工具 2：SQL 数据库查询
sql_database = SQLDatabase.from_uri("sqlite:///analytics.db")
sql_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    tables=["metrics", "costs", "usage_logs"],
)
sql_tool = QueryEngineTool(
    query_engine=sql_engine,
    metadata=ToolMetadata(
        name="analytics_db",
        description="查询分析数据库，适合获取具体的数据指标、成本统计、使用量等数值信息",
    ),
)
 
# 创建 Agent（自动选择工具）
agent = ReActAgent.from_tools(
    tools=[vector_tool, sql_tool],
    llm=llm,
    verbose=True,
    max_iterations=5,
)
 
# Agent 会自动判断使用哪个工具
response = agent.chat("上个月 RAG 系统的查询成本是多少？有什么优化建议？")
# Agent 会先查 SQL 获取成本数据，再查知识库获取优化建议

你是一个 Agentic RAG 系统架构师。请设计一个智能检索 Agent：

**业务需求：**
- 应用场景：[客服问答/技术支持/研究助手/数据分析]
- 数据源：[向量库/SQL数据库/API/网页/混合]
- 查询复杂度：[简单事实/多步推理/对比分析/数据+知识混合]
- 延迟要求：[实时<2s/准实时<10s/离线]

**请输出：**
1. Agent 架构设计（路由→检索→生成→验证的流程图）
2. 路由策略（什么查询走什么路径）
3. 工具定义（每个数据源对应的工具描述）
4. 重试和自我修正策略
5. 成本控制机制（避免无限循环）
6. 推荐的框架和技术栈

┌─────────────────────────────────────────────────────┐
│                  用户查询入口                          │
└──────────────────────┬──────────────────────────────┘
                       ▼
┌─────────────────────────────────────────────────────┐
│              查询路由 Agent                           │
│  判断：简单查询 → 直接检索                              │
│        复杂查询 → 查询分解                              │
│        模糊查询 → HyDE 转换                            │
└──────────────────────┬──────────────────────────────┘
                       ▼
┌─────────────────────────────────────────────────────┐
│              混合检索层                                │
│  BM25（精确关键词）+ 向量搜索（语义）                     │
│  → RRF 融合 → Top-50                                 │
└──────────────────────┬──────────────────────────────┘
                       ▼
┌─────────────────────────────────────────────────────┐
│              重排序层                                  │
│  Cohere Rerank v3.5 → Top-8                          │
└──────────────────────┬──────────────────────────────┘
                       ▼
┌─────────────────────────────────────────────────────┐
│              生成 + 质量检查                            │
│  GPT-4o 生成回答 → 事实一致性验证                        │
│  不满意 → 改写查询重试（最多 2 次）                       │
└─────────────────────────────────────────────────────┘