Skip to content

teff.flow.team

teff.flow.team

Supervised team builder for :class:~teff.flow.Flow.

:class:TeamBuilder composes a supervisor decider plus routed agents in a single call (:meth:~teff.flow.Flow.team). :class:Flow owns an instance and delegates to it.

Classes:

Name Description
TeamBuilder

Compose a supervised agent team on a :class:~teff.flow.Flow.

TeamBuilder

Compose a supervised agent team on a :class:~teff.flow.Flow.

Parameters:

Name Type Description Default
flow Flow

The owning Flow whose graph state is mutated.

 required

Methods:

Name Description
team

Compose a supervised agent team in one call.
Source code in teff/flow/team.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class TeamBuilder:
    """Compose a supervised agent team on a :class:`~teff.flow.Flow`.

    Args:
        flow: The owning ``Flow`` whose graph state is mutated.
    """

    def __init__(self, flow: "Flow"):
        self.flow = flow

    def team(
        self,
        system: str = "",
        *,
        roles: dict,
        model: str | None = None,
        provider: str | None = None,
        messages_key: str = "messages",
        sections: dict[str, str] | None = None,
        route_keys: dict[str, str] | None = None,
        done_keys: list[str] | None = None,
        done_mode: str = "all",
        fallback: str = "",
        max_rounds: int = 6,
        finish: "Node | list[Node] | None" = None,
        id: str | None = None,
    ) -> "Flow":
        """Compose a supervised agent team in one call.

        Builds a :class:`~teff.node.supervisor.Supervisor` decider plus one
        routed agent per *role* and wires the supervisor loop in a single
        step — the programmatic twin of the ``team:`` flow.yaml idiom::

            flow.team(
                "Route to planner or coder, then finish.",
                roles={
                    "planner": AgentRole("You plan.", output_key="plan"),
                    "coder": AgentRole("You code.", output_key="code"),
                },
                fallback="planner",
            )

        Each *role* value is an :class:`~teff.flow.AgentRole` (the
        recommended spelling), a plain dict recipe accepted for YAML parity
        (``{system, output_key, use_tools, ...}``), a :class:`Node` /
        :class:`~teff.flow.SubFlow` used as-is, a :class:`Flow` embedded as a
        :class:`SubFlow`, or a list of nodes run in sequence for that route.

        The leader decider inherits *model*/*provider* (or the flow's
        ``default_model``/``default_provider``).  *route_keys* defaults to
        ``{role: output_key}``; *done_keys*/*done_mode*/*fallback*/``max_rounds``
        drive the built-in safety guards (see :class:`Supervisor`).  When the
        decider replies ``finish`` the flow continues through *finish* (or
        terminates when omitted).  *id* names the supervisor node.

        Returns ``self`` for chaining.
        """
        from teff.flow.agent import AgentRole
        from teff.flow.sub_flow import SubFlow
        from teff.node.supervisor import Supervisor

        flow = self.flow
        flow._check_continuation()
        if not isinstance(roles, dict) or not roles:
            raise ValueError("team requires a non-empty `roles` mapping")
        model = model or flow._default_model
        provider = provider or flow._default_provider
        if not model or not provider:
            raise ValueError(
                "team requires `model` and `provider` (or `default_model`/"
                "`default_provider` on the flow)"
            )

        if route_keys is None:
            route_keys = {}
            for role_name, spec in roles.items():
                if isinstance(spec, AgentRole):
                    out = spec.output_key or role_name
                elif isinstance(spec, dict):
                    out = spec.get("output_key") or role_name
                else:
                    out = role_name
                route_keys[role_name] = out

        def build_role(spec):
            """Normalize one role spec into node(s) for ``route()``.

            An ``AgentRole`` / recipe dict / ``Flow`` becomes a single node
            (``SubFlow`` for agents/flow); a plain ``Node`` is used as-is; a
            list (a route chain such as agent → interrupt) is normalized
            element-wise so roles can be mixed with plain nodes.
            """
            if isinstance(spec, list):
                return [build_role(item) for item in spec]  # type: ignore[misc]
            if isinstance(spec, AgentRole):
                return spec.build(model=model, provider=provider, id=role_name)
            if isinstance(spec, dict):
                return AgentRole.from_mapping(spec, name=role_name).build(
                    model=model, provider=provider, id=role_name
                )
            if isinstance(spec, self.flow.__class__):
                return SubFlow(graph=spec.compile())
            return spec

        agents: dict = {}
        for role_name, spec in roles.items():
            if isinstance(spec, list):
                chain = build_role(spec)
                agents[role_name] = flow._as_chain(chain)
            else:
                agents[role_name] = build_role(spec)

        supervisor = Supervisor(
            system=system,
            model=model,
            provider=provider,
            messages_key=messages_key,
            sections=sections,
            route_keys=route_keys,
            done_keys=set(done_keys or ()),
            done_mode=done_mode,
            fallback_agent=fallback,
            max_rounds=max_rounds,
        )
        self.flow.supervisor(supervisor, id=id)
        self.flow.route("next_agent", finish=finish, **agents)
        return self.flow

team 

team(
    system="",
    *,
    roles,
    model=None,
    provider=None,
    messages_key="messages",
    sections=None,
    route_keys=None,
    done_keys=None,
    done_mode="all",
    fallback="",
    max_rounds=6,
    finish=None,
    id=None,
)

Compose a supervised agent team in one call.

Builds a :class:~teff.node.supervisor.Supervisor decider plus one routed agent per role and wires the supervisor loop in a single step — the programmatic twin of the team: flow.yaml idiom::

flow.team(
    "Route to planner or coder, then finish.",
    roles={
        "planner": AgentRole("You plan.", output_key="plan"),
        "coder": AgentRole("You code.", output_key="code"),
    },
    fallback="planner",
)

Each role value is an :class:~teff.flow.AgentRole (the recommended spelling), a plain dict recipe accepted for YAML parity ({system, output_key, use_tools, ...}), a :class:Node / :class:~teff.flow.SubFlow used as-is, a :class:Flow embedded as a :class:SubFlow, or a list of nodes run in sequence for that route.

The leader decider inherits model/provider (or the flow's default_model/default_provider). route_keys defaults to {role: output_key}; done_keys/done_mode/fallback/max_rounds drive the built-in safety guards (see :class:Supervisor). When the decider replies finish the flow continues through finish (or terminates when omitted). id names the supervisor node.

Returns self for chaining.

Source code in teff/flow/team.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def team(
    self,
    system: str = "",
    *,
    roles: dict,
    model: str | None = None,
    provider: str | None = None,
    messages_key: str = "messages",
    sections: dict[str, str] | None = None,
    route_keys: dict[str, str] | None = None,
    done_keys: list[str] | None = None,
    done_mode: str = "all",
    fallback: str = "",
    max_rounds: int = 6,
    finish: "Node | list[Node] | None" = None,
    id: str | None = None,
) -> "Flow":
    """Compose a supervised agent team in one call.

    Builds a :class:`~teff.node.supervisor.Supervisor` decider plus one
    routed agent per *role* and wires the supervisor loop in a single
    step — the programmatic twin of the ``team:`` flow.yaml idiom::

        flow.team(
            "Route to planner or coder, then finish.",
            roles={
                "planner": AgentRole("You plan.", output_key="plan"),
                "coder": AgentRole("You code.", output_key="code"),
            },
            fallback="planner",
        )

    Each *role* value is an :class:`~teff.flow.AgentRole` (the
    recommended spelling), a plain dict recipe accepted for YAML parity
    (``{system, output_key, use_tools, ...}``), a :class:`Node` /
    :class:`~teff.flow.SubFlow` used as-is, a :class:`Flow` embedded as a
    :class:`SubFlow`, or a list of nodes run in sequence for that route.

    The leader decider inherits *model*/*provider* (or the flow's
    ``default_model``/``default_provider``).  *route_keys* defaults to
    ``{role: output_key}``; *done_keys*/*done_mode*/*fallback*/``max_rounds``
    drive the built-in safety guards (see :class:`Supervisor`).  When the
    decider replies ``finish`` the flow continues through *finish* (or
    terminates when omitted).  *id* names the supervisor node.

    Returns ``self`` for chaining.
    """
    from teff.flow.agent import AgentRole
    from teff.flow.sub_flow import SubFlow
    from teff.node.supervisor import Supervisor

    flow = self.flow
    flow._check_continuation()
    if not isinstance(roles, dict) or not roles:
        raise ValueError("team requires a non-empty `roles` mapping")
    model = model or flow._default_model
    provider = provider or flow._default_provider
    if not model or not provider:
        raise ValueError(
            "team requires `model` and `provider` (or `default_model`/"
            "`default_provider` on the flow)"
        )

    if route_keys is None:
        route_keys = {}
        for role_name, spec in roles.items():
            if isinstance(spec, AgentRole):
                out = spec.output_key or role_name
            elif isinstance(spec, dict):
                out = spec.get("output_key") or role_name
            else:
                out = role_name
            route_keys[role_name] = out

    def build_role(spec):
        """Normalize one role spec into node(s) for ``route()``.

        An ``AgentRole`` / recipe dict / ``Flow`` becomes a single node
        (``SubFlow`` for agents/flow); a plain ``Node`` is used as-is; a
        list (a route chain such as agent → interrupt) is normalized
        element-wise so roles can be mixed with plain nodes.
        """
        if isinstance(spec, list):
            return [build_role(item) for item in spec]  # type: ignore[misc]
        if isinstance(spec, AgentRole):
            return spec.build(model=model, provider=provider, id=role_name)
        if isinstance(spec, dict):
            return AgentRole.from_mapping(spec, name=role_name).build(
                model=model, provider=provider, id=role_name
            )
        if isinstance(spec, self.flow.__class__):
            return SubFlow(graph=spec.compile())
        return spec

    agents: dict = {}
    for role_name, spec in roles.items():
        if isinstance(spec, list):
            chain = build_role(spec)
            agents[role_name] = flow._as_chain(chain)
        else:
            agents[role_name] = build_role(spec)

    supervisor = Supervisor(
        system=system,
        model=model,
        provider=provider,
        messages_key=messages_key,
        sections=sections,
        route_keys=route_keys,
        done_keys=set(done_keys or ()),
        done_mode=done_mode,
        fallback_agent=fallback,
        max_rounds=max_rounds,
    )
    self.flow.supervisor(supervisor, id=id)
    self.flow.route("next_agent", finish=finish, **agents)
    return self.flow