latent.agents.base¶

BaseAgent -- conversational agent matching agent-studio AgentProtocol (FR-1.1).

Classes¶

BaseAgent¶

BaseAgent(name: str, skip_guardrails: bool = False, lifecycle_before_hooks: list[Any] | None = None, lifecycle_after_hooks: list[Any] | None = None, kwargs: Any = {})

General-purpose conversational agent.

Matches the agent-studio AgentProtocol interface so any BaseAgent can be used as an agent-studio backend.

Generic in TContext — the type of the per-invocation :attr:context. Context-less agents subclass bare (TContext defaults to None); agents that take typed context bind it (class MyAgent(ReActAgent[MyContext])) so self.context reads as MyContext | None inside hooks and tools.

Methods¶

BaseAgent.invoke¶

invoke(messages: list[Message], stream_wrapper: StreamWrapper | None = None, config: dict[str, Any] | None = None) -> InvokeResult

Buffered call. Consumes stream(), returns per-step structure.

result.text is the terminal step's text. Derived from next((s for s in reversed(steps) if self.is_terminal_step(s)), None) post-construction — NOT from a live mutable buffer cleared on a boundary signal. PR #75 of the consuming runtime used live-clear and shipped a regression when boundary ordering didn't match expectations. Structural derivation is insulated.

stream_wrapper, if given, wraps the internal event stream before consumption. Used by external observers like agent-studio's TracedStream that emit OTEL spans as a side effect of consuming the same events invoke() consumes.

BaseAgent.is_terminal_step¶

is_terminal_step(step: Step) -> bool

Return True iff step is the customer-facing terminal step.

Default rule: a step is terminal iff it produced no tool calls. This matches ReAct loop semantics in ReActAgent — the loop exits exactly when an iteration emits no tool_calls, so the same condition identifies the answer step.

Subclasses with different loop semantics (guided / pipeline / etc.) override.

BaseAgent.name¶

Human-readable name for logging and MLflow tracking.

BaseAgent.on_session_start¶

on_session_start() -> dict[str, Any]

Return a session config snapshot for logging.

Called by ChatController when a session file is configured. Pure — no I/O. Subclasses may override to add agent-specific metadata.

BaseAgent.reset¶

reset() -> None

Reset agent state between conversations.

BaseAgent.run¶

run(messages: list[Message]) -> str

Backward-compat: returns the terminal-step text only.

Existing callers (await agent.run(messages)) see no behaviour change. Internally delegates to invoke() — the per-step structure is built and discarded.

BaseAgent.stream¶

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

Run one turn, applying lifecycle hooks around the concrete stream.

BaseAgent.tools¶

Tool definitions available to this agent.

Auto-discovers methods decorated with @tool (those carrying _tool_meta). Subclasses may override to provide a static list.

BaseAgent.turn_metadata¶

turn_metadata() -> dict[str, MetadataValue]