ReAct Agent¶

ReActAgent is the standard Latent agent: a streaming, async, tool-calling ReAct loop powered by LiteLLM (any provider). This page covers the agent itself and its specialized subclasses (Judge, Classifier, RAGAgent). For the surrounding features see Tools & Retrievers, Guardrails, and the Agent Protocol. New here? Start with the Building Agents overview.

from latent.agents import (
    ReActAgent, Judge, Classifier, RAGAgent,
    ScoredModel, OrdinalScore, BinaryScore, ContinuousScore,
)
from latent.protocol import Message  # canonical wire types

Installation

The ReAct agent runtime ships in the base install (pip install latent). Judges and Classifiers used inside flows need the eval extra (pip install "latent[eval]"); the interactive REPL needs chat.

Async-only since 5.0.0

Every agent entry point — stream(), invoke(), run(), ask(), and __call__ — is a coroutine. Call them with await inside an async def, or wrap a top-level call in asyncio.run(...). (reset() is the one synchronous method.)

LiteLLMAgent was renamed to ReActAgent in 5.4.0

The old name still resolves to the same class but emits a DeprecationWarning and will be removed in 6.0.0. Update imports to from latent.agents import ReActAgent.

The wire types every agent emits (Message, AgentEvent and its subclasses, Step, InvokeResult, AgentProtocol) are documented in the Agent Protocol reference.

BaseAgent¶

The abstract base class. All agents implement a single method:

async def stream(
    self,
    messages: list[Message],
    *,
    config: dict[str, Any] | None = None,
) -> AsyncIterator[AgentEvent]:
    ...

stream() yields a sequence of typed events (TextDelta, ToolCall, ToolResult, Usage, StepBoundary, Error, etc.) that represent the agent's reasoning and output. Guardrail middleware is auto-discovered from @guardrail-decorated methods and injected at construction time.

ReActAgent¶

Concrete conversational agent powered by LiteLLM. Supports any model provider (OpenAI, Anthropic, Bedrock, Azure, etc.) and implements the ReAct pattern: the LLM can call tools, observe results, and iterate. (Formerly LiteLLMAgent.)

Construction¶

from latent.agents import ReActAgent

agent = ReActAgent(
    name="assistant",
    model="gpt-4o",
    system_prompt="You are a helpful assistant.",
    temperature=0.0,
    max_tokens=4096,
    max_iterations=10,       # ReAct loop cap
    response_format=None,    # Pydantic model for structured output
    tools=None,              # list of @tool-decorated functions
    optimized_prompt=None,   # path or OptimizedPrompt object
    # max_iterations_fallback_message=None,  # canned final text when loop is exhausted
    # max_iterations_fallback_tool=None,     # or force this tool as the final action
    # max_iterations_fallback_args=None,     # args for the forced fallback tool
    # thinking_config=None,    # provider extended-thinking / reasoning config
)

Parameters

Name	Type	Default	Description
`name`	`str`	required	Human-readable name for logging and MLflow
`model`	`str`	required	LiteLLM model identifier (e.g. `gpt-4o`, `anthropic/claude-sonnet-4-20250514`)
`system_prompt`	`str \\| None`	`None`	System message prepended to every call
`tools`	`list[Callable] \\| None`	`None`	Functions decorated with `@tool`
`temperature`	`float`	`0.0`	Sampling temperature
`max_tokens`	`int`	`4096`	Max output tokens per LLM call
`max_iterations`	`int`	`10`	Maximum ReAct loop iterations
`max_iterations_fallback_message`	`str \\| None`	`None`	On hitting `max_iterations`, emit this canned text as the final response instead of erroring
`max_iterations_fallback_tool`	`str \\| None`	`None`	On hitting `max_iterations`, force a final call to this tool (e.g. so a structured-output agent still returns something)
`max_iterations_fallback_args`	`dict \\| None`	`None`	Arguments passed to the forced `max_iterations_fallback_tool`
`thinking_config`	`dict \\| None`	`None`	Enable provider extended-thinking/reasoning; passed through to the model
`response_format`	`type[BaseModel] \\| dict \\| None`	`None`	Structured output schema (strict mode enforced)
`optimized_prompt`	`str \\| Path \\| None`	`None`	Load an optimized prompt from file

Usage patterns¶

Stateless by design

A Latent agent holds no conversation state between calls. Every stream() / invoke() / run() receives the full message history as its messages argument and returns based solely on that — so the same agent instance can safely serve many independent conversations concurrently. Persisting and replaying conversation history is the platform's job (e.g. Agent Studio), not the agent's.

The ask() helper is the one exception: it keeps a small transient in-memory history so you can chat turn-by-turn in a REPL or script. reset() clears that buffer. It does not imply the agent is stateful — for production, drive the agent with invoke(messages) and store history yourself (or let Agent Studio do it).

Single-shotMulti-turn chatAsync streamingStructured output

from latent.agents import ReActAgent
from latent.protocol import Message

agent = ReActAgent(name="qa", model="gpt-4o")
response = await agent.run([Message(role="user", content="What is 2+2?")])
print(response)  # "4"

agent = ReActAgent(
    name="tutor",
    model="gpt-4o",
    system_prompt="You are a math tutor.",
)

print(await agent.ask("What is a derivative?"))
print(await agent.ask("Can you give me an example?"))  # retains history

agent.reset()  # drop the transient in-memory ask() history (sync)

from latent.agents import ReActAgent
from latent.protocol import Message, TextDelta

agent = ReActAgent(name="streamer", model="gpt-4o")

async for event in agent.stream([Message(role="user", content="Tell me a joke")]):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)

from pydantic import BaseModel
from latent.agents import ReActAgent
from latent.protocol import Message

class City(BaseModel):
    name: str
    country: str
    population: int

agent = ReActAgent(
    name="extractor",
    model="gpt-4o",
    response_format=City,
)

text = await agent.run([Message(role="user", content="Tell me about Tokyo")])
city = City.model_validate_json(text)

Callable interface¶

await agent(question) resets context and returns a response string. Designed for evaluation frameworks where each call should be stateless:

agent = ReActAgent(name="eval_target", model="gpt-4o")
response = await agent("What is the capital of France?")  # resets, asks, returns text

`invoke()` — buffered, structured result¶

stream() is the wire-level API. invoke() is the buffered API: it consumes the stream internally and returns an InvokeResult with the terminal-step text in .text and the full per-step breakdown in .steps.

from latent.agents import ReActAgent
from latent.protocol import Message

agent = ReActAgent(name="qa", model="gpt-4o", tools=[my_tool])
result = await agent.invoke([Message(role="user", content="Use the tool, then answer")])

print(result.text)            # customer-facing answer (terminal step)
for step in result.steps:     # one Step per ReAct iteration
    print(step.tool_calls)    # list[ToolCall] the LLM invoked
    print(step.tool_messages) # list[Message] with role="tool" results

invoke() is the entry point batch runners, Temporal activities, and dataset replays use; stream() is for live UIs and tracers that want every event. Both are defined by the AgentProtocol contract, so isinstance(agent, AgentProtocol) verifies an object's shape. See the Agent Protocol reference for the full event hierarchy and the Step / InvokeResult shapes.

Tools¶

Give an agent capabilities with the @tool decorator — both standalone functions and @tool-decorated methods on a subclass (auto-discovered at construction):

from latent.agents import ReActAgent, tool

@tool
async def search_web(query: str) -> str:
    """Search the web for the given query and return results."""
    return await do_search(query)

agent = ReActAgent(name="researcher", model="gpt-4o", tools=[search_web])

See Tools & Retrievers for the full reference: schema generation, tool methods on subclasses, @retriever instrumentation, and the end_call tool.

Judges & Classifiers¶

Judge and Classifier are ReActAgent subclasses specialized for LLM-as-a-judge evaluation — structured scores, auto-generated rubrics, and rationale fields. They, along with ScoredModel, the score annotations, and the pre-built judges, have their own page:

➡️ Judges & Scoring

Agent registry¶

The @agent decorator registers a BaseAgent subclass for CLI discovery (latent agents, latent chat):

from latent.agents import agent, ReActAgent

@agent("support_bot")
class SupportBot(ReActAgent):
    """Customer support agent with knowledge base access."""

    def __init__(self):
        super().__init__(
            name="support_bot",
            model="gpt-4o",
            system_prompt="You are a customer support agent.",
        )

latent agents          # lists registered agents
latent chat support_bot  # interactive REPL

RAGAgent¶

RAGAgent extends ReActAgent with a config-driven RAG pipeline. It auto-registers a search_knowledge_base tool so the LLM can retrieve context during conversation.

from latent.agents import RAGAgent
from latent.protocol import Message

agent = RAGAgent(
    name="support",
    model="gpt-4o",
    rag_config={
        "embeddings": {"provider": "voyage", "model": "voyage-3"},
        "reranker": {"type": "cross_encoder"},
        "post_retrieval": {"type": "reorder"},
    },
)

await agent.index_documents(
    texts=["Reset your password at settings > security...", ...],
    sources=["help/passwords.md", ...],
)

response = await agent.run([Message(role="user", content="How do I reset my password?")])

Full documentation

See RAG for the complete RAG pipeline configuration reference, including chunking strategies, embedding providers, backend options, reranking, and caching.

Events¶

Every agent emits a stream of typed events (TextDelta, ToolCall, ToolResult, StepBoundary, Usage, Error, …) — the basis for logging, UI rendering, and metrics. The full event hierarchy, the Message history type, and the buffered Step / InvokeResult shapes are documented in the Agent Protocol reference.

from latent.agents import ReActAgent
from latent.protocol import Message, TextDelta

agent = ReActAgent(name="demo", model="gpt-4o")
async for event in agent.stream([Message(role="user", content="Hi")]):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)

Context-Aware Agents¶

An agent can manage its own context window with @context_check methods — they work like @guardrail but emit ContextCheckViolation events and add a tier classification for tracing. Token-budget auditing, observation masking, and auto-compaction are all available.

See Context Engineering for the @context_check decorator, the full context-scanner catalog, utility functions, offline linting with review_context(), and production architecture recommendations.