12221 字
约 40 分钟
1
第20章 RAG-检索增强生成
2026-09-18
第20章 RAG-检索增强生成
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch16-rag.html 所属：第六篇-综合实战
第五篇：综合实战 — 让 Agent 拥有"外挂知识库"——从向量检索到生产级 RAG
20.1 为什么 Agent 需要 RAG？
大模型有两个硬伤：知识截止日期（训练数据有时效性）和领域知识不足（通用模型不懂你的业务）。微调可以解决一部分，但成本高、周期长、更新慢。**RAG（Retrieval-Augmented Generation，检索增强生成）**提供了更灵活的方案：先检索，再生成。 💡 RAG 的核心思想
不是把知识"塞进"模型参数，而是在生成回答前，先从外部知识库检索相关文档，把检索到的内容作为上下文喂给 LLM。模型只需要"阅读理解"，不需要"死记硬背"。
🔴 纯 LLM（无 RAG）
- 知识截止到训练日期，不知道最新信息
- 企业内部文档一概不知
- 容易产生幻觉（编造事实）
- 无法引用信息来源
- 更新知识需要重新训练或微调
🟢 LLM + RAG
- 知识实时更新，随时新增文档
- 接入企业内部知识库（Wiki/文档/工单）
- 基于检索到的真实内容回答，减少幻觉
- 可以引用来源文档，可追溯
- 无需重新训练，成本极低
RAG vs 微调：什么时候用什么？ | 对比维度 | RAG | 微调 | | --- | --- | --- | | 知识更新 | 实时，加文档即可 | 需要重新微调 | | 适合场景 | 事实性问答、文档检索 | 风格/格式/推理模式定制 | | 成本 | 低（向量数据库 + 检索） | 高（GPU 训练） | | 延迟 | 多一次检索（+200-500ms） | 无额外延迟 | | 幻觉控制 | ✅ 强（基于检索内容） | ⚠️ 弱（仍可能编造） | | 最佳实践 | RAG + 微调结合：微调改善风格，RAG 提供知识 | ## 20.2 RAG 的完整流程
RAG 不是简单的"搜索 + 拼接"，一个生产级 RAG 系统包含五个阶段：
阶段一：文档处理（Chunking）
文档不能整篇丢给模型——上下文窗口有限，检索精度也不够。需要将长文档分块（Chunking）。
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 递归字符分割器：按优先级依次尝试分隔符
splitter = RecursiveCharacterTextSplitter(
chunk_size=512, # 每块最大 512 字符
chunk_overlap=64, # 块间重叠 64 字符（保持上下文连贯）
separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""],
keep_separator=True
)
chunks = splitter.split_text(long_document)
print(f"文档被分为 {len(chunks)} 块")
# chunk_size 太大 → 检索不精确、浪费 Token
# chunk_size 太小 → 上下文断裂、语义不完整
# 经验值：512-1024 字符，overlap 10-15%
import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter';
// 递归字符分割器：按优先级依次尝试分隔符
const splitter = new RecursiveCharacterTextSplitter({
chunkSize: 512, // 每块最大 512 字符
chunkOverlap: 64, // 块间重叠 64 字符（保持上下文连贯）
separators: ['\n\n', '\n', '。', '！', '？', '，', ' ', ''],
keepSeparator: true
});
const chunks = await splitter.splitText(longDocument);
console.log(`文档被分为 ${chunks.length} 块`);
// chunk_size 太大 → 检索不精确、浪费 Token
// chunk_size 太小 → 上下文断裂、语义不完整
// 经验值：512-1024 字符，overlap 10-15%
package main
import (
"fmt"
"strings"
)
// 递归字符分割器：按优先级依次尝试分隔符
func splitText(text string, chunkSize, chunkOverlap int, separators []string) []string {
var chunks []string
for len(text) > chunkSize {
// 在 chunkSize 范围内找最佳分割点
end := chunkSize
for _, sep := range separators {
idx := strings.LastIndex(text[:end], sep)
if idx > 0 {
end = idx + len(sep)
break
}
}
chunks = append(chunks, text[:end])
// 保留 overlap 重叠部分
start := end - chunkOverlap
if start 0 {
chunks = append(chunks, text)
}
return chunks
}
func main() {
separators := []string{"\n\n", "\n", "。", "！", "？", "，", " ", ""}
chunks := splitText(longDocument, 512, 64, separators)
fmt.Printf("文档被分为 %d 块\n", len(chunks))
// chunk_size 太大 → 检索不精确、浪费 Token
// chunk_size 太小 → 上下文断裂、语义不完整
// 经验值：512-1024 字符，overlap 10-15%
}
import java.util.*;
public class TextSplitter {
// 递归字符分割器：按优先级依次尝试分隔符
public static List splitText(String text, int chunkSize, int chunkOverlap, List separators) {
List chunks = new ArrayList<>();
while (text.length() > chunkSize) {
int end = chunkSize;
for (String sep : separators) {
int idx = text.substring(0, end).lastIndexOf(sep);
if (idx > 0) {
end = idx + sep.length();
break;
}
}
chunks.add(text.substring(0, end));
int start = Math.max(0, end - chunkOverlap);
text = text.substring(start);
}
if (!text.isEmpty()) {
chunks.add(text);
}
return chunks;
}
public static void main(String[] args) {
List separators = Arrays.asList("\n\n", "\n", "。", "！", "？", "，", " ", "");
List chunks = splitText(longDocument, 512, 64, separators);
System.out.println("文档被分为 " + chunks.size() + " 块");
// chunk_size 太大 → 检索不精确、浪费 Token
// chunk_size 太小 → 上下文断裂、语义不完整
// 经验值：512-1024 字符，overlap 10-15%
}
}
⚠️ 分块策略选择
固定大小分块
- 简单快速，适合大多数场景
- 可能在句子中间截断
语义分块
- 按段落/标题/语义边界分块
- 检索质量更高，但实现复杂
阶段二：向量化（Embedding）
把文本块转成向量（高维浮点数组），使语义相近的文本在向量空间中距离也近。
from langchain_openai import OpenAIEmbeddings
# 选择 Embedding 模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 单文本向量化
vector = embeddings.embed_query("什么是 RAG？")
# 返回 1536 维浮点数组，如 [0.012, -0.034, 0.056, ...]
# 批量向量化（生产环境推荐批量操作）
texts = [chunk for chunk in chunks]
vectors = embeddings.embed_documents(texts)
import { OpenAIEmbeddings } from '@langchain/openai';
// 选择 Embedding 模型
const embeddings = new OpenAIEmbeddings({
model: 'text-embedding-3-small'
});
// 单文本向量化
const vector = await embeddings.embedQuery('什么是 RAG？');
// 返回 1536 维浮点数组，如 [0.012, -0.034, 0.056, ...]
// 批量向量化（生产环境推荐批量操作）
const texts = [...chunks];
const vectors = await embeddings.embedDocuments(texts);
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
)
// 调用 OpenAI Embedding API
type EmbeddingRequest struct {
Model string `json:"model"`
Input []string `json:"input"`
}
type EmbeddingResponse struct {
Data []struct {
Embedding []float64 `json:"embedding"`
} `json:"data"`
}
func embed(texts []string) ([][]float64, error) {
reqBody := EmbeddingRequest{
Model: "text-embedding-3-small",
Input: texts,
}
jsonData, _ := json.Marshal(reqBody)
req, _ := http.NewRequest("POST", "https://api.openai.com/v1/embeddings", bytes.NewReader(jsonData))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+os.Getenv("OPENAI_API_KEY"))
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var result EmbeddingResponse
json.Unmarshal(body, &result)
var vectors [][]float64
for _, d := range result.Data {
vectors = append(vectors, d.Embedding)
}
return vectors, nil
}
func main() {
// 单文本向量化
vectors, _ := embed([]string{"什么是 RAG？"})
fmt.Printf("返回 %d 维浮点数组\n", len(vectors[0]))
// 返回 1536 维浮点数组，如 [0.012, -0.034, 0.056, ...]
// 批量向量化（生产环境推荐批量操作）
batchVectors, _ := embed(chunks)
fmt.Printf("批量向量化 %d 条文本\n", len(batchVectors))
}
import dev.langchain4j.model.openai.OpenAiEmbeddingModel;
import dev.langchain4j.data.embedding.Embedding;
import java.util.*;
public class EmbeddingExample {
public static void main(String[] args) {
// 选择 Embedding 模型
OpenAiEmbeddingModel embeddings = OpenAiEmbeddingModel.builder()
.apiKey(System.getenv("OPENAI_API_KEY"))
.modelName("text-embedding-3-small")
.build();
// 单文本向量化
Embedding vector = embeddings.embed("什么是 RAG？").content();
// 返回 1536 维浮点数组，如 [0.012, -0.034, 0.056, ...]
System.out.println("向量维度: " + vector.vector().length);
// 批量向量化（生产环境推荐批量操作）
List vectors = embeddings.embedAll(chunks).content();
System.out.println("批量向量化 " + vectors.size() + " 条文本");
}
}
``` | Embedding 模型 | 维度 | 特点 | 适合场景 | | --- | --- | --- | --- | | **text-embedding-3-small** | 1536 | OpenAI，性价比高 | 通用英文/中文 | | **text-embedding-3-large** | 3072 | 精度更高，成本翻倍 | 高精度检索 | | **bge-large-zh-v1.5** | 1024 | 开源，中文最强 | 中文场景/私有部署 | | **jina-embeddings-v3** | 1024 | 开源，多语言 | 多语言混合 | ### 阶段三：向量数据库存储
向量数据库专门存储和检索高维向量，支持**近似最近邻搜索（ANN）**，比暴力遍历快几个数量级。 | 向量数据库 | 类型 | 特点 | 适合场景 | | --- | --- | --- | --- | | **Chroma** | 嵌入式 | 零配置，Python 原生 | 原型/小项目 | | **Pinecone** | 云服务 | 全托管，弹性扩展 | 生产环境/无运维 | | **Weaviate** | 自部署 | GraphQL API，混合检索 | 企业自建 | | **Milvus** | 自部署 | 亿级向量，分布式 | 大规模生产 | | **pgvector** | PostgreSQL 插件 | SQL 查询，事务支持 | 已有 PG 的项目 | ```
import chromadb
# 创建客户端（数据持久化到磁盘）
client = chromadb.PersistentClient(path="./vector_db")
# 创建集合（相当于数据库表）
collection = client.get_or_create_collection(
name="knowledge_base",
metadata={"hnsw:space": "cosine"} # 使用余弦相似度
)
# 写入向量
collection.add(
documents=chunks, # 原始文本
embeddings=vectors, # 向量
metadatas=[{"source": "doc1"} for _ in chunks], # 元数据
ids=[f"chunk_{i}" for i in range(len(chunks))] # 唯一 ID
)
import { ChromaClient } from 'chromadb';
// 创建客户端（数据持久化到磁盘）
const client = new ChromaClient({ path: './vector_db' });
// 创建集合（相当于数据库表）
const collection = await client.getOrCreateCollection({
name: 'knowledge_base',
metadata: { 'hnsw:space': 'cosine' } // 使用余弦相似度
});
// 写入向量
await collection.add({
documents: chunks, // 原始文本
embeddings: vectors, // 向量
metadatas: chunks.map(() => ({ source: 'doc1' })), // 元数据
ids: chunks.map((_, i) => `chunk_${i}`) // 唯一 ID
});
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
)
// 通过 ChromaDB REST API 操作
type CollectionRequest struct {
Name string `json:"name"`
Metadata map[string]string `json:"metadata"`
}
func main() {
baseURL := "http://localhost:8000/api/v1"
// 创建集合（相当于数据库表）
colReq := CollectionRequest{
Name: "knowledge_base",
Metadata: map[string]string{"hnsw:space": "cosine"}, // 使用余弦相似度
}
jsonData, _ := json.Marshal(colReq)
resp, _ := http.Post(baseURL+"/collections", "application/json", bytes.NewReader(jsonData))
resp.Body.Close()
// 写入向量
addPayload := map[string]interface{}{
"documents": chunks, // 原始文本
"embeddings": vectors, // 向量
"metadatas": metadatas, // 元数据
"ids": ids, // 唯一 ID
}
jsonData, _ = json.Marshal(addPayload)
resp, _ = http.Post(baseURL+"/collections/knowledge_base/add", "application/json", bytes.NewReader(jsonData))
body, _ := io.ReadAll(resp.Body)
fmt.Println("写入结果:", string(body))
resp.Body.Close()
}
import dev.langchain4j.store.embedding.chroma.ChromaEmbeddingStore;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.data.segment.Metadata;
import java.util.*;
public class VectorStoreExample {
public static void main(String[] args) {
// 创建客户端（数据持久化到磁盘）
ChromaEmbeddingStore store = ChromaEmbeddingStore.builder()
.baseUrl("http://localhost:8000")
.collectionName("knowledge_base")
// 使用余弦相似度
.build();
// 写入向量
List ids = new ArrayList<>();
List segments = new ArrayList<>();
List embeddings = new ArrayList<>();
for (int i = 0; i results[j].Score
})
if len(results) > topK {
results = results[:topK]
}
return results
}
// 向量相似度检索（调用 Embedding API + 余弦相似度）
func vectorSearch(query string, topK int) []Doc {
queryVec := embed(query)
var results []Doc
for _, d := range allDocs {
sim := cosineSimilarity(queryVec, d.Vector)
results = append(results, Doc{Text: d.Text, Score: sim})
}
sort.Slice(results, func(i, j int) bool {
return results[i].Score > results[j].Score
})
if len(results) > topK {
results = results[:topK]
}
return results
}
// 混合检索：BM25 + 向量 加权融合
func hybridSearch(query string, topK int) []Doc {
bm25Results := bm25Search(query, chunks, topK)
vectorResults := vectorSearch(query, topK)
// 加权融合：BM25 占 40%，向量占 60%
merged := make(map[string]float64)
for _, d := range bm25Results {
merged[d.Text] += d.Score * 0.4
}
for _, d := range vectorResults {
merged[d.Text] += d.Score * 0.6
}
var results []Doc
for text, score := range merged {
results = append(results, Doc{Text: text, Score: score})
}
sort.Slice(results, func(i, j int) bool {
return results[i].Score > results[j].Score
})
if len(results) > 5 {
results = results[:5]
}
return results
}
func cosineSimilarity(a, b []float64) float64 {
dot, normA, normB := 0.0, 0.0, 0.0
for i := range a {
dot += a[i] * b[i]
normA += a[i] * a[i]
normB += b[i] * b[i]
}
return dot / (math.Sqrt(normA) * math.Sqrt(normB))
}
import dev.langchain4j.store.embedding.EmbeddingStore;
import dev.langchain4j.store.embedding.EmbeddingMatch;
import dev.langchain4j.data.embedding.Embedding;
import dev.langchain4j.data.segment.TextSegment;
import java.util.*;
import java.util.stream.Collectors;
public class HybridRetriever {
// 1. 向量检索（语义匹配）
public List vectorSearch(String query, int topK) {
Embedding queryEmbedding = embeddingModel.embed(query).content();
List> matches = embeddingStore
.findRelevant(queryEmbedding, topK)
.matches();
return matches.stream()
.map(EmbeddingMatch::embedded)
.collect(Collectors.toList());
}
// 2. BM25 关键词检索（简化版）
public List bm25Search(String query, int topK) {
String[] queryTerms = query.split("\\s+");
Map scores = new HashMap<>();
for (TextSegment doc : allDocs) {
double score = 0.0;
for (String term : queryTerms) {
if (doc.text().contains(term)) {
score += 1.0;
}
}
scores.put(doc, score);
}
return scores.entrySet().stream()
.sorted(Map.Entry.comparingByValue().reversed())
.limit(topK)
.map(Map.Entry::getKey)
.collect(Collectors.toList());
}
// 3. 混合检索：BM25 + 向量 加权融合
public List hybridSearch(String query, int topK) {
List bm25Results = bm25Search(query, topK);
List vectorResults = vectorSearch(query, topK);
// 加权融合：BM25 占 40%，向量占 60%
Map merged = new LinkedHashMap<>();
for (int i = 0; i comparingByValue().reversed())
.limit(5)
.map(Map.Entry::getKey)
.collect(Collectors.toList());
}
}
20.3.3 进阶：重排序（Reranker）
向量检索速度快但精度有限，返回 Top-20 中可能有噪声。重排序模型对这 20 条结果做精细打分，选出最相关的 Top-5。 📊 两阶段检索：快召回 + 精排序
-
召回阶段：向量检索快速从百万文档中召回 Top-50（毫秒级）
-
重排阶段：Cross-Encoder 模型对 50 条逐一精细打分，选 Top-5（百毫秒级）
Cross-Encoder 比 Bi-Encoder（向量检索）精度高，但速度慢 100 倍，所以只用在小规模候选集上。
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
# 重排序模型
reranker = CohereRerank(top_n=5) # 从候选集中选 Top-5
# 压缩检索器：先向量检索召回 → 再重排序
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=vector_retriever # 先召回 Top-20
)
# 最终结果：精度远超纯向量检索
docs = compression_retriever.get_relevant_documents("Agent 如何管理长期记忆？")
import { ContextualCompressionRetriever } from 'langchain/retrievers/contextual_compression';
import { CohereRerank } from '@langchain/cohere';
// 重排序模型
const reranker = new CohereRerank({ topN: 5 }); // 从候选集中选 Top-5
// 压缩检索器：先向量检索召回 → 再重排序
const compressionRetriever = new ContextualCompressionRetriever({
baseCompressor: reranker,
baseRetriever: vectorRetriever // 先召回 Top-20
});
// 最终结果：精度远超纯向量检索
const docs = await compressionRetriever.getRelevantDocuments('Agent 如何管理长期记忆？');
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"sort"
)
type RerankRequest struct {
Query string `json:"query"`
Documents []string `json:"documents"`
TopN int `json:"top_n"`
}
type RerankResponse struct {
Results []struct {
Index int `json:"index"`
RelevanceScore float64 `json:"relevance_score"`
} `json:"results"`
}
// 调用 Cohere Rerank API 对候选文档重排序
func rerank(query string, candidates []string, topN int) []string {
reqBody := RerankRequest{
Query: query,
Documents: candidates,
TopN: topN,
}
jsonData, _ := json.Marshal(reqBody)
req, _ := http.NewRequest("POST", "https://api.cohere.ai/v1/rerank", bytes.NewReader(jsonData))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+os.Getenv("COHERE_API_KEY"))
resp, err := http.DefaultClient.Do(req)
if err != nil {
return candidates[:topN]
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var result RerankResponse
json.Unmarshal(body, &result)
// 按 relevance_score 降序排序
sort.Slice(result.Results, func(i, j int) bool {
return result.Results[i].RelevanceScore > result.Results[j].RelevanceScore
})
var reranked []string
for _, r := range result.Results {
reranked = append(reranked, candidates[r.Index])
}
return reranked
}
func main() {
// 先向量检索召回 Top-20
candidates := vectorSearch("Agent 如何管理长期记忆？", 20)
// 重排序选 Top-5
reranked := rerank("Agent 如何管理长期记忆？", candidates, 5)
fmt.Println("重排序结果:", reranked)
}
import dev.langchain4j.rag.content.retriever.EmbeddingStoreContentRetriever;
import dev.langchain4j.http.okhttp.OkHttpApiBuilder;
import dev.langchain4j.data.segment.TextSegment;
import okhttp3.*;
import com.fasterxml.jackson.databind.*;
import java.util.*;
import java.util.stream.Collectors;
public class RerankExample {
// 调用 Cohere Rerank API 对候选文档重排序
public List rerank(String query, List candidates, int topN) throws Exception {
OkHttpClient client = new OkHttpClient();
ObjectMapper mapper = new ObjectMapper();
Map requestBody = new HashMap<>();
requestBody.put("query", query);
requestBody.put("documents", candidates);
requestBody.put("top_n", topN);
RequestBody body = RequestBody.create(
mapper.writeValueAsString(requestBody),
MediaType.parse("application/json")
);
Request request = new Request.Builder()
.url("https://api.cohere.ai/v1/rerank")
.post(body)
.addHeader("Authorization", "Bearer " + System.getenv("COHERE_API_KEY"))
.build();
Response response = client.newCall(request).execute();
JsonNode root = mapper.readTree(response.body().string());
List reranked = new ArrayList<>();
for (JsonNode item : root.path("results")) {
int index = item.path("index").asInt();
reranked.add(candidates.get(index));
}
return reranked;
}
public static void main(String[] args) throws Exception {
RerankExample reranker = new RerankExample();
// 先向量检索召回 Top-20
List candidates = vectorSearch("Agent 如何管理长期记忆？", 20);
// 重排序选 Top-5
List reranked = reranker.rerank("Agent 如何管理长期记忆？", candidates, 5);
System.out.println("重排序结果: " + reranked);
}
}
20.4 RAG 高级优化技巧
20.4.1 Query 改写
用户提问往往模糊或口语化，直接检索效果差。先用 LLM 改写 Query，提升检索质量。
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
# 改写策略
rewrite_prompt = """你是一个搜索查询优化器。将用户的口语化问题改写为更适合检索的查询。
用户问题：{question}
改写要求：
1. 提取核心关键词
2. 补充可能的同义词
3. 生成 3 个不同角度的查询
输出格式：JSON 数组，如 ["query1", "query2", "query3"]
"""
# 原始问题："Agent 怎么记住东西的？"
# 改写后：["Agent 记忆系统架构", "Agent 短期记忆 长期记忆", "LLM memory management"]
# 多查询并行检索，结果合并去重 → 召回率大幅提升
import {ChatOpenAI} from 'langchain_openai';
const llm = ChatOpenAI(model="gpt-4o-mini");
// 改写策略
const rewrite_prompt = """你是一个搜索查询优化器。将用户的口语化问题改写为更适合检索的查询。;
// 用户问题：{question}
// 改写要求：
// 1. 提取核心关键词
// 2. 补充可能的同义词
// 3. 生成 3 个不同角度的查询
// 输出格式：JSON 数组，如 ["query1", "query2", "query3"]
/** docstring */
// 原始问题："Agent 怎么记住东西的？"
// 改写后：["Agent 记忆系统架构", "Agent 短期记忆 长期记忆", "LLM memory management"]
// 多查询并行检索，结果合并去重 → 召回率大幅提升
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from langchain_openai import ChatOpenAI
// Python: llm = ChatOpenAI(model="gpt-4o-mini")
// 改写策略
// Python: rewrite_prompt = """你是一个搜索查询优化器。将用户的口语化问题改写为更适合检索的查询。
// Python: 用户问题：{question}
// Python: 改写要求：
// Python: 1. 提取核心关键词
// Python: 2. 补充可能的同义词
// Python: 3. 生成 3 个不同角度的查询
// Python: 输出格式：JSON 数组，如 ["query1", "query2", "query3"]
// 原始问题："Agent 怎么记住东西的？"
// 改写后：["Agent 记忆系统架构", "Agent 短期记忆 长期记忆", "LLM memory management"]
// 多查询并行检索，结果合并去重 → 召回率大幅提升
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from langchain_openai import ChatOpenAI
// Python: llm = ChatOpenAI(model="gpt-4o-mini")
// 改写策略
// Python: rewrite_prompt = """你是一个搜索查询优化器。将用户的口语化问题改写为更适合检索的查询。
// Python: 用户问题：{question}
// Python: 改写要求：
// Python: 1. 提取核心关键词
// Python: 2. 补充可能的同义词
// Python: 3. 生成 3 个不同角度的查询
// Python: 输出格式：JSON 数组，如 ["query1", "query2", "query3"]
// 原始问题："Agent 怎么记住东西的？"
// 改写后：["Agent 记忆系统架构", "Agent 短期记忆 长期记忆", "LLM memory management"]
// 多查询并行检索，结果合并去重 → 召回率大幅提升
}
20.4.2 HyDE（假设性文档嵌入）
有时候用户的问题太短，向量化后语义信息不足。HyDE 的思路是：先让 LLM 生成一个"假回答"，用假回答去检索，再用检索结果生成真回答。 🔄 HyDE 流程
**用户问题**："RAG 怎么做？" →
**LLM 生成假回答**："RAG 是检索增强生成技术，通过先检索相关文档再生成回答……" →
**用假回答向量检索**（假回答比短问题语义更丰富）→
**用检索结果生成真回答**
20.4.3 父子分块（Parent-Child Chunking）
检索时用小块（精确匹配），生成时用大块（上下文完整）。
策略说明
# 大块（Parent Chunk）：512 字符 → 存储、生成时使用
# 小块（Child Chunk）：128 字符 → 检索时使用
# 流程：
# 1. 文档先分成 512 字符的大块
# 2. 每个大块再细分为 128 字符的小块
# 3. 小块做 Embedding 并存储，但保留 parent_id
# 4. 检索时匹配小块 → 找到对应的大块 → 用大块作为上下文
# 优势：检索精度高（小块精确），上下文完整（大块信息丰富）
20.5 Agent + RAG：让 Agent 自主决定何时检索
最简单的 RAG 是"每次提问都检索"，但有些问题不需要检索（如"你好"、"1+1=?"）。把 RAG 包装成 Agent 的一个工具，让 Agent 自主决定何时检索。
from langchain.tools import Tool
from langchain.agents import create_react_agent, AgentExecutor
def rag_search(query: str) -> str:
"""检索知识库并返回相关文档"""
# 1. Query 改写
rewritten = llm.invoke(rewrite_prompt.format(question=query))
# 2. 多查询并行检索
all_docs = []
for q in parse_queries(rewritten):
docs = ensemble_retriever.get_relevant_documents(q)
all_docs.extend(docs)
# 3. 去重 + 重排序
unique_docs = deduplicate(all_docs)
ranked_docs = reranker.compress_documents(unique_docs, query)
# 4. 返回拼接的上下文
return "\n\n".join([doc.page_content for doc in ranked_docs[:5]])
# 将 RAG 注册为 Agent 工具
tools = [
Tool(
name="knowledge_search",
description="搜索内部知识库，获取产品文档、技术文档、FAQ等信息。当需要查找具体信息时使用。",
func=rag_search
),
# ... 其他工具
]
# Agent 自主决定何时检索
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 用户："你好" → Agent 不检索，直接回答
# 用户："我们的 API 限流策略是什么？" → Agent 调用 knowledge_search
import {Tool} from 'langchain.tools';
import {create_react_agent, AgentExecutor} from 'langchain.agents';
// def rag_search(query: str) -> str:
/** docstring */
// 1. Query 改写
const rewritten = llm.invoke(rewrite_prompt.format(question=query));
// 2. 多查询并行检索
const all_docs = [];
for (const q of parse_queries(rewritten)) {
const docs = ensemble_retriever.get_relevant_documents(q);
// all_docs.extend(docs)
// 3. 去重 + 重排序
const unique_docs = deduplicate(all_docs);
const ranked_docs = reranker.compress_documents(unique_docs, query);
// 4. 返回拼接的上下文
return "\n\n".join([doc.page_content for doc in ranked_docs[:5]]);
// 将 RAG 注册为 Agent 工具
const tools = [;
// Tool(
const name = "knowledge_search",;
const description = "搜索内部知识库，获取产品文档、技术文档、FAQ等信息。当需要查找具体信息时使用。",;
const func = rag_search;
// ),
// ... 其他工具
// ]
// Agent 自主决定何时检索
const agent = create_react_agent(llm, tools, prompt);
const executor = AgentExecutor(agent=agent, tools=tools, verbose=true);
// 用户："你好" → Agent 不检索，直接回答
// 用户："我们的 API 限流策略是什么？" → Agent 调用 knowledge_search
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from langchain.tools import Tool
// from langchain.agents import create_react_agent, AgentExecutor
// Python: def rag_search(query: str) -> str:
// 1. Query 改写
// Python: rewritten = llm.invoke(rewrite_prompt.format(question=query))
// 2. 多查询并行检索
// Python: all_docs = []
for _, q := range parse_queries(rewritten) {
// Python: docs = ensemble_retriever.get_relevant_documents(q)
// Python: all_docs.extend(docs)
// 3. 去重 + 重排序
// Python: unique_docs = deduplicate(all_docs)
// Python: ranked_docs = reranker.compress_documents(unique_docs, query)
// 4. 返回拼接的上下文
return "\n\n".join([doc.page_content for doc in ranked_docs[:5]])
// 将 RAG 注册为 Agent 工具
// Python: tools = [
// Python: Tool(
// Python: name="knowledge_search",
// Python: description="搜索内部知识库，获取产品文档、技术文档、FAQ等信息。当需要查找具体信息时使用。",
// Python: func=rag_search
// Python: ),
// ... 其他工具
// Python: ]
// Agent 自主决定何时检索
// Python: agent = create_react_agent(llm, tools, prompt)
// Python: executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
// 用户："你好" → Agent 不检索，直接回答
// 用户："我们的 API 限流策略是什么？" → Agent 调用 knowledge_search
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from langchain.tools import Tool
// from langchain.agents import create_react_agent, AgentExecutor
// Python: def rag_search(query: str) -> str:
// 1. Query 改写
// Python: rewritten = llm.invoke(rewrite_prompt.format(question=query))
// 2. 多查询并行检索
// Python: all_docs = []
for (var q : parse_queries(rewritten)) {
// Python: docs = ensemble_retriever.get_relevant_documents(q)
// Python: all_docs.extend(docs)
// 3. 去重 + 重排序
// Python: unique_docs = deduplicate(all_docs)
// Python: ranked_docs = reranker.compress_documents(unique_docs, query)
// 4. 返回拼接的上下文
return "\n\n".join([doc.page_content for doc in ranked_docs[:5]]);
// 将 RAG 注册为 Agent 工具
// Python: tools = [
// Python: Tool(
// Python: name="knowledge_search",
// Python: description="搜索内部知识库，获取产品文档、技术文档、FAQ等信息。当需要查找具体信息时使用。",
// Python: func=rag_search
// Python: ),
// ... 其他工具
// Python: ]
// Agent 自主决定何时检索
// Python: agent = create_react_agent(llm, tools, prompt)
// Python: executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
// 用户："你好" → Agent 不检索，直接回答
// 用户："我们的 API 限流策略是什么？" → Agent 调用 knowledge_search
}
✅ Agent + RAG 的优势
- 智能路由：简单问题不检索，省 Token、降延迟
- 多轮检索：复杂问题可以多次检索不同关键词
- 结果验证：Agent 可以判断检索结果是否相关，不相关则换关键词重搜
- 引用来源：Agent 可以在回答中标注"根据 [文档名] ..."
20.6 生产级 RAG 架构
🏭 企业级 RAG 系统架构
**数据接入层**
文档加载器（PDF/Word/HTML/Confluence）→ 文档清洗 → 元数据提取
**索引层**
分块 → 向量化 → 写入向量数据库（增量更新、版本管理）
**检索层**
Query 改写 → 混合检索（BM25 + 向量）→ 重排序 → 上下文组装
**生成层**
Prompt 模板 → LLM 生成 → 回答后处理（去幻觉、加引用）
**评估层**
检索准确率、回答忠实度、用户反馈 → 持续优化
20.7 GraphRAG：知识图谱增强检索
传统向量 RAG 有一个致命弱点：只能检索局部相似片段，无法回答全局性问题。比如"整个系统架构包含哪些模块？"这样的问题需要跨文档、跨段落的全局视角。GraphRAG 通过构建知识图谱，让 RAG 具备"关联推理"的能力。 💡 GraphRAG 的核心思想
不只是把文档切成块存向量库，而是先从文档中抽取实体和关系，构建一张知识图谱。检索时不是做向量相似度搜索，而是在图上进行子图检索和图遍历，找到与问题相关的实体及其关联信息。
20.7.1 GraphRAG 的四步流程
from llama_index.core import KnowledgeGraphIndex
from llama_index.core.graph_stores import SimpleGraphStore
from llama_index.llms.openai import OpenAI
# 1. 实体抽取 + 关系构建（LLM 自动完成）
llm = OpenAI(model="gpt-4o-mini")
# KnowledgeGraphIndex 自动从文档中抽取实体和关系
kg_index = KnowledgeGraphIndex.from_documents(
documents,
llm=llm,
max_triplets_per_chunk=5, # 每个块最多抽取 5 个三元组
include_embeddings=True # 同时生成实体嵌入
)
# 抽取的三元组示例：
# ("RAG", "属于", "检索增强生成技术")
# ("向量数据库", "用于", "RAG")
# ("Agent", "调用", "RAG")
# 2. 子图检索：从问题实体出发，沿关系边遍历
query_engine = kg_index.as_query_engine(
include_text=True, # 同时返回原文片段
similarity_top_k=5, # 返回 Top-5 相关子图
graph_depth=2 # 图遍历深度 2 跳
)
# 3. 生成回答时，上下文 = 子图结构 + 原文片段
response = query_engine.query("Agent 和 RAG 是什么关系？")
# → 沿 "Agent" → "调用" → "RAG" → "属于" → "检索增强生成技术" 遍历
# → 上下文包含完整关系链，而非孤立的文本块
import {KnowledgeGraphIndex} from 'llama_index.core';
import {SimpleGraphStore} from 'llama_index.core.graph_stores';
import {OpenAI} from 'llama_index.llms.openai';
// 1. 实体抽取 + 关系构建（LLM 自动完成）
const llm = OpenAI(model="gpt-4o-mini");
// KnowledgeGraphIndex 自动从文档中抽取实体和关系
const kg_index = KnowledgeGraphIndex.from_documents(;
// documents,
const llm = llm,;
const max_triplets_per_chunk = 5, # 每个块最多抽取 5 个三元组;
const include_embeddings = true # 同时生成实体嵌入;
// )
// 抽取的三元组示例：
// ("RAG", "属于", "检索增强生成技术")
// ("向量数据库", "用于", "RAG")
// ("Agent", "调用", "RAG")
// 2. 子图检索：从问题实体出发，沿关系边遍历
const query_engine = kg_index.as_query_engine(;
const include_text = true, # 同时返回原文片段;
const similarity_top_k = 5, # 返回 Top-5 相关子图;
const graph_depth = 2 # 图遍历深度 2 跳;
// )
// 3. 生成回答时，上下文 = 子图结构 + 原文片段
const response = query_engine.query("Agent 和 RAG 是什么关系？");
// → 沿 "Agent" → "调用" → "RAG" → "属于" → "检索增强生成技术" 遍历
// → 上下文包含完整关系链，而非孤立的文本块
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from llama_index.core import KnowledgeGraphIndex
// from llama_index.core.graph_stores import SimpleGraphStore
// from llama_index.llms.openai import OpenAI
// 1. 实体抽取 + 关系构建（LLM 自动完成）
// Python: llm = OpenAI(model="gpt-4o-mini")
// KnowledgeGraphIndex 自动从文档中抽取实体和关系
// Python: kg_index = KnowledgeGraphIndex.from_documents(
// Python: documents,
// Python: llm=llm,
// Python: max_triplets_per_chunk=5, # 每个块最多抽取 5 个三元组
// Python: include_embeddings=True # 同时生成实体嵌入
// Python: )
// 抽取的三元组示例：
// ("RAG", "属于", "检索增强生成技术")
// ("向量数据库", "用于", "RAG")
// ("Agent", "调用", "RAG")
// 2. 子图检索：从问题实体出发，沿关系边遍历
// Python: query_engine = kg_index.as_query_engine(
// Python: include_text=True, # 同时返回原文片段
// Python: similarity_top_k=5, # 返回 Top-5 相关子图
// Python: graph_depth=2 # 图遍历深度 2 跳
// Python: )
// 3. 生成回答时，上下文 = 子图结构 + 原文片段
// Python: response = query_engine.query("Agent 和 RAG 是什么关系？")
// → 沿 "Agent" → "调用" → "RAG" → "属于" → "检索增强生成技术" 遍历
// → 上下文包含完整关系链，而非孤立的文本块
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from llama_index.core import KnowledgeGraphIndex
// from llama_index.core.graph_stores import SimpleGraphStore
// from llama_index.llms.openai import OpenAI
// 1. 实体抽取 + 关系构建（LLM 自动完成）
// Python: llm = OpenAI(model="gpt-4o-mini")
// KnowledgeGraphIndex 自动从文档中抽取实体和关系
// Python: kg_index = KnowledgeGraphIndex.from_documents(
// Python: documents,
// Python: llm=llm,
// Python: max_triplets_per_chunk=5, # 每个块最多抽取 5 个三元组
// Python: include_embeddings=True # 同时生成实体嵌入
// Python: )
// 抽取的三元组示例：
// ("RAG", "属于", "检索增强生成技术")
// ("向量数据库", "用于", "RAG")
// ("Agent", "调用", "RAG")
// 2. 子图检索：从问题实体出发，沿关系边遍历
// Python: query_engine = kg_index.as_query_engine(
// Python: include_text=True, # 同时返回原文片段
// Python: similarity_top_k=5, # 返回 Top-5 相关子图
// Python: graph_depth=2 # 图遍历深度 2 跳
// Python: )
// 3. 生成回答时，上下文 = 子图结构 + 原文片段
// Python: response = query_engine.query("Agent 和 RAG 是什么关系？")
// → 沿 "Agent" → "调用" → "RAG" → "属于" → "检索增强生成技术" 遍历
// → 上下文包含完整关系链，而非孤立的文本块
}
20.7.2 GraphRAG vs 向量 RAG：什么时候用什么？ | 对比维度 | 向量 RAG | GraphRAG | | --- | --- | --- | | 检索方式 | 语义相似度搜索 | 实体关系图遍历 | | 擅长问题 | 事实性问答（"X 是什么？"） | 关联性问答（"X 和 Y 什么关系？"） | | 全局视角 | ❌ 只看局部相似片段 | ✅ 跨文档关联推理 | | 多跳推理 | ❌ 最多 1 跳（检索到的片段） | ✅ 多跳图遍历（A→B→C→D） | | 构建成本 | 低（分块 + Embedding） | 高（LLM 抽取实体关系） | | 更新成本 | 低（增量加向量） | 中（需更新图谱边） | | 延迟 | 毫秒级 | 百毫秒~秒级 | | 最佳实践 | 混合使用：向量 RAG 做日常问答，GraphRAG 做复杂关联分析 | ⚠️ GraphRAG 的代价
- 构建成本高：需要用 LLM 从每篇文档中抽取实体和关系，Token 消耗大
- 维护复杂：文档更新时需要同步更新图谱结构，比加向量复杂得多
- 实体歧义："苹果"是公司还是水果？需要实体消歧（Entity Disambiguation）
- 不适合简单问答：杀鸡用牛刀，"RAG 是什么"用向量检索就够了 ✅ GraphRAG 的黄金场景
- 多跳推理问答："A 公司的 CEO 毕业于哪所大学？" → 需要跨文档追溯
- 关系分析："这个故障影响了哪些下游服务？" → 微服务依赖图
- 全局摘要："整个系统架构包含哪些模块？" → 需要全局视角
- 因果追溯："为什么这个指标下降了？" → 因果链路遍历
20.8 Agentic RAG：Agent 驱动的自适应检索
传统 RAG 是一个固定流水线：用户提问 → 检索 → 拼接上下文 → LLM 生成。无论问题难易，流程都一样。但这不合理——有些问题不需要检索，有些需要检索多次，有些需要先检索再反思再检索。Agentic RAG 把 Agent 引入检索循环，让 Agent 动态决策是否检索、检索几次、何时停止。 💡 Agentic RAG 的核心思想
不再是一条固定的"检索→生成"流水线，而是让 Agent 作为检索编排者，根据问题难度动态决定检索策略：简单问题直接回答，复杂问题多轮检索+反思，检索不充分则换角度重搜，直到信息充分才生成最终回答。
20.8.1 传统 RAG vs Agentic RAG
🔴 传统 RAG Pipeline
- 固定流程：Query → Retrieve → Generate
- 每次提问都检索，不管是否需要
- 只检索一次，不管结果够不够
- 检索结果直接拼接，不做质量判断
- 无法根据结果调整检索策略
🟢 Agentic RAG
- 动态流程：Agent 自主决策每一步
- 判断是否需要检索（简单问题跳过）
- 支持多轮检索，直到信息充分
- 每次检索后反思：结果是否相关？
- 可以改写 Query 换角度重新检索
20.8.2 Agentic RAG 的核心循环
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
class RAGState(TypedDict):
question: str
retrieved_docs: List[str]
answer: str
reflection: str # 反思结果
search_count: int # 检索次数
sufficient: bool # 信息是否充分
def decide_to_search(state: RAGState) -> str:
"""Agent 决策：是否需要检索"""
question = state["question"]
# 简单问题不检索（如打招呼、算术题）
if is_simple_question(question):
return "generate_direct"
# 已有文档是否充分？
if state.get("sufficient", False):
return "generate"
# 最多检索 3 轮，防止死循环
if state.get("search_count", 0) >= 3:
return "generate"
return "search"
def search(state: RAGState) -> RAGState:
"""执行检索：Query 改写 + 混合检索 + 重排序"""
question = state["question"]
count = state.get("search_count", 0)
# 第 2 轮起，根据反思结果改写 Query
if count > 0:
question = llm_rewrite_query(question, state.get("reflection", ""))
docs = hybrid_search(question, top_k=10)
ranked_docs = reranker.rerank(docs, question, top_n=5)
state["retrieved_docs"] = state.get("retrieved_docs", []) + ranked_docs
state["search_count"] = count + 1
return state
def reflect(state: RAGState) -> RAGState:
"""反思：检索结果是否足以回答问题？"""
prompt = f"""问题：{state['question']}
已有文档：{state['retrieved_docs']}
判断：当前文档是否足以完整回答问题？
如果不足，指出缺少什么信息。"""
result = llm.invoke(prompt)
state["reflection"] = result
state["sufficient"] = "信息充分" in result
return state
def generate(state: RAGState) -> RAGState:
"""生成最终回答"""
context = "\n\n".join(state["retrieved_docs"])
state["answer"] = llm.invoke(f"基于以下信息回答问题：\n{context}\n\n问题：{state['question']}")
return state
# 构建工作流图
workflow = StateGraph(RAGState)
workflow.add_node("decide", decide_to_search)
workflow.add_node("search", search)
workflow.add_node("reflect", reflect)
workflow.add_node("generate", generate)
workflow.set_entry_point("decide")
workflow.add_conditional_edges("decide", decide_to_search, {
"search": "search",
"generate": "generate",
"generate_direct": "generate"
})
workflow.add_edge("search", "reflect")
workflow.add_conditional_edges("reflect", lambda s: "search" if not s["sufficient"] and s["search_count"] str:
/** docstring */
// question = state["question"]
// 简单问题不检索（如打招呼、算术题）
if (is_simple_question(question)) {
return "generate_direct";
// 已有文档是否充分？
if (state.get("sufficient", false)) {
return "generate";
// 最多检索 3 轮，防止死循环
if (state.get("search_count", 0) >= 3) {
return "generate";
return "search";
// def search(state: RAGState) -> RAGState:
/** docstring */
// question = state["question"]
// count = state.get("search_count", 0)
// 第 2 轮起，根据反思结果改写 Query
if (count > 0) {
// question = llm_rewrite_query(question, state.get("reflection", ""))
// docs = hybrid_search(question, top_k=10)
// ranked_docs = reranker.rerank(docs, question, top_n=5)
// state["retrieved_docs"] = state.get("retrieved_docs", []) + ranked_docs
// state["search_count"] = count + 1
return state;
// def reflect(state: RAGState) -> RAGState:
/** docstring */
// prompt = f"""问题：{state['question']}
// 已有文档：{state['retrieved_docs']}
// 判断：当前文档是否足以完整回答问题？
// 如果不足，指出缺少什么信息。"""
// result = llm.invoke(prompt)
// state["reflection"] = result
// state["sufficient"] = "信息充分" in result
return state;
// def generate(state: RAGState) -> RAGState:
/** docstring */
// context = "\n\n".join(state["retrieved_docs"])
// state["answer"] = llm.invoke(f"基于以下信息回答问题：\n{context}\n\n问题：{state['question']}")
return state;
// 构建工作流图
// workflow = StateGraph(RAGState)
// workflow.add_node("decide", decide_to_search)
// workflow.add_node("search", search)
// workflow.add_node("reflect", reflect)
// workflow.add_node("generate", generate)
// workflow.set_entry_point("decide")
// workflow.add_conditional_edges("decide", decide_to_search, {
// "search": "search",
// "generate": "generate",
// "generate_direct": "generate"
// })
// workflow.add_edge("search", "reflect")
// workflow.add_conditional_edges("reflect", lambda s: "search" if not s["sufficient"] and s["search_count"] str:
// Python: question = state["question"]
// 简单问题不检索（如打招呼、算术题）
if is_simple_question(question) {
return "generate_direct"
// 已有文档是否充分？
if state.get("sufficient", false) {
return "generate"
// 最多检索 3 轮，防止死循环
if state.get("search_count", 0) >= 3 {
return "generate"
return "search"
// Python: def search(state: RAGState) -> RAGState:
// Python: question = state["question"]
// Python: count = state.get("search_count", 0)
// 第 2 轮起，根据反思结果改写 Query
if count > 0 {
// Python: question = llm_rewrite_query(question, state.get("reflection", ""))
// Python: docs = hybrid_search(question, top_k=10)
// Python: ranked_docs = reranker.rerank(docs, question, top_n=5)
// Python: state["retrieved_docs"] = state.get("retrieved_docs", []) + ranked_docs
// Python: state["search_count"] = count + 1
return state
// Python: def reflect(state: RAGState) -> RAGState:
// Python: prompt = f"""问题：{state['question']}
// Python: 已有文档：{state['retrieved_docs']}
// Python: 判断：当前文档是否足以完整回答问题？
// Python: 如果不足，指出缺少什么信息。"""
// Python: result = llm.invoke(prompt)
// Python: state["reflection"] = result
// Python: state["sufficient"] = "信息充分" in result
return state
// Python: def generate(state: RAGState) -> RAGState:
// Python: context = "\n\n".join(state["retrieved_docs"])
// Python: state["answer"] = llm.invoke(f"基于以下信息回答问题：\n{context}\n\n问题：{state['question']}")
return state
// 构建工作流图
// Python: workflow = StateGraph(RAGState)
// Python: workflow.add_node("decide", decide_to_search)
// Python: workflow.add_node("search", search)
// Python: workflow.add_node("reflect", reflect)
// Python: workflow.add_node("generate", generate)
// Python: workflow.set_entry_point("decide")
// Python: workflow.add_conditional_edges("decide", decide_to_search, {
// Python: "search": "search",
// Python: "generate": "generate",
// Python: "generate_direct": "generate"
// Python: })
// Python: workflow.add_edge("search", "reflect")
// Python: workflow.add_conditional_edges("reflect", lambda s: "search" if not s["sufficient"] and s["search_count"] str:
// Python: question = state["question"]
// 简单问题不检索（如打招呼、算术题）
if (is_simple_question(question)) {
return "generate_direct";
// 已有文档是否充分？
if (state.get("sufficient", false)) {
return "generate";
// 最多检索 3 轮，防止死循环
if (state.get("search_count", 0) >= 3) {
return "generate";
return "search";
// Python: def search(state: RAGState) -> RAGState:
// Python: question = state["question"]
// Python: count = state.get("search_count", 0)
// 第 2 轮起，根据反思结果改写 Query
if (count > 0) {
// Python: question = llm_rewrite_query(question, state.get("reflection", ""))
// Python: docs = hybrid_search(question, top_k=10)
// Python: ranked_docs = reranker.rerank(docs, question, top_n=5)
// Python: state["retrieved_docs"] = state.get("retrieved_docs", []) + ranked_docs
// Python: state["search_count"] = count + 1
return state;
// Python: def reflect(state: RAGState) -> RAGState:
// Python: prompt = f"""问题：{state['question']}
// Python: 已有文档：{state['retrieved_docs']}
// Python: 判断：当前文档是否足以完整回答问题？
// Python: 如果不足，指出缺少什么信息。"""
// Python: result = llm.invoke(prompt)
// Python: state["reflection"] = result
// Python: state["sufficient"] = "信息充分" in result
return state;
// Python: def generate(state: RAGState) -> RAGState:
// Python: context = "\n\n".join(state["retrieved_docs"])
// Python: state["answer"] = llm.invoke(f"基于以下信息回答问题：\n{context}\n\n问题：{state['question']}")
return state;
// 构建工作流图
// Python: workflow = StateGraph(RAGState)
// Python: workflow.add_node("decide", decide_to_search)
// Python: workflow.add_node("search", search)
// Python: workflow.add_node("reflect", reflect)
// Python: workflow.add_node("generate", generate)
// Python: workflow.set_entry_point("decide")
// Python: workflow.add_conditional_edges("decide", decide_to_search, {
// Python: "search": "search",
// Python: "generate": "generate",
// Python: "generate_direct": "generate"
// Python: })
// Python: workflow.add_edge("search", "reflect")
// Python: workflow.add_conditional_edges("reflect", lambda s: "search" if not s["sufficient"] and s["search_count"] < 3 else "generate")
// Python: workflow.add_edge("generate", END)
// Python: agent_rag = workflow.compile()
// 用户："你好" → decide 判断简单问题 → 直接生成
// 用户："我们系统的限流策略和降级策略分别是什么？" →
// 第1轮检索限流策略 → 反思：缺少降级策略 →
// 第2轮检索降级策略 → 反思：信息充分 → 生成完整回答
}
}
20.8.3 Agentic RAG 什么时候触发检索？
这是面试高频追问点。Agent 通过以下信号判断是否触发检索： | 触发信号 | 说明 | 示例 | | --- | --- | --- | | 问题类型判断 | 事实性问题 → 检索；闲聊/算术 → 跳过 | "今天天气" → 检索；"1+1" → 跳过 | | 知识边界检测 | 问题涉及模型训练截止后的信息 → 检索 | "2025年新发布模型" → 检索 | | 领域关键词匹配 | 问题包含企业专有名词 → 检索内部知识库 | "我们的SLA策略" → 检索 | | 反思触发再检索 | 已有文档不足以回答 → 改写Query重搜 | 第1轮只有限流，缺降级 → 第2轮搜降级 | | 用户显式要求 | 用户要求查证或引用来源 → 检索 | "帮我查一下文档" → 检索 | 📊 Agentic RAG 的关键设计：防止死循环
- 最大检索次数限制：通常设 3 轮，超过则用已有信息生成（宁可答案不完美，也不能死循环）
- 反思质量评估：反思模型判断信息充分性，避免无意义的重复检索
- Query 改写递进：每轮检索改写 Query，避免相同查询返回相同结果
- 成本控制：每轮检索消耗 Token，Agent 需要在质量和成本间权衡
🔥 2026 面试视角：Agentic RAG vs 传统 RAG
**传统 RAG**是"流水线工人"——固定流程，不管实际情况。
**Agentic RAG**是"调研员"——先判断要不要查资料，查了一轮发现不够再查，查到够为止。
**核心区别**：Agentic RAG 引入了**决策环节（是否检索）** + **反思环节（够不够）** + **迭代能力（换角度重搜）**，这是从"工具"升级为**智能体**的关键。
📋 八股总结 — 面试高频考点
Q1: RAG 的原理是什么？和微调有什么区别？
**RAG 原理**：在生成回答前，先从外部知识库检索相关文档，将检索结果作为上下文喂给 LLM 生成回答。
**与微调的区别**：RAG 是"外挂知识库"，不改模型参数，知识实时更新，成本低；微调是"内化知识"，修改模型参数，适合定制风格/格式，成本高。最佳实践是两者结合：微调改善输出风格，RAG 提供最新知识。
Q2: 向量数据库和传统数据库有什么区别？
传统数据库做精确匹配（WHERE id=123），向量数据库做**相似度搜索**（找到最接近的 K 个向量）。向量数据库使用 ANN（近似最近邻）算法索引，如 HNSW、IVF，能在百万级向量中毫秒返回结果。主流方案：Chroma（原型）、Pinecone（云托管）、Milvus（大规模自部署）、pgvector（PostgreSQL 插件）。
Q3: 什么是混合检索？为什么要做重排序？
**混合检索**：BM25（关键词匹配）+ 向量检索（语义匹配）加权融合。BM25 擅长精确关键词（如"GPT-4o"），向量擅长语义（如"大语言模型"）。
**重排序**：向量检索召回快但精度有限，用 Cross-Encoder 对候选集逐一精细打分。两阶段：先快召回 Top-50（毫秒级），再精排序 Top-5（百毫秒级），兼顾速度和精度。
Q4: RAG 系统如何评估？
从两个阶段评估：
**检索质量**：召回率（相关文档是否被检索到）、精确率（检索结果中有多少相关）、MRR（平均倒数排名）。
**生成质量**：忠实度（回答是否基于检索内容，有没有幻觉）、相关性（回答是否切题）、完整性（是否覆盖了用户问题的所有方面）。可用 Ragas 框架自动化评估。
Q5: 如何解决 RAG 检索不到相关文档的问题？
① **Query 改写**：用 LLM 改写用户问题，生成多个角度的查询并行检索；
② **HyDE**：先让 LLM 生成假回答，用假回答去检索（语义更丰富）；
③ **调整分块策略**：chunk_size 太大不精确，太小语义不完整，尝试 512-1024；
④ **混合检索**：加入 BM25 关键词检索，弥补向量检索关键词匹配弱的缺陷；
⑤ **增加召回量**：先召回 Top-20，再用 Reranker 精选 Top-5。
Q6: RAG 检索出的内容相互冲突，Agent 该怎么取舍？
企业知识库文档版本多、来源杂，冲突是常态。三种策略：
**① 元数据加权排序**：根据文档的发布时间、权威等级赋予不同权重，优先采信**高权威 + 最新**的内容。时间维度和权威维度双重保障。
**② 多智能体辩论（Multi-Agent Debate）**：让不同 Agent 分别对应冲突的文档内容，相互对比论证，梳理出矛盾点，筛选出逻辑最通顺的答案。用"辩论"机制替代"单点裁决"。
**③ 强制溯源**：要求 Agent 输出答案必须附带检索来源，方便用户最终校验核对。把最终裁决权交给人类。
面试要点：不能只说"取最新的"，要展示元数据加权+多Agent辩论+溯源三重保障的组合思维。
Q7: 企业 RAG 怎么解决权限隔离问题？会不会泄露涉密数据？
这是企业落地 RAG 的硬性安全要求，面试高频追问。我的解决方案是**RAG权限对齐**：
**存入阶段**：在向量数据库存入数据时，给每一条向量绑定对应的访问控制权限元数据（如部门、角色、保密等级）。
**检索阶段**：用户发起检索请求时，系统自动带入当前用户的身份权限做过滤。
**关键点**：不是在"结果返回后"做过滤（可能已泄露敏感信息到上下文），而是**在向量检索的源头就完成数据隔离**。从根本上杜绝普通用户查到高管敏感数据的问题。
面试要点：强调"源头过滤"而非"事后过滤"，体现对数据安全闭环的理解。
Q8: 面对新闻/股价等实时更新知识库，RAG 怎么适配？
实时性是 RAG 落地的关键挑战。三种机制：
**① 动态路由 Agent**：先判断用户问题是否需要实时数据。需要实时 → 优先调用搜索/实时 API，不检索静态向量库；不需要 → 正常走向量库检索。类似"先思考再行动"的智能分发。
**② 流式增量更新**：通过消息队列监听知识库的实时变动，动态新增/更新向量数据，不用全量重构。增量更新效率高。
**③ 缓存失效机制**：给高频问题设置缓存时效，元数据更新后立刻清空旧缓存，保存答案的时效性，避免返回过期信息。
面试要点：三机制覆盖"判断要不要实时"（路由）、"怎么更新"（增量）、"怎么保证不过期"（缓存失效），体现全链路思维。
Q9: 怎么系统提升 RAG 问答准确率？三层体系是什么？
这是最综合的一道题，需要展示"从解析到检索到生成全链路"的完整体系思维：
**第一层：深度解析层——解决文档拆分错乱**
传统按字符切分会打断表格/标题/正文的关联。我用**布局感知解析模型**，把文档识别成标题、正文、表格、列表等完整模块，**按标题层级切块**，保证每一段内容语义完整不会断裂。核心思路：先理解文档结构，再按结构切。
**第二层：检索增强层——多阶段精准检索**
① **向量检索 + BM25 关键词检索混合模式**：兼顾语义匹配和专有名词匹配。单一向量检索会丢掉精确关键词匹配能力。
② **重排序模型对初筛内容精排**：这是提升准确率**性价比最高**的操作。初筛粗选 → 精排精选，二次提纯。
③ **扩展 Query 生成多个同义问题并行检索**：解决用户提问太简短、信息不全的问题。一个问题变成多个角度同时检索。
**第三层：生成校验层——实现自我纠错**
答案生成前，Agent 做两次自检：
**第一检**：判断检索内容是否足够回答问题 → 不足就重新检索。
**第二检**：核对答案内容是否全部来自检索结果 → 杜绝无依据的幻觉输出。
面试要点：三层体系覆盖解析（输入端）、检索（中间端）、生成（输出端），体现全链路精度提升而非单点优化。
Q10: GraphRAG 和向量 RAG 的区别是什么？各自适合什么场景？
**核心区别在检索方式**：向量 RAG 做语义相似度搜索，检索的是"像"的文本块；GraphRAG 从文档中抽取实体和关系构建知识图谱，检索时做图遍历，找到的是"有关联"的信息。
**向量 RAG 擅长**：事实性问答（"RAG 是什么？"），单点查询，毫秒级响应，构建成本低。
**GraphRAG 擅长**：多跳推理（"A 公司 CEO 毕业于哪所大学？"需要跨文档追溯），关系分析（"这个故障影响哪些下游服务？"），全局摘要，跨文档关联。
**GraphRAG 的代价**：构建成本高（LLM 抽取实体关系消耗大量 Token），维护复杂（文档更新需同步更新图谱），有实体歧义问题（"苹果"是公司还是水果）。
**最佳实践**：混合使用。向量 RAG 做日常问答（90% 场景），GraphRAG 做复杂关联分析（10% 场景）。面试要点是展示"不是 GraphRAG 替代向量 RAG，而是互补"的思维。
Q11: Agentic RAG 什么时候触发检索？和传统 RAG pipeline 有什么本质区别？
**本质区别**：传统 RAG 是固定流水线（Query → Retrieve → Generate），每次提问都检索且只检索一次；Agentic RAG 引入 Agent 做动态决策，具备三个传统 RAG 没有的能力：
**① 决策能力（是否检索）**：Agent 判断问题类型——简单问题（"你好"、"1+1"）直接回答不检索，事实性问题才触发检索。触发信号包括：问题类型判断、知识边界检测（超出训练数据）、领域关键词匹配、用户显式要求。
**② 反思能力（够不够）**：每轮检索后，Agent 评估已有文档是否足以完整回答问题。不足则指出缺什么，触发再检索。
**③ 迭代能力（换角度重搜）**：根据反思结果改写 Query，换角度重新检索。比如第 1 轮搜"限流策略"，反思发现还缺"降级策略"，第 2 轮改搜"降级策略"。
**防止死循环**：设最大检索次数（通常 3 轮），超过则用已有信息生成。面试要点：强调"决策+反思+迭代"三重能力，这是从"工具"升级为"智能体"的关键。
评论
0 条
还没有评论，先写一条吧。