paper2skills Playbook

MAA 多 Agent 行动建议 - 从评论到产品改进决策链

Skill-MAA-Review-to-Action-Decision · 14-用户分析

causalexperimentrecommendationmulti_agentpricing广告与投放客服与VOC推荐与搜索数据采集与治理MAS与智能体工程定价与利润WF-B 广告优化WF-C 客服分诊WF-D 选品扫描WF-E Review监控WF-F 动态定价WF-H 复购增长
年化 ROI510-920 万元/年
实现难度⭐⭐⭐⭐☆
业务优先级⭐⭐⭐⭐☆
业务视角
适用角色运营负责人 / 用户增长负责人 · CMO · 产品经理
适用平台Amazon 买家分层 · DTC 站 LTV 预测 · Klaviyo/Brevo 邮件分群
什么情况下用有大量老客户,但不知道谁是高价值客户、谁快要流失;新客获取成本越来越高,老客复购却上不去
成功是什么样的用户按 RFM/LTV 分层精准触达,高价值用户留存率提升,老客贡献收入占比从 30% 提升到 50%
业务痛点
老客复购率上不去不知道哪些用户要流失了所有用户用同一套活动买过一次就不见了

1. 解决的问题

Momcozy M5 吸奶器在美国/德国/中国三市场销售,各市场用户痛点完全不同(美国关注续航便携、德国关注静音认证、中国关注清洗方便). 现有运营复盘只产出"差评列表",无法直接驱动产品改进决策——产品经理拿到差评列表还要花 1-2 周二次提炼 - 数据要求:三市场 Amazon Review API + market 标签 - MAA 配置: - 按市场分别聚类(K=5,每市场 5

2. 核心算法逻辑

传统评论分析停留在"描述性"层面(情感/属性). MAA 将其升级为"规范性决策链路":评论 → 问题 → 建议 → 评估 → 排序. 通过 5 个 Agent 分工,把大规模评论语料蒸馏成企业可直接执行的行动清单,而非分析报告.

3. 业务应用场景

- 业务问题:Momcozy M5 吸奶器在美国/德国/中国三市场销售,各市场用户痛点完全不同(美国关注续航便携、德国关注静音认证、中国关注清洗方便). 现有运营复盘只产出"差评列表",无法直接驱动产品改进决策——产品经理拿到差评列表还要花 1-2 周二次提炼 - 数据要求:三市场 Amazon Review API + market 标签 - MAA 配置: - 按市场分别聚类(K=5,每市场 5 个主题簇) - Issue Agent 提取每个簇的核心痛点(如"夜间使用嘈杂吵醒宝宝") - Recommendation Agent 生成 3-4 条改进建议(如"加静音模式降至 35dB"/

- 业务问题:Q1-Q4 季度复盘,人工归纳 3 个产品 × 4 季度 = 12 次,每次 3-5 PM 天,总计 36-60 PM 天/年;复盘结论高度依赖个人经验,新人接手质量大幅波动 - 数据要求:季度 Amazon + Wayfair 评论合并 - MAA 配置: AGRS 摘要 → MAA 5 Agent 决策链 → 输出"季度 Top 3-5 优先改进项 + ROI 排序" - 业务价值: - 节省人工:36-60 PM 天 × 3000元/天 = 10-18 万/年 - 决策质量标准化(SRAC 评分客观可比),新人接手不衰退 - 改进 ROI 提升 20-30% = 额外 10

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

  • 易处:论文 prompt 模板完整公开,5 Agent 模块清晰
  • 难处:需要业务专家初始化 Ranking Agent 的成本/周期评分规则
  • 难处:5 Agent 串联推理 token 成本较高(单产品估算 $0.05-0.15 GPT-4o-mini)

7. 代码模板

代码块数量:1 · 路径:未检测到

"""
MAA Multi-Agent Actionable Advice 最小骨架
论文 arXiv:2601.12024 (Bhandari et al., 2026-01)
完整实现见 paper2skills-code/nlp_voc/maa_actionable_advice/model.py
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List


@dataclass
class Review:
    text: str
    review_id: str = ""
    market: str = ""


@dataclass
class SRAC:
    S: int = 0
    R: int = 0
    A: int = 0
    C: int = 0

    @property
    def score(self) -> float:
        return 0.25 * (self.S + self.R + self.A + self.C)


def clustering_agent(reviews: List[Review], k: int = 5) -> Dict[int, List[Review]]:
    """阶段 1: 聚类 + 选代表(生产替换为 TF-IDF + K-Means)"""
    keyword_buckets = {
        0: ["noise", "loud", "quiet"],
        1: ["clean", "wash", "hygiene"],
        2: ["battery", "portable", "wireless"],
        3: ["price", "expensive", "value"],
        4: ["build", "sturdy", "broken"],
    }
    clusters = defaultdict(list)
    for r in reviews:
        text_low = r.text.lower()
        best_cluster, best_match = -1, 0
        for cid, kws in keyword_buckets.items():
            match = sum(1 for kw in kws if kw in text_low)
            if match > best_match:
                best_cluster, best_match = cid, match
        if best_cluster >= 0:
            clusters[best_cluster].append(r)
    return dict(clusters)


def issue_agent(cluster_reps: Dict[int, List[Review]]) -> List[Dict]:
    """阶段 2: 抽取主题 + 具体问题"""
    issues = []
    for cid, reps in cluster_reps.items():
        if not reps:
            continue
        themes = ["noise", "cleaning", "portability", "value", "build_quality"]
        theme = themes[cid] if cid < len(themes) else "general"

8. 论文来源

  • 2601.12024