"""Cypher 生成节点 输入: state['input'] 用户原始提问 state['matched_*'] 标准化后的实体名，拿来写 WHERE 条件 Config().TCM_METADATA 图谱的 schema（labels / relationships / triples） 输出: state['cypher_query']: List[str] 生成的 cypher 语句列表 为什么要喂元数据：不给 schema，模型会凭空造标签和关系名（`:Medicine`、`:CURES`）， 查出来永远是空。`tcm_metadata.json` 是 `_003_` 从库里导出的真实骨架，14 个三元组 限定了哪些 (起点标签, 关系, 终点标签) 组合真实存在。 为什么要喂 matched_* 而不是用户原词：用户说"拉肚子"，图里存的是"腹泻"， `WHERE n.name = '拉肚子'` 查不到东西。matched_* 已经是库里的标准名了， 所以 prompt 里明确要求 WHERE 只能用这些名字的原文。 """ import json import re from typing import List from langchain_core.messages import HumanMessage, SystemMessage from _004_langgraph_more_nodes.agent_state import AgentState from common.config import Config from common.llm import my_llm conf = Config() llm = my_llm # 六类实体在 state 里的键 -> prompt 里展示用的中文标签 MATCHED_FIELDS = { "matched_herbs": ("Herb", "药材"), "matched_formulas": ("Formula", "方剂"), "matched_diseases": ("Disease", "疾病"), "matched_symptoms": ("Symptom", "症状"), "matched_effects": ("Effect", "功效"), "matched_sources": ("Source", "出处"), } SYSTEM_PROMPT = ( "你是一个 Neo4j Cypher 生成专家，负责把中医知识图谱上的自然语言提问翻译成 Cypher 查询。\n\n" "【图谱 schema】以下是数据库的真实骨架，只能使用这里出现的标签、关系类型和属性，" "不要发明任何新的标签或关系名：\n" "{metadata}\n\n" "【生成要求】\n" "1. 只使用 schema 中 triples 里真实存在的 (起点标签)-[关系]->(终点标签) 组合，方向不要写反。\n" "2. WHERE 条件里的实体名必须原样使用下面给出的『可用实体名』，一个字都不要改，" "也不要用提问里的原词（图里存的是标准名）。多个候选名用 IN [...] 覆盖。\n" "3. 只生成读查询：只能用 MATCH / OPTIONAL MATCH / WHERE / WITH / RETURN / ORDER BY / LIMIT / DISTINCT。" "严禁 CREATE、MERGE、SET、DELETE、REMOVE、DETACH、DROP、LOAD CSV、CALL。\n" "4. 每条语句必须有 RETURN，并且给返回列起别名（AS）。结果集可能很大，一律加 LIMIT（默认 20）。\n" "5. 一条语句只回答一个角度。提问涉及多个角度时拆成多条语句，最多 5 条。\n" "6. 如果可用实体名全为空，或提问无法用这个图谱回答，返回空数组 []。\n\n" "【输出格式】只输出一个 JSON 数组，每个元素是一条完整的 cypher 字符串，" "不要 markdown 代码块，不要任何解释文字。例如：\n" '["MATCH (h:Herb)-[:HAS_EFFECT]->(e:Effect) WHERE h.name IN [\'人参\'] ' 'RETURN h.name AS herb, e.name AS effect LIMIT 20"]' ) # 只读白名单：出现这些写操作关键字就把语句丢掉。 # 生成的 cypher 会直接打到生产库上，不能只靠 prompt 里那句"严禁"。 FORBIDDEN_KEYWORDS = ( "CREATE", "MERGE", "SET", "DELETE", "REMOVE", "DETACH", "DROP", "LOAD CSV", "CALL", "FOREACH", "IMPORT", ) MAX_CYPHER = 5 def _build_entity_block(state: AgentState) -> str: """把 matched_* 拼成给模型看的『可用实体名』清单""" lines = [] for key, (label, zh) in MATCHED_FIELDS.items(): names = state.get(key) or [] if names: listed = "、".join(f"'{n}'" for n in names) lines.append(f"- {label}（{zh}）: {listed}") return "\n".join(lines) if lines else "（无，没有匹配到任何实体）" def _strip_code_fence(text: str) -> str: """去掉模型可能套上的 ```json ... ``` 外壳""" text = text.strip() if text.startswith("```"): text = re.sub(r"^```[a-zA-Z]*\s*", "", text) text = re.sub(r"\s*```$", "", text) return text.strip() def _parse_cypher_list(raw: str) -> List[str]: """把模型输出解析成 cypher 字符串列表。 这里不用 JsonOutputParser：需要在解析失败时退回到"按行捞 MATCH 语句"， 而不是抛异常整条链断掉 —— 模型偶尔会在数组外面多写一句话。 """ text = _strip_code_fence(raw) try: data = json.loads(text) except json.JSONDecodeError: # 从整段文本里截出第一个 [...]，模型在数组前后加了废话时救一把 match = re.search(r"\[.*\]", text, re.DOTALL) try: data = json.loads(match.group(0)) if match else None except json.JSONDecodeError: data = None if isinstance(data, list): items = data elif isinstance(data, dict): # 模型有时包一层 {"cypher_query": [...]} items = next( (v for v in data.values() if isinstance(v, list)), [data.get("cypher", "")] ) else: # 彻底解析不了，按行捞出以 MATCH 开头的语句 items = [ line.strip().rstrip(",").strip('"') for line in text.splitlines() if line.strip().upper().startswith(("MATCH", "OPTIONAL MATCH")) ] return [str(item).strip() for item in items if str(item).strip()] def _is_readonly(cypher: str) -> bool: """确认这条语句只读。 按词边界匹配，避免把实体名里的字母误判 —— 比如属性名 `processing` 含有 "cess" 不成问题，但 `MERGE` 这种词必须整词命中才算。 """ upper = cypher.upper() for keyword in FORBIDDEN_KEYWORDS: if re.search(rf"\b{re.escape(keyword)}\b", upper): return False return "RETURN" in upper and upper.lstrip().startswith(("MATCH", "OPTIONAL MATCH", "WITH")) def generate_neo4j_cypher_node(state: AgentState) -> AgentState: """根据提问 + 匹配到的实体 + 图谱元数据生成 cypher，写回 cypher_query""" # 检查节点不通过时会再回到这里，所以这个节点一轮里可能执行多次，计数加在最前面 retry_count = (state.get("cypher_retry_count") or 0) + 1 state["cypher_retry_count"] = retry_count print(f"cypher生成节点执行开始了（第 {retry_count} 次生成）") # 读转写结果而不是 state['input']：指代词还原过的问题才生成得出正确的 cypher user_input = (state.get("input_semantic_trans") or "").strip() entity_block = _build_entity_block(state) # 一个实体都没匹配上时，生成的 WHERE 必然查不到东西，省一次调用 has_entity = any(state.get(key) for key in MATCHED_FIELDS) if not user_input or not has_entity: state["cypher_query"] = [] print("cypher生成节点：没有匹配到实体，不生成") return state # 重试时把上一轮不合法的语句回喂给模型。不带这段，模型看到的输入和上一次 # 完全一样，大概率原样再生成一遍，重试就白花了。 retry_block = "" previous = state.get("cypher_query") or [] if retry_count > 1 and previous: listed = "\n".join(f" {i}. {c}" for i, c in enumerate(previous, 1)) retry_block = ( f"\n\n【注意】下面这些语句是你上一次生成的，它们没有通过 Neo4j 的语法/schema " f"校验，请找出问题并重新生成，不要再输出同样的语句：\n{listed}" ) messages = [ SystemMessage(content=SYSTEM_PROMPT.format(metadata=conf.TCM_METADATA)), HumanMessage( content=( f"用户提问：{user_input}\n\n" f"可用实体名（WHERE 条件只能用这些）：\n{entity_block}" f"{retry_block}" ) ), ] raw = llm.invoke(messages).content cyphers, seen = [], set() for cypher in _parse_cypher_list(raw): if not _is_readonly(cypher): print(f"[warn] 丢弃非只读或格式异常的 cypher: {cypher[:80]}") continue if cypher not in seen: seen.add(cypher) cyphers.append(cypher) if len(cyphers) >= MAX_CYPHER: break state["cypher_query"] = cyphers print(f"cypher生成节点执行完成: {len(cyphers)} 条") return state if __name__ == "__main__": cases = [ { "input_semantic_trans": "人参有什么功效？", "matched_herbs": ["人参"], }, { "input_semantic_trans": "四君子汤由哪些药材组成，出自哪里？", "matched_formulas": ["四君子汤"], }, { "input_semantic_trans": "我拉肚子，有哪些方剂或药材可以缓解？", "matched_symptoms": ["腹泻", "泄泻", "泻痢"], }, { "input_semantic_trans": "哪些药材能补气，又能治肾虚？", "matched_effects": ["补气"], "matched_diseases": ["肾虚"], }, { "input_semantic_trans": "《伤寒论》里记载了哪些方剂？", "matched_sources": ["《伤寒论》", "《伤寒杂病论》"], }, # 没有匹配到实体：预期空列表，不调用大模型 {"input_semantic_trans": "今天天气怎么样"}, ] for case in cases: result = generate_neo4j_cypher_node(dict(case)) print(f"\n提问: {case['input_semantic_trans']}") if not result["cypher_query"]: print(" <无 cypher>") for i, cypher in enumerate(result["cypher_query"], 1): print(f" {i}. {cypher}")