File size: 9,660 Bytes
a925f29 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | """SciWorld AgentLoop for verl —— 把 AgentGym 的 HTTP 环境接进 verl 的多轮 rollout。
verl 的 multi_turn 是围绕 tool call 设计的; SciWorld 是"环境返回纯文本 observation"
的形态, 所以按 notes 的建议实现一个 AgentLoop 子类。
协议与我们自建评测/ROSE/OPD 完全一致(便于横向对比):
system(通用 Qwen) + user(24 动作 REACT 说明) + assistant("OK. I'll follow…")
+ user(task_description + look around) + [assistant 动作 / user observation] * N
动作解析取 "Action:" 之后首行; 每条轨迹的 reward = 最终 score(0~1)。
数据里每行需要:
item_id : "sciworld_{games下标}"
raw_prompt : 开场三条消息(list[dict])
agent_name : "sciworld"
"""
import asyncio
import logging
import os
from typing import Any
from uuid import uuid4
import aiohttp
from verl.experimental.agent_loop.agent_loop import AgentLoopBase, AgentLoopOutput, register
from verl.utils.profiler import simple_timer
from verl.utils.rollout_trace import rollout_trace_op
from verl.workers.rollout.replica import TokenOutput
logger = logging.getLogger(__file__)
logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN"))
ENV_BASE = os.getenv("SCIWORLD_ENV_BASE", "http://127.0.0.1:36001")
ENV_TIMEOUT = float(os.getenv("SCIWORLD_ENV_TIMEOUT", "300"))
def parse_action(text: str) -> str:
tail = text.split("Action:")[-1] if "Action:" in text else text
for line in tail.strip().splitlines():
line = line.strip()
if line:
return line.strip("`* ")
return "look around"
@register("sciworld")
class SciWorldAgentLoop(AgentLoopBase):
"""多轮 SciWorld: 学生出动作 -> HTTP 环境返回 observation -> 拼回对话, 直到 done 或轮数上限。"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.prompt_length = self.rollout_config.prompt_length
self.response_length = self.rollout_config.response_length
mt = getattr(self.rollout_config, "multi_turn", None)
self.max_turns = int(getattr(mt, "max_user_turns", 0) or 20)
self.max_obs_len = int(getattr(mt, "max_tool_response_length", 0) or 2048)
async def _post(self, session, path, payload, retries=3):
"""环境侧任何接口都可能返回 {"error": ...} 而非期望字段, 这里统一重试。"""
last = None
for attempt in range(retries):
try:
async with session.post(f"{ENV_BASE}/{path}", json=payload,
timeout=aiohttp.ClientTimeout(total=ENV_TIMEOUT)) as r:
js = await r.json()
if isinstance(js, dict) and "error" in js:
last = js["error"]
await asyncio.sleep(1.0 + attempt)
continue
return js
except Exception as e:
last = repr(e)
await asyncio.sleep(1.0 + attempt)
logger.warning("sciworld %s failed after %d retries: %s", path, retries, last)
return {"error": last}
@rollout_trace_op
async def run(self, sampling_params: dict[str, Any], priority: int = 0, **kwargs) -> AgentLoopOutput:
priority = int(priority)
try:
with open("/home/haojinw2/agentloop_trace.log", "a") as _f:
_f.write(f"RUN kwargs={sorted(kwargs.keys())}\n")
except Exception:
pass
messages = list(kwargs["raw_prompt"])
item_id = kwargs.get("item_id", "sciworld_0")
data_idx = int(str(item_id).split("_")[-1])
metrics = {}
self._assert_mm_supported(False)
prompt_ids = await self.ct_build_initial_tokens(messages)
runtime_ids = list(prompt_ids)
min_gs, max_gs = None, None # 记录每轮生成所用的权重版本(异步框架用来算 staleness)
response_mask: list[int] = []
response_logprobs: list[float] = []
score = 0.0
turns = 0
async with aiohttp.ClientSession() as session:
try:
with open("/home/haojinw2/agentloop_trace.log", "a") as _f:
_f.write(f"CREATE item={item_id}\n")
except Exception:
pass
created = await self._post(session, "create", {}, retries=8)
if "id" not in created:
await asyncio.sleep(20) # 等看护重启环境服务
created = await self._post(session, "create", {}, retries=8)
if "id" not in created:
logger.warning("sciworld create failed, skip this trajectory: %s", created)
return AgentLoopOutput(
prompt_ids=prompt_ids, response_ids=[], response_mask=[],
reward_score=0.0, num_turns=0, metrics={"num_preempted": -1},
extra_fields={"sciworld_score": 0.0, "sciworld_turns": 0,
"min_global_steps": 0, "max_global_steps": 0,
"sciworld_env_error": 1})
env_id = created["id"]
try:
obs = await self._post(session, "reset", {"id": env_id, "data_idx": data_idx})
if "observation" not in obs and "task_description" not in obs:
logger.warning("sciworld reset failed: %s", obs)
obs = {"task_description": "", "observation": "", "done": True, "score": 0}
first = (obs.get("task_description", "") + "\n" + obs.get("observation", "")).strip()
convo = list(messages) + [{"role": "user", "content": first}]
merge_result, response_mask, response_logprobs = await self.ct_merge_non_assistant_msg(
messages, convo, runtime_ids, response_mask, response_logprobs)
runtime_ids = merge_result.token_ids
done = bool(obs.get("done", False))
score = float(obs.get("score", 0) or 0) / 100.0
while not done and turns < self.max_turns:
if len(runtime_ids) - len(prompt_ids) >= self.response_length:
break
with simple_timer("generate_sequences", metrics):
rid = (f"det-{priority}"
if getattr(self.rollout_config, "full_determinism", False) else uuid4().hex)
out: TokenOutput = await self.server_manager.generate(
request_id=rid, prompt_ids=runtime_ids,
sampling_params=sampling_params, priority=priority)
gs = (out.extra_fields or {}).get("global_steps")
if gs is not None:
min_gs = gs if min_gs is None else min(min_gs, gs)
max_gs = gs if max_gs is None else max(max_gs, gs)
merge_result, response_mask, response_logprobs = await self.ct_merge_assistant_token(
runtime_ids, out.token_ids, response_mask, response_logprobs,
assistant_logprobs=getattr(out, "log_probs", None))
runtime_ids = merge_result.token_ids
action_text = self.tokenizer.decode(out.token_ids, skip_special_tokens=True)
turns += 1
try:
st = await self._post(session, "step",
{"id": env_id, "action": parse_action(action_text)})
obs_text = str(st.get("observation", ""))[: self.max_obs_len]
done = bool(st.get("done", False))
score = max(score, float(st.get("score", 0) or 0) / 100.0)
except Exception as e: # 环境异常 = 该轨迹终止, 与 AgentGym 一致
logger.warning("sciworld step failed: %s", e)
obs_text, done = f"Error: {e}", True
prev = list(convo)
convo = prev + [{"role": "assistant", "content": action_text},
{"role": "user", "content": obs_text}]
if not done:
merge_result, response_mask, response_logprobs = await self.ct_merge_non_assistant_msg(
prev + [{"role": "assistant", "content": action_text}], convo,
runtime_ids, response_mask, response_logprobs)
runtime_ids = merge_result.token_ids
finally:
try:
await self._post(session, "close", {"id": env_id})
except Exception:
pass
response_ids = runtime_ids[len(prompt_ids):]
response_mask = response_mask[: len(response_ids)]
if len(response_mask) < len(response_ids):
response_mask += [0] * (len(response_ids) - len(response_mask))
metrics.setdefault("num_preempted", -1)
return AgentLoopOutput(
prompt_ids=prompt_ids,
response_ids=response_ids[: self.response_length],
response_mask=response_mask[: self.response_length],
response_logprobs=(response_logprobs[: self.response_length] if response_logprobs else None),
reward_score=score,
num_turns=turns,
metrics=metrics,
extra_fields={"sciworld_score": score, "sciworld_turns": turns,
"min_global_steps": min_gs if min_gs is not None else 0,
"max_global_steps": max_gs if max_gs is not None else 0},
)
|