汽车供应链风险监控与影响穿透 Skill
端到端一体化 skill(自包含,无外部依赖):仅以联网搜索为信息源,采集汽车供应链风险事件 → 三维评级 → 多层打标 → 确定性穿透到内部供应商/零件主数据 → 全景报告 + Excel。核心增量是重点下级部件供应商(龙头/垄断企业)驱动的搜索 + 命中确定性预打标。
全流程分两段,职责边界=「事件里写了什么 ↔ 命中了我们什么」(先评级落契约,再打标穿透,不让 LLM 一把梭):
- 采集段=事件原文实体抽取(涉及原材料/涉及供应商)+ 三维评级 +(监控层)企业种子预打标,不碰主数据。
- 穿透段=对真实 BOM 受控词表多层打标(LLM)+ 确定性穿透(脚本)→ 受影响供应商/零件/地区。
配置变量
执行前读 skill 根目录 config.json(所有脚本经此读取):
- time_window_days:检索时间窗口天数,默认 7
- output_dir:交付物输出目录,默认 outputs/
- search_config:搜索词预算(p0_l1≤7, p0_l2≤3, company_watchlist≤10, p1=1/类, cross=1)
- company_watchlist.path:references/company_watchlist.json(重点企业清单 + 类别搜索包 + 轮动游标)
- data_sources.supplier_name_map:运行时口子(references/supplier_name_map.csv,名↔编码,零构建)。底座原始口子(A/新表)在 build-kit/,不在 skill。
- feishu_upload:飞书云盘上传配置(enabled 默认 false)
所有时间窗口过滤统一按新闻发布时间判断。来源类型固定为 联网搜索。
环境依赖(首次在新平台运行先装齐)
pip install pandas openpyxl # 数据/Excel(必需)
pip install playwright && python -m playwright install chromium # PDF 主方案
pip install reportlab # PDF 兜底(可选)
- 联网搜索由主会话的联网搜索工具执行(步骤 1),skill 不内置搜索 API。
- WebFetch(步骤 2.6)由主会话工具执行。
lark-cli(步骤 12 飞书上传,可选):未装则跳过上传,不影响主流程。- 本 skill 完全自包含 + 构建/运行分离:报告套件在
report_kit/、契约在references/、主数据底座已物化在data/master/,无任何../跨目录依赖;整个 skill 目录可直接搬运(整包 ~25M,zip <10M)。 - 穿透底座以纯文本随包(目标平台禁
.sqlite/.xlsx):data/master/parts_agg.csv(A库一级广度·聚合版:供应商码/省市区县/大类/n_parts)+data/master/nt_chain.json(新表三层深度链)。penetrate.py启动时载入内存 sqlite(:memory:) 查询,包内无 .sqlite 文件。受影响零件数=SUM(n_parts) 约2%跨组合重复(广度估算),供应商数/大类分布精确;零件级精确底座在build-kit/archive/。 - 构建脚本与原始数据均不在 skill 内——
clean_new_table.py/build_base.py(及旧build_master.py等)、原始 A/新表口子都在项目级build-kit/,补数重建见build-kit/README.md。新平台只跑分析流程,无需 build-kit。 - 供应商名→编码 运行时口子:目标环境把两列 CSV(
供应商名称,供应商编码,UTF-8)放到references/supplier_name_map.csv即生效,零构建;缺失则供应商名匹配降级(按码显示)。模板见references/supplier_name_map.example.csv。
执行流程
严格按步骤顺序执行。每次新任务前清空 outputs/(含 intermediate/)。
步骤 0:确认主数据底座就绪(穿透段前置)
data/master/parts_agg.csv + nt_chain.json 已随包,运行时载入内存 sqlite 直接用,不在 skill 内构建。仅当主数据更新时,去 build-kit/ 重建后把成品拷回(见 build-kit/README.md):
python build-kit/scripts/clean_new_table.py <新表.xlsx> -o build-kit/intermediate/nt_chains.json
python build-kit/scripts/build_base.py --old-sqlite build-kit/archive/penetration_legacy.sqlite \
--chains build-kit/intermediate/nt_chains.json \
--out-dir data/master
运行时口子(零构建):references/supplier_name_map.csv(供应商名↔编码),目标环境自投即生效。
步骤 1:联网搜索词条生成
按 三组 系统化生成搜索词,合计 ≈30 条。
词条设计规范(所有组通用)
- 关键词 ≤5~6 个(含信号词+时间锚定词)
- 必含 1 个新闻信号词:
最新通报预警突发本周 - 必含 1 个时间锚定词:
近期本周2026年X月(动态取当前月) - 禁用词:
分析趋势展望回顾纵览全景报告综述盘点 - 优先事件性名词(爆炸/封闭/停运/断供/减产),少用描述性词汇
组 A — 原材料波动 P0(L1≤7 + L2≤3)
参考 references/material_keywords.md 与 references/semiconductor_supply_map.md:
- L1 大类层(≤7):原材料名称 + 风险场景词(涨价/断供/产能不足/出口管制)
- L2 品类层(≤3):具体型号 + 风险场景词(如「DDR4 DDR5 车规级 产能 分料 最新」)
- 体现汽车语境:「车规级」「汽车用」等限定词
> 旧 L3「巨头层」已被组 B 系统化覆盖,不再单列。
组 B — ★重点企业监控层(≤10,本版核心)
读取 references/company_watchlist.json:
1. 核心档 core:每次必搜。
2. 轮动档 rotation:按 rotation_config.cursor 取 batch_size 家本次覆盖;跑完把 cursor 推进 batch_size 并对清单长度回绕(就地写回 company_watchlist.json 的 cursor,实现多轮跑遍全表)。
3. 词条来自 search_bundles:每条把一个高优类别的 2-4 家头部企业打包(如「美光 三星 SK海力士 存储芯片 减产 涨价 断供 最新」),并携带该 bundle 的 tag_category 作为命中预打标依据。
4. 合计 ≤ search_config.company_watchlist_max_terms(默认 10)。
> 设计意图:词按企业播种 → 命中事件可确定性预打标企业名+部件类别(见步骤 3),下游穿透置信度天然高。这是监控层相比通用搜索的核心增量。
组 C — P1 其他 9 类(每类 1 条)+ 跨类型综合(1 条)
完整事件类型清单与词条示例见 references/event_types.md。9 类:自然灾害/生产事故/交通事故/公共事件/限电限产/海关贸易/供应商经营/品牌口碑市场舆情/物流港口。
执行方式
- 先生成全部词条列表(标注每条所属组;组 B 词条附带
命中企业候选=该 bundle 企业、命中部件类别=tag_category)。 - 列表返回主会话,由主会话调用联网搜索工具逐条执行。
- 结果回传本技能继续处理;执行中输出当前词条;记录完整搜索日志。
搜索结果原始数据保存(search_events_{ts}.json,严禁改写原始字段)
| 字段 | 说明 |
|---|---|
title/url/snippet/published_date |
搜索工具原样输出 |
snippet_scope |
per_link(该 URL 独立摘要)/ search_term(搜索词级总结,覆盖该词下所有链接共享) |
search_term/search_time |
词条与执行时间 |
source_type |
固定「联网搜索」 |
date_status |
日期预过滤状态(valid/unknown) |
watchlist_group |
是否来自组 B(true/false) |
watchlist_companies / watchlist_category |
组 B 词条携带的企业候选与部件类别(供步骤 3 预打标) |
搜索工具适配规则(按工具能力分支填 snippet/snippet_scope)
| 工具类型 | 代表 | snippet 填充 | scope |
|---|---|---|---|
| 返回 per-link 摘要 | Tavily/Google CSE/Bing | 每条 link 自带 snippet | per_link |
| 仅返回词级合成总结 | Claude Code 内置 WebSearch | 合成总结写入该词下所有 link 的 snippet(共享) | search_term |
| 仅 title+url | 极简工具 | snippet 留空,靠步骤 2.6 WebFetch 补 | per_link(补全后) |
日期预过滤
明确超窗口→不保存(记日志);缺失/模糊→保留标 unknown;窗口内→保留标 valid。
步骤 2:数据预过滤
python scripts/pre_filter.py outputs/search_events_{ts}.json --time-window 7
- 输入:
search_events_{ts}.json(单一联网搜索源)→ 输出:cleaned_events_{ts}.json+pre_filter_stats_{ts}.json - 规则:导航页垃圾(标题≤4字无时间)丢弃 / 首页·极短路径 URL 丢弃 / URL 全等去重 / 归一化标题重叠系数 ≥0.6 去重(保留更丰富者)/ 时间窗口(含 URL 多模式日期提取)/ 空内容(<20字)丢弃。
- 气象预警 guard:标题含「预警」且命中气象·地质灾害关键词的事件不参与标题相似度合并(各区县独立,模板字串高度雷同极易误并)。
步骤 2.6:WebFetch 日期 + 正文摘要补全
对 date_status=unknown 且 URL 日期提取失败、或 snippet_scope=search_term 的事件,WebFetch 其 URL,一次同时取:发布日期(meta/JSON-LD/正文 YYYY-MM-DD)+ 正文前 500 字摘要。
- 日期:窗口内→valid;超窗→移除;失败→留 unknown。
- 摘要:成功→覆盖 snippet 并把 scope 改 per_link;失败→留原值。
- 覆盖更新 cleaned_events_{ts}.json。只取这两类信息,不抓全文、不做语义分析。
步骤 3:原始事件标准化 + 评级 +(监控层)企业预打标
评分前必须完整阅读 references/scoring_guide.md(三维评分细则 + 公司供应商地理梯队)。按 8-10 条一批 做 LLM 语义分析,输出 llm_analysis_{ts}.json(符合 references/llm_output_schema.json)。
字段分类
- A 类·事实透传(原样复制):事件标题、来源URL、发布时间(YYYY/MM/DD HH:mm,禁模糊到月,无法确定填「未知」)。
- C 类·内容派生(忠于原文):事件详情——素材来源优先级:
per_linksnippet(可保留具体数字)>search_term总结(不得归因数字到单条 URL)> 仅 title(最小化扩展,不编造)。 - B 类·语义生成:事件类型(10 枚举或「待过滤-非风险事件」)、省/市/区县/泛区域海域、
涉及原材料(原文锚点)、涉及供应商(原文锚点)、相关度(写具体理由,禁套话)、关联度评分依据(必填,写命中地理梯队或非地理依据)、运输方式、风险评级结果及三维得分。
★企业命中预打标(监控层确定性,本版新增)
对来自组 B 监控层的事件(watchlist_group=true),在 LLM 标准化时确定性回填:
1. 命中重点企业:正文确认提及该词条携带的 watchlist_companies 中某企业 → 是;否则 否。搜索词召回≠正文命中,必须正文确认,避免误标。
2. 命中企业名:命中的清单企业官方中文名(多个逗号分隔;填清单标准名,便于供应商名匹配)。
3. 命中部件类别:命中企业所属 watchlist_category(如 存储芯片)。
4. 同时把 命中企业名 并入 涉及供应商、命中部件类别 可佐证 涉及原材料/事件类型判断。
> 非监控层事件:命中重点企业=否,后两字段留空。但若通用搜索/原材料词命中了清单企业,也可据正文置 是 并回填(清单见 company_watchlist.json)。
被过滤事件精简输出
仅 5 字段:事件标题、来源URL、来源类型、是否过滤标志位、被过滤的原因。
LLM 只做语义判断并输出 JSON,格式化交脚本。严禁临时写新 Python 脚本。
步骤 4:原始事件过滤(含全局去重)
- 时间窗口过滤(无明确时间跳过)。
- 语义相关度过滤(LLM 语义推理,禁关键词匹配)。
- 全局事件去重(跨批次强制):所有批次分析完后回看全部保留事件,只比对轻量字段(标题/类型/省/市/区县/发布时间)。聚类同一现实事件,每簇留 1 条代表(信息最全→风险最高)。其余改判
是否过滤=是,原因填「与「<代表>」为同一事件,重复仅保留一条」。气象/地质预警按区县判独立性,不得合并;拿不准不并。
差异化过滤标准
- 原材料波动(P0):宽松但非无限制——价格异动/供应中断/产能变化/政策管制才保留。
- 重点企业命中事件:优先保留(即便影响偏间接,也保留为低风险供知晓,因穿透价值高);但纯正面/经营动态仍按通用排除清单过滤。
- P1 其他类型:严格——须确实可能对供应链物流供货产生实质性影响。
通用排除清单(一律归「待过滤-非风险事件」,无一例外)
政企/外交动态、技术研发/产品发布、正面景气分析、行业趋势/综述、常规统计数据(除非伴随骤降/暴跌)、政府长期规划、人事任免、公司经营动态(融资/上市/扩产计划未开工/季报)、非重点区域常规气象·地质预警(低关注省份,或 T2/T3 省未命中核心区县;T1 沪苏浙照常保留走评级;且未伴随人员伤亡/干线封闭/港口机场停运/波及汽车产业链)。详见 references/scoring_guide.md 地理梯队。
原材料波动适用范围(满足≥1 项才保留)
价格异动(量化涨跌)/ 供应中断(断供·出口管制·限产·罢工·矿难·交期延长)/ 产能变化(扩产已开工·停产·关停·检修)/ 政策管制(关税·配额·反倾销·新增管制)。
核心原则:是否在未来 1-4 周内可能对汽车供应链物流或零部件供货产生实质性负面影响?正面/意向/技术/综述一律不算。
步骤 5:风险评级
详细细则见 references/scoring_guide.md。3 维度各 1-3 分加权:
- 事件严重程度 ×0.4 / 供应链关联度 ×0.35 / 影响紧迫性 ×0.25。
- ≥2.50 高 / 1.70-2.49 中 / <1.70 低。兜底:任一维度=3 且另一维度≥2 → 至少中风险。
- 命中重点企业可作为「供应链关联度」的加分依据(命中清单内寡头/单一来源企业 = 直接命中核心供应节点)。
步骤 5.5:确定性去重兜底
python scripts/dedup_events.py outputs/llm_analysis_{ts}.json
按 (事件类型+省+市+发布日) 分组,组内标题重叠系数 ≥0.5 判同一事件,把步骤 4 LLM 漏掉的翻为 是否过滤=是。气象预警 guard 不参与。不调 LLM、可复现,就地写回。
步骤 6:数据标准化输出(产出契约)
python scripts/data_processor.py outputs/llm_analysis_{ts}.json
分离保留/过滤事件、一致性自检(全量=过滤+保留)、输出 all_events_{ts} + filtered_events_{ts}(JSON+Excel)、保存 suggestions_{ts}.json。filtered_events_{ts}.json 即穿透段输入契约(含 命中重点企业/命中企业名/命中部件类别)。严禁省略被过滤事件。
步骤 7:载入并校验契约(穿透段入口)
按 config.json → input_contract.events_glob(outputs/filtered_events_*.json)取最新契约,按 references/impact_schema.json → event_contract.required 逐字段校验:任一保留事件缺必填(尤其 涉及原材料/涉及供应商/命中重点企业/省市区县/三维得分)→ 报错停止,提示缺哪条哪字段。
步骤 8:多层打标(LLM,受控词表)
打标前必须完整阅读 references/tagging_guide.md 与 data/master/part_dictionary.json。对每条保留事件分步打标:原材料标签(保留原词)→ 大类标签(语义映射到真实 10 大类,匹配不到留空)→ 材料类标签(真实 13 材料类)→ 一级零件标签(多为「全品类」)。每层给命中依据+置信度。
- ★命中重点企业的事件:命中部件类别 直接为大类标签提供高置信 grounding(如 存储芯片→电子/电池电源/电器按 part_dictionary 映射),命中企业名 进 涉及供应商 供穿透名匹配。
输出 outputs/tagged_events.json(追加 多层标签,见 schema)。LLM 只打标输出 JSON,不写脚本。
步骤 9:确定性穿透(脚本)
python scripts/penetrate.py outputs/tagged_events.json -o outputs/penetrated_events.json
两库独立并列(编码两套命名空间,不 join),各库内三匹配器并集 + 命中层 provenance + 去重 + 相关度排序:
- A 库(一级广度,可信地理) parts:地理∩大类(地理绝不裸用——原材料波动=全国×大类;区域性事件=地理×大类,避免「上海暴雨→全上海」)。产出 A_广度:受影响 T1 供应商/零件数、清单、大类分布、Top集中度。
- 新表库(三层深度,试用基座) nt_chain:① 实体自底向上上溯(事件原材料/零部件实体→命中 T3/T2/T1 名→带出完整上游链,原始名不剥前缀靠分词召回)② 供应商名(事件供应商名→references/supplier_name_map.csv 运行时直读→编码→命中任一层)③ 地理∩实体(仅对已被实体命中的链,再以 T2/T3 地点收敛)。产出 新表_深度:受影响 T1/T2/T3 供应商数、命中层统计、层级化链(每链含命中层标注 + 三层名/码/地点 + 采购周期)。
纯查底座,不调 LLM。supplier_name_map.csv 缺失→供应商名硬匹配降级(按码原样显示),其余穿透不受影响。
步骤 10:多维建议(LLM)→ 报告 + Excel
- 基于穿透结果写多维建议存
outputs/suggestions.md(受影响大类分布 / 地区集中度 / 供应商集中度 / 采购周期 /命中重点企业清单专项)。 - 全景报告(HTML 自包含 + 同名 PDF):
python scripts/report.py outputs/penetrated_events.json --suggestions outputs/suggestions.md
套 report_kit/risk_report_shell.html(ECharts + 中国地图内联,无 CDN)。骨架:执行摘要 / 风险态势(等级·类型·省份分布 + 中国地图热力)/ 风险事件清单(按等级高→中→低、组内按风险评级加权总分降序,含地理敏感层级标签[T1/T2/T3/低关注,不涉及留空,从 关联度评分依据 取最高梯队] + 逐条穿透块[A库一级广度 + 新表库三层深度链(命中层标注·试用水印)两区] + 多层标签 + 命中重点企业高亮)/ 影响穿透汇总(A库广度图 + 新表深度 T1/T2/T3 + T2地点分布 + 高采购周期链)/ 多维建议。输出 outputs/risk_monitor_report.html + .pdf。
3. Excel 双表交付物:
python scripts/export_impact_excel.py outputs/penetrated_events.json
impact_deliverable_{ts}.xlsx:① 风险事件清单(一行一事件,含原文锚点+命中企业+多层打标+A库/新表穿透汇总)② 受影响一级供应商明细(A库,一行一事件×T1供应商,事件ID 关联)③ 新表·三层深度链(一行一 T1→T2→T3 链,含命中层+三层名码地点+采购周期,仅新表库有命中时生成)。
步骤 11:交付物整理
mkdir -p outputs/intermediate
mv outputs/search_events_*.json outputs/cleaned_events_*.json outputs/pre_filter_stats_*.json \
outputs/llm_analysis_*.json outputs/all_events_*.json outputs/suggestions_*.json \
outputs/tagged_events.json outputs/penetrated_events.json outputs/intermediate/ 2>/dev/null
最终交付物(outputs/ 主目录):风险监控报告(.html/.pdf) + all_events_*.xlsx + filtered_events_*.xlsx + impact_deliverable_*.xlsx。
步骤 12:飞书云盘上传(可选,默认关)
仅用户明确要求时执行:
python scripts/feishu_uploader.py --timestamp {ts}
读 config.json → feishu_upload(enabled 默认 false,临时改 true 上传完改回)。在 parent_folder_token 下建日期子文件夹(风险监控报告_YYYYMMDD)。只传 html/pdf/xlsx,不传 json。前置:lark-cli 已认证;单文件 ≤20MB;失败不阻塞。
脚本一览
> 构建脚本(clean_new_table.py/build_base.py/旧 build_master.py/build_geo_weights.py/export_master_excel.py)已移出 skill,在项目级 build-kit/,见 build-kit/README.md。下表只列 skill 内运行时脚本。
| 脚本 | 步骤 | 输入 → 输出 |
|---|---|---|
pre_filter.py |
2 | search_events → cleaned_events + stats(仅联网搜索,气象预警 guard 按标题识别) |
dedup_events.py |
5.5 | llm_analysis → llm_analysis(确定性去重兜底) |
data_processor.py |
6 | llm_analysis → all_events + filtered_events(JSON/Excel) + suggestions |
penetrate.py |
9 | tagged_events → penetrated_events + 影响汇总 |
report.py |
10 | penetrated_events(+suggestions.md) → risk_monitor_report.html + .pdf(自动) |
pdf_exporter.py |
10 | html → pdf(Playwright 主 / ReportLab 兜底) |
export_impact_excel.py |
10 | penetrated_events → impact_deliverable_{ts}.xlsx(多 sheet) |
feishu_uploader.py |
12(可选) | outputs/ 交付物(html/pdf/xlsx) → 飞书云盘 |
数据流
步骤1 联网搜索(组A原材料 + 组B重点企业监控 + 组C通用) → search_events_{ts}.json
步骤2 pre_filter.py → cleaned_events_{ts}.json
步骤2.6 WebFetch 日期/正文补全 → 更新 cleaned_events
步骤3-5 LLM 标准化+评级+企业预打标(分批8-10) + 全局去重 → llm_analysis_{ts}.json
步骤5.5 dedup_events.py → 确定性去重兜底
步骤6 data_processor.py → all_events + filtered_events(契约)
步骤7 契约校验(impact_schema.json) fail-fast
步骤8 LLM 多层打标(受控词表 tagging_guide) → tagged_events.json
步骤9 penetrate.py → penetrated_events.json
步骤10 LLM 多维建议 + report.py(全景报告 html/pdf) + export_impact_excel.py(双表)
步骤11 中间产物移 intermediate/
步骤12 feishu_uploader.py(可选)
核心设计原则
- 信息源单一(联网搜索),来源类型固定「联网搜索」。
- 采集↔穿透职责边界保留:LLM 语义判断,脚本确定性处理。
- 多层打标只落真实 BOM 受控词表,匹配不到留空、不硬填。
- 重点企业监控层:词按企业播种 → 命中确定性预打标 → 穿透高置信。
- 报告完全自包含(ECharts + 中国地图内联,禁 CDN),套
shared/统一外壳。
配置与参考文件
config.json(skill 根)— 路径/口子/搜索预算/穿透阈值/飞书references/company_watchlist.json— ★重点企业清单 + 类别搜索包 + 轮动游标references/event_types.md— 10 类事件定义与词条示例references/material_keywords.md/semiconductor_supply_map.md— 原材料词条与芯片供应链 groundingreferences/scoring_guide.md— 三维评分 + 地理梯队references/tagging_guide.md— 多层打标分步指引references/llm_output_schema.json— LLM 输出格式(含企业预打标字段)references/impact_schema.json— 采集↔穿透事件契约(唯一真相源)report_kit/(render_report.py + risk_report_shell.html + assets)— 自包含报告渲染套件