阶段三：导入流程 — 知识图谱（对应课程 day08~10）

当前状态：AI 处理链路已就绪（7 节点）→ 新增知识图谱节点 + Neo4J 基础

知识标记总览：

知识点	出现次数	扮演的角色
Neo4J 图数据库	🔴 第 1 次（全新）	实体-关系存储
Cypher 查询语言	🔴 第 1 次（全新）	图数据库的 SQL
GraphRAG vs 普通 RAG	🔴 第 1 次（全新）	两种知识组织方式的对比
LLM 抽取实体关系	🟢 第 2 次（类似 Ch17 商品名识别）	用 LLM 从文本中提取结构化信息

---

3.1 Neo4J 快速入门（🔴 全新知识）

3.1.1 四个核心概念

概念	类比 SQL	类比电商场景
节点 (Node)	表中的一行记录	一个商品实体
关系 (Relationship)	外键连接	"包含关系"、"关联产品"
标签 (Label)	表名	:Product、:Brand、:Category
属性 (Property)	列字段	name、price、brand

3.1.2 Cypher 基础语法

// 创建节点
CREATE (n:Product {name: '万用表', price: 129})
CREATE (n:Brand {name: '胜利仪器'})

// 创建关系
MATCH (a:Product {name: '万用表'})
MATCH (b:Brand {name: '胜利仪器'})
CREATE (a)-[:MADE_BY]->(b)

// 查询：找出所有胜利仪器品牌的商品
MATCH (b:Brand {name: '胜利仪器'})<-[:MADE_BY]-(p:Product)
RETURN p.name, p.price

// 查询：找到"万用表"的关联品类
MATCH (p:Product {name: '万用表'})-[:RELATED_TO]->(related)
RETURN related.name

3.1.3 普通 RAG vs GraphRAG

对比维度	普通 RAG（Ch16 方式）	GraphRAG（本项目方式）
数据组织	文档切片（扁平的向量列表）	实体-关系图（结构化的知识网络）
检索方式	向量余弦相似度	Cypher 关系遍历
能回答的问题	"万用表怎么使用？"（基于文档段落）	"万用表和哪些品类相关？"（基于关系链接）
局限性	无法回答多跳关系问题	需要先准确提取实体和关系
互补性	语义匹配强	结构化关系强

---

3.2 阶段起始状态

---

3.3 核心代码

# → knowledge/processor/import_process/nodes/kg_graph_node.py
import json
from knowledge.processor.import_process.base import BaseNode
from knowledge.processor.import_process.state import ImportGraphState

class KnowledgeGraphNode(BaseNode):
    """用 LLM 抽取实体和关系，写入 Neo4J 图数据库"""
    
    name = "kg_node"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        chunks = state.get('chunks', [])
        llm_client = LLMClient(self.config)
        neo4j_util = Neo4jUtil(self.config)
        
        # 1. 每批 chunks 调用 LLM 抽取实体和关系
        for i in range(0, len(chunks), 3):  # 每批 3 个 chunks
            batch = chunks[i:i+3]
            context = "\n".join(c['content'] for c in batch)
            
            # LLM 抽取（返回 JSON）
            prompt = self._build_extract_prompt(context)
            response = llm_client.chat([
                {"role": "system", "content": "抽取实体和关系，以 JSON 返回。"},
                {"role": "user", "content": prompt}
            ])
            
            try:
                data = json.loads(response)
                entities = data.get('entities', [])
                relations = data.get('relations', [])
                
                # 2. 写入 Neo4J
                for entity in entities:
                    neo4j_util.create_node(
                        label=entity['type'],    # Product / Brand / Category
                        name=entity['name'],
                        properties=entity.get('properties', {})
                    )
                
                for rel in relations:
                    neo4j_util.create_relationship(
                        from_label=rel['from_type'],
                        from_name=rel['from_name'],
                        to_label=rel['to_type'],
                        to_name=rel['to_name'],
                        rel_type=rel['relation']  # MADE_BY / RELATED_TO
                    )
            except json.JSONDecodeError:
                self.logger.warning(f"LLM 返回非 JSON: {response}")
                continue
        
        return state
    
    def _build_extract_prompt(self, context: str) -> str:
        return f"""从以下文档中抽取商品实体和关系。

输出格式：
{{
  "entities": [
    {{"type": "Product", "name": "万用表", "properties": {{"price": 129}}}}
  ],
  "relations": [
    {{"from_type": "Product", "from_name": "万用表", "to_type": "Brand", "to_name": "胜利仪器", "relation": "MADE_BY"}}
  ]
}}

文档内容：
{context}"""


# → knowledge/utils/neo4j_util.py
from neo4j import GraphDatabase

class Neo4jUtil:
    def __init__(self, config):
        self.driver = GraphDatabase.driver(
            config.neo4j_uri,
            auth=(config.neo4j_username, config.neo4j_password)
        )
    
    def create_node(self, label: str, name: str, properties: dict = None):
        with self.driver.session() as session:
            props = properties or {}
            props['name'] = name
            session.run(
                f"MERGE (n:{label} {{name: $name}}) SET n += $props",
                name=name, props=props
            )
    
    def create_relationship(self, from_label, from_name, to_label, to_name, rel_type):
        with self.driver.session() as session:
            session.run(f"""
                MATCH (a:{from_label} {{name: $from_name}})
                MATCH (b:{to_label} {{name: $to_name}})
                MERGE (a)-[:{rel_type}]->(b)
            """, from_name=from_name, to_name=to_name)
    
    def query(self, cypher: str, params: dict = None) -> list:
        with self.driver.session() as session:
            result = session.run(cypher, params or {})
            return [record.data() for record in result]

---

3.4 设计决策

决策	选项	理由
每批 3 个 chunks 调用一次 LLM	1 vs 3 vs 全文	3 个 chunks 够上下文又不超 token 限制
MERGE 而非 CREATE	MERGE / CREATE	MERGE 可防止重复节点，幂等
name 作为唯一标识	name / 独立 ID	电商领域 name 天然唯一，省去 ID 映射复杂度
用 LLM 抽取实体关系	LLM / 规则 / NER 模型	LLM 对开放域实体抽取最灵活，无需训练

---

3.5 导入流程最终架构图

导入流程至此全部完成（8 个节点）。从第 day11 开始进入查询流程。

📂 对应的原始代码快照：day08/ → day10/ 之间的逐日增量（含 Neo4J 入门和知识图谱节点）

---

3.6 验证命令

# 1. 验证 Neo4J 已启动
curl -s http://localhost:7474 | grep -q "neo4j" && echo "✅ Neo4J OK"

# 2. 验证图谱有数据
python -c "
from knowledge.utils.neo4j_util import Neo4jUtil
from knowledge.processor.import_process.config import get_config
util = Neo4jUtil(get_config())
nodes = util.query('MATCH (n) RETURN count(n) as cnt')
print(f'✅ 图谱中实体数: {nodes[0][\"cnt\"]}')
"

# 3. 验证关联查询
python -c "
result = util.query('MATCH (p:Product)-[:MADE_BY]->(b:Brand) RETURN p.name, b.name LIMIT 5')
print(f'✅ 产品-品牌关联数: {len(result)}')
"