"""find_agent 的确定性年龄画像标准化与流程审计工具。""" from __future__ import annotations import json import re from typing import Any from supply_agent.tools import tool _ROOT_SOURCE_TYPES = {"demand", "seed", "point", "mixed"} def _number(value: Any) -> float | None: if value is None or isinstance(value, bool): return None text = str(value).strip().replace("%", "") if not text: return None try: return float(text) except ValueError: return None def _ratio(value: Any) -> float | None: number = _number(value) if number is None: return None if number > 1: number /= 100 return min(max(number, 0.0), 1.0) def _age_dimension(portrait: dict[str, Any] | None) -> dict[str, Any]: if not isinstance(portrait, dict): return {} for wrapper_key in ("portrait_data", "content", "account"): wrapped = portrait.get(wrapper_key) if isinstance(wrapped, dict): wrapped_dimension = _age_dimension(wrapped) if wrapped_dimension: return wrapped_dimension for key in ("年龄", "age", "Age", "年龄分布"): value = portrait.get(key) if isinstance(value, dict): return value if portrait and all(isinstance(value, dict) for value in portrait.values()): return portrait return {} def _bucket_kind(label: str) -> str: compact = ( label.strip() .lower() .replace("岁", "") .replace("以上", "+") .replace("及", "") .replace(" ", "") ) if compact in {"50-", "50+", ">=50", "≥50", "51+", "60+", ">=60", "≥60"}: return "older" if re.search(r"(?:^|[^0-9])(50|51|60)\+$", compact): return "older" numbers = [int(value) for value in re.findall(r"\d+", compact)] if not numbers: return "unknown" if len(numbers) == 1: if numbers[0] >= 50 and any( marker in compact for marker in ("+", ">=", "≥", "以上") ): return "older" return "unknown" lower, upper = min(numbers), max(numbers) if lower >= 50: return "older" if lower >= 40 or upper >= 50: return "mature" return "younger" def _normalize_portrait(portrait: dict[str, Any] | None) -> dict[str, Any]: dimension = _age_dimension(portrait) buckets: list[dict[str, Any]] = [] older_ratio = 0.0 mature_ratio = 0.0 older_tgi_values: list[tuple[float, float]] = [] for label, raw_metrics in dimension.items(): metrics = raw_metrics if isinstance(raw_metrics, dict) else {} percentage = _ratio( metrics.get("percentage") if "percentage" in metrics else metrics.get("ratio") ) tgi = _number( metrics.get("preference") if "preference" in metrics else metrics.get("tgi") ) kind = _bucket_kind(str(label)) buckets.append( { "label": str(label), "kind": kind, "ratio": percentage, "tgi": tgi, } ) if percentage is None: continue if kind == "older": older_ratio += percentage if tgi is not None: older_tgi_values.append((tgi, percentage)) elif kind == "mature": mature_ratio += percentage weighted_tgi = None tgi_weight = sum(weight for _, weight in older_tgi_values) if tgi_weight: weighted_tgi = sum(tgi * weight for tgi, weight in older_tgi_values) / tgi_weight if not dimension: strength = "missing" elif ( older_ratio >= 0.45 or (older_ratio >= 0.35 and (weighted_tgi or 0) >= 100) or (older_ratio >= 0.20 and (weighted_tgi or 0) >= 130) ): strength = "strong" elif ( older_ratio >= 0.20 or (older_ratio > 0 and (weighted_tgi or 0) >= 100) or mature_ratio >= 0.30 ): strength = "moderate" else: strength = "weak" return { "has_age_portrait": bool(dimension), "older_ratio": round(older_ratio, 6), "older_tgi": round(weighted_tgi, 4) if weighted_tgi is not None else None, "mature_ratio": round(mature_ratio, 6), "strength": strength, "buckets": buckets, } def normalize_age_portrait_pair( content_portrait: dict[str, Any] | None, account_portrait: dict[str, Any] | None = None, ) -> dict[str, Any]: """Return the deterministic two-sided age normalization payload.""" content = _normalize_portrait(content_portrait) account = _normalize_portrait(account_portrait) content_has = content["has_age_portrait"] account_has = account["has_age_portrait"] if content_has and account_has: strong_set = {"strong", "moderate"} consistency = ( "aligned" if (content["strength"] in strong_set) == (account["strength"] in strong_set) else "conflict" ) cap = 1.0 elif account_has: consistency = "account_only" cap = 0.65 elif content_has: consistency = "content_only" cap = 1.0 else: consistency = "missing" cap = 0.35 return { "content": content, "account": account, "consistency": consistency, "elder_score_cap": cap, } @tool def normalize_age_portraits( content_portrait: dict[str, Any], account_portrait: dict[str, Any] | None = None, ) -> str: """ 标准化视频点赞画像与作者粉丝画像中的年龄桶。 能识别接口实际返回的 `50-`,以及 `50+ / 50岁以上 / >=50 / 41-50` 等表达,统一输出直接老年比例、TGI、成熟人群代理比例和证据强度。 本工具只标准化证据,不把点赞用户伪装成转发用户。 Args: content_portrait: 视频 portrait_data,或其中的年龄字典。 account_portrait: 可选作者 portrait_data,或其中的年龄字典。 Returns: JSON,包含 content、account、consistency 和 elder_score_cap。 """ normalized = normalize_age_portrait_pair(content_portrait, account_portrait) content = normalized["content"] account = normalized["account"] consistency = normalized["consistency"] cap = normalized["elder_score_cap"] result = { "title": "年龄画像标准化", **normalized, "output": ( f"视频侧={content['strength']},作者侧={account['strength']}," f"一致性={consistency},E上限={cap}" ), } return json.dumps(result, ensure_ascii=False) def _float_score(candidate: dict[str, Any], key: str) -> float | None: value = candidate.get(key) try: number = float(value) except (TypeError, ValueError): return None return number if 0 <= number <= 1 else None @tool def audit_video_discovery_process( searches: list[dict[str, Any]], candidates: list[dict[str, Any]], intended_status: str = "finished", ) -> str: """ 在结束找片前审计搜索树、翻页、标签扩展、证据完备性和最终分流。 Args: searches: 已执行搜索页。建议包含 search_id、keyword、source_type、 parent_search_id、cursor、page_no、has_more、new_candidate_count。 candidates: 已评估候选。建议包含 aweme_id、decision_bucket、R/E/S 分数、 detail_verified、content_portrait_attempted、account_portrait_attempted、 age_portraits_normalized、expansion_worthy_tags。 intended_status: 准备设置的运行状态,通常为 finished。 Returns: JSON,包含 can_finish、critical_violations、warnings 和 coverage。 """ critical: list[str] = [] warnings: list[str] = [] coverage_violations: list[str] = [] valid_searches = [item for item in searches if isinstance(item, dict)] valid_candidates = [item for item in candidates if isinstance(item, dict)] roots = [ item for item in valid_searches if item.get("source_type") in _ROOT_SOURCE_TYPES and not item.get("parent_search_id") and int(item.get("page_no") or 1) == 1 ] root_keywords = { str(item.get("keyword") or "").strip() for item in roots if item.get("keyword") } if len(root_keywords) < 2: coverage_violations.append("独立根搜索词少于2个") by_parent = { int(item["parent_search_id"]) for item in valid_searches if item.get("parent_search_id") is not None } keyword_pages = { (str(item.get("keyword") or ""), int(item.get("page_no") or 1)) for item in valid_searches } for item in valid_searches: page_no = int(item.get("page_no") or 1) if ( page_no != 1 or not item.get("has_more") or int(item.get("new_candidate_count") or 0) <= 0 ): continue search_id = item.get("search_id") keyword = str(item.get("keyword") or "") followed = ( search_id is not None and int(search_id) in by_parent ) or (keyword, page_no + 1) in keyword_pages if not followed: coverage_violations.append( f"生产性搜索页未翻页: search_id={search_id}, keyword={keyword}" ) tag_searches = [ item for item in valid_searches if item.get("source_type") == "tag" ] worthy_tags = { str(tag) for candidate in valid_candidates for tag in (candidate.get("expansion_worthy_tags") or []) if str(tag).strip() } if worthy_tags and not tag_searches: coverage_violations.append("存在值得扩展的标签,但没有 tag 搜索分支") bucket_counts = { "primary": 0, "rejected": 0, "pending_evaluation": 0, } pending_evaluation_messages: list[str] = [] rejected_misclass_messages: list[str] = [] for candidate in valid_candidates: aweme_id = str(candidate.get("aweme_id") or "unknown") bucket = str( candidate.get("decision_bucket") or "pending_evaluation" ) if bucket == "unreviewed": bucket = "pending_evaluation" if bucket not in bucket_counts: critical.append(f"{aweme_id} 使用了不支持的分池: {bucket}") continue bucket_counts[bucket] += 1 relevance = _float_score(candidate, "relevance_score") elder = _float_score(candidate, "elder_score") share = _float_score(candidate, "share_score") if bucket == "primary": if not candidate.get("detail_verified"): critical.append(f"{aweme_id} 未核验详情") if not candidate.get("content_portrait_attempted"): critical.append(f"{aweme_id} 未尝试视频画像") if not candidate.get("account_portrait_attempted"): critical.append(f"{aweme_id} 未尝试作者画像") if not candidate.get("age_portraits_normalized"): critical.append(f"{aweme_id} 未标准化年龄画像") if bucket == "primary" and not ( relevance is not None and relevance >= 0.55 and elder is not None and elder >= 0.55 and share is not None and share >= 0.50 ): critical.append(f"{aweme_id} 不满足 primary 阈值") meets_primary = ( relevance is not None and relevance >= 0.55 and elder is not None and elder >= 0.55 and share is not None and share >= 0.50 ) if bucket == "rejected" and meets_primary: rejected_misclass_messages.append( f"{aweme_id} 达到 primary 线却被错误淘汰" ) if bucket == "pending_evaluation": message = f"{aweme_id} 等待 Agent 补证和评估" if intended_status == "finished": pending_evaluation_messages.append(message) else: warnings.append(message) retained_count = bucket_counts.get("primary", 0) if retained_count >= 5 and intended_status == "finished": warnings.extend(pending_evaluation_messages) warnings.extend(rejected_misclass_messages) else: critical.extend(pending_evaluation_messages) critical.extend(rejected_misclass_messages) if retained_count < 5: warnings.append( f"当前保留 {retained_count} 条,低于优先目标 5 条;" "若仍有高价值搜索前沿应继续探索,候选确实不足时允许结束" ) if retained_count > 0: exploration_note = ( "已达到 5 条优先目标" if retained_count >= 5 else "尚未达到 5 条优先目标;仅在剩余前沿价值较低时允许结束" ) warnings.extend( f"{exploration_note};未继续探索: {item}" for item in coverage_violations ) else: critical.extend(coverage_violations) if not valid_candidates: warnings.append("没有候选;应确认是搜索无结果而非提前停止") can_finish = intended_status != "finished" or not critical result = { "title": "视频发现流程审计", "can_finish": can_finish, "critical_violations": list(dict.fromkeys(critical)), "warnings": list(dict.fromkeys(warnings)), "coverage": { "search_pages": len(valid_searches), "root_keywords": sorted(root_keywords), "tag_search_count": len(tag_searches), "candidate_count": len(valid_candidates), "retained_candidate_count": retained_count, "bucket_counts": bucket_counts, }, "output": ( f"can_finish={can_finish},严重问题 {len(set(critical))} 个," f"警告 {len(set(warnings))} 个" ), } return json.dumps(result, ensure_ascii=False)