Agentic SCKG Risk Analyzer — 供应链知识图谱智能风险分析框架

1. 解决的问题

手动 ERP 排查可能要 2-3 周，到时候竞争对手早把备货扫空了

2. 核心算法逻辑

Agentic SCKG Risk Analyzer 解决出海品牌面临的生死级挑战：当全球某处发生黑天鹅事件（罢工/地震/制裁），你的旗舰产品会在多少天后断供？传统方案要么靠 ERP 人工逐层排查（耗时数周），要么让 LLM 在非结构化新闻里盲目搜索（漏掉隐藏链路）。

3. 业务应用场景

某国内出海智能家电品牌，V8 旗舰款吸尘器依赖一条跨越越南、韩国、深圳的四级供应链。某天新闻出现"越南工业区大规模罢工"，采购总监知道直接供应商里没有越南企业，但不知道越南工厂是否是自己三级供应商的材料独家来源。手动 ERP 排查可能要 2-3 周，到时候竞争对手早把备货扫空了。

| 数据类型 | 格式 | 说明 | |---------|------|------| | 供应商层级图 | 节点：{supplier_id, name, country, tier, default_prob, inventory_days} | ERP 或供应链系统导出 | | 供应关系图 | 边：{src, dst, lead_time_days, dependency_ratio, annual_volume} | 采购订单/BOM 汇总 | | 风险事件 | {event_type, location, affected_nodes, severity, description}

- 断供预警从"出事后救火（2-3 周）"提前到"事件触发即秒级诊断（< 10 秒）" - 旗舰 SKU 断货每天损失约 50-200 万元（依规模），提前 15 天响应直接避免断货损失 - 供应链弹性（Resilience）从"被动反应"升级为"主动链路穿透预警"

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• ROI 极高：供应链断货是出海品牌生死级风险，一次黑天鹅事件的损失可覆盖系统建设成本数十倍，且本框架无需昂贵专用图数据库，仅依赖 Python 标准库 + numpy
• 难度中等：核心算法已封装（见 model.py），主要实施成本在于①供应商数据治理（Tier 2+ 数据录入，约 4-8 周）；②新闻/风险事件监控接入（1-2 周）；③报告模版调优（1 周）
• 优先级最高：属于 WF-A（供应链工作流）的 P0 战略防御基础设施，且当前知识图谱域 Skill 库中唯一覆盖"网络科学 × 风险传播"方向，填补关键缺口
• 技术壁垒：将图论的中心度算法与 LLM Context Shell 结合，形成竞对难以快速复制的"链路穿透预警"护城河

7. 代码模板

代码块数量：6 · 路径：未检测到

from paper2skills_code._08_知识图谱.supply_chain_kg_2025.model import (
    AgenticSCKGRiskAnalyzer, SupplierNode, SupplyEdge, RiskEvent
)

# 1. 初始化分析器（指定品牌方节点 ID）
analyzer = AgenticSCKGRiskAnalyzer(brand_node_id="brand_001")

# 2. 构建供应链知识图谱
nodes = [
    SupplierNode("brand_001", "XX智能家电品牌", "中国", tier=0,
                 default_prob=0.01, inventory_days=30,
                 capacity_utilization=0.8, component_type="assembly"),
    SupplierNode("factory_d", "深圳整机厂D",   "中国", tier=1,
                 default_prob=0.03, inventory_days=20,
                 capacity_utilization=0.9, component_type="assembly"),
    SupplierNode("supplier_c", "马达供应商C",  "中国", tier=2,
                 default_prob=0.05, inventory_days=15,
                 capacity_utilization=0.85, component_type="motor"),
    SupplierNode("supplier_b", "韩国电机B",    "韩国", tier=3,
                 default_prob=0.04, inventory_days=10,
                 capacity_utilization=0.9, component_type="motor"),
    SupplierNode("factory_a", "越南材料厂A",   "越南", tier=4,
                 default_prob=0.08, inventory_days=5,
                 capacity_utilization=0.95, component_type="material"),
]
edges = [
    SupplyEdge("factory_a",  "supplier_b", lead_time_days=21, dependency_ratio=0.9,  annual_volume=500),
    SupplyEdge("supplier_b", "supplier_c", lead_time_days=14, dependency_ratio=0.75, annual_volume=800),
    SupplyEdge("supplier_c", "factory_d",  lead_time_days=7,  dependency_ratio=0.6,  annual_volume=1200),
    SupplyEdge("factory_d",  "brand_001",  lead_time_days=3,  dependency_ratio=1.0,  annual_volume=3000),
]
analyzer.build_kg(nodes, edges)

# 3. 触发风险事件分析
event = RiskEvent(
    event_id="evt_001", event_type="strike",
    location="越南胡志明工业区",
    affected_node_ids=["factory_a"],
    severity=0.8,
    description="越南胡志明工业区大规模罢工，预计持续4周",
)

chains, shells = analyzer.analyze_risk_event(
    event=event,
    brand_inventory_days=30,
    alternative_suppliers=["备用材料商-泰国F", "国内替代材料商G"],
    top_k_paths=3,
)

# 4. 输出诊断报告
for i, (chain, shell) in enumerate(zip(chains, shells)):
    print(f"\n【风险链 #{i+1}】级联风险: {chain.cascade_risk_score:.1%}")
    print(f"传播时间: {chain.total_lead_time_days} 天")
    print(shell)

# 5. 查看中心度摘要（节点重要性排名）
summary = analyzer.get_centrality_summary()
for nid, info in sorted(summary.items(), key=lambda x: -x[1]["pagerank"]):
    print(f"{info['name']:30s}  PR={info['pagerank']:.4f}  BT={info['betweenness']:.4f}")

8. 论文来源

• 2510.01115