jeevesagent.team
================

.. py:module:: jeevesagent.team

.. autoapi-nested-parse::

   Team — sibling-style builders for multi-agent architectures.

   Coming from LangGraph, CrewAI, AutoGen, or the OpenAI Agents SDK,
   you'd expect to construct a multi-agent team like this::

       team = create_supervisor([researcher, writer], model="gpt-4")
       result = await team.invoke(...)

   In JeevesAgent the same shape is :class:`Team`::

       team = Team.supervisor(
           workers={"researcher": researcher, "writer": writer},
           instructions="manage the pipeline",
           model="gpt-4.1-mini",
       )
       result = await team.run("write me a blog post")

   Under the hood :class:`Team` returns a regular :class:`Agent` —
   exactly what ``Agent(architecture=Supervisor(...))`` would produce.
   The two shapes are interchangeable; :class:`Team` is the
   "familiar-from-other-frameworks" facade for the common case, while
   the nested ``Agent(architecture=...)`` form remains the path for
   **recursive composition** (wrapping a Supervisor in Reflexion, etc.).

   Provided builders
   -----------------

   * :meth:`Team.supervisor` — coordinator + workers
   * :meth:`Team.swarm` — peer agents handing off control
   * :meth:`Team.router` — classify-and-dispatch
   * :meth:`Team.debate` — N debaters + optional judge
   * :meth:`Team.actor_critic` — actor + critic pair
   * :meth:`Team.blackboard` — shared workspace + coordinator

   Each method exposes **every** :class:`Agent` constructor kwarg
   explicitly (rather than forwarding through ``**kwargs``) so that
   IDEs and type-checkers surface the full parameter list with proper
   types, defaults, and docstrings on hover.

   Standalone-run helper
   ---------------------

   :func:`run_architecture` builds a minimal :class:`Agent` shell
   around any :class:`Architecture` instance and runs it. Useful for
   **testing orchestrators in isolation** without wiring a full Agent
   yourself::

       sup = Supervisor(workers={"a": agent_a})
       result = await run_architecture(sup, "do the thing", model="gpt-4.1-mini")



Classes
-------

.. autoapisummary::

   jeevesagent.team.Team


Functions
---------

.. autoapisummary::

   jeevesagent.team.run_architecture


Module Contents
---------------

.. py:class:: Team

   Namespace for multi-agent team builders.

   Every classmethod returns a fully-built :class:`Agent` whose
   architecture is the corresponding multi-agent strategy. The
   returned Agent has the standard ``run`` / ``stream`` / etc.
   interface — call sites don't change between single-agent and
   team agents.


   .. py:method:: actor_critic(actor: jeevesagent.agent.api.Agent, critic: jeevesagent.agent.api.Agent, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_rounds: int = 3, approval_threshold: float = 0.9, critique_template: str | None = None, refine_template: str | None = None) -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build an actor-critic pair where the critic reviews the
      actor's output (with structured JSON scoring + rubric) and
      the actor refines below ``approval_threshold``.



   .. py:method:: blackboard(agents: dict[str, jeevesagent.agent.api.Agent], *, coordinator: jeevesagent.agent.api.Agent | None = None, decider: jeevesagent.agent.api.Agent | None = None, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_rounds: int = 10, coordinator_instructions: str | None = None, decider_instructions: str | None = None) -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build a blackboard team where ``agents`` collaborate via
      a shared workspace; an optional ``coordinator`` selects who
      acts each round and an optional ``decider`` decides when
      the work is done.



   .. py:method:: debate(debaters: list[jeevesagent.agent.api.Agent], *, judge: jeevesagent.agent.api.Agent | None = None, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, rounds: int = 2, convergence_check: bool = True, convergence_similarity: float = 0.85, debater_instructions: str | None = None, judge_instructions: str | None = None) -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build a multi-agent debate where ``debaters`` argue for
      ``rounds`` (with optional convergence early-exit). If
      ``judge`` is provided, the judge synthesizes a final
      answer; otherwise majority vote wins.



   .. py:method:: router(routes: list[jeevesagent.architecture.RouterRoute], *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, fallback_route: str | None = None, require_confidence_above: float = 0.0, classifier_prompt: str | None = None) -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build a router that classifies once and dispatches to
      ONE specialist :class:`Agent`. Cheaper than Supervisor for
      tasks with clear specialist boundaries (one classifier call
      + one specialist run, no synthesis pass).



   .. py:method:: supervisor(workers: dict[str, jeevesagent.agent.api.Agent], *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, instructions_template: str | None = None, delegate_tool_name: str = 'delegate', forward_tool_name: str = 'forward_message') -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build a coordinator Agent that delegates to ``workers``.

      The coordinator can call ``delegate(worker, instructions)``
      to dispatch a subtask, or ``forward_message(worker)`` to
      return a worker's output verbatim. Multiple delegations in
      one turn run in parallel.



   .. py:method:: swarm(agents: dict[str, jeevesagent.agent.api.Agent | jeevesagent.architecture.swarm.Handoff], entry_agent: str, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False, skills: list[Any] | None = None, max_handoffs: int = 8, detect_cycles: bool = True, pass_full_history: bool = True, handoff_tool_name: str = 'handoff') -> jeevesagent.agent.api.Agent
      :staticmethod:


      Build a peer-swarm of agents that hand off control via a
      ``handoff`` tool (or per-target ``transfer_to_<name>`` tools
      when peers are wrapped in :class:`Handoff` with an
      ``input_type``).

      ``entry_agent`` is the peer that receives the first message.



.. py:function:: run_architecture(architecture: jeevesagent.architecture.Architecture, prompt: str, *, instructions: str = '', model: jeevesagent.core.protocols.Model | str | None = None, memory: jeevesagent.core.protocols.Memory | None = None, runtime: jeevesagent.core.protocols.Runtime | None = None, budget: jeevesagent.core.protocols.Budget | None = None, permissions: jeevesagent.core.protocols.Permissions | None = None, hooks: jeevesagent.security.hooks.HookRegistry | None = None, tools: list[jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object]] | jeevesagent.core.protocols.ToolHost | jeevesagent.tools.registry.Tool | collections.abc.Callable[Ellipsis, object] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, audit_log: jeevesagent.security.audit.AuditLog | None = None, max_turns: int = DEFAULT_MAX_TURNS, auto_consolidate: bool = False) -> jeevesagent.core.types.RunResult
   :async:


   Run an :class:`Architecture` once with a minimal Agent shell.

   Useful for testing orchestrators in isolation or for one-shot
   scripts where you don't want to construct an Agent yourself.

   The default ``model`` is the framework's resolver default (set
   via ``model=`` or env / config); pass an explicit model or
   string id to override.

   Example::

       sup = Supervisor(workers={"a": agent_a})
       result = await run_architecture(
           sup, "do the thing", model="gpt-4.1-mini"
       )


