Skip to content

agent

Agent

Bases: Generic[AgentInputT, AgentOutputT]

Source code in agenty/agent/base.py 
 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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class Agent(Generic[AgentInputT, AgentOutputT], metaclass=AgentMeta):
    # agenty
    name: str = ""
    system_prompt: str = ""
    input_schema: Type[AgentIO] = str
    output_schema: Type[AgentIO] = str
    chat_history: ChatHistory

    # pydantic-ai
    model: Optional[Model] = None
    model_settings: Optional[ModelSettings] = None
    retries: int = 1
    result_retries: Optional[int] = None
    end_strategy: EndStrategy = "early"

    _pai_agent: Optional[pai.Agent["Agent[AgentInputT, AgentOutputT]", AgentIO]]
    _input_hooks: List[
        Callable[["Agent[AgentInputT, AgentOutputT]", AgentInputT], AgentInputT]
    ]
    _output_hooks: List[
        Callable[["Agent[AgentInputT, AgentOutputT]", AgentOutputT], AgentOutputT]
    ]

    def __init__(
        self,
        model: Union[KnownModelName, Model, NotGiven, None] = NOT_GIVEN,
        model_settings: Union[Optional[ModelSettings], NotGiven] = NOT_GIVEN,
        name: Union[str, NotGiven] = NOT_GIVEN,
        system_prompt: Union[str, NotGiven] = NOT_GIVEN,
        input_schema: Union[Type[AgentIO], NotGiven] = NOT_GIVEN,
        output_schema: Union[Type[AgentIO], NotGiven] = NOT_GIVEN,
        retries: Union[int, NotGiven] = NOT_GIVEN,
        result_retries: Union[Optional[int], NotGiven] = NOT_GIVEN,
        end_strategy: Union[EndStrategy, NotGiven] = NOT_GIVEN,
        chat_history: Union[ChatHistory, NotGiven] = NOT_GIVEN,
        # usage: Optional[AgentUsage] = None,
        # usage_limits: Optional[AgentUsageLimits] = None,
    ) -> None:
        _create_pai_agent = False
        # infer model so that we always have a Model instance or None
        if not isinstance(model, NotGiven):
            if model is None:
                self.model = None
            else:
                self.model = infer_model(model)
                _create_pai_agent = True

        if not isinstance(name, NotGiven):
            self.name = name
        if not isinstance(system_prompt, NotGiven):
            self.system_prompt = system_prompt
        if not isinstance(input_schema, NotGiven):
            self.input_schema = input_schema
        if isinstance(chat_history, NotGiven):
            self.chat_history = ChatHistory()
        else:
            self.chat_history = chat_history
        # self.usage = usage or AgentUsage()
        # self.usage_limits = usage_limits or AgentUsageLimits()

        # If any of these attributes are not NOT_GIVEN, set them and create an object-specific pydantic-ai agent
        create_pai_agent_attributes = {
            "model_settings": model_settings,
            "output_schema": output_schema,
            "retries": retries,
            "result_retries": result_retries,
            "end_strategy": end_strategy,
        }
        for attr, val in create_pai_agent_attributes.items():
            if val is not None and not isinstance(val, NotGiven):
                setattr(self, attr, val)
                _create_pai_agent = True

        if _create_pai_agent and self.model is not None:
            logger.debug("Creating object-specific pydantic-ai agent")
            self._pai_agent = None
            if model is not None:
                self._pai_agent = pai.Agent(
                    self.model,
                    result_type=self.output_schema,
                    deps_type=self.__class__,
                    model_settings=self.model_settings,
                    retries=self.retries,
                    result_retries=self.result_retries,
                    end_strategy=self.end_strategy,
                )

    async def run(
        self,
        input_data: Optional[AgentInputT],
        name: Optional[str] = None,
    ) -> AgentOutputT:
        """Run the agent with the provided input.

        Args:
            input_data: The input data for the agent to process

        Returns:
            The processed output data
        """

        _chat_history = self.chat_history.to_pydantic_ai(ctx=self.template_context())
        _input_data: Any = input_data
        if _input_data is not None:
            for input_hook in self._input_hooks:
                _type: Any = type(_input_data)
                _input_data = input_hook(self, _input_data)
                if not isinstance(_input_data, _type):
                    raise exc.AgentyValueError(
                        f"Input hook {input_hook.__name__} returned invalid type"
                    )
            try:
                TypeAdapter(self.input_schema).validate_python(input_data)
            except ValidationError as e:
                raise exc.AgentyTypeError(e)
            self.chat_history.add("user", _input_data, name=name)

        system_prompt = ChatMessage(
            role="system", content=self.system_prompt
        ).to_pydantic_ai(ctx=self.template_context())

        try:
            output = await self.pai_agent.run(
                str(_input_data),
                message_history=[system_prompt] + _chat_history,
                deps=self,
            )
        except pai.exceptions.UnexpectedModelBehavior as e:
            raise exc.InvalidResponse(e)

        _output_data: Any = output.data
        for output_hook in self._output_hooks:
            _type = type(_output_data)
            _output_data = output_hook(self, _output_data)
            if not isinstance(_output_data, _type):
                raise exc.AgentyValueError(
                    f"Output hook {output_hook.__name__} returned invalid type"
                )

        self.chat_history.add("assistant", _output_data, name=name)
        return cast(AgentOutputT, _output_data)

    def run_sync(
        self,
        input_data: Optional[AgentInputT],
        name: Optional[str] = None,
    ) -> AgentOutputT:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop:
            return loop.run_until_complete(
                asyncio.create_task(self.run(input_data, name))
            )
        else:
            return asyncio.run(self.run(input_data, name))

    @property
    def model_name(self) -> str:
        """Get the name of the current model.

        Returns:
            str: The model name

        Raises:
            ValueError: If no model is set
        """
        if self.model is None:
            return ""

        return self.model.name()

    @property
    def pai_agent(self) -> pai.Agent["Agent[AgentInputT, AgentOutputT]", AgentIO]:
        """Get the underlying pydantic-ai agent instance.

        Returns:
            pai.Agent: The pydantic-ai agent instance
        """
        if self.model is None:
            raise exc.AgentyAttributeError("Model is not set")
        if self._pai_agent is None:
            raise exc.AgentyAttributeError("Pydantic AI agent does not exist")
        return self._pai_agent

    @property
    def system_prompt_rendered(self) -> str:
        """Render the system prompt with the current template context.

        Returns:
            str: The rendered system prompt
        """
        return apply_template(self.system_prompt, self.template_context())

    def template_context(self) -> Dict[str, Any]:
        """Get a dictionary of instance variables for use in templates. By default, this includes all variables that start with an uppercase letter.

        Returns:
            Dict[str, Any]: Dictionary of variables
        """
        return {key: getattr(self, key) for key in dir(self) if key[0].isupper()}

    def reset(self) -> None:
        """Reset the agent to its initial state."""
        self.chat_history.clear()

    def __or__(
        self, other: AgentIOProtocol[AgentOutputT, PipelineOutputT]
    ) -> AgentIOProtocol[AgentInputT, PipelineOutputT]:
        """Chain this agent with another using the | operator.

        Args:
            other: Another agent to chain with. Its input schema must match this agent's output schema.

        Returns:
            A new Pipeline instance containing both agents, preserving type information.
        """
        return Pipeline[AgentInputT, PipelineOutputT](
            agents=[self, other],
            input_schema=self.input_schema,
            output_schema=other.output_schema,
        )

model_name property

Get the name of the current model.

Returns:

Name Type Description
str str

The model name

Raises:

Type Description
ValueError

If no model is set

pai_agent property

Get the underlying pydantic-ai agent instance.

Returns:

Type Description
Agent[Agent[AgentInputT, AgentOutputT], AgentIO]

pai.Agent: The pydantic-ai agent instance

system_prompt_rendered property

Render the system prompt with the current template context.

Returns:

Name Type Description
str str

The rendered system prompt

reset()

Reset the agent to its initial state.

Source code in agenty/agent/base.py 
236
237
238
def reset(self) -> None:
    """Reset the agent to its initial state."""
    self.chat_history.clear()

run(input_data, name=None) async

Run the agent with the provided input.

Parameters:

Name Type Description Default
input_data Optional[AgentInputT]

The input data for the agent to process

 required

Returns:

Type Description
AgentOutputT

The processed output data
Source code in agenty/agent/base.py 
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
async def run(
    self,
    input_data: Optional[AgentInputT],
    name: Optional[str] = None,
) -> AgentOutputT:
    """Run the agent with the provided input.

    Args:
        input_data: The input data for the agent to process

    Returns:
        The processed output data
    """

    _chat_history = self.chat_history.to_pydantic_ai(ctx=self.template_context())
    _input_data: Any = input_data
    if _input_data is not None:
        for input_hook in self._input_hooks:
            _type: Any = type(_input_data)
            _input_data = input_hook(self, _input_data)
            if not isinstance(_input_data, _type):
                raise exc.AgentyValueError(
                    f"Input hook {input_hook.__name__} returned invalid type"
                )
        try:
            TypeAdapter(self.input_schema).validate_python(input_data)
        except ValidationError as e:
            raise exc.AgentyTypeError(e)
        self.chat_history.add("user", _input_data, name=name)

    system_prompt = ChatMessage(
        role="system", content=self.system_prompt
    ).to_pydantic_ai(ctx=self.template_context())

    try:
        output = await self.pai_agent.run(
            str(_input_data),
            message_history=[system_prompt] + _chat_history,
            deps=self,
        )
    except pai.exceptions.UnexpectedModelBehavior as e:
        raise exc.InvalidResponse(e)

    _output_data: Any = output.data
    for output_hook in self._output_hooks:
        _type = type(_output_data)
        _output_data = output_hook(self, _output_data)
        if not isinstance(_output_data, _type):
            raise exc.AgentyValueError(
                f"Output hook {output_hook.__name__} returned invalid type"
            )

    self.chat_history.add("assistant", _output_data, name=name)
    return cast(AgentOutputT, _output_data)

template_context()

Get a dictionary of instance variables for use in templates. By default, this includes all variables that start with an uppercase letter.

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: Dictionary of variables
Source code in agenty/agent/base.py 
228
229
230
231
232
233
234
def template_context(self) -> Dict[str, Any]:
    """Get a dictionary of instance variables for use in templates. By default, this includes all variables that start with an uppercase letter.

    Returns:
        Dict[str, Any]: Dictionary of variables
    """
    return {key: getattr(self, key) for key in dir(self) if key[0].isupper()}

AgentUsage

Bases: MutableMapping[str, Usage]

Tracks usage statistics for multiple models in an agent.

A dictionary-like container that maps model names to their usage statistics. Automatically creates new Usage entries for unknown models and provides aggregated statistics across all models.

Source code in agenty/agent/usage.py 
  9
 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
 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
class AgentUsage(MutableMapping[str, Usage]):
    """Tracks usage statistics for multiple models in an agent.

    A dictionary-like container that maps model names to their usage statistics.
    Automatically creates new Usage entries for unknown models and provides
    aggregated statistics across all models.
    """

    def __init__(self):
        """Initialize an empty usage tracker."""
        self._usages: dict[str, Usage] = {}

    def __getitem__(self, key: str) -> Usage:
        """Get usage statistics for a model, creating a new entry if needed.

        Args:
            key: Name of the model

        Returns:
            Usage: Usage statistics for the specified model. If the model doesn't
                  exist, a new Usage instance is created and returned.
        """
        try:
            return self._usages[key]
        except KeyError:
            self._usages[key] = Usage()
            return self._usages[key]

    def __setitem__(self, key: str, value: Usage) -> None:
        """Set usage statistics for a model.

        Args:
            key: Name of the model
            value: Usage statistics to set for the model
        """
        self._usages[key] = value

    def __delitem__(self, key: str) -> None:
        """Remove usage statistics for a model.

        Args:
            key: Name of the model to remove

        Raises:
            KeyError: If the model doesn't exist
        """
        del self._usages[key]

    def __iter__(self) -> Iterator[str]:
        """Iterate over model names.

        Returns:
            Iterator[str]: Iterator over model names with usage statistics
        """
        return iter(self._usages)

    def __len__(self) -> int:
        """Get the number of models being tracked.

        Returns:
            int: Number of models with usage statistics
        """
        return len(self._usages)

    def __contains__(self, key: object) -> bool:
        return key in self._usages

    @property
    def requests(self) -> int:
        """Get the total number of requests across all models.

        Returns:
            int: Total number of API requests made across all models
        """
        return sum(usage.requests for usage in self._usages.values())

    @property
    def request_tokens(self) -> int:
        """Get the total number of request tokens across all models.

        Returns:
            int: Total number of tokens used in requests across all models
        """
        return sum(usage.request_tokens or 0 for usage in self._usages.values())

    @property
    def response_tokens(self) -> int:
        """Get the total number of response tokens across all models.

        Returns:
            int: Total number of tokens in model responses across all models
        """
        return sum(usage.response_tokens or 0 for usage in self._usages.values())

    @property
    def total_tokens(self) -> int:
        """Get the total number of tokens across all models.

        Returns:
            int: Total number of tokens used (requests + responses) across all models
        """
        return sum(usage.total_tokens or 0 for usage in self._usages.values())

request_tokens property

Get the total number of request tokens across all models.

Returns:

Name Type Description
int int

Total number of tokens used in requests across all models

requests property

Get the total number of requests across all models.

Returns:

Name Type Description
int int

Total number of API requests made across all models

response_tokens property

Get the total number of response tokens across all models.

Returns:

Name Type Description
int int

Total number of tokens in model responses across all models

total_tokens property

Get the total number of tokens across all models.

Returns:

Name Type Description
int int

Total number of tokens used (requests + responses) across all models

AgentUsageLimits

Bases: MutableMapping[str, UsageLimits]

Tracks usage limits for multiple models in an agent.

A dictionary-like container that maps model names to their UsageLimits. Automatically creates new UsageLimits entries for unknown models.

Source code in agenty/agent/usage.py 
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
class AgentUsageLimits(MutableMapping[str, UsageLimits]):
    """Tracks usage limits for multiple models in an agent.

    A dictionary-like container that maps model names to their UsageLimits.
    Automatically creates new UsageLimits entries for unknown models.
    """

    def __init__(self):
        """Initialize an empty UsageLimits tracker."""
        self._usage_limits: dict[str, UsageLimits] = {}

    def __getitem__(self, key: str) -> UsageLimits:
        """Get usage limits for a model, creating a new entry if needed.

        Args:
            key: Name of the model

        Returns:
            UsageLimits: Usage limits for the specified model. If the model doesn't
                        exist, a new UsageLimits instance is created and returned.
        """
        try:
            return self._usage_limits[key]
        except KeyError:
            self._usage_limits[key] = UsageLimits()
            return self._usage_limits[key]

    def __setitem__(self, key: str, value: UsageLimits) -> None:
        """Set UsageLimits for a model.

        Args:
            key: Name of the model
            value: UsageLimits to set for the model
        """
        self._usage_limits[key] = value

    def __delitem__(self, key: str) -> None:
        """Remove UsageLimits for a model.

        Args:
            key: Name of the model to remove

        Raises:
            KeyError: If the model doesn't exist
        """
        del self._usage_limits[key]

    def __iter__(self) -> Iterator[str]:
        """Iterate over model names.

        Returns:
            Iterator[str]: Iterator over model names with usage limits
        """
        return iter(self._usage_limits)

    def __len__(self) -> int:
        """Get the number of models being tracked.

        Returns:
            int: Number of models with UsageLimits
        """
        return len(self._usage_limits)

    def __contains__(self, key: object) -> bool:
        return key in self._usage_limits

ChatHistory

Bases: MutableSequence[ChatMessage]

Manages conversation history for an AI agent.

Implements MutableSequence for list-like access to message history. Handles conversation turns and optional history length limits.

Parameters:

Name Type Description Default
max_messages int

Max messages to keep (-1 for unlimited)

 -1
messages Optional[Sequence[ChatMessage]]

Optional initial messages

 None
Source code in agenty/agent/chat_history.py 
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class ChatHistory(MutableSequence[ChatMessage]):
    """Manages conversation history for an AI agent.

    Implements MutableSequence for list-like access to message history.
    Handles conversation turns and optional history length limits.

    Args:
        max_messages: Max messages to keep (-1 for unlimited)
        messages: Optional initial messages
    """

    def __init__(
        self, max_messages: int = -1, messages: Optional[Sequence[ChatMessage]] = None
    ) -> None:
        self._messages: list[ChatMessage] = list(messages) if messages else []
        self.max_messages = max_messages
        self.current_turn_id: Optional[str] = None

    def initialize_turn(self) -> None:
        """Start a new conversation turn with a fresh UUID.

        This method generates a new UUID for the current turn and sets it as the current_turn_id.
        All messages added after initializing a turn will share this turn_id until end_turn() is called.
        """
        self.current_turn_id = str(uuid.uuid4())

    def add(
        self,
        role: Role,
        content: AgentIO,
        name: Optional[str] = None,
        inject_name: bool = False,
    ) -> None:
        """Add a message to history.

        Args:
            role: Message sender's role
            content: Message content
            name: Optional sender name

        Note:
            Automatically initializes a new turn if none is active. Messages added in the same turn share a turn_id.
        """
        if self.current_turn_id is None:
            self.initialize_turn()

        message = ChatMessage(
            role=role,
            content=content,
            turn_id=self.current_turn_id,
            name=name,
        )
        message.set_inject_name(inject_name)
        self.append(message)

    def clear(self) -> None:
        """Clear all messages from memory and reset the current turn.

        This method removes all stored messages and resets the current_turn_id to None,
        effectively starting fresh with an empty memory state.
        """
        self.current_turn_id = None
        return super().clear()

    def _cull_history(self) -> None:
        """Remove oldest messages if exceeding max_messages.

        This is called automatically when adding new messages if max_messages
        is set to a non-negative value. Messages are removed from the start
        of the history (oldest first) until the length is within the max_messages limit.
        """
        if self.max_messages >= 0:
            while len(self._messages) > self.max_messages:
                self._messages.pop(0)

    def end_turn(self) -> None:
        """End current conversation turn.

        After calling this, the next message added will start a new turn
        with a new turn_id. This helps organize messages into logical groups
        or turns in the conversation.

        Note:
            This does not remove any messages, it only resets the current_turn_id
            to None so that the next message will start a new turn.
        """
        self.current_turn_id = None

    def to_openai(self, ctx: dict[str, Any] = {}) -> list[ChatCompletionMessageParam]:
        """Get history in OpenAI API format.

        Converts all messages in memory to the format expected by OpenAI's Chat API.
        Each message is rendered with the provided template context before conversion.

        Args:
            ctx: Template context dictionary for variable substitution. Used to render
                any template variables in message content.

        Returns:
            list[ChatCompletionMessageParam]: Messages formatted for OpenAI API, with each
                message containing role, content, and optional name fields.
        """
        return [msg.to_openai(ctx) for msg in self._messages]

    def to_pydantic_ai(
        self,
        ctx: dict[str, Any] = {},
    ) -> list[ModelMessage]:
        """Get history in Pydantic-AI format.

        Converts all messages in memory to the format expected by Pydantic-AI.
        Each message is rendered with the provided template context before conversion.

        Args:
            ctx: Template context dictionary for variable substitution. Used to render
                any template variables in message content.

        Returns:
            list[ModelMessage]: Messages formatted for Pydantic-AI, with each message
                converted to the appropriate ModelMessage subtype based on its role
                (ModelRequest for user/system, ModelResponse for assistant).

        Raises:
            ValueError: If a message has an unsupported role that cannot be converted
                to a Pydantic-AI message type.
        """
        # TODO: Do we really need to template the entire message history? Maybe just the last message?
        return [msg.to_pydantic_ai(ctx) for msg in self._messages]

    @overload
    def __getitem__(self, index: int) -> ChatMessage: ...

    @overload
    def __getitem__(self, index: slice) -> MutableSequence[ChatMessage]: ...

    def __getitem__(
        self, index: Union[int, slice]
    ) -> Union[ChatMessage, MutableSequence[ChatMessage]]:
        """Get message(s) by index/slice.

        Args:
            index: Integer index or slice

        Returns:
            Single message or sequence of messages
        """
        return self._messages[index]

    def __setitem__(
        self,
        index: Union[int, slice],
        value: Union[ChatMessage, Iterable[ChatMessage]],
    ) -> None:
        """Set message(s) at index/slice.

        Args:
            index: Integer index or slice
            value: Message or sequence of messages to set

        Raises:
            TypeError: If value type doesn't match index type
        """
        if isinstance(index, slice):
            adapter = TypeAdapter(List[ChatMessage])
            try:
                value = adapter.validate_python(value)
            except ValidationError:
                raise exc.AgentyTypeError("Can only assign sequence of ChatMessages")
            self._messages[index] = value
        else:
            if not isinstance(value, ChatMessage):
                raise exc.AgentyTypeError("Can only assign ChatMessage")
            self._messages[index] = value

    def __delitem__(self, index: Union[int, slice]) -> None:
        """Delete message(s) at index/slice.

        Args:
            index: Integer index or slice
        """
        del self._messages[index]

    def __len__(self) -> int:
        """Get number of messages in history.

        Returns:
            int: Number of messages
        """
        return len(self._messages)

    def insert(self, index: int, value: ChatMessage) -> None:
        """Insert message at index.

        Args:
            index: Position to insert at
            value: Message to insert

        Note:
            May trigger history culling if max_messages is exceeded
        """
        self._messages.insert(index, value)
        self._cull_history()

    def __repr__(self) -> str:
        """Get string representation of memory.

        Returns:
            str: String representation of memory
        """
        return f"AgentMemory({self._messages!r})"

    def __str__(self) -> str:
        """Get string representation of memory.

        Returns:
            str: String representation of memory
        """
        return f"AgentMemory({self._messages!r})"

    def __rich__(self) -> JSON:
        """Create a rich console representation of the model.

        Returns:
            JSON: Rich-formatted JSON representation
        """
        return JSON(json.dumps([msg.to_openai() for msg in self._messages]))

add(role, content, name=None, inject_name=False)

Add a message to history.

Parameters:

Name Type Description Default
role Role

Message sender's role

 required
content AgentIO

Message content

 required
name Optional[str]

Optional sender name

 None
Note

Automatically initializes a new turn if none is active. Messages added in the same turn share a turn_id.

Source code in agenty/agent/chat_history.py 
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
def add(
    self,
    role: Role,
    content: AgentIO,
    name: Optional[str] = None,
    inject_name: bool = False,
) -> None:
    """Add a message to history.

    Args:
        role: Message sender's role
        content: Message content
        name: Optional sender name

    Note:
        Automatically initializes a new turn if none is active. Messages added in the same turn share a turn_id.
    """
    if self.current_turn_id is None:
        self.initialize_turn()

    message = ChatMessage(
        role=role,
        content=content,
        turn_id=self.current_turn_id,
        name=name,
    )
    message.set_inject_name(inject_name)
    self.append(message)

clear()

Clear all messages from memory and reset the current turn.

This method removes all stored messages and resets the current_turn_id to None, effectively starting fresh with an empty memory state.

Source code in agenty/agent/chat_history.py 
196
197
198
199
200
201
202
203
def clear(self) -> None:
    """Clear all messages from memory and reset the current turn.

    This method removes all stored messages and resets the current_turn_id to None,
    effectively starting fresh with an empty memory state.
    """
    self.current_turn_id = None
    return super().clear()

end_turn()

End current conversation turn.

After calling this, the next message added will start a new turn with a new turn_id. This helps organize messages into logical groups or turns in the conversation.

Note

This does not remove any messages, it only resets the current_turn_id to None so that the next message will start a new turn.

Source code in agenty/agent/chat_history.py 
216
217
218
219
220
221
222
223
224
225
226
227
def end_turn(self) -> None:
    """End current conversation turn.

    After calling this, the next message added will start a new turn
    with a new turn_id. This helps organize messages into logical groups
    or turns in the conversation.

    Note:
        This does not remove any messages, it only resets the current_turn_id
        to None so that the next message will start a new turn.
    """
    self.current_turn_id = None

initialize_turn()

Start a new conversation turn with a fresh UUID.

This method generates a new UUID for the current turn and sets it as the current_turn_id. All messages added after initializing a turn will share this turn_id until end_turn() is called.

Source code in agenty/agent/chat_history.py 
159
160
161
162
163
164
165
def initialize_turn(self) -> None:
    """Start a new conversation turn with a fresh UUID.

    This method generates a new UUID for the current turn and sets it as the current_turn_id.
    All messages added after initializing a turn will share this turn_id until end_turn() is called.
    """
    self.current_turn_id = str(uuid.uuid4())

insert(index, value)

Insert message at index.

Parameters:

Name Type Description Default
index int

Position to insert at

 required
value ChatMessage

Message to insert

 required
Note

May trigger history culling if max_messages is exceeded

Source code in agenty/agent/chat_history.py 
331
332
333
334
335
336
337
338
339
340
341
342
def insert(self, index: int, value: ChatMessage) -> None:
    """Insert message at index.

    Args:
        index: Position to insert at
        value: Message to insert

    Note:
        May trigger history culling if max_messages is exceeded
    """
    self._messages.insert(index, value)
    self._cull_history()

to_openai(ctx={})

Get history in OpenAI API format.

Converts all messages in memory to the format expected by OpenAI's Chat API. Each message is rendered with the provided template context before conversion.

Parameters:

Name Type Description Default
ctx dict[str, Any]

Template context dictionary for variable substitution. Used to render any template variables in message content.

 {}

Returns:

Type Description
list[ChatCompletionMessageParam]

list[ChatCompletionMessageParam]: Messages formatted for OpenAI API, with each message containing role, content, and optional name fields.
Source code in agenty/agent/chat_history.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def to_openai(self, ctx: dict[str, Any] = {}) -> list[ChatCompletionMessageParam]:
    """Get history in OpenAI API format.

    Converts all messages in memory to the format expected by OpenAI's Chat API.
    Each message is rendered with the provided template context before conversion.

    Args:
        ctx: Template context dictionary for variable substitution. Used to render
            any template variables in message content.

    Returns:
        list[ChatCompletionMessageParam]: Messages formatted for OpenAI API, with each
            message containing role, content, and optional name fields.
    """
    return [msg.to_openai(ctx) for msg in self._messages]

to_pydantic_ai(ctx={})

Get history in Pydantic-AI format.

Converts all messages in memory to the format expected by Pydantic-AI. Each message is rendered with the provided template context before conversion.

Parameters:

Name Type Description Default
ctx dict[str, Any]

Template context dictionary for variable substitution. Used to render any template variables in message content.

 {}

Returns:

Type Description
list[ModelMessage]

list[ModelMessage]: Messages formatted for Pydantic-AI, with each message converted to the appropriate ModelMessage subtype based on its role (ModelRequest for user/system, ModelResponse for assistant).

Raises:

Type Description
ValueError

If a message has an unsupported role that cannot be converted to a Pydantic-AI message type.
Source code in agenty/agent/chat_history.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def to_pydantic_ai(
    self,
    ctx: dict[str, Any] = {},
) -> list[ModelMessage]:
    """Get history in Pydantic-AI format.

    Converts all messages in memory to the format expected by Pydantic-AI.
    Each message is rendered with the provided template context before conversion.

    Args:
        ctx: Template context dictionary for variable substitution. Used to render
            any template variables in message content.

    Returns:
        list[ModelMessage]: Messages formatted for Pydantic-AI, with each message
            converted to the appropriate ModelMessage subtype based on its role
            (ModelRequest for user/system, ModelResponse for assistant).

    Raises:
        ValueError: If a message has an unsupported role that cannot be converted
            to a Pydantic-AI message type.
    """
    # TODO: Do we really need to template the entire message history? Maybe just the last message?
    return [msg.to_pydantic_ai(ctx) for msg in self._messages]