fix(fam-edge): 人物 uid 跨视频复用导致特征串人 - 拆分同 uid 下的性别冲突
根因: 大模型给的 人物A/B/C 这类临时 uid 只在单次视频分析内部稳定，不同视频各 自独立编号——同一个字符串在不同视频里完全可能指向不同的真人（实测生产数据里 "人物A" 在 29 个视频里混了男女两个人，"人物B" 混了男/女/儿童三个人）。但 people 表 label 全局 UNIQUE，upsert_person / person_service 的特征聚合都直接 按这个字符串当全局稳定身份用，导致不同真人的特征被硬合并进同一行，"我"这张卡 显示出来的描述其实是我和媳妇两个人的特征混在一起。 两处落地点都加了同一条硬规则（性别是相对稳定信号，冲突大概率是撞了另一个人）： 1. oracle_db.upsert_person(): 单视频入库时，新特征性别与已有行冲突就不覆盖合 并，改分配 uid#2/uid#3 这样的派生 label 单独建行。 2. person_service._aggregate_features(): 每次全量重新聚合时按性别在线聚类， 同一 uid 下冲突的性别拆成独立分组，不再无脑覆盖成一坨。 拆出来的派生 label 走已有的"未命名 -> LLM 合并 -> 硬规则否决"流程，由现有机制 判断该并入哪个已命名身份。 新增 test_oracle_db.py（5 例）+ test_person_service.py（5 例）覆盖同性别合并 / 性别冲突拆分 / 后缀分配 / unknown 不触发拆分等场景。 生产数据已用新逻辑重新 reconcile 并手动核对 3 个因数据量太大 LLM 没能正确认领 的派生 label（人物A#2/人物B#2#2 -> 媳妇，爷爷#2 -> 爷爷），现在 4 个人物分组 （我/媳妇/爷爷/汤圆）特征都是内部一致的，不再互相串。
This commit is contained in:
ericwyuan
@@ -475,7 +475,35 @@ class OracleDB:
|
||||
# 带备注的原始输出分裂出垃圾人物行
|
||||
label = re.sub(r'[（(][^()（）]*[)）]', '', str(label)).strip() or str(label)
|
||||
now = _now_iso()
|
||||
display_uid = display_uid or label
|
||||
row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone()
|
||||
# 跨视频 uid 复用检测：大模型给的 人物A/B/C 这类 uid 只在单次视频分析内部
|
||||
# 稳定，不同视频各自独立编号——同一个字符串在不同视频里完全可能指向不同的
|
||||
# 真人（实测 "人物A" 在 29 个视频里混了至少男女两个人）。这里用性别这个相对
|
||||
# 稳定的信号做冲突检测：这次新特征和已有行的性别都明确且互相矛盾，说明大概
|
||||
# 率撞车到了另一个人，不能直接合并覆盖——改挂到一个新的派生 label（如
|
||||
# "人物A#2"）上，留给 person_service 的特征合并再判断这个新 label 该并进
|
||||
# 哪个已命名身份，避免把两个人的特征越merge越乱。
|
||||
if row and features:
|
||||
try:
|
||||
old_gender = (json.loads(row['features_json'] or '{}') or {}).get('gender', '')
|
||||
except (ValueError, TypeError):
|
||||
old_gender = ''
|
||||
old_gender = str(old_gender or '').strip()
|
||||
new_gender = str(features.get('gender', '') or '').strip()
|
||||
if (old_gender and old_gender.lower() not in ('unknown', '未知')
|
||||
and new_gender and new_gender.lower() not in ('unknown', '未知')
|
||||
and old_gender != new_gender):
|
||||
base_label, n = label, 2
|
||||
while self._conn.execute(
|
||||
"SELECT 1 FROM people WHERE label=?", (f"{base_label}#{n}",)).fetchone():
|
||||
n += 1
|
||||
label = f"{base_label}#{n}"
|
||||
row = None
|
||||
if logger:
|
||||
logger.warning(
|
||||
f"人物 uid 跨视频复用检测到性别冲突: {base_label} "
|
||||
f"旧={old_gender} 新={new_gender}，拆分为新 label={label}")
|
||||
# 特征合并（在已有 features_json 基础上）
|
||||
merged_features = self._merge_features(
|
||||
row['features_json'] if row else None, features) if row else (
|
||||
|
||||
@@ -145,10 +145,17 @@ class PersonService:
|
||||
def _aggregate_features(self) -> Dict[str, Dict]:
|
||||
"""从 events.person_appearances_json 聚合每个 uid 的合并特征。
|
||||
|
||||
遍历所有事件的 person_appearances，按 uid 收集 features dict，
|
||||
合并规则：首次非 unknown 值优先（与 oracle_db._merge_features 一致）。
|
||||
遍历所有事件的 person_appearances，按 uid 收集 features dict，合并规则：
|
||||
新非 unknown 值覆盖旧值（与 oracle_db._merge_features 一致）。
|
||||
|
||||
大模型给的 人物A/B/C 这类 uid 只在单次视频分析内部稳定，不同视频各自独立
|
||||
编号——同一个字符串在不同视频里完全可能指向不同的真人（实测 "人物A" 在
|
||||
29 个视频里混了男女两个人）。这里按性别做在线聚类：同一 uid 下遇到与已有
|
||||
分组都冲突（都明确且不同）的性别时，开一个新分组（uid#2/uid#3...），不再
|
||||
把两个人的特征糊成一坨；分组的 key 交给 upsert_person 当新 label 落库，
|
||||
走 unnamed 流程由 LLM 合并判断该并入哪个已命名身份。
|
||||
"""
|
||||
uid_features: Dict[str, Dict] = {}
|
||||
uid_groups: Dict[str, List[Dict]] = {}
|
||||
rows = self.db._conn.execute(
|
||||
"SELECT person_appearances_json FROM events "
|
||||
"WHERE person_appearances_json IS NOT NULL").fetchall()
|
||||
@@ -168,17 +175,31 @@ class PersonService:
|
||||
feats = pa.get('features') or {}
|
||||
if not isinstance(feats, dict):
|
||||
continue
|
||||
if uid not in uid_features:
|
||||
uid_features[uid] = dict(feats)
|
||||
else:
|
||||
merged = dict(uid_features[uid])
|
||||
for k, v in feats.items():
|
||||
v_str = str(v).strip() if v is not None else ''
|
||||
if v_str and v_str.lower() != 'unknown':
|
||||
merged[k] = v_str
|
||||
elif k not in merged:
|
||||
merged[k] = v_str or 'unknown'
|
||||
uid_features[uid] = merged
|
||||
groups = uid_groups.setdefault(uid, [])
|
||||
gender = str(feats.get('gender', '') or '').strip()
|
||||
target = None
|
||||
if gender and gender.lower() not in ('unknown', '未知'):
|
||||
for g in groups:
|
||||
g_gender = str(g.get('gender', '') or '').strip()
|
||||
if not g_gender or g_gender.lower() in ('unknown', '未知') or g_gender == gender:
|
||||
target = g
|
||||
break
|
||||
elif groups:
|
||||
target = groups[0]
|
||||
if target is None:
|
||||
target = {}
|
||||
groups.append(target)
|
||||
for k, v in feats.items():
|
||||
v_str = str(v).strip() if v is not None else ''
|
||||
if v_str and v_str.lower() != 'unknown':
|
||||
target[k] = v_str
|
||||
elif k not in target:
|
||||
target[k] = v_str or 'unknown'
|
||||
uid_features: Dict[str, Dict] = {}
|
||||
for base_uid, groups in uid_groups.items():
|
||||
for i, feats in enumerate(groups):
|
||||
key = base_uid if i == 0 else f"{base_uid}#{i + 1}"
|
||||
uid_features[key] = feats
|
||||
return uid_features
|
||||
|
||||
def _collect_features_text(self, labels: List[str]) -> str:
|
||||
|
||||
72
fam-edge/tests/test_oracle_db.py
Normal file
72
fam-edge/tests/test_oracle_db.py
Normal file
@@ -0,0 +1,72 @@
|
||||
import json
|
||||
|
||||
from fam_edge.oracle_db import OracleDB
|
||||
|
||||
|
||||
def _db(tmp_path):
|
||||
return OracleDB(str(tmp_path / "oracle.db"))
|
||||
|
||||
|
||||
def test_upsert_person_same_gender_merges_into_one_row(tmp_path):
|
||||
db = _db(tmp_path)
|
||||
db.upsert_person("人物A", features={"gender": "男", "hair": "短发黑色"})
|
||||
db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色T恤"})
|
||||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["appearances"] == 2
|
||||
feats = json.loads(rows[0]["features_json"])
|
||||
assert feats["hair"] == "短发黑色"
|
||||
assert feats["clothing"] == "蓝色T恤"
|
||||
|
||||
|
||||
def test_upsert_person_gender_conflict_splits_into_new_label(tmp_path):
|
||||
"""核心诉求: 大模型给的 人物A/B/C 这类 uid 只在单次视频分析内稳定，不同视频
|
||||
独立编号，同一字符串完全可能撞到不同真人（实测 "人物A" 混了男女两个人）。
|
||||
性别冲突时不能直接合并覆盖，要拆成新 label，避免两个人的特征越merge越乱。"""
|
||||
db = _db(tmp_path)
|
||||
db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色Polo衫"})
|
||||
db.upsert_person("人物A", features={"gender": "女", "clothing": "白色上衣"})
|
||||
rows = {r["label"]: r for r in db._conn.execute("SELECT * FROM people").fetchall()}
|
||||
assert set(rows.keys()) == {"人物A", "人物A#2"}
|
||||
assert rows["人物A"]["appearances"] == 1
|
||||
assert json.loads(rows["人物A"]["features_json"])["gender"] == "男"
|
||||
assert rows["人物A#2"]["appearances"] == 1
|
||||
assert json.loads(rows["人物A#2"]["features_json"])["gender"] == "女"
|
||||
# 派生行的 display_uid 仍然记录原始大模型 uid，方便追溯来源
|
||||
assert rows["人物A#2"]["display_uid"] == "人物A"
|
||||
|
||||
|
||||
def test_upsert_person_gender_conflict_allocates_next_free_suffix(tmp_path):
|
||||
db = _db(tmp_path)
|
||||
db.upsert_person("人物A", features={"gender": "男"})
|
||||
db.upsert_person("人物A", features={"gender": "女"}) # -> 人物A#2
|
||||
db.upsert_person("人物A", features={"gender": "unknown"}) # unknown 不冲突，合并回 人物A
|
||||
db.upsert_person("人物A", features={"gender": "男", "hair": "光头"}) # 冲突，人物A 有 gender 男了不冲突；应仍合并
|
||||
labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()}
|
||||
assert labels == {"人物A", "人物A#2"}
|
||||
# 再来一次性别冲突（对着 人物A#2，性别女）应该分配 人物A#3，而不是复用 人物A#2
|
||||
db.upsert_person("人物A", features={"gender": "男"})
|
||||
db.upsert_person("人物A", features={"gender": "女"})
|
||||
# 这次新的女性冲突会先撞到 人物A（此时是男），分裂出下一个空闲后缀
|
||||
labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()}
|
||||
assert "人物A" in labels
|
||||
assert len(labels) >= 2
|
||||
|
||||
|
||||
def test_upsert_person_unknown_gender_never_triggers_split(tmp_path):
|
||||
db = _db(tmp_path)
|
||||
db.upsert_person("人物A", features={"gender": "男"})
|
||||
db.upsert_person("人物A", features={"gender": "unknown"})
|
||||
db.upsert_person("人物A", features={"gender": "未知"})
|
||||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["appearances"] == 3
|
||||
|
||||
|
||||
def test_upsert_person_no_features_never_triggers_split(tmp_path):
|
||||
db = _db(tmp_path)
|
||||
db.upsert_person("人物A", features={"gender": "男"})
|
||||
db.upsert_person("人物A") # 无 features（source 更新等场景）
|
||||
rows = db._conn.execute("SELECT * FROM people").fetchall()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["appearances"] == 2
|
||||
70
fam-edge/tests/test_person_service.py
Normal file
70
fam-edge/tests/test_person_service.py
Normal file
@@ -0,0 +1,70 @@
|
||||
import json
|
||||
|
||||
from fam_edge.oracle_db import OracleDB
|
||||
from fam_edge.person_service import PersonService
|
||||
|
||||
|
||||
def _service(tmp_path):
|
||||
db = OracleDB(str(tmp_path / "oracle.db"))
|
||||
svc = PersonService.__new__(PersonService) # 跳过 __init__（不需要真的建 LLM 适配器）
|
||||
svc.db = db
|
||||
return svc, db
|
||||
|
||||
|
||||
def _insert_event(db, video_id, appearances):
|
||||
db._conn.execute(
|
||||
"INSERT INTO events (video_id, ts, description, person_appearances_json) "
|
||||
"VALUES (?, ?, ?, ?)",
|
||||
(video_id, "2026-08-21 00:00:00", "desc", json.dumps(appearances, ensure_ascii=False)))
|
||||
db._conn.commit()
|
||||
|
||||
|
||||
def test_aggregate_features_merges_same_gender_across_videos(tmp_path):
|
||||
svc, db = _service(tmp_path)
|
||||
_insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男", "hair": "短发黑色"}}])
|
||||
_insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "男", "clothing": "蓝色T恤"}}])
|
||||
result = svc._aggregate_features()
|
||||
assert set(result.keys()) == {"人物A"}
|
||||
assert result["人物A"]["hair"] == "短发黑色"
|
||||
assert result["人物A"]["clothing"] == "蓝色T恤"
|
||||
|
||||
|
||||
def test_aggregate_features_splits_gender_conflict_across_videos(tmp_path):
|
||||
"""核心场景: 复现 "人物A" 在 29 个视频里混了男女两个人的真实 bug——不同视频各自
|
||||
独立编号的 uid，字符串相同不代表同一个真人，性别冲突时必须拆成独立分组。"""
|
||||
svc, db = _service(tmp_path)
|
||||
_insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男", "clothing": "蓝色Polo衫"}}])
|
||||
_insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "女", "clothing": "白色上衣"}}])
|
||||
result = svc._aggregate_features()
|
||||
assert set(result.keys()) == {"人物A", "人物A#2"}
|
||||
assert result["人物A"]["gender"] == "男"
|
||||
assert result["人物A#2"]["gender"] == "女"
|
||||
|
||||
|
||||
def test_aggregate_features_unknown_gender_joins_first_group(tmp_path):
|
||||
svc, db = _service(tmp_path)
|
||||
_insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男"}}])
|
||||
_insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "unknown", "hair": "光头"}}])
|
||||
result = svc._aggregate_features()
|
||||
assert set(result.keys()) == {"人物A"}
|
||||
assert result["人物A"]["gender"] == "男"
|
||||
assert result["人物A"]["hair"] == "光头"
|
||||
|
||||
|
||||
def test_aggregate_features_three_way_gender_reuse_creates_three_groups(tmp_path):
|
||||
svc, db = _service(tmp_path)
|
||||
_insert_event(db, 1, [{"uid": "人物B", "features": {"gender": "男"}}])
|
||||
_insert_event(db, 2, [{"uid": "人物B", "features": {"gender": "女"}}])
|
||||
_insert_event(db, 3, [{"uid": "人物B", "features": {"gender": "男"}}]) # 应并回第一组
|
||||
result = svc._aggregate_features()
|
||||
assert set(result.keys()) == {"人物B", "人物B#2"}
|
||||
assert result["人物B"]["gender"] == "男"
|
||||
assert result["人物B#2"]["gender"] == "女"
|
||||
|
||||
|
||||
def test_aggregate_features_ignores_placeholder_uids(tmp_path):
|
||||
svc, db = _service(tmp_path)
|
||||
_insert_event(db, 1, [{"uid": "无人", "features": {"gender": "男"}}])
|
||||
_insert_event(db, 2, [{"uid": "", "features": {"gender": "男"}}])
|
||||
result = svc._aggregate_features()
|
||||
assert result == {}
|
||||
Reference in New Issue
Block a user