jeevesagent.architecture

Architecture layer.

An Architecture is a strategy for driving the agent loop. The canonical default is ReAct — observe / think / act in a tight loop. Other architectures (Plan-and-Execute, Reflexion, Self-Refine, Tree of Thoughts, Supervisor, Router, …) plug into the same Agent by satisfying the Architecture protocol.

See Subagent.md in the repo root for the full architecture catalogue and design rationale.

Public surface:

• Architecture — the protocol architectures implement

• AgentSession — mutable per-run state

• Dependencies — bundled protocol implementations

• ReAct — the canonical default (observe / think / act)

• resolve_architecture() — string -> Architecture instance

Submodules

• jeevesagent.architecture.actor_critic
• jeevesagent.architecture.base
• jeevesagent.architecture.blackboard
• jeevesagent.architecture.debate
• jeevesagent.architecture.helpers
• jeevesagent.architecture.plan_and_execute
• jeevesagent.architecture.react
• jeevesagent.architecture.reflexion
• jeevesagent.architecture.resolver
• jeevesagent.architecture.rewoo
• jeevesagent.architecture.router
• jeevesagent.architecture.self_refine
• jeevesagent.architecture.supervisor
• jeevesagent.architecture.swarm
• jeevesagent.architecture.tool_host_wrappers
• jeevesagent.architecture.tree_of_thoughts

Classes

ActorCritic	Actor + adversarial critic with optional different models.
AgentSession	Mutable per-run state shared between Agent and an
Architecture	Strategy interface for driving the agent loop.
Blackboard	Public + per-agent private state for the architecture.
BlackboardArchitecture	Coordinator + agents + decider, mediated by a shared
BlackboardEntry	One contribution on the blackboard.
Dependencies	Bundled protocol implementations passed to every architecture.
Handoff	Per-peer handoff configuration.
MultiAgentDebate	N debaters + optional judge orchestration.
Plan	A list of plan steps in execution order.
PlanAndExecute	Planner → step executor → synthesizer.
PlanStep	One step of a plan.
ReAct	Observe-think-act in a tight loop.
ReWOO	Plan-then-tool-execute with placeholder substitution.
ReWOOPlan	A list of ReWOO steps (no required ordering — dependencies
ReWOOStep	One step of a ReWOO plan: id + tool + args.
ReWOOStepResult	!!! abstract "Usage Documentation"
Reflexion	Wrap a base architecture with evaluator + reflector + lesson
Router	Classify input → dispatch to ONE specialist Agent.
RouterRoute	One specialist + classification metadata.
SelfRefine	Wrap a base architecture with iterative critique / refine.
StepResult	The output of executing one step.
Supervisor	Coordinator + workers, glued by a delegate tool.
Swarm	Peer agents passing control through handoff tools.
ThoughtNode	One node in the Tree-of-Thoughts search tree.
TreeOfThoughts	Branch + evaluate + prune. BFS beam search over thoughts.

Functions

resolve_architecture(...)

Coerce spec to a concrete Architecture.

Package Contents

class jeevesagent.architecture.ActorCritic(*, actor: jeevesagent.agent.api.Agent, critic: jeevesagent.agent.api.Agent, max_rounds: int = 3, approval_threshold: float = 0.9, critique_template: str | None = None, refine_template: str | None = None)

Actor + adversarial critic with optional different models.

Constructor parameters:

• actor (required): the generating Agent. Sees the original prompt on round 0 and a refine prompt on subsequent rounds.

• critic (required): the reviewing Agent. Sees the original prompt + the actor’s current output and produces structured JSON critique.

• max_rounds: cap on critique-refine cycles after the initial generation. Default 3.

• approval_threshold: terminate when critique.score is at or above this value. Default 0.9.

• critique_template / refine_template: override the default prompts. Templates use {prompt}, {output}, {critique}, {issues_bulleted}.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'actor-critic'

class jeevesagent.architecture.AgentSession

Mutable per-run state shared between Agent and an Architecture.

The Agent constructs this once per run, the architecture mutates it as iteration progresses, and the Agent reads the final state to build a RunResult.

metadata is a free-form dict architectures use for things that don’t deserve their own field — multi-agent architectures stash worker handoff state, planners stash plans, etc.

cumulative_usage: jeevesagent.core.types.Usage

id: str

instructions: str

interrupted: bool = False

interruption_reason: str | None = None

messages: list[jeevesagent.core.types.Message] = []

metadata: dict[str, Any]

output: str = ''

turns: int = 0

class jeevesagent.architecture.Architecture

Bases: Protocol

Strategy interface for driving the agent loop.

Implementations are async generators: they yield Event values for every milestone they want surfaced (model chunks, tool calls, tool results, budget warnings, errors, architecture-specific progress events).

See Subagent.md for the catalogue of architectures and the design rationale behind the protocol shape.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

Sub-Agents this architecture composes, keyed by role name.

Used by multi-agent architectures (Supervisor, Actor-Critic, Debate, Router, Blackboard, Swarm) to expose their workers for introspection (logging, telemetry, eval). Single-agent architectures return {}.

run(session: AgentSession, deps: Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

Drive iteration; yield events as they happen.

The architecture mutates session (turns, output, cumulative_usage, messages, interrupted, interruption_reason, metadata) as it iterates and yields Events for the caller to forward (or ignore, in non-streaming runs).

Implementations are async generators — declared async def run(...) -> AsyncIterator[Event]: with yield statements in the body.

name: str

class jeevesagent.architecture.Blackboard

Public + per-agent private state for the architecture.

post(author: str, content: str, *, kind: str = 'contribution', private_to: str | None = None) → BlackboardEntry

render_for(agent_name: str) → str

Format the blackboard state as a string for agent_name.

Includes every public entry and the agent’s own private scratchpad if any.

private: dict[str, list[BlackboardEntry]]

public: list[BlackboardEntry] = []

class jeevesagent.architecture.BlackboardArchitecture(*, agents: dict[str, jeevesagent.agent.api.Agent], coordinator: jeevesagent.agent.api.Agent | None = None, decider: jeevesagent.agent.api.Agent | None = None, max_rounds: int = 10, coordinator_instructions: str | None = None, decider_instructions: str | None = None)

Coordinator + agents + decider, mediated by a shared blackboard.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'blackboard'

class jeevesagent.architecture.BlackboardEntry

One contribution on the blackboard.

author: str

content: str

kind: str = 'contribution'

timestamp: datetime.datetime

class jeevesagent.architecture.Dependencies

Bundled protocol implementations passed to every architecture.

Constructed once per run from the Agent’s configured backends. Architectures treat this as read-only — they call methods on the contained protocols but don’t mutate the struct itself.

Multi-agent architectures (Supervisor, Router, etc.) will grow helper methods on this class — fresh_session, scope_for_worker, with_extra_tools, spawn_child — as they land in v0.5+. v0.3 keeps it as a passive struct.

audit_log: jeevesagent.security.audit.AuditLog | None

budget: jeevesagent.core.protocols.Budget

context: jeevesagent.core.context.RunContext

Typed scope for the run — user_id (memory namespace), session_id (conversation thread), run_id (this specific invocation), and metadata (free-form app context). See RunContext for the per-field semantics.

fast_audit: bool = True

Skip _audit(...) calls when audit_log is None.

fast_budget: bool = True

Skip budget.allows_step() and budget.consume(...) when budget is NoBudget.

fast_hooks: bool = True

Skip hooks.pre_tool / hooks.post_tool dispatch when no hooks have been registered.

fast_permissions: bool = True

Skip per-tool permissions.check(...) when permissions is the no-op AllowAll.

fast_runtime: bool = True

Inline await fn(*args) (skipping runtime.step(...) wrapping + idempotency-key derivation) when runtime is InProcRuntime.

fast_telemetry: bool = True

Skip telemetry.trace(...) contextmanagers + emit_metric calls when telemetry is NoTelemetry.

hooks: jeevesagent.security.hooks.HookRegistry

max_turns: int

memory: jeevesagent.core.protocols.Memory

model: jeevesagent.core.protocols.Model

permissions: jeevesagent.core.protocols.Permissions

runtime: jeevesagent.core.protocols.Runtime

streaming: bool = False

Whether a downstream consumer is reading from agent.stream(). When True, architectures should preserve real-time event-arrival semantics so a consumer that breaks out of the iterator triggers prompt cancellation. When False (the default for agent.run()), architectures may batch events for fewer task-group / channel allocations on the hot path.

telemetry: jeevesagent.core.protocols.Telemetry

tools: jeevesagent.core.protocols.ToolHost

class jeevesagent.architecture.Handoff

Per-peer handoff configuration.

• agent — the peer Agent.

• input_type — optional Pydantic model. When set, the generated handoff tool’s input schema mirrors this model’s fields, so the calling model gets a typed schema (instead of a string message). The validated payload is exposed to input_filter and surfaces in the swarm.handoff event.

• input_filter — optional callback (history, payload) → prompt for selective context forwarding. Default behavior respects the Swarm’s pass_full_history flag.

• description — override the generated tool’s description. Useful when the agent’s name is opaque (“billing_v2”) but the description should be user-friendly.

• tool_name — override the auto-generated tool name. Default is "transfer_to_<key>" where <key> is the peer’s key in the swarm’s agents dict.

agent: jeevesagent.agent.api.Agent

description: str | None = None

input_filter: InputFilter | None = None

input_type: type[pydantic.BaseModel] | None = None

tool_name: str | None = None

class jeevesagent.architecture.MultiAgentDebate(*, debaters: list[jeevesagent.agent.api.Agent], judge: jeevesagent.agent.api.Agent | None = None, rounds: int = 2, convergence_check: bool = True, convergence_similarity: float = 0.85, debater_instructions: str | None = None, judge_instructions: str | None = None)

N debaters + optional judge orchestration.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'debate'

class jeevesagent.architecture.Plan(/, **data: Any)

Bases: pydantic.BaseModel

A list of plan steps in execution order.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

steps: list[PlanStep] = None

class jeevesagent.architecture.PlanAndExecute(*, max_steps: int = 8, planner_prompt: str | None = None, executor_prompt: str | None = None, synthesizer_prompt: str | None = None)

Planner → step executor → synthesizer.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'plan-and-execute'

class jeevesagent.architecture.PlanStep(/, **data: Any)

Bases: pydantic.BaseModel

One step of a plan.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

description: str

id: str

class jeevesagent.architecture.ReAct(*, max_turns: int | None = None)

Observe-think-act in a tight loop.

The default architecture for every Agent. Other architectures wrap or replace this strategy; see Subagent.md.

max_turns overrides Dependencies.max_turns for this architecture only — useful when wrapping ReAct inside another architecture that sets its own per-leaf cap (Reflexion, Plan-and-Execute, etc.). None means “use whatever the Agent was configured with”.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'react'

class jeevesagent.architecture.ReWOO(*, max_steps: int = 8, planner_prompt: str | None = None, solver_prompt: str | None = None, parallel_levels: bool = True)

Plan-then-tool-execute with placeholder substitution.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'rewoo'

class jeevesagent.architecture.ReWOOPlan(/, **data: Any)

Bases: pydantic.BaseModel

A list of ReWOO steps (no required ordering — dependencies are inferred from {{En}} placeholders).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

steps: list[ReWOOStep] = None

class jeevesagent.architecture.ReWOOStep(/, **data: Any)

Bases: pydantic.BaseModel

One step of a ReWOO plan: id + tool + args.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

args: dict[str, Any] = None

property depends_on: list[str]

Extract {{En}} step ids referenced in args.

id: str

tool: str

class jeevesagent.architecture.ReWOOStepResult(/, **data: Any)

Bases: pydantic.BaseModel

!!! abstract “Usage Documentation”

[Models](../concepts/models.md)

A base class for creating Pydantic models.

__class_vars__

The names of the class variables defined on the model.

__private_attributes__

Metadata about the private attributes of the model.

__signature__

The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__

Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__

The core schema of the model.

__pydantic_custom_init__

Whether the model has a custom __init__ function.

__pydantic_decorators__

Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__

Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__

Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__

The name of the post-init method for the model, if defined.

__pydantic_root_model__

Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__

The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__

The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__

A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__

A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__

A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to ‘allow’.

__pydantic_fields_set__

The names of fields explicitly set during instantiation.

__pydantic_private__

Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

error: str | None = None

output: str

step_id: str

tool: str

class jeevesagent.architecture.Reflexion(*, base: jeevesagent.architecture.base.Architecture | None = None, max_attempts: int = 3, threshold: float = 0.8, evaluator_prompt: str | None = None, reflector_prompt: str | None = None, lessons_block_name: str = 'reflexion_lessons', lesson_store: jeevesagent.vectorstore.base.VectorStore | None = None, top_k_lessons: int = 5)

Wrap a base architecture with evaluator + reflector + lesson memory.

See module docstring for the full mechanism. Constructor parameters:

• base — architecture to retry. Default ReAct.

• max_attempts — cap on retries within a single run. Default 3.

• threshold — minimum evaluator score to terminate as success. Default 0.8.

• evaluator_prompt / reflector_prompt — override the default system prompts.

• lessons_block_name — memory working-block name for persisted lessons. Default "reflexion_lessons". Multiple Reflexion-wrapped agents in the same memory should pick distinct names.

• lesson_store — optional VectorStore enabling selective recall. When set, lessons are stored as embedded chunks and only the top-top_k_lessons most relevant lessons are surfaced on each attempt (instead of all past lessons). Avoids context bloat as lessons accumulate.

• top_k_lessons — how many lessons to recall per attempt (selective-recall mode only). Default 5.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'reflexion'

class jeevesagent.architecture.Router(*, routes: list[RouterRoute], fallback_route: str | None = None, require_confidence_above: float = 0.0, classifier_prompt: str | None = None)

Classify input → dispatch to ONE specialist Agent.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'router'

class jeevesagent.architecture.RouterRoute

One specialist + classification metadata.

name is what the classifier emits in its route: line and must be unique within a Router. description is shown to the classifier alongside the name — keep it specific and distinguishing so the classifier picks reliably.

agent: jeevesagent.agent.api.Agent

description: str = ''

name: str

class jeevesagent.architecture.SelfRefine(*, base: jeevesagent.architecture.base.Architecture | None = None, max_rounds: int = 3, critic_prompt: str | None = None, refiner_prompt: str | None = None, stop_phrase: str = 'no issues')

Wrap a base architecture with iterative critique / refine.

base defaults to ReAct; the round-0 generator runs the base architecture’s full strategy. Subsequent rounds are text-only model calls — no tools, just critique and rewrite.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'self-refine'

class jeevesagent.architecture.StepResult(/, **data: Any)

Bases: pydantic.BaseModel

The output of executing one step.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

description: str

output: str

step_id: str

class jeevesagent.architecture.Supervisor(*, workers: dict[str, jeevesagent.agent.api.Agent], base: jeevesagent.architecture.base.Architecture | None = None, instructions_template: str | None = None, delegate_tool_name: str = 'delegate', forward_tool_name: str = 'forward_message')

Coordinator + workers, glued by a delegate tool.

The supervisor’s base architecture (default ReAct) sees a fresh delegate(worker, instructions) tool that routes calls to the named worker Agent. Worker outputs come back as tool results just like any other tool call.

Constructor

• workers: dict mapping role-names to fully-built Agent instances. Names must be valid identifiers (the model emits them as the worker argument).

• base: the architecture the supervisor itself runs. Default ReAct. Wrap inside Reflexion to learn delegation patterns across runs.

• instructions_template: format string with {worker_descriptions}. Default teaches the supervisor to delegate effectively. The agent’s own instructions are prepended (so domain context survives).

• delegate_tool_name: defaults to "delegate". Customize to avoid clashes with user-defined tools that happen to have the same name.

• forward_tool_name: defaults to "forward_message". The supervisor calls this with a worker name to return that worker’s last output VERBATIM as the supervisor’s final response. Skips a synthesis round-trip — the `langchain.com/blog/benchmarking-multi-agent-architectures`_ benchmark showed +50% quality on tasks where the supervisor would otherwise paraphrase a worker’s output.

add_worker(name: str, agent: jeevesagent.agent.api.Agent) → None

Register a worker between runs.

Safe to call between Agent.run() invocations on the agent that owns this supervisor; the new worker becomes available for delegate(name, ...) on the next run. Calling mid-run is undefined — the supervisor’s prompt is composed at run start.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

remove_worker(name: str) → jeevesagent.agent.api.Agent | None

Unregister a worker by name. Returns the removed Agent if it was registered, None otherwise. Same lifecycle rules as add_worker().

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'supervisor'

class jeevesagent.architecture.Swarm(*, agents: dict[str, jeevesagent.agent.api.Agent | Handoff], entry_agent: str, max_handoffs: int = 8, detect_cycles: bool = True, pass_full_history: bool = True, handoff_tool_name: str = 'handoff')

Peer agents passing control through handoff tools.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'swarm'

class jeevesagent.architecture.ThoughtNode(/, **data: Any)

Bases: pydantic.BaseModel

One node in the Tree-of-Thoughts search tree.

Children are stored implicitly (each node has a parent_id). The full tree is reconstructable from the node list ToT keeps in its session metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

content: str

depth: int

id: str

parent_id: str | None

score: float = 0.0

class jeevesagent.architecture.TreeOfThoughts(*, branch_factor: int = 3, max_depth: int = 3, beam_width: int = 2, solved_threshold: float = 1.0, min_score: float = 0.0, parallel: bool = True, proposer_prompt: str | None = None, evaluator_prompt: str | None = None)

Branch + evaluate + prune. BFS beam search over thoughts.

declared_workers() → dict[str, jeevesagent.agent.api.Agent]

async run(session: jeevesagent.architecture.base.AgentSession, deps: jeevesagent.architecture.base.Dependencies, prompt: str) → collections.abc.AsyncIterator[jeevesagent.core.types.Event]

name = 'tree-of-thoughts'

jeevesagent.architecture.resolve_architecture(spec: jeevesagent.architecture.base.Architecture | str | None) → jeevesagent.architecture.base.Architecture

Coerce spec to a concrete Architecture.

• None → ReAct (the default)

• str → looked up in KNOWN (only "react" in v0.3)

• Architecture instance → returned as-is

Unknown strings raise ConfigError with a list of known names.