知识图谱

// 唯一性约束：同类型节点按 name 去重（同时自动建索引）
CREATE CONSTRAINT company_name IF NOT EXISTS
FOR (c:Company) REQUIRE c.name IS UNIQUE;

CREATE CONSTRAINT product_id IF NOT EXISTS
FOR (p:Product) REQUIRE p.id IS UNIQUE;

CREATE CONSTRAINT material_name IF NOT EXISTS
FOR (m:Material) REQUIRE m.name IS UNIQUE;

docker run -d --name neo4j \
  -p 7474:7474 -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/your-password \
  neo4j:5

# 浏览器界面（可视化查询）: http://localhost:7474
# 程序连接地址（Bolt）:       bolt://localhost:7687

// 写入一：CREATE，适合一次性手工造数据
CREATE (:Company {name: "华芯科技", country: "CN"})
CREATE (:Product  {name: "电源管理芯片", id: "PM-8321"})

// 写入二：MERGE，幂等写入（查到即复用，查不到才创建）
// 生产管道只用它：重复执行不产生重复数据
MERGE (c:Company {name: "华芯科技"})
MERGE (p:Product {name: "电源管理芯片"})
MERGE (c)-[s:SUPPLIES]->(p)
ON CREATE SET s.since = 2021, s.share = 0.35

// 查询：MATCH 声明模式，WHERE 过滤，RETURN 投影
MATCH (c:Company)-[:SUPPLIES]->(p:Product)
WHERE c.country = "CN"
RETURN c.name, collect(p.name) AS products
ORDER BY size(products) DESC
LIMIT 5

from typing import Literal
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

llm = init_chat_model("openai:gpt-4o-mini")

NODE_TYPES = Literal["Company", "Product", "Material", "Country", "Region"]
REL_TYPES = Literal["SUPPLIES", "REQUIRES", "PRODUCED_IN",
                    "LOCATED_IN", "ALTERNATIVE_TO"]

class Triple(BaseModel):
    subject: str = Field(description="主体实体名，使用规范全称")
    subject_type: NODE_TYPES
    predicate: REL_TYPES
    object: str = Field(description="客体实体名，使用规范全称")
    object_type: NODE_TYPES
    evidence: str = Field(description="支撑该三元组的原文片段")

class Extraction(BaseModel):
    triples: list[Triple] = Field(description="文中全部符合本体的三元组")

extractor = llm.with_structured_output(Extraction)

def triples_from(text: str) -> list[Triple]:
    return extractor.invoke(
        "从文本抽取符合本体约束的三元组。只抽取文本明确陈述的事实，"
        "不要推断；无符合项则返回空列表。\n\n" + text
    ).triples

MERGE_TMPL = """
MERGE (s:{stype} {{name: $subject}})
MERGE (o:{otype} {{name: $object}})
MERGE (s)-[r:{pred}]->(o)
ON CREATE SET r.evidence = $evidence
"""

def write_triples(tx, t: Triple):
    tx.run(
        MERGE_TMPL.format(stype=t.subject_type, otype=t.object_type,
                          pred=t.predicate),
        subject=t.subject, object=t.object, evidence=t.evidence,
    )
# with driver.session() as s: s.execute_write(write_triples, t)

from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_neo4j import Neo4jGraph

transformer = LLMGraphTransformer(
    llm=init_chat_model("openai:gpt-4o-mini"),
    allowed_nodes=["Company", "Product", "Material", "Country", "Region"],
    allowed_relationships=[
        "SUPPLIES", "REQUIRES", "PRODUCED_IN", "LOCATED_IN", "ALTERNATIVE_TO",
    ],
    node_properties=["country"],
    relationship_properties=["since", "share"],
)

graph_docs = transformer.convert_to_graph_documents(docs)

neo4j = Neo4jGraph(url="bolt://localhost:7687",
                   username="neo4j", password="your-password")
neo4j.add_graph_documents(graph_docs, include_source=True)
# include_source：每个抽取节点连回原文档节点，图谱事实可溯源

// 稀土镨断供会波及哪些公司？（沿 SUPPLIES / REQUIRES 反向最多 3 跳）
MATCH path = (c:Company)-[:SUPPLIES|REQUIRES*1..3]->(m:Material {name: "稀土镨"})
RETURN c.name AS company,
       [n IN nodes(path) | coalesce(n.name, n.id)] AS chain,
       length(path) AS hops
ORDER BY hops

// 每个国家掌握的材料与产品面：分组聚合一次完成
MATCH (c:Company)-[:SUPPLIES]->(p:Product)-[:REQUIRES]->(m:Material)
RETURN c.country AS country,
       count(DISTINCT p) AS product_count,
       collect(DISTINCT m.name)[..5] AS top_materials
ORDER BY product_count DESC
LIMIT 10

// 两家公司之间最短的关联路径（最多 5 跳，任意关系方向）
MATCH p = shortestPath(
  (a:Company {name: "华芯科技"})-[*..5]-(b:Company {name: "北方精密"})
)
RETURN [n IN nodes(p) | n.name] AS path, length(p) AS hops

from langchain_neo4j import Neo4jGraph
from langchain.chat_models import init_chat_model

neo4j = Neo4jGraph(url="bolt://localhost:7687",
                   username="neo4j", password="your-password")
llm = init_chat_model("openai:gpt-4o-mini")

def graph_context(entity: str, limit: int = 50) -> str:
    """以实体为中心展开 1 跳邻域，拼成三元组文本。"""
    rows = neo4j.query(
        """
        MATCH (e {name: $entity})-[r]-(n)
        RETURN e.name AS s, type(r) AS p, n.name AS o
        LIMIT $limit
        """,
        {"entity": entity, "limit": limit},
    )
    return "\n".join(f"{row['s']} --{row['p']}--> {row['o']}" for row in rows)

def answer_from_graph(question: str, entity: str) -> str:
    ctx = graph_context(entity)
    return llm.invoke(
        "依据以下知识图谱事实回答问题；图谱未覆盖的部分明确说明。\n\n"
        "图谱事实：\n" + ctx + f"\n\n问题：{question}"
    ).content

from langchain_neo4j import GraphCypherChain

qa = GraphCypherChain(
    graph=neo4j,
    llm=init_chat_model("openai:gpt-4o-mini"),
    allow_dangerous_operations=False,   # 默认拒绝写操作，保持开启
)

result = qa.invoke({"query": "哪些公司间接依赖稀土镨？列出传导链。"})
# 内部流程：读 graph.schema（节点/关系类型）→ 生成 Cypher → 执行 → 生成答案

def retrieve(question: str) -> str:
    """按问题类型路由到不同检索器（分类器用结构化输出实现）。"""
    kind = classify(question)                 # fact / relation / global
    if kind == "fact":
        return vector_context(question)       # 第 3 章的混合检索管道
    if kind == "relation":
        entities = link_entities(question)    # 向量库召回实体名
        return "\n".join(graph_context(e) for e in entities)
    return community_context(question)        # 社区摘要

// 规范名 + 别名表：抽取与查询的统一入口
MATCH (c:Company {name: "华芯科技"})
SET c.aliases = ["华芯", "HX Tech", "华芯半导体"];

// 实体链接：按规范名或别名都能命中
MATCH (c:Company)
WHERE c.name = $name OR $name IN c.aliases
RETURN c.name AS canonical;

// 为 Document 节点的嵌入建向量索引
CREATE VECTOR INDEX doc_chunks IF NOT EXISTS
FOR (d:Document) ON (d.embedding)
OPTIONS {indexConfig: {
  `vector.dimensions`: 1536,
  `vector.similarity_function`: "cosine"
}}

// 相似检索：图与向量一次查询打通
CALL db.index.vector.queryNodes("doc_chunks", 5, $query_vector)
YIELD node, score
MATCH (node)-[:MENTIONS]->(e:Entity)
RETURN node.text, score, collect(e.name) AS entities