Skip to content

Team

Multiagent building blocks for orchestrating multiple agents in a team.

Classes:

  • ActiveTeamAgentView –
  • Chain –

    Calls agents sequentially. Copies thoughts of previous agents for the next agents.

  • TeamAgent –

    Agent designed to work in the team with similar other agents performing different kinds

ActiveTeamAgentView

Methods:

  • __init__ –

    ActiveTeamAgentView contains the ephemeral state computed from the tape. This class extracts the data relevant to

Source code in tapeagents/team.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class ActiveTeamAgentView:
    def __init__(self, agent: TeamAgent, tape: TeamTape):
        """
        ActiveTeamAgentView contains the ephemeral state computed from the tape. This class extracts the data relevant to
        the given agent and also computes some additional information from it, e.g. whether the agent
        should call the LLM to generate a message or respond with an already available one.
        """
        view = TapeViewStack.compute(tape)
        self.messages = view.messages_by_agent[agent.full_name]
        self.last_non_empty_message = next((m for m in reversed(self.messages) if m.content), None)
        self.node = agent.select_node(tape)
        self.steps = view.top.steps
        self.steps_by_kind = view.top.steps_by_kind
        self.exec_result = self.steps[-1] if self.steps and isinstance(self.steps[-1], CodeExecutionResult) else None
        self.should_generate_message = (
            isinstance(self.node, (CallNode, RespondNode))
            and self.messages
            and not self.exec_result
            and "system" in agent.templates
        )
        self.should_stop = (
            agent.max_calls and (agent.max_calls and len(self.steps_by_kind.get("call", [])) >= agent.max_calls)
        ) or (self.messages and ("TERMINATE" in self.messages[-1].content))

__init__(agent, tape)

ActiveTeamAgentView contains the ephemeral state computed from the tape. This class extracts the data relevant to the given agent and also computes some additional information from it, e.g. whether the agent should call the LLM to generate a message or respond with an already available one.

Source code in tapeagents/team.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, agent: TeamAgent, tape: TeamTape):
    """
    ActiveTeamAgentView contains the ephemeral state computed from the tape. This class extracts the data relevant to
    the given agent and also computes some additional information from it, e.g. whether the agent
    should call the LLM to generate a message or respond with an already available one.
    """
    view = TapeViewStack.compute(tape)
    self.messages = view.messages_by_agent[agent.full_name]
    self.last_non_empty_message = next((m for m in reversed(self.messages) if m.content), None)
    self.node = agent.select_node(tape)
    self.steps = view.top.steps
    self.steps_by_kind = view.top.steps_by_kind
    self.exec_result = self.steps[-1] if self.steps and isinstance(self.steps[-1], CodeExecutionResult) else None
    self.should_generate_message = (
        isinstance(self.node, (CallNode, RespondNode))
        and self.messages
        and not self.exec_result
        and "system" in agent.templates
    )
    self.should_stop = (
        agent.max_calls and (agent.max_calls and len(self.steps_by_kind.get("call", [])) >= agent.max_calls)
    ) or (self.messages and ("TERMINATE" in self.messages[-1].content))

Chain

Bases: Agent[TapeType], Generic[TapeType]

Calls agents sequentially. Copies thoughts of previous agents for the next agents.

Source code in tapeagents/team.py 
333
334
335
336
337
338
339
340
341
class Chain(Agent[TapeType], Generic[TapeType]):
    """Calls agents sequentially. Copies thoughts of previous agents for the next agents."""

    @classmethod
    def create(cls, nodes: list[CallSubagent], **kwargs) -> Self:
        subagents = []
        for node in nodes:
            subagents.append(node.agent)
        return super().create(nodes=nodes + [RespondIfNotRootNode()], subagents=subagents, **kwargs)

TeamAgent

Bases: Agent[TeamTape]

Agent designed to work in the team with similar other agents performing different kinds

Methods:

  • create –

    Create a simple agent that can execute code, think and respond to messages

  • create_initiator –

    Create an agent that sets the team's initial message and calls the team manager

  • create_team_manager –

    Create a team manager that broadcasts the last message to all subagents, selects one of them to call, call it and

Source code in tapeagents/team.py 
 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
class TeamAgent(Agent[TeamTape]):
    """
    Agent designed to work in the team with similar other agents performing different kinds
    """

    max_calls: int | None = None
    init_message: str | None = None

    model_config = ConfigDict(use_enum_values=True)

    @classmethod
    def create(
        cls,
        name: str,
        system_prompt: str | None = None,
        llm: LLM | None = None,
        execute_code: bool = False,
    ):  # type: ignore
        """
        Create a simple agent that can execute code, think and respond to messages
        """
        return cls(
            name=name,
            templates={"system": system_prompt} if system_prompt else {},
            llms={DEFAULT: llm} if llm else {},
            nodes=([ExecuteCodeNode()] if execute_code else []) + [RespondNode()],  # type: ignore
        )

    @classmethod
    def create_team_manager(
        cls,
        name: str,
        subagents: list[Agent[TeamTape]],
        llm: LLM,
        templates: dict[str, str],
        max_calls: int = 1,
    ):
        """
        Create a team manager that broadcasts the last message to all subagents, selects one of them to call, call it and
        responds to the last message if the termination message is not received.
        """
        return cls(
            name=name,
            subagents=subagents,
            nodes=[
                BroadcastLastMessageNode(),
                SelectAndCallNode(),
                RespondOrRepeatNode(next_node="broadcast_last_message"),
            ],
            max_calls=max_calls,
            templates=templates,
            llms={DEFAULT: llm},
        )

    @classmethod
    def create_initiator(
        cls,
        name: str,
        teammate: Agent[TeamTape],
        init_message: str,
        system_prompt: str = "",
        llm: LLM | None = None,
        max_calls: int = 1,
        execute_code: bool = False,
    ):
        """
        Create an agent that sets the team's initial message and calls the team manager
        """
        nodes = []
        if execute_code:
            nodes = [ExecuteCodeNode(), CallNode(), TerminateOrRepeatNode(next_node="execute_code")]
        else:
            nodes = [CallNode(), TerminateOrRepeatNode(next_node="call")]
        return cls(
            name=name,
            templates={
                "system": system_prompt,
            },
            llms={DEFAULT: llm} if llm else {},
            subagents=[teammate],
            nodes=nodes,
            max_calls=max_calls,
            init_message=init_message,
        )

create(name, system_prompt=None, llm=None, execute_code=False) classmethod

Create a simple agent that can execute code, think and respond to messages

Source code in tapeagents/team.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@classmethod
def create(
    cls,
    name: str,
    system_prompt: str | None = None,
    llm: LLM | None = None,
    execute_code: bool = False,
):  # type: ignore
    """
    Create a simple agent that can execute code, think and respond to messages
    """
    return cls(
        name=name,
        templates={"system": system_prompt} if system_prompt else {},
        llms={DEFAULT: llm} if llm else {},
        nodes=([ExecuteCodeNode()] if execute_code else []) + [RespondNode()],  # type: ignore
    )

create_initiator(name, teammate, init_message, system_prompt='', llm=None, max_calls=1, execute_code=False) classmethod

Create an agent that sets the team's initial message and calls the team manager

Source code in tapeagents/team.py 
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
@classmethod
def create_initiator(
    cls,
    name: str,
    teammate: Agent[TeamTape],
    init_message: str,
    system_prompt: str = "",
    llm: LLM | None = None,
    max_calls: int = 1,
    execute_code: bool = False,
):
    """
    Create an agent that sets the team's initial message and calls the team manager
    """
    nodes = []
    if execute_code:
        nodes = [ExecuteCodeNode(), CallNode(), TerminateOrRepeatNode(next_node="execute_code")]
    else:
        nodes = [CallNode(), TerminateOrRepeatNode(next_node="call")]
    return cls(
        name=name,
        templates={
            "system": system_prompt,
        },
        llms={DEFAULT: llm} if llm else {},
        subagents=[teammate],
        nodes=nodes,
        max_calls=max_calls,
        init_message=init_message,
    )

create_team_manager(name, subagents, llm, templates, max_calls=1) classmethod

Create a team manager that broadcasts the last message to all subagents, selects one of them to call, call it and responds to the last message if the termination message is not received.

Source code in tapeagents/team.py 
 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
@classmethod
def create_team_manager(
    cls,
    name: str,
    subagents: list[Agent[TeamTape]],
    llm: LLM,
    templates: dict[str, str],
    max_calls: int = 1,
):
    """
    Create a team manager that broadcasts the last message to all subagents, selects one of them to call, call it and
    responds to the last message if the termination message is not received.
    """
    return cls(
        name=name,
        subagents=subagents,
        nodes=[
            BroadcastLastMessageNode(),
            SelectAndCallNode(),
            RespondOrRepeatNode(next_node="broadcast_last_message"),
        ],
        max_calls=max_calls,
        templates=templates,
        llms={DEFAULT: llm},
    )