From f798c31cab8df354b4d30db127278d9dd449e0ad Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Sat, 22 Aug 2026 08:45:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(fam-edge):=20=E4=BA=BA=E7=89=A9=20uid=20?= =?UTF-8?q?=E8=B7=A8=E8=A7=86=E9=A2=91=E5=A4=8D=E7=94=A8=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E7=89=B9=E5=BE=81=E4=B8=B2=E4=BA=BA=20-=20=E6=8B=86=E5=88=86?= =?UTF-8?q?=E5=90=8C=20uid=20=E4=B8=8B=E7=9A=84=E6=80=A7=E5=88=AB=E5=86=B2?= =?UTF-8?q?=E7=AA=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: 大模型给的 人物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 个人物分组 （我/媳妇/爷爷/汤圆）特征都是内部一致的，不再互相串。 --- fam-edge/src/fam_edge/oracle_db.py | 28 ++++++++++ fam-edge/src/fam_edge/person_service.py | 49 ++++++++++++----- fam-edge/tests/test_oracle_db.py | 72 +++++++++++++++++++++++++ fam-edge/tests/test_person_service.py | 70 ++++++++++++++++++++++++ 4 files changed, 205 insertions(+), 14 deletions(-) create mode 100644 fam-edge/tests/test_oracle_db.py create mode 100644 fam-edge/tests/test_person_service.py diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index e1f09ef..2b87f8c 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -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 ( diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index 7457ab3..8147b28 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/fam-edge/src/fam_edge/person_service.py @@ -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: diff --git a/fam-edge/tests/test_oracle_db.py b/fam-edge/tests/test_oracle_db.py new file mode 100644 index 0000000..8fc0742 --- /dev/null +++ b/fam-edge/tests/test_oracle_db.py @@ -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 diff --git a/fam-edge/tests/test_person_service.py b/fam-edge/tests/test_person_service.py new file mode 100644 index 0000000..83f022c --- /dev/null +++ b/fam-edge/tests/test_person_service.py @@ -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 == {}