Skip to content

Agent Chains

Agent Chains

Agent Chain Element (AgentChainElement)

Element of an agent chain

llama_cpp_agent.chain.AgentChainElement

Represents a single element in the chain of an agent-based framework. This element holds all necessary data to manage the prompt, process the input/output, and adjust the processing behavior based on given parameters.

Attributes:

  • output_identifier (str) –

    Unique identifier for the output of this chain element.

  • system_prompt (str) –

    Template string for the system prompt.

  • prompt (str) –

    Template string for the main prompt to the language model.

  • preprocessor (Callable[[str, str, dict], tuple[str, str, dict]]) –

    Function to preprocess the input before sending it to the language model. Takes the system prompt with template fields replaced, the prompt with template fields replaced, and the additional information dictionary as arguments and returns the modified tuple of it.

  • postprocessor (Callable[[str, str, dict, str], str]) –

    Function to postprocess the output from the language model. Takes the system prompt, the prompt, the additional information dictionary and the result as arguments and returns the modified result.

  • function_tool_registry (LlamaCppFunctionToolRegistry) –

    Registry for LlamaCppFunctionTool and enables function calling.

  • add_prompt_to_chat_history (bool) –

    Flag to determine if the prompt should be added to the chat history.

  • add_response_to_chat_history (bool) –

    Flag to determine if the response should be added to the chat history.

Methods:

  • __init__ –

    Constructs an instance of the AgentChain class.

Source code in llama_cpp_agent/chain.py 
10
11
12
13
14
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
class AgentChainElement:
    """
    Represents a single element in the chain of an agent-based framework. This element holds all necessary data to
    manage the prompt, process the input/output, and adjust the processing behavior based on given parameters.

    Attributes:
        output_identifier (str): Unique identifier for the output of this chain element.
        system_prompt (str): Template string for the system prompt.
        prompt (str): Template string for the main prompt to the language model.

        preprocessor (Callable[[str, str, dict], tuple[str, str, dict]], optional): Function to preprocess the input
                before sending it to the language model. Takes the system prompt with template fields replaced, the prompt
                with template fields replaced, and the additional information dictionary as arguments and returns the
                modified tuple of it.
        postprocessor (Callable[[str, str, dict, str], str], optional): Function to postprocess the output from
                the language model. Takes the system prompt, the prompt, the additional information dictionary and the
                result as arguments and returns the modified result.
        function_tool_registry (LlamaCppFunctionToolRegistry): Registry for LlamaCppFunctionTool and enables function calling.
        add_prompt_to_chat_history (bool): Flag to determine if the prompt should be added to the chat history.
        add_response_to_chat_history (bool): Flag to determine if the response should be added to the chat history.

    Methods:
        __init__: Constructs an instance of the AgentChain class.
    """

    def __init__(
        self,
        output_identifier: str,
        system_prompt: str,
        prompt: str,
        tools: List[LlamaCppFunctionTool] = None,
        structured_output_settings: LlmStructuredOutputSettings = None,
        preprocessor: Callable[[str, str, dict], tuple[str, str, dict]] = None,
        postprocessor: Callable[[str, str, dict, str], str] = None,
        add_prompt_to_chat_history: bool = False,
        add_response_to_chat_history: bool = False,
        llm_sampling_settings: LlmSamplingSettings = None,
    ):
        """
        Constructs an instance of the AgentChainElement class.

        Args:
            output_identifier (str): Unique identifier for the output of this chain element.
            system_prompt (str): Template string for the system prompt.
            prompt (str): Template string for the main prompt to the language model.
            tools (List[LlamaCppFunctionTool], optional): List of function tools available for use in this chain element.
            preprocessor (Callable[[str, str, dict], tuple[str, str, dict]], optional): Function to preprocess the input
                before sending it to the language model. Takes the system prompt with template fields replaced, the prompt
                with template fields replaced, and the additional information dictionary as arguments and returns the
                modified tuple of it.
            postprocessor (Callable[[str, str, dict, str], str], optional): Function to postprocess the output from
                the language model. Takes the system prompt, the prompt, the additional information dictionary and the
                result as arguments and returns the modified result.
            add_prompt_to_chat_history (bool): Flag to determine if the prompt should be added to the chat history.
            add_response_to_chat_history (bool): Flag to determine if the response should be added to the chat history.
        """
        self.output_identifier = output_identifier
        self.system_prompt = system_prompt
        self.prompt = prompt
        self.add_prompt_to_chat_history = add_prompt_to_chat_history
        self.add_response_to_chat_history = add_response_to_chat_history
        self.structured_output_settings = structured_output_settings
        self.preprocessor = preprocessor
        self.postprocessor = postprocessor
        self.function_tool_registry = None
        if tools is not None:
            self.structured_output_settings = (
                LlmStructuredOutputSettings.from_llama_cpp_function_tools(tools)
            )
        self.llm_sampling_settings = llm_sampling_settings

__init__(output_identifier, system_prompt, prompt, tools=None, structured_output_settings=None, preprocessor=None, postprocessor=None, add_prompt_to_chat_history=False, add_response_to_chat_history=False, llm_sampling_settings=None)

Constructs an instance of the AgentChainElement class.

Parameters:

  • output_identifier (str) –

    Unique identifier for the output of this chain element.

  • system_prompt (str) –

    Template string for the system prompt.

  • prompt (str) –

    Template string for the main prompt to the language model.

  • tools (List[LlamaCppFunctionTool], default: None ) –

    List of function tools available for use in this chain element.

  • preprocessor (Callable[[str, str, dict], tuple[str, str, dict]], default: None ) –

    Function to preprocess the input before sending it to the language model. Takes the system prompt with template fields replaced, the prompt with template fields replaced, and the additional information dictionary as arguments and returns the modified tuple of it.

  • postprocessor (Callable[[str, str, dict, str], str], default: None ) –

    Function to postprocess the output from the language model. Takes the system prompt, the prompt, the additional information dictionary and the result as arguments and returns the modified result.

  • add_prompt_to_chat_history (bool, default: False ) –

    Flag to determine if the prompt should be added to the chat history.

  • add_response_to_chat_history (bool, default: False ) –

    Flag to determine if the response should be added to the chat history.

Source code in llama_cpp_agent/chain.py 
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
def __init__(
    self,
    output_identifier: str,
    system_prompt: str,
    prompt: str,
    tools: List[LlamaCppFunctionTool] = None,
    structured_output_settings: LlmStructuredOutputSettings = None,
    preprocessor: Callable[[str, str, dict], tuple[str, str, dict]] = None,
    postprocessor: Callable[[str, str, dict, str], str] = None,
    add_prompt_to_chat_history: bool = False,
    add_response_to_chat_history: bool = False,
    llm_sampling_settings: LlmSamplingSettings = None,
):
    """
    Constructs an instance of the AgentChainElement class.

    Args:
        output_identifier (str): Unique identifier for the output of this chain element.
        system_prompt (str): Template string for the system prompt.
        prompt (str): Template string for the main prompt to the language model.
        tools (List[LlamaCppFunctionTool], optional): List of function tools available for use in this chain element.
        preprocessor (Callable[[str, str, dict], tuple[str, str, dict]], optional): Function to preprocess the input
            before sending it to the language model. Takes the system prompt with template fields replaced, the prompt
            with template fields replaced, and the additional information dictionary as arguments and returns the
            modified tuple of it.
        postprocessor (Callable[[str, str, dict, str], str], optional): Function to postprocess the output from
            the language model. Takes the system prompt, the prompt, the additional information dictionary and the
            result as arguments and returns the modified result.
        add_prompt_to_chat_history (bool): Flag to determine if the prompt should be added to the chat history.
        add_response_to_chat_history (bool): Flag to determine if the response should be added to the chat history.
    """
    self.output_identifier = output_identifier
    self.system_prompt = system_prompt
    self.prompt = prompt
    self.add_prompt_to_chat_history = add_prompt_to_chat_history
    self.add_response_to_chat_history = add_response_to_chat_history
    self.structured_output_settings = structured_output_settings
    self.preprocessor = preprocessor
    self.postprocessor = postprocessor
    self.function_tool_registry = None
    if tools is not None:
        self.structured_output_settings = (
            LlmStructuredOutputSettings.from_llama_cpp_function_tools(tools)
        )
    self.llm_sampling_settings = llm_sampling_settings

Sequential Chain (AgentChain)

Sequentially invoked chain.

llama_cpp_agent.chain.AgentChain

Represents a chain of AgentChainElements that are processed in a sequence to handle an interaction within an agent-based system.

Attributes:

  • agent (LlamaCppAgent) –

    The agent responsible for managing and processing the chain.

  • chain (List[AgentChainElement]) –

    A list of AgentChainElement instances that make up the chain.

Methods:

  • __init__ –

    Constructs an instance of the AgentChain class.

  • run_chain –

    Processes the entire chain of elements using provided inputs.

Source code in llama_cpp_agent/chain.py 
 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
class AgentChain:
    """
    Represents a chain of AgentChainElements that are processed in a sequence to handle an interaction within an
    agent-based system.

    Attributes:
        agent (LlamaCppAgent): The agent responsible for managing and processing the chain.
        chain (List[AgentChainElement]): A list of AgentChainElement instances that make up the chain.

    Methods:
        __init__: Constructs an instance of the AgentChain class.
        run_chain: Processes the entire chain of elements using provided inputs.
    """

    def __init__(self, agent: LlamaCppAgent, chain_elements: List[AgentChainElement]):
        """
        Constructs an instance of the AgentChain class.
        Args:
            agent (LlamaCppAgent): The agent responsible for managing and processing the chain.
            chain_elements (List[AgentChainElement]): A list of AgentChainElement instances that make up the chain.
        """
        self.agent = agent
        self.chain = chain_elements

    def run_chain(self, user_message: str = None, additional_fields: dict = None):
        """
        Executes the chain of agent elements using the initial user message and additional fields, and
        returns the final output and state of the outputs dictionary.

        Args:
            user_message (str, optional): Initial user message to be processed by the chain.
            additional_fields (dict, optional): Dictionary of additional data to be used in the processing of the chain.

        Returns:
            tuple[str, dict]: A tuple containing the concatenated output string from the final element and
            the dictionary of all outputs.
        """
        outputs = {}

        if additional_fields is not None:
            for field, value in additional_fields.items():
                outputs[field] = value

        for chain_element in self.chain:
            sys_prompter = PromptTemplate.from_string(chain_element.system_prompt)
            sys_prompt = sys_prompter.generate_prompt(outputs)
            prompter = PromptTemplate.from_string(chain_element.prompt)
            prompt = prompter.generate_prompt(outputs)
            if chain_element.function_tool_registry is not None:
                sys_prompt += f"\n\nYou can call functions in JSON format.\n\nAvailable Function Tools:\n\n{chain_element.function_tool_registry.get_documentation().strip()}"
            if chain_element.preprocessor is not None:
                sys_prompt, prompt, outputs = chain_element.preprocessor(
                    sys_prompt, prompt, outputs
                )

            outputs[chain_element.output_identifier] = self.agent.get_chat_response(
                user_message if user_message is not None else prompt,
                system_prompt=sys_prompt,
                structured_output_settings=chain_element.structured_output_settings,
                add_response_to_chat_history=chain_element.add_response_to_chat_history,
                add_message_to_chat_history=chain_element.add_prompt_to_chat_history,
                llm_sampling_settings=chain_element.llm_sampling_settings,
            )
            if chain_element.function_tool_registry is not None:
                outputs[chain_element.output_identifier] = outputs[
                    chain_element.output_identifier
                ][0]["return_value"]
            if chain_element.postprocessor is not None:
                outputs[
                    chain_element.output_identifier
                ] = chain_element.postprocessor = chain_element.postprocessor(
                    sys_prompt,
                    prompt,
                    outputs,
                    outputs[chain_element.output_identifier],
                )
        output = "\n".join(
            [val if isinstance(val, str) else str(val) for val in outputs.values()]
        )
        return output, outputs

__init__(agent, chain_elements)

Constructs an instance of the AgentChain class. Args: agent (LlamaCppAgent): The agent responsible for managing and processing the chain. chain_elements (List[AgentChainElement]): A list of AgentChainElement instances that make up the chain.

Source code in llama_cpp_agent/chain.py 
 96
 97
 98
 99
100
101
102
103
104
def __init__(self, agent: LlamaCppAgent, chain_elements: List[AgentChainElement]):
    """
    Constructs an instance of the AgentChain class.
    Args:
        agent (LlamaCppAgent): The agent responsible for managing and processing the chain.
        chain_elements (List[AgentChainElement]): A list of AgentChainElement instances that make up the chain.
    """
    self.agent = agent
    self.chain = chain_elements

run_chain(user_message=None, additional_fields=None)

Executes the chain of agent elements using the initial user message and additional fields, and returns the final output and state of the outputs dictionary.

Parameters:

  • user_message (str, default: None ) –

    Initial user message to be processed by the chain.

  • additional_fields (dict, default: None ) –

    Dictionary of additional data to be used in the processing of the chain.

Returns:

  • –

    tuple[str, dict]: A tuple containing the concatenated output string from the final element and

  • –

    the dictionary of all outputs.

Source code in llama_cpp_agent/chain.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
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
def run_chain(self, user_message: str = None, additional_fields: dict = None):
    """
    Executes the chain of agent elements using the initial user message and additional fields, and
    returns the final output and state of the outputs dictionary.

    Args:
        user_message (str, optional): Initial user message to be processed by the chain.
        additional_fields (dict, optional): Dictionary of additional data to be used in the processing of the chain.

    Returns:
        tuple[str, dict]: A tuple containing the concatenated output string from the final element and
        the dictionary of all outputs.
    """
    outputs = {}

    if additional_fields is not None:
        for field, value in additional_fields.items():
            outputs[field] = value

    for chain_element in self.chain:
        sys_prompter = PromptTemplate.from_string(chain_element.system_prompt)
        sys_prompt = sys_prompter.generate_prompt(outputs)
        prompter = PromptTemplate.from_string(chain_element.prompt)
        prompt = prompter.generate_prompt(outputs)
        if chain_element.function_tool_registry is not None:
            sys_prompt += f"\n\nYou can call functions in JSON format.\n\nAvailable Function Tools:\n\n{chain_element.function_tool_registry.get_documentation().strip()}"
        if chain_element.preprocessor is not None:
            sys_prompt, prompt, outputs = chain_element.preprocessor(
                sys_prompt, prompt, outputs
            )

        outputs[chain_element.output_identifier] = self.agent.get_chat_response(
            user_message if user_message is not None else prompt,
            system_prompt=sys_prompt,
            structured_output_settings=chain_element.structured_output_settings,
            add_response_to_chat_history=chain_element.add_response_to_chat_history,
            add_message_to_chat_history=chain_element.add_prompt_to_chat_history,
            llm_sampling_settings=chain_element.llm_sampling_settings,
        )
        if chain_element.function_tool_registry is not None:
            outputs[chain_element.output_identifier] = outputs[
                chain_element.output_identifier
            ][0]["return_value"]
        if chain_element.postprocessor is not None:
            outputs[
                chain_element.output_identifier
            ] = chain_element.postprocessor = chain_element.postprocessor(
                sys_prompt,
                prompt,
                outputs,
                outputs[chain_element.output_identifier],
            )
    output = "\n".join(
        [val if isinstance(val, str) else str(val) for val in outputs.values()]
    )
    return output, outputs

Map Chain (MapChain)

Maps over a list of items and then combines the results using another chain

llama_cpp_agent.chain.MapChain

Represents a specialized chain that maps over a list of items and then combines the results using another chain.

Attributes:

  • agent (LlamaCppAgent) –

    The agent responsible for managing and processing the map and combine chains.

  • map_chain (AgentChain) –

    An AgentChain instance used to process each item in the list.

  • combine_chain (AgentChain) –

    An AgentChain instance used to combine the results of the map chain.

  • item_identifier (str) –

    The identifier used to insert each item into the additional_fields dictionary.

  • map_output_identifier (str) –

    The identifier used to store the results of the map chain before passing to the

Methods:

  • __init__ –

    Constructs an instance of the MapChain class.

  • run_map_chain –

    Executes the map chain on a list of items and then processes the results with the combine chain.

Source code in llama_cpp_agent/chain.py 
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class MapChain:
    """
    Represents a specialized chain that maps over a list of items and then combines the results using another chain.

    Attributes:
        agent (LlamaCppAgent): The agent responsible for managing and processing the map and combine chains.
        map_chain (AgentChain): An AgentChain instance used to process each item in the list.
        combine_chain (AgentChain): An AgentChain instance used to combine the results of the map chain.
        item_identifier (str): The identifier used to insert each item into the additional_fields dictionary.
        map_output_identifier (str): The identifier used to store the results of the map chain before passing to the
        combine chain.

    Methods:
        __init__: Constructs an instance of the MapChain class.
        run_map_chain: Executes the map chain on a list of items and then processes the results with the combine chain.
    """

    def __init__(
        self,
        agent: LlamaCppAgent,
        map_chain: List[AgentChainElement],
        combine_chain: List[AgentChainElement],
        item_identifier: str = "item",
        map_output_identifier: str = "map_output",
    ):
        """
        Constructs an instance of the MapChain class. This class is designed to process a list of items through
        a mapping chain and then combine the results using a combining chain.

        Args:
            agent (LlamaCppAgent): The agent responsible for managing and processing the map and combine chains.
            map_chain (List[AgentChainElement]): A list of AgentChainElement instances that make up the map chain.
            combine_chain (List[AgentChainElement]): A list of AgentChainElement instances that make up the combine chain.
            item_identifier (str, optional): The identifier used to insert each item into the additional_fields dictionary.
            map_output_identifier (str, optional): The identifier used to store the results of the map chain before passing to the combine chain.
        """
        self.agent = agent
        self.map_chain = AgentChain(agent, map_chain)
        self.combine_chain = AgentChain(agent, combine_chain)
        self.item_identifier = item_identifier
        self.map_output_identifier = map_output_identifier

    def run_map_chain(
        self,
        items_to_map: list[str],
        user_message: str = None,
        additional_fields: dict = None,
    ):
        """
        Executes the map chain over a list of items and then uses the combine chain to process the concatenated
        results.

        Args:
            items_to_map (list[str]): List of strings to be individually processed by the map chain.
            user_message (str, optional): Initial user message to be included in the processing.
            additional_fields (dict, optional): Additional data to be used throughout the map and combine chains.

        Returns:
            tuple: A tuple containing the final output string from the combine chain and the outputs dictionary.
        """
        outputs = {}

        if additional_fields is not None:
            for field, value in additional_fields.items():
                outputs[field] = value
        else:
            additional_fields = {}
        results = []
        for item in items_to_map:
            additional_fields[self.item_identifier] = item
            result, result_dic = self.map_chain.run_chain(
                user_message=user_message, additional_fields=additional_fields
            )
            results.append(result_dic[self.map_chain.chain[-1].output_identifier])
        additional_fields[self.map_output_identifier] = "\n\n".join(results)
        return self.combine_chain.run_chain(additional_fields=additional_fields)

__init__(agent, map_chain, combine_chain, item_identifier='item', map_output_identifier='map_output')

Constructs an instance of the MapChain class. This class is designed to process a list of items through a mapping chain and then combine the results using a combining chain.

Parameters:

  • agent (LlamaCppAgent) –

    The agent responsible for managing and processing the map and combine chains.

  • map_chain (List[AgentChainElement]) –

    A list of AgentChainElement instances that make up the map chain.

  • combine_chain (List[AgentChainElement]) –

    A list of AgentChainElement instances that make up the combine chain.

  • item_identifier (str, default: 'item' ) –

    The identifier used to insert each item into the additional_fields dictionary.

  • map_output_identifier (str, default: 'map_output' ) –

    The identifier used to store the results of the map chain before passing to the combine chain.

Source code in llama_cpp_agent/chain.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def __init__(
    self,
    agent: LlamaCppAgent,
    map_chain: List[AgentChainElement],
    combine_chain: List[AgentChainElement],
    item_identifier: str = "item",
    map_output_identifier: str = "map_output",
):
    """
    Constructs an instance of the MapChain class. This class is designed to process a list of items through
    a mapping chain and then combine the results using a combining chain.

    Args:
        agent (LlamaCppAgent): The agent responsible for managing and processing the map and combine chains.
        map_chain (List[AgentChainElement]): A list of AgentChainElement instances that make up the map chain.
        combine_chain (List[AgentChainElement]): A list of AgentChainElement instances that make up the combine chain.
        item_identifier (str, optional): The identifier used to insert each item into the additional_fields dictionary.
        map_output_identifier (str, optional): The identifier used to store the results of the map chain before passing to the combine chain.
    """
    self.agent = agent
    self.map_chain = AgentChain(agent, map_chain)
    self.combine_chain = AgentChain(agent, combine_chain)
    self.item_identifier = item_identifier
    self.map_output_identifier = map_output_identifier

run_map_chain(items_to_map, user_message=None, additional_fields=None)

Executes the map chain over a list of items and then uses the combine chain to process the concatenated results.

Parameters:

  • items_to_map (list[str]) –

    List of strings to be individually processed by the map chain.

  • user_message (str, default: None ) –

    Initial user message to be included in the processing.

  • additional_fields (dict, default: None ) –

    Additional data to be used throughout the map and combine chains.

Returns:

  • tuple –

    A tuple containing the final output string from the combine chain and the outputs dictionary.

Source code in llama_cpp_agent/chain.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def run_map_chain(
    self,
    items_to_map: list[str],
    user_message: str = None,
    additional_fields: dict = None,
):
    """
    Executes the map chain over a list of items and then uses the combine chain to process the concatenated
    results.

    Args:
        items_to_map (list[str]): List of strings to be individually processed by the map chain.
        user_message (str, optional): Initial user message to be included in the processing.
        additional_fields (dict, optional): Additional data to be used throughout the map and combine chains.

    Returns:
        tuple: A tuple containing the final output string from the combine chain and the outputs dictionary.
    """
    outputs = {}

    if additional_fields is not None:
        for field, value in additional_fields.items():
            outputs[field] = value
    else:
        additional_fields = {}
    results = []
    for item in items_to_map:
        additional_fields[self.item_identifier] = item
        result, result_dic = self.map_chain.run_chain(
            user_message=user_message, additional_fields=additional_fields
        )
        results.append(result_dic[self.map_chain.chain[-1].output_identifier])
    additional_fields[self.map_output_identifier] = "\n\n".join(results)
    return self.combine_chain.run_chain(additional_fields=additional_fields)