Agent Cookbook

Real-world, class-based recipes for every agent type — distilled from production agents built on latent. Each is a faithful skeleton (sanitized of business logic); follow the link for the full reference. All entry points are async.

Recipe	Pattern
Tool-using ReAct agent	ReActAgent subclass + @tool methods + @agent
Tools: the factory pattern	closure-bound dependencies
Retrievers: @retriever for KB search	RAG with retrieval span events
RAG agent	config-driven retrieval pipeline
Guardrails: the mixin pattern	reusable @guardrail methods
Guided agent	guidelines + journeys
Pipeline agent	deterministic phases
LLM-as-judge	structured scoring
Simulated human	conversation simulation

Tool-using ReAct agent

A ReActAgent subclass. @tool and @guardrail methods are auto-discovered at construction. @agent registers it for latent agents / latent chat.

from latent.agents import ReActAgent, agent, tool

@agent("support", description="Customer support assistant")
class SupportAgent(ReActAgent):
    def __init__(self, *, model: str = "gpt-4o", **kwargs):
        super().__init__(
            name="support",
            model=model,
            system_prompt="You are a helpful support agent.",
            temperature=0.0,
            max_tokens=4096,
            **kwargs,
        )

    @tool
    async def lookup_order(self, order_id: str) -> str:
        """Look up an order by its ID."""
        return await db.fetch_order(order_id)

latent agents            # lists "support"
latent chat support      # interactive REPL

→ ReAct Agent · Tools & Retrievers

Tools: the factory pattern

Production agents usually build tools from a factory that closure-binds dependencies (a retriever, a customer id, a config) so the tool can't reach outside its scope. Pass the resulting list to tools=[...].

from latent.agents import ReActAgent, tool

def make_kb_tool(retriever, *, k: int = 5):
    @tool
    async def search_knowledge_base(queries: list[str]) -> str:
        """Search the knowledge base. Pass several phrasings to improve recall."""
        results = await retriever.search(queries[0], k=k)
        return "\n\n".join(f"[{c.source}] {c.content}" for c in results)
    return search_knowledge_base

agent = ReActAgent(name="kb", model="gpt-4o", tools=[make_kb_tool(my_retriever)])

Docstrings are the tool description

A @tool function must have a docstring (it becomes what the model sees) and type-annotated parameters (they generate the JSON schema). Tools must be async def.

→ Tools & Retrievers

Retrievers: @retriever for KB search

@retriever is @tool plus observability: it registers the function as an LLM-callable tool and emits RetrieverStart / RetrieverEnd span events around the whole call (vector lookup + filtering + formatting), so retrieval shows up in traces and eval harnesses. Use it for the read side of RAG.

from latent.agents import ReActAgent, retriever
from latent.rag import ChromaRetriever, RetrievedChunk

kb = ChromaRetriever(
    collection_name="support-kb",
    persist_directory="./data/chroma_db",
    embedding_provider="litellm",
    embedding_model="gemini/gemini-embedding-001",
)

@retriever  # implies @tool — LLM-callable AND emits RetrieverStart/RetrieverEnd
async def search_support_kb(query: str) -> str:
    """Search the support docs for content relevant to the question."""
    chunks: list[RetrievedChunk] = await kb.search(query, k=15)
    chunks = [c for c in chunks if c.score >= 0.4][:5]
    if not chunks:
        return "No relevant documentation found."
    return "\n\n---\n\n".join(
        f"[{c.metadata.get('title') or c.source}] (score={c.score:.2f})\n{c.content}"
        for c in chunks
    )

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

RetrievedChunk carries content, source, section_path, score, and a metadata dict. A @retriever must be a plain async def (not an async generator).

→ Tools & Retrievers · RAG Pipelines

RAG agent

When you want the whole retrieval pipeline wired for you (chunking, embeddings, reranking) rather than a hand-rolled retriever, subclass RAGAgent — it auto-registers a search_knowledge_base tool.

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"}},
)
await agent.index_documents(texts=[...], sources=[...])
answer = await agent.run([Message(role="user", content="How do I reset my password?")])

→ RAG Pipelines

Guardrails: the mixin pattern

Bundle safety rules into a reusable mixin and inherit it across agents. @guardrail factory methods (signature (self)) return a scanner; on_block is an async callback that can rewrite the block message. Pass skip_guardrails=True to disable them (handy for ablations).

from latent.agents import ReActAgent
from latent.guardrails import guardrail
from latent.guardrails.base import GuardrailBlockContext
from latent.guardrails.scanners.builtin import LanguageScanner, TokenLimitScanner
from latent.guardrails.scanners.llmguard import AnonymizeScanner   # needs [guardrails-llmguard]

async def _soft_block(ctx: GuardrailBlockContext) -> str:
    # ctx.rule_name, ctx.score, ctx.prompt, ctx.message, ctx.metadata, ctx.agent
    return ctx.message

class GuardrailsMixin:
    @guardrail(timing="pre", outcome="active", message="We support English only.")
    def language(self):
        return LanguageScanner(allowed_languages=["en"], min_length=30)

    @guardrail(timing="pre", outcome="active", message="That request is too long.")
    def token_cap(self):
        return TokenLimitScanner(max_tokens=8000)

    @guardrail(timing="pre", outcome="active", on_block=_soft_block,
               message="Credit cards are redacted for your security.")
    def redact_pii(self):
        return AnonymizeScanner(entity_types=["CREDIT_CARD"])

class SupportAgent(GuardrailsMixin, ReActAgent):
    def __init__(self, **kwargs):
        super().__init__(name="support", model="gpt-4o", tools=[search_support_kb], **kwargs)

→ Guardrails · Scanners

Guided agent

A GuidedAgent injects only the guidelines whose when condition matches the turn, gates tools per guideline, and can walk a multi-step journey. (Experimental.)

from latent.agents import GuidedAgent
from latent.agents.guided import Guideline, Journey, Step

GUIDELINES = [
    Guideline(
        id="upgrade_eligible_only",
        when="the customer asks to upgrade or change plans",
        then="Call get_current_user first, then search the KB; offer only eligible tiers.",
        priority=70,
        tools=("get_current_user",),     # unlocked only when this guideline is active
    ),
]

JOURNEY = Journey(name="support", steps=[
    Step(
        name="anchor_identity",
        instructions="Call get_current_user when the user's tier could affect the answer.",
        advance_when="the authenticated user and tier are known",
    ),
])

agent = GuidedAgent(
    name="concierge",
    model="gpt-4o",
    system_prompt="You are a concierge.",
    guidelines=GUIDELINES,
    journeys=[JOURNEY],
    match_strategy="embedding",          # "always" | "batch_llm" | "embedding"
    embedding_model="gemini/gemini-embedding-001",
    match_threshold=0.55,
    mode="fluid",
    tools=[search_support_kb, get_current_user],
)

→ Guided Agents

Pipeline agent

A PipelineAgent where code decides the routing. @phase (LLM step, optional Pydantic output_schema) and @subagent (bounded ReAct loop) methods return a string and may be sync; @tool / @respond methods must be async def. transitions are lambda output, state: bool predicates — first truthy wins, None is the default edge.

from typing import Literal
from pydantic import BaseModel, Field
from latent.agents.pipeline import PipelineAgent, PipelineState, phase, subagent, respond
from latent.protocol import Message

class Triage(BaseModel):
    action: Literal["answer", "escalate", "clarify"]
    queries: list[str] = Field(default_factory=list)

class SupportPipeline(PipelineAgent):
    @phase(output_schema=Triage, transitions={
        "clarify": lambda o, s: o.action == "clarify",
        "escalate": lambda o, s: o.action == "escalate",
        "retrieve": None,                            # default next phase
    })
    def classify(self, state: PipelineState, messages: list[Message]) -> str:
        return "Classify the request as answer / escalate / clarify (JSON per schema)."

    @subagent(tools=[search_support_kb], max_iterations=3, transitions={"draft": None})
    def retrieve(self, state: PipelineState, messages: list[Message]) -> str:
        return f"Search the KB for: {', '.join(state.outputs['classify'].queries)}"

    @phase(transitions={"answer": None})
    def draft(self, state: PipelineState, messages: list[Message]) -> str:
        return f"Context:\n{state.outputs['retrieve']}\n\nDraft a concise reply."

    @respond(model="gpt-4o")
    async def answer(self, state: PipelineState, messages: list[Message]) -> str:
        return f"Polish this reply for a warm, professional tone:\n{state.outputs['draft']}"

→ Pipeline Agents

LLM-as-judge

A Judge returns a typed, validated score object built from a ScoredModel.

from typing import Annotated
from latent.agents import Judge
from latent.agents.scores import ScoredModel, OrdinalScore

class Quality(ScoredModel):
    quality: Annotated[int, OrdinalScore(scale=(1, 2, 3, 4, 5), pass_threshold=3)]

judge = Judge("quality", model="gpt-4o", output_type=Quality)
score = await judge.evaluate({"question": "...", "answer": "..."})

→ Judges & Scoring

Simulated human

A HumanAgent plays the user side in conversation simulations.

from latent.simulators import HumanAgent, ConversationSimulator

human = HumanAgent(
    goals=["book a flight to NYC", "ask about baggage fees"],
    model="gpt-4o-mini",
)
sim = ConversationSimulator(agent_under_test=SupportAgent(), human_agent=human)
result = await sim.simulate(scenario_context="The user opens a support chat.")

→ Simulators

---

See Also

• Building Agents — the feature map for the whole stack
• ReAct Agent · Tools & Retrievers · Guardrails
• Agent Protocol — the contract every class above satisfies