手把手教你用Neo4j+Python+Claude,3小时搞定一个供应链知识图谱

1  数据源梳理 → 本体设计 → 知识抽取 → 知识融合 → 图谱存储 → 查询应用

 1  Thing 2  ├── Organization 3  │   ├── Supplier          # 供应商 4  │   ├── Manufacturer      # 制造商 5  │   └── Distributor       # 分销商 6  ├── Product 7  │   ├── RawMaterial       # 原材料 8  │   ├── Component         # 零部件 9  │   └── FinishedGoods     # 成品10  ├── Location11  │   ├── Factory           # 工厂12  │   ├── Warehouse         # 仓库13  │   └── Port              # 港口14  └── Risk                  # 风险事件15      ├── NaturalDisaster   # 自然灾害16      ├── PolicyChange      # 政策变动17      └── QualityIssue      # 质量问题

1  docker run -d \2    --name neo4j-kg \3    -p 7474:7474 \4    -p 7687:7687 \5    -e NEO4J_AUTH=neo4j/your_password_here \6    -e NEO4J_PLUGINS='["apoc","graph-data-science"]' \7    -v neo4j_data:/data \8    neo4j:5.26-community

 1  // 唯一性约束——防止重复节点 2  CREATE CONSTRAINT supplier_name IF NOT EXISTS 3  FOR (s:Supplier) REQUIRE s.name IS UNIQUE; 4   5  CREATE CONSTRAINT manufacturer_name IF NOT EXISTS 6  FOR (m:Manufacturer) REQUIRE m.name IS UNIQUE; 7   8  CREATE CONSTRAINT product_sku IF NOT EXISTS 9  FOR (p:Product) REQUIRE p.sku IS UNIQUE;10  11  // 全文索引——支持模糊搜索12  CREATE FULLTEXT INDEX entity_name IF NOT EXISTS13  FOR (n:Supplier|Manufacturer|Product) ON EACH [n.name, n.alias];

import csvfrom neo4j import GraphDatabasedriver = GraphDatabase.driver(    "bolt://localhost:7687",    auth=("neo4j", "your_password_here"))def load_suppliers(csv_path):    """从CSV导入供应商数据"""    with open(csv_path) as f:        reader = csv.DictReader(f)        with driver.session() as session:            for row in reader:                session.run("""                    MERGE (s:Supplier {name: $name})                    SET s.creditRating = $rating,                        s.country = $country,                        s.leadTime = toInteger($lead_time),                        s.annualRevenue = toFloat($revenue)                """, {                    "name": row["supplier_name"],                    "rating": row["credit_rating"],                    "country": row["country"],                    "lead_time": row["lead_time_days"],                    "revenue": row["annual_revenue"]                })def load_supply_relations(csv_path):    """从CSV导入供应关系"""    with open(csv_path) as f:        reader = csv.DictReader(f)        with driver.session() as session:            for row in reader:                session.run("""                    MATCH (s:Supplier {name: $supplier})                    MATCH (m:Manufacturer {name: $manufacturer})                    MERGE (s)-[r:SUPPLIES_TO]->(m)                    SET r.product = $product,                        r.contractStart = date($start),                        r.contractEnd = date($end)                """, {                    "supplier": row["supplier"],                    "manufacturer": row["manufacturer"],                    "product": row["product"],                    "start": row["contract_start"],                    "end": row["contract_end"]                })

import anthropicimport jsonclient = anthropic.Anthropic()# 定义抽取的结构化输出格式EXTRACTION_TOOLS = [{    "name": "extract_supply_chain_knowledge",    "description": "从文本中抽取供应链相关的实体和关系",    "input_schema": {        "type": "object",        "properties": {            "entities": {                "type": "array",                "items": {                    "type": "object",                    "properties": {                        "name": {"type": "string"},                        "type": {                            "type": "string",                            "enum": ["Supplier", "Manufacturer",                                     "Product", "Location", "Risk"]                        },                        "attributes": {"type": "object"}                    },                    "required": ["name", "type"]                }            },            "relations": {                "type": "array",                "items": {                    "type": "object",                    "properties": {                        "source": {"type": "string"},                        "target": {"type": "string"},                        "relation": {                            "type": "string",                            "enum": ["SUPPLIES_TO", "PRODUCES",                                     "HAS_COMPONENT", "LOCATED_IN",                                     "AFFECTED_BY", "ALTERNATIVE_FOR"]                        },                        "attributes": {"type": "object"}                    },                    "required": ["source", "target", "relation"]                }            }        },        "required": ["entities", "relations"]    }}]SYSTEM_PROMPT = """你是一个供应链知识抽取专家。从给定的文本中抽取供应链相关的实体（供应商、制造商、产品、位置、风险）和关系（供应给、生产、包含组件、位于、受影响于、替代供应商）。规则：1. 实体名称使用官方全称，附带常用简称作为alias属性2. 只抽取文本中明确提到的关系，不要推测3. 时间、数量等信息放在关系的attributes中4. 如果实体类型不确定，选择最合理的类型"""def extract_knowledge(text):    """用Claude从文本中抽取三元组"""    response = client.messages.create(        model="claude-sonnet-4-6",  # 抽取任务用Sonnet够了        max_tokens=4096,        system=SYSTEM_PROMPT,        tools=EXTRACTION_TOOLS,        messages=[{            "role": "user",            "content": f"请从以下文本中抽取供应链知识：\n\n{text}"        }]    )    for block in response.content:        if block.type == "tool_use":            return json.loads(json.dumps(block.input))    return None# 示例：从新闻中抽取news = """路透社2026年5月报道：由于日本九州地区发生6.2级地震，瑞萨电子（Renesas Electronics）位于熊本县的300mm晶圆厂被迫停产至少两周。该工厂主要生产车规级MCU芯片，是丰田汽车、本田汽车和博世（Bosch）的核心供应商。行业分析师指出，意法半导体（STMicroelectronics）和德州仪器（Texas Instruments）有望成为短期替代供应商。"""result = extract_knowledge(news)print(json.dumps(result, indent=2, ensure_ascii=False))

{  "entities": [    {"name": "瑞萨电子", "type": "Supplier",     "attributes": {"alias": "Renesas Electronics", "country": "日本"}},    {"name": "丰田汽车", "type": "Manufacturer",     "attributes": {"alias": "Toyota"}},    {"name": "本田汽车", "type": "Manufacturer",     "attributes": {"alias": "Honda"}},    {"name": "博世", "type": "Manufacturer",     "attributes": {"alias": "Bosch", "country": "德国"}},    {"name": "车规级MCU芯片", "type": "Product", "attributes": {}},    {"name": "熊本县晶圆厂", "type": "Location",     "attributes": {"type": "Factory", "waferSize": "300mm"}},    {"name": "九州地震", "type": "Risk",     "attributes": {"type": "NaturalDisaster", "magnitude": 6.2,                     "date": "2026-05"}},    {"name": "意法半导体", "type": "Supplier",     "attributes": {"alias": "STMicroelectronics"}},    {"name": "德州仪器", "type": "Supplier",     "attributes": {"alias": "Texas Instruments"}}  ],  "relations": [    {"source": "瑞萨电子", "target": "丰田汽车",     "relation": "SUPPLIES_TO", "attributes": {"product": "车规级MCU芯片"}},    {"source": "瑞萨电子", "target": "本田汽车",     "relation": "SUPPLIES_TO", "attributes": {"product": "车规级MCU芯片"}},    {"source": "瑞萨电子", "target": "博世",     "relation": "SUPPLIES_TO", "attributes": {"product": "车规级MCU芯片"}},    {"source": "瑞萨电子", "target": "熊本县晶圆厂",     "relation": "LOCATED_IN", "attributes": {}},    {"source": "瑞萨电子", "target": "九州地震",     "relation": "AFFECTED_BY",     "attributes": {"impact": "停产至少两周"}},    {"source": "意法半导体", "target": "瑞萨电子",     "relation": "ALTERNATIVE_FOR", "attributes": {"timeframe": "短期"}},    {"source": "德州仪器", "target": "瑞萨电子",     "relation": "ALTERNATIVE_FOR", "attributes": {"timeframe": "短期"}}  ]}

def write_knowledge_to_neo4j(knowledge, session):    """将抽取的知识写入Neo4j"""    # 写入实体    for entity in knowledge["entities"]:        label = entity["type"]        props = {"name": entity["name"]}        props.update(entity.get("attributes", {}))        # 动态构建SET子句        set_clauses = ", ".join(            f"n.{k} = ${k}" for k in props if k != "name"        )        set_part = f"SET {set_clauses}" if set_clauses else ""        session.run(            f"MERGE (n:{label} {{name: $name}}) {set_part}",            props        )    # 写入关系    for rel in knowledge["relations"]:        attrs = rel.get("attributes", {})        set_clauses = ", ".join(            f"r.{k} = ${k}" for k in attrs        )        set_part = f"SET {set_clauses}" if set_clauses else ""        session.run(f"""            MATCH (a {{name: $source}})            MATCH (b {{name: $target}})            MERGE (a)-[r:{rel['relation']}]->(b)            {set_part}        """, {"source": rel["source"],              "target": rel["target"], **attrs})

from neo4j import GraphDatabaseimport anthropicclient = anthropic.Anthropic()driver = GraphDatabase.driver(    "bolt://localhost:7687",    auth=("neo4j", "your_password_here"))def find_similar_entities(session):    """用APOC找名称相似的实体对"""    result = session.run("""        MATCH (a), (b)        WHERE id(a) < id(b)          AND labels(a) = labels(b)          AND apoc.text.jaroWinklerDistance(a.name, b.name) > 0.85        RETURN a.name AS name1, b.name AS name2,               labels(a)[0] AS label,               apoc.text.jaroWinklerDistance(a.name, b.name) AS similarity        ORDER BY similarity DESC        LIMIT 50    """)    return [dict(r) for r in result]def llm_verify_alignment(entity_pairs):    """用Claude验证实体是否是同一个"""    prompt = "以下是一些可能指向同一实体的名称对。\n"    prompt += "请判断每一对是否是同一个实体，返回JSON数组。\n\n"    for pair in entity_pairs:        prompt += (f"- \"{pair['name1']}\" vs \"{pair['name2']}\" "                   f"(类型: {pair['label']})\n")    response = client.messages.create(        model="claude-sonnet-4-6",        max_tokens=2048,        messages=[{"role": "user", "content": prompt}]    )    return response.content[0].textdef merge_entities(session, name_keep, name_remove):    """合并两个实体节点，保留第一个，删除第二个"""    session.run("""        MATCH (keep {name: $keep})        MATCH (remove {name: $remove})        // 迁移所有入边        CALL {            WITH keep, remove            MATCH (remove)<-[r]-()            WITH keep, remove, collect(r) AS rels            UNWIND rels AS r            WITH keep, startNode(r) AS other, type(r) AS relType,                 properties(r) AS props            CALL apoc.create.relationship(other, relType, props, keep)            YIELD rel            RETURN count(rel) AS inMoved        }        // 迁移所有出边        CALL {            WITH keep, remove            MATCH (remove)-[r]->()            WITH keep, remove, collect(r) AS rels            UNWIND rels AS r            WITH keep, endNode(r) AS other, type(r) AS relType,                 properties(r) AS props            CALL apoc.create.relationship(keep, relType, props, other)            YIELD rel            RETURN count(rel) AS outMoved        }        // 将被合并实体的名称作为别名保留        SET keep.alias = CASE            WHEN keep.alias IS NULL THEN remove.name            ELSE keep.alias + ',' + remove.name        END        DETACH DELETE remove    """, {"keep": name_keep, "remove": name_remove})

def run_quality_checks(session):    """运行一系列质量检查"""    checks = {        "孤立节点（没有任何关系的实体）": """            MATCH (n) WHERE NOT (n)--() RETURN count(n) AS count        """,        "自环关系（自己指向自己）": """            MATCH (n)-[r]->(n) RETURN count(r) AS count        """,        "重复关系": """            MATCH (a)-[r1]->(b), (a)-[r2]->(b)            WHERE id(r1) < id(r2) AND type(r1) = type(r2)            RETURN count(r1) AS count        """,        "缺少关键属性的供应商": """            MATCH (s:Supplier)            WHERE s.creditRating IS NULL               OR s.country IS NULL            RETURN count(s) AS count        """,        "没有供应商的产品": """            MATCH (p:Product)            WHERE NOT ()-[:SUPPLIES_TO]->()-[:PRODUCES]->(p)              AND NOT ()-[:SUPPLIES_TO {product: p.name}]->()            RETURN count(p) AS count        """    }    print("=" * 60)    print("知识图谱质量报告")    print("=" * 60)    for name, query in checks.items():        result = session.run(query).single()        count = result["count"]        status = "✓ PASS" if count == 0 else f"✗ WARN ({count})"        print(f"  {status}{name}")    print("=" * 60)

1  ============================================================2  知识图谱质量报告3  ============================================================4    ✗ WARN (12)  孤立节点（没有任何关系的实体）5    ✓ PASS  自环关系（自己指向自己）6    ✗ WARN (3)   重复关系7    ✗ WARN (8)   缺少关键属性的供应商8    ✓ PASS  没有供应商的产品9  ============================================================

// 查询某供应商停产的影响扩散范围（3跳以内）MATCH path = (risk:Risk)-[:AFFECTED_BY*0..1]-(supplier:Supplier)             -[:SUPPLIES_TO*1..3]->(affected)WHERE supplier.name = "瑞萨电子"RETURN path// 更精确的版本：返回所有受影响的制造商和产品MATCH (s:Supplier {name: "瑞萨电子"})-[:SUPPLIES_TO]->(m:Manufacturer)OPTIONAL MATCH (m)-[:PRODUCES]->(p:Product)RETURN m.name AS affected_manufacturer,       collect(DISTINCT p.name) AS affected_products,       s.creditRating AS supplier_ratingORDER BY m.name

// 查找某供应商的替代方案MATCH (original:Supplier {name: "瑞萨电子"})      -[:SUPPLIES_TO]->(customer:Manufacturer)MATCH (alt:Supplier)-[:ALTERNATIVE_FOR]->(original)WHERE alt.creditRating IN ["A", "B"]  // 只要信用好的RETURN customer.name AS affected_customer,       alt.name AS alternative_supplier,       alt.creditRating AS rating,       alt.leadTime AS lead_time_daysORDER BY alt.creditRating, alt.leadTime

// 查找最长的供应链路径MATCH path = (start)-[:SUPPLIES_TO|PRODUCES|HAS_COMPONENT*]->(end)WHERE NOT ()-[:SUPPLIES_TO|PRODUCES|HAS_COMPONENT]->(start)  // 起点  AND NOT (end)-[:SUPPLIES_TO|PRODUCES|HAS_COMPONENT]->()    // 终点RETURN [n IN nodes(path) | n.name] AS supply_chain,       length(path) AS depthORDER BY depth DESCLIMIT 10

import jsonimport anthropicfrom neo4j import GraphDatabaseclient = anthropic.Anthropic()driver = GraphDatabase.driver(    "bolt://localhost:7687",    auth=("neo4j", "your_password_here"))def graph_rag_query(user_question):    """基于知识图谱的RAG问答"""    # Step 1: 用Claude将自然语言转为Cypher查询    cypher_response = client.messages.create(        model="claude-sonnet-4-6",        max_tokens=1024,        system="""你是一个Cypher查询生成专家。根据用户的自然语言问题，生成Neo4j Cypher查询语句。图谱Schema:- 节点: Supplier, Manufacturer, Product, Location, Risk- 关系: SUPPLIES_TO, PRODUCES, HAS_COMPONENT, LOCATED_IN,        AFFECTED_BY, ALTERNATIVE_FOR- 供应商属性: name, creditRating, country, leadTime- 产品属性: name, sku, category只返回Cypher语句，不要解释。""",        messages=[{            "role": "user",            "content": user_question        }]    )    cypher = cypher_response.content[0].text.strip()    # 去掉大模型返回中可能包含的代码围栏标记    fence = chr(96) * 3    cypher = cypher.replace(fence + "cypher", "").replace(fence, "").strip()    # Step 2: 执行Cypher查询    with driver.session() as session:        try:            result = session.run(cypher)            records = [dict(r) for r in result]        except Exception as e:            records = [{"error": str(e)}]    # Step 3: 用Claude基于查询结果生成自然语言回答    answer_response = client.messages.create(        model="claude-sonnet-4-6",        max_tokens=2048,        system="基于知识图谱的查询结果，用中文回答用户的问题。"               "如果结果为空，说明图谱中没有相关数据。",        messages=[{            "role": "user",            "content": (f"用户问题：{user_question}\n\n"                        f"Cypher查询：{cypher}\n\n"                        f"查询结果：{json.dumps(records, ensure_ascii=False)}")        }]    )    return {        "question": user_question,        "cypher": cypher,        "raw_results": records,        "answer": answer_response.content[0].text    }# 测试result = graph_rag_query("瑞萨电子停产会影响哪些汽车制造商？有替代供应商吗？")print(f"问题：{result['question']}")print(f"Cypher：{result['cypher']}")print(f"回答：{result['answer']}")

问题：瑞萨电子停产会影响哪些汽车制造商？有替代供应商吗？Cypher：MATCH (s:Supplier {name: "瑞萨电子"})-[:SUPPLIES_TO]->(m:Manufacturer)OPTIONAL MATCH (alt:Supplier)-[:ALTERNATIVE_FOR]->(s)RETURN m.name AS affected, collect(DISTINCT alt.name) AS alternatives回答：根据知识图谱数据，瑞萨电子停产将直接影响以下汽车制造商：1. 丰田汽车2. 本田汽车3. 博世（Bosch，虽然博世是零部件供应商，但它也从瑞萨采购MCU芯片）目前图谱中记录的替代供应商有：- 意法半导体（STMicroelectronics）——短期替代- 德州仪器（Texas Instruments）——短期替代建议关注这两家替代供应商的产能和交付周期，确认能否满足需求。