RCR-Router角色感知上下文路由 — Token预算约束下的多Agent记忆子集动态分配
Skill-RCR-Router-Role-Aware-Context-Routing · 10-MAS
1. 解决的问题
让所有Agent访问完整记忆库既浪费Token又引入噪声干扰——RCR-Router按角色和任务阶段动态选择语义相关记忆子集，在严格Token预算下回答质量提升13-22%（2025 arXiv:2508.04903）
2. 核心算法逻辑
反直觉洞察：大多数MAS系统让每个Agent访问完整的共享记忆池——这看起来"信息最丰富"，实际上是一种浪费甚至有害：财务Agent看到的大量研究原始数据对它毫无用处（噪声），而合规Agent需要的法规细节却被稀释在海量市场数据中。RCRRouter的关键发现：按角色和任务阶段动态选择语义相关的记忆子集，不仅节省Token，还能提高回答质量（减少噪声干扰）。
3. 业务应用场景
- 业务问题：母婴MAS有一个包含5000条记忆的共享库（市场数据+法规文件+财务记录+品牌指南）。每次调用要给所有Agent传入完整记忆库，严重超出Token预算，而且Finance Agent收到大量无关的品牌指南信息，导致ROI计算时被不相关信息干扰 - RCR-Router方案： - Research Agent (阶段1)：路由市场数据+竞品记忆（B=2048） - Compliance Agent：路由法规文件+认证记录（B=1536） - Finance Agent (阶段2)：路由财务模板+历史ROI记录（B=1024） - Report Agent (阶段3)：路由所有Age
- **业务问题**：Prime Day实时分析中，早期阶段的研究记忆（"吸奶器竞品分析"）在财务决策阶段仍被频繁路由，但此时最相关的是实时销售数据 - **迭代精炼机制**：随着任务阶段推进，重要性评分自动降低旧研究数据的权重，提升实时数据权重；Agent在大促后期收到的上下文越来越聚焦于"当前销售状态+历史决策"，而非初期的"市场背景" - **预期产出**：大促后期决策质量提升15%，Token消耗减少25%
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：月调用5000次MAS的平台，RCR-Router节省约20%Token（无关信息不传入Agent），同时提升关键Agent质量13-22%；年化Token节省约$600，质量提升间接减少错误决策成本更高；系统成本$4万，综合ROI≈200%（首年），后续年ROI持续提升
- 实施难度：⭐⭐⭐☆☆（规则基础版较简单；学习版重要性评分需要角色标签库；需要改造MAS的记忆访问接口）
- 优先级：⭐⭐⭐⭐☆（在记忆库>50条、Agent数>3的MAS中，无选择性地传入所有记忆会严重降低质量，RCR-Router是必要组件）
- 适用规模：共享记忆库>30条、多角色Agent（>3个不同角色）的MAS系统
- 数据依赖：需要为每个角色定义相关标签（可从Agent的SRL蓝图中自动提取）；不需要额外训练数据
7. 代码模板
代码块数量：4 · 路径：未检测到
"""
RCR-Router角色感知上下文路由系统
功能：Token预算分配 + 重要性评分 + 语义过滤 + 迭代上下文精炼
基于 arXiv:2508.04903 + 2602.06025 (2025-2026)
"""
import numpy as np
import re
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from collections import defaultdict
import warnings
warnings.filterwarnings('ignore')
class BudgetTier:
LOW = "low" # 简化检索，< 512 tokens
MID = "mid" # 标准检索，512-2048 tokens
HIGH = "high" # 深度检索，> 2048 tokens
@dataclass
class MemoryItem:
"""记忆库中的单条记忆"""
item_id: str
content: str
source_agent: str
task_stage: int # 0=信息收集, 1=分析, 2=决策, 3=报告
relevance_tags: List[str] = field(default_factory=list) # 相关角色标签
citation_count: int = 0 # 被引用次数
creation_round: int = 0 # 创建轮次
importance_score: float = 0.5 # 当前重要性分数（动态更新）
@property
def token_count(self) -> int:
return max(len(self.content) // 4, 1)
class ImportanceScorer:
"""重要性评分器"""
ROLE_TAG_MAP = {
'research_agent': ['market', 'competitor', 'trend', '市场', '竞品', '增长'],
'compliance_agent': ['regulation', 'cpsc', 'fda', 'compliance', '合规', '认证', '法规'],
'finance_agent': ['roi', 'cost', 'revenue', 'financial', '成本', '利润', 'fba', '财务'],
'report_agent': ['conclusion', 'recommendation', 'summary', '结论', '建议', '报告'],
}
STAGE_RELEVANCE = {
0: ['market', 'competitor', '市场', '竞品'], # 信息收集阶段
1: ['analysis', 'trend', '分析', '趋势'], # 分析阶段
2: ['decision', 'roi', 'risk', '决策', 'ROI', '风险'], # 决策阶段
3: ['conclusion', 'summary', '结论', '建议'], # 报告阶段
}
def score(self, item: MemoryItem, agent_role: str,
task_stage: int, current_round: int) -> float:
"""计算记忆项对特定角色在特定阶段的重要性"""
score = 0.5 # 基础分
# 1. 角色相关性8. 论文来源
- 2508.04903
- 2602.06025