Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/user-guide/agentic-rollout.md
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,12 @@ sequence, trims model-specific boundary tokens, and builds the training sample.

### Choose the session behavior

Session purpose is fixed at creation. `POST /sessions` accepts `{"evaluation": true}` for evaluation; omitting the body or field defaults to training. The field must be a boolean. The agentic generator passes `GenerateFnInput.evaluation` automatically, and the purpose survives v1 rollback and v2 branching.

Evaluation sessions force `return_sampling_mask`, `return_routed_experts`, and `return_indexer_topk` to `false`, including when a request or model rule asks to enable them. They ignore `routed_experts_start_len`. Sampling defaults and override order stay unchanged, and temperature can vary between turns. TITO rendering, token IDs, logprobs, and template compatibility checks still apply; agentic evaluation still collects samples.

Disabling replay outputs does not disable all internal capture work in a shared SGLang engine with capture enabled globally.

History handling depends on the selected server version:

- **v1 is linear.** Each request must extend the previous messages at the tail.
Expand Down
2 changes: 1 addition & 1 deletion miles/rollout/generate_hub/agentic_tool_call.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ async def generate(input: GenerateFnInput) -> GenerateFnOutput:
)
use_v2 = getattr(input.args, "use_session_server", None) == "v2"
collect_spec_metrics = use_v2 and input.args.sglang_speculative_algorithm is not None
tracer = await OpenAIEndpointTracer.create(input.args)
tracer = await OpenAIEndpointTracer.create(input.args, evaluation=input.evaluation)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of passing evaluation flag, how about explicitly passing "is_replay"? In case, we need to turn it off for non-evaluation requests too. What do you think?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I also thought of this... My idea is the evaluation semantic is easy to be understood by user, and also we might skip more TITO components for evaluation to ensure perf. So I think keep this param name is ok.


custom_agent_function: Callable = load_function(input.args.custom_agent_function_path)
assert (
Expand Down
4 changes: 2 additions & 2 deletions miles/rollout/generate_utils/openai_endpoint_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ def session_server_id(self) -> str:
return self.router_url.removeprefix("http://")

@staticmethod
async def create(args: Namespace):
async def create(args: Namespace, *, evaluation: bool = False):
session_addrs = getattr(args, "session_server_addrs", None)
if not session_addrs:
raise RuntimeError(
Expand All @@ -56,7 +56,7 @@ async def create(args: Namespace):
session_url = f"http://{session_addr}"
instance_ids = getattr(args, "session_server_instance_ids", None) or {}
session_server_instance_id = instance_ids.get(session_addr)
response = await post(f"{session_url}/sessions", {}, action="post")
response = await post(f"{session_url}/sessions", {"evaluation": evaluation}, action="post")
session_id = response["session_id"]
use_v2 = getattr(args, "use_session_server", None) == "v2"
return OpenAIEndpointTracer(
Expand Down
4 changes: 2 additions & 2 deletions miles/rollout/session/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,8 +231,8 @@ async def health(self) -> Response:
body["session_server_instance_id"] = self.instance_id
return Response(content=_render_json(body), status_code=200, media_type=JSON_MEDIA_TYPE)

async def create_session(self) -> Response:
session_id = self.registry.create_session()
async def create_session(self, *, evaluation: bool = False) -> Response:
session_id = self.registry.create_session(evaluation=evaluation)
return Response(content=_render_json({"session_id": session_id}), status_code=200, media_type=JSON_MEDIA_TYPE)

def _session_metadata(self, session_id: str, session) -> dict:
Expand Down
9 changes: 6 additions & 3 deletions miles/rollout/session/linear_trajectory.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ class LinearTrajectory:
generated_checkpoint_message_ends: list[int] = field(default_factory=list)
num_assistant: int = 0
turn_args_history: list[dict[str, Any]] = field(default_factory=list)
evaluation: bool = False

@property
def turn_args(self) -> dict[str, Any]:
Expand Down Expand Up @@ -126,7 +127,9 @@ def prepare_token_ids_and_request_args(
request_messages = client_args.get("messages", [])
checkpoint_index = self._find_rollback_checkpoint(request_messages, matcher)
turn_args = self.turn_args_history[checkpoint_index] if checkpoint_index >= 0 else None
prepared = prepare_chat_request(client_args, tito_tokenizer, config=config, turn_args=turn_args)
prepared = prepare_chat_request(
client_args, tito_tokenizer, config=config, turn_args=turn_args, evaluation=self.evaluation
)
prepared.body["input_ids"] = self._render_token_ids(
request_messages,
checkpoint_index=checkpoint_index,
Expand Down Expand Up @@ -373,9 +376,9 @@ def __init__(
message_matcher if message_matcher is not None else strict_message_matches
)

def create_session(self) -> str:
def create_session(self, *, evaluation: bool = False) -> str:
session_id = uuid.uuid4().hex
self.sessions[session_id] = LinearTrajectory()
self.sessions[session_id] = LinearTrajectory(evaluation=evaluation)
return session_id

def get_session(self, session_id: str) -> LinearTrajectory:
Expand Down
17 changes: 11 additions & 6 deletions miles/rollout/session/request_args.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,24 +53,29 @@ def prepare_chat_request(
*,
config: SessionServerConfig,
turn_args: dict[str, Any] | None,
evaluation: bool = False,
) -> PreparedChatRequest:
"""Resolve an owned request using server rules, model rules, and prior turn args.

``turn_args`` is the continued turn's full request; ``None`` starts a root.
Client input and recorded history remain unchanged.
"""
request_args, client_stream = resolve_request_args_by_config(deepcopy(client_args), config)
request_args, client_stream = resolve_request_args_by_config(deepcopy(client_args), config, evaluation=evaluation)
try:
request_args = tito_tokenizer.resolve_request_args(request_args, turn_args=turn_args)
except ValueError as e:
raise MessageValidationError(str(e)) from e
if evaluation:
# Model rules must not re-enable training replay outputs for evaluation.
request_args.update(return_sampling_mask=False, return_routed_experts=False, return_indexer_topk=False)
request_args.pop("routed_experts_start_len", None)
return PreparedChatRequest(
body=request_args, template_args=extract_template_args(request_args), client_stream=client_stream
)


def resolve_request_args_by_config(
request_args: dict[str, Any], config: SessionServerConfig
request_args: dict[str, Any], config: SessionServerConfig, *, evaluation: bool = False
) -> tuple[dict[str, Any], bool]:
"""Apply server constraints in place and return the same request and stream intent.

Expand All @@ -82,9 +87,9 @@ def resolve_request_args_by_config(
# Must be False so stop-token text is trimmed from assistant content;
# token IDs still come from logprobs below.
request_args["no_stop_trim"] = False
# R3 replay follows the launch flags, on or off.
request_args["return_routed_experts"] = bool(config.use_rollout_routing_replay)
request_args["return_indexer_topk"] = bool(config.use_rollout_indexer_replay)
# Training replay follows the launch flags; eval never requests replay outputs.
request_args["return_routed_experts"] = not evaluation and bool(config.use_rollout_routing_replay)
request_args["return_indexer_topk"] = not evaluation and bool(config.use_rollout_indexer_replay)

# The served adapter is selected by training; SGLang lets a ``base:adapter``
# model parameter beat ``lora_path``, so that spelling is refused too.
Expand All @@ -108,7 +113,7 @@ def resolve_request_args_by_config(
f"input_ids={value!r} is not accepted: TITO token ids are rendered by the session server"
)
request_args.pop("input_ids", None)
if (value := request_args.get("routed_experts_start_len")) is not None:
if not evaluation and (value := request_args.get("routed_experts_start_len")) is not None:
raise MessageValidationError(
f"routed_experts_start_len={value!r} is not accepted: R3 offsets are computed by the session server"
)
Expand Down
10 changes: 8 additions & 2 deletions miles/rollout/session/sessions.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from fastapi import Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from sglang.srt.entrypoints.openai.protocol import ChatCompletionResponse
from sglang.srt.parser.template_detection import detect_inline_system_support
from starlette.responses import Response
Expand Down Expand Up @@ -43,6 +44,7 @@
from miles.rollout.session.core import JSON_MEDIA_TYPE, SessionCore, _render_json
from miles.rollout.session.errors import SessionError
from miles.rollout.session.linear_trajectory import SessionRegistry
from miles.rollout.session.types import CreateSessionRequest
from miles.utils.chat_template_utils import get_tito_tokenizer
from miles.utils.chat_template_utils.message_matcher_hub import (
SessionMessageMatcherError,
Expand Down Expand Up @@ -100,8 +102,12 @@
return Response(content=_render_json(body), status_code=response.status_code, media_type=JSON_MEDIA_TYPE)

@app.post("/sessions")
async def create_session():
return await core.create_session()
async def create_session(request: Request):
try:
params = CreateSessionRequest.model_validate_json(await request.body() or b"{}")
except ValidationError as exc:
return JSONResponse(status_code=400, content={"error": str(exc)})
return await core.create_session(evaluation=params.evaluation)

@app.get("/sessions/{session_id}")
async def get_session(session_id: str):
Expand Down
6 changes: 5 additions & 1 deletion miles/rollout/session/types.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, StrictBool


class CreateSessionRequest(BaseModel):
evaluation: StrictBool = False
Comment thread
claude[bot] marked this conversation as resolved.
Resolved


class SessionRecord(BaseModel):
Expand Down
11 changes: 8 additions & 3 deletions miles/rollout/session/v2/session_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ class SessionStateV2:
lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False, compare=False)
closing: bool = field(default=False, repr=False, compare=False)
tree: SessionTree = field(default_factory=SessionTree)
evaluation: bool = False

def latest(self) -> TrajectoryNode | None:
"""The most recently committed generation (always a leaf), or ``None``
Expand Down Expand Up @@ -99,7 +100,11 @@ def prepare_token_ids_and_request_args(
request_messages = client_args.get("messages", [])
parent = attach_point_for_request(state, request_messages, message_matcher=message_matcher).node
prepared = prepare_chat_request(
client_args, tito_tokenizer, config=config, turn_args=parent.turn_args if parent is not None else None
client_args,
tito_tokenizer,
config=config,
turn_args=parent.turn_args if parent is not None else None,
evaluation=state.evaluation,
)
prepared.body["input_ids"] = _render_token_ids(
parent, request_messages, template_args=prepared.template_args, tito_tokenizer=tito_tokenizer
Expand Down Expand Up @@ -209,9 +214,9 @@ class SessionRegistryV2(SessionRegistry):

sessions: dict[str, SessionStateV2]

def create_session(self) -> str:
def create_session(self, *, evaluation: bool = False) -> str:
session_id = uuid.uuid4().hex
self.sessions[session_id] = SessionStateV2()
self.sessions[session_id] = SessionStateV2(evaluation=evaluation)
return session_id

def compute_mismatch(
Expand Down
16 changes: 10 additions & 6 deletions tests/fast/rollout/generate_hub/test_agentic_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ async def collect_samples(self, input_sample, *, max_seq_len, agent_metadata=Non
return self.reply


def _generate_input(**args_kwargs) -> GenerateFnInput:
def _generate_input(*, evaluation=False, **args_kwargs) -> GenerateFnInput:
args = SimpleNamespace(
**{
"session_server_addrs": ["127.0.0.1:12345"],
Expand All @@ -48,7 +48,7 @@ def _generate_input(**args_kwargs) -> GenerateFnInput:
label="label",
metadata={"source": "test"},
)
return GenerateFnInput(state=state, sample=sample, sampling_params={}, evaluation=False)
return GenerateFnInput(state=state, sample=sample, sampling_params={}, evaluation=evaluation)


async def _fake_agent(**kwargs):
Expand All @@ -65,21 +65,25 @@ def _session_metadata(spec_info=None):


def _patch_agent(monkeypatch, tracer):
async def fake_create(args):
async def fake_create(args, *, evaluation=False):
tracer.evaluation = evaluation
return tracer

monkeypatch.setattr(agentic_tool_call.OpenAIEndpointTracer, "create", fake_create)
monkeypatch.setattr(agentic_tool_call, "load_function", lambda path: _fake_agent)


@pytest.mark.asyncio
async def test_success_returns_list_and_forwards_agent_metadata(monkeypatch):
@pytest.mark.parametrize("evaluation", [False, True])
async def test_success_returns_list_and_forwards_agent_metadata(monkeypatch, evaluation):
sample = Sample(status=Sample.Status.COMPLETED, response="done", response_length=1, tokens=[1])
tracer = _Tracer(SamplesReply(samples=[sample], session_metadata={}, empty_reason=None))
_patch_agent(monkeypatch, tracer)

output = await agentic_tool_call.generate(_generate_input())
generate_input = _generate_input(evaluation=evaluation)
output = await agentic_tool_call.generate(generate_input)

assert tracer.evaluation is evaluation
assert output.samples == [sample]
assert output.samples[0].rollout_id is None
assert tracer.agent_metadata == {"agent_result": "done"}
Expand Down Expand Up @@ -212,7 +216,7 @@ async def test_empty_session_server_addrs_is_rejected(self, monkeypatch, addrs):
"""generate() raises the documented AssertionError when session_server_addrs is absent, null or empty, without creating a tracer."""
created_for: list[object] = []

async def fake_create(args):
async def fake_create(args, *, evaluation=False):
created_for.append(args)
return _Tracer(SamplesReply(samples=[], session_metadata={}, empty_reason="no_records"))

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,13 @@


@pytest.mark.asyncio
async def test_create_reads_session_server_instance_id_from_args(monkeypatch):
@pytest.mark.parametrize("create_kwargs", [{}, {"evaluation": False}, {"evaluation": True}])
async def test_create_reads_session_server_instance_id_from_args(monkeypatch, create_kwargs):
calls: list[tuple[str, str]] = []

async def fake_post(url: str, payload: dict, action: str = "post"):
calls.append((action, url))
assert payload == {"evaluation": create_kwargs.get("evaluation", False)}
assert action == "post"
assert url == "http://127.0.0.1:12345/sessions"
return {"session_id": "session-123"}
Expand All @@ -37,7 +39,7 @@ async def fake_post(url: str, payload: dict, action: str = "post"):
session_server_addrs=["127.0.0.1:12345"],
session_server_instance_ids={"127.0.0.1:12345": "server-instance-123"},
)
tracer = await OpenAIEndpointTracer.create(args)
tracer = await OpenAIEndpointTracer.create(args, **create_kwargs)

assert tracer.base_url == "http://127.0.0.1:12345/sessions/session-123"
assert tracer.session_server_id == "127.0.0.1:12345"
Expand Down
42 changes: 42 additions & 0 deletions tests/fast/rollout/session/test_request_args.py
Original file line number Diff line number Diff line change
Expand Up @@ -210,3 +210,45 @@ def test_filter_turn_args_accepts_an_explicit_drop_list():
turn_args = {"input_ids": [1], "messages": [{"role": "user", "content": "hi"}], "seed": 42}
assert filter_turn_args(turn_args, drop_keys=("input_ids",)) == {"messages": turn_args["messages"], "seed": 42}
assert filter_turn_args(turn_args, drop_keys=()) == turn_args


@pytest.mark.parametrize("sampling", [{}, {"temperature": 0.0}, {"temperature": 1.2, "top_p": 0.8, "top_k": 16}])
def test_eval_keeps_sampling_resolution_and_overrides_model_replay(sampling):
class Model(TITOTokenizer):
def resolve_request_args(self, request_args, *, turn_args):
request_args = super().resolve_request_args(request_args, turn_args=turn_args)
request_args.setdefault("temperature", 0.7)
request_args.setdefault("top_p", 0.9)
request_args.setdefault("top_k", 32)
request_args.update(return_sampling_mask=True, return_routed_experts=True, return_indexer_topk=True)
request_args["routed_experts_start_len"] = 7
return request_args

config = make_session_server_config(use_rollout_routing_replay=True, use_rollout_indexer_replay=True)
client_args = {**sampling, "return_sampling_mask": True, "routed_experts_start_len": 99}
history = {"temperature": 0.3, "chat_template_kwargs": {}}
original, original_history = deepcopy(client_args), deepcopy(history)
prepared = prepare_chat_request(client_args, Model(MagicMock()), config=config, turn_args=history, evaluation=True)
assert {key: prepared.body[key] for key in ("temperature", "top_p", "top_k")} == {
"temperature": 0.7,
"top_p": 0.9,
"top_k": 32,
**sampling,
}
assert all(
prepared.body[key] is False for key in ("return_sampling_mask", "return_routed_experts", "return_indexer_topk")
)
assert "routed_experts_start_len" not in prepared.body
assert client_args == original and history == original_history


@pytest.mark.parametrize("evaluation", [False, True])
def test_template_compatibility_applies_to_train_and_eval(evaluation):
with pytest.raises(MessageValidationError, match="tools changed"):
prepare_chat_request(
{"tools": [{"type": "function", "function": {"name": "new_tool"}}]},
TITOTokenizer(MagicMock()),
config=make_session_server_config(),
turn_args={"chat_template_kwargs": {}},
evaluation=evaluation,
)
Loading
Loading