latent.agents.lifecycle¶

Agent lifecycle hooks and stream orchestration helpers.

Classes¶

LifecyclePhase¶

LifecyclePhase()

LifecycleRunner¶

LifecycleRunner(agent: Any, before: list[_LifecycleHook], after: list[_LifecycleHook])

Runs lifecycle hooks for an agent without wrapping stream().

TurnContext¶

TurnContext(agent: Any, messages: list[Message], events: list[AgentEvent])

Outcome context passed to after_turn lifecycle hooks.

TurnSelector¶

TurnSelector()

Functions¶

apply_system_suffixes¶

apply_system_suffixes(content: str | SystemContent | list[dict[str, Any]], suffixes: list[str] | tuple[str, ...] | None) -> str | SystemContent | list[dict[str, Any]]

Append lifecycle system suffixes without modifying cached static text.

discover_lifecycle¶

discover_lifecycle(instance: Any) -> tuple[list[_LifecycleHook], list[_LifecycleHook]]

Discover decorated lifecycle hooks.

inject_prefix¶

inject_prefix(events: AsyncIterator[AgentEvent], prefix: str) -> AsyncIterator[AgentEvent]

Inject prefix before the first user-visible text step.

Tool-calling steps are buffered and skipped so the prefix lands on the answer text. If the turn emits no text, the prefix is flushed as its own final text step.

is_terminal_turn¶

is_terminal_turn(ctx: TurnContext) -> bool

Return True when a configured terminal tool fired in the turn.

lifecycle¶

lifecycle(event: LifecycleEvent, when: Callable[..., bool] | None = None, turn: TurnFilter | list[TurnFilter] | tuple[TurnFilter, ...] | None = None, first_turn: bool = False, terminal_turn: bool = False, priority: int = 0) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Mark an agent method as a lifecycle hook.

before_turn hooks receive the input messages and run before the agent stream starts. after_turn hooks receive a TurnContext after the stream completes. LIFECYCLE.BEFORE_TURN / LIFECYCLE.AFTER_TURN are available to avoid hand-typed event strings.

Use first_turn=True or turn=TURN.FIRST to restrict either phase to the first user turn. Use terminal_turn=True or turn=TURN.TERMINAL to restrict an after_turn hook to turns where a configured terminal tool fired. Terminal status is only knowable after the stream has produced tool calls, so terminal-turn filters are not valid for before_turn hooks.

responder¶

responder(fn: Callable[..., Any] | None = None, when: BeforePredicate | None = None, priority: int = 0) -> Callable[..., Any]

Sugar for a before_turn hook that returns reply text.

Methods¶

LifecycleRunner.build_context¶

build_context(messages: list[Message], events: list[AgentEvent]) -> TurnContext

LifecycleRunner.record_reply¶

record_reply(reply: Reply, hook: _LifecycleHook | None) -> None

LifecycleRunner.run_after¶

run_after(ctx: TurnContext) -> None

LifecycleRunner.run_before¶

run_before(messages: list[Message]) -> _TurnActions

TurnContext.called¶

called(name: str) -> bool