PRIME-RL 中的多智能体系统
今天,Prime Intellect 的 RL 技术栈从训练单个智能体扩展到了多智能体系统。现在,你可以编程定义智能体之间的任意交互方式,选择哪些角色参与学习,并在整个交互过程中进行信用分配。
两个近期引入的抽象为这一进展奠定了基础:verifiers v1 引入了在可替换的 harness 和 runtime 上运行单个智能体以完成可编程任务的基础原语;prime-rl 则增加了算法层,使从 rollouts 到训练信号的映射变得可编程。
今天,我们将这两部分整合起来,引入了支持多智能体训练和评估的一级抽象。一些现在可以表达的典型示例包括:
- Agentic Judging — 由评判智能体对求解智能体的轨迹进行评分
- Self-Play — 模型与自身对弈
- User-Sim — 用户智能体与助手智能体交互
本文将介绍两个核心抽象——Agent 和 Env,解释其背后的设计选择,并展示几种我们认为能为 Agentic RL 开辟新方向的交互模式。
Agent
verifiers v1 原生支持了定义单个智能体 rollout 的所有基础原语:
Taskset提供一组Task对象Harness定义驱动模型的程序Runtime定义程序的执行环境
Agent 自然成为这些组件的承载者。其核心签名是 Agent.run(task: Task) -> Trace —— 当传入一个 Task 时,智能体产生一个 Trace,即 rollout 生成的产物。
class Agent:
async def run(self, task: Task, *, runtime: Runtime) -> Trace:
"""Produce a trace given a task."""
...
在这背后,agent 负责管理所有隐藏机制的完整生命周期,让你可以在任意 runtime 中、针对任意任务运行任意 agent。把这个抽象提升为一等公民后,你就能轻松地在其上编写脚本,而多智能体环境不过是基于 agents 编写程序的一种实例化。
Env
现在,Env 成为了多智能体训练和评估的载体。它的核心签名是 Env.run(task: Task, agents: Agents) -> None:接收一个初始任务和一组预先初始化好的 agents,然后编排完整的多智能体控制流;每个完成的 agent 运行会自动汇入最终的 Episode。
class Env(ABC):
@abstractmethod
async def run(self, task: Task, agents: Agents) -> None:
"""Run a single multi-agent episode."""
...
verifiers v1 中能用的一切表达,在 SingleAgentEnv 里都收敛成一个单行的 run 方法。
class SingleAgentEnvConfig(EnvConfig):
agent: AgentConfig = AgentConfig()
class SingleAgentEnv(Env[SingleAgentEnvConfig]):
async def run(self, task: Task, agents: Agents) -> None:
await agents.agent.run(task)
接下来我们将介绍四个已实现的多智能体环境:它们不仅本身实用,也很好地展示了这些抽象所能带来的可能性。
Agentic Judging
我们之前那篇关于扩展 agentic RL 的文章结尾提出了一个确定性评分无法解决的问题:固定评分器往往过于狭窄。比如在软件工程中,测试可能会断言某个特定的实现细节,导致明明有效的解法却错误地拿到零奖励。
朴素的 LLM 评判者解决不了这个问题——受限于单次调用,指望它给出准确判断并不现实。而 agentic 评判者则可以探索代码库、查看失败的测试,进而推翻确定性测试的判定。
AgenticJudgeEnv 定义了 solver 与 judge 两个 agent 之间的顺序交互。
class AgenticJudgeEnvConfig(vf.EnvConfig):
solver: vf.AgentConfig = vf.AgentConfig()
judge: vf.AgentConfig = vf.AgentConfig()
class AgenticJudgeEnv(vf.Env[AgenticJudgeEnvConfig]):
async def run(self, task, agents) -> None:
solution = await agents.solver.run(task)
if not solution.ok:
raise
await agents.judge.run(JudgeTask.from_trace(solution))
示例。完整实现见此处。
Self-Play
任务稀缺是 LLM RL 的核心瓶颈。有效的学习信号依赖于接近 Agent 当前能力水平的任务。Self-Play 允许模型生成自身课程,作为静态任务集之外的替代方案。随着模型进步,它可以构建更难的任务、成为更强的对手,或发现新的失败案例。我们认为,Self-Play 将成为推动前沿发展的重要技术。
本节介绍 Self-Play 的两个变体:Proposer-Solver 和 Kuhn-Poker。
Proposer-Solver
在 ProposerSolverEnv 中,Proposer 接收一个种子主题并构建新任务,随后由一组 Solver 解决。每个 Solver 答对获得奖励,而 Proposer 则根据与 Solver 群体的校准程度获得奖励。若所有 Solver 均成功,说明任务过易;若无人成功,则可能过难或无效。环境的可学习性在 50% 求解率时达到峰值,这鼓励 Proposer 生成能产生最大 RL 训练信号的任务(受 Zhao et al. (2025) 的 Absolute Zero 启发)。
该交互还需要角色感知的 Credit Assignment。Solver 的尝试应与针对同一生成问题的尝试进行比较,而 Proposer 的轨迹则应与由同一种子任务生成的其他提案进行比较。经典的 GRPO 无法表达这种层级结构,因此我们实现了Hierarchical GRPO,以保持这些比较集,避免混合角色或问题难度。
class ProposerSolverEnvConfig(vf.EnvConfig):
proposer: vf.AgentConfig = vf.AgentConfig()
solver: vf.AgentConfig = vf.AgentConfig()
n: int = Field(4, ge=1)
class ProposerSolverEnv(vf.Env[ProposerSolverEnvConfig]):
async def run(self, task: vf.Task, agents: vf.Agents) -> None:
proposed = await agents.proposer.run(task)
solve_task = SolveTask.from_trace(proposed)
async with asyncio.TaskGroup() as tg:
for _ in range(self.config.n):
tg.create_task(agents.solver.run(solve_task))
async def finalize(self, task: vf.Task, episode: vf.Episode) -> None:
rate = solve_rate(episode.traces) # 平均求解率
for trace in episode.traces:
if trace.agent.name == "proposer":
trace.record_metric("solve_rate", rate)
trace.record_reward("learnability", 4.0 * rate * (1.0 - rate))
示例代码。完整实现见此处。
Kuhn-Poker
在 KuhnPokerEnv 中,两个模型进行扑克游戏。环境维护底牌、公共游戏状态及合法动作。随着策略优化,模型同时成为更强的对手,从而生成动态课程,无需独立的对手服务。
由于不同角色的奖励分布可能在结构上存在差异,我们的训练系统在此场景下支持角色条件优势估计(Role-Conditioned Advantage Estimation, RAE)。不同于让所有智能体对比同一共享基线,RAE 基于各角色自身的奖励历史进行评估。
User-Sim
回合级交错(turn-level interleaving)不仅适用于回合制游戏。许多助手任务无法用单一的提示词和响应来表征:用户拥有私有上下文,随时间逐步透露信息,对助手的反应做出回应,并自行判断目标是否达成。
UserSimEnv 将用户建模为智能体,整个剧集(episode)即为用户智能体与助手智能体之间逐轮进行的对话。
两条 Traces 都保留在 Episode 中。默认情况下模拟用户是冻结的,而助手的 Trace 会对照原始 Task 进行评分,用于训练。
这样一来,用户模拟就成了同一套基础组件的另一种组合方式,而不是一条独立的评测路径。不同类型的用户群体、人设、隐藏目标和交互策略都可以插入同一接口,而助手侧则可以单独配置。
class UserSimEnvConfig(vf.EnvConfig):
assistant: vf.AgentConfig = vf.AgentConfig()
user: vf.AgentConfig = vf.AgentConfig()
class UserSimEnv(vf.Env[UserSimEnvConfig]):
async def run(self, task, agents):
user_task, assistant_task = ..., ...
async with (
agents.user.interaction(user_task) as user,
agents.assistant.interaction(assistant_task) as assistant,
):
ask = await user.turn("Hello! How can I help you today?")
while True:
answer = await assistant.turn(ask.last_reply)
if answer.terminated:
break
ask = await user.turn(answer.last_reply)
if ask.terminated:
break
示例仅供参考。完整实现请见这里。
RL 之外的 Agent 应用
我们预期这套 agent 抽象在训练和评测之外也能派上用场。我们已经在构建强大的合成数据生成与整理流水线。每个 agent 的 trace 都是统一且可审计的数据产物。
举个例子,下面是 Laguna-S2.1 在 pool harness 中查找最新发布的 verifiers 版本。
import asyncio
import verifiers.v1 as vf
async def main():
agent = vf.make_agent(
vf.AgentConfig(
model="poolside/laguna-s-2.1",
harness={"id": "pool"},
runtime=vf.PrimeConfig()
)
)
task = vf.Task(
vf.TaskData(
prompt="Find the latest released version of the `verifiers` Python "
"package on PyPI. Answer with just the version string."
)
)
async with agent:
trace = await agent.run(task)
print(trace.last_reply)
asyncio.run(main())
Multi-agent 功能今日正式上线,支持版本包括 verifiers 0.3.0 和 prime-rl 0.8.0。我们致力于赋予研究者表达 multi-agent RL 的能力,推动开源 AGI 的前沿探索。
引用
@article{primeintellect2026multiagent,
author = {Konstantin Dunas and Mika Senghaas and Eli Gottlieb and Prime Intellect Team},
title = {Multi-Agent Systems in PRIME-RL},
journal = {Prime Intellect Blog},
year = {2026},
month = {August},
note = {https://www.primeintellect.ai/blog/multi-agent-systems}
}