这一节只解决一个问题

看到 async def generate(...) 时,不能把“定义函数”“调用函数”“执行函数体”当成同一件事。

你已经熟悉 def、函数参数和 return 就可以开始。本节暂时不需要了解线程、网络请求或 Slime 的训练过程。

运行一个小实验

下面是独立的语言实验,沿用 Slime 测试里的 generate(rollout_id) 和 exercise() 名称;把实际生成逻辑替换为打印和返回编号。它不是 Slime 的完整实现。

import asyncio

async def generate(rollout_id):
    print("run", rollout_id)
    return rollout_id

async def exercise():
    coro = generate(10)
    print("created")
    result = await coro
    print("result", result)

if __name__ == "__main__":
    asyncio.run(exercise())

下载并运行本节 example.py。只使用 Python 标准库;推荐 Python 3.12,最低 Python 3.11。

运行结果:

created
run 10
result 10

按发生顺序看:

  1. async def generate 定义了一个协程函数。
  2. generate(10) 创建协程对象,此时还没打印 run 10。
  3. await coro 开始执行这个协程,并等待它返回。
  4. asyncio.run(exercise()) 建立并运行本次实验的事件循环,最后收尾。

这里没有同时执行两个任务。先把一次调用读懂,再进入并发。

回到 Slime

原测试用 generate(rollout_id) 创建两份工作,再交给 gather。现在只需要认出函数定义与调用,下一节再读 await,第 3 节再展开 gather。

展开 Slime 源码:generate / exercise

generate / exercise
tests/test_rollout_sample_hooks.py · L61–L70
快照 2f2318653f6f794dddd321eff7c9d4b7b174643f· 查看固定版本源码

    async def generate(rollout_id):
        with rollout_context(rollout_id):
            await asyncio.sleep(0)
            sample = await apply_rollout_sample_hooks(args, Sample(index=rollout_id))
            return sample.metadata["sync_hook"]

    async def exercise():
        assert await asyncio.gather(generate(10), generate(11)) == [("seen", 10), ("seen", 11)]

    asyncio.run(exercise())

先自己回答

为什么 created 出现在 run 10 前面?把 await coro 删除后,generate 的函数体还会执行吗?

查看解释

创建协程对象本身不执行函数体。删除 await 且不把对象交给其他调度入口时,它不会执行,并可能产生“协程未被等待”的运行时警告。

语义核对:Python:协程与等待对象。

本节对应的全部源码位置

下面把同类用法在 Slime 中的每个扫描位置接回本节讲解;点击文件名可查看完整源码快照。重复出现的 API 共用解释,具体输入与清理由所在函数决定。

展开本节 393 条源码记录
记录 / 讲解源码位置原代码
occ-000001examples/coding_agent_rl/generate.py:20
<module>
import asyncio
occ-000019examples/coding_agent_rl/generate.py:182
<module>
async def generate(args, base_sample: Sample, sampling_params: dict[str, Any], evaluation: bool = False):
occ-000034examples/coding_agent_rl/swe.py:27
<module>
import asyncio
occ-000071examples/coding_agent_rl/swe.py:177
<module>
async def prepare_workspace(sb: Sandbox, workdir: str, md: dict) -> None:
occ-000081examples/coding_agent_rl/swe.py:201
<module>
async def apply_before_repo_set_cmd(sb: Sandbox, workdir: str, swepro: dict) -> None:
occ-000086examples/coding_agent_rl/swe.py:214
<module>
async def apply_pre_commands(sb: Sandbox, workdir: str, pre: list[str] | str) -> None:
occ-000090examples/coding_agent_rl/swe.py:230
<module>
async def git_diff(sb: Sandbox, workdir: str) -> str:
occ-000092examples/coding_agent_rl/swe.py:239
<module>
async def run_evaluation(md: dict, *, diff_text: str, timeout_sec: int) -> EvalResult:
occ-000096examples/coding_agent_rl/swe.py:253
<module>
async def _grade_scaleswe(md: dict, diff_text: str, timeout_sec: int) -> EvalResult:
occ-000111examples/coding_agent_rl/swe.py:290
<module>
async def _setup_swepro_assets(ev: Sandbox, swepro: dict) -> None:
occ-000116examples/coding_agent_rl/swe.py:299
<module>
async def _apply_diff(ev: Sandbox, workdir: str, diff_text: str) -> bool:
occ-000120examples/coding_agent_rl/swe.py:316
<module>
async def _run_swepro(ev: Sandbox, workdir: str, swepro: dict, timeout: int) -> float:
occ-000130examples/coding_agent_rl/swe.py:341
<module>
async def _run_eval_cmd(ev: Sandbox, workdir: str, cmd: str, timeout: int) -> float:
occ-000132examples/coding_agent_rl/swe.py:346
<module>
async def _run_f2p_script(ev: Sandbox, workdir: str, script: str, timeout: int) -> float:
occ-000135examples/coding_agent_rl/swe.py:364
<module>
async def _apply_model_patch(ev: Sandbox, workdir: str) -> bool:
occ-000147examples/coding_agent_rl/swe.py:440
<module>
async def _grade_swebench(md: dict, diff_text: str, timeout_sec: int) -> EvalResult:
occ-000160examples/geo3k_vlm_multi_turn/rollout.py:192
<module>
async def _run_inference_step(url: str, tokens: list[int], sampling_params: dict, image_data, tokenizer):
occ-000162examples/geo3k_vlm_multi_turn/rollout.py:315
<module>
async def generate(args: Any, sample: Sample, sampling_params) -> Sample:
occ-000166examples/multi_agent/agent_system.py:1
<module>
import asyncio
occ-000168examples/multi_agent/agent_system.py:14
<module>
async def generate_response(args, prompt, key):
occ-000170examples/multi_agent/agent_system.py:95
Agent
async def run(self, args, prompt, max_retries: int = 1, key: str = None) -> str:
occ-000173examples/multi_agent/agent_system.py:114
SolverAgent
async def generate_initial_solution(self, args, problem_statement) -> str:
occ-000175examples/multi_agent/agent_system.py:126
RewriterAgent
async def rewrite(self, args, problem_statement, previous_solutions: list[str]) -> str:
occ-000177examples/multi_agent/agent_system.py:147
SelectorAgent
async def select(self, args, problem_statement, candidate_solutions: list[str]) -> str:
occ-000179examples/multi_agent/agent_system.py:176
<module>
async def rewrite_worker(args, previous_solutions, problem_statement, worker_id):
occ-000181examples/multi_agent/agent_system.py:182
<module>
async def solver_worker(args, problem_statement, worker_id):
occ-000183examples/multi_agent/agent_system.py:198
<module>
async def run_agent_system(args, sample):
occ-000191examples/multi_agent/rollout_with_multi_agents.py:16
<module>
async def generate_with_multi_agents(args, sample: Sample, sampling_params, evaluation=False) -> list[Sample]:
occ-000196examples/retool/generate_with_retool.py:183
<module>
async def execute_predictions(prediction: str) -> str:
occ-000199examples/retool/generate_with_retool.py:215
<module>
async def generate(args, sample: Sample, sampling_params) -> Sample:
occ-000203examples/retool/generate_with_retool.py:407
<module>
async def reward_func(args, sample, **kwargs):
occ-000204examples/retool/tool_sandbox.py:10
<module>
import asyncio
occ-000207examples/retool/tool_sandbox.py:193
PythonSandbox
async def execute_code(self, code: str) -> str:
occ-000216examples/retool/tool_sandbox.py:337
ToolRegistry
async def execute_tool(self, tool_name: str, arguments: dict[str, Any]) -> str:
occ-000219examples/retool/tool_sandbox.py:348
ToolRegistry
async def _execute_python(self, arguments: dict[str, Any]) -> str:
occ-000222examples/search-r1/generate_with_search.py:4
<module>
import asyncio
occ-000225examples/search-r1/generate_with_search.py:59
<module>
async def search(query: str) -> str:
occ-000228examples/search-r1/generate_with_search.py:124
<module>
async def execute_predictions(prediction: str) -> str:
occ-000231examples/search-r1/generate_with_search.py:145
<module>
async def generate(args, sample: Sample, sampling_params) -> Sample:
occ-000235examples/search-r1/generate_with_search.py:277
<module>
async def reward_func(args, sample, **kwargs):
occ-000236examples/search-r1/google_search_server.py:1
<module>
import asyncio
occ-000238examples/search-r1/google_search_server.py:42
<module>
async def fetch(session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore) -> str:
occ-000246examples/search-r1/google_search_server.py:63
<module>
async def fetch_all(urls: list[str], limit: int = 8) -> list[str]:
occ-000252examples/search-r1/google_search_server.py:94
<module>
async def google_search(api_key, query, top_k=5, timeout: int = 60, proxy=None, snippet_only=False) -> list[dict]:
occ-000264examples/search-r1/local_search_server.py:25
<module>
async def local_search(
occ-000270examples/strands_sglang/generate_with_strands.py:41
<module>
async def generate(args, sample: Sample, sampling_params) -> Sample:
occ-000273examples/strands_sglang/generate_with_strands.py:90
<module>
async def reward_func(args, sample: Sample, **kwargs):
occ-000294examples/tau-bench/generate_with_tau.py:100
<module>
async def generate(args: dict[str, Any], sample: Sample, sampling_params: dict) -> Sample:
occ-000296examples/tau-bench/trainable_agents.py:91
TrainableAgentMixin
async def _call_llm(self, url: str, payload: dict[str, Any]) -> dict[str, Any]:
occ-000298examples/tau-bench/trainable_agents.py:116
TrainableAgentMixin
async def _execute_tool(self, env, action: Action):
occ-000301examples/tau-bench/trainable_agents.py:177
TrainableAgentMixin
async def asolve(
occ-000309slime/agent/adapters/anthropic.py:71
AnthropicAdapter
async def _respond(self, request, body, reply, in_tok, out_tok, stream) -> web.StreamResponse:
occ-000332slime/agent/adapters/anthropic.py:211
<module>
async def _render_stream(request, blocks, stop_reason, in_tok, out_tok) -> web.StreamResponse:
occ-000346slime/agent/adapters/anthropic.py:279
<module>
async def _count_tokens(request: web.Request) -> web.Response:
occ-000355slime/agent/adapters/common.py:15
<module>
import asyncio
occ-000363slime/agent/adapters/common.py:197
BaseAdapter
async def _respond(
occ-000364slime/agent/adapters/common.py:225
BaseAdapter
async def shutdown_session(self, sid: str, *, wait_timeout: float = 5.0) -> None:
occ-000366slime/agent/adapters/common.py:232
BaseAdapter.shutdown_session
async def _drain() -> None:
occ-000377slime/agent/adapters/common.py:245
BaseAdapter
async def finish_session(
occ-000379slime/agent/adapters/common.py:278
BaseAdapter
async def drop_session(self, sid: str, *, wait_timeout: float = 5.0) -> None:
occ-000383slime/agent/adapters/common.py:318
BaseAdapter
async def _run_turn(self, request: web.Request) -> web.StreamResponse:
occ-000400slime/agent/adapters/common.py:442
<module>
async def call_sglang_generate(
occ-000412slime/agent/adapters/common.py:517
<module>
async def _abort_sglang_request(sglang_url: str, rid: str, logger: logging.Logger) -> None:
occ-000414slime/agent/adapters/common.py:523
_abort_sglang_request
async def abort_worker(url: str) -> None:
occ-000424slime/agent/adapters/common.py:557
<module>
async def _health(request: web.Request) -> web.Response:
occ-000427slime/agent/adapters/openai.py:69
OpenAIAdapter
async def _respond(self, request, body, reply, in_tok, out_tok, stream) -> web.StreamResponse:
occ-000451slime/agent/adapters/openai.py:324
<module>
async def _render_stream(
occ-000453slime/agent/adapters/openai.py:351
_render_stream
async def emit(delta: dict[str, Any], finish_reason: str | None = None, usage: dict | None = None) -> None:
occ-000467slime/agent/aiohttp_threaded.py:5
<module>
import asyncio
occ-000471slime/agent/aiohttp_threaded.py:34
AppHandle.stop
async def _shutdown() -> None:
occ-000489slime/agent/harness/claude_code.py:36
ClaudeCodeHarness
async def install_cli(self, sb: Sandbox) -> None:
occ-000491slime/agent/harness/claude_code.py:44
ClaudeCodeHarness
async def write_config(self, sb: Sandbox, ctx: HarnessContext) -> None:
occ-000493slime/agent/harness/claude_code.py:57
ClaudeCodeHarness
async def launch_and_wait(self, sb: Sandbox, ctx: HarnessContext, prompt: str, time_budget_sec: int) -> int:
occ-000495slime/agent/harness/codex.py:48
CodexHarness
async def install_cli(self, sb: Sandbox) -> None:
occ-000497slime/agent/harness/codex.py:56
CodexHarness
async def write_config(self, sb: Sandbox, ctx: HarnessContext) -> None:
occ-000499slime/agent/harness/codex.py:69
CodexHarness
async def launch_and_wait(self, sb: Sandbox, ctx: HarnessContext, prompt: str, time_budget_sec: int) -> int:
occ-000501slime/agent/harness/common.py:18
<module>
import asyncio
occ-000502slime/agent/harness/common.py:64
BaseHarness
async def install_cli(self, sb: Sandbox) -> None:
occ-000503slime/agent/harness/common.py:69
BaseHarness
async def write_config(self, sb: Sandbox, ctx: HarnessContext) -> None:
occ-000504slime/agent/harness/common.py:73
BaseHarness
async def launch_and_wait(self, sb: Sandbox, ctx: HarnessContext, prompt: str, time_budget_sec: int) -> int:
occ-000505slime/agent/harness/common.py:81
BaseHarness
async def run(
occ-000509slime/agent/harness/common.py:107
<module>
async def run_agent(sb: Sandbox, *, workdir: str, start_cmd: str, env: dict[str, str], time_budget_sec: int) -> int:
occ-000512slime/agent/harness/common.py:125
<module>
async def install_npm_cli(
occ-000518slime/agent/harness/common.py:154
<module>
async def install_node22(sb: Sandbox, host_tarball: Path) -> None:
occ-000521slime/agent/sandbox.py:11
<module>
import asyncio
occ-000522slime/agent/sandbox.py:42
Sandbox
async def __aenter__(self) -> Sandbox: ...
occ-000523slime/agent/sandbox.py:44
Sandbox
async def __aexit__(self, exc_type, exc, tb) -> None: ...
occ-000524slime/agent/sandbox.py:46
Sandbox
async def exec(
occ-000525slime/agent/sandbox.py:57
Sandbox
async def write_file(self, sandbox_path: str, content: FileContent, *, user: str = "root") -> None: ...
occ-000526slime/agent/sandbox.py:59
Sandbox
async def read_file(self, sandbox_path: str, *, user: str = "root") -> str: ...
occ-000527slime/agent/sandbox.py:65
<module>
async def _await_done_marker(sb: Sandbox, done_file: str, *, user: str, time_budget_sec: int) -> int:
occ-000531slime/agent/sandbox.py:82
<module>
async def exec_and_wait(
occ-000538slime/agent/sandbox.py:237
E2BSandbox
async def _rpc_retry(self, op_name: str, coro_factory, *, idempotent: bool = True):
occ-000543slime/agent/sandbox.py:273
E2BSandbox
async def _reset_conn_pool(self) -> None:
occ-000545slime/agent/sandbox.py:281
E2BSandbox
async def __aenter__(self) -> E2BSandbox:
occ-000547slime/agent/sandbox.py:302
E2BSandbox
async def __aexit__(self, exc_type, exc, tb) -> None:
occ-000550slime/agent/sandbox.py:309
E2BSandbox
async def exec(
occ-000552slime/agent/sandbox.py:342
E2BSandbox
async def write_file(self, sandbox_path: str, content: FileContent, *, user: str = "root") -> None:
occ-000553slime/agent/sandbox.py:346
E2BSandbox.write_file
async def _do_path():
occ-000557slime/agent/sandbox.py:362
E2BSandbox.write_file
async def _do_bytes():
occ-000563slime/agent/sandbox.py:380
E2BSandbox
async def read_file(self, sandbox_path: str, *, user: str = "root") -> str:
occ-000566slime/agent/sandbox.py:390
<module>
async def ensure_agent_user(sb: Sandbox, workdir: str) -> None:
occ-000568slime/backends/megatron_utils/server/megatron_server.py:1
<module>
import asyncio
occ-000590slime/backends/megatron_utils/server/megatron_server.py:355
<module>
async def _ray_get(ref):
occ-000593slime/backends/megatron_utils/server/megatron_server.py:359
<module>
async def _read_json_payload(request: web.Request) -> dict[str, Any]:
occ-000597slime/backends/megatron_utils/server/megatron_server.py:391
<module>
async def _wait_until_idle(sample_manager, timeout_s: float) -> dict[str, Any]:
occ-000605slime/backends/megatron_utils/server/megatron_server.py:431
_build_http_app
async def detect(_request: web.Request) -> web.Response:
occ-000606slime/backends/megatron_utils/server/megatron_server.py:434
_build_http_app
async def healthz(_request: web.Request) -> web.Response:
occ-000607slime/backends/megatron_utils/server/megatron_server.py:437
_build_http_app
async def info(_request: web.Request) -> web.Response:
occ-000608slime/backends/megatron_utils/server/megatron_server.py:440
_build_http_app
async def get_loads(_request: web.Request) -> web.Response:
occ-000610slime/backends/megatron_utils/server/megatron_server.py:443
_build_http_app
async def update_from_disk(request: web.Request) -> web.Response:
occ-000627slime/backends/megatron_utils/server/megatron_server.py:517
_build_http_app
async def generate(request: web.Request) -> web.Response:
occ-000668slime/backends/sglang_utils/server_control.py:1
<module>
import asyncio
occ-000670slime/backends/sglang_utils/server_control.py:31
<module>
async def get_live_router_workers(router_url: str, *, timeout: float) -> list[dict]:
occ-000674slime/backends/sglang_utils/server_control.py:42
get_live_router_workers
async def check(worker):
occ-000694slime/backends/sglang_utils/server_control.py:152
<module>
async def _abort_server_once(url: str, request_timeout: float) -> None:
occ-000697slime/backends/sglang_utils/server_control.py:159
<module>
async def _get_server_load(url: str, request_timeout: float) -> Any:
occ-000700slime/backends/sglang_utils/server_control.py:166
<module>
async def abort_server_until_idle(
occ-000706slime/backends/sglang_utils/server_control.py:237
<module>
async def abort_servers_until_idle(
occ-000745slime/data/queue_data_source.py:11
<module>
import asyncio
occ-000808slime/data/queue_data_source.py:1113
QueueReader
async def get_samples_async(self, num_samples):
occ-000822slime/data/transport.py:5
<module>
import asyncio
occ-000831slime/data/transport.py:364
<module>
async def publish_rollout_async(value, args, rollout_id, *, group=False, controller=None):
occ-000834slime/data/transport.py:370
<module>
async def run_rollout_io(args, function, *values, **kwargs):
occ-000849slime/data/transport.py:410
<module>
async def unpack_published_payload(value):
occ-000885slime/observability/trace_utils.py:463
trace_function.decorator
async def async_wrapper(*args, **kwargs):
occ-000910slime/rollout/fully_async_distributed.py:3
<module>
import asyncio
occ-000935slime/rollout/fully_async_distributed.py:279
_GenerationActor
async def start(self, rollout_id):
occ-000939slime/rollout/fully_async_distributed.py:283
_GenerationActor
async def _start(self, rollout_id):
occ-000944slime/rollout/fully_async_distributed.py:294
_GenerationActor
async def pause(self):
occ-000948slime/rollout/fully_async_distributed.py:298
_GenerationActor
async def _pause(self):
occ-000949slime/rollout/fully_async_distributed.py:301
_GenerationActor
async def next(self):
occ-000953slime/rollout/fully_async_distributed.py:305
_GenerationActor
async def _next(self):
occ-000967slime/rollout/fully_async_distributed.py:319
_GenerationActor
async def _produce(self):
occ-000968slime/rollout/fully_async_distributed.py:320
_GenerationActor._produce
async def generate_slot(group):
occ-000981slime/rollout/fully_async_distributed.py:344
_GenerationActor
async def _generate_group(self, rollout_id, group=None):
occ-000992slime/rollout/fully_async_distributed.py:397
_GenerationActor
async def _rebuffer(self, group):
occ-000998slime/rollout/fully_async_distributed.py:407
_GenerationActor
async def _flush_rebuffers(self):
occ-001007slime/rollout/fully_async_distributed.py:434
_GenerationActor
async def close(self):
occ-001011slime/rollout/fully_async_distributed.py:438
_GenerationActor
async def _close(self):
occ-001022slime/rollout/fully_async_rollout.py:31
<module>
import asyncio
occ-001041slime/rollout/fully_async_rollout.py:165
AsyncRolloutWorker._thread_main
asyncio.run(self._loop())
occ-001043slime/rollout/fully_async_rollout.py:173
AsyncRolloutWorker
async def _loop(self) -> None:
occ-001062slime/rollout/fully_async_rollout.py:268
<module>
async def _generate_rollout_async(args, rollout_id: int, data_buffer) -> RolloutFnTrainOutput | list[list[Sample]]:
occ-001071slime/rollout/on_policy_distillation.py:8
<module>
async def reward_func(args, sample, **kwargs):
occ-001076slime/rollout/rm_hub/__init__.py:1
<module>
import asyncio
occ-001077slime/rollout/rm_hub/__init__.py:34
<module>
async def remote_rm(args, sample: Sample, max_retries: int = 10):
occ-001082slime/rollout/rm_hub/__init__.py:55
<module>
async def async_rm(args, sample: Sample, **kwargs):
occ-001087slime/rollout/rm_hub/__init__.py:99
<module>
async def batched_async_rm(
occ-001108slime/rollout/sample_hooks.py:37
<module>
async def _apply_to_sample(args, sample: Sample, paths: list[str], **kwargs) -> Sample:
occ-001111slime/rollout/sample_hooks.py:52
<module>
async def apply_rollout_sample_hooks(args, value, **kwargs):
occ-001115slime/rollout/sglang_rollout.py:1
<module>
import asyncio
occ-001120slime/rollout/sglang_rollout.py:158
<module>
async def generate(args: Namespace, sample: Sample, sampling_params: dict[str, Any]) -> Sample:
occ-001123slime/rollout/sglang_rollout.py:258
<module>
async def _run_request_abortable_generate(
occ-001128slime/rollout/sglang_rollout.py:277
<module>
async def _run_server_abort_generate(
occ-001131slime/rollout/sglang_rollout.py:289
<module>
async def generate_and_rm(
occ-001140slime/rollout/sglang_rollout.py:384
<module>
async def generate_and_rm_group(
occ-001145slime/rollout/sglang_rollout.py:426
<module>
async def abort(args: Namespace, rollout_id: int) -> list[list[Sample]]:
occ-001156slime/rollout/sglang_rollout.py:476
<module>
async def generate_rollout_async(
occ-001172slime/rollout/sglang_rollout.py:603
<module>
async def eval_rollout(args: Namespace, rollout_id: int) -> tuple[dict[str, dict[str, list[Any]]], list[list[Sample]]]:
occ-001175slime/rollout/sglang_rollout.py:616
<module>
async def eval_rollout_single_dataset(
occ-001180slime/rollout/sglang_streaming_rollout.py:48
<module>
async def generate_streaming(args: Namespace, sample: Sample, sampling_params: dict[str, Any]) -> Sample:
occ-001196slime/utils/async_utils.py:1
<module>
import asyncio
occ-001198slime/utils/async_utils.py:24
AsyncPacer
async def wait(self):
occ-001247slime/utils/http_utils.py:1
<module>
import asyncio
occ-001248slime/utils/http_utils.py:180
_ShardedHTTPTransport
async def handle_async_request(self, request):
occ-001250slime/utils/http_utils.py:185
_ShardedHTTPTransport
async def aclose(self):
occ-001253slime/utils/http_utils.py:189
<module>
async def _post(client, url, payload, max_retries=60, headers=None, timeout=None, score_centering_top_k=None):
occ-001262slime/utils/http_utils.py:297
_init_ray_distributed_post._HttpPosterActor
async def do_post(self, url, payload, max_retries=60, headers=None, timeout=None, score_centering_top_k=None):
occ-001264slime/utils/http_utils.py:331
<module>
async def post(url, payload, max_retries=60, headers=None, timeout=None, score_centering_top_k=None):
occ-001267slime/utils/http_utils.py:367
<module>
async def get(url, *, timeout=None):
occ-001312slime_plugins/rollout_buffer/buffer.py:113
<module>
async def set_body_size(request: Request, call_next):
occ-001318slime_plugins/rollout_buffer/buffer.py:260
<module>
async def write_to_buffer(request: Request):
occ-001321slime_plugins/rollout_buffer/buffer.py:278
<module>
async def get_rollout_data(request: Request):
occ-001326slime_plugins/rollout_buffer/buffer.py:326
<module>
async def start_rollout(request: Request, background: BackgroundTasks):
occ-001350slime_plugins/rollout_buffer/rollout_buffer_example.py:1
<module>
import asyncio
occ-001352slime_plugins/rollout_buffer/rollout_buffer_example.py:138
<module>
async def get_rollout_data(api_base_url: str) -> tuple[list[dict[str, Any]], dict[str, Any]]:
occ-001358slime_plugins/rollout_buffer/rollout_buffer_example.py:207
<module>
async def generate_rollout_async(args, rollout_id: int, data_buffer, evaluation: bool = False) -> dict[str, Any]:
occ-001401tests/fanout_test_helpers.py:39
<module>
async def compact_generate(args, sample, sampling_params):
occ-001406tests/pipeline_rl_test_helpers.py:21
<module>
async def reward(args, sample, **kwargs):
occ-001426tests/plugin_contracts/test_plugin_generate_contracts.py:3
<module>
import asyncio
occ-001427tests/plugin_contracts/test_plugin_generate_contracts.py:75
<module>
async def custom_generate(args, sample: Sample, sampling_params: dict):
occ-001428tests/plugin_contracts/test_plugin_generate_contracts.py:84
<module>
async def custom_generate_with_evaluation(args, sample: Sample, sampling_params: dict, evaluation: bool = False):
occ-001429tests/plugin_contracts/test_plugin_generate_contracts.py:108
_DummySemaphore
async def __aenter__(self):
occ-001430tests/plugin_contracts/test_plugin_generate_contracts.py:111
_DummySemaphore
async def __aexit__(self, exc_type, exc, tb):
occ-001431tests/plugin_contracts/test_plugin_generate_contracts.py:133
test_generate_and_rm_default_generate_branch_is_stable
async def official_default_generate(args, sample: Sample, sampling_params: dict):
occ-001432tests/plugin_contracts/test_plugin_generate_contracts.py:143
test_generate_and_rm_default_generate_branch_is_stable
result = asyncio.run(
occ-001433tests/plugin_contracts/test_plugin_generate_contracts.py:158
test_generate_and_rm_prefers_per_sample_generate_function
result = asyncio.run(generate_and_rm(args, sample, sampling_params={"temperature": 0.3}, evaluation=True))
occ-001434tests/plugin_contracts/test_plugin_generate_contracts.py:169
test_custom_generate_function_path_supports_user_override
result = asyncio.run(
occ-001435tests/plugin_contracts/test_plugin_generate_contracts.py:183
test_generate_and_rm_group_rm_accepts_list_result_from_custom_generate
async def custom_generate_list(args, sample: Sample, sampling_params: dict):
occ-001436tests/plugin_contracts/test_plugin_generate_contracts.py:190
test_generate_and_rm_group_rm_accepts_list_result_from_custom_generate
result = asyncio.run(
occ-001437tests/plugin_contracts/test_plugin_path_loading_contracts.py:3
<module>
import asyncio
occ-001439tests/plugin_contracts/test_plugin_path_loading_contracts.py:143
<module>
async def reference_single_rm(args, sample: Sample, **kwargs):
occ-001440tests/plugin_contracts/test_plugin_path_loading_contracts.py:147
<module>
async def reference_batched_rm(args, samples: list[Sample], **kwargs):
occ-001441tests/plugin_contracts/test_plugin_path_loading_contracts.py:334
test_custom_rm_default_behavior_is_stable
reward = asyncio.run(async_rm(make_args(rm_type="random"), make_sample(4)))
occ-001442tests/plugin_contracts/test_plugin_path_loading_contracts.py:335
test_custom_rm_default_behavior_is_stable
rewards = asyncio.run(
occ-001443tests/plugin_contracts/test_plugin_path_loading_contracts.py:347
test_custom_rm_path_aligns_with_expected_format
rewards = asyncio.run(
occ-001444tests/plugin_contracts/test_plugin_path_loading_contracts.py:360
test_custom_rm_path_aligns_with_expected_format
reward = asyncio.run(
occ-001445tests/rollout_health_test_helpers.py:15
<module>
async def reward(args, sample, **kwargs):
occ-001446tests/rollout_health_test_helpers.py:36
generate_rollout
async def crash_then_abort(config, current_rollout):
occ-001461tests/test_agent/_fakes.py:157
FakeSGLangServer
async def _handle(self, request: web.Request) -> web.Response:
occ-001464tests/test_agent/_fakes.py:171
FakeSGLangServer
async def __aenter__(self) -> FakeSGLangServer:
occ-001466tests/test_agent/_fakes.py:181
FakeSGLangServer
async def __aexit__(self, *exc) -> None:
occ-001469tests/test_agent/_fakes.py:197
fake_call_sglang_generate
async def _fake(prompt_ids, session, body, *, adapter, session_id=None) -> TurnRecord:
occ-001471tests/test_agent/_fakes.py:260
FakeSandbox
async def __aenter__(self) -> FakeSandbox:
occ-001472tests/test_agent/_fakes.py:263
FakeSandbox
async def __aexit__(self, *exc) -> None:
occ-001473tests/test_agent/_fakes.py:266
FakeSandbox
async def exec(self, cmd, *, user="root", env=None, timeout=120, check=False, idempotent=True):
occ-001475tests/test_agent/_fakes.py:290
FakeSandbox
async def write_file(self, sandbox_path, content, *, user="root") -> None:
occ-001476tests/test_agent/_fakes.py:293
FakeSandbox
async def read_file(self, sandbox_path, *, user="root") -> str:
occ-001478tests/test_agent/test_adapters.py:16
<module>
import asyncio
occ-001480tests/test_agent/test_adapters.py:73
<module>
async def _drain(adapter, sid) -> list[Sample]:
occ-001482tests/test_agent/test_adapters.py:169
test_anthropic_messages_nonstream_records_token_segments
async def run_case():
occ-001490tests/test_agent/test_adapters.py:201
test_anthropic_messages_nonstream_records_token_segments
asyncio.run(run_case())
occ-001491tests/test_agent/test_adapters.py:205
test_openai_chat_completions_nonstream_records_token_segments
async def run_case():
occ-001499tests/test_agent/test_adapters.py:230
test_openai_chat_completions_nonstream_records_token_segments
asyncio.run(run_case())
occ-001500tests/test_agent/test_adapters.py:239
test_anthropic_messages_streams_blocks
async def run_case():
occ-001509tests/test_agent/test_adapters.py:269
test_anthropic_messages_streams_blocks
asyncio.run(run_case())
occ-001510tests/test_agent/test_adapters.py:273
test_openai_chat_completions_streams_chunks_until_done
async def run_case():
occ-001518tests/test_agent/test_adapters.py:297
test_openai_chat_completions_streams_chunks_until_done
asyncio.run(run_case())
occ-001519tests/test_agent/test_adapters.py:313
test_anthropic_multiturn_wire_roundtrip_and_token_capture
async def run_case():
occ-001529tests/test_agent/test_adapters.py:375
test_anthropic_multiturn_wire_roundtrip_and_token_capture
asyncio.run(run_case())
occ-001530tests/test_agent/test_adapters.py:384
test_max_turns_per_sid_returns_429
async def run_case():
occ-001538tests/test_agent/test_adapters.py:402
test_max_turns_per_sid_returns_429
asyncio.run(run_case())
occ-001540tests/test_agent/test_agent_rollout_cpu.py:26
<module>
import asyncio
occ-001542tests/test_agent/test_agent_rollout_cpu.py:56
<module>
if not hasattr(asyncio, "timeout"):
occ-001548tests/test_agent/test_agent_rollout_cpu.py:81
<module>
async def _fast_sleep(_secs): # collapse run_command's 5s poll loop
occ-001550tests/test_agent/test_agent_rollout_cpu.py:108
<module>
async def _anthropic_agent(env: dict, *, n_turns: int = 2) -> int:
occ-001554tests/test_agent/test_agent_rollout_cpu.py:157
<module>
async def _noop_install(self, sb) -> None:
occ-001555tests/test_agent/test_agent_rollout_cpu.py:176
test_generate_produces_trained_samples
async def run_case(monkeypatch):
occ-001558tests/test_agent/test_agent_rollout_cpu.py:194
test_generate_produces_trained_samples
asyncio.run(run_case(mp))
occ-001559tests/test_agent/test_agent_rollout_cpu.py:201
test_generate_aborts_on_empty_trajectory
async def silent_agent(_env) -> int:
occ-001560tests/test_agent/test_agent_rollout_cpu.py:204
test_generate_aborts_on_empty_trajectory
async def run_case(monkeypatch):
occ-001563tests/test_agent/test_agent_rollout_cpu.py:216
test_generate_aborts_on_empty_trajectory
asyncio.run(run_case(mp))
occ-001564tests/test_agent/test_agent_rollout_cpu.py:220
test_generate_aborts_on_missing_image
async def run_case(monkeypatch):
occ-001567tests/test_agent/test_agent_rollout_cpu.py:230
test_generate_aborts_on_missing_image
asyncio.run(run_case(mp))
occ-001568tests/test_agent/test_agent_rollout_cpu.py:238
<module>
async def _codex_agent(env: dict, *, n_turns: int = 2) -> int:
occ-001573tests/test_agent/test_agent_rollout_cpu.py:260
test_codex_openai_rollout_closes_loop
async def run_case():
occ-001577tests/test_agent/test_agent_rollout_cpu.py:292
test_codex_openai_rollout_closes_loop
asyncio.run(run_case())
occ-001578tests/test_agent/test_harness.py:14
<module>
import asyncio
occ-001580tests/test_agent/test_harness.py:38
<module>
async def _fast_sleep(_secs):
occ-001582tests/test_agent/test_harness.py:56
test_run_agent_returns_marker_exit_code
async def run_case():
occ-001583tests/test_agent/test_harness.py:59
test_run_agent_returns_marker_exit_code.run_case
async def fake_agent(env):
occ-001585tests/test_agent/test_harness.py:75
test_run_agent_returns_marker_exit_code
asyncio.run(run_case())
occ-001586tests/test_agent/test_harness.py:79
test_run_agent_propagates_nonzero_exit
async def run_case():
occ-001587tests/test_agent/test_harness.py:80
test_run_agent_propagates_nonzero_exit.run_case
async def fail_agent(_env):
occ-001589tests/test_agent/test_harness.py:88
test_run_agent_propagates_nonzero_exit
asyncio.run(run_case())
occ-001590tests/test_agent/test_harness.py:92
test_run_agent_times_out_when_marker_never_appears
async def run_case():
occ-001592tests/test_agent/test_harness.py:98
test_run_agent_times_out_when_marker_never_appears
asyncio.run(run_case())
occ-001593tests/test_agent/test_harness.py:107
test_claude_code_write_config_preacks_bypass_permissions
async def run_case():
occ-001596tests/test_agent/test_harness.py:115
test_claude_code_write_config_preacks_bypass_permissions
asyncio.run(run_case())
occ-001597tests/test_agent/test_harness.py:119
test_claude_code_launch_command_and_env
async def run_case():
occ-001598tests/test_agent/test_harness.py:122
test_claude_code_launch_command_and_env.run_case
async def agent(env):
occ-001600tests/test_agent/test_harness.py:143
test_claude_code_launch_command_and_env
asyncio.run(run_case())
occ-001601tests/test_agent/test_harness.py:152
test_codex_write_config_base64_roundtrips_inline_base_url
async def run_case():
occ-001603tests/test_agent/test_harness.py:163
test_codex_write_config_base64_roundtrips_inline_base_url
asyncio.run(run_case())
occ-001604tests/test_agent/test_harness.py:167
test_codex_launch_command_and_env
async def run_case():
occ-001605tests/test_agent/test_harness.py:170
test_codex_launch_command_and_env.run_case
async def agent(env):
occ-001607tests/test_agent/test_harness.py:186
test_codex_launch_command_and_env
asyncio.run(run_case())
occ-001608tests/test_agent/test_harness.py:195
test_ensure_agent_user_provisions_user_and_git_safe_dir
async def run_case():
occ-001610tests/test_agent/test_harness.py:203
test_ensure_agent_user_provisions_user_and_git_safe_dir
asyncio.run(run_case())
occ-001611tests/test_agent/test_harness.py:212
test_base_harness_run_calls_steps_in_order
async def run_case():
occ-001612tests/test_agent/test_harness.py:213
test_base_harness_run_calls_steps_in_order.run_case
async def agent(_env):
occ-001615tests/test_agent/test_harness.py:232
test_base_harness_run_calls_steps_in_order
asyncio.run(run_case())
occ-001616tests/test_agent/test_sandbox_exec_and_wait.py:23
<module>
import asyncio
occ-001617tests/test_agent/test_sandbox_exec_and_wait.py:59
ShellFakeSandbox
async def __aenter__(self):
occ-001618tests/test_agent/test_sandbox_exec_and_wait.py:62
ShellFakeSandbox
async def __aexit__(self, *exc):
occ-001619tests/test_agent/test_sandbox_exec_and_wait.py:65
ShellFakeSandbox
async def write_file(self, path, content, *, user="root"):
occ-001620tests/test_agent/test_sandbox_exec_and_wait.py:68
ShellFakeSandbox
async def read_file(self, path, *, user="root"):
occ-001622tests/test_agent/test_sandbox_exec_and_wait.py:71
ShellFakeSandbox
async def exec(self, cmd, *, user="root", env=None, timeout=120, check=False, idempotent=True):
occ-001624tests/test_agent/test_sandbox_exec_and_wait.py:121
fast_marker_polls
async def _instant(_seconds):
occ-001625tests/test_agent/test_sandbox_exec_and_wait.py:134
test_same_tag_reinvocation_actually_reruns
async def _two_attempts():
occ-001628tests/test_agent/test_sandbox_exec_and_wait.py:139
test_same_tag_reinvocation_actually_reruns
(first_code, first_out), (second_code, second_out) = asyncio.run(_two_attempts())
occ-001629tests/test_agent/test_sandbox_exec_and_wait.py:152
test_transport_retry_of_the_spawn_stays_deduped
asyncio.run(exec_and_wait(sb, cmd="true", time_budget_sec=60, tag="job"))
occ-001630tests/test_agent/test_sandbox_exec_and_wait.py:158
test_transport_retry_of_the_spawn_stays_deduped
asyncio.run(sb.exec(spawn_cmds[0]))
occ-001631tests/test_distributed_rollout.py:3
<module>
import asyncio
occ-001660tests/test_distributed_rollout.py:612
test_slow_task_acquire_does_not_block_async_generation
async def exercise():
occ-001669tests/test_distributed_rollout.py:625
test_slow_task_acquire_does_not_block_async_generation
asyncio.run(exercise())
occ-001677tests/test_distributed_rollout.py:683
scheduler_factory.create.make_generator
async def execute(rollout_id, group=None):
occ-001681tests/test_distributed_rollout.py:702
scheduler_factory.create
async def initial_groups(count):
occ-001684tests/test_distributed_rollout.py:731
scheduler_factory
async def cleanup():
occ-001716tests/test_distributed_rollout.py:940
test_distributed_completed_groups_keep_rewards_and_masks
async def get_samples(count):
occ-001717tests/test_distributed_rollout.py:944
test_distributed_completed_groups_keep_rewards_and_masks
async def unexpected_generation(*args):
occ-001718tests/test_distributed_rollout.py:947
test_distributed_completed_groups_keep_rewards_and_masks
async def republish_buffered(*args, **kwargs):
occ-001719tests/test_distributed_rollout.py:963
test_distributed_completed_groups_keep_rewards_and_masks
result, _ = asyncio.run(worker._generate_group(3))
occ-001721tests/test_distributed_rollout.py:982
test_worker_refills_locally_and_slow_group_does_not_block_results
async def generate_group(rollout_id, group=None):
occ-001726tests/test_distributed_rollout.py:1000
test_worker_refills_locally_and_slow_group_does_not_block_results
async def initial_groups(count):
occ-001727tests/test_distributed_rollout.py:1006
test_worker_refills_locally_and_slow_group_does_not_block_results
async def exercise():
occ-001748tests/test_distributed_rollout.py:1039
test_worker_refills_locally_and_slow_group_does_not_block_results
asyncio.run(exercise())
occ-001749tests/test_distributed_rollout.py:1052
<module>
async def _generate_locally(args, sample, sampling_params):
occ-001754tests/test_distributed_rollout.py:1087
<module>
async def _score_fanout(args, samples):
occ-001755tests/test_distributed_rollout.py:1094
<module>
async def _score_group(args, samples):
occ-001773tests/test_distributed_rollout.py:1830
test_cancelled_rebuffer_waiter_does_not_drop_its_continuation
async def exercise():
occ-001780tests/test_distributed_rollout.py:1839
test_cancelled_rebuffer_waiter_does_not_drop_its_continuation
asyncio.run(exercise())
occ-001799tests/test_distributed_rollout.py:2071
test_queue_fetch_finishing_after_pause_preserves_group_without_new_generation
async def run():
occ-001802tests/test_distributed_rollout.py:2078
test_queue_fetch_finishing_after_pause_preserves_group_without_new_generation.run
async def fetch(count):
occ-001806tests/test_distributed_rollout.py:2083
test_queue_fetch_finishing_after_pause_preserves_group_without_new_generation.run
async def rebuffer(value):
occ-001807tests/test_distributed_rollout.py:2086
test_queue_fetch_finishing_after_pause_preserves_group_without_new_generation.run
async def unexpected(*args):
occ-001814tests/test_distributed_rollout.py:2103
test_queue_fetch_finishing_after_pause_preserves_group_without_new_generation
asyncio.run(run())
occ-001823tests/test_fully_async_rollout.py:21
<module>
import asyncio
occ-001826tests/test_fully_async_rollout.py:63
test_custom_data_source_keeps_original_entrypoint
async def generate(received_args, rollout_id, received_source):
occ-001827tests/test_fully_async_rollout.py:70
test_custom_data_source_keeps_original_entrypoint
monkeypatch.setattr(fa, "run", asyncio.run)
occ-001829tests/test_fully_async_rollout.py:118
test_rollout_takes_target_groups_and_leaves_surplus_queued
out = asyncio.run(fa._generate_rollout_async(args, rollout_id=0, data_buffer=None))
occ-001832tests/test_fully_async_rollout.py:166
test_dynamic_filter_drops_groups_and_refills
result = asyncio.run(fa._generate_rollout_async(args, rollout_id=7, data_buffer=None))
occ-001837tests/test_fully_async_rollout.py:234
test_loop_backpressure_stops_topping_up_when_queue_is_full
async def _instant_generate(args, group, sampling_params, evaluation):
occ-001841tests/test_fully_async_rollout.py:266
test_local_worker_lifecycle_is_scoped_to_source
async def generate(args, group, sampling_params, evaluation):
occ-001860tests/test_optional_straw.py:85
_run_default_without_straw
async def generate(*values):
occ-001862tests/test_published_payload_retry.py:3
<module>
import asyncio
occ-001869tests/test_published_payload_retry.py:42
test_result_and_shelve_payloads_recover_at_the_storage_layer
result = asyncio.run(transport.unpack_published_payload(ref))
occ-001871tests/test_published_payload_retry.py:64
test_native_failure_is_not_retried_again_by_slime
asyncio.run(transport.unpack_published_payload(ref))
occ-001935tests/test_rollout_sample_hooks.py:1
<module>
import asyncio
occ-001936tests/test_rollout_sample_hooks.py:16
<module>
async def async_hook(args, sample, *, evaluation=False):
occ-001937tests/test_rollout_sample_hooks.py:34
test_rollout_sample_hooks_preserve_nested_shape_and_filter_kwargs
result = asyncio.run(apply_rollout_sample_hooks(args, samples, evaluation=True, ignored="value"))
occ-001938tests/test_rollout_sample_hooks.py:47
test_rollout_sample_hook_rejects_invalid_return_type
asyncio.run(apply_rollout_sample_hooks(args, Sample(index=0)))
occ-001939tests/test_rollout_sample_hooks.py:55
test_rollout_sample_hooks_are_noop_when_unconfigured
assert asyncio.run(apply_rollout_sample_hooks(args, sample)) is sample
occ-001940tests/test_rollout_sample_hooks.py:61
test_concurrent_groups_keep_their_own_rollout_hook_context
async def generate(rollout_id):
occ-001944tests/test_rollout_sample_hooks.py:67
test_concurrent_groups_keep_their_own_rollout_hook_context
async def exercise():
occ-001947tests/test_rollout_sample_hooks.py:70
test_concurrent_groups_keep_their_own_rollout_hook_context
asyncio.run(exercise())
occ-001951tests/test_rollout_straw_transport.py:329
test_fully_async_stores_groups_while_collecting_the_batch
import asyncio
occ-001952tests/test_rollout_straw_transport.py:354
test_fully_async_stores_groups_while_collecting_the_batch
output = asyncio.run(fa._generate_rollout_async(args, 0, None))
occ-001953tests/test_rollout_straw_transport.py:362
test_synchronous_rollout_stores_during_generation_and_preserves_legacy_hook
import asyncio
occ-001954tests/test_rollout_straw_transport.py:375
test_synchronous_rollout_stores_during_generation_and_preserves_legacy_hook
async def exercise():
occ-001957tests/test_rollout_straw_transport.py:379
test_synchronous_rollout_stores_during_generation_and_preserves_legacy_hook.exercise
async def generate(index):
occ-001962tests/test_rollout_straw_transport.py:395
test_synchronous_rollout_stores_during_generation_and_preserves_legacy_hook.exercise
async def abort(*a):
occ-001965tests/test_rollout_straw_transport.py:411
test_synchronous_rollout_stores_during_generation_and_preserves_legacy_hook
output, aborted = asyncio.run(exercise())
occ-001983tests/test_score_centering_binary.py:1
<module>
import asyncio
occ-001987tests/test_score_centering_binary.py:111
test_http_worker_decodes_off_event_loop.Response
async def aread(self):
occ-001988tests/test_score_centering_binary.py:114
test_http_worker_decodes_off_event_loop.Response
async def aclose(self):
occ-001989tests/test_score_centering_binary.py:118
test_http_worker_decodes_off_event_loop.Client
async def post(self, *args, **kwargs):
occ-001990tests/test_score_centering_binary.py:121
test_http_worker_decodes_off_event_loop
result = asyncio.run(
occ-001991tests/test_score_centering_transport.py:3
<module>
import asyncio
occ-001992tests/test_score_centering_transport.py:85
test_generate_requests_sampler_topk
async def post(url, payload, **kwargs):
occ-001993tests/test_score_centering_transport.py:93
test_generate_requests_sampler_topk
sample = asyncio.run(rollout.generate(a, Sample(prompt="test"), {"max_new_tokens": 8}))
occ-001994tests/test_score_centering_transport.py:127
test_r3_resume_appends_routes_without_replacing_the_persisted_prefix
async def post(url, payload, **kwargs):
occ-001995tests/test_score_centering_transport.py:145
test_r3_resume_appends_routes_without_replacing_the_persisted_prefix
asyncio.run(rollout.generate(a, sample, {"max_new_tokens": 8}))
occ-001996tests/test_score_centering_transport.py:148
test_r3_resume_appends_routes_without_replacing_the_persisted_prefix
actual = asyncio.run(rollout.generate(a, sample, {"max_new_tokens": 8}))
occ-001997tests/test_score_centering_transport.py:159
test_streaming_score_centering_rejected
asyncio.run(generate_streaming(args(), Sample(), {}))
occ-001999tests/test_score_centering_transport.py:204
test_evaluation_preserves_training_score_centering
async def generate(received, sample, params):
occ-002000tests/test_score_centering_transport.py:208
test_evaluation_preserves_training_score_centering
async def hooks(received, sample, **kwargs):
occ-002001tests/test_score_centering_transport.py:215
test_evaluation_preserves_training_score_centering
async def run():
occ-002004tests/test_score_centering_transport.py:219
test_evaluation_preserves_training_score_centering
asyncio.run(run())
occ-002005tests/test_score_centering_transport.py:552
test_generate_requests_complete_top_p_probabilities
async def post(url, payload, **kwargs):
occ-002006tests/test_score_centering_transport.py:561
test_generate_requests_complete_top_p_probabilities
sample = asyncio.run(rollout.generate(a, Sample(prompt="test"), {"max_new_tokens": 8}))
occ-002007tests/test_server_control.py:1
<module>
import asyncio
occ-002008tests/test_server_control.py:19
test_rollout_boundary_prunes_failed_router_workers
async def respond(request):
occ-002009tests/test_server_control.py:37
test_rollout_boundary_prunes_failed_router_workers
workers = asyncio.run(server_control.get_live_router_workers("http://router", timeout=0.1))
occ-002012tests/test_server_control.py:108
test_abort_uses_supported_load_sections_and_waits_for_pd_transfers
async def post(url, payload, **kwargs):
occ-002013tests/test_server_control.py:114
test_abort_uses_supported_load_sections_and_waits_for_pd_transfers
async def get(url, **kwargs):
occ-002014tests/test_server_control.py:139
test_abort_uses_supported_load_sections_and_waits_for_pd_transfers
asyncio.run(server_control.abort_server_until_idle("http://engine", retry_interval=0, timeout=0.1))
occ-002015tests/test_server_control.py:168
test_abort_retries_load_failure_until_idle
async def abort_once(url, request_timeout):
occ-002016tests/test_server_control.py:171
test_abort_retries_load_failure_until_idle
async def get_load(url, request_timeout):
occ-002017tests/test_server_control.py:181
test_abort_retries_load_failure_until_idle
asyncio.run(server_control.abort_server_until_idle("http://engine", retry_interval=0, timeout=1))
occ-002018tests/test_server_control.py:187
test_abort_servers_surfaces_partial_failure
async def abort(url, **kwargs):
occ-002019tests/test_server_control.py:193
test_abort_servers_surfaces_partial_failure
asyncio.run(server_control.abort_servers_until_idle(["http://good", "http://bad"]))
occ-002020tests/test_server_control.py:205
test_http_post_forwards_timeout_and_closes_response.Response
async def aread(self):
occ-002021tests/test_server_control.py:208
test_http_post_forwards_timeout_and_closes_response.Response
async def aclose(self):
occ-002022tests/test_server_control.py:214
test_http_post_forwards_timeout_and_closes_response.Client
async def post(self, url, **kwargs):
occ-002023tests/test_server_control.py:218
test_http_post_forwards_timeout_and_closes_response
assert asyncio.run(http_utils._post(Client(), "http://engine", {}, max_retries=1, timeout=2.5)) == {"ok": True}
occ-002024tests/test_server_control.py:226
test_http_transport_preserves_streams_timeouts_and_connection_budget
async def check():
occ-002025tests/test_server_control.py:230
test_http_transport_preserves_streams_timeouts_and_connection_budget.check
async def serve(reader, writer):
occ-002043tests/test_server_control.py:266
test_http_transport_preserves_streams_timeouts_and_connection_budget.check
async def request(i):
occ-002055tests/test_server_control.py:284
test_http_transport_preserves_streams_timeouts_and_connection_budget
asyncio.run(check())
occ-002056tests/test_straw_fully_async_recovery.py:9
<module>
import asyncio
occ-002062tests/test_straw_fully_async_recovery.py:45
<module>
async def _generate_with_interrupt(args, sample, sampling_params):
occ-002071tests/test_straw_fully_async_recovery.py:86
<module>
async def _reward_once(args, sample):
occ-002091tests/test_straw_r3.py:1
<module>
import asyncio
occ-002093tests/test_straw_r3.py:61
test_straw_publishes_replay_tensors_with_the_group_without_hook_or_spill_directory
result = asyncio.run(apply_rollout_sample_hooks(args, nested, rollout_id=3))
occ-002094tests/test_straw_r3.py:105
test_hooks_leave_ray_evaluation_and_aborted_samples_resident
assert asyncio.run(apply_rollout_sample_hooks(args, sample, evaluation=evaluation)) is sample
occ-002099tests/test_straw_r3.py:159
test_cancelled_straw_publication_drains_before_releasing_capacity
async def run():
occ-002115tests/test_straw_r3.py:179
test_cancelled_straw_publication_drains_before_releasing_capacity
asyncio.run(run())
occ-002117tests/test_streaming_rollout.py:1
<module>
import asyncio
occ-002128tests/test_streaming_rollout.py:258
test_generate_streaming_preserves_metadata_across_stream_intervals
result = asyncio.run(
occ-002129tests/test_streaming_rollout.py:299
test_generate_streaming_handles_encoded_terminal_metadata
result = asyncio.run(
occ-002130tests/test_streaming_rollout.py:320
test_streaming_generator_fails_closed_on_unexpected_eof
asyncio.run(
occ-002133tests/test_streaming_rollout.py:338
test_server_abort_keeps_last_observed_prefix
result = asyncio.run(
occ-002144tests/test_streaming_rollout.py:375
test_stream_cancellation_closes_request_and_keeps_prefix
async def server_generate(args, sample, sampling_params):
occ-002148tests/test_streaming_rollout.py:381
test_stream_cancellation_closes_request_and_keeps_prefix
async def get_workers(_url, *, timeout):
occ-002149tests/test_streaming_rollout.py:384
test_stream_cancellation_closes_request_and_keeps_prefix
async def abort_servers(urls):
occ-002151tests/test_streaming_rollout.py:392
test_stream_cancellation_closes_request_and_keeps_prefix
async def exercise():
occ-002161tests/test_streaming_rollout.py:405
test_stream_cancellation_closes_request_and_keeps_prefix
result, server_result = asyncio.run(exercise())
occ-002170tests/test_streaming_rollout.py:427
test_unrelated_stream_cancellation_propagates
async def exercise():
occ-002177tests/test_streaming_rollout.py:441
test_unrelated_stream_cancellation_propagates
asyncio.run(exercise())
occ-002178tests/test_streaming_rollout.py:466
test_partial_abort_buffers_and_resumes_only_aborted_siblings
async def exercise():
occ-002182tests/test_streaming_rollout.py:489
test_partial_abort_buffers_and_resumes_only_aborted_siblings.exercise
async def resume_generate(args, sample, sampling_params):
occ-002183tests/test_streaming_rollout.py:500
test_partial_abort_buffers_and_resumes_only_aborted_siblings.exercise
async def reward(_args, sample):
occ-002186tests/test_streaming_rollout.py:514
test_partial_abort_buffers_and_resumes_only_aborted_siblings
buffered_groups, resumed_group = asyncio.run(exercise())
occ-002187tests/test_streaming_rollout.py:533
test_generation_pacing_preserves_concurrency_cancellation_and_reuse
async def exercise():
occ-002193tests/test_streaming_rollout.py:550
test_generation_pacing_preserves_concurrency_cancellation_and_reuse.exercise
async def generate(_args, sample, params, evaluation=False):
occ-002213tests/test_streaming_rollout.py:589
test_generation_pacing_preserves_concurrency_cancellation_and_reuse
asyncio.run(exercise())
occ-002214tests/test_streaming_rollout.py:593
test_abort_while_waiting_for_generation_pacer_does_not_start_request
async def exercise():
occ-002217tests/test_streaming_rollout.py:600
test_abort_while_waiting_for_generation_pacer_does_not_start_request.exercise
async def wait_for_release():
occ-002222tests/test_streaming_rollout.py:605
test_abort_while_waiting_for_generation_pacer_does_not_start_request.exercise
async def unexpected_generate(*args, **kwargs):
occ-002228tests/test_streaming_rollout.py:618
test_abort_while_waiting_for_generation_pacer_does_not_start_request
asyncio.run(exercise())
occ-002243tests/training_recovery_test_helpers.py:15
<module>
async def reward(args, sample, **kwargs):

专题首页 · 下一节:02 · await:等待结果,不等于自动并发 · 源码与教程对照 · 全部用法