Skip to content
Open
Show file tree
Hide file tree
Changes from all 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
79 changes: 47 additions & 32 deletions responses_api_agents/hermes_agent/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
NeMoGymResponseCreateParamsNonStreaming,
NeMoGymResponseFunctionToolCall,
NeMoGymResponseInputTokensDetails,
NeMoGymResponseOutputMessage,
NeMoGymResponseOutputMessageForTraining,
NeMoGymResponseOutputText,
NeMoGymResponseOutputTokensDetails,
Expand All @@ -60,19 +61,38 @@ def _trajectory_to_output_items(messages, n_input):
if isinstance(content, list):
content = "".join(c.get("text", "") if isinstance(c, dict) else getattr(c, "text", "") for c in content)
if role == "assistant":
output_items.append(
NeMoGymResponseOutputMessageForTraining(
id=f"msg-{len(output_items)}",
content=[NeMoGymResponseOutputText(type="output_text", text=content, annotations=[])],
role="assistant",
status="completed",
type="message",
prompt_token_ids=item.get("prompt_token_ids") or [],
generation_token_ids=item.get("generation_token_ids") or [],
generation_log_probs=item.get("generation_log_probs") or [],
routed_experts=item.get("routed_experts"),
text_content = [NeMoGymResponseOutputText(type="output_text", text=content, annotations=[])]
if item.get("generation_token_ids"):
output_items.append(
NeMoGymResponseOutputMessageForTraining(
id=f"msg-{len(output_items)}",
content=text_content,
role="assistant",
status="completed",
type="message",
prompt_token_ids=item.get("prompt_token_ids") or [],
generation_token_ids=item.get("generation_token_ids") or [],
generation_log_probs=item.get("generation_log_probs") or [],
routed_experts=item.get("routed_experts"),
)
)
else:
# Not everything hermes appends with role="assistant" came from the policy: the
# max-iterations summary and the repeated-error apology are both synthesised
# locally and carry no token ids. Emitting those as ...ForTraining hands the
# trainer an assistant turn with an empty token sequence, which breaks the
# contiguity of the prompt/generation token chain across turns. Use the
# non-training message type instead — /verify still sees the text, so the rollout
# keeps its reward, but the turn is not presented as policy output.
output_items.append(
NeMoGymResponseOutputMessage(
id=f"msg-{len(output_items)}",
content=text_content,
role="assistant",
status="completed",
type="message",
)
)
)
for tc in item.get("tool_calls") or []:
fn = tc.get("function") if isinstance(tc, dict) else None
if not fn:
Expand Down Expand Up @@ -324,37 +344,32 @@ def _patched_build_api_kwargs(api_messages):
output_items = _trajectory_to_output_items(messages, n_input)

has_assistant_message = any(
getattr(item, "type", None) == "message" and getattr(item, "role", None) == "assistant"
getattr(item, "type", None) == "message"
and getattr(item, "role", None) == "assistant"
and getattr(item, "generation_token_ids", None)
for item in output_items
)
if not has_assistant_message:
LOG.warning(
"Hermes agent ended without an assistant message. Padding empty assistant message. This should not happen often, investigate: error=%r",
"Hermes agent ended without a trainable assistant message. Emitting a non-trainable assistant message so the rollout can still be verified. This should not happen often, investigate: error=%r",
result.get("error"),
)
last_valid = next(
(
m
for m in reversed(messages)
if isinstance(m, dict) and m.get("role") == "assistant" and m.get("generation_token_ids")
),
None,
)
pti = last_valid["prompt_token_ids"] if last_valid else [0]
gti = last_valid["generation_token_ids"] if last_valid else [0]
glp = (last_valid.get("generation_log_probs") if last_valid else None) or [0.0]
routed_experts = last_valid.get("routed_experts") if last_valid else None
# Deliberately NOT ...ForTraining, and deliberately no fabricated token ids. The
# previous implementation substituted either literal [0] placeholders or the token ids
# of the last assistant message anywhere in `messages` — which includes the input
# history, i.e. tokens the policy did not generate for this rollout. Both handed the
# trainer fabricated samples that are indistinguishable from real ones downstream.
# A rollout with no policy output has nothing to train on; say so, and let the trainer
# decide what to do with it.
output_items.append(
NeMoGymResponseOutputMessageForTraining(
NeMoGymResponseOutputMessage(
id=f"msg_{uuid4().hex}",
content=[NeMoGymResponseOutputText(text=result.get("error") or "", annotations=[])],
content=[
NeMoGymResponseOutputText(type="output_text", text=result.get("error") or "", annotations=[])
],
role="assistant",
status="completed",
type="message",
prompt_token_ids=pti,
generation_token_ids=gti,
generation_log_probs=glp,
routed_experts=routed_experts,
)
)

Expand Down
106 changes: 104 additions & 2 deletions responses_api_agents/hermes_agent/tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@
from nemo_gym.openai_utils import (
NeMoGymEasyInputMessage,
NeMoGymFunctionCallOutput,
NeMoGymResponseCreateParamsNonStreaming,
NeMoGymResponseFunctionToolCall,
NeMoGymResponseOutputMessage,
NeMoGymResponseOutputMessageForTraining,
)
from nemo_gym.server_utils import ServerClient
Expand Down Expand Up @@ -200,7 +202,7 @@ def test_empty(self) -> None:
def test_drops_input_prefix(self) -> None:
msgs = [
{"role": "user", "content": "q"},
{"role": "assistant", "content": "a"},
{"role": "assistant", "content": "a", "generation_token_ids": [3, 4]},
]
out = _trajectory_to_output_items(msgs, 1)
assert len(out) == 1
Expand Down Expand Up @@ -235,6 +237,7 @@ def test_assistant_with_tool_call_and_tool_result(self) -> None:
{
"role": "assistant",
"content": "",
"generation_token_ids": [5, 6],
"tool_calls": [{"id": "c1", "function": {"name": "terminal", "arguments": '{"cmd":"ls"}'}}],
},
{"role": "tool", "tool_call_id": "c1", "content": "file.txt\n"},
Expand All @@ -250,10 +253,52 @@ def test_assistant_with_tool_call_and_tool_result(self) -> None:
assert out[2].output == "file.txt\n"

def test_skips_non_dict_items(self) -> None:
msgs = [None, "string", {"role": "assistant", "content": "ok"}]
msgs = [None, "string", {"role": "assistant", "content": "ok", "generation_token_ids": [7]}]
out = _trajectory_to_output_items(msgs, 0)
assert len(out) == 1

def test_assistant_without_tokens_is_not_trainable(self) -> None:
"""Locally-synthesised assistant turns (max-iteration summary, error apology) carry no
token ids. They must not be presented to the trainer as policy output."""
msgs = [{"role": "assistant", "content": "I hit the iteration limit."}]
out = _trajectory_to_output_items(msgs, 0)
assert len(out) == 1
assert isinstance(out[0], NeMoGymResponseOutputMessage)
assert not isinstance(out[0], NeMoGymResponseOutputMessageForTraining)
assert out[0].content[0].text == "I hit the iteration limit."

def test_empty_token_ids_are_not_trainable(self) -> None:
"""An explicitly empty generation_token_ids is as untrainable as an absent one."""
msgs = [{"role": "assistant", "content": "x", "prompt_token_ids": [], "generation_token_ids": []}]
out = _trajectory_to_output_items(msgs, 0)
assert isinstance(out[0], NeMoGymResponseOutputMessage)
assert not isinstance(out[0], NeMoGymResponseOutputMessageForTraining)

def test_assistant_without_tokens_still_emits_tool_calls(self) -> None:
"""Downgrading the message type must not drop the tool calls attached to it."""
msgs = [
{
"role": "assistant",
"content": "",
"tool_calls": [{"id": "c1", "function": {"name": "terminal", "arguments": '{"cmd":"ls"}'}}],
},
{"role": "tool", "tool_call_id": "c1", "content": "file.txt\n"},
]
out = _trajectory_to_output_items(msgs, 0)
assert len(out) == 3
assert isinstance(out[0], NeMoGymResponseOutputMessage)
assert not isinstance(out[0], NeMoGymResponseOutputMessageForTraining)
assert isinstance(out[1], NeMoGymResponseFunctionToolCall)
assert out[1].name == "terminal"
assert isinstance(out[2], NeMoGymFunctionCallOutput)

def test_routed_experts_survive_on_trainable_turns(self) -> None:
"""The trainable branch must keep forwarding routed_experts (added upstream in #1716)."""
msgs = [{"role": "assistant", "content": "a", "generation_token_ids": [2], "routed_experts": [[[1, 2]]]}]
out = _trajectory_to_output_items(msgs, 0)
assert isinstance(out[0], NeMoGymResponseOutputMessageForTraining)
assert out[0].routed_experts == [[[1, 2]]]


class TestRolloutCorrelation:
"""The prefixed self-call path makes responses() build AIAgent with the same model URL prefix."""
Expand Down Expand Up @@ -290,3 +335,60 @@ def run_conversation(self, *args, **kwargs) -> dict:

asyncio.run(agent.responses(request=None, body=NeMoGymResponseCreateParamsNonStreaming(input="hi")))
assert seen["base_url"] == "http://h:1/v1"


class TestResponsesEmitsNoFabricatedTokens:
"""The /responses fallback path must never invent training token ids."""

async def _run(self, monkeypatch, messages, error="boom") -> list:
import asyncio as _asyncio

import run_agent

import nemo_gym.base_responses_api_agent as base_agent

# resolve_model_base_url is a public method on a pydantic model, so it cannot be
# monkeypatched on the instance. Stub out what it reads instead, as TestRolloutCorrelation does.
monkeypatch.setattr(base_agent, "get_first_server_config_dict", lambda _gc, _name: {"host": "h", "port": 1})
server_client = MagicMock(spec=ServerClient)
server_client.global_config_dict = {}
server_client._build_server_base_url = lambda _cfg: "http://h:1"
agent = HermesAgent(config=_config(), server_client=server_client)
monkeypatch.setattr(run_agent, "AIAgent", MagicMock())

async def _fake_to_thread(fn, *args, **kwargs):
return {"messages": messages, "error": error}

monkeypatch.setattr(_asyncio, "to_thread", _fake_to_thread)
body = NeMoGymResponseCreateParamsNonStreaming(input="q", model="m")
resp = await agent.responses(MagicMock(), body)
return list(resp.output)

async def test_untrainable_trajectory_is_not_padded_with_fake_tokens(self, monkeypatch) -> None:
"""Previously this emitted ...ForTraining with generation_token_ids=[0]."""
out = await self._run(monkeypatch, [{"role": "assistant", "content": "synthetic summary"}])
assert not any(isinstance(item, NeMoGymResponseOutputMessageForTraining) for item in out)
assert any(isinstance(item, NeMoGymResponseOutputMessage) for item in out)

async def test_does_not_reuse_token_ids_from_input_history(self, monkeypatch) -> None:
"""Previously the fallback copied token ids off the last assistant message anywhere in
`messages` — including the prompt history, i.e. tokens this rollout never generated."""
messages = [
{"role": "assistant", "content": "from history", "prompt_token_ids": [1, 2], "generation_token_ids": [9]},
{"role": "user", "content": "q"},
{"role": "assistant", "content": "synthetic summary"},
]
out = await self._run(monkeypatch, messages)
assert all(getattr(item, "generation_token_ids", None) != [9] for item in out)
assert not any(isinstance(item, NeMoGymResponseOutputMessageForTraining) for item in out)

async def test_real_generation_is_still_trainable(self, monkeypatch) -> None:
"""A genuine policy turn is unaffected and keeps its token ids."""
messages = [
{"role": "user", "content": "q"},
{"role": "assistant", "content": "a", "prompt_token_ids": [1], "generation_token_ids": [2, 3]},
]
out = await self._run(monkeypatch, messages, error=None)
trainable = [item for item in out if isinstance(item, NeMoGymResponseOutputMessageForTraining)]
assert len(trainable) == 1
assert trainable[0].generation_token_ids == [2, 3]
Loading