Skip to content
Merged
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
29 changes: 29 additions & 0 deletions docs/source/reference/cli.md
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,35 @@ openenv skills add --dest /path/to/my-agent/skills/

[[autodoc]] openenv.cli.commands.skills.skills_preview

## `openenv collect`

Collect rollouts from a running environment with a teacher model and write
them as an SFT-ready `results.jsonl`. The teacher can be a hosted provider
(`--provider openai|anthropic`) or any self-hosted OpenAI-compatible server
such as vLLM, TGI or Ollama via `--llm-endpoint`.

```bash
# Scripted teacher, no API key needed
openenv collect openspiel:tic_tac_toe --base-url http://localhost:8001 \
--output-dir ./rollouts -n 10 --provider scripted

# Self-hosted model (vLLM serving on port 8000)
openenv collect reasoning_gym:chain_sum --base-url http://localhost:8001 \
--output-dir ./rollouts -n 50 \
--llm-endpoint http://localhost:8000 --model Qwen/Qwen3-1.7B
```

`--llm-endpoint` takes a full base URL. `/v1` is appended when the URL has no
path, so `http://localhost:8000` and `http://localhost:8000/v1` are equivalent;
a URL with a path (for example a gateway prefix) is used as-is. `--llm-port` is
only needed when the URL does not include a port; it has no default, so
`--llm-endpoint http://localhost` means port 80 (earlier releases assumed 8000).
The resolved endpoint is printed when the run starts. Only `http(s)` URLs are
accepted, and credentials, query strings and fragments in the URL are rejected:
pass the key through `OPENAI_API_KEY` instead.

[[autodoc]] openenv.cli.commands.collect.collect

# API Reference

## Entry point
Expand Down
15 changes: 15 additions & 0 deletions docs/source/tutorials/sft-warmup.md
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,21 @@ hub_repo_arg = shlex.quote(f"{YOUR_HF_USERNAME}/chain-sum-rollouts")
--output-dir ./rollouts
```

To use a self-hosted teacher instead of a hosted provider, point `--llm-endpoint` at any
OpenAI-compatible server (vLLM, TGI, Ollama) and pass the model id it serves:

```bash
openenv collect reasoning_gym:chain_sum \
--base-url https://sergiopaniego-reasoning-gym.hf.space \
--llm-endpoint http://localhost:8000 \
--model Qwen/Qwen3-1.7B \
--num-episodes 300 \
--output-dir ./rollouts
```

The teacher drives the environment through tool calls, so the server must have tool calling
enabled (for vLLM: `--enable-auto-tool-choice --tool-call-parser <parser for your model>`).

The command prints a live progress summary and pushes the collected episodes to the Hub as
`{YOUR_HF_USERNAME}/chain-sum-rollouts`. Pull them back to start filtering:

Expand Down
19 changes: 15 additions & 4 deletions examples/ttt_collect_with_llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
# Local vLLM (OpenAI-compatible) with Qwen:
python examples/ttt_collect_with_llm.py \\
--base-url http://localhost:8000 \\
--llm-endpoint http://localhost --llm-port 8001 \\
--llm-endpoint http://localhost:8001 \\
--model Qwen/Qwen2.5-7B-Instruct \\
--num-episodes 20

Expand Down Expand Up @@ -136,9 +136,20 @@ def parse_args() -> argparse.Namespace:
parser.add_argument(
"--llm-endpoint",
default=None,
help="OpenAI-compatible endpoint (e.g. http://localhost for vLLM).",
help=(
"Base URL of an OpenAI-compatible server (e.g. http://localhost:8001 "
"for vLLM). /v1 is appended when the URL has no path."
),
)
parser.add_argument(
"--llm-port",
type=int,
default=None,
help=(
"Port appended to --llm-endpoint when the URL does not include one. "
"No default: earlier versions assumed 8000."
),
)
parser.add_argument("--llm-port", type=int, default=8000)
parser.add_argument("--temperature", type=float, default=0.2)
parser.add_argument("--max-tokens", type=int, default=200)

Expand All @@ -156,7 +167,7 @@ def main() -> None:

llm_client = build_llm_client(args)
teacher_label = (
f"{args.llm_endpoint}:{args.llm_port}/{args.model}"
f"{llm_client.base_url}/{args.model}"
if args.llm_endpoint
else f"{args.provider}/{args.model}"
)
Expand Down
69 changes: 42 additions & 27 deletions src/openenv/cli/commands/collect.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ def _build_llm_model_step(
model: str,
*,
llm_endpoint: str | None,
llm_port: int,
llm_port: int | None,
temperature: float,
max_tokens: int,
system_prompt: str | None = None,
Expand All @@ -152,15 +152,19 @@ def _build_llm_model_step(
# Self-hosted OpenAI-compatible endpoint (vLLM, TGI, Ollama, ...).
from openenv.core.llm_client import OpenAIClient

client: LLMClient = OpenAIClient(
endpoint=llm_endpoint,
port=llm_port,
model=model,
api_key=os.getenv("OPENAI_API_KEY") or "not-needed",
system_prompt=effective_system_prompt,
temperature=temperature,
max_tokens=max_tokens,
)
try:
client: LLMClient = OpenAIClient(
endpoint=llm_endpoint,
port=llm_port,
model=model,
api_key=os.getenv("OPENAI_API_KEY") or "not-needed",
system_prompt=effective_system_prompt,
temperature=temperature,
max_tokens=max_tokens,
)
except ValueError as exc:
raise typer.BadParameter(str(exc), param_hint="--llm-endpoint") from exc
console.print(f"[cyan]LLM endpoint:[/cyan] {client.base_url}")
else:
client = create_llm_client(
provider=provider,
Expand Down Expand Up @@ -280,13 +284,23 @@ def collect(
str | None,
typer.Option(
"--llm-endpoint",
help="OpenAI-compatible endpoint URL (for self-hosted vLLM/TGI/Ollama).",
help=(
"Base URL of a self-hosted OpenAI-compatible server "
"(vLLM/TGI/Ollama), e.g. http://localhost:8000. /v1 is appended "
"when the URL has no path; a URL with a path is used as-is."
),
),
] = None,
llm_port: Annotated[
int,
typer.Option("--llm-port", help="Port for self-hosted LLM endpoint."),
] = 8000,
int | None,
typer.Option(
"--llm-port",
help=(
"Port appended to --llm-endpoint when the URL does not include one. "
"No default: earlier releases assumed 8000."
),
),
] = None,
temperature: Annotated[
float, typer.Option("--temperature", help="Sampling temperature.")
] = 0.2,
Expand Down Expand Up @@ -364,6 +378,20 @@ def collect(
factory = _build_session_factory(
env, base_url, dataset_config=parsed_dataset_config
)

if uses_llm_teacher:
model_step = _build_llm_model_step(
provider=provider,
model=model, # type: ignore[arg-type] # validated above
llm_endpoint=llm_endpoint,
llm_port=llm_port,
temperature=temperature,
max_tokens=max_tokens,
system_prompt=system_prompt,
)
else:
model_step = _build_scripted_model_step()

serializer = RolloutSerializer(output_dir)
serializer.write_metadata(
{
Expand All @@ -379,19 +407,6 @@ def collect(
}
)

if uses_llm_teacher:
model_step = _build_llm_model_step(
provider=provider,
model=model, # type: ignore[arg-type] # validated above
llm_endpoint=llm_endpoint,
llm_port=llm_port,
temperature=temperature,
max_tokens=max_tokens,
system_prompt=system_prompt,
)
else:
model_step = _build_scripted_model_step()

should_keep = None if keep_losses else (lambda record: record.reward >= 0.0)

collect_runner = CollectRunner(
Expand Down
6 changes: 6 additions & 0 deletions src/openenv/core/harness/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,12 @@ OPENAI_API_KEY=... openenv collect openspiel:tic_tac_toe \
--output-dir /tmp/ttt-sft-v1 \
-n 200 --provider openai --model gpt-5-mini \
--push-to-hub <user>/ttt-sft-v1

# Self-hosted teacher (vLLM/TGI/Ollama) via any OpenAI-compatible base URL
openenv collect openspiel:tic_tac_toe \
--base-url https://<user>-<space>.hf.space \
--output-dir /tmp/ttt-sft-local \
-n 200 --llm-endpoint http://localhost:8000 --model Qwen/Qwen3-1.7B
```

Programmatic use:
Expand Down
110 changes: 95 additions & 15 deletions src/openenv/core/llm_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@
client = OpenAIClient("http://localhost", 8000, model="meta-llama/...")
response = await client.complete("What is 2+2?")

# The endpoint may carry its own port and path prefix:
client = OpenAIClient("http://localhost:8000/v1", port=None, model="meta-llama/...")

# Or use the factory for hosted APIs:
client = create_llm_client("openai", model="gpt-4", api_key="sk-...")
response = await client.complete_with_tools(messages, tools)
Expand All @@ -25,9 +28,79 @@
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import urlsplit, urlunsplit

from openai import AsyncOpenAI

_OPENAI_API_PREFIX = "/v1"
_MIN_PORT, _MAX_PORT = 1, 65535


def _redact_userinfo(endpoint: str) -> str:
"""Drop `user:password@` from a URL so it is safe to echo in an error."""
scheme, sep, rest = endpoint.partition("://")
authority, slash, tail = rest.partition("/")
if not sep or "@" not in authority:
return endpoint
return f"{scheme}{sep}***@{authority.rsplit('@', 1)[-1]}{slash}{tail}"


def _join_endpoint_port(endpoint: str, port: int | None) -> str:
"""Validate an endpoint URL and combine it with an optional port.

Only `http`/`https` URLs with a host are accepted. Credentials, query
strings and fragments are rejected: the endpoint is logged and persisted in
rollout metadata, and the SDKs build request URLs by appending to the path,
which corrupts a query string. `port` is appended only when the URL does
not name one; an explicit port that differs from the one in the URL is
rejected rather than silently overridden. A trailing slash on the path is
dropped. Invalid endpoints raise `ValueError`.
"""
safe = _redact_userinfo(endpoint)
try:
parts = urlsplit(endpoint)
if parts.scheme not in ("http", "https"):
raise ValueError("expected an http:// or https:// URL")
if not parts.hostname:
raise ValueError("missing host")
if parts.username is not None or parts.password is not None:
raise ValueError("credentials in the URL are not supported, use api_key")
if parts.query or parts.fragment:
raise ValueError("query strings and fragments are not supported")
if parts.netloc.endswith(":"):
raise ValueError("empty port")
url_port = parts.port
for candidate in (url_port, port):
if candidate is not None and not _MIN_PORT <= candidate <= _MAX_PORT:
raise ValueError(
f"port {candidate} is out of range {_MIN_PORT}-{_MAX_PORT}"
)
except ValueError as exc:
raise ValueError(f"Invalid endpoint URL {safe!r}: {exc}") from exc

if url_port is None:
if port is not None:
parts = parts._replace(netloc=f"{parts.netloc}:{port}")
elif port is not None and port != url_port:
raise ValueError(
f"Endpoint URL {safe!r} already specifies port {url_port}, "
f"which conflicts with port={port}"
)
return urlunsplit(parts._replace(path=parts.path.rstrip("/")))


def _openai_base_url(base_url: str) -> str:
"""Append the OpenAI `/v1` prefix when `base_url` has no path.

A URL that already names a path (`/v1`, or a gateway prefix such as
`/openai/v1`) is used as-is, matching the OpenAI SDK convention that
`base_url` includes the API prefix.
"""
parts = urlsplit(base_url)
if parts.path in ("", "/"):
return urlunsplit(parts._replace(path=_OPENAI_API_PREFIX))
return base_url


@dataclass
class ToolCall:
Expand Down Expand Up @@ -70,12 +143,15 @@ class LLMClient(ABC):

Args:
endpoint (`str`):
The base URL of the LLM service (e.g. "http://localhost").
port (`int`):
The port the service listens on.
The `http(s)` base URL of the LLM service (e.g. "http://localhost").
May include a port and a path (e.g. "http://localhost:8000/v1").
Credentials, query strings and fragments are rejected.
port (`int` or `None`):
The port the service listens on. Appended to `endpoint` when the
URL does not name one; must match the URL's port when both are given.
"""

def __init__(self, endpoint: str, port: int):
def __init__(self, endpoint: str, port: int | None):
self.endpoint = endpoint
self.port = port

Expand Down Expand Up @@ -122,8 +198,8 @@ async def complete_with_tools(

@property
def base_url(self) -> str:
"""Construct base URL from endpoint and port."""
return f"{self.endpoint}:{self.port}"
"""Base URL of the service: `endpoint` plus `port` when the URL names none."""
return _join_endpoint_port(self.endpoint, self.port)


class OpenAIClient(LLMClient):
Expand All @@ -134,9 +210,12 @@ class OpenAIClient(LLMClient):

Args:
endpoint (`str`):
The base URL (e.g. "http://localhost").
port (`int`):
The port number.
The base URL (e.g. "http://localhost"). May include a port and a
path (e.g. "http://localhost:8000/v1"). The `/v1` API prefix is
appended when the URL has no path; a URL with a path is used as-is.
port (`int` or `None`):
The port number, appended when `endpoint` does not name one; must
match the URL's port when both are given.
model (`str`):
Model name to pass to the API.
api_key (`str`, *optional*):
Expand All @@ -155,7 +234,7 @@ class OpenAIClient(LLMClient):
def __init__(
self,
endpoint: str,
port: int,
port: int | None,
model: str,
api_key: str | None = None,
system_prompt: str | None = None,
Expand All @@ -174,7 +253,7 @@ def __init__(
self._omit_temperature = use_max_completion_tokens

self._client = AsyncOpenAI(
base_url=f"{self.base_url}/v1",
base_url=_openai_base_url(self.base_url),
api_key=api_key if api_key is not None else "not-needed",
)

Expand Down Expand Up @@ -246,9 +325,10 @@ class AnthropicClient(LLMClient):

Args:
endpoint (`str`):
The base URL (e.g. `https://api.anthropic.com`).
port (`int`):
The port number.
The base URL (e.g. `https://api.anthropic.com`). May include a port.
port (`int` or `None`):
The port number, appended when `endpoint` does not name one; must
match the URL's port when both are given.
model (`str`):
Model name (e.g. "claude-sonnet-4-20250514").
api_key (`str`, *optional*):
Expand All @@ -264,7 +344,7 @@ class AnthropicClient(LLMClient):
def __init__(
self,
endpoint: str,
port: int,
port: int | None,
model: str,
api_key: str | None = None,
system_prompt: str | None = None,
Expand Down
Loading
Loading