Text2Cypher 与 LLM 图查询 Agent：从自然语言到可执行 Cypher

用户问题（自然语言）
   ↓ 提示 + schema + few-shot
LLM 生成 Cypher
   ↓ 校验（语法 / 语义 / 白名单）
执行（只读沙箱） → 组织答案

1. 词汇鸿沟：用户说「供应商」，图里叫 :Vendor
2. 结构鸿沟：用户说「关联方」，图里是 3 跳路径
3. 度量鸿沟：「大额」在图里是 amount > 100000
4. 时间鸿沟：「上个月」要翻译成日期区间谓词
→ 前两个靠 schema 提示，后两个靠业务词典

- 模型没见过你的 schema → 编造标签/属性名
- 模型不知道数据分布 → 生成的查询可能全表扫描
- 模型不理解安全边界 → 可能生成写操作
- 模型不保证语法 → Cypher 方言版本差异会翻车
→ Text2Cypher = 模型生成 + 工程约束

- 简单单跳查询（找某实体的属性）：准确率可到 90% 以上
- 中等多跳 / 聚合：70%~85%，需要校验兜底
- 复杂时序 + 多约束：50%~70%，建议人工确认
→ 不要承诺「100% 准确」，要设计「错了也能兜住」的链路

- 给太少：模型不知道有哪些标签和关系 → 编造
- 给太多：token 爆掉、关键信息被淹没 → 精度下降
→ 目标：用最小 token 表达「可查询的结构」

形式 1（紧凑 DSL）：
(:Person {id, name, age})-[:FRIEND]->(:Person)
(:Company {id, name})-[:SUPPLIES {amount, at}]->(:Company)

形式 2（JSON Schema）：
{"nodes":{"Person":{"props":["id","name"]}}, ...}

形式 3（样例查询）：
MATCH (p:Person)-[:FRIEND]->(f) RETURN f.name
→ 紧凑 DSL 最省 token，JSON 最结构化，样例最直观

- 只保留高频属性（覆盖率 > 5%），低频属性省略
- 枚举型属性给 3~5 个样例值（如 status: active/suspended）
- 敏感属性（手机号、身份证）不出现在提示里
- 标签数 > 100 时按业务域分组，按问题路由子 schema
- schema 提示 300~800 token、few-shot 3~6 条、总输入控制在 2K~4K
→ 提示里的 schema 是「导航地图」，不是「完整字典」

你是图数据库查询专家。下面是图模式：

节点：
(:Person {id, name, age, city})
(:Company {id, name, industry})
(:Account {id, balance})

关系：
(:Person)-[:WORKS_AT {since}]->(:Company)
(:Person)-[:FRIEND]->(:Person)
(:Account)-[:TRANSFER {amount, at}]->(:Account)
(:Person)-[:OWNS]->(:Account)

规则：
1. 只生成只读查询（MATCH / RETURN / WITH）
2. 禁止 CREATE / MERGE / SET / DELETE / DROP
3. 结果集必须带 LIMIT（默认 100）
4. 时间用 datetime('2026-09-01T00:00:00Z') 形式

- 教模型「你的图里这类问题怎么写」
- 覆盖：多跳路径、聚合、时间过滤、排序取 TopN
- 比自然语言规则更有效（模型模仿示例 > 理解规则）
→ 示例质量 > 示例数量，3~6 条精选胜过 20 条泛泛

{
  "question": "查张三的朋友里在北京的",
  "cypher": "MATCH (p:Person {name:'张三'})-[:FRIEND]->(f:Person) WHERE f.city='北京' RETURN f.name LIMIT 100",
  "tags": ["单跳", "属性过滤"],
  "verified": true
}

静态：固定 5 条塞进提示
  - 优点：简单、稳定；缺点：与当前问题无关时是噪声
检索式：按用户问题检索最相似的 K 条示例
  - 向量检索（问题 embedding 相似）或按标签/意图路由
  - 优点：示例相关性高，准确率显著提升
→ 生产推荐检索式，示例库可持续积累

def build_prompt(question, schema, store, k=4):
    hits = store.search(question, k=k)          # 语义检索
    shots = [e for e in hits if e["verified"]]
    shots = ensure_coverage(shots, ["多跳", "聚合"])  # 保证结构多样
    return TEMPLATE.format(schema=schema, examples=render(shots), question=question)

- 与目标问题「结构相似」优于「字面相似」
- 覆盖难点模式：变长路径、聚合分组、时间窗口
- 示例里的标签属性必须与当前 schema 一致（防过时）
- 可保留 1 条「反面示例」说明禁止的写法
→ 示例库要版本化，schema 变更时同步失效

- 模型输出不可信：可能语法错、可能语义错、可能危险
- 直接丢给数据库 = 把数据库暴露给概率模型
- 校验层是「模型世界」与「数据库世界」之间的防火墙
→ 校验不是可选项，是上线的前提

闸门 1 语法：能否被 Cypher 解析器解析（EXPLAIN 编译，不执行）
闸门 2 语义：标签 / 关系 / 属性是否在 schema 白名单内
闸门 3 安全：是否只读、是否有 LIMIT、是否触碰敏感属性
→ 三道全过才允许执行

def syntax_check(driver, cypher):
    try:
        with driver.session() as s:
            s.run("EXPLAIN " + cypher).consume()
        return True, None
    except Exception as e:
        return False, str(e)

import re
ALLOWED_LABELS = {"Person", "Company", "Account"}
ALLOWED_RELS = {"WORKS_AT", "FRIEND", "TRANSFER", "OWNS"}
ALLOWED_PROPS = {"id", "name", "age", "city", "industry", "balance", "amount", "at", "since"}

def semantic_check(cypher):
    errs = []
    for lab in re.findall(r':\s*([A-Za-z_]\w*)', cypher):
        if lab not in ALLOWED_LABELS and lab not in ALLOWED_RELS:
            errs.append(f"未知标签或关系: {lab}")
    for prop in re.findall(r'\.([A-Za-z_]\w*)', cypher):
        if prop not in ALLOWED_PROPS:
            errs.append(f"未知属性: {prop}")
    return errs

FORBIDDEN = re.compile(
    r'\b(CREATE|MERGE|SET|DELETE|DETACH|REMOVE|DROP|FOREACH|LOAD\s+CSV|'
    r'CALL\s+\{?db\.|apoc\.(create|merge|refactor|periodic|do\.))\b', re.IGNORECASE)

def security_check(cypher):
    errs = []
    if FORBIDDEN.search(cypher):
        errs.append("包含禁止的写操作或高危过程")
    if not re.search(r'\bLIMIT\b', cypher, re.IGNORECASE):
        cypher = cypher.rstrip().rstrip(';') + "\nLIMIT 100"   # 自动补齐
    return errs

- 校验层是「静态」的，正则可能被绕过
- 模型可能生成资源消耗型查询（笛卡尔积、无上界变长路径）
- 生产库上直接跑 LLM 生成的查询 = 高风险
→ 沙箱是「最后一道物理防线」

1. 账号层：只读账号，无写权限（数据库层兜底）
2. 查询层：强制超时 + 强制 LIMIT + 只读事务
3. 资源层：独立实例 / 只读副本，与写入实例隔离
4. 网络层：从副本读，主库不受影响
→ 层层递进，任何一层被绕过都还有下一层

def run_safe(driver, cypher, timeout_s=10, max_rows=1000):
    with driver.session(default_access_mode="READ") as s:   # 写操作直接报错
        result = s.run("CALL apoc.cypher.runTimeboxed($q, {}, $ms)",
                       q=cypher, ms=timeout_s * 1000)
        return result.data()[:max_rows]

形态 A：只读副本（写入集群 → 复制 → 只读副本 ← LLM 查询）
  优点：完全隔离、成本低；缺点：有复制延迟
形态 B：独立分析实例（定期快照导入，LLM 只查分析实例）
  优点：物理隔离最彻底；缺点：数据非实时
形态 C：同实例只读角色（限制权限 + 超时）
  优点：实时；缺点：仍有资源竞争风险
→ 生产优先 A 或 B，C 只用于低风险内部工具

- 强制 LIMIT（校验层已加，沙箱再兜底）
- 大字段截断（长文本属性只返回前 500 字符）
- 敏感属性返回前脱敏（手机号打码）
- 返回行数超过阈值即告警
→ 保护数据库，也保护下游（避免海量结果灌给模型）

- 模型看不到数据库的真实报错（除非你回灌）
- 报错信息太原始（"Invalid input ')'"）模型看不懂
- 没有「修正方向」时，模型倾向于原样重试
→ 自愈的关键：把错误翻译成模型能懂的话 + 给出修正方向

def text2cypher_with_retry(question, schema, llm, driver, max_attempts=3):
    cypher = llm.generate(question, schema)
    for _ in range(max_attempts):
        ok, err = syntax_check(driver, cypher)
        errs = [] if ok else [err]
        if ok:
            errs = semantic_check(cypher) + security_check(cypher)
        if not errs:
            rows = run_safe(driver, cypher)
            if rows:
                return cypher, rows
            errs = ["查询返回 0 行，请放宽条件"]
        cypher = llm.repair(question, schema, cypher, errs)   # 回灌修复
    return None, None   # 放弃，走兜底模板

你上一次生成的 Cypher 无法执行：

Cypher: {cypher}
错误: {errors}
可用标签: {labels} / 可用关系: {rels} / 可用属性: {props}

请只输出修正后的 Cypher，不要解释。

- 硬上限：最多 2~3 次重试
- 相同错误重复出现 → 立即放弃（模型卡住了）
- 重试预算：总耗时超阈值就降级（返回「无法回答」）
- 降级路径：模板查询兜底（预置常见问题的固定查询）
- 空结果：先判断「条件是否过严」（去掉一个 WHERE 再试）
  放宽后仍空 → 返回「未找到匹配数据」，绝不硬造答案
→ 自愈不是万能，要有「放弃」和「兜底」

Text2Cypher：精确的结构化查询（谁是谁的朋友、转账总额）
GraphRAG：语义检索 + 图扩展（相关背景、多跳证据）
→ 不是二选一，而是「精确问句走 Text2Cypher，开放问句走 GraphRAG」

走 Text2Cypher：问题里有明确实体/关系/聚合意图
  例：「A 公司有哪些供应商」「张三的账户余额」
走 GraphRAG：问题开放、需要背景知识
  例：「A 公司有哪些潜在风险」
混合：先 GraphRAG 召回相关子图 → 再 Text2Cypher 精确计算
→ 用一个轻量分类器或让 LLM 自己选工具

def answer(question):
    intent = classify(question)          # cypher / rag / hybrid
    if intent == "cypher":
        return text2cypher_with_retry(question, schema, llm, driver)
    if intent == "rag":
        return graphrag_answer(question)
    anchors = vector_search(question, top_k=5)     # hybrid
    subgraph = expand_subgraph(anchors, hops=2)    # 2 跳内子图
    return llm.answer(question, context=serialize(subgraph))

场景 1：Text2Cypher 生成失败 → 用 GraphRAG 兜底给「近似答案」
场景 2：结果需要解释 → 用 GraphRAG 召回路径作为「证据链」
→ 两者互为兜底与增强

- 实体链接：问题里的「A 公司」能否准确映射到图里的节点
- 子图规模：2 跳内可能已很大，要按相关性剪枝
- 序列化格式：紧凑 DSL 比自然语言描述更省 token
→ 实体链接是 GraphRAG 准确率的第一瓶颈

单次：问题 → Cypher → 答案（一步到位）
Agent：问题 → 思考 → 调工具 → 观察 → 再思考 → 答案
→ 复杂问题需要多步：先查实体 ID，再查关系，再聚合

tool: get_schema()          → 返回当前 schema（含域路由）
tool: search_entity(name)   → 实体链接，返回候选节点与 ID
tool: run_cypher(query)     → 执行校验后的只读查询
tool: vector_search(text)   → 向量检索节点 / 文档
tool: expand(node_id, hops) → 扩展子图
→ 工具要「窄而清晰」，每个工具职责单一

Thought: 需要先找到「A 公司」的节点 ID
Action: search_entity("A 公司")
Observation: [{id:"c_1024", name:"A 公司", score:0.97}]
Thought: 有了 ID，查它的供应商及供应商的风险等级
Action: run_cypher("MATCH (c:Company {id:'c_1024'})<-[:SUPPLIES]-(s:Company) RETURN s.name, s.risk LIMIT 100")
Observation: [{s.name:"B", s.risk:"高"}, ...]
Answer: ...

- run_cypher 的输入必须过校验层（复用第 4 节的三道闸门）
- 每步工具调用有超时，总步数有上限（如 6 步）
- 观察结果要截断（避免把大结果集塞回上下文）
- 工具报错作为 Observation 回灌，让 Agent 自我修正
- 每步 Observation 只保留必要摘要；历史超 N 步做摘要压缩
→ Agent 自由度越高，约束必须越硬；上下文膨胀是延迟与成本主因

适用：需要先消歧实体、再按结果决定下一步的探索式问题
不适用：单跳事实查询（多步只会增加延迟与出错面）
→ 简单问题走单次生成，复杂问题才上 Agent

1. 查询级：生成的 Cypher 是否可执行（执行率）
2. 结果级：执行结果是否与标准答案一致（结果准确率）
3. 端到端：最终答案是否被用户接受（人工 / LLM 评审）
→ 只测第 1 层会「语法全对但结果全错」

- 从真实问题日志采样（覆盖高频意图）
- 每条包含：问题 + 标准 Cypher + 期望结果
- 分层：单跳 / 多跳 / 聚合 / 时间 / 组合
- 规模：200~500 条即可覆盖主要失效模式
- 匹配方式：不比字符串，比「执行结果集」（顺序无关）

def exec_match(pred, gold, driver):
    p = set(map(frozenset, run_safe(driver, pred)))
    g = set(map(frozenset, run_safe(driver, gold)))
    return p == g

运维：全量日志（问题 / Cypher / 校验结果 / 耗时）
      失败样本回流待标注池，人工修正后进示例库
      灰度：新提示 / 新模型先小流量对比，指标不降才全量
      监控：执行率、空结果率、超时率、用户追问率
产品：展示生成的 Cypher（可解释、可信任）
      结果可一键转可视化图；支持「改一句再问」的交互
      明确标注「AI 生成，仅供参考」
→ 上线不是终点，是持续迭代的起点；透明 + 可纠正优于假装全对

总延迟 = LLM 生成（1~5s）+ 校验（<50ms）+ 执行（10ms~10s）+ 答案组织（1~3s）
- LLM 生成是大头：用小模型做简单意图、流式输出降低体感延迟
- 执行可能失控：靠索引 + LIMIT + 超时控制
- 多步 Agent 延迟翻倍：限制步数
成本 = 输入 token（schema + few-shot + 问题）+ 输出 token + 每步 Agent 输入
降本三招：schema 精简、few-shot 缓存（prompt caching）、简单问题走小模型
→ 优化优先级：执行正确性 > 生成延迟 > 答案润色

坑 1：schema 硬编码在提示里 → 图演进后模型仍用旧标签
坑 2：不做校验直接执行 → 一次误生成的写操作毁库
坑 3：不强制 LIMIT → 模型生成全表返回，打爆内存
坑 4：把原始报错直接回灌 → 模型看不懂，反复犯同一错
坑 5：空结果当「没有数据」返回 → 实际是查询条件写错
坑 6：few-shot 示例过时 → 示例里的标签已不存在，误导模型
坑 7：多步 Agent 无步数上限 → 无限循环烧钱
坑 8：结果集直接塞回模型 → 上下文爆炸
坑 9：评测只看生成不看结果 → 上线后发现结果大面积错
坑 10：无兜底路径 → 生成失败时用户体验断崖

[ ] schema 提示自动生成（不硬编码）
[ ] 三道校验闸门齐全（语法 / 语义 / 安全）
[ ] 只读账号 + 超时 + 强制 LIMIT
[ ] 错误自愈有硬上限与兜底模板
[ ] 全量日志与失败样本回流
[ ] 评测集覆盖五类意图
[ ] 灰度发布与指标对比
[ ] 用户侧展示 Cypher 并可纠正