Generation

Connect the acting model. A policy backend generates tokens; PolicyFormat binds its tokenizer and chat template to an action parser. Backends resolve the format from the model name unless you supply one. Import these from focus.backends; build a policy step by step in Quick start: MATH.

GenerateResult

Record one model response, including generated tokens, text, probabilities, and stopping information. Backend adapters return this record to the policy, which hands it to the agent.

from focus.backends import GenerateResult

Definition

@dataclass
class GenerateResult

Fields

tokens: tuple[TokenId, ...] = field(default_factory=tuple)
prefix_tokens: tuple[TokenId, ...] = field(default_factory=tuple)
text: str = ''
stop_reason: str = 'stop'
matched_stop: int | str | None = None
logprobs: list[float] = field(default_factory=list)
top_logprobs: list[dict[str, float]] | None = None
prompt_logprobs: list[float | None] = field(default_factory=list)
prompt_top_logprobs: list[dict[str, float] | None] | None = None
entropy: float = 0.0
exact_generation: bool = True
native_tool_calls: tuple[NativeToolCall, ...] | None = None
native_output: tuple[dict[str, Any], ...] = ()

HermesToolCallParser

Read Hermes-style JSON tool-call blocks from a model response and separate them from reasoning and visible text.

from focus.backends import HermesToolCallParser

parser = HermesToolCallParser()

Use in PolicyFormat for a model whose chat template emits Hermes tool-call blocks.

Definition

class HermesToolCallParser(ActionParser[ToolTurn])

HermesToolCallParser.parse

def parse(self, result: GenerateResult) -> ToolTurn

NativeToolCallParser

Convert a backend’s structured tool calls into ToolTurn actions for ToolEnv.

from focus.backends import NativeToolCallParser

parser = NativeToolCallParser()

Use with a backend that returns tool calls in structured response fields.

Definition

class NativeToolCallParser(ActionParser[ToolTurn])

NativeToolCallParser.parse

def parse(self, result: GenerateResult) -> ToolTurn

OpenAIChatPolicy

Run agent interactions through a Chat Completions endpoint with native message and tool-call handling. Local token accounting is estimated.

from focus.backends import OpenAIChatPolicy

policy = OpenAIChatPolicy(model=model_name)

This backend uses the configured API credentials and structured tool calls for agent execution and evaluation. Exact actor-training spans require a backend that preserves exact token IDs.

Definition

class OpenAIChatPolicy(NativePolicy)

OpenAIChatPolicy.init

def __init__(self, *, model: str, format: PolicyFormat[Any] | None=None, default_params: SamplingParams | None=None, api_key: str | None=None, base_url: str | None=None, system_prompt: str | None=None, locked_sampling: bool=False) -> None

OpenAIChatPolicy.agenerate

async def agenerate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

OpenAIResponsesPolicy

Run agent interactions through a Responses endpoint, preserving its native response and tool-call structure.

from focus.backends import OpenAIResponsesPolicy

policy = OpenAIResponsesPolicy(model=model_name)

This backend uses the configured API credentials for agent execution and evaluation. Use a backend that preserves exact token IDs when collecting exact actor-training spans.

Definition

class OpenAIResponsesPolicy(NativePolicy)

OpenAIResponsesPolicy.init

def __init__(self, *, model: str, format: PolicyFormat[Any] | None=None, default_params: SamplingParams | None=None, api_key: str | None=None, base_url: str | None=None, system_prompt: str | None=None, locked_sampling: bool=False, reasoning_effort: str | None=None) -> None

OpenAIResponsesPolicy.agenerate

async def agenerate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

Policy

Generate model output through a backend or a custom completion function. Its format supplies the codec and the agent’s default action parser.

from focus.backends import Policy

policy = Policy(format=policy_format, complete=complete)

policy_format determines how inputs and outputs are represented. complete returns generated tokens and their log probabilities, which remain attached to those tokens through training-data preparation. The context manager prepares each request. Keep the actor and critic fixed for each tree; the training loop owns model snapshots and batch-age checks. Several collectors can share local execution capacity while calling their own policy objects directly.

Definition

class Policy[A]

Policy.init

def __init__(self, *, format: PolicyFormat[A], model: str='custom', served_model: str | None=None, default_params: SamplingParams | None=None, complete: Completion | None=None, exact: bool=True, native: bool=False) -> None

Policy.format

@property
def format(self) -> PolicyFormat[A]

Policy.model

@property
def model(self) -> str

Policy.served_model

@property
def served_model(self) -> str

Policy.trainable

@property
def trainable(self) -> bool

Policy.prepare

def prepare(self, messages: Sequence[Message], tools: Sequence[dict[str, Any]]=()) -> PreparedPrompt

Policy.validate_context

def validate_context(self, messages: Sequence[Message]) -> None

Check an edited conversation before committing it. Native API policies check tool-call and reply ordering; context managers enforce their own prefix-preservation requirements. This check makes no model request.

Policy.agenerate

async def agenerate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

Policy.generate

def generate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

Policy.agenerate_tokens

async def agenerate_tokens(self, tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

Generate directly from token IDs using the effective sampling settings. Use this path for rollout requests and replay.

Policy.generate_tokens

def generate_tokens(self, tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

Policy.decode

def decode(self, tokens: Sequence[int], *, skip_special_tokens: bool=False) -> str

Policy.parse

def parse(self, result: GenerateResult) -> A

Policy.is_exact

def is_exact(self, result: GenerateResult) -> bool

Report whether this particular response preserves exact token provenance. trainable describes the policy’s capability before collection; is_exact() records what happened on an individual call, including when a replay policy mixes exact and inexact responses.

Policy.record_reply

def record_reply(self, result: GenerateResult, messages: list[Message]) -> None

Policy.startup_check

def startup_check(self) -> None

Check that the policy is ready to serve requests. Backends can override this hook to contact their server.

Policy.close

def close(self) -> None

Release policy-owned resources.

Policy.aclose

async def aclose(self) -> None

Release resources owned by an asynchronous policy client.

PolicyFormat

Bind a codec for encoding model input to a parser for interpreting model output.

from focus.backends import PolicyFormat

policy_format = PolicyFormat.resolve(
    "Qwen/Qwen3.5-4B", tokenizer=tokenizer, profile="token_chunk"
)

The codec renders messages as model tokens. The parser interprets output in the model’s format and serves as the agent’s default. resolve() loads the tokenizer and selects a parser from profile, or infers a supported tool-call dialect from the model name. Profiles are qwen_xml, hermes, native, text, and token_chunk. The agent can adapt the parser for text or token-chunk actions; Agent(parser=...) selects an explicit parser. Tool environments need a parser that matches the model’s tool-call syntax. Construct PolicyFormat(name, codec, parser) directly for a custom format.

Definition

@dataclass(frozen=True)
class PolicyFormat[A]

Fields

name: str
codec: ChatCodec
parser: ActionParser[A]

PolicyFormat.prepare

def prepare(self, messages: Sequence[Message], tools: Sequence[dict[str, Any]]=()) -> PreparedPrompt

Render the supplied conversation and tool schemas into a prompt snapshot.

PolicyFormat.record_reply

def record_reply(self, result: GenerateResult, messages: list[Message]) -> None

Append an assistant message, preserving native tool-call fields when supplied.

PolicyFormat.resolve

@classmethod
def resolve(cls, model: str, *, tokenizer: HFTokenizer | None=None, profile: str | None=None) -> PolicyFormat[Any]

The tokenizer, chat template, and parser for a known model family.

profile names the parser: "qwen_xml" (Qwen3.5 tool calls), "hermes" (Qwen2.5 tool calls), "native" (API tool calls), "text", or "token_chunk" (exact segments). Without one, the model name selects the tool dialect.

PreparedPrompt

Store the messages, tool schemas, and tokens for one generation request. The agent’s context manager prepares it; policy.prepare(messages) renders an ordinary conversation.

from focus.backends import PreparedPrompt

Definition

@dataclass(frozen=True)
class PreparedPrompt

Fields

tokens: tuple[TokenId, ...]
format: PolicyFormat[Any]
_messages: tuple[Message, ...]
_tools: tuple[dict[str, Any], ...]
exact_continuation: bool = False

PreparedPrompt.messages

@property
def messages(self) -> list[Message]

PreparedPrompt.tools

@property
def tools(self) -> list[dict[str, Any]]

QwenXMLToolCallParser

Read Qwen3.5-style XML tool calls and convert their function names and arguments into ToolTurn actions.

from focus.backends import QwenXMLToolCallParser

parser = QwenXMLToolCallParser()

Use in PolicyFormat for a model whose template emits Qwen XML tool calls.

Definition

class QwenXMLToolCallParser(ActionParser[ToolTurn])

QwenXMLToolCallParser.parse

def parse(self, result: GenerateResult) -> ToolTurn

SGLangPolicy

Generate token continuations through an independently hosted SGLang server, preserving their exact token IDs.

from focus.backends import SGLangPolicy

policy = SGLangPolicy(
    model=model_name,
    format=policy_format,
    base_url="http://localhost:30000",
)

This policy uses the server’s native /generate endpoint and owns its HTTP clients. Supply the model’s format so token rendering and action parsing match the server.

Definition

class SGLangPolicy(Policy[Any])

SGLangPolicy.init

def __init__(self, *, model: str, base_url: str, format: PolicyFormat[Any] | None=None, profile: str | None=None, default_params: SamplingParams | None=None, api_key: str | None=None, headers: Mapping[str, str] | None=None, timeout: float | None=None) -> None

SGLangPolicy.generate

def generate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

SGLangPolicy.generate_tokens

def generate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

SGLangPolicy.agenerate_tokens

async def agenerate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

SGLangPolicy.startup_check

def startup_check(self) -> None

SGLangPolicy.close

def close(self) -> None

SGLangPolicy.aclose

async def aclose(self) -> None

SamplingParams

Set generation controls for a request, including sampling temperature, output length, and requested token probabilities.

from focus.backends import SamplingParams

params = SamplingParams(
    temperature=0.7, max_tokens=512, logprobs=0
)

This request allows up to 512 generated tokens and asks for each sampled token’s log probability. Those probabilities support training and token-based allocation.

Definition

@dataclass
class SamplingParams

Fields

temperature: float | None = DEFAULT_TEMPERATURE
max_tokens: int | None = DEFAULT_MAX_TOKENS
top_p: float | None = DEFAULT_TOP_P
top_k: int | None = None
stop: list[str] | None = None
logprobs: int | None = None
prompt_logprobs: int | None = None
repetition_penalty: float | None = None
seed: int | None = None

SlimePolicy

Generate tokens through Slime’s SGLang router using the training framework’s HTTP transport.

from focus.backends import SlimePolicy

policy = SlimePolicy(
    post, url, model=model_name, format=policy_format
)

post supplies Slime’s asynchronous transport, and url identifies the generation endpoint. Slime controls admission to that transport.

Definition

class SlimePolicy(Policy[Any])

SlimePolicy.init

def __init__(self, post: PostFn, url: str, *, model: str, format: PolicyFormat[Any] | None=None, profile: str | None=None, default_params: SamplingParams | None=None, headers: Mapping[str, str] | None=None) -> None

SlimePolicy.agenerate_tokens

async def agenerate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

TextActionParser

Use the stripped completion text as the environment action.

from focus.backends import TextActionParser
from focus.backends import GenerateResult

parser = TextActionParser()

The agent uses this parser for an environment that reads text, even when the policy’s format parses tool calls.

Definition

class TextActionParser(ActionParser[str])

TextActionParser.parse

def parse(self, result: GenerateResult) -> str

TinkerPolicy

Use a Tinker sampling client as the generation backend for a rollout.

from focus.backends import TinkerPolicy

policy = TinkerPolicy(
    sampling_client, model=model_name, format=policy_format
)

sampling_client generates from a fixed policy snapshot. policy_format supplies prompt encoding and a default action parser; the agent’s context manager prepares each request.

Definition

class TinkerPolicy(Policy[Any])

TinkerPolicy.init

def __init__(self, sampling_client: tinker.SamplingClient, *, model: str, format: PolicyFormat[Any] | None=None, profile: str | None=None, default_params: SamplingParams | None=None)

TinkerPolicy.agenerate_tokens

async def agenerate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

TinkerPolicy.close

def close(self) -> None

TinkerPolicy.enter

def __enter__(self) -> TinkerPolicy

TinkerPolicy.exit

def __exit__(self, *exc: object) -> None

TokenChunk

Carry one generated segment into an environment: the original token IDs and whether the model finished its response.

from focus.backends import TokenChunk

chunk = TokenChunk(tokens=(10, 20), finished=False)

The environment appends these two tokens to its state. finished=False means generation can continue from that prefix; finished=True tells a final-answer environment to grade the completed response.

Definition

@dataclass(frozen=True)
class TokenChunk

Fields

tokens: tuple[TokenId, ...]
finished: bool

TokenChunkParser

Extract a generated segment’s original token IDs and whether the response has finished.

from focus.backends import TokenChunkParser

parser = TokenChunkParser()

A length finish produces finished=False; a stop finish produces finished=True. Other finish reasons raise an error. Segments uses this parser and sets the generation limit; the execution service validates each response against that limit before the parser consumes it.

Definition

class TokenChunkParser(ActionParser[TokenChunk])

TokenChunkParser.parse

def parse(self, result: GenerateResult) -> TokenChunk

TokenId

An integer identifying one token in the policy’s tokenizer vocabulary. Token IDs must come from the tokenizer bound to that policy.

from focus.backends import TokenId

Definition

TokenId = int

ToolTurn

Represent a parsed assistant response as visible text, reasoning, and structured tool calls. Tool parsers produce this record and ToolEnv consumes it.

from focus.backends import ToolTurn

Definition

@dataclass(frozen=True)
class ToolTurn

Fields

text: str
tool_calls: tuple[ToolCall, ...] = field(default_factory=tuple)
reasoning: str = ''

VLLMPolicy

Generate from token prompts through a vLLM server and retain exact token IDs and behavior log probabilities for training.

from focus.backends import SamplingParams, VLLMPolicy

policy = VLLMPolicy(
    model=model_name,
    format=policy_format,
    base_url="http://localhost:8000/v1",
    api_key="EMPTY",
    default_params=SamplingParams(max_tokens=512, logprobs=0),
)

model_name identifies the model and tokenizer, and policy_format supplies the template and parser. Set served_model if the server uses a different model name. The endpoint must support vLLM’s token-completion fields.

Definition

class VLLMPolicy(Policy[Any])

VLLMPolicy.init

def __init__(self, *, model: str, format: PolicyFormat[Any] | None=None, profile: str | None=None, served_model: str | None=None, base_url: str, api_key: str, default_params: SamplingParams | None=None, timeout: float | None=None) -> None

Connect to the vLLM endpoint with the supplied model format. timeout sets the request deadline in seconds; None uses the SDK default. Allow for queueing time when many requests share the server, since timeouts and retries can add work to an already busy queue.

VLLMPolicy.startup_check

def startup_check(self) -> None

VLLMPolicy.generate

def generate(self, prompt: PreparedPrompt, sampling_params: SamplingParams | None=None) -> GenerateResult

VLLMPolicy.generate_tokens

def generate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

VLLMPolicy.agenerate_tokens

async def agenerate_tokens(self, prefix_tokens: Sequence[TokenId], sampling_params: SamplingParams | None=None) -> GenerateResult

VLLMPolicy.close

def close(self) -> None

VLLMPolicy.aclose

async def aclose(self) -> None

ActionKind

What an action is: "text" (the reply’s text), "tool" (a tool-calling turn), or "tokens" (an exact token chunk). Parsers declare the kind they produce; environments and context managers declare the kind they need, and the agent adapts the policy format’s parser to match.

from focus.extend import ActionKind

Definition

type ActionKind = Literal['text', 'tool', 'tokens']

NativeToolCall

Represent a structured tool call returned by a model backend, separately from completion text.

from focus.generation import NativeToolCall

Definition

@dataclass(frozen=True)
class NativeToolCall

Fields

name: str
arguments: str
call_id: str = ''

NativePolicy

Provide the shared message validation and client lifecycle inherited by OpenAI chat and Responses policies. Instantiate the concrete backend classes to select a transport.

from focus.generation._openai_shared import NativePolicy

Definition

class NativePolicy(Policy[Any])

NativePolicy.init

def __init__(self, *, model: str, format: PolicyFormat[Any] | None=None, default_params: SamplingParams | None=None, api_key: str | None=None, base_url: str | None=None, system_prompt: str | None=None, locked_sampling: bool=False) -> None

NativePolicy.validate_context

def validate_context(self, messages: Sequence[Message]) -> None

NativePolicy.close

def close(self) -> None

NativePolicy.aclose

async def aclose(self) -> None

ActionParser

Decode a GenerateResult into the action type consumed by Env.step(). Implement parse() and declare whether the parser supports token-level training or requires native tool calls.

from focus.generation.parsing import ActionParser

class IntegerAction(ActionParser[int]):
    def parse(self, result):
        return int(result.text.strip())

parser = IntegerAction()

Bind this parser in PolicyFormat when your environment consumes integer actions. The parser decodes the generated text; Env.step() defines what that integer does in the task.

Definition

class ActionParser[A](ABC)

Fields

trainable: bool = True
requires_native_channel: bool = False
action: ActionKind | None = None

ActionParser.parse

Required implementation.

@abstractmethod
def parse(self, result: GenerateResult) -> A

Return the action result expresses (or a task-defined malformed one).

ToolCall

Hold a parsed tool’s name and JSON argument string.

from focus.generation.parsing import ToolCall

Definition

@dataclass(frozen=True)
class ToolCall

Fields

name: str
arguments: str
call_id: str = ''

TOOL_CALL

Match serialized tool-call blocks in a model completion.

from focus.generation.parsing.base import TOOL_CALL

Definition

TOOL_CALL = re.compile('<tool_call>(.*?)</tool_call>', re.DOTALL)

tool_turn

Parse a completion into a tool-calling assistant turn using a supplied decoder for each call block.

from focus.generation.parsing.base import tool_turn

Definition

def tool_turn(text: str, decode: Callable[[str], ToolCall]) -> ToolTurn

Completion

Specify an asynchronous completion callback: accept prefix token IDs and optional sampling parameters, then return a GenerateResult.

from focus.generation.policy import Completion

async def complete(tokens, params):
    return await backend.agenerate_tokens(tokens, params)

completion: Completion = complete

The callback receives the exact prefix token IDs and optional per-call sampling settings. It returns a GenerateResult asynchronously. Bind backend to the policy transport used by your application.

Definition

Completion = Callable[[Sequence[TokenId], SamplingParams | None], Awaitable[GenerateResult]]

PostFn

Specify the asynchronous HTTP callback used by SlimePolicy to send generation requests.

from focus.generation.slime import PostFn

Definition

PostFn = Callable[..., Awaitable[Mapping[str, Any]]]

DEFAULT_MAX_TOKENS

Set the output-token limit used when sampling parameters leave it unspecified.

from focus.generation.types import DEFAULT_MAX_TOKENS

Definition

DEFAULT_MAX_TOKENS = 512

DEFAULT_TEMPERATURE

Set the sampling temperature used when no explicit value is supplied.

from focus.generation.types import DEFAULT_TEMPERATURE

Definition

DEFAULT_TEMPERATURE = 0.7

DEFAULT_TOP_P

Set the nucleus-sampling probability used when no explicit value is supplied.

from focus.generation.types import DEFAULT_TOP_P

Definition

DEFAULT_TOP_P = 1.0