AgentEvolver 深度研究：自顶向下跑通第一版训练

https://modelscope.github.io/AgentEvolver/

AgentEvolver 源码解读 · 第 1 章：工程骨架与训练入口

> 此篇为 C:/GitHub/AgentEvolver/ 真本源码之通读。前作《深度研究》已把三大机制讲成「鸟瞰」，此番自顶向下，先剖骨架——进程如何编排、训练主循环何在、配置如何收口。读毕此章，再去第 2–6 章看各模块，便知它们挂在哪根梁上。

---

一、仓库三层与「壳」「核」之分

AgentEvolver 不是从零造轮，而是骑在 veRL 之上的自进化改装车。目录分三层：

---

二、launcher.py：进程编排器

launcher.py 干四件事，件件是「脏活」：

1. 起伴生服务（pty_launch）

def pty_launch(service_name, success_std_string="Starting server on"):
    service_path = os.environ.get(f'{service_name.upper()}_PATH')
    service_script = os.environ.get(f'{service_name.upper()}_SCRIPT')
    companion = LaunchCommandWhenAbsent(
        full_argument_list=[service_script], dir=service_path,
        tag="appworld_env_service", use_pty=True)
    companion.launch(launch_wait_time=1800, success_std_string=success_std_string)

本质是 utils/daemon.py 的 LaunchCommandWhenAbsent：用伪终端起子进程，等日志里冒出 success_std_string 才算活。ReMe 等的是 Uvicorn running on，AppWorld 等的是 Starting server on。这就解释了上一章「curl /healthz 验活」——服务起没起，launcher 和 curl 看的是同一根弦。

2. 配置备份与占位符回填

config = _replace_placeholder_in_config(
    config, placeholder="${trainer.experiment_name}", replacement=exp_name)

_replace_placeholder_in_config 是个深度 walk——遍历 dict/list，凡字符串里含 ${trainer.experiment_name} 一律换成实名。此举让「实验名」一处定、处处随，避免多服务各喊各名。

3. 训练子进程

cmd = [sys.executable, '-m', args.target,   # agentevolver.main_ppo
       '--config-path', os.path.abspath(exe_exp_base),
       '--config-name', os.path.basename(exe_yaml_path)]

注意 args.target 默认 agentevolver.main_ppo——壳到此交棒。

4. 善后

---

三、main_ppo.py：训练主循环真核

@hydra.main(config_path="../config", config_name="script_config") 是入口。run_ppo 里先 ray.init，注入 runtime_env（TOKENIZERS_PARALLELISM、VLLM_USE_V1=1、WANDB_* 等），再断言 max_prompt+max_response ≤ max_model_len，而后把活甩给 Ray actor：

runner = TaskRunner.remote()
ray.get(runner.run.remote(config))

TaskRunner.run 才是重头戏，按序：

1. 落盘模型：copy_to_local 把 HF checkpoint 拉到本地；hf_tokenizer / hf_processor 初始化。 2. 选 worker 类：fsdp 策略下，从 module/exp_manager/het_fsdp_worker.py 取 HETAsyncActorRolloutRefWorker（异步 rollout）或 HETActorRolloutRefWorker。这「HET」前缀即第 3 章异策略损失的载体。 3. 资源与角色映射：ResourcePoolManager + role_worker_mapping（ActorRollout / Critic / RewardModel / RefPolicy）。 4. 奖励函数：load_reward_manager 同时给训练集与验证集各造一个。 5. 双 TaskManager：

train_task_manager = TaskManager(..., mixture_strategy=UnifiedMixtureStrategy(...), ...)
val_task_manager   = TaskManager(..., mixture_strategy=OriginalOnlyStrategy(), ...)

训练集用混合策略（原任务+合成），验证集只用 OriginalOnlyStrategy——保证评测不被合成数据污染。 6. 交棒训练器：AgentEvolverRayPPOTrainer(...).init_workers() 后 .fit()。

另有两个小机关值得记：kl_penalty 可按环境变量 DEBUG_ARG=kl_control monkeypatch（前端 token 加权 KL）；get_custom_reward_fn 用 importlib 动态载入用户自定义奖励函数。

---

四、config/agentevolver.yaml：总配置图

一份 yaml 收尽六段。要点：

• algorithm.adv_estimator: grpo——默认纯 GRPO（关 ADCA 时即上一章「最小化训练」）。
• data.max_response_length: 21580、actor_rollout_ref.rollout.max_model_len: 25580——长轨迹是大头。
• rollout.n: 8、multi_turn.max_steps: 30、context_template: linear——多轮、每组 8 条、线性上下文。
• env_service.env_url: http://127.0.0.1:8080——指向上一章起的 AppWorld 服务。
• task_manager / exp_manager / attribution_driven_credit_assignment 三段，正是第 2/3/4 章的配置落点。

---

五、自顶向下入口链（一张图记牢）

launcher.py
  └─ pty_launch          → 起 appworld / reme / logview 伴生服务
  └─ subprocess          → python -m agentevolver.main_ppo
        └─ run_ppo       → ray.init
              └─ TaskRunner.run
                    ├─ copy_to_local + tokenizer
                    ├─ HET*Worker (fsdp)
                    ├─ load_reward_manager
                    ├─ TaskManager ×2 (train/val)
                    └─ AgentEvolverRayPPOTrainer.fit()
                          ├─ env_client.step → env_service(/step)
                          ├─ AgentFlow.execute（第 5 章）
                          ├─ adv_processor（第 4 章，开则覆盖 advantages）
                          └─ exp_manager（第 3 章，开则注入/剥离经验）

一句话收口：launcher 管「伴生服务与配置」，main_ppo 管「Ray 编排与 worker 选型」，ae_ray_trainer 管「fit 主循环」；三大机制作为可插拔开关，在第 4/3 章的钩子里改写优势与上下文。下一章，看任务从哪来——Self-Questioning 之 task_manager。

AgentEvolver 源码解读 · 第 2 章：任务自生成（Self-Questioning / Task Manager）

> 上章把骨架立起。此章看「米」从何来——AgentEvolver 三大机制之首 Self-Questioning：自动生成训练任务，免却人工造数据集之苦。核心在 agentevolver/module/task_manager/。

---

一、Why：任务稀缺是 RL Agent 的头号瓶颈

RL 训练公式化作「对环境的轨迹调优」，数据质量直接定模型上限。真人环境里，任务难造、质量难控。Task Manager 用一套动态工作流破局：探环境 → 抽任务 → 滤质量 → 配比例，全程可独立跑、也可嵌进训练。

官方三阶段（见 docs/guidelines/task_manager.md）：环境刻画（Profile）→ 任务派生（Derivation）→ 任务策展（Curation）。代码里对应 TaskManager + strategies/ + filters/ + data_mixture/。

---

二、Environment Profile：给环境一个「概念骨架」

API 只说「能调什么」，不说「高层概念为何」。Profile 用三元补全：

---

三、TaskManager：四件武器

TaskManager.__init__ 把依赖焊死：

self._exploration_strategy = get_exploration_strategy(exploration_strategy, ...)
self._mixture_strategy     = mixture_strategy
self._realtime_filters     = [NaiveTaskPostFilter()]
self._post_filter          = [LlmFilter(env_service_url, llm_client, ...)]
self._exploration_strategy._inject_deps(
    self._old_retrival, self._llm_client,
    DashScopeClient(model_name='qwen3-235b-a22b-instruct-2507', max_tokens=8192),
    env_profile=env_profile)

注意两处「藏刀」：

• 注入的依赖里塞了个 qwen3-235b 专做 summarize——探索用轻模型，总结用强模型，省钱。
• 两层过滤器：实时 NaiveTaskPostFilter（边探边滤）+ 后置 LlmFilter（LLM 判可行性），外加 old_retrival（NaiveTaskObjectiveRetrieval，按 task_id 存已生成目标，避免重复）。

四、任务派生：探索 + 总结

generate_task 是主入口，伪代码即真相：

task_q = list(tasks) * self._n          # 每个种子任务滚 n 次
parallel_num = min(self._num_exploration_threads, len(tasks))
with ThreadPoolExecutor(...) as pool:
    for batch in batches(task_q, parallel_num):
        futures = [pool.submit(self._exlore_and_summarize, task, ...)
                   for task in batch]
        res.extend(sum(f.result() for f in futures))
        res = reduce(lambda x,f: f.filter(x), self._realtime_filters, res)
        self._old_retrival.reset(); add_objectives(res)
# 落盘检查点 → 后置过滤 → 打乱

_exlore_and_summarize 拆两步：_step_explore（策略去环境里逛）→ _step_summarize（把轨迹抽成任务）。批大小取 min(线程数, 任务数)，正是注释所言「避免同时探索同一任务产生重复查询」。检查点 .generate_task.checkpoint.json 带 md5 摘要，断了能续。

RandomWalk 策略

• explore：起 EnvWorker + ModifiedAgentFlow（enable_context_generator=False），max_steps=max_explore_step，在环境里随机踩点，吐一条 Trajectory。
• summarize：把轨迹里真实 query 用 [MASKED] 遮掉（steps[1]、steps[2]），再交给强模型 get_task_summarize_prompt → parse_tasks_from_response，产出一串 TaskObjective，且强制 task.evaluator='synthetic'。

trajectory.steps[1]['content'] = '[MASKED]'
trajectory.steps[2]['content'] = '[MASKED]'
tasks = parse_tasks_from_response(task, llm_output)

遮真 query 是为防「总结」时把答案泄露回任务本身——合成任务须是开放问句，不是复述标准答案。

---

五、任务策展：滤与混

---

六、内置合成奖励（兜底）

无用户自定义奖励时，synthetic_grader: llm 兜底。奖励三件套：相关性（轨迹对不对题）、成败（任务成否）、效率（步数是否合理）。官方明言：兜底能用，但领域奖励函数才能拉满性能。配置里 original_grader: env / synthetic_grader: llm 即分工。

---

七、接驳点：第 1 章那两个 TaskManager

回到 main_ppo.py：

train_task_manager = TaskManager(..., mixture_strategy=UnifiedMixtureStrategy(
    use_original=config.task_manager.mixture.use_original_tasks,      # basic.yaml=True
    synthetic_ratio=config.task_manager.mixture.synthetic_data_ratio, # basic.yaml=0.0
    shuffle=True, seed=42))
val_task_manager   = TaskManager(..., mixture_strategy=OriginalOnlyStrategy())

最小化训练为何只吃内置数据？ 因为 synthetic_data_ratio=0.0 且 use_original_tasks=True——UnifiedMixtureStrategy 只吐原任务，合成流水线被 n=0 熄火（n=0 时 task_q 空）。这正接上《深度研究》里「basic.yaml 只跑内置数据集」的结论，源头在此 TaskManager 与混合策略。

---

八、小结

Self-Questioning 的代码骨架：Profile 给概念 → RandomWalk 探环境 → 强模型总结遮真 query 抽任务 → 双层过滤器去芜 → 混合策略配比进 RL 集。本质是把「人工造题」自动化成一趟可断点续跑、可独立调试的流水线。NaiveTaskObjectiveRetrieval 做去重记忆，qwen3-235b 专司总结，n=0 一键关停合成——处处是工程上的省钱与防错心机。

下一章，看经验如何被复用——Self-Navigating 之 exp_manager 与 ReMe。

AgentEvolver 源码解读 · 第 3 章：经验导航（Self-Navigating / ExpManager + ReMe）

> 三大机制之二 Self-Navigating：把败仗变避坑的肌肉记忆。人靠经验少走弯路，agent 也该如此。代码核心在 agentevolver/module/exp_manager/，外加一个外部服务 ReMe（Reflective Memory Engine）。

---

一、Why：从瞎撞到借力

传统 RL 靠 trial-and-error，轨迹冗余、收敛慢。Self-Navigating 让 agent 把过往轨迹提炼成「经验」，rollout 时注入、训练时按需剥离——把探索从「无引导乱撞」扭成「有知识的自进化」。

官方分两类：ExperienceManager（调度分配）+ ExperienceWorker（上下文注入/剥离）。另有一座 EMClient 桥接 ReMe 服务（默认 http://127.0.0.1:8001）。

---

二、ExperienceManager：两级动态分配

1. 任务级分配 allocate_train_mode

mode_to_ratio = {"allkeep": 1.0, "alldiscard": 0.0, "hybrid": self.train_sample_keepratio}
keep_ratio = mode_to_ratio.get(self.train_sample_mode, self.train_sample_keepratio)
keep_count = int(len(tasks) * keep_ratio)
exp_modes = ['keep'] * keep_count + ['discard'] * (len(tasks) - keep_count)
random.shuffle(exp_modes)

2. Rollout 级分配 allocate_add_exp

rollout_n = self.rollout_config.n   # 或 val_kwargs.n
add_exp_choices = {
    "woexp": [False] * rollout_n,                       # 纯探索
    "mixed": sorted([i < round(rollout_n*self.rollout_ratio)
                     for i in range(rollout_n)], key=lambda _: random.random()),
    "all":   [True] * rollout_n,                         # 全引导
}[exp_mode]

---

三、ExperienceWorker：注入与剥离

1. rollout 注入 manage_rollout_context

trajectory = Trajectory(data_id=..., steps=init_messages, query=traj_exp_config.query)
history_experience = self.em_client.call_context_generator(
    trajectory=trajectory, retrieve_top_k=reme_config.retrieve_top_k, workspace_id=...)
formatted_experience = self.experience_template.format(history_experience)
new_content = formatted_experience + trajectory.steps[-1]["content"]
trajectory.steps[-1]["content"] = new_content

2. 训练剥离 manage_training_context

if metadata_config.get("task_train_mode", "discard") == "discard":
    pattern = re.escape(self.experience_template).replace(r'\{\}', '(.*?)')
    match = re.search(pattern, message, re.DOTALL)
    if match:
        experience = match.group(1)
        cleaned_message = re.sub(pattern, '', message, flags=re.DOTALL)

3. 接驳点

---

四、EMClient ↔ ReMe：异步总结

ExperienceManager 持 EMClient(base_url=...)，两套投递：

def summarize_in_batch(self, trajectories):
    grouped = groupby(sorted(trajectories, key=lambda t: t.task_id), ...)
    futures = [self.thread_pool.submit(self.em_client.call_summarizer,
                trajectories=b, workspace_id=...) for b in batches]
    # as_completed 收结果

def _should_submit_summary(self, global_steps):
    return (self.reme_config.enable_summarizer and self.reme_config.updated_freq
            and global_steps % self.reme_config.updated_freq == 0)

ThreadPoolExecutor 异步把原始轨迹蒸馏成经验片段；updated_freq 把关——0 即关更新，非 0 每 updated_freq 步灌一次。经验池随训练并行增长。

---

五、异策略损失：on/off 一念之间

最硬核在 module/exp_manager/het_core_algos.py 的 het_compute_token_on_off_policy_loss（官方 docs/guidelines/exp_manager.md 直引）：

negative_approx_kl = log_prob - old_log_prob
ratio = torch.exp(negative_approx_kl)
# 1️⃣ on-policy（exp_mask=0）
on_pg_losses, _, _ = compute_pg_losses(cliprange_low, cliprange_high)
on_pg_loss = masked_mean(on_pg_losses, (1.0 - exp_mask) * response_mask)
# 3️⃣ off-policy（exp_mask=1）
off_pg_losses, _, _ = compute_pg_losses(off_cliprange_low, off_cliprange_high)
off_pg_loss = masked_mean(off_pg_losses, exp_mask * response_mask)
# 4️⃣ 合并
pg_losses = off_pg_losses * exp_mask + on_pg_losses * (1.0 - exp_mask)
pg_loss  = agg_loss(loss_mat=pg_losses, loss_mask=response_mask, loss_agg_mode=loss_agg_mode)

exp_mask 是定海神针：把「新鲜 rollout」与「经验回放」干净分开。off 分支用更紧的 off_cliprange_high（默认 0.6）当信任域，防止旧经验把更新带飞。loss_agg_mode 默认 token-mean。这就把「复用过去」与「不稳定学习」解耦——借力而不翻车。

---

六、六模式配置矩阵

经验池支持六种用法，矩阵记牢（updated_freq / init_exp_only / init_exp_before_training / enable_summarizer / enable_context_generator / workspace_id）：

---

七、小结

Self-Navigating 的代码骨架：Manager 做两级分配（任务级 keep/discard、rollout 级 woexp/mixed/all）→ Worker 在 rollout 注入经验、训练时正则剥离 → EMClient 异步把轨迹喂给 ReMe → 异策略损失用 exp_mask 把 on/off 策略分开、off 分支加信任域。整条线回答一个问题：如何让 agent 借经验之力，又不让旧经验劫持新学习。

下一章，看最锋利的一环——Self-Attributing 之 ADCA-GRPO，如何把「步级信用」算到 token。

AgentEvolver 源码解读 · 第 4 章：步骤归因（Self-Attributing / ADCA-GRPO）

> 三大机制之三 Self-Attributing，也是最锋利的一环。代码全在 agentevolver/module/adv_processor/：semantic_attribution.py（标注）+ adca_grpo.py（融合优势）。

---

一、Why：长轨迹的信用分配难题

GRPO 只拿「终局成败」当组间Baseline——一条 30 步轨迹，成则步步有功、败则步步背锅，分不清哪步真妙、哪步瞎搞。Self-Attributing 借 LLM 的因果判断，给每步贴 GOOD/BAD，把学习信号从「结果」拆成「过程质量 + 结果有效性」，做步级精细归因。论文报：约 +10% 性能、−40% 训练步数。

---

二、入口：compute_prm_grpo_advantages

adca_grpo.py 的入口把原始 batch 变成 token 级优势：

def compute_prm_grpo_advantages(batch, step_flags, hyper=None, scheme="decouple"):
    step_ids = _ensure_tensor(batch.batch["step_ids"], ..., dtype=torch.long)  # (B,L_resp), -1=非响应
    # 对齐实际 response 长度（补 -1 或截断）
    group_ids = _ensure_tensor(batch.batch["group_ids"], ..., dtype=torch.long).view(-1)
    token_level_rewards = _ensure_tensor(batch.batch["token_level_rewards"], ...)
    orm_scores = token_level_rewards.sum(dim=1)          # 轨迹级 ORM 分
    step_rewards, extra_metrics = _build_decouple(orm_scores, step_flags, step_ids, group_ids, hyper)
    step_adv = suffix_sum_on_steps(step_rewards)         # 步级 suffix sum
    advantages = broadcast_step_adv_to_tokens(step_adv, step_ids)  # 广播到 token
    return {"advantages": advantages, "orm_scores": orm_scores, "metrics": extra_metrics}

step_ids 是关键索引：标记每个 token 属第几步，-1 为非响应 token。它串起「步 → token」的桥。

---

三、语义标注：每样本一次 API

semantic_attribution.py 的 evaluate_step_flags_parallel 是标注主力：

api_client = AsyncOpenAI(api_key=..., base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")
semaphore = asyncio.Semaphore(max_concurrent)
# 每样本建一个 EvaluationTask（不是每步！），一次 API 调用评完全部步
for task in all_tasks:
    coroutines.append(_evaluate_single_sample_api(api_client, model_name, task, semaphore, ...))
batch_results = await asyncio.gather(*coroutines, return_exceptions=True)

效率要点：一个样本一次调用（不是每步一次），max_concurrent 控并发，整体省下「步数/样本数」倍的 API。

省钱的 skip

if skip_type == "skip_small_adv":
    if abs(advantage) < 1e-8:           # 优势≈0 不值得标
        should_skip = True
elif skip_type == "skip_all_neg":
    if orm_reward <= THRESHOLD:         # ORM 非正不标
        should_skip = True

解析与容错

def parse_batch_evaluation_result(response, num_steps):
    numbered = {int(m.group(1)): m.group(2).upper()=="GOOD"
                for m in re.finditer(r"Step\s+(\d+)\s+Judgment:\s*(GOOD|BAD)", response, re.I)}
    if len(numbered) == num_steps: return [numbered[i] for i in range(num_steps)]
    ...

---

四、decouple 融合：过程与结果各自归一

_build_decouple 是算法心脏，四步：

# 1. 原始 PRM 奖励：GOOD=+fix_base, BAD=-fix_base（fix_base 默认 0.2）
prm_rewards_raw = [[hyper.fix_base if f else -hyper.fix_base for f in flags] ...]

# 2. PRM 步级 z-score（组内）
prm_rewards_std = _group_zscore_on_steps(prm_rewards_raw, group_ids, hyper)

# 3. ORM 组级 z-score
orm_scores_std[i] = (orm - mean) / (std + 1e-12)

# 4. 融合
for j, prm_reward in enumerate(prm_std):
    if orm_distribution == "last_step":
        combined = alpha*prm_reward + orm_std if j == K-1 else alpha*prm_reward
    elif orm_distribution == "all_steps":
        combined = alpha*prm_reward + orm_std

decouple 的精髓：PRM（过程质量）与 ORM（结果）先各自独立 z-score 归一，再按 alpha 融合。归一是防一个信号淹没另一个——过程与结果各守各的尺度，互不抢戏。equal_trajectory_weight=True（GRPO）时每组轨迹等权；False 则扁平成 GSPO 大池（噪声环境更稳）。length_normalization 开启则乘 1/sqrt(K) 平衡长短轨迹。prm_steps 让标注只在前 N epoch 跑——后期模型已学会，关掉 API 省钱。

产出的 decouple_stats 一堆监控：outcome_share_last_mean（末步 ORM 占比）、alpha_effective（实际 α 比）、sum_step_reward_sign_agree（合成奖励与 ORM 符号一致率）——可当作「归因健康度仪表盘」。

---

五、步级 → token 级：两道桥

def suffix_sum_on_steps(step_rewards):
    t = torch.tensor(r)
    s = torch.flip(torch.cumsum(torch.flip(t,0),0),0)   # 未来奖励后缀和
    return s

def broadcast_step_adv_to_tokens(step_adv, step_ids):
    out = torch.zeros((B, L), device=device)
    for i in range(B):
        valid = step_ids[i] >= 0
        out[i, valid] = adv_i[step_ids[i][valid]]      # 按 step_ids 映射
    return out

suffix_sum_on_steps 给每步算「从这一步往后还能拿多少未来奖励」——这恰是优势函数的直觉定义。broadcast_step_adv_to_tokens 借 step_ids 把步优势铺到它名下每个 token；-1（非响应）填 0，不污染损失。

---

六、接驳点：优势如何改写训练

回到第 1 章 ae_ray_trainer 的 fit：当 attribution_driven_credit_assignment.enable=true，训练器在算出 token_level_rewards 后调 evaluate_step_flags_parallel 拿 step_flags，再 compute_prm_grpo_advantages 覆盖 batch.batch["advantages"]。整条 GRPO 链路不变，只换掉优势这一层——这就是「可插拔」的真意。

basic.yaml 里 enable: false，故最小化训练走原版 GRPO；全开时把 alpha 从 0.05~0.2 起调、prm_steps=20 控成本、skip_type=skip_small_adv 省 API，便是《深度研究》里 +10% 的来源。

---

七、小结

Self-Attributing 的代码骨架：语义标注（每样本一 API、skip 省钱、正则解析、重试壳）→ decouple 融合（GOOD/BAD 转 ±fix_base → 独立 z-score → α 平衡过程与结果）→ 后缀和 + step_ids 广播到 token → 覆盖 advantages。一句话：用 LLM 当「分步阅卷老师」，把终局成败拆成每步好坏，再各自归一、谨慎融合，让长轨迹里每一脚好棋都算数。

下一章，看 agent 到底怎么一步一步跟环境过招——Agent 执行流与上下文管理。

AgentEvolver 源码解读 · 第 5 章：Agent 执行流与上下文管理

> 前三章讲了「任务从哪来」「经验怎么攒」「奖励怎么拆」。然则 agent 究竟如何一步一步与环境过招？答案在 agentevolver/module/agent_flow/ 与 context_manager/。此章专攻「一回合之内」的微观机制：消息如何拼接、token 如何计、loss 如何遮、长文如何剪。

---

一、Why：多轮交互不是「拼字符串」那么简单

普通单轮 RL 只需把一条 prompt 喂给模型、收回一条回答。AgentEvolver 是多轮——agent 每步都要看环境回信、再写下一步代码，循环至多 max_steps=30 次。其间有三件麻烦事必须解决：

1. 上下文拼接：每步要把「历史全部消息 + 刚刚环境回信」重新拼成模型输入，且不同 CMT（上下文管理器）拼法不同。 2. loss 掩码：只让模型的生成参与梯度，环境的回信、初始指令不当作「学习对象」。 3. 长轨迹治理：30 步累积下来，上下文极易爆窗；需有策略压缩历史而不丢关键信息。

AgentFlow 是「导演」，ContextManager（CMT）是「场记」——导演喊 action，场记负责把每一镜记进带 token 与掩码的长卷。

---

二、导演：AgentFlow.execute 的主循环

agent_flow.py 的 execute 是整段执行流的心脏，九步一目了然：

def execute(self, context_manager, init_messages, env, instance_id, ...):
    add_nothink = self.config...use_qwen3          # qwen3 加 <think:6124c78e> 抑制
    # ① 经验注入：把 ReMe 召回的失败经验拼进 init_messages 末条
    init_messages, traj_exp_config = self.exp_worker.manage_rollout_context(init_messages, traj_exp_config)
    self.cmt.metadata["task_train_exp_mode"] = traj_exp_config.train_mode
    self.cmt.save_init_input(init_messages, add_nothink)

    for act_step in range(self.max_steps):
        # ② 准备下一步模型输入（各 CMT 拼法不同）
        step_input = self.cmt.prepare_next_llm_context()
        # ③ token 溢出检查
        if not self.cmt.check_context_token_num_safe(step_input):
            self.cmt.is_terminated = False ; break      # 没跑完，但先撤
        # ④ 调模型
        llm_output = self.llm_chat_fn(step_input, request_id=request_id)
        # ⑤ 落盘模型输出（author=llm）
        self.cmt.save_llm_output(llm_output, input_msg_ref=step_input)
        # ⑥ 与世界交互：把刚生成的内容交给环境执行
        env_output = env.step(instance_id, {"content": self.cmt.prepare_world_interaction(),
                                            "role": "assistant"})
        if env_output["state"]["role"] == "tool":
            env_output["state"] = convert_tool_to_user_message(env_output["state"], tokenizer, format="qwen")
        # ⑦ 落盘环境输出（author=env，截断到 max_env_len）
        state = env_output["state"] ; state.pop('tool_calls', None)
        self.cmt.save_env_output(state, input_msg_ref=step_input, add_nothink=add_nothink)
        # ⑧ 终止判定
        self.cmt.is_terminated = env_output["is_terminated"]
        if self.cmt.is_terminated: break

    # ⑨ 算分：优先用 RewardCalculator，否则直接 env.evaluate
    if self._reward_calculator is not None:
        grader_res = self._reward_calculator.calculate_reward(self.cmt, env, instance_id)
        score = grader_res["score"] ; reason = grader_res["reason"]
    else:
        score = env.evaluate(instance_id, params={"sparse": self.sparse})
        reason = "Outcome 1 = success, 0 = failure."
    self.cmt.reward = Reward(outcome=score, success_rate=1.0 if score>=1 else 0.0,
                              madness=self.cmt.compute_madness(), description=reason)
    self.cmt.remove_last_context()                     # 末条若非 llm，剔除，保证以模型结尾
    self.cmt.generate_log(task_id=task_id)             # 彩色日志
    return self.cmt

两个关键拐弯：

• 经验注入在循环之前（第①步）：这是第 3 章 ReMe 的落地点——manage_rollout_context 已把召回的经验塞进 init_messages 末尾，故 agent 从第一步起就「带着前车之鉴」。
• 终止的两种语义：is_terminated=True 表示任务正常完成（complete_task 被调用）；若因 token 溢出跳出，则 is_terminated=False——这条轨迹算「没跑完」，训练时仍能部分利用（取决于 remove_last_context 是否清掉不完整的末步）。

BaseAgentFlow（base_agent_flow.py）只是抽象基类，守住三把尺：max_steps、max_model_len、max_env_len，具体 execute 由子类实现。

---

三、场记之基：ExtendedMessage 与 loss 掩码

一切上下文操作的原子是 ExtendedMessage（cmt_base.py）：

class ExtendedMessage:
    author          # "llm" / "env" / "initialization" / "memory" / "llm(do_not_train)"
    role            # "assistant" / "user" / "system"
    content         # 原始文本
    token_arr       # 该消息自身的增量 token（非整段！）
    clip            # 是否截断
    uuid            # 全局唯一，供 SelfContextClip 反查定位

need_training 决定谁进梯度：

NEED_TRAIN_AUTHORS   = ["llm"]
NON_TRAIN_AUTHORS    = ["env", "initialization", "user", "memory", "llm(do_not_train)"]
return self.author in NEED_TRAIN_AUTHORS

get_loss_mask 是掩码的精密手术：

def get_loss_mask(self, blackout_token_combo):
    if self.need_training:
        mask = [1] * len(self.token_arr)
        # 1. 首次出现 `<|im_start|>assistant\n` 之后的 token 全黑（不让模型学「扮演自己开场」的模板）
        mask = blackout_specific_token_ids_first_encounter(mask, self.token_arr, blackout_token_combo)
        # 2. EOS 之后的所有 token 全黑（生成结束后的尾巴不学）
        mask = blackout_everything_after_eos_but_keep_eos(mask, self.token_arr, self.eos_token_id)
        return mask
    else:
        return [0] * len(self.token_arr)   # 环境/指令/记忆：一律不学

一句话：只有 author=llm 的消息参与损失，且其中「开场模板 token」与「EOS 之后」被抠掉。这正是多轮 RL 里「只训模型决策、不训环境回信」的标准做法。

clip=True 的消息（典型是环境回信）在构造时即按 clip_token_limit=max_env_len（默认 8192）截断，长回信只留前段、尾加 ... truncate ...，避免单步环境输出撑爆上下文。

---

四、五式 CMT：线性 / 思考 / 幽灵提示 / 记忆 / 自剪

ContextManagerBase 定义抽象接口（save_init_input / prepare_next_llm_context / check_context_token_num_safe / prepare_world_interaction / save_llm_output / save_env_output / group_tokenize）。五套实现各管一类训练范式：

1. Linear_CMT（线性，最朴素）

• prepare_next_llm_context 直接返回 full_context 全量（'future' 模式：用 content_for_future）。
• save_env_output 把环境回信记作 author=env, role=user，并 clip。
• tokenize_steps 是 tokenizer 总装：遍历所有消息拼 input_ids/attention_mask/loss_mask，按首个 need_training 消息切分 prompt / response，再算 position_ids（用 verl 的 compute_position_id_with_mask）。
• group_tokenize 只产 1 个 Sample（整条轨迹当一条样本）。
• compute_madness 查 llm_output_mistakes：若某步触发重复惩罚（repetition_penalty_reward_scalar），且低于阈值，则 madness=-1.0，最终 reward.outcome 被 reward_patch 压成 −1——对「胡言乱语」直接判负。

2. LinearThinkCMT（先思考，后行动）

# 历史里的 llm 消息：剥掉 <think>...</think>，且改 author="llm(do_not_train)"（历史思考不当作学习对象）
new_content = re.sub(r'<think>.*?</think>', '', ext_msg.content, flags=re.DOTALL)
author_override = "llm(do_not_train)"
# 末条（最新环境回信）：追加 think_hint，逼模型「先想后动」
if is_last:
    content += self.think_hint

think_hint 随 force_think 与 env_feedin_preference 而变：box 模式要 \{...} 收尾，code 模式要 ___CODE_BLOCK_5___python

save_llm_output 时，把本步抽成一条 memory 消息

class RewardCalculator(ABC): def __init__(self, task: Task): self._task = task @abstractmethod def calculate_reward(self, trajectory, env, instance_id) -> GraderResult: ... `

AgentFlow 有 RewardCalculator 则走它算分（可做稀疏/稠密、可带 reason 文本），否则退回 env.evaluate。评分结果打包成 Reward(outcome, success_rate, madness, description)——注意 madness 来自 CMT 的重复惩罚检测，与 outcome 共同决定最终奖励。description 会进 reward_scores 随样本落盘，便于复盘「这条为啥正负」。

---

六、小结

第 5 章的骨架：AgentFlow.execute 是九步主循环（经验注入→拼上下文→查 token→调模型→落盘→交环境→落盘→判终止→算分）；ExtendedMessage 是带 token 与掩码的原子，借 need_training + get_loss_mask 实现「只训模型决策、不训环境回信」；五式 CMT 各司其职——Linear 朴素全量、Think 剥历史思考逼当下想、Memory 用 markdown 四段积累摘要、SelfContextClip 唤外部模型自动归档防爆窗、phantom_hint` 为预留位。一句话：上下文管理器，是把「多轮对话」编译成「带掩码的 token 序列」的编译器；导演管流程，场记管记法。

下一章，看这套执行流对外依赖什么——环境服务与基建（env_service / AppworldEnv / schema）。

AgentEvolver 源码解读 · 第 6 章：环境服务与基建

> 前五章皆在训练进程之「内」。然 agent 每步 env.step 打出去的，是一台独立的环境服务。此章看「墙外」：env_service/ 如何用 FastAPI + Ray 把 AppWorld 这类环境封装成可远程调用的沙箱，以及贯穿全程的数据模型（schema / trajectory）。读完便知：何以最小化训练要先 python env_service/env_service.py 起服务，再用 curl 探 /healthz。

---

一、Why：为何要把环境拆成独立服务

训练主进程要绷紧 GPU 跑 rollout；环境（尤其 AppWorld，起一个 world 实例很重）若塞进同一进程，既抢资源又难并发。AgentEvolver 的解法：环境服务独立成 HTTP 服务，Ray actor 作远程沙箱，多实例并行。主进程里的 EnvClient 只发 REST 请求，不关心环境怎么跑。隔离带来三利：

1. 并发：一个服务可同时持数十个 instance_id，对应一个 batch 的多条轨迹。 2. 容错：EnvClient 有重试与 fallback，环境偶崩不拖垮 RL 主循环。 3. 可换：换环境只需实现 BaseEnv 并 @Registry.register，主进程零改动。

---

二、服务端：EnvService + FastAPI + Ray

env_service.py 是服务的全部。启动方式：python env_service/env_service.py --env appworld --port 8080。

2.1 进程内单例与 Ray

class EnvService:
    def __init__(self):
        if not ray.is_initialized():
            ray.init(address='local')          # 本地 Ray，远程环境跑在 actor 上
        self.env_actors = {}                   # instance_id -> Ray actor
        self.remote_env = {}                    # env_type -> 动态生成的 RemoteEnv 类
        self.last_access_time = {}
        self.max_idle_time = 3600               # 闲置 1 小时自动回收

    async def cleanup_inactive_instances(self):
        # 后台 loop 每隔 cleanup_interval=300s 扫一遍，超时实例 release

2.2 动态注册：import_and_register_env

def import_and_register_env(env_name, env_file=None):
    env_file = env_file or f"{env_name}_env"
    env_module_path = os.path.join(SERVER_DIR, "env_service/environments", env_name, f"{env_file}.py")
    spec = importlib.util.spec_from_file_location(f"{env_name}_env", env_module_path)
    module = importlib.util.module_from_spec(spec) ; sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    envir_class = getattr(module, f"{env_name.capitalize()}Env")   # 如 AppworldEnv
    Registry.register(env_name)(envir_class)
    return envir_class

2.3 远程沙箱：@ray.remote RemoteEnv

@ray.remote
class RemoteEnv:
    def __init__(self, task_id, instance_id, params):
        module = importlib.import_module(f"env_service.environments.{env_type}.{env_type}_env")
        self.env = getattr(module, f"{env_type.capitalize()}Env")(task_id, instance_id, params)
    def get_init_state(self, params): return self.env.get_init_state(params)
    def step(self, action, params):     return self.env.step(action, params)
    def evaluate(self, messages, params): return self.env.evaluate(messages, params)
    def get_info(self, messages, params):  return self.env.get_info(messages, params)
    def close(self): return self.env.close()

2.4 六条 REST 路由

---

三、客户端：EnvClient 与重试兜底

env_client.py 是训练侧入口，纯 requests + 自定义 retry_call：

class EnvClient:
    def __init__(self, base_url="http://localhost:8000"):
        self.timeout = 150.0 + random.uniform(50, 200)   # 抖动超时，避免集体卡死

    def _make_request(self, endpoint, env_type, task_id, instance_id, messages, params):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        data = {"env_type":env_type,"task_id":task_id,"instance_id":instance_id,
                "messages":messages or {},"params":params or {}}
        response = requests.post(url, json=data, timeout=self.timeout)
        response.raise_for_status() ; return response.json()

    def step(self, instance_id, action, params, max_retry=3):
        fallback = {"state":[{"role":"assistant","content":"Step failed (timeout or exception),please retry"}],
                    "reward":0,"is_terminated":False,"info":{...}}
        def call(): return self._make_request("step", instance_id=instance_id, messages=action, params=params)["data"]
        return retry_call(call, max_retry=max_retry, fail_return=fallback, ...)

设计要点：

• 每个方法都带 fallback 字典——环境崩时返回一个「假装这步失败、不终止」的占位状态，让主循环继续而非整体挂掉。
• retry_call 指数退避（random.uniform(3,10) 秒），失败达上限返回 fail_return，绝不抛到 RL 主进程。
• safe_log 写错误日志用 fsync 强落盘，且包 try/except——日志本身失败也不能影响训练。这正是生产级 RL 基建的谨慎。

---

四、环境基类与 AppWorld 实现

4.1 BaseEnv：抽象的契约

4.2 AppworldEnv：状态化的 Python REPL

@Registry.register("appworld")
class AppworldEnv(BaseEnv):
    def get_init_state(self, params):
        self.world = AppWorld(task_id=self.task_id, experiment_name=self.instance_id)
        self.query = self._get_query(params)
        return {"state":[{"role":"system","content":self.query},
                         {"role":"user","content":self.world.task.instruction}],
                "info":{"instance_id":self.instance_id,"task_id":self.task_id}}

• PROMPT_TEMPLATE / simple_prompt：两段 jinja2 模板，把「你是自主 AI 助理、用 Python REPL 调 AppWorld API」的长示例与关键指令（查看 API 文档、分页遍历、调用 complete_task(answer=...) 等）渲染进 system 提示。simple 模式用精简版。
• step：收 ActionMessage → transition 解析代码并执行 → 查 world.task_completed() → 若终止则 evaluate 算分：

def step(self, action, params=None):
    action_msg = ActionMessage(**action)
    state_message = self.transition(action_msg)
    is_terminated = self.world.task_completed()
    reward = self.evaluate(params) if is_terminated else 0.0
    return {"state":[state_message.simple_dict], "reward":reward,
            "is_terminated":is_terminated, "info":{}}

• transition：核心解析器。若 model 输出含 ``` `python\n(.*?)` `` 代码块，抽出来；或从 tool_calls 取 appworld 工具的 arguments["code"]，调 appworld_excute → self.world.execute(code)（即真实 Python REPL，变量跨步保留）。执行异常则把错误回信塞进 result，逼模型自纠。非 appworld 工具名一律报「tool not found」——环境只认这一道门。
• evaluate：调 AppWorld 官方 evaluate_task，sparse=True 时「零失败即 1、否则 0」；否则返回 passes/(passes+failures) 的稠密分。sparse 正是第 1 章 actor_rollout_ref.rollout.sparse 的落点。
• get_query_list：静态方法，load_task_ids(split) 直接取 AppWorld 官方任务清单（train/val/test）——/get_env_profile 与 TaskManager 都靠它拿任务池。

• task.py：Task（task_id / env_type / open_query / query / ground_truth / evaluator）——ground_truth 与 evaluator 是 TaskManager 第 2 阶段合成的；TaskObjective（task+confidence+reward）用于探索期。
• trajectory.py：Reward（outcome / success_rate / madness / description）——恰是 AgentFlow 算分打包的对象；Trajectory（steps/query/is_terminated/reward/metadata）；Sample（RL 数据集正样本：messages + 全套 input_ids/prompt_ids/response_ids + 各 loss_mask + position_ids + reward_scores）。truncate_output_ids 在 group_tokenize 后裁超长 response（prompt 超长则直接 raise，逼人调 max_prompt_length）。

convert_tool_to_user_message（utils/utils.py）在 agent_flow 里把 role="tool" 的回信转成 qwen 格式 user 消息，确保多轮上下文里环境反馈以「用户」角色呈现——与 AppWorld 模板里 USER: Output: ... 的对话结构严丝合缝。

---

七、小结

第 6 章的骨架：环境服务 = EnvService（Ray 本地 + env_actors 字典 + 闲置回收）跑 FastAPI，六路由经 import_and_register_env 动态挂 BaseEnv 子类；每个实例是一个 @ray.remote RemoteEnv actor，天然隔离、可并发。EnvClient 用 retry_call + fallback 把环境故障挡在 RL 主循环之外。AppworldEnv 是默认环境——jinja2 模板注入「自主助理」人设，world.execute 当状态化 Python REPL，evaluate_task 出稀疏/稠密分。两份 schema 各管一侧：环境侧 ActionMessage/StateMessage、训练侧 Reward/Sample/Trajectory`。一句话：环境服务是 agent 的「外部世界」，用 HTTP 隔、用 Ray 隔、用 Registry 解耦；训练进程只管发请求、收回信，世界的重活在墙外 actor 里悄悄跑完。

至此六章源码解读毕。自顶向下：工程骨架（1）→ 任务自生成（2）→ 经验导航（3）→ 步骤归因（4）→ 执行流与上下文（5）→ 环境服务与基建（6）。源码之下，AgentEvolver 之「自进化」全貌已现。

非常棒~