AI 知识图谱指南：GraphRAG、来源与安全查询

{
  "claim_id": "clm_01J...",
  "subject_id": "org:acme",
  "predicate": "HEADQUARTERED_IN",
  "object_id": "place:singapore",
  "source_document_id": "doc_842",
  "source_revision": "sha256:8f1...",
  "source_span": {"start": 418, "end": 476},
  "extractor_version": "relations-2026-08-01",
  "valid_from": "2026-04-01",
  "valid_to": null,
  "tenant_id": "tenant_17",
  "policy_scope": ["research"],
  "state": "candidate"
}

from dataclasses import dataclass
from typing import Optional

ALLOWED_RELATIONS = {"HEADQUARTERED_IN", "OWNS", "DEPENDS_ON"}

@dataclass(frozen=True)
class Claim:
    subject_id: str
    predicate: str
    object_id: str
    source_document_id: str
    source_revision: str
    span_start: int
    span_end: int
    extractor_version: str
    tenant_id: str
    valid_to: Optional[str] = None

def validate_claim(claim: Claim, source_length: int) -> None:
    if claim.predicate not in ALLOWED_RELATIONS:
        raise ValueError("关系类型不在版本化词表中")
    if not (0 <= claim.span_start < claim.span_end <= source_length):
        raise ValueError("来源片段超出指定来源版本")
    required = (
        claim.subject_id,
        claim.object_id,
        claim.source_document_id,
        claim.source_revision,
        claim.extractor_version,
        claim.tenant_id,
    )
    if any(not value.strip() for value in required):
        raise ValueError("主张谱系与租户范围不能为空")

example = Claim(
    subject_id="org:acme",
    predicate="HEADQUARTERED_IN",
    object_id="place:singapore",
    source_document_id="doc_842",
    source_revision="sha256:8f1",
    span_start=10,
    span_end=42,
    extractor_version="relations-v3",
    tenant_id="tenant_17",
)
validate_claim(example, source_length=100)
print("valid")
# 输出：valid

from dataclasses import dataclass
from enum import Enum
from typing import Any

class Intent(str, Enum):
    ENTITY_PROFILE = "entity_profile"
    DEPENDENCY_NEIGHBORS = "dependency_neighbors"

@dataclass(frozen=True)
class QueryPlan:
    cypher: str
    parameters: dict[str, Any]

TEMPLATES = {
    Intent.ENTITY_PROFILE: """
        MATCH (e:Entity {tenant_id: $tenant_id, id: $entity_id})
        WHERE e.policy_scope IN $allowed_scopes
        RETURN e.id AS id, e.name AS name, e.type AS type
        LIMIT 1
    """,
    Intent.DEPENDENCY_NEIGHBORS: """
        MATCH (e:Entity {tenant_id: $tenant_id, id: $entity_id})
              -[r:DEPENDS_ON]->(n:Entity {tenant_id: $tenant_id})
        WHERE e.policy_scope IN $allowed_scopes
          AND n.policy_scope IN $allowed_scopes
        RETURN n.id AS id, n.name AS name, r.claim_id AS claim_id
        ORDER BY n.id
        LIMIT $row_limit
    """,
}

def build_plan(
    intent: Intent,
    *,
    entity_id: str,
    authoritative_tenant_id: str,
    authoritative_scopes: list[str],
    requested_limit: int = 25,
) -> QueryPlan:
    if not entity_id or len(entity_id) > 128:
        raise ValueError("实体标识不合法")
    if not authoritative_scopes:
        raise PermissionError("调用方没有已授权策略范围")
    return QueryPlan(
        cypher=TEMPLATES[intent],
        parameters={
            "entity_id": entity_id,
            "tenant_id": authoritative_tenant_id,
            "allowed_scopes": authoritative_scopes,
            "row_limit": min(max(requested_limit, 1), 100),
        },
    )

plan = build_plan(
    Intent.DEPENDENCY_NEIGHBORS,
    entity_id="service:checkout",
    authoritative_tenant_id="tenant_17",
    authoritative_scopes=["operations"],
    requested_limit=500,
)
print(plan.parameters["row_limit"])
# 输出：100

def retrieval_metrics(
    required_evidence: set[str],
    retrieved_evidence: list[str],
    cited_evidence: list[str],
) -> dict[str, float]:
    retrieved = set(retrieved_evidence)
    cited = set(cited_evidence)
    recall = len(required_evidence & retrieved) / max(len(required_evidence), 1)
    citation_precision = len(cited & retrieved) / max(len(cited), 1)
    return {
        "evidence_recall": round(recall, 3),
        "citation_precision": round(citation_precision, 3),
    }

print(retrieval_metrics(
    required_evidence={"span:A", "span:B"},
    retrieved_evidence=["span:A", "span:C"],
    cited_evidence=["span:A", "span:X"],
))
# 输出：{'evidence_recall': 0.5, 'citation_precision': 0.5}

来源版本
  -> 文本分段
  -> 候选主张
  -> 消歧实体
  -> 已接受图快照
  -> 关键词/向量索引
  -> 社区划分
  -> 生成摘要
  -> 检索与答案缓存