RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
69 lines
3.4 KiB
Python
69 lines
3.4 KiB
Python
"""演化分析测试：构造两季度（多窗口）数据验证 新增/增强/减弱/退出 事件。"""
|
||
from __future__ import annotations
|
||
|
||
from app.evolution.analyzer import analyze
|
||
from app.graph.tkg import TemporalKG
|
||
from app.models import GraphObservation, JobDefinition
|
||
|
||
|
||
def _obs(job: str, skill: str, window: str, n: int) -> list[GraphObservation]:
|
||
"""生成 n 条同窗口观察（模拟窗口内 n 条 JD 提到该技能）。"""
|
||
return [GraphObservation(job_id=job, skill=skill, importance="required",
|
||
weight=1.0, window=window, source_ids=[f"{window}#{i}"])
|
||
for i in range(n)]
|
||
|
||
|
||
def test_quarter_window_events():
|
||
kg = TemporalKG()
|
||
kg.add_job(JobDefinition(job_id="job_A", title="岗位A"))
|
||
observations: list[GraphObservation] = []
|
||
# 老技能 Legacy：2024Q1-Q3 稳定出现（3 窗口），Q4 与次年 Q1 连续消失 → 退出
|
||
for w in ["2024Q1", "2024Q2", "2024Q3"]:
|
||
observations += _obs("job_A", "Legacy技能", w, 3)
|
||
# 新技能 NewSkill：2024Q4 首次出现且支撑 >=2 → 新增
|
||
observations += _obs("job_A", "NewSkill", "2024Q4", 3)
|
||
# 增强技能：3 → 9（>=1.5 倍且差 >=3）
|
||
observations += _obs("job_A", "GrowSkill", "2024Q3", 3)
|
||
observations += _obs("job_A", "GrowSkill", "2024Q4", 9)
|
||
# 减弱技能：8 → 3（<=0.5 倍且差 >=3）
|
||
observations += _obs("job_A", "ShrinkSkill", "2024Q3", 8)
|
||
observations += _obs("job_A", "ShrinkSkill", "2024Q4", 3)
|
||
|
||
for o in observations: # 与 pipeline 一致：观察先写入图谱，再分析演化
|
||
kg.observe(o)
|
||
events = analyze(kg, observations=observations)
|
||
by_type = {}
|
||
for e in events:
|
||
by_type.setdefault(e.change_type, []).append(e)
|
||
|
||
# 新增：LegacySkill 首现 + NewSkill 首现
|
||
new_skills = {e.skill for e in by_type.get("新增", [])}
|
||
assert "NewSkill" in new_skills and "Legacy技能" in new_skills
|
||
# 退出：Legacy 技能在 2024Q4 判定（Q4、次年 Q1 连续为 0，且已稳定 3 窗口）
|
||
exits = [e for e in by_type.get("退出", []) if e.skill == "Legacy技能"]
|
||
assert exits and exits[0].window == "2024Q4"
|
||
assert exits[0].prev_weight > 0 and exits[0].curr_weight == 0
|
||
assert "数据源" in exits[0].description # 中文说明注明数据源
|
||
# 增强 / 减弱
|
||
assert any(e.skill == "GrowSkill" and e.window == "2024Q4"
|
||
for e in by_type.get("增强", []))
|
||
assert any(e.skill == "ShrinkSkill" and e.window == "2024Q4"
|
||
for e in by_type.get("减弱", []))
|
||
# 退出事件关闭了图谱区间
|
||
assert kg.g["job_A"]["Legacy技能"]["intervals"][-1][1] == "2024Q4"
|
||
# 事件保存前后权重、时间窗、来源
|
||
grow = next(e for e in by_type["增强"] if e.skill == "GrowSkill")
|
||
assert grow.prev_weight == 3.0 and grow.curr_weight == 9.0
|
||
assert grow.source_ids
|
||
|
||
|
||
def test_no_spurious_exit_for_new_skill():
|
||
"""只出现 1-2 个窗口的技能消失不报“退出”（抑制稀疏噪声）。"""
|
||
kg = TemporalKG()
|
||
kg.add_job(JobDefinition(job_id="job_B", title="岗位B"))
|
||
observations = _obs("job_B", "一次性技能", "2024Q1", 3) # 只出现 1 个窗口
|
||
observations += _obs("job_B", "常驻技能", "2024Q1", 2)
|
||
observations += _obs("job_B", "常驻技能", "2024Q2", 2)
|
||
events = analyze(kg, observations=observations)
|
||
assert not [e for e in events if e.change_type == "退出" and e.skill == "一次性技能"]
|