"""Cypher 合法性检查节点 输入: state['cypher_query']: List[str] 上游生成的 cypher 语句 输出: state['is_all_validate_cypher']: bool 是否全部合法 为什么生成完还要再查一遍：`generate_neo4j_cypher_node` 里的过滤只看"是不是只读"， 拦不住"语法对、但标签/属性图里没有"这类语句 —— 它们能正常执行，只是恒返回空行， 下游会当成"图里没这个知识"，比报错更难查。所以这里做三层检查： 1. 静态只读：复用生成节点的 FORBIDDEN_KEYWORDS，同一份白名单，避免两处漂移 2. 对 schema 核账：标签、关系类型、`变量.属性` 都要在 tcm_metadata.json 里存在 3. Neo4j EXPLAIN：只编译不执行，把语法错误在打库之前拦掉 第 3 层要连库，连不上时只跳过这一层（前两层的结论照样有效），不让整个节点挂掉。 """ import json import re from typing import Dict, List, Set, Tuple from _004_langgraph_more_nodes.agent_state import AgentState from _004_langgraph_more_nodes.nodes.generate_neo4j_cypher_node import ( FORBIDDEN_KEYWORDS, ) from common.config import Config conf = Config() # 空的 cypher_query 判 False。空集合"全部合法"在逻辑上成立，但这个标志位是给路由用的： # 没有语句可执行时，必须走兜底分支而不是查库分支，判 True 会让下游拿着空列表白跑一趟。 EMPTY_IS_VALID = False # ====================== # schema：从 tcm_metadata.json 读出标签 / 关系 / 属性 # ====================== def _load_schema() -> Tuple[Set[str], Set[str], Set[str]]: """返回 (标签集合, 关系类型集合, 所有属性名集合)。 属性不按标签分开：cypher 里 `n.name` 的 `n` 要回溯到它的标签才能精确校验， 而 WITH / OPTIONAL MATCH 之后变量的标签可能已经丢了。这里只判"这个属性名在图里 存在吗"，宁可漏判也不误杀。 """ meta = json.loads(conf.TCM_METADATA) labels = {label["name"] for label in meta.get("labels", [])} rel_types = {rel["type"] for rel in meta.get("relationships", [])} props = set() for label in meta.get("labels", []): props.update(prop["name"] for prop in label.get("properties", [])) for rel in meta.get("relationships", []): props.update(prop["name"] for prop in rel.get("properties", [])) return labels, rel_types, props LABELS, REL_TYPES, PROP_KEYS = _load_schema() # `(n:Herb)`、`(f:Formula:Prescription)` 里的标签。冒号后面允许空格，多标签用 | 或 : LABEL_PATTERN = re.compile(r"\(\s*\w*\s*:\s*([A-Za-z_][\w|:\s]*?)\s*[){\s]") # `-[:HAS_EFFECT]->`、`-[r:HAS_EFFECT|HAS_INGREDIENT*1..2]-` REL_PATTERN = re.compile(r"\[\s*\w*\s*:\s*([A-Za-z_][\w|\s]*)") # `n.name`、`h.property_flavor`。前面不能是字母/数字，排开 `count(n).x` 这种不存在的写法 PROP_PATTERN = re.compile(r"\b[A-Za-z_]\w*\.([A-Za-z_]\w+)") # cypher 内建函数和关键字后面也可能跟点号或被当成标签，先排掉常见的 IGNORED_TOKENS = {"DISTINCT", "AS", "AND", "OR", "NOT", "IN", "NULL", "TRUE", "FALSE"} STRING_LITERAL = re.compile(r"'[^']*'|\"[^\"]*\"") def _strip_literals(cypher: str) -> str: """把字符串字面量换成空串再做正则匹配。 实体名是数据不是代码，`WHERE s.name = '桂枝汤.加减'` 里的 `.加减` 不该被当成属性， `'Herb'` 这种字面量也不该被当成标签。 """ return STRING_LITERAL.sub("''", cypher) def _split_names(raw: str) -> List[str]: """`Formula:Prescription`、`HAS_EFFECT|HAS_INGREDIENT*1..2` 拆成名字列表""" names = re.split(r"[|:]", raw) # 关系里可能带 `*1..2` 变长路径，星号后面的部分不是名字 return [name.split("*")[0].strip() for name in names if name.split("*")[0].strip()] # ====================== # 第 1 层：静态只读 # ====================== def _check_readonly(cypher: str) -> List[str]: """返回问题列表，空列表表示这一层通过""" upper = _strip_literals(cypher).upper() problems = [ f"包含写操作关键字 {keyword}" for keyword in FORBIDDEN_KEYWORDS if re.search(rf"\b{re.escape(keyword)}\b", upper) ] if "RETURN" not in upper: problems.append("没有 RETURN，查不出任何结果") if not upper.lstrip().startswith(("MATCH", "OPTIONAL MATCH", "WITH")): problems.append("不是以 MATCH / OPTIONAL MATCH / WITH 开头") return problems # ====================== # 第 2 层：对 schema 核账 # ====================== def _check_schema(cypher: str) -> List[str]: """标签、关系类型、属性名都要在 tcm_metadata.json 里存在""" text = _strip_literals(cypher) problems = [] for raw in LABEL_PATTERN.findall(text): for label in _split_names(raw): if label.upper() in IGNORED_TOKENS: continue if label not in LABELS: problems.append(f"标签不存在: {label}") for raw in REL_PATTERN.findall(text): for rel in _split_names(raw): if rel not in REL_TYPES: problems.append(f"关系类型不存在: {rel}") for prop in PROP_PATTERN.findall(text): if prop not in PROP_KEYS: problems.append(f"属性不存在: .{prop}") # 同一个错误可能在一条语句里出现多次，去重保序 return list(dict.fromkeys(problems)) # ====================== # 第 3 层：Neo4j EXPLAIN # ====================== _explain_available = None def _check_explain(cypher: str) -> List[str]: """用 EXPLAIN 让 Neo4j 自己编译一遍，只查语法不执行。 连不上库时把 _explain_available 置 False，后续语句直接跳过这一层， 不用每条都重试一次连接。 """ global _explain_available if _explain_available is False: return [] try: from common.neo4j_manager import neo4j_client neo4j_client.run_cypher(f"EXPLAIN {cypher}") _explain_available = True return [] except Exception as exc: message = str(exc) # 区分"库连不上"和"这条语句本身有问题"：连接类异常不该算语句不合法 if _explain_available is None and ( "Unable to retrieve routing information" in message or "ServiceUnavailable" in message or "AuthError" in type(exc).__name__ or "Couldn't connect" in message ): _explain_available = False print(f"[warn] Neo4j 连不上，跳过 EXPLAIN 校验: {type(exc).__name__}") return [] _explain_available = True return [f"EXPLAIN 失败: {message.splitlines()[0][:120]}"] # ====================== # 对外接口 # ====================== def validate_cypher(cypher: str, use_explain: bool = True) -> List[str]: """检查单条 cypher，返回问题列表。空列表表示合法。 三层依次短路：静态检查没过就不必再对 schema 和连库了，问题已经确定。 """ cypher = (cypher or "").strip() if not cypher: return ["空语句"] problems = _check_readonly(cypher) if problems: return problems problems = _check_schema(cypher) if problems: return problems return _check_explain(cypher) if use_explain else [] def check_cypher_node01(state: AgentState, use_explain: bool = True) -> AgentState: """检查 cypher_query 里所有语句是否合法，结果写回 is_all_validate_cypher""" cyphers = state.get("cypher_query") or [] if not cyphers: state["is_all_validate_cypher"] = EMPTY_IS_VALID return state all_valid = True for cypher in cyphers: problems = validate_cypher(cypher, use_explain=use_explain) if problems: all_valid = False # 打日志：不合法时下游只看到一个 False，不记下来就无法定位是哪条、错在哪 print(f"[warn] 非法 cypher: {cypher[:80]}") for problem in problems: print(f" - {problem}") state["is_all_validate_cypher"] = all_valid return state if __name__ == "__main__": print(f"schema: {len(LABELS)} 标签 / {len(REL_TYPES)} 关系 / {len(PROP_KEYS)} 属性") # (语句, 期望合法) —— 前四条来自 generate_neo4j_cypher_node 的真实输出 cases: List[Tuple[str, bool]] = [ ( "MATCH (h:Herb)-[:HAS_EFFECT]->(e:Effect) WHERE h.name IN ['人参'] " "RETURN h.name AS herb, e.name AS effect LIMIT 20", True, ), ( "MATCH (f:Formula)-[:HAS_INGREDIENT]->(h:Herb) WHERE f.name = '四君子汤' " "RETURN h.name AS herb LIMIT 20", True, ), ( "MATCH (s:Source)<-[:FROM_SOURCE]-(f:Formula) WHERE s.name IN ['《伤寒论》'] " "RETURN f.name AS formula LIMIT 20", True, ), ( "MATCH (h:Herb) WHERE h.name = '人参' " "RETURN h.property_flavor AS flavor, h.meridian AS meridian LIMIT 20", True, ), # 标签不在 schema 里：语法合法，执行恒空 —— 第 2 层要抓的就是这种 ( "MATCH (m:Medicine)-[:HAS_EFFECT]->(e:Effect) RETURN m.name AS name LIMIT 20", False, ), # 关系名编造 ( "MATCH (h:Herb)-[:CURES]->(d:Disease) RETURN d.name AS disease LIMIT 20", False, ), # 属性编造 ("MATCH (h:Herb) RETURN h.price AS price LIMIT 20", False), # 写操作 ("MATCH (h:Herb {name: '人参'}) SET h.name = '党参' RETURN h.name AS n", False), # 没有 RETURN ("MATCH (h:Herb) WHERE h.name = '人参'", False), # 语法错误：括号不闭合，前两层看不出来，靠 EXPLAIN ("MATCH (h:Herb RETURN h.name AS herb LIMIT 20", False), # 实体名里带点号和标签同名词，不该被误判 ( "MATCH (h:Herb) WHERE h.name IN ['Herb', '桂枝.加减'] RETURN h.name AS n LIMIT 5", True, ), ("", False), ] print("\n=== 单条校验 ===") wrong = 0 for cypher, expected in cases: problems = validate_cypher(cypher) actual = not problems flag = "OK " if actual == expected else "MISS" if actual != expected: wrong += 1 print(f" [{flag}] {'合法' if actual else '非法'} <- {cypher[:64] or '<空>'}") for problem in problems: print(f" {problem}") print(f"\n预期不符 {wrong} 条") print("\n=== 节点整体 ===") for cypher_query in [ [cases[0][0], cases[1][0]], # 全合法 -> True [cases[0][0], cases[4][0]], # 混了一条非法 -> False [], # 空列表 -> False ]: result = check_cypher_node01({"cypher_query": list(cypher_query)}) print(f" {len(cypher_query)} 条 -> {result['is_all_validate_cypher']}") #我们不用这个node了,逻辑有些复杂 #逻辑: #1. 用正则进行cypher语句的检查如果没问题 #2. 用neo4j的元数据进行cypher语句的检查如果没问题 # 3.用Explain 语句进行检查