RAG 技术深度解析：从基础到 GraphRAG，以及大 Context 时代的思考

用户查询 → 检索相关文档 → 将文档作为上下文 → LLM 生成答案

┌─────────────┐
│  文档语料库  │
└──────┬──────┘
       │ 离线处理
       ▼
┌─────────────┐
│  文档切块    │ (Chunking)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  向量化      │ (Embedding)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  向量数据库  │ (Vector DB)
└──────┬──────┘
       │
       │ 在线查询
       │
┌──────┴──────┐
│  用户查询    │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  查询向量化  │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  相似度检索  │ (Top-K)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  重排序      │ (可选)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  构建 Prompt │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  LLM 生成    │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  返回答案    │
└─────────────┘

def fixed_size_chunking(text, chunk_size=512, overlap=50):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks

# 常用 Embedding 模型
- OpenAI text-embedding-3-large (3072 维)
- Cohere embed-v3
- BGE-large (1024 维)
- text2vec

# 余弦相似度
similarity = cosine_similarity(query_vector, doc_vectors)
top_k_docs = get_top_k(similarity, k=5)

# 结合关键词检索和向量检索
bm25_score = bm25_search(query, documents)
vector_score = vector_search(query_embedding, doc_embeddings)
final_score = alpha * vector_score + (1 - alpha) * bm25_score

# 使用更强的模型对初检结果重新排序
initial_results = vector_search(query, top_k=20)
reranked_results = reranker_model(query, initial_results, top_k=5)

文档 A: "张三是项目经理，负责 X 项目"
文档 B: "X 项目在 2024 年 Q2 完成"
文档 C: "项目经理向李四汇报"

用户问题: "X 项目的负责人向谁汇报？"
传统 RAG: 可能检索到 A 和 B，但缺少 C，无法完整回答

问题: "Claude Sonnet 4.5 的开发公司的 CEO 是谁？"
需要：
1. 找到 Claude Sonnet 4.5 → Anthropic
2. 找到 Anthropic → CEO 是 Dario Amodei

Chunk 1: "...该算法的时间复杂度取决于"
Chunk 2: "输入数据的规模，当 n > 10000 时..."

文档 → 实体/关系抽取 → 知识图谱 → Community 检测 → Community 摘要 → 多层次检索

def build_knowledge_graph(documents):
    entities = []
    relationships = []
    for doc in documents:
        # 使用 LLM 提取实体和关系
        extracted = llm_extract(
            doc,
            prompt="""
            提取文档中的：
            1. 实体（人物、组织、概念、事件等）
            2. 实体之间的关系（类型、方向、强度）
            """,
        )
        entities.extend(extracted.entities)
        relationships.extend(extracted.relationships)

    # 实体消解（Entity Resolution）
    unified_entities = entity_resolution(entities)

    # 构建图
    graph = build_graph(unified_entities, relationships)
    return graph

实体:
- Claude Sonnet 4 (产品)
- Anthropic (公司)
- Dario Amodei (人物)
- LLM (技术概念)

关系:
- Claude Sonnet 4 --[开发商]--> Anthropic
- Dario Amodei --[CEO_OF]--> Anthropic
- Claude Sonnet 4 --[类型]--> LLM
- Anthropic --[专注于]--> AI 安全

communities = leiden_algorithm(graph)

# 示例结果
# Community 1: [Claude, Anthropic, Dario Amodei, AI 安全]
# Community 2: [RAG, 向量数据库, Embedding]
# Community 3: [GraphRAG, 知识图谱, 实体抽取]

def generate_community_summary(community, graph):
    entities = community.entities
    relationships = get_relationships(entities, graph)
    summary = llm_summarize(
        entities,
        relationships,
        prompt="总结这个 Community 的核心主题和关键信息",
    )
    return summary

def local_search(query, graph):
    query_entities = extract_entities(query)
    located_entities = graph.find_entities(query_entities)
    subgraph = graph.get_neighbors(located_entities, hops=2)
    context = subgraph.to_context()
    return llm_generate(query, context)

def global_search(query, communities):
    relevant_communities = retrieve_communities(
        query,
        [c.summary for c in communities],
        top_k=10,
    )
    global_context = combine_summaries(relevant_communities)
    return llm_generate(query, global_context)

问题: "Anthropic CEO 负责的产品支持哪些功能？"

Graph 路径:
Dario Amodei --[CEO_OF]--> Anthropic
            --[开发]--> Claude Sonnet 4
            --[支持]--> [函数调用, 视觉理解, ...]

传统 RAG:
Query → 检索 chunks → 拼接上下文 → 生成

GraphRAG:
Query → 确定相关 Community → 获取 Community 摘要 → 理解全局主题 → 细化检索 → 生成

文档 A: "Claude is an AI assistant"
文档 B: "克劳德是一个 AI 助手"
文档 C: "Claude AI can help with..."

GraphRAG: 将这三个表述统一为同一实体 "Claude"

节点类型：
  Vulnerability (CVE)
  Software / Version
  WeaknessType (CWE)
  AttackVector
  Exploit (PoC)
  Patch

边 / 关系：
  (CVE)-[AFFECTS]->(Software)-[HAS_VERSION]->(Version)
  (CVE)-[CLASSIFIED_AS]->(CWE)
  (CVE)-[HAS_EXPLOIT]->(Exploit)
  (CVE)-[FIXED_BY]->(Patch)
  (CVE)-[CHAIN_WITH]->(CVE)   // 利用链

// 1. 某个版本受哪些高危漏洞影响
MATCH (cve:Vulnerability)-[:AFFECTS]->(:Software {name: "OpenSSL"})
      -[:HAS_VERSION]->(v:VersionRange)
WHERE "1.1.1k" >= v.min_version AND "1.1.1k" <= v.max_version
  AND cve.cvss_score >= 7.0
RETURN cve.id, cve.description, cve.cvss_score
ORDER BY cve.cvss_score DESC;

// 2. 可组合利用的漏洞链（多跳推理）
MATCH path = (cve1:Vulnerability)-[:CHAIN_WITH*1..3]->(cve2:Vulnerability)
WHERE cve1.id = "CVE-2024-XXXX" AND cve2 <> cve1
RETURN path, length(path) AS chain_length
ORDER BY chain_length
LIMIT 10;

// 3. 相似漏洞历史分析（用于预测）
MATCH (cve:Vulnerability {id: "CVE-2024-XXXX"})-[:CLASSIFIED_AS]->(cwe)
      <-[:CLASSIFIED_AS]-(similar:Vulnerability)
WHERE similar.published_date < cve.published_date
RETURN similar.id, similar.description, cwe.name
ORDER BY similar.published_date DESC
LIMIT 20;

class Edge:
    valid_at: datetime          # 事件实际发生的时间
    created_at: datetime        # 数据被系统摄入的时间
    invalidated_at: Optional[datetime]  # 事实失效的时间

# 示例：员工职位变更
Edge(
    source="张三",
    target="技术总监",
    relation="WORKS_AS",
    valid_at="2024-01-01",
    created_at="2024-01-15",       # 15 天后才录入系统
    invalidated_at="2024-06-01",   # 6 月 1 日升职，此关系失效
)

class GraphitiRetrieval:
    def search(self, query, strategy="hybrid"):
        vector_results = self.semantic_search(query)    # 语义搜索 (BGE-m3)
        bm25_results = self.fulltext_search(query)      # 全文搜索 (Lucene)
        graph_results = self.graph_traverse(query)      # 图遍历 (BFS)
        community_results = self.community_search(query)  # Community 搜索

        all_results = vector_results + bm25_results + graph_results
        return self.rerank(all_results, query)

def on_new_document(doc):
    new_facts = extract_facts(doc)
    for fact in new_facts:
        existing = graph.find_fact(fact.subject, fact.predicate)
        if existing:
            graph.update_fact(fact, timestamp=now())
        else:
            graph.add_fact(fact, timestamp=now())
    vector_db.update(doc)

graph.add_relationship(
    source="Claude Sonnet 4",
    target="Anthropic",
    type=auto_infer_relationship_type(source, target, context),
)
# 系统自动聚合相似关系类型
# "developed_by", "created_by", "made_by" → "CREATOR_OF"

传统 RAG:
  ✓ 简单 Q&A
  ✓ 文档检索
  ✓ 静态知识库

GraphRAG:
  ✓ 复杂推理
  ✓ 关系查询
  ✓ 全局理解
  ✗ 实时更新
  ✗ 时序分析

Graphiti:
  ✓ 复杂推理
  ✓ 关系查询
  ✓ 全局理解
  ✓ 实时更新
  ✓ 时序分析
  ✓ 知识演进

def choose_strategy(data_size, query_type, update_frequency):
    # 小规模 + 静态 = Long Context
    if data_size < 200_000 and update_frequency == "low":
        return "long_context"
    # 大规模 = 必须 RAG
    if data_size > 1_000_000:
        return "rag"
    # 需要精确引用 = RAG
    if query_type == "fact_lookup":
        return "rag"
    # 需要全局理解 = Long Context
    if query_type == "global_analysis":
        return "long_context"
    # 中等规模 = 混合
    return "hybrid"

开始
  │
  ├─ 数据规模 < 100K tokens?
  │    ├─ 是 → 考虑 Long Context
  │    │      ├─ 静态文档 → Long Context ✓
  │    │      └─ 动态更新 → 轻量 RAG
  │    │
  │    └─ 否 → 继续评估
  │
  ├─ 数据规模 > 10M tokens?
  │    └─ 是 → 必须 RAG ✓
  │
  ├─ 查询频率 > 1000 次/天?
  │    └─ 是 → RAG (成本考虑) ✓
  │
  ├─ 需要精确引用?
  │    └─ 是 → RAG ✓
  │
  ├─ 需要关系推理?
  │    └─ 是 → GraphRAG ✓
  │
  ├─ 需要时序分析?
  │    └─ 是 → Graphiti ✓
  │
  └─ 其他 → 混合方案