DARA - LLM+RL 双阶段广告预算分配 Agent

1. 解决的问题

婴儿推车季节性爆款上线,Google Ads 历史只有 3-5 周数据,传统规则策略难快速找到最优出价时段 - 数据要求:近 3-5 周 Google Ads ROAS 时段数据 + 月度总预算 - DARA 配置:T = 7(一周)或 24(一天时段);Phase 1 LLM 读历史生成日预算向量;Phase 2 每日 ROAS 反馈调整下一日 - 业务价值:冷启动期 ROAS 提升

2. 核心算法逻辑

固定总预算下,广告主需在 T 个时段/渠道间分配预算 $b_1, \ldots, b_T$ 最大化总回报。DARA 双阶段架构:Phase 1 Fewshot Reasoner(LLM 读历史 35 周数据生成初始分配) + Phase 2 Finegrained Optimizer(滑窗 ROI 反馈精调),用 GRPOAdaptive(动态参考策略 RL)训练 LLM,解决冷启动 + 数值精度双难题。

3. 业务应用场景

- 业务问题:婴儿推车季节性爆款上线,Google Ads 历史只有 3-5 周数据,传统规则策略难快速找到最优出价时段 - 数据要求:近 3-5 周 Google Ads ROAS 时段数据 + 月度总预算 - DARA 配置:T = 7(一周)或 24(一天时段);Phase 1 LLM 读历史生成日预算向量;Phase 2 每日 ROAS 反馈调整下一日 - 业务价值:冷启动期 ROAS 提升 15-30%,新品 GMV 增量 30-60 万元/月;以月预算 100 万元计 = 年化收益 360-720 万元

- 业务问题:母婴品牌同投 Google Shopping + Meta DPA,每周需调整两渠道预算比例,但 marginal ROAS 随节促(618/双11)动态变化 - 数据要求:跨渠道历史 ROAS + CPC/CPM + 竞品节奏数据 - DARA 配置:T = 渠道数(2-5);Phase 1 LLM 读历史 + 竞品输出初始权重;Phase 2 实时 ROAS 反馈精调,目标"边际 ROAS 等价" - 业务价值:跨渠道 ROAS 提升 10-20%,大盘 GMV 增量 5-8%;以中型品牌月广告 500 万元计 = 年化增量 600-1200 万元

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 难处:论文无官方代码,GRPO-Adaptive 需基于 TRL 库自行实现
• 难处:LLM 微调需大量历史数据 + RL 训练经验
• 难处:Phase 2 实时反馈需要 Google/Meta Ads API 集成

7. 代码模板

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

"""
DARA-lite: 双阶段 LLM 广告预算分配骨架
论文 arXiv:2601.14711 (WWW 2026, 阿里巴巴)
注: 实际部署需要 OpenAI API 或本地 LLM (Qwen2.5/DeepSeek-R1) + TRL.GRPOTrainer
本骨架使用纯规则模拟 LLM 决策,验证 算法骨架。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Tuple


@dataclass
class AdChannel:
    name: str
    budget: float
    actual_roas: float


def phase1_reasoner(history: List[AdChannel], total_budget: float, channels: List[str]) -> Dict[str, float]:
    """Few-shot Reasoner: 历史数据 → 初始预算分配
    简化版: 按历史 ROAS 比例分配(实际生产替换为 LLM 调用)
    """
    roas_by_channel: Dict[str, float] = {}
    for h in history:
        roas_by_channel[h.name] = roas_by_channel.get(h.name, 0.0) + h.actual_roas
    counts = {c: sum(1 for h in history if h.name == c) for c in channels}
    avg_roas = {c: roas_by_channel.get(c, 1.0) / max(counts.get(c, 1), 1) for c in channels}
    total_roas = sum(avg_roas.values()) or 1.0
    return {c: total_budget * avg_roas[c] / total_roas for c in channels}


def phase2_optimizer(
    allocation: Dict[str, float],
    feedback: Dict[str, float],
    total_budget: float,
    learning_rate: float = 0.1,
) -> Dict[str, float]:
    """Fine-grained Optimizer: 实时 ROAS 反馈 → 精调
    目标: 各渠道边际 ROAS 趋于相等
    """
    if not feedback:
        return allocation

    avg_roas = sum(feedback.values()) / len(feedback)
    new_alloc = {}
    for c, b in allocation.items():
        roas = feedback.get(c, avg_roas)
        delta = (roas - avg_roas) * learning_rate * b
        new_alloc[c] = max(b + delta, 100.0)

    total = sum(new_alloc.values())
    return {c: v * total_budget / total for c, v in new_alloc.items()}


def simulate_marginal_roas(allocation: Dict[str, float], base_roas: Dict[str, float], saturation: float = 1000.0) -> Dict[str, float]:
    """模拟边际 ROAS 衰减(实际场景从 Ads API 拉取)"""
    return {c: base * (saturation / (saturation + allocation[c])) for c, base in base_roas.items()}


def main() -> None:

8. 论文来源

• 2601.14711