知识图谱构建节点

本文档详细介绍知识库导入流程中的知识图谱构建节点（KnowledgeGraphNode），该节点负责从文档切片中提取实体与关系，经过清洗后分别写入 Milvus 向量数据库和 Neo4j 图数据库，是整个导入流程中业务最复杂的节点。

---

学习理念：知识图谱构建是本项目的"皇冠"节点——KnowledgeGraphNode 协调 LLM、Milvus、Neo4j 三个外部服务，对每个切片执行"LLM 提取实体关系 → JSON 清洗 → 实体向量化写入 Milvus → 三元组写入 Neo4j"四步流程。核心价值：将非结构化文本转化为结构化的图数据，让 RAG 检索从"搜段落"升级为"搜关系"。

海外对标：KnowledgeGraphNode 的"LLM → 实体提取 → 向量存储 → 图存储"四步架构对标 Google Knowledge Graph API 的 Entity Extraction + Freebase 方案，以及 Neo4j 官方推荐的 "LLM + GraphRAG" 模式（Neo4j + OpenAI + Vector Index）。三元组提取的 Prompt Engineering 设计与微软 GraphRAG 项目的实体提取 Prompt 思路一致。

本节 AI 替代率：~65% | 人工干预率：~35%

角色	能力范围
🤖 AI 擅长	LLM 调用骨架、JSON 清洗代码、Milvus Schema/索引、BGE-M3 向量化、Neo4j Cypher 模板、测试代码
👤 人类需理解	SYSTEM_PROMPT 的设计（实体类型/关系类型/命名规则的定义）、Cypher 注入安全（关系类型白名单）、事务原子性（execute_write）、双写一致性（Milvus + Neo4j）

---

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习目标明确
🟢	§2 核心概念	~95%	~5%	全部是图片说明
🟡	§3 整体流程	~90%	~10%	理解三大阶段即可
🔴	§4.1 主流程	~75%	~25%	遍历切片 + _process_single_chunk 编排
🔴	§4.2 LLM 提取	~70%	~30%	SYSTEM_PROMPT 设计是核心难点
🔴	§4.3 JSON 清洗	~80%	~20%	去围栏 + 实体/关系清洗 + 白名单校验
🟠	§4.4 Milvus 写入	~70%	~30%	实体名去重 + Schema + 双向量 + 索引
🔴	§4.5 Neo4j 写入	~60%	~40%	最复杂部分：事务 + Cypher 注入安全 + MERGE 幂等
🟢	§5 单元测试	~90%	~10%	看预期输出 + Neo4j Browser 验证
🟡	§6 总结	~90%	~10%	设计要点回顾

---

技术栈健康度标签体系

技术	健康度	建议
Neo4j	🟢 稳定	图数据库第一品牌。Cypher 查询语言成熟稳定。社区版免费，适合学习和中小项目。生产环境需考虑集群部署。
Cypher MERGE	🟢 稳定	Neo4j 的幂等写入语句。ON CREATE SET / ON MATCH SET 的分支逻辑是图数据写入的标准模式。
execute_write	🔥 巅峰	Neo4j Python Driver 的事务包装器，确保原子性。GraphRAG 项目中的标准做法。
Prompt Engineering for KG	⏳ 成长期	用 LLM 提取结构化知识的核心技能。2025-2026 年逐步被 Structured Output / JSON Mode 替代。
IVF_FLAT	🟢 稳定	Milvus 向量索引算法，适合中等规模数据集（百万级）。AUTOINDEX 的底层会自动选择类似算法。
BGE-M3	🔥 巅峰	混合嵌入模型，同时输出稠密+稀疏向量。知识图谱实体向量化的标配。

体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。

---

中英文对照表

English	中文	本质
Knowledge Graph	知识图谱	以"实体-关系-实体"三元组表示知识的图结构
Triple / Triple	三元组	知识图谱的最小单元：(头实体, 关系, 尾实体)
Entity	实体	知识图谱中的节点，代表现实世界的事物
Relation	关系	连接两个实体的有向边，描述它们之间的语义关联
Cypher	Cypher 查询语言	Neo4j 的图查询语言，类似 SQL 但专为图结构设计
MERGE	合并写入	Cypher 的幂等写入语句：存在则匹配，不存在则创建
execute_write	事务写入	Neo4j Python Driver 的原子事务执行方法
Dangling Relation	悬空关系	引用了不存在的头/尾实体的关系
Whitelist Validation	白名单校验	只允许预定义值通过的安全检查机制
Cypher Injection	Cypher 注入	通过用户输入篡改 Cypher 语句的安全攻击

---

💡 程序员比喻

• KnowledgeGraphNode 就像 terraform apply——先 plan（LLM 提取），再 validate（JSON 清洗），最后 apply（Milvus + Neo4j 写入）。
• 三元组 (Entity-Relation-Entity) 就像 git commit 中的 (author, authored_date, message)——头=作者，关系=authored_date，尾=message。
• Neo4j MERGE 就像 kubectl apply——声明式配置，存在就不动，不存在就创建。
• Cypher 注入防御 就像 SQL 注入的 Prepared Statement——能参数化的就参数化（$name），不能参数化的（关系类型）就用白名单。
• 双写 Milvus + Neo4j 就像数据库的 CQRS 模式——Milvus = 读模型（向量检索），Neo4j = 写模型（图结构）。
• JSON 多层清洗 就像 CI/CD 的 lint + test + deploy 流水线——每一层把关，不合格就 reject。

---

1. 任务目标

1.1 本章目标

通过本章学习，你将掌握：

1. 知识图谱基本概念：理解实体、关系、三元组等核心术语
2. LLM 信息抽取：学会用 Prompt 工程驱动大模型从文本中提取结构化知识
3. JSON 解析与清洗：掌握 LLM 返回结果的鲁棒解析与多层清洗策略
4. Milvus 实体向量化存储：理解将实体名称向量化并存入 Milvus 的完整流程
5. Neo4j 图数据写入：学会用 Cypher 语句构建实体节点与关系边
6. 事务化写入设计：理解在单个事务内完成所有图操作的原子性保障

1.2 涉及文件

knowledge/
├── processor/import_process/nodes/
│   └── knowledge_graph.py            # 知识图谱构建节点（本章重点）
│
└── tools/
    ├── llm_utils.py                  # LLM 客户端封装
    ├── embedding_utils.py            # BGE-M3 混合嵌入工具
    ├── neo4j_utils.py                # Neo4j 连接管理
    └── normalize_sparse_vector.py    # 稀疏向量 L2 归一化

1.3 节点在流程中的位置

---

2. 核心概念扫盲

2.1 什么是知识图谱？

知识图谱是一种用图结构来表示现实世界知识的方式。它把信息组织成"谁和谁有什么关系"的形式，而不是传统的表格或文本：

2.2 三元组：知识图谱的最小单元

知识图谱中的每条知识都可以表示为一个三元组（Triple）：

2.3 实体与实体类型

本项目定义了 7 种实体类型（label），每种对应操作手册中的不同概念：

2.4 关系类型

本项目定义了 8 种关系类型，描述实体之间的语义连接：

2.5 Neo4j 与 Cypher 语言基础

Neo4j 是一个图数据库，专门用来存储和查询图结构数据。Cypher 是它的查询语言：

2.6 实体名称也要存入 Milvus？

在 RAG 检索时，除了对切片内容做相似度搜索外，还需要对实体名称做向量检索，以便在知识图谱中定位相关节点，实现"图增强检索"：

---

3. 知识图谱构建业务处理流程（总）

知识图谱节点是整个导入流程中业务最复杂的节点，它需要协调 LLM、Milvus、Neo4j 三个外部服务。整体流程可以概括为对每个切片执行"提取 → 清洗 → 双写"三大阶段：

关键设计：此节点不修改 state，所有产出都直接写入外部存储（Milvus 和 Neo4j）。这意味着即使某个切片处理失败，也不会影响 state 的完整性。

---

4. 知识图谱构建业务处理流程（分）

整个节点的代码逻辑可以拆分为以下 5 个子流程，我们逐一分析。

---

4.1 主流程控制

4.1.1 目标

遍历所有切片，对每个有效切片调用"提取 → 清洗 → 写入"子流程。

4.1.2 需求分析

主流程需要解决以下问题：

• 空数据保护：chunks 为空时直接跳过，不报错
• 数据校验：跳过非字典类型、缺少 content 或 item_name 的切片
• 单切片异常隔离：一个切片处理失败，不影响其他切片继续处理
• 流程编排：按"提取 → 清洗 → 写入"顺序串联子流程

4.1.3 实现流程

4.1.3.1 实现流程图

4.1.3.2 具体实现步骤

Step1：读取切片列表

从 state 中获取 chunks 列表。如果列表为空，记录日志并直接返回 state，避免后续无意义的处理。

🟡 【P1 看注释就行】 Step1 空数据保护——chunks 为空则跳过。

chunks = state.get("chunks", [])
if not chunks:
    self.logger.info("chunks 为空，跳过知识图谱构建")
    return state

Step2：遍历切片并校验

使用 for 循环遍历每个切片。对每个切片依次检查：是否为字典类型、content 是否非空、item_name 是否存在。任何一项不满足，就跳过该切片。

🟡 【P1 看注释就行】 Step2 遍历切片——校验是否为 dict、content 非空、item_name 存在。注意 chunk_id 回退逻辑。

for i, chunk in enumerate(chunks):
    if not isinstance(chunk, dict):
        continue

    content = chunk.get("content", "")
    chunk_id = str(chunk.get("chunk_id", f"temp_{i}"))
    item_name = chunk.get("item_name") or state.get("item_name", "")

    if not content or not item_name:
        continue

注意：chunk_id 可能不存在（比如 Milvus 写入失败时），此时用 temp_{i} 作为临时 ID。item_name 优先取切片自身的值，其次取 state 全局值。

Step3：调用单切片处理

将校验通过的切片交给 _process_single_chunk 方法处理。该方法内部依次执行提取、清洗、写入三个子流程。

🟡 【P1 看注释就行】 Step3 调用单切片处理——_process_single_chunk 是核心调度方法。

self._process_single_chunk(content, chunk_id, item_name)

Step4：单切片处理流程

_process_single_chunk 是整个节点的核心调度方法，它按顺序调用四个子方法，每一步的输出是下一步的输入。任何一步返回空结果，后续步骤都会被跳过。

🔥 【P0 必须要学】 核心调度方法——_process_single_chunk 的执行链：提取(_llm_extract) → 清洗(_parse_and_clean) → Milvus(_save_entities_to_milvus) → Neo4j(_save_graph_to_neo4j)。每一步返回值决定下一步是否执行。

def _process_single_chunk(self, content, chunk_id, item_name):
    config = get_config()

    # 1. LLM 提取实体和关系
    raw_response = self._llm_extract(content)
    if not raw_response:
        return

    # 2. 解析 JSON 并清洗
    graph_data = self._parse_and_clean(raw_response)
    if not graph_data.get("entities"):
        return

    # 3. 实体写入 Milvus
    self._save_entities_to_milvus(
        graph_data.get("entities", []), chunk_id, content, item_name, config,
    )

    # 4. 图数据写入 Neo4j
    self._save_graph_to_neo4j(graph_data, chunk_id, item_name, config)

---

4.2 LLM 实体与关系提取

4.2.1 目标

调用大语言模型，从一段文本切片中提取出结构化的实体和关系信息，返回 JSON 字符串。

4.2.2 需求分析

• Prompt 设计：需要精心设计 System Prompt，定义允许的实体类型、关系类型、命名规则和输出格式
• 错误处理：LLM 调用可能超时或报错，需要优雅降级（返回空字符串）
• 输出格式约束：要求 LLM 只输出 JSON 对象，不输出解释和 Markdown

4.2.3 实现流程

4.2.3.1 实现流程图

4.2.3.2 具体实现步骤

Step1：获取 LLM 客户端

调用 get_llm_client() 获取已配置好的 ChatOpenAI 客户端单例。

🟡 【P1 看注释就行】 获取 LLM 客户端——get_llm_client() 单例。

llm = get_llm_client()

Step2：调用 LLM

使用 LangChain 的 SystemMessage 和 HumanMessage 构建消息列表。System Prompt 中详细定义了所有允许的实体类型、关系类型、命名规则和输出 JSON Schema。

🟡 【P1 看注释就行】 LLM 调用——SystemMessage + HumanMessage 构造消息。捕获异常返回空字符串实现降级。

response = llm.invoke([
    SystemMessage(content=self.SYSTEM_PROMPT),
    HumanMessage(content=f"请处理以下文本切片：\n\n{content}"),
])
return (response.content or "").strip()

Step3：异常捕获与降级

如果 LLM 调用失败（网络异常、超时、模型错误等），捕获异常并返回空字符串，确保不影响其他切片的处理。

🔴 【P0 必须要学】实体清洗——_clean_entities 的过滤逻辑：(1)空值过滤 (2)去重 (3)保留 description。seen = set() 用 (name, label) 元组作为去重键。

except Exception as e:
    self.logger.warning(f"LLM 提取失败: {e}")
    return ""

关于 SYSTEM_PROMPT 的设计：这是整个节点最核心的部分之一。一个好的 Prompt 可以让 LLM 输出规范、准确的 JSON，减少后续清洗的工作量。Prompt 中包含了实体类型定义、命名规则、关系类型定义、抽取原则和输出 Schema。

---

4.3 JSON 解析与清洗

4.3.1 目标

将 LLM 返回的原始 JSON 字符串解析为结构化数据，并通过多层清洗确保数据质量。

4.3.2 需求分析

LLM 的输出并不总是完美的，常见问题包括：

• Markdown 围栏：LLM 可能用 ```json ... ``` 包裹输出
• 实体名过长：LLM 可能把整句话作为实体名
• 重复实体：同一个实体被提取多次
• 悬空关系：关系引用了不存在的实体名
• 字段名不一致：关系类型字段可能叫 type，也可能叫 label
• 非法关系类型：LLM 可能自创关系类型

4.3.3 实现流程

4.3.3.1 实现流程图

4.3.3.2 具体实现步骤

Step1：去除 Markdown 围栏

LLM 经常在 JSON 外面包裹 Markdown 代码块（如 ```json ... ```），需要用正则表达式去除。

🔥 【P0 必须要学】关系清洗——_clean_relations 的三层过滤：(1)空值过滤 (2)字段名修正 + 白名单校验（非法类型降级为 RELATED_TO） (3)悬空引用过滤（head/tail 必须在 valid_entity_names 中）。这是数据质量的核心保障。

cleaned_text = re.sub(r"^```(?:json)?\s*", "", raw_text.strip())
cleaned_text = re.sub(r"\s*```$", "", cleaned_text)

Step2：JSON 解析

使用 json.loads() 将字符串解析为 Python 字典。如果解析失败，记录日志并返回空结果。

🟡 【P1 看注释就行】 按 name 去重——_dedup_entities_by_name 合并同名不同类型的实体。labels 存为 set 供后面使用。

try:
    data = json.loads(cleaned_text)
except json.JSONDecodeError as e:
    self.logger.warning(f"JSON 解析失败: {e}, 原文前200字: {raw_text[:200]}")
    return {"entities": [], "relations": []}

Step3：清洗实体列表

_clean_entities 方法对每个实体执行以下检查：

1. 过滤无效项：name 或 label 为空的直接跳过
2. 去重：使用 (name, label) 元组作为去重键，通过 set 去重
3. 构建清洗后的实体字典：只保留 name、label 和可选的 description

🔥 【P0 必须要学】Milvus 实体集合 Schema——注意索引类型不同：稠密向量用 IVF_FLAT + COSINE 度量（非 IP），稀疏向量用 SPARSE_INVERTED_INDEX + IP。enable_dynamic_field=True 允许灵活添加字段。

def _clean_entities(self, entities):
    seen = set()
    cleaned = []

    for entity in entities:
        name = str(entity.get("name", "")).strip()
        label = str(entity.get("label", "")).strip()
        description = str(entity.get("description", "")).strip()

        if not name or not label:
            continue

        dedup_key = (name, label)
        if dedup_key in seen:
            continue
        seen.add(dedup_key)

        result = {"name": name, "label": label}
        if description:
            result["description"] = description
        cleaned.append(result)

    return cleaned

Step4：收集有效实体名

清洗完实体后，提取所有有效实体的 name 组成集合，用于后续关系清洗时判断"悬空引用"。

🟡 【P1 看注释就行】 BGE-M3 encode_documents——一次性向量化所有实体名称。注意返回的稀疏向量是 CSR 矩阵格式。

valid_names = {e["name"] for e in cleaned_entities}

Step5：清洗关系列表

_clean_relations 方法对每条关系执行三层过滤：

1. 空值过滤：head 或 tail 为空的直接跳过
2. 字段名修正与白名单校验：关系类型可能在 type 或 label 字段中，统一取值后与白名单比对，不在白名单内的降级为 RELATED_TO
3. 悬空引用过滤：关系的 head 和 tail 必须都在 valid_names 集合中，否则跳过

🟡 【P1 看注释就行】 组装 Milvus 记录——从 CSR 矩阵提取稀疏向量（indptr[idx] 到 indptr[idx+1]），normalize_sparse_vector 做 L2 归一化。

def _clean_relations(self, relations, valid_entity_names):
    cleaned = []

    for rel in relations:
        head = str(rel.get("head", "")).strip()
        tail = str(rel.get("tail", "")).strip()
        if not head or not tail:
            continue

        # 兼容 "type" 和 "label" 两种字段名
        rel_type = str(rel.get("type") or rel.get("label") or "RELATED_TO").strip()
        if rel_type not in self.ALLOWED_RELATION_TYPES:
            rel_type = "RELATED_TO"

        # 过滤悬空引用
        if head not in valid_entity_names or tail not in valid_entity_names:
            self.logger.debug(f"悬空关系已跳过: {head} -[{rel_type}]-> {tail}")
            continue

        cleaned.append({"head": head, "tail": tail, "type": rel_type})

    return cleaned

---

4.4 实体向量化写入 Milvus

4.4.1 目标

将清洗后的实体名称通过 BGE-M3 模型向量化（稠密 + 稀疏双向量），然后写入 Milvus 的实体名称集合，为后续的图增强检索提供向量基础。

4.4.2 需求分析

• 实体去重合并：同一个切片中可能出现同名但不同 label 的实体，需按 name 合并
• 集合自动创建：Milvus 集合可能不存在，需要幂等地创建
• Schema 设计：实体集合需要 entity_name、dense_vector、sparse_vector、source_chunk_id、context、item_name 等字段
• 批量向量化与插入：将所有去重后的实体名一次性向量化，再批量插入

4.4.3 实现流程

4.4.3.1 实现流程图

4.4.3.2 具体实现步骤

Step1：按 name 去重合并实体

使用 _dedup_entities_by_name 方法，将同名实体的 label 合并到一个 set 中，只保留一条记录。

🟢 【P2 后面可以查】 插入 + load——insert 后调用 load_collection 使数据可检索。

@staticmethod
def _dedup_entities_by_name(entities):
    dedup = {}
    for entity in entities:
        name = str(entity.get("name", "")).strip()
        if not name:
            continue
        label = str(entity.get("label", "")).strip()
        description = str(entity.get("description", "")).strip()

        if name not in dedup:
            dedup[name] = {"labels": set(), "description": description}
        if label:
            dedup[name]["labels"].add(label)
    return dedup

为什么要合并？ 同一个实体可能在不同关系中以不同 label 出现（比如"万用表"既是 Device 也是某关系的头实体），在 Milvus 中只需要存一条向量记录。

Step2：确保实体集合存在

使用 _ensure_entity_collection 方法幂等地创建集合。如果集合已存在则直接返回。

集合的 Schema 包含 6 个字段：

• pk：自增主键
• entity_name：实体名称（VARCHAR）
• dense_vector：稠密向量（FLOAT_VECTOR, dim=1024）
• sparse_vector：稀疏向量（SPARSE_FLOAT_VECTOR）
• source_chunk_id：来源切片 ID
• context：切片原文前 200 字（用于检索时快速预览上下文）
• item_name：所属商品名称

索引配置：

• 稠密向量：IVF_FLAT 索引，COSINE 度量
• 稀疏向量：SPARSE_INVERTED_INDEX 索引，IP 度量

🟡 【P1 看注释就行】 Neo4j 写入入口——get_neo4j_driver() 获取驱动，session.execute_write 在事务内执行。注意 execute_write 的原子性保障。

@staticmethod
def _ensure_entity_collection(client, collection_name):
    if client.has_collection(collection_name):
        return

    schema = client.create_schema(enable_dynamic_field=True)
    schema.add_field(field_name="pk",              datatype=DataType.INT64,
                     is_primary=True, auto_id=True)
    schema.add_field(field_name="entity_name",     datatype=DataType.VARCHAR,
                     max_length=65535)
    schema.add_field(field_name="dense_vector",    datatype=DataType.FLOAT_VECTOR,
                     dim=1024)
    schema.add_field(field_name="sparse_vector",   datatype=DataType.SPARSE_FLOAT_VECTOR)
    schema.add_field(field_name="source_chunk_id", datatype=DataType.VARCHAR,
                     max_length=65535)
    schema.add_field(field_name="context",         datatype=DataType.VARCHAR,
                     max_length=65535)
    schema.add_field(field_name="item_name",       datatype=DataType.VARCHAR,
                     max_length=65535)

    # 配置索引
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="dense_vector",
        index_name="dense_vector_index",
        index_type="IVF_FLAT",
        metric_type="COSINE",
        params={"nlist": 128},
    )
    index_params.add_index(
        field_name="sparse_vector",
        index_name="sparse_vector_index",
        index_type="SPARSE_INVERTED_INDEX",
        metric_type="IP",
    )

    client.create_collection(
        collection_name=collection_name,
        schema=schema,
        index_params=index_params,
    )

Step3：BGE-M3 向量化实体名称

调用 BGE-M3 模型的 encode_documents 方法，一次性将所有实体名称向量化。返回的结果包含稠密向量和稀疏向量（CSR 格式）。

🔥 【P0 必须要学】事务总控——_write_graph_in_tx 在单个事务内完成三类操作：(1)MERGE Chunk 节点 (2)MERGE Entity 节点 + MENTIONED_IN 关联 (3)MERGE 实体间关系。中途出错自动回滚。

bge_ef = get_bge_m3_model()
names = list(dedup_map.keys())
vectors = bge_ef.encode_documents(names)

注意：这里使用的是 encode_documents 而不是前面切片向量化章节中的 generate_hybrid_embeddings，因为 encode_documents 是 pymilvus 封装的 BGEM3EmbeddingFunction 的原生方法，返回的稀疏向量是 CSR 矩阵格式，需要在 _build_milvus_records 中手动拆解。

Step4：组装 Milvus 插入记录

_build_milvus_records 方法将向量化结果与元数据组装成 Milvus 可插入的记录列表。对稀疏向量需要从 CSR 矩阵中手动提取并归一化。

🟡 【P1 看注释就行】 MERGE Chunk——以 chunk_id + item_name 为唯一标识。

@staticmethod
def _build_milvus_records(names, vectors, chunk_id, content, item_name):
    from knowledge.tools.embedding_utils import normalize_sparse_vector

    dense_list = vectors.get("dense", [])
    sparse_raw = vectors.get("sparse")

    records = []
    for idx, name in enumerate(names):
        if idx >= len(dense_list):
            break

        dense_vector = dense_list[idx]
        if hasattr(dense_vector, "tolist"):
            dense_vector = dense_vector.tolist()

        record = {
            "entity_name": name,
            "dense_vector": dense_vector,
            "source_chunk_id": chunk_id,
            "context": content[:200],
            "item_name": item_name,
        }

        # 从 CSR 矩阵中提取稀疏向量
        if sparse_raw is not None:
            start, end = sparse_raw.indptr[idx], sparse_raw.indptr[idx + 1]
            indices = sparse_raw.indices[start:end].tolist()
            data = sparse_raw.data[start:end].tolist()
            sparse_vector = {k: v for k, v in zip(indices, data)}
            record["sparse_vector"] = normalize_sparse_vector(sparse_vector)

        records.append(record)

    return records

Step5：批量插入并加载集合

将组装好的记录批量插入 Milvus，然后调用 load_collection 使数据可被检索。

🔥 【P0 必须要学】MERGE Entity——关键设计：types 数组存储实体的所有类型标签。ON CREATE SET 初始化 types，ON MATCH SET 追加新 label（不覆盖）。coalesce(n.types, []) 处理 null 值。

if insert_data:
    milvus_client.insert(collection_name=collection_name, data=insert_data)
    milvus_client.load_collection(collection_name=collection_name)

---

4.5 图数据写入 Neo4j

4.5.1 目标

将提取并清洗后的实体和关系以图结构写入 Neo4j 数据库。每个切片的写入在一个事务内完成，保证原子性。

4.5.2 需求分析

• 事务原子性：一个切片的所有节点和关系要么全部写入，要么全部回滚
• 幂等性：使用 MERGE 语句，重复执行不会创建重复数据
• Chunk 节点关联：每个实体要与来源切片建立 MENTIONED_IN 关系
• 关系类型白名单：只允许预定义的关系类型，非法类型降级为 RELATED_TO
• Cypher 注入安全：关系类型直接拼接到 Cypher 中，必须通过白名单校验

4.5.3 实现流程

4.5.3.1 实现流程图

4.5.3.2 具体实现步骤

Step1：获取 Neo4j 驱动并开启会话

从工具模块获取 Neo4j 驱动单例，然后开启一个指定数据库的 Session。使用 execute_write 在写入事务中执行所有操作。

🟡 【P1 看注释就行】 LINK Entity → Chunk——通过 MENTIONED_IN 关系追溯实体来源。

def _save_graph_to_neo4j(self, graph_data, chunk_id, item_name, config):
    entities = graph_data.get("entities", [])
    relations = graph_data.get("relations", [])

    if not entities:
        return

    try:
        from knowledge.tools.neo4j_utils import get_neo4j_driver
        driver = get_neo4j_driver()

        with driver.session(database=config.neo4j_database) as session:
            session.execute_write(
                self._write_graph_in_tx,
                entities, relations, chunk_id, item_name,
            )
    except Exception as e:
        self.logger.warning(f"Neo4j 写入失败: {e}")

execute_write 的含义：这是 Neo4j Python 驱动提供的事务执行方法。传入的函数 _write_graph_in_tx 会在一个事务中执行，如果中途出错会自动回滚。

Step2：事务总控方法

_write_graph_in_tx 是一个静态方法，在单个事务 tx 中依次完成三类写入操作：

🔥 【P0 必须要学】MERGE 关系 + Cypher 注入防护——关系类型通过 str.format() 拼入 Cypher（不支持参数化），所以必须通过白名单校验。{{name: $head}} 的双花括号是 Python .format() 的转义写法。

@staticmethod
def _write_graph_in_tx(tx, entities, relations, chunk_id, item_name):
    cls = KnowledgeGraphNode

    # 1. 创建 Chunk 节点
    cls._tx_merge_chunk(tx, chunk_id, item_name)

    # 2. 创建实体节点 + 关联 Chunk
    for entity in entities:
        name = str(entity.get("name", "")).strip()
        if not name:
            continue
        label = str(entity.get("label", "")).strip()
        description = str(entity.get("description", "")).strip()

        cls._tx_merge_entity(tx, name, label, description, chunk_id, item_name)
        cls._tx_link_entity_to_chunk(tx, name, chunk_id, item_name)

    # 3. 创建实体间关系
    for rel in relations:
        head = str(rel.get("head", "")).strip()
        tail = str(rel.get("tail", "")).strip()
        if not head or not tail:
            continue
        rel_type = str(rel.get("type", "RELATED_TO")).strip() or "RELATED_TO"
        cls._tx_merge_relation(tx, head, tail, rel_type, item_name)

Step3：MERGE Chunk 节点

创建或匹配 Chunk 节点，以 chunk_id 和 item_name 为唯一标识。

🔥 【P0 必须要学】KnowledgeGraphNode 完整架构——四个阶段：提取→清洗→Milvus→Neo4j。三个安全机制：JSON 多层清洗、关系类型白名单、execute_write 事务。两个外部存储：Milvus（向量检索）+ Neo4j（图遍历）。

CYPHER_MERGE_CHUNK = """
    MERGE (c:Chunk {id: $chunk_id, item_name: $item_name})
"""

@staticmethod
def _tx_merge_chunk(tx, chunk_id, item_name):
    tx.run(KnowledgeGraphNode.CYPHER_MERGE_CHUNK,
           chunk_id=chunk_id, item_name=item_name)

Step4：MERGE Entity 节点

创建或合并实体节点。使用 ON CREATE SET 和 ON MATCH SET 分别处理新建和已存在两种情况。

关键设计：types 字段是一个数组，存储实体的所有类型标签。当同名实体从不同切片被提取时，新的 label 会被追加到数组中（而不是覆盖）。

🟢 【P2 后面可以查】 测试代码——模拟一个切片跑通全流程。看预期输出即可。

CYPHER_MERGE_ENTITY = """
    MERGE (n:Entity {name: $name, item_name: $item_name})
    ON CREATE SET
        n.source_chunk_id = $chunk_id,
        n.description     = $description,
        n.types           = CASE
                                WHEN $label = "" THEN []
                                ELSE [$label]
                            END
    ON MATCH SET
        n.description = CASE
                            WHEN $description <> "" THEN $description
                            ELSE coalesce(n.description, "")
                        END,
        n.types       = CASE
                            WHEN $label = ""                       THEN coalesce(n.types, [])
                            WHEN $label IN coalesce(n.types, [])   THEN n.types
                            ELSE coalesce(n.types, []) + $label
                        END
"""

Cypher 语法解读：

• MERGE：存在就匹配，不存在就创建
• ON CREATE SET：只在创建时执行
• ON MATCH SET：只在匹配到已有节点时执行
• coalesce(n.types, [])：如果 n.types 为 null 则返回空数组
• $label IN coalesce(n.types, [])：检查 label 是否已在数组中，避免重复添加

Step5：关联 Entity 到 Chunk

通过 MENTIONED_IN 关系将实体与来源切片关联。这样在查询时可以追溯"这个实体是从哪个切片提取出来的"。

CYPHER_LINK_ENTITY_TO_CHUNK = """
    MATCH (n:Entity {name: $name, item_name: $item_name})
    MATCH (c:Chunk  {id: $chunk_id, item_name: $item_name})
    MERGE (n)-[:MENTIONED_IN]->(c)
"""

Step6：MERGE 实体间关系

创建实体之间的有向关系。关键安全措施：关系类型 rel_type 是通过字符串格式化拼入 Cypher 的（因为 Cypher 不支持参数化的关系类型），所以必须通过白名单校验，防止 Cypher 注入。

ALLOWED_RELATION_TYPES = {
    "HAS_OPERATION", "HAS_PART", "HAS_STEP", "USES_TOOL",
    "HAS_WARNING", "NEXT_STEP", "AFFECTS", "REQUIRES",
    "MENTIONED_IN", "RELATED_TO",
}

CYPHER_MERGE_RELATION_TEMPLATE = """
    MATCH (h:Entity {{name: $head, item_name: $item_name}})
    MATCH (t:Entity {{name: $tail, item_name: $item_name}})
    MERGE (h)-[:{rel_type}]->(t)
"""

@staticmethod
def _tx_merge_relation(tx, head, tail, rel_type, item_name):
    if rel_type not in KnowledgeGraphNode.ALLOWED_RELATION_TYPES:
        rel_type = "RELATED_TO"
    cypher = KnowledgeGraphNode.CYPHER_MERGE_RELATION_TEMPLATE.format(
        rel_type=rel_type
    )
    tx.run(cypher, head=head, tail=tail, item_name=item_name)

为什么关系类型不能参数化？ 在 Cypher 中，$name 形式的参数只能用于属性值，不能用于关系类型名称（如 [:$rel_type] 是非法语法）。因此必须使用 Python 的 str.format() 拼接，同时用白名单确保安全。注意模板中的 {{name: $head}} 双花括号是为了在 .format() 中转义，最终生成 {name: $head}。

---

4.5.4 代码实现

以下是 knowledge_graph.py 的完整代码：

"""
知识图谱构建节点

从切片中提取实体和关系，写入 Neo4j 和 Milvus。

代码结构：
    1. LLM 提示词
    2. 主流程 (process → _process_single_chunk)
    3. LLM 提取
    4. JSON 解析与清洗
    5. Milvus 写入
    6. Neo4j 写入（Cypher 常量 + 原子操作）
"""

import json
import re
from typing import Dict, List, Any, Set, Optional

from knowledge.processor.import_process.base import BaseNode, setup_logging
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.config import get_config


from knowledge.tools.llm_utils import get_llm_client
from knowledge.tools.embedding_utils import (
    get_openai_embeddings_client, embed_text_to_dense_vector,
    get_bge_m3_model, generate_hybrid_embeddings,
)
from langchain_core.messages import SystemMessage, HumanMessage
from pymilvus import MilvusClient, DataType


class KnowledgeGraphNode(BaseNode):
    """知识图谱构建节点。

    对每个文本切片执行：LLM 实体/关系提取 → JSON 清洗 → Milvus 写入 → Neo4j 写入。
    """

    name = "knowledge_graph"

    # 实体名最大长度（超过则触发截断）
    MAX_ENTITY_NAME_LENGTH = 20

    # ================================================================== #
    #                       1. LLM 提取提示词                              #
    # ================================================================== #

    SYSTEM_PROMPT = """你是知识图谱信息抽取器。给你一段设备操作手册的文本切片，你必须抽取实体与关系，并只输出一个 JSON 对象（不要输出解释、不要 Markdown）。

## 允许的实体类型（label）
- Device：设备整体（如"万用表""仪表"）
- Part：部件或零件（如"电池后盖""螺母""表笔"）
- Operation：操作/功能名称（如"电池安装""电阻测量"），通常对应章节标题
- Step：操作步骤，name 用"步骤N-动作短语"格式（如"步骤1-断开表笔"），description 存原文
- Warning：警告/注意事项，name 用"警告-核心要点"格式（如"警告-操作前断开电源"），description 存原文
- Condition：前置条件或约束（如"电阻小于30Ω"）
- Tool：工具（如"螺丝刀"）

## 实体命名规则（非常重要）
- name 必须简短，不超过15个字。这是硬性要求。
- 禁止将整句原文作为 name。
- Step 格式：name="步骤N-动作短语"，description="原文完整步骤"
- Warning 格式：name="警告-核心要点"，description="原文完整警告"
- 同名同类型的实体只保留一个，不要重复。

## 允许的关系类型（type）
- HAS_OPERATION：Device → Operation
- HAS_PART：Device → Part
- HAS_STEP：Operation → Step
- USES_TOOL：Step → Tool
- HAS_WARNING：Operation/Step → Warning
- NEXT_STEP：Step → Step（按步骤顺序串联）
- AFFECTS：Step → Part（该步骤操作了哪个部件）
- REQUIRES：Step/Operation → Condition

## 抽取原则
- 只抽取文本中明确出现或可直接对应的实体与关系，禁止臆造。
- 步骤编号(1/2/3)时：每条作为 Step，并按顺序生成 NEXT_STEP 关系链。
- 关系的 head 和 tail 必须使用实体的 name 值（简短名），不要用 description。
- 如果无法判断某个关系，不要输出该关系。

## 输出 JSON Schema
{
  "entities": [
    {"name": "简短名称", "label": "类型", "description": "可选，原文内容或补充说明"}
  ],
  "relations": [
    {"head": "头实体name", "tail": "尾实体name", "type": "关系类型"}
  ]
}
"""

    # ================================================================== #
    #                        2. 主流程                                     #
    # ================================================================== #

    def process(self, state: ImportGraphState) -> ImportGraphState:
        """执行知识图谱构建。"""
        chunks = state.get("chunks", [])
        if not chunks:
            self.logger.info("chunks 为空，跳过知识图谱构建")
            return state

        self.log_step("start", f"开始处理 {len(chunks)} 个切片")

        for i, chunk in enumerate(chunks):
            if not isinstance(chunk, dict):
                continue

            content = chunk.get("content", "")
            chunk_id = str(chunk.get("chunk_id", f"temp_{i}"))
            item_name = chunk.get("item_name") or state.get("item_name", "")

            if not content or not item_name:
                continue

            self.logger.debug(f"处理切片 {i + 1}/{len(chunks)}: {chunk_id}")
            self._process_single_chunk(content, chunk_id, item_name)

        self.log_step("end", "知识图谱构建完成")
        return state

    def _process_single_chunk(self, content: str, chunk_id: str, item_name: str):
        """处理单个切片：提取 → 清洗 → 写入。"""
        config = get_config()

        # 提取
        raw_response = self._llm_extract(content)
        if not raw_response:
            return

        # 清洗
        graph_data = self._parse_and_clean(raw_response)
        if not graph_data.get("entities"):
            return

        self.logger.info(
            f"切片 {chunk_id}: "
            f"提取到 {len(graph_data['entities'])} 个实体, "
            f"{len(graph_data['relations'])} 条关系"
        )

        # 写入
        self._save_entities_to_milvus(
            graph_data.get("entities", []), chunk_id, content, item_name, config,
        )
        self._save_graph_to_neo4j(graph_data, chunk_id, item_name, config)

    # ================================================================== #
    #                      3. LLM 提取                                    #
    # ================================================================== #

    def _llm_extract(self, content: str) -> str:
        """调用 LLM 提取实体和关系。"""
        try:
            llm = get_llm_client()
            response = llm.invoke([
                SystemMessage(content=self.SYSTEM_PROMPT),
                HumanMessage(content=f"请处理以下文本切片：\n\n{content}"),
            ])
            return (response.content or "").strip()
        except Exception as e:
            self.logger.warning(f"LLM 提取失败: {e}")
            return ""

    # ================================================================== #
    #                   4. JSON 解析与清洗                                 #
    # ================================================================== #

    # 关系类型白名单
    ALLOWED_RELATION_TYPES = {
        "HAS_OPERATION", "HAS_PART", "HAS_STEP", "USES_TOOL",
        "HAS_WARNING", "NEXT_STEP", "AFFECTS", "REQUIRES",
        "MENTIONED_IN", "RELATED_TO",
    }

    def _parse_and_clean(self, raw_text: str) -> Dict[str, Any]:
        """解析 LLM 返回的 JSON 并执行清洗。"""
        if not raw_text:
            return {"entities": [], "relations": []}

        # 去除 Markdown 代码围栏
        cleaned_text = re.sub(r"^```(?:json)?\s*", "", raw_text.strip())
        cleaned_text = re.sub(r"\s*```$", "", cleaned_text)

        try:
            data = json.loads(cleaned_text)
        except json.JSONDecodeError as e:
            self.logger.warning(f"JSON 解析失败: {e}, 原文前200字: {raw_text[:200]}")
            return {"entities": [], "relations": []}

        cleaned_entities = self._clean_entities(data.get("entities", []))
        valid_names = {e["name"] for e in cleaned_entities}
        cleaned_relations = self._clean_relations(data.get("relations", []), valid_names)

        return {"entities": cleaned_entities, "relations": cleaned_relations}

    # ---------- 实体清洗 ----------

    def _clean_entities(self, entities: List[Dict]) -> List[Dict]:
        """清洗实体：过滤无效项、截断过长名称、去重。"""
        seen: Set[tuple] = set()
        cleaned: List[Dict] = []

        for entity in entities:
            name = str(entity.get("name", "")).strip()
            label = str(entity.get("label", "")).strip()
            description = str(entity.get("description", "")).strip()

            if not name or not label:
                continue

            dedup_key = (name, label)
            if dedup_key in seen:
                continue
            seen.add(dedup_key)

            result = {"name": name, "label": label}
            if description:
                result["description"] = description
            cleaned.append(result)

        return cleaned

    # ---------- 关系清洗 ----------

    def _clean_relations(
        self,
        relations: List[Dict],
        valid_entity_names: Set[str],
    ) -> List[Dict]:
        """清洗关系：修正字段、白名单校验、过滤悬空引用。"""
        cleaned: List[Dict] = []

        for rel in relations:
            head = str(rel.get("head", "")).strip()
            tail = str(rel.get("tail", "")).strip()
            if not head or not tail:
                continue

            rel_type = str(rel.get("type") or rel.get("label") or "RELATED_TO").strip()
            if rel_type not in self.ALLOWED_RELATION_TYPES:
                rel_type = "RELATED_TO"

            if head not in valid_entity_names or tail not in valid_entity_names:
                self.logger.debug(f"悬空关系已跳过: {head} -[{rel_type}]-> {tail}")
                continue

            cleaned.append({"head": head, "tail": tail, "type": rel_type})

        return cleaned

    # ================================================================== #
    #                     5. Milvus 写入                                  #
    # ================================================================== #

    def _save_entities_to_milvus(
        self,
        entities: List[Dict],
        chunk_id: str,
        content: str,
        item_name: str,
        config,
    ):
        """将实体向量化并写入 Milvus（稠密 + 稀疏双向量）。"""
        if not entities or not config.entity_name_collection:
            return

        dedup_map = self._dedup_entities_by_name(entities)
        if not dedup_map:
            return

        try:
            bge_ef = get_bge_m3_model()
            milvus_client = MilvusClient(uri=config.milvus_url)
            collection_name = config.entity_name_collection

            self._ensure_entity_collection(milvus_client, collection_name)

            names = list(dedup_map.keys())

            vectors = bge_ef.encode_documents(names)
            insert_data = self._build_milvus_records(
                names, vectors, chunk_id, content, item_name,
            )
            if insert_data:
                milvus_client.insert(collection_name=collection_name, data=insert_data)
                milvus_client.load_collection(collection_name=collection_name)
                self.logger.debug(f"写入 {len(insert_data)} 个实体到 Milvus")

        except Exception as e:
            self.logger.warning(f"Milvus 写入失败: {e}")

    @staticmethod
    def _dedup_entities_by_name(entities: List[Dict]) -> Dict[str, Dict]:
        """按 name 去重，合并同名不同类型的 label。"""
        dedup: Dict[str, Dict] = {}
        for entity in entities:
            name = str(entity.get("name", "")).strip()
            if not name:
                continue
            label = str(entity.get("label", "")).strip()
            description = str(entity.get("description", "")).strip()

            if name not in dedup:
                dedup[name] = {"labels": set(), "description": description}
            if label:
                dedup[name]["labels"].add(label)
        return dedup

    @staticmethod
    def _ensure_entity_collection(client: MilvusClient, collection_name: str):
        """确保实体集合存在，不存在则创建完整 schema 和索引。"""
        if client.has_collection(collection_name):
            return

        schema = client.create_schema(enable_dynamic_field=True)
        schema.add_field(field_name="pk",              datatype=DataType.INT64,
                         is_primary=True, auto_id=True)
        schema.add_field(field_name="entity_name",     datatype=DataType.VARCHAR,
                         max_length=65535)
        schema.add_field(field_name="dense_vector",    datatype=DataType.FLOAT_VECTOR,
                         dim=1024)
        schema.add_field(field_name="sparse_vector",   datatype=DataType.SPARSE_FLOAT_VECTOR)
        schema.add_field(field_name="source_chunk_id", datatype=DataType.VARCHAR,
                         max_length=65535)
        schema.add_field(field_name="context",         datatype=DataType.VARCHAR,
                         max_length=65535)
        schema.add_field(field_name="item_name",       datatype=DataType.VARCHAR,
                         max_length=65535)

        index_params = client.prepare_index_params()
        index_params.add_index(
            field_name="dense_vector",
            index_name="dense_vector_index",
            index_type="IVF_FLAT",
            metric_type="COSINE",
            params={"nlist": 128},
        )
        index_params.add_index(
            field_name="sparse_vector",
            index_name="sparse_vector_index",
            index_type="SPARSE_INVERTED_INDEX",
            metric_type="IP",
        )

        client.create_collection(
            collection_name=collection_name,
            schema=schema,
            index_params=index_params,
        )

    @staticmethod
    def _build_milvus_records(
        names: List[str],
        vectors: Dict[str, Any],
        chunk_id: str,
        content: str,
        item_name: str,
    ) -> List[Dict]:
        """组装 Milvus 插入记录（稠密 + 稀疏双向量）。"""
        from knowledge.tools.embedding_utils import normalize_sparse_vector

        dense_list = vectors.get("dense", [])
        sparse_raw = vectors.get("sparse")

        records = []
        for idx, name in enumerate(names):
            if idx >= len(dense_list):
                break

            dense_vector = dense_list[idx]
            if hasattr(dense_vector, "tolist"):
                dense_vector = dense_vector.tolist()

            record = {
                "entity_name": name,
                "dense_vector": dense_vector,
                "source_chunk_id": chunk_id,
                "context": content[:200],
                "item_name": item_name,
            }

            if sparse_raw is not None:
                start, end = sparse_raw.indptr[idx], sparse_raw.indptr[idx + 1]
                indices = sparse_raw.indices[start:end].tolist()
                data = sparse_raw.data[start:end].tolist()
                sparse_vector = {k: v for k, v in zip(indices, data)}
                record["sparse_vector"] = normalize_sparse_vector(sparse_vector)

            records.append(record)

        return records

    # ================================================================== #
    #                      6. Neo4j 写入                                  #
    # ================================================================== #

    # ---------- Cypher 常量 ----------

    CYPHER_MERGE_CHUNK = """
        MERGE (c:Chunk {id: $chunk_id, item_name: $item_name})
    """

    CYPHER_MERGE_ENTITY = """
        MERGE (n:Entity {name: $name, item_name: $item_name})
        ON CREATE SET
            n.source_chunk_id = $chunk_id,
            n.description     = $description,
            n.types           = CASE
                                    WHEN $label = "" THEN []
                                    ELSE [$label]
                                END
        ON MATCH SET
            n.description = CASE
                                WHEN $description <> "" THEN $description
                                ELSE coalesce(n.description, "")
                            END,
            n.types       = CASE
                                WHEN $label = ""                       THEN coalesce(n.types, [])
                                WHEN $label IN coalesce(n.types, [])   THEN n.types
                                ELSE coalesce(n.types, []) + $label
                            END
    """

    CYPHER_LINK_ENTITY_TO_CHUNK = """
        MATCH (n:Entity {name: $name, item_name: $item_name})
        MATCH (c:Chunk  {id: $chunk_id, item_name: $item_name})
        MERGE (n)-[:MENTIONED_IN]->(c)
    """

    CYPHER_MERGE_RELATION_TEMPLATE = """
        MATCH (h:Entity {{name: $head, item_name: $item_name}})
        MATCH (t:Entity {{name: $tail, item_name: $item_name}})
        MERGE (h)-[:{rel_type}]->(t)
    """

    # ---------- 入口 ----------

    def _save_graph_to_neo4j(self, graph_data, chunk_id, item_name, config):
        """将图数据保存到 Neo4j。"""
        entities = graph_data.get("entities", [])
        relations = graph_data.get("relations", [])

        if not entities:
            return

        try:
            from knowledge.tools.neo4j_utils import get_neo4j_driver
            driver = get_neo4j_driver()

            with driver.session(database=config.neo4j_database) as session:
                session.execute_write(
                    self._write_graph_in_tx,
                    entities, relations, chunk_id, item_name,
                )

            self.logger.debug(
                f"写入 {len(entities)} 个实体, {len(relations)} 条关系到 Neo4j"
            )
        except Exception as e:
            self.logger.warning(f"Neo4j 写入失败: {e}")

    # ---------- 事务总控 ----------

    @staticmethod
    def _write_graph_in_tx(tx, entities, relations, chunk_id, item_name):
        """在单个事务内完成所有写入。"""
        cls = KnowledgeGraphNode

        cls._tx_merge_chunk(tx, chunk_id, item_name)

        for entity in entities:
            name = str(entity.get("name", "")).strip()
            if not name:
                continue
            label = str(entity.get("label", "")).strip()
            description = str(entity.get("description", "")).strip()

            cls._tx_merge_entity(tx, name, label, description, chunk_id, item_name)
            cls._tx_link_entity_to_chunk(tx, name, chunk_id, item_name)

        for rel in relations:
            head = str(rel.get("head", "")).strip()
            tail = str(rel.get("tail", "")).strip()
            if not head or not tail:
                continue
            rel_type = str(rel.get("type", "RELATED_TO")).strip() or "RELATED_TO"
            cls._tx_merge_relation(tx, head, tail, rel_type, item_name)

    # ---------- 原子操作 ----------

    @staticmethod
    def _tx_merge_chunk(tx, chunk_id, item_name):
        tx.run(KnowledgeGraphNode.CYPHER_MERGE_CHUNK,
               chunk_id=chunk_id, item_name=item_name)

    @staticmethod
    def _tx_merge_entity(tx, name, label, description, chunk_id, item_name):
        tx.run(KnowledgeGraphNode.CYPHER_MERGE_ENTITY,
               name=name, label=label, description=description,
               chunk_id=chunk_id, item_name=item_name)

    @staticmethod
    def _tx_link_entity_to_chunk(tx, name, chunk_id, item_name):
        tx.run(KnowledgeGraphNode.CYPHER_LINK_ENTITY_TO_CHUNK,
               name=name, chunk_id=chunk_id, item_name=item_name)

    @staticmethod
    def _tx_merge_relation(tx, head, tail, rel_type, item_name):
        if rel_type not in KnowledgeGraphNode.ALLOWED_RELATION_TYPES:
            rel_type = "RELATED_TO"
        cypher = KnowledgeGraphNode.CYPHER_MERGE_RELATION_TEMPLATE.format(
            rel_type=rel_type
        )
        tx.run(cypher, head=head, tail=tail, item_name=item_name)


# ================================================================== #
#                        兼容 & 测试                                   #
# ================================================================== #

node_knowledge_graph = KnowledgeGraphNode()


def test_kg_extraction():
    """测试：模拟单个切片，跑通 LLM 提取 → 解析清洗全流程。"""
    print("=== 开始测试知识图谱构建流程 ===\n")

    mock_state = {
        "chunks": [
            {
                "content": """# 电池安装
警告: 为防触电, 打开电池后盖前后，请勿操作仪表并把表笔与电源断开。
1. 把表笔与仪表断开。
2. 用螺丝刀拧开电池后盖上的螺母。
3. 正确安装电池，正负极应一致。
4. 盖上电池后盖并拧紧螺丝钉。
警告: 为防触电,在电池后盖安装和固定之前，请勿操作仪表。
注意: 若仪表出现工作不正常，请检测保险丝和电池是否完好以及是否放在正确的位置。""",
                "chunk_id": "chunk_test_001",
                "item_name": "万用表",
            }
        ]
    }

    node_knowledge_graph.process(mock_state)
    print("\n=== 测试完成 ===")


if __name__ == "__main__":
    setup_logging()
    test_kg_extraction()

---

5. 单元测试

在 knowledge_graph.py 文件末尾已包含测试代码，下面详细说明测试流程。

5.1 测试代码说明

测试代码模拟了一个包含"电池安装"操作步骤的文本切片，覆盖了以下测试场景：

• 多类型实体：Device（万用表）、Operation（电池安装）、Step（步骤1-4）、Warning（警告）、Part（电池后盖、螺母）、Tool（螺丝刀）
• NEXT_STEP 链：步骤1 → 步骤2 → 步骤3 → 步骤4 的顺序关系
• 多种关系类型：HAS_OPERATION、HAS_STEP、USES_TOOL、HAS_WARNING、AFFECTS 等
• 完整流程：LLM 提取 → JSON 清洗 → Milvus 写入 → Neo4j 写入

if __name__ == "__main__":
    setup_logging()
    test_kg_extraction()

5.2 测试执行

# 进入项目目录
cd D:\develop\develop\workspace\pycharm\usage\shopkeeper_brain_v260213

# 执行测试
python -m knowledge.processor.import_process.nodes.knowledge_graph

5.3 预期输出

=== 开始测试知识图谱构建流程 ===

2026-xx-xx - import.knowledge_graph - INFO - --- knowledge_graph 开始 ---
2026-xx-xx - import.knowledge_graph - INFO - [start] 开始处理 1 个切片
2026-xx-xx - import.knowledge_graph - INFO - 切片 chunk_test_001: 提取到 12 个实体, 15 条关系
2026-xx-xx - import.knowledge_graph - INFO - [end] 知识图谱构建完成
2026-xx-xx - import.knowledge_graph - INFO - --- knowledge_graph 完成 ---

=== 测试完成 ===

5.4 验证 Neo4j 数据

测试完成后，可以使用 Neo4j Browser 验证图数据：

1. 打开浏览器访问 http://localhost:7474（Neo4j Browser）
2. 使用 Cypher 查询查看写入的图结构：

// 查看所有实体
MATCH (n:Entity {item_name: "万用表"}) RETURN n

// 查看某个操作的完整步骤链
MATCH (op:Entity {name: "电池安装"})-[:HAS_STEP]->(s:Entity)
RETURN op, s

// 查看步骤顺序
MATCH (s1:Entity)-[:NEXT_STEP]->(s2:Entity)
WHERE s1.item_name = "万用表"
RETURN s1.name, s2.name

// 查看完整图结构
MATCH (n)-[r]->(m)
WHERE n.item_name = "万用表"
RETURN n, r, m

---

6. 总结

6.1 节点功能概览

功能模块	说明
LLM 实体抽取	通过精心设计的 Prompt 驱动 LLM 从文本中提取实体和关系
JSON 清洗	去围栏、过滤无效项、截断名称、去重、白名单校验、悬空引用过滤
Milvus 实体写入	实体名称 BGE-M3 向量化 → 创建集合 → 批量插入（稠密+稀疏双向量）
Neo4j 图写入	事务内完成 Chunk 节点 → Entity 节点 → MENTIONED_IN 关联 → 实体间关系

6.2 设计要点

1. Prompt 工程是核心

   • 精确定义实体类型、关系类型、命名规则
   • 约束输出格式（纯 JSON，无 Markdown）
   • 提供清晰的 JSON Schema 示例

2. 多层清洗保障数据质量

   • Markdown 围栏去除 → JSON 解析 → 实体清洗 → 关系清洗
   • 每一层都有明确的过滤规则和降级策略

3. 白名单防注入

   • 关系类型必须通过白名单校验才能拼入 Cypher
   • 非法类型统一降级为 RELATED_TO

4. 事务原子性

   • 使用 session.execute_write 在单个事务内完成所有 Neo4j 操作
   • 保证一个切片的图数据要么全部写入，要么全部回滚

5. 幂等性设计

   • MERGE 语句保证重复执行不会创建重复节点或关系
   • _ensure_entity_collection 检测集合存在性后再创建

6. 双写架构

   • 实体名称 → Milvus（支持向量检索，用于图增强 RAG）
   • 实体+关系 → Neo4j（支持图遍历查询，用于结构化推理）

---

企业痛点映射

痛点	传统方案	AI Agent 知识图谱方案	效率提升
非结构化文档无法精准检索	关键词搜索匹配整段文本	实体 + 关系图结构 + 向量检索	检索精度提升 ~100%（实体级精确命中）
设备手册的步骤关系难维护	人工阅读并手写关系	LLM 自动提取实体和关系链（NEXT_STEP）	每份文档从 2h 降至 ~2min
LLM 输出不可靠需清洗	肉眼逐条审核 JSON	多层自动清洗（去围栏 + 实体去重 + 关系白名单 + 悬空过滤）	清洗准确率 ~95%
实体名重复导致图数据冗余	查重全靠人工	MERGE 幂等 + (name, label) 去重	冗余节点数降为 ~0
关系类型注入安全风险	直接拼接用户输入到 Cypher	白名单校验 + 非法类型降级 RELATED_TO	注入风险降为 ~0%
图谱写入中途失败数据不一致	手动清理已写入的数据	execute_write 事务原子性保障	写失败回滚率 ~100%

---

Remote & Agent 应用场景价值

• Remote 场景价值：KnowledgeGraphNode 同时依赖 LLM API、Milvus 和 Neo4j 三个远程服务，全部通过 .env 配置。远程团队成员只需配置环境变量即可在本地运行完整的图谱构建流程。execute_write 事务机制确保分布式环境下即使网络中断也不会产生脏数据。

• Agent 落地场景：KnowledgeGraphNode 是"文档导入 Agent"的最后一个也是最复杂的子 Agent。可拆分为三个子 Agent：(1) 实体提取 Agent——调用 LLM 提取实体关系 (2) 数据清洗 Agent——JSON 解析 + 过滤 + 白名单校验 (3) 图存储 Agent——写入 Milvus + Neo4j。三个子 Agent 通过 Pipeline 编排，每步输出是下一步的输入。

---

Git Commit 对应

本节知识图谱构建节点对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议在项目仓库中搜索 "knowledge_graph.py" 相关提交>

cd shopkeeper_brain
git log --oneline --all -- knowledge/processor/import_process/nodes/knowledge_graph.py