Calling language models

behaviors is a Python library for calling language models from fxtr experiments. It provides:

model adapters for Anthropic, OpenAI, and OpenAI-compatible services;
conversation sessions and context policies, which decide what the model sees next: scripted user messages, tool results, or custom logic;
retries for rate limits and server errors;
conversation records that are fxtr entities, so they’re stored, linked, and shown in the viewer.

Set up

Add behaviors, with its provider SDKs, to your project. Like fxtr, it comes from your fxtr checkout by path. In pyproject.toml:

pyproject.toml

[project]
dependencies = ["fxtr[runner,debug-web]", "behaviors[models]"]

[tool.uv.sources]
fxtr = { path = "/path/to/fxtr/fxtr", editable = true }
behaviors = { path = "/path/to/fxtr/behaviors", editable = true }

Then run uv sync and commit the updated uv.lock. Set your provider’s API key in the environment: OPENAI_API_KEY or ANTHROPIC_API_KEY. Never store credentials in experiment inputs or entities.

Provider	Adapter
OpenAI Responses	`from behaviors.implementations.models.openai_responses import OpenAIResponsesAPI`
Anthropic Messages	`from behaviors.implementations.models.anthropic import AnthropicMessagesAPI`
OpenAI Chat Completions-compatible	`from behaviors.implementations.models.openai_compatible import OpenAICompatibleAPI`

Construct an adapter with the provider’s model ID, such as OpenAIResponsesAPI("gpt-6-luna"). The OpenAI adapters also accept base_url for compatible services.

A conversation as a step

This step holds a two-turn conversation: it asks a question, then asks the model to explain its answer. The model and sampling settings arrive as a configuration entity, so every conversation points at the exact settings that produced it.

from dataclasses import dataclass
from typing import Annotated, cast

import anyio

from behaviors.implementations.context_policies import (
    ScriptedUserSimulator,
    context_policy_from_user_simulator,
)
from behaviors.implementations.models import chat_session_model_from_api
from behaviors.implementations.models.openai_responses import OpenAIResponsesAPI
from behaviors.rollouts.chat import run_chat_rollout
from behaviors.types import (
    ChatRollout, ContentText, RetryConfig, SamplingParams, SetSystemMessage,
    SystemMessage, UserMessage,
)
from fxtr.core.entities import BoundID
from fxtr.entity_defns.dataclass_entity import DataclassEntity
from fxtr.experiment.handles import ArrayHandle
from fxtr.experiment.steps import StepContext, step
from fxtr.experiment.workflows import WorkflowContext, workflow


@dataclass(frozen=True)
class ConversationConfig(DataclassEntity, fxtr_type="org.example.my_experiment.ConversationConfig.v1"):
    model_name: str
    sampling: SamplingParams
    system_prompt: str


def make_api(model_name: str) -> OpenAIResponsesAPI:
    return OpenAIResponsesAPI(model_name)


@step(name="my_experiment.conversation")
async def converse(context: StepContext, config: ConversationConfig, prompt: str) -> ChatRollout:
    """Answer a prompt, then explain the answer in a follow-up turn."""
    api = make_api(config.model_name)
    try:
        model = chat_session_model_from_api(api, config.sampling, retry=RetryConfig())
        user = ScriptedUserSimulator(messages=(
            UserMessage(content=(ContentText(text=prompt),)),
            UserMessage(content=(ContentText(text="Explain your answer."),)),
        ))
        policy = context_policy_from_user_simulator(
            user,
            initialization=SetSystemMessage(
                message=SystemMessage(content=(ContentText(text=config.system_prompt),)),
            ),
        )
        # One more turn than there are scripted prompts, so the rollout can end as completed.
        return await run_chat_rollout(model, policy, max_turns=3)
    finally:
        with anyio.CancelScope(shield=True):
            await api.aclose()


@workflow(name="my_experiment.run")
async def experiment(
    context: WorkflowContext,
    configs: Annotated[ArrayHandle[BoundID[ConversationConfig]], "[model: str]"],
    prompts: Annotated[ArrayHandle[str], "[prompt: str]"],
) -> Annotated[ArrayHandle[BoundID[ChatRollout]], "[model: str, prompt: str]"]:
    """Hold one conversation per model and prompt."""
    rollouts = context.run_step(
        "conversations",
        converse,
        {"config": configs, "prompt": prompts},
        map_over=["model", "prompt"],
    )
    return cast(ArrayHandle[BoundID[ChatRollout]], rollouts)

In this step:

chat_session_model_from_api wraps the adapter as a conversation model. Pass retry=RetryConfig() to retry rate limits, overloads, and server errors; without it, calls aren’t retried.
ScriptedUserSimulator sends one scripted message per turn and then ends the conversation. context_policy_from_user_simulator turns it into a context policy, here starting with a system message.
run_chat_rollout alternates between the policy and the model, and returns a ChatRollout: the whole conversation, with its status. The step’s output is an entity, so fxtr stores it.
Building the adapter in one helper, make_api, lets you substitute a fake model for credential-free checks.
The adapter is closed in finally, because run_chat_rollout closes the conversation sessions but not the provider client.

Launch it with a stored configuration per model:

import anyio

from fxtr.entity_defns.array import Array
from fxtr.project.running import open_local_client, run_job
from my_project.conversations import ConversationConfig, experiment


async def main() -> None:
    async with open_local_client(__file__) as client:
        luna = await client.store(ConversationConfig(
            model_name="gpt-6-luna",
            sampling=SamplingParams(max_tokens=2000),
            system_prompt="Answer concisely.",
        ))
        configs = Array.from_items([(("luna",), luna)], ("[model: str]", BoundID[ConversationConfig]))
        prompts = Array.from_items([(("greeting",), "Hello!")], ("[prompt: str]", str))
        await run_job(client, experiment, {"configs": configs, "prompts": prompts}, root="conv-v1")


if __name__ == "__main__":
    anyio.run(main)

For repeated samples of each conversation, add replicas={"sample": range(n)} to the run_step call (see Replicas).

A judge step

Judging needs no conversation state, so a judge calls the model once with generate_with_retries. This step grades a conversation’s final answer and returns a Verdict entity, which the viewer can link to. Put it in the same module as the conversation step, whose make_api it uses:

import re
from dataclasses import dataclass
from typing import Annotated

import anyio

from behaviors.implementations.models import (
    DEFAULT_FAILURE_CATEGORIES_TO_RETRY,
    generate_with_retries,
)
from behaviors.types import (
    AssistantMessage, ChatMessage, ChatRollout, ContentText, ModelCallFailure,
    ModelRequest, RetryConfig, RolloutCompleted, SamplingParams, UserMessage,
)
from fxtr.core.entities import BoundID
from fxtr.entity_defns.array import Array
from fxtr.entity_defns.dataclass_entity import DataclassEntity
from fxtr.experiment.steps import StepContext, step


@dataclass(frozen=True)
class JudgeConfig(DataclassEntity, fxtr_type="org.example.my_experiment.JudgeConfig.v1"):
    model_name: str
    sampling: SamplingParams
    rubric: str


@dataclass(frozen=True)
class Verdict(DataclassEntity, fxtr_type="org.example.my_experiment.Verdict.v1"):
    score: int | None  # None when there was nothing to grade or the reply had no score
    judge_reply: str


def answer_text(messages: tuple[ChatMessage, ...]) -> str:
    """The assistant's text, without reasoning or tool-use blocks."""
    return "".join(
        block.text
        for message in messages
        if isinstance(message, AssistantMessage)
        for block in message.content
        if isinstance(block, ContentText)
    )


@step(name="my_experiment.judge")
async def judge(context: StepContext, judge_config: JudgeConfig, rollout: ChatRollout) -> Verdict:
    """Grade the conversation's final answer against the rubric from 0 to 10."""
    if not isinstance(rollout.status, RolloutCompleted) or not rollout.model_turns:
        return Verdict(score=None, judge_reply=f"not graded: {rollout.status.kind}")
    prompt = (
        f"{judge_config.rubric}\n\n"
        f"Answer to grade:\n{answer_text(rollout.model_turns[-1].messages)}\n\n"
        "End your reply with a line of the form 'Score: N', where N is from 0 to 10."
    )
    request = ModelRequest(
        context=(UserMessage(content=(ContentText(text=prompt),)),),
        tools=(),
        tool_choice=None,
        parallel_tool_calls=None,
        sampling=judge_config.sampling,
    )
    api = make_api(judge_config.model_name)
    try:
        response = await generate_with_retries(
            api,
            request,
            retry=RetryConfig(),
            failure_categories_to_retry=DEFAULT_FAILURE_CATEGORIES_TO_RETRY,
        )
    finally:
        with anyio.CancelScope(shield=True):
            await api.aclose()
    if isinstance(response, ModelCallFailure):
        # Raise rather than return: a returned failure would be cached as the result.
        raise RuntimeError(f"judge call failed ({response.category}): {response.description}")
    reply = answer_text(response.output)
    match = re.search(r"Score:\s*(\d+)\s*$", reply)
    score = int(match.group(1)) if match else None
    return Verdict(score=score if score is not None and 0 <= score <= 10 else None, judge_reply=reply)

Schedule it in the workflow, mapped over the conversations, with the judge configuration as a single stored input. The judge_config: JudgeConfig and rollout: ChatRollout annotations load the entities before the body runs:

verdicts = context.run_step(
    "judge_final_answers",
    judge,
    {"judge_config": judge_config, "rollout": rollouts},
    map_over=["model", "prompt"],
)

The judge follows the return or raise rule: it returns a Verdict for outcomes every attempt would reproduce, such as a conversation that never completed or a reply without a score. It raises when the model call still fails after its retries, so resuming the job calls the judge again.

Summarize the verdicts

A reduction step takes one model’s verdict references and loads them:

@step(name="my_experiment.mean_score")
async def mean_score(
    context: StepContext,
    verdicts: Annotated[Array[BoundID[Verdict]], "[prompt: str]"],
) -> float | None:
    """Average the judge's scores across prompts."""
    scores: list[int] = []
    for ref in verdicts.values():
        verdict = await context.load(ref)
        if verdict.score is not None:
            scores.append(verdict.score)
    return sum(scores) / len(scores) if scores else None


# In the workflow:
mean_scores = context.run_step("mean_scores", mean_score, {"verdicts": verdicts}, map_over=["model"])

Rollouts, settings, and testing

Check a rollout's status before scoring it

A rollout ends as RolloutCompleted, RolloutMaxTurns, or RolloutFailed. max_turns counts model turns: a script with N prompts needs max_turns=N + 1 to end as completed. A model stop reason such as StopMaxTokens is separate from the rollout’s status, and may mean the content was truncated even in a completed rollout.

Recorded failures are cached as results

By default, a rollout that exceeds the model’s context length is returned as RolloutFailed, keeping the conversation so far; other failures raise. A returned failed rollout is a successful step result, so it’s cached. Decide deliberately how scoring treats such outcomes.

Sampling settings vary by provider

SamplingParams covers max_tokens, temperature, top_p, stop_sequences, reasoning, and provider-specific extra. Which settings take effect depends on the adapter and model: for example, OpenAIResponsesAPI doesn’t apply stop_sequences. Leave a setting unset unless your adapter and model support it.

Test without credentials

Any object with an async generate(request) method returning a ModelResponse, and an aclose() method, can stand in for a provider adapter. Return it from make_api in tests, and run against a scratch schema.

Tools and custom conversation flows

context_policy_from_user_and_tools combines a user simulator with a tool policy you implement, which defines the tools and executes their calls. For fully custom flows, context_policy_from_generator turns an async generator into a context policy: it yields each turn’s messages and receives the model’s reply.

Get started

Guides

Calling language models

Set up

A conversation as a step

A judge step

Summarize the verdicts

Rollouts, settings, and testing

​Set up

​A conversation as a step

​A judge step

​Summarize the verdicts

​Rollouts, settings, and testing

Set up

A conversation as a step

A judge step

Summarize the verdicts

Rollouts, settings, and testing