EvoPool — 进化式多智能体弱监督标注

Skill-LLM-Annotation-Weak-Supervision · 09-DataAgent-LLM

causalexperimentmulti_agentdata_collection广告与投放客服与VOC推荐与搜索数据采集与治理MAS与智能体工程定价与利润风控与合规WF-B 广告优化WF-C 客服分诊WF-E Review监控WF-I 智能体工程WF-K 全域风险防御

收录于数据治理基础手册 AI Agent 替人手册

年化 ROI80-150 万元

实现难度⭐⭐⭐☆☆

业务优先级⭐⭐⭐⭐☆

业务视角

适用角色数据分析师 / 运营负责人 · CEO · 供应链负责人

适用平台Amazon SP API · Shopify · TikTok Ads API · 多平台数据整合

什么情况下用数据需求太多，数据团队排期 2 周；非技术人员（采购/客服/运营）有数据问题但不会 SQL；重复报表占用大量时间

成功是什么样的业务方用自然语言自助查数据，常规报表自动化，数据驱动决策响应速度从「天」变「分钟」

业务痛点

数据需求排期太长不会 SQL 只能等数据团队老板临时要数据没法马上出分析师时间都花在取数上

1. 解决的问题

母婴跨境卖家面临海量多语言评论与工单需标注的成本困境——EvoPool 进化式弱监督框架将 LLM 直接标注 10 万条的 10 万次 API 调用压缩为 50 次，标注速度提升 4500x，年化节省标注人力成本 80-150 万元

2. 核心算法逻辑

EvoPool 的核心洞察：LLM 直接标注 100K 条数据成本极高（每条需一次 API 调用），但 LLM 能以极低成本生成标注规则代码，而规则代码运行几乎零成本。

3. 业务应用场景

业务问题：某母婴跨境品牌每月积累 30 万条全球买家评论（英/德/法），需要对每条评论打「质量投诉」「安全隐患」「正向体验」「物流问题」四个标签，用于训练合规预警模型。人工标注 10 条/人/小时，30 万条需 3 万人时（约 90 万元/轮）。

数据要求： - 原始评论文本（无需预处理） - 500 条人工验证集（用于 Fitness Gate 评估） - 可选：产品类目标签（奶瓶/推车/睡袋等）

执行流程： 1. LLM 生成初代 20 个标注函数（关键词匹配 + 正则 + 简单分类器） 2. 经 3-5 代进化，池扩充至 40-60 个函数 3. EvoAgg 聚合软标签，训练下游 BERT 分类器 4. 下游模型上线后持续接收新评论，零成本标注

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

需要 200-500 条高质量种子标注数据（可由业务专家一次性标注）
初代标注函数需要领域专家参与设计（半天工作量）
进化迭代需要调参经验
高优先级场景：评论/工单数据量 > 10 万条/月，且有持续标注需求
中优先级场景：数据量 1-10 万条/月，可考虑与人工标注混合使用
不适用场景：任务需要极高精度（医疗诊断），或数据量 < 1 万条

7. 代码模板

代码块数量：2 · 路径：未检测到

"""
EvoPool 简化实现：母婴评论弱监督标注
场景：亚马逊母婴产品评论的安全合规 + 情感二分类

核心思路：
1. 生成多个基于关键词/规则的标注函数（模拟进化池）
2. Fitness Gate 过滤低质量/冗余函数
3. EvoAgg 软标签聚合（加权投票 + 语义特征）
4. 输出软标签用于下游分类器训练
"""

import re
import numpy as np
from collections import Counter
from typing import List, Tuple, Optional

# ==============================================================
# 1. 标注函数定义（模拟 LLM 进化生成的函数池）
# ==============================================================

def annotate_safety_concern(text: str) -> int:
    """检测安全隐患相关评论 (1=有隐患, 0=无, -1=弃权)"""
    safety_keywords = [
        "choke", "choking", "hazard", "danger", "injury", "hurt",
        "recall", "toxic", "sharp", "unsafe", "accident", "warning",
        "危险", "安全", "割伤", "吞咽", "有毒", "召回"
    ]
    text_lower = text.lower()
    hits = sum(1 for kw in safety_keywords if kw in text_lower)
    if hits >= 2:
        return 1
    if hits == 1:
        return -1  # 弃权（不确定）
    return 0

def annotate_quality_complaint(text: str) -> int:
    """检测质量投诉 (1=投诉, 0=无, -1=弃权)"""
    complaint_patterns = [
        r"broke? (after|in|within)",
        r"stopped? work",
        r"poor quality",
        r"cheap|cheaply made",
        r"fell apart",
        r"doesn'?t? work",
        r"defect",
        r"broken|crack",
    ]
    text_lower = text.lower()
    for pat in complaint_patterns:
        if re.search(pat, text_lower):
            return 1
    negative_words = ["disappointed", "waste", "terrible", "awful", "horrible"]
    if any(w in text_lower for w in negative_words):
        return 1
    return -1  # 大多数情况弃权，让其他函数决定

def annotate_positive_sentiment(text: str) -> int:
    """检测正向情感 (1=正向, 0=负向, -1=弃权)"""
    positive_strong = ["love", "excellent", "perfect", "amazing", "wonderful", "great", "best"]
    negative_strong = ["hate", "terrible", "awful", "horrible", "worst", "never again"]

8. 论文来源

2606.01617