GraphRAG 实战：从向量检索到知识图谱增强检索

知识图谱 = 节点（实体）+ 边（关系）+ 属性（标签/值）

示例（保险理赔场景）：
  (客户:Person {name:"王强", age:42})
  (保单:Policy {id:"P-8842", amount:500000})
  (险种:Product {name:"重疾险"})
  (王强)-[:购买]->(P-8842)
  (P-8842)-[:属于]->(重疾险)
  (王强)-[:申请]->(理赔:Claim {id:"C-331", status:"审核中"})

# graph_builder.py — LLM 抽取实体关系三元组
import json
from typing import List, Dict

EXTRACT_SYSTEM = """你是知识图谱构建助手。从给定文本中抽取实体与关系，
输出 JSON：{"entities": [{"name", "type", "description"}],
            "relations": [{"from", "to", "type", "description"}]}
规则：
- 实体类型限：Person/Organization/Product/Event/Location/Concept
- 关系类型限：购买/属于/参与/导致/位于/服务/竞争
- 只抽取文本明确表达的信息，不要臆造
"""

def extract_triples(text: str, llm_call) -> dict:
    raw = llm_call(EXTRACT_SYSTEM, f"文本：\n{text}")
    return json.loads(raw)


def build_graph(documents: List[str], llm_call, neo4j_driver):
    """批量抽取并写入图谱，带实体消歧。"""
    with neo4j_driver.session() as session:
        for doc in documents:
            triples = extract_triples(doc, llm_call)
            for ent in triples["entities"]:
                # MERGE 保证同名实体合并，避免重复节点
                session.run(
                    "MERGE (e:{type} {{name:$name, description:$desc}})"
                    .format(type=ent["type"]),
                    name=ent["name"], desc=ent.get("description", ""))
            for rel in triples["relations"]:
                session.run(
                    """MATCH (a {name:$from}), (b {name:$to})
                       MERGE (a)-[:{type} {{description:$desc}}]->(b)"""
                    .format(type=rel["type"]),
                    from=rel["from"], to=rel["to"], desc=rel.get("description",""))

// 抽取结果示例（保险条款文本）
{
  "entities": [
    {"name": "王强", "type": "Person", "description": "投保人"},
    {"name": "意外险A", "type": "Product", "description": "一年期意外险"}
  ],
  "relations": [
    {"from": "王强", "to": "意外险A", "type": "购买",
     "description": "2025-03-01 投保，保额 50 万"},
    {"from": "意外险A", "to": "高风险运动免责", "type": "属于",
     "description": "含免责条款"}
  ]
}

def disambiguate_entities(entities: List[dict], existing_aliases: dict,
                          embed_fn) -> List[str]:
    """将抽取实体映射到既有规范名。alias 表：别名 → 规范名。"""
    resolved = []
    for ent in entities:
        name = ent["name"]
        # 1. 精确别名表
        if name in existing_aliases:
            resolved.append(existing_aliases[name])
            continue
        # 2. 向量相似度匹配（阈值 0.92 才合并）
        best, best_score = find_similar_entity(name, embed_fn, existing_aliases)
        if best and best_score >= 0.92:
            resolved.append(best)
        else:
            resolved.append(name)   # 新实体，保留原名
    return resolved

def incremental_update(driver, changed_docs: List[str], llm_call,
                       doc_ids: List[str]):
    """按文档级增量更新：先删旧子图再重建，保持一致性。"""
    with driver.session() as session:
        for doc_id in changed_docs:
            # 1. 删除该文档对应的旧子图（通过文档锚点节点）
            session.run(
                """MATCH (d:Document {id:$doc_id})-[r]-(e)
                   DETACH DELETE d, r""", doc_id=doc_id)
            # 2. 重建该文档的三元组
            triples = extract_triples(load_doc(doc_id), llm_call)
            write_triples(session, doc_id, triples)

def local_graph_search(driver, entity_name: str, hops: int = 2) -> list[dict]:
    """局部检索：从实体出发做 BFS 遍历，返回邻域路径。"""
    query = """
        MATCH (e)-[r*1..{hops}]-(n)
        WHERE e.name = $name
        RETURN collect(DISTINCT e.name + ' --' + head(r).type + '--> ' + n.name)
               AS paths
        LIMIT 200
    """.format(hops=hops)
    with driver.session() as session:
        result = session.run(query, name=entity_name).single()
        return [{"path": p} for p in (result["paths"] if result else [])]

def answer_with_local_graph(question: str, driver, llm_call, entity_extractor):
    """局部 GraphRAG 问答：先抽取问题中的实体，再查图谱，最后让 LLM 组织回答。"""
    # 1. 从问题抽取核心实体
    entities = entity_extractor(question)
    if not entities:
        return fallback_to_vector_rag(question)

    # 2. 图谱遍历，收集邻域
    graph_evidence = []
    for ent in entities:
        graph_evidence += local_graph_search(driver, ent["name"])

    # 3. 图谱证据 + 问题 → LLM
    evidence_text = "\n".join(g["path"] for g in graph_evidence[:50])
    return llm_call(
        f"根据以下知识图谱中的事实回答用户问题：\n{evidence_text}\n\n问题：{question}",
        system="回答要基于给定的图谱事实，不要臆造关系。")

全局检索流程：
文档 → 实体图 → Leiden 社区检测（分层） → 每层社区生成摘要
→ 全局问题时，检索相关社区摘要 → LLM 综合回答

# global_search.py — 基于社区摘要的全局检索
import pandas as pd

def detect_communities(driver) -> list[dict]:
    """用图算法做社区检测（Neo4j GDS 的 Leiden/Louvain）。"""
    query = """
        CALL gds.leiden.stream({nodeProjection:'*', relationshipProjection:'*'})
        YIELD nodeId, communityId
        RETURN communityId, count(*) AS size
        ORDER BY size DESC
    """
    with driver.session() as session:
        return [dict(r) for r in session.run(query)]

def summarize_community(driver, community_entities, llm_call) -> str:
    """对社区内的实体与关系生成主题摘要，供全局问答检索。"""
    # 取出社区子图，让 LLM 总结主题
    subgraph = extract_subgraph(driver, community_entities)
    return llm_call(
        f"总结以下实体的共同主题与关键关系：\n{subgraph}",
        system="输出 2-3 句话的主题摘要。")

def global_answer(question: str, community_summaries: list[str],
                  llm_call) -> str:
    """基于社区摘要回答全局性问题。"""
    matched = rank_communities(question, community_summaries, embed_fn)
    top_summaries = "\n\n".join(matched[:5])
    return llm_call(
        f"问题：{question}\n\n相关资料摘要：\n{top_summaries}",
        system="综合多份摘要回答，标注信息来自哪些主题。")

模式 A：先向量后图谱（级联）
  问题 → 向量 Top-K 文档 → 抽取实体 → 图谱遍历补充关系
  │ 适用：文档为主、关系辅助

模式 B：先图谱后向量（级联）
  问题 → 抽取实体 → 图谱定位 → 缺失信息向量补全
  │ 适用：实体明确的领域

模式 C：并行融合（重排）
  问题 → 向量结果 + 图谱结果 → 统一评分器 → 合成排序
  │ 适用：综合场景，质量最好但开销最大

# hybrid_retriever.py — 向量 + 图谱并行检索后融合
def hybrid_retrieve(question: str, vector_store, graph_driver,
                    embed_fn, entity_extractor,
                    alpha: float = 0.6, top_k: int = 10) -> list[dict]:
    """
    alpha 控制向量与图谱的权重配比。
    返回融合排序后的证据列表。
    """
    # 向量侧
    vector_hits = vector_store.search(
        embed_fn(question), top_k=top_k,
        payload=True)   # 每项含 score 0-1

    # 图谱侧：抽取实体后做局部遍历
    graph_hits = []
    entities = entity_extractor(question)
    for ent in entities:
        paths = local_graph_search(graph_driver, ent["name"], hops=2)
        for p in paths:
            graph_hits.append({"content": p["path"],
                               "score": estimate_relevance(p, question)})

    # 融合：归一化分数，按 alpha 加权排序
    merged = []
    for hit in vector_hits:
        merged.append({"content": hit["content"],
                       "score": alpha * hit["score"],
                       "source": "vector"})
    for hit in graph_hits:
        merged.append({"content": hit["content"],
                       "score": (1 - alpha) * hit["score"],
                       "source": "graph"})
    merged.sort(key=lambda x: x["score"], reverse=True)
    return merged[:top_k]

def serialize_graph_evidence(paths: list[dict]) -> str:
    """把图谱路径转成 LLM 可读的事实句。"""
    facts = []
    for p in paths:
        # "王强 --购买--> 意外险A" → "王强购买了意外险A"
        parts = p["path"].split("-->")
        if len(parts) == 2:
            facts.append(f"{parts[0].strip()} {verbalize(parts[1].strip())}")
    return "\n".join(facts)

def verbalize(rel_path: str) -> str:
    """关系动词化（简化实现，生产用映射表）。"""
    return {"--购买-->": "购买了", "--属于-->": "属于",
            "--参与-->": "参与了", "--导致-->": "导致",
            "--位于-->": "位于", "--服务-->": "服务"}

# multi_hop_eval.py — 多跳推理评测集
MULTI_HOP_SET = [
    {
        "query": "王强购买的意外险是否覆盖高风险运动？",
        # 需两跳：王强→意外险A→免责条款→高风险运动
        "hops": 3,
        "ground_truth": "不覆盖，高风险运动属于免责范围",
        "expected_path": ["王强", "购买", "意外险A", "免责", "高风险运动"],
    },
    {
        "query": "同时服务 A 公司和 B 公司的供应商有哪些？",
        "hops": 2,
        "ground_truth": "供应商 X",
        "expected_path": ["A公司", "服务", "供应商X", "服务", "B公司"],
    },
]

def eval_multi_hop(graphrag_system, eval_set, judge_fn):
    results = []
    for case in eval_set:
        answer = graphrag_system.answer(case["query"])
        score = judge_fn(case["query"], answer, case["ground_truth"])
        path_correct = verify_path_reconstructed(answer, case["expected_path"])
        results.append({**case, "quality_score": score, "path_correct": path_correct})
    pass_rate = sum(r["path_correct"] for r in results) / len(results)
    return {"pass_rate": pass_rate, "avg_quality": mean(s["quality_score"] for s in results)}

def ablate_rag_variants(question_set, vector_rag, graphrag, hybrid, judge_fn):
    """对比三种检索方案在评测集上的质量与成本。"""
    results = {}
    for name, system in [("vector", vector_rag), ("graph", graphrag),
                         ("hybrid", hybrid)]:
        scores = [judge_fn(q["query"], system.answer(q["query"]),
                           q["ground_truth"]) for q in question_set]
        results[name] = {
            "avg_quality": mean(s for s in scores),
            "multi_hop_pass": multi_hop_pass_rate(system, MULTI_HOP_SET),
            "avg_cost": system.avg_cost_per_query(),
        }
    return results

# 输出示例：
# vector:  avg_quality 0.78 | multi_hop_pass 0.35 | cost 0.011
# graph:   avg_quality 0.74 | multi_hop_pass 0.62 | cost 0.008
# hybrid:  avg_quality 0.86 | multi_hop_pass 0.71 | cost 0.013

def build_with_budget(docs, llm_call, budget_usd, cost_per_doc=0.02):
    """预算内构建：优先处理高价值文档。"""
    affordable = int(budget_usd / cost_per_doc)
    # 按文档价值（点击率、更新频率）排序后截断
    ranked = sorted(docs, key=lambda d: d["priority"], reverse=True)
    selected = ranked[:affordable]
    build_graph(selected, llm_call, driver)
    print(f"本次构建 {len(selected)} 篇文档，剩余 {len(docs)-len(selected)} 篇走纯向量")

# 缓存热门路径，降低重复遍历
class GraphPathCache:
    def __init__(self, redis):
        self.r = redis
    def get(self, entity: str, hops: int) -> list | None:
        cached = self.r.get(f"graph:{entity}:{hops}")
        return json.loads(cached) if cached else None
    def set(self, entity: str, hops: int, paths: list, ttl=3600):
        self.r.setex(f"graph:{entity}:{hops}", ttl, json.dumps(paths))

def sync_check(driver, doc_store, sample_ratio=0.05) -> list[str]:
    """定期抽查：文档中的实体/关系是否在图谱中一致。"""
    inconsistencies = []
    for doc in doc_store.random_sample(sample_ratio):
        fresh = extract_triples(doc["text"], llm_call)
        in_graph = query_triples_in_graph(driver, fresh)
        if not in_graph:
            inconsistencies.append(doc["id"])
    return inconsistencies   # 非空则触发增量重建

# fraud_graph_check.py — 图谱驱动的理赔风控
from neo4j import GraphDatabase

class ClaimFraudChecker:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def check_shared_contact(self, customer_id: str) -> list[dict]:
        """检查该客户是否与其他人共享联系方式（欺诈信号）。"""
        query = """
            MATCH (c:Customer {id:$cid})-[:拥有]->(contact)
            MATCH (other:Customer)-[:拥有]->(contact)
            WHERE other.id <> c.id
            RETURN other.name AS name, contact.value AS shared
        """
        with self.driver.session() as s:
            return [dict(r) for r in s.run(query, cid=customer_id)]

    def check_exclusive_policies(self, customer_id: str) -> list[dict]:
        """检查是否持有互斥险种（销售合规问题）。"""
        query = """
            MATCH (c:Customer {id:$cid})-[:购买]->(:Policy)-[:属于]->(p:Product)
            WITH c, collect(p.name) AS products
            WHERE '定期寿险' IN products AND '终身寿险' IN products
            RETURN c.name AS customer, products
        """
        with self.driver.session() as s:
            return [dict(r) for r in s.run(query, cid=customer_id)]

    def close(self):
        self.driver.close()

import pytest

def test_fraud_check_detects_shared_contact(fixture_graph):
    checker = ClaimFraudChecker(uri="bolt://localhost:7687",
                                user="neo4j", password="test")
    # 构造已知风险数据：王强与李雷共享手机号
    risks = checker.check_shared_contact("customer-wangqiang")
    assert any(r["name"] == "李雷" for r in risks), "应识别共享联系方式"

def test_exclusive_policy_detection(fixture_graph):
    checker = ClaimFraudChecker(...)
    found = checker.check_exclusive_policies("customer-chen")
    assert found, "持有互斥险种应被标记"

def memorize_to_graph(conversation, driver, llm_call):
    """从对话中抽取实体关系，写入 Agent 记忆图谱。"""
    triples = extract_triples(
        json.dumps(conversation, ensure_ascii=False), llm_call)
    write_triples(driver, "agent_memory", triples)

def recall_from_graph(question, driver, llm_call):
    """Agent 回答前，先查记忆图谱补充背景。"""
    entities = extract_entities(question, llm_call)
    return local_graph_search(driver, entities[0]["name"], hops=2)

def plan_with_graph(task: str, domain_graph, llm_call):
    """利用图谱中的实体依赖关系生成执行计划。"""
    # 例如：任务"发布新功能" → 图谱中前置节点（测试→部署）
    entities = extract_entities(task, llm_call)
    deps = []
    for ent in entities:
        deps += query_predecessors(domain_graph, ent["name"])
    return llm_call(
        f"任务：{task}\n依赖关系：{deps}\n请按依赖顺序给出执行步骤。",
        system="步骤必须满足给定的依赖顺序。")