当前位置：首页 > news >正文

GraphRAG实战：demo跑通很容易，为什么联调时权限和日志先翻车？

news 2026/8/16 3:54:52

聊《GraphRAG跑通那天，我才发现前面的学习顺序反了》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

去年做企业知识库项目，RAG pipeline 在本地跑通了，准确率看着不错，就急着上线。结果联调第一天就被打脸——权限系统没打通，用户能搜到不该看的内容；日志完全不可观测，出问题只能盲猜。GraphRAG 方案当时还在 demo 阶段，根本没考虑这些生产级问题。

后来花了两个月补工程化，才真正跑通。今天复盘，想说的不是 GraphRAG 多香，而是：图谱建好了，联调时权限和日志才是真实门槛。

传统 RAG 的瓶颈，不止是检索不准

先说结论：传统 RAG 的核心问题是跨文档关联能力太弱。

我们之前的项目，用户问"张三和李四在哪个项目上有过合作"，基于向量检索的 RAG 基本答不上来。因为问题涉及两个实体之间的关系，而向量检索只能找到相似片段，找不到"关系"本身。

还有一个被忽视的问题：权限隔离。传统 RAG 检索完直接拼接返回，没有考虑不同用户能看哪些文档。联调时才发现，法务部的文档被普通员工搜到了，这个问题在 demo 阶段完全暴露不了。

所以 GraphRAG 的价值，不只是提升准确率，更是为结构化检索和权限控制提供了基础。

知识图谱建模：别一上来就搞复杂 schema

很多教程一上来就讲本体设计、OWL 语义网，实际项目里根本用不上。

我们的做法很简单：先定义核心实体类型和关系类型，够用就行。

# 实体类型定义（精简版） ENTITY_TYPES = { "person": {"fields": ["name", "role", "dept"]}, "project": {"fields": ["name", "status", "budget"]}, "document": {"fields": ["title", "author", "sensitivity"]}, "product": {"fields": ["name", "category"]} } # 关系类型定义 RELATION_TYPES = { "works_on": {"source": "person", "target": "project"}, "owns": {"source": "person", "target": "product"}, "references": {"source": "document", "target": "project"}, "has_access": {"source": "person", "target": "document"} # 权限关系 }

关键点：has_access关系是权限控制的基石。图谱里直接存储"谁能访问哪些文档"，后续检索时自动过滤，比在应用层做权限校验更可靠。

建模时我犯过的错误：一开始设计了 20 多种关系类型，结果抽取质量很差。后来砍到 6 种核心关系，效果反而更好。图谱质量 > 图谱复杂度。

实体关系抽取：别全指望大模型

很多教程说"用 LLM 做信息抽取"，实际跑下来问题很多：

1. 幻觉严重：LLM 会编造不存在的实体和关系
2. 格式不稳定：JSON 经常解析失败
3. 成本高：每份文档都要调 API

我们的解决方案：规则 + 小模型 + 大模型校验。

import spacy from typing import List, Dict, Tuple # 第一步：用规则+小模型做初筛 nlp = spacy.load("zh_core_web_sm") def extract_entities(text: str) -> List[Dict]: """规则抽取实体""" doc = nlp(text) entities = [] # 人名：基于命名实体识别 for ent in doc.ents: if ent.label_ == "PERSON": entities.append({ "text": ent.text, "type": "person", "confidence": 0.8 }) # 项目名：基于关键词匹配 project_keywords = ["项目", "工程", "计划"] for kw in project_keywords: if kw in text: # 简单启发式：关键词前后 10 字 idx = text.find(kw) start = max(0, idx - 10) end = min(len(text), idx + len(kw) + 10) entities.append({ "text": text[start:end], "type": "project", "confidence": 0.5 }) return entities # 第二步：用大模型做关系抽取和校验 async def extract_relations(entities: List[Dict], text: str) -> List[Dict]: """LLM 抽取关系""" prompt = f""" 从以下文本中抽取实体间的关系： 文本：{text} 已识别实体： {entities} 请以 JSON 格式返回关系列表，格式： [{{"source": "实体1", "relation": "关系类型", "target": "实体2"}}] """ response = await call_llm(prompt) return parse_json(response)

关键判断：规则覆盖 80% 的常见情况，LLM 处理 20% 的边缘情况。这样既能控制成本，又能保证质量。

还有一个坑：抽取后的实体需要去重和合并。同一个人可能有"张三"、"zhang san"、"ZS"等多种写法，需要归一化。我们用了简单的字符串相似度 + 人工确认的方式，效果不错。

图检索增强：查询解析是关键

GraphRAG 的核心是把自然语言查询翻译成图查询。这一步做不好，后面全白搭。

from neo4j import GraphDatabase class GraphRAGRetriever: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def query(self, natural_query: str, user_id: str) -> List[Dict]: """ 核心逻辑： 1. 解析查询，提取实体和关系 2. 生成 Cypher 查询 3. 注入权限过滤 4. 执行并返回结果 """ # 第一步：解析查询 parsed = self.parse_query(natural_query) # 第二步：注入权限条件 cypher = self.build_cypher(parsed, user_id) # 第三步：执行查询 results = self.execute(cypher) # 第四步：返回结果 return self.format_results(results) def build_cypher(self, parsed: Dict, user_id: str) -> str: """构建带权限过滤的 Cypher 查询""" # 基础查询模板 base = """ MATCH (p:Person {name: $person_name}) MATCH (d:Document) WHERE // 权限过滤：用户必须能访问该文档 (p)-[:HAS_ACCESS]->(d) OR EXISTS { MATCH (admin:Person {name: $admin_name}) WHERE admin.dept = 'admin' } RETURN d.title, d.content """ return base

这里有个关键点：权限过滤必须在图查询层面完成，而不是在应用层后处理。原因有两个：

1. 性能：图数据库的 MATCH + WHERE 比应用层过滤快几个数量级
2. 安全性：应用层过滤可以被绕过，图层面过滤更可靠

联调时我们发现，有些查询返回了不该返回的文档，排查发现是权限关系没建全。有些文档的has_access关系缺失，导致任何人都能搜到。图谱数据质量直接决定权限安全。

评估与优化：准确率不是唯一指标

传统 RAG 评估只看准确率，GraphRAG 需要额外关注：

1. 关系覆盖率：图谱中有多少真实关系被正确抽取
2. 权限正确率：用户能否访问到应该能访问的文档，以及不能访问的文档是否被正确过滤
3. 查询响应时间：图查询通常比向量检索慢，需要优化

我们内部有个评估脚本：

def evaluate_graphrag(test_cases: List[Dict]) -> Dict: """ test_cases 格式： [ { "query": "张三能访问哪些文档", "expected": ["doc1", "doc2"], "user_id": "zhangsan" } ] """ results = { "accuracy": 0.0, "permission_correct": 0.0, "avg_response_time": 0.0 } total = len(test_cases) correct = 0 perm_correct = 0 total_time = 0 for case in test_cases: start = time.time() response = retriever.query(case["query"], case["user_id"]) elapsed = time.time() - start total_time += elapsed # 准确率评估 if set(response) == set(case["expected"]): correct += 1 # 权限正确性评估：检查是否有越权访问 if check_permission_safety(response, case["user_id"]): perm_correct += 1 results["accuracy"] = correct / total results["permission_correct"] = perm_correct / total results["avg_response_time"] = total_time / total return results

优化方向：缓存热点查询结果、索引优化（给常用关系建立索引）、查询改写（把模糊查询转成精确查询）。

联调复盘：权限和日志才是真实门槛

回到开头说的联调翻车。复盘下来，问题出在三个地方：

1. 权限边界不清晰

demo 阶段只关注了"能不能搜到"，没关注"谁能搜到什么"。上线后发现，法务部的合同文档被普通员工搜到了，原因是图谱里缺少权限关系，或者权限关系建错了。

教训：权限关系必须和实体关系同等重要，建模时就要考虑。

2. 日志不可观测

查询失败了，不知道是图谱查询失败、权限过滤失败、还是结果解析失败。没有结构化日志，排查成本极高。

教训：每个关键步骤都要有日志，包括查询解析、权限过滤、结果返回。日志要包含 userid、query、resultcount、elapsed_time 等关键字段。

import logging logger = logging.getLogger("graphrag") def query(self, natural_query: str, user_id: str) -> List[Dict]: logger.info(f"Query started: user={user_id}, query={natural_query}") try: parsed = self.parse_query(natural_query) logger.info(f"Query parsed: {parsed}") cypher = self.build_cypher(parsed, user_id) logger.debug(f"Cypher built: {cypher}") results = self.execute(cypher) logger.info(f"Query executed: {len(results)} results") return results except Exception as e: logger.error(f"Query failed: {e}", exc_info=True) raise

3. 责任边界不清晰

权限问题出了，到底是图谱数据的问题、查询逻辑的问题、还是应用层配置的问题？没有埋点，说不清楚。

教训：关键路径要有埋点，能定位到是哪个环节的问题。