Skip to content

main_agent

MainAgent

Bases: BaseAgent

Main Agent. This class manages sub agents that in turn use the LLM.

Source code in cat/agents/main_agent.py 
 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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
class MainAgent(BaseAgent):
    """Main Agent.
    This class manages sub agents that in turn use the LLM.
    """

    def __init__(self):
        self.mad_hatter = MadHatter()

        if get_env("CCAT_LOG_LEVEL") in ["DEBUG", "INFO"]:
            self.verbose = True
        else:
            self.verbose = False

    async def execute(self, stray) -> AgentOutput:
        """Execute the agents.

        Returns
        -------
        agent_output : AgentOutput
            Reply of the agent, instance of AgentOutput.
        """

        # prepare input to be passed to the agent.
        #   Info will be extracted from working memory
        # Note: agent_input works both as a dict and as an object
        agent_input : BaseModelDict = self.format_agent_input(stray)
        agent_input = self.mad_hatter.execute_hook(
            "before_agent_starts", agent_input, cat=stray
        )

        # store the agent input inside the working memory
        stray.working_memory.agent_input = agent_input

        # should we run the default agents?
        fast_reply = {}
        fast_reply = self.mad_hatter.execute_hook(
            "agent_fast_reply", fast_reply, cat=stray
        )
        if isinstance(fast_reply, AgentOutput):
            return fast_reply
        if isinstance(fast_reply, dict) and "output" in fast_reply:
            return AgentOutput(**fast_reply)

        # obtain prompt parts from plugins
        prompt_prefix = self.mad_hatter.execute_hook(
            "agent_prompt_prefix", prompts.MAIN_PROMPT_PREFIX, cat=stray
        )
        prompt_suffix = self.mad_hatter.execute_hook(
            "agent_prompt_suffix", prompts.MAIN_PROMPT_SUFFIX, cat=stray
        )

        # run tools and forms
        procedures_agent = ProceduresAgent()
        procedures_agent_out : AgentOutput = await procedures_agent.execute(stray)
        if procedures_agent_out.return_direct:
            return procedures_agent_out

        # we run memory agent if:
        # - no procedures were recalled or selected or
        # - procedures have all return_direct=False
        memory_agent = MemoryAgent()
        memory_agent_out : AgentOutput = await memory_agent.execute(
            # TODO: should all agents only receive stray?
            stray, prompt_prefix, prompt_suffix
        )

        memory_agent_out.intermediate_steps += procedures_agent_out.intermediate_steps

        return memory_agent_out

    def format_agent_input(self, stray):
        """Format the input for the Agent.

        The method formats the strings of recalled memories and chat history that will be provided to the Langchain
        Agent and inserted in the prompt.

        Returns
        -------
        BaseModelDict
            Formatted output to be parsed by the Agent executor. Works both as a dict and as an object.

        Notes
        -----
        The context of memories and conversation history is properly formatted before being parsed by the and, hence,
        information are inserted in the main prompt.
        All the formatting pipeline is hookable and memories can be edited.

        See Also
        --------
        agent_prompt_episodic_memories
        agent_prompt_declarative_memories
        agent_prompt_chat_history
        """

        # format memories to be inserted in the prompt
        episodic_memory_formatted_content = self.agent_prompt_episodic_memories(
            stray.working_memory.episodic_memories
        )
        declarative_memory_formatted_content = self.agent_prompt_declarative_memories(
            stray.working_memory.declarative_memories
        )

        # format conversation history to be inserted in the prompt
        # TODOV2: take away
        conversation_history_formatted_content = stray.stringify_chat_history()

        return BaseModelDict(**{
            "episodic_memory": episodic_memory_formatted_content,
            "declarative_memory": declarative_memory_formatted_content,
            "tools_output": "",
            "input": stray.working_memory.user_message_json.text,  # TODOV2: take away
            "chat_history": conversation_history_formatted_content, # TODOV2: take away
        })

    def agent_prompt_episodic_memories(
        self, memory_docs: List[Tuple[Document, float]]
    ) -> str:
        """Formats episodic memories to be inserted into the prompt.

        Parameters
        ----------
        memory_docs : List[Document]
            List of Langchain `Document` retrieved from the episodic memory.

        Returns
        -------
        memory_content : str
            String of retrieved context from the episodic memory.
        """

        # convert docs to simple text
        memory_texts = [m[0].page_content.replace("\n", ". ") for m in memory_docs]

        # add time information (e.g. "2 days ago")
        memory_timestamps = []
        for m in memory_docs:
            # Get Time information in the Document metadata
            timestamp = m[0].metadata["when"]

            # Get Current Time - Time when memory was stored
            delta = timedelta(seconds=(time.time() - timestamp))

            # Convert and Save timestamps to Verbal (e.g. "2 days ago")
            memory_timestamps.append(f" ({verbal_timedelta(delta)})")

        # Join Document text content with related temporal information
        memory_texts = [a + b for a, b in zip(memory_texts, memory_timestamps)]

        # Format the memories for the output
        memories_separator = "\n  - "
        memory_content = (
            "## Context of things the Human said in the past: "
            + memories_separator
            + memories_separator.join(memory_texts)
        )

        # if no data is retrieved from memory don't erite anithing in the prompt
        if len(memory_texts) == 0:
            memory_content = ""

        return memory_content

    def agent_prompt_declarative_memories(
        self, memory_docs: List[Tuple[Document, float]]
    ) -> str:
        """Formats the declarative memories for the prompt context.
        Such context is placed in the `agent_prompt_prefix` in the place held by {declarative_memory}.

        Parameters
        ----------
        memory_docs : List[Document]
            list of Langchain `Document` retrieved from the declarative memory.

        Returns
        -------
        memory_content : str
            String of retrieved context from the declarative memory.
        """

        # convert docs to simple text
        memory_texts = [m[0].page_content.replace("\n", ". ") for m in memory_docs]

        # add source information (e.g. "extracted from file.txt")
        memory_sources = []
        for m in memory_docs:
            # Get and save the source of the memory
            source = m[0].metadata["source"]
            memory_sources.append(f" (extracted from {source})")

        # Join Document text content with related source information
        memory_texts = [a + b for a, b in zip(memory_texts, memory_sources)]

        # Format the memories for the output
        memories_separator = "\n  - "

        memory_content = (
            "## Context of documents containing relevant information: "
            + memories_separator
            + memories_separator.join(memory_texts)
        )

        # if no data is retrieved from memory don't write anithing in the prompt
        if len(memory_texts) == 0:
            memory_content = ""

        return memory_content

agent_prompt_declarative_memories(memory_docs)

Formats the declarative memories for the prompt context. Such context is placed in the agent_prompt_prefix in the place held by {declarative_memory}.

Parameters:

Name Type Description Default
memory_docs List[Document]

list of Langchain Document retrieved from the declarative memory.

 required

Returns:

Name Type Description
memory_content str

String of retrieved context from the declarative memory.
Source code in cat/agents/main_agent.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def agent_prompt_declarative_memories(
    self, memory_docs: List[Tuple[Document, float]]
) -> str:
    """Formats the declarative memories for the prompt context.
    Such context is placed in the `agent_prompt_prefix` in the place held by {declarative_memory}.

    Parameters
    ----------
    memory_docs : List[Document]
        list of Langchain `Document` retrieved from the declarative memory.

    Returns
    -------
    memory_content : str
        String of retrieved context from the declarative memory.
    """

    # convert docs to simple text
    memory_texts = [m[0].page_content.replace("\n", ". ") for m in memory_docs]

    # add source information (e.g. "extracted from file.txt")
    memory_sources = []
    for m in memory_docs:
        # Get and save the source of the memory
        source = m[0].metadata["source"]
        memory_sources.append(f" (extracted from {source})")

    # Join Document text content with related source information
    memory_texts = [a + b for a, b in zip(memory_texts, memory_sources)]

    # Format the memories for the output
    memories_separator = "\n  - "

    memory_content = (
        "## Context of documents containing relevant information: "
        + memories_separator
        + memories_separator.join(memory_texts)
    )

    # if no data is retrieved from memory don't write anithing in the prompt
    if len(memory_texts) == 0:
        memory_content = ""

    return memory_content

agent_prompt_episodic_memories(memory_docs)

Formats episodic memories to be inserted into the prompt.

Parameters:

Name Type Description Default
memory_docs List[Document]

List of Langchain Document retrieved from the episodic memory.

 required

Returns:

Name Type Description
memory_content str

String of retrieved context from the episodic memory.
Source code in cat/agents/main_agent.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def agent_prompt_episodic_memories(
    self, memory_docs: List[Tuple[Document, float]]
) -> str:
    """Formats episodic memories to be inserted into the prompt.

    Parameters
    ----------
    memory_docs : List[Document]
        List of Langchain `Document` retrieved from the episodic memory.

    Returns
    -------
    memory_content : str
        String of retrieved context from the episodic memory.
    """

    # convert docs to simple text
    memory_texts = [m[0].page_content.replace("\n", ". ") for m in memory_docs]

    # add time information (e.g. "2 days ago")
    memory_timestamps = []
    for m in memory_docs:
        # Get Time information in the Document metadata
        timestamp = m[0].metadata["when"]

        # Get Current Time - Time when memory was stored
        delta = timedelta(seconds=(time.time() - timestamp))

        # Convert and Save timestamps to Verbal (e.g. "2 days ago")
        memory_timestamps.append(f" ({verbal_timedelta(delta)})")

    # Join Document text content with related temporal information
    memory_texts = [a + b for a, b in zip(memory_texts, memory_timestamps)]

    # Format the memories for the output
    memories_separator = "\n  - "
    memory_content = (
        "## Context of things the Human said in the past: "
        + memories_separator
        + memories_separator.join(memory_texts)
    )

    # if no data is retrieved from memory don't erite anithing in the prompt
    if len(memory_texts) == 0:
        memory_content = ""

    return memory_content

execute(stray) async

Execute the agents.

Returns:

Name Type Description
agent_output AgentOutput

Reply of the agent, instance of AgentOutput.
Source code in cat/agents/main_agent.py 
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
async def execute(self, stray) -> AgentOutput:
    """Execute the agents.

    Returns
    -------
    agent_output : AgentOutput
        Reply of the agent, instance of AgentOutput.
    """

    # prepare input to be passed to the agent.
    #   Info will be extracted from working memory
    # Note: agent_input works both as a dict and as an object
    agent_input : BaseModelDict = self.format_agent_input(stray)
    agent_input = self.mad_hatter.execute_hook(
        "before_agent_starts", agent_input, cat=stray
    )

    # store the agent input inside the working memory
    stray.working_memory.agent_input = agent_input

    # should we run the default agents?
    fast_reply = {}
    fast_reply = self.mad_hatter.execute_hook(
        "agent_fast_reply", fast_reply, cat=stray
    )
    if isinstance(fast_reply, AgentOutput):
        return fast_reply
    if isinstance(fast_reply, dict) and "output" in fast_reply:
        return AgentOutput(**fast_reply)

    # obtain prompt parts from plugins
    prompt_prefix = self.mad_hatter.execute_hook(
        "agent_prompt_prefix", prompts.MAIN_PROMPT_PREFIX, cat=stray
    )
    prompt_suffix = self.mad_hatter.execute_hook(
        "agent_prompt_suffix", prompts.MAIN_PROMPT_SUFFIX, cat=stray
    )

    # run tools and forms
    procedures_agent = ProceduresAgent()
    procedures_agent_out : AgentOutput = await procedures_agent.execute(stray)
    if procedures_agent_out.return_direct:
        return procedures_agent_out

    # we run memory agent if:
    # - no procedures were recalled or selected or
    # - procedures have all return_direct=False
    memory_agent = MemoryAgent()
    memory_agent_out : AgentOutput = await memory_agent.execute(
        # TODO: should all agents only receive stray?
        stray, prompt_prefix, prompt_suffix
    )

    memory_agent_out.intermediate_steps += procedures_agent_out.intermediate_steps

    return memory_agent_out

format_agent_input(stray)

Format the input for the Agent.

The method formats the strings of recalled memories and chat history that will be provided to the Langchain Agent and inserted in the prompt.

Returns:

Type Description
BaseModelDict

Formatted output to be parsed by the Agent executor. Works both as a dict and as an object.
Notes

The context of memories and conversation history is properly formatted before being parsed by the and, hence, information are inserted in the main prompt. All the formatting pipeline is hookable and memories can be edited.

See Also

agent_prompt_episodic_memories agent_prompt_declarative_memories agent_prompt_chat_history

Source code in cat/agents/main_agent.py 
 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
def format_agent_input(self, stray):
    """Format the input for the Agent.

    The method formats the strings of recalled memories and chat history that will be provided to the Langchain
    Agent and inserted in the prompt.

    Returns
    -------
    BaseModelDict
        Formatted output to be parsed by the Agent executor. Works both as a dict and as an object.

    Notes
    -----
    The context of memories and conversation history is properly formatted before being parsed by the and, hence,
    information are inserted in the main prompt.
    All the formatting pipeline is hookable and memories can be edited.

    See Also
    --------
    agent_prompt_episodic_memories
    agent_prompt_declarative_memories
    agent_prompt_chat_history
    """

    # format memories to be inserted in the prompt
    episodic_memory_formatted_content = self.agent_prompt_episodic_memories(
        stray.working_memory.episodic_memories
    )
    declarative_memory_formatted_content = self.agent_prompt_declarative_memories(
        stray.working_memory.declarative_memories
    )

    # format conversation history to be inserted in the prompt
    # TODOV2: take away
    conversation_history_formatted_content = stray.stringify_chat_history()

    return BaseModelDict(**{
        "episodic_memory": episodic_memory_formatted_content,
        "declarative_memory": declarative_memory_formatted_content,
        "tools_output": "",
        "input": stray.working_memory.user_message_json.text,  # TODOV2: take away
        "chat_history": conversation_history_formatted_content, # TODOV2: take away
    })