Simulators¶

Simulators drive an agent through multi-turn conversations for evaluation. You pit an agent under test against a simulated human and run the exchange to completion, producing a transcript you can score with the conversation eval flows.

Everything is exported from latent.simulators:

from latent.simulators import (
    ConversationAgent,
    ConversationSimulator,
    SimulationConfig,
    SimulationResult,
    HumanAgent,
)

HumanAgent¶

HumanAgent simulates the human side of a conversation. It is a ReActAgent subclass and operates in one of three modes, selected by constructor arguments:

Mode	Trigger	Behavior
Script	`script=[...]`	Replays fixed messages in order. No LLM calls.
Goal-directed	`goals=[...]`	LLM generates messages pursuing objectives.
Free-form	neither	LLM generates messages with no specific goals.

script and goals are mutually exclusive — passing both raises ValueError.

Script mode¶

from latent.simulators import HumanAgent

human = HumanAgent(
    context="A customer with a billing question.",
    script=[
        "Hi, I was charged twice this month.",
        "My account email is alex@example.com.",
        "Thanks, that fixes it.",
    ],
)

Scripted humans replay messages deterministically and signal should_end once the list is exhausted. They register no tools.

Goal-directed mode¶

human = HumanAgent(
    context="A frustrated user whose order never arrived.",
    goals=[
        "Get a refund for the missing order.",
        "Stay polite but firm.",
    ],
    model="gpt-4o",
)

Free-form mode¶

human = HumanAgent(
    context="A curious user exploring the product.",
    model="gpt-4o",
)

Goal-directed and free-form humans get an end_call tool so the LLM can end the conversation naturally.

Constructor¶

HumanAgent(
    context: str = "",
    *,
    goals: list[str] | None = None,
    script: list[str] | None = None,
    system_prompt: str | None = None,   # override the default prompt entirely
    model: str = "gpt-4o-mini",
    temperature: float = 0.5,
    name: str = "human",
    max_tokens: int = 4096,
)

The `respond(...)` API¶

respond is the primary entry point used by the simulator. It is async:

response = await human.respond(context=[
    {"role": "assistant", "content": "How can I help you today?"},
])
# response.text, response.tool_calls, response.should_end

context is the conversation history as role/content dicts. The other agent's messages are role="assistant" and this agent's own messages are role="user"; HumanAgent swaps roles internally so the other side reads as its user input. It returns an AgentResponse. Call human.reset() to clear state between conversations.

ConversationSimulator¶

ConversationSimulator runs the loop between an agent under test and a HumanAgent. Any object satisfying the ConversationAgent protocol — i.e. it has async def respond(context) -> AgentResponse — works, as does a plain async callable.

from latent.agents import ReActAgent
from latent.simulators import ConversationSimulator, HumanAgent, SimulationConfig

agent_under_test = MyAgent(name="support_bot", model="gpt-4o")  # ConversationAgent
human = HumanAgent(context="...", goals=["..."])

simulator = ConversationSimulator(
    agent_under_test=agent_under_test,
    human_agent=human,
    config=SimulationConfig(max_turns=20),
)

result = await simulator.simulate(scenario_context="The user opens a support chat.")

`simulate(...)`¶

async def simulate(self, scenario_context: str | None = None) -> SimulationResult: ...

scenario_context, if given, is sent as the first user message to the agent under test. The loop alternates agent and human turns until either side signals should_end=True, returns empty text, or max_turns is reached. The simulator resets agents that support reset() before starting and always returns a SimulationResult, even on error.

SimulationConfig¶

@dataclass
class SimulationConfig:
    max_turns: int = 30   # maximum turns before forced termination

SimulationResult¶

@dataclass
class SimulationResult:
    conversation: list[dict[str, str]]   # turn dicts with "role" / "content"
    turn_count: int                      # number of user+assistant exchange pairs
    success: bool                        # completed without errors
    error: str | None = None             # error message if it failed
    metadata: dict[str, Any] = field(default_factory=dict)

The conversation dicts map directly onto Turn(**turn_dict), so a result feeds straight into the conversation eval flows.

The response contract¶

Simulation exchanges values via latent.agents.response. This is the non-streaming view of an agent turn — a single resolved response, distinct from the incremental Agent Protocol stream.

AgentResponse¶

@dataclass
class AgentResponse:
    text: str
    tool_calls: list[ToolCallRecord] = field(default_factory=list)
    should_end: bool = False

should_end is the simulation's termination signal.

ToolCallRecord¶

@dataclass
class ToolCallRecord:
    tool_name: str
    arguments: dict[str, Any] = field(default_factory=dict)

ToolCallRecord is a combined call record used here in the simulation view. This differs from the streaming Agent Protocol, which emits separate ToolCall events plus role="tool" result messages — there a call and its result are distinct items, whereas ToolCallRecord summarizes just the invocation (name + arguments).

respond_from_events¶

respond_from_events is the standard respond() implementation for ReActAgent-based agents. It calls agent.run(messages) (which repopulates agent.events), then extracts tool calls and the end signal:

async def respond_from_events(
    agent: StreamingAgent,
    messages: list[Message],
    *,
    end_tool_name: str = END_CALL_TOOL_NAME,   # "end_call"
) -> AgentResponse: ...

It scans agent.events for ToolCall events, maps each to a ToolCallRecord, and sets should_end=True if any call matches end_tool_name.

StreamingAgent¶

StreamingAgent is the protocol respond_from_events requires: an async run(messages) -> str method plus an events list. It is satisfied by ReActAgent and its subclasses (including HumanAgent).

class StreamingAgent(Protocol):
    events: list[AgentEvent]
    async def run(self, messages: list[Message]) -> str: ...

The wire-level Message, ToolCall, and related stream types are importable from latent.protocol:

from latent.protocol import Message, ToolCall