Base AI interface¶

For developers only

This section is addressed to developers who want to extend the BaseAIInterface class.

End users should use classes such as PlanningAIInterface or ExecutionAIInterface (see AI interfaces).

The Base AI Interface "contract"¶

The general idea of an AI interface is to abstract the conversation between the agent and the LLM. You give it a list of messages, and it will return a payload object, which contains the response from the LLM, the cost, the finish reason, and potentially a structured output. This is useful if you want to create a multi-step agent: you can have an AI interface for the planning step, and another one for the execution step.

Anatomy of the `run` method¶

The BaseAIInterface class exposes two methods:

run: This method takes a list of ConversationAIMessage objects, and returns an AIInterfacePayload object.
arun: This method is the asynchronous version of the run method. It takes the same arguments as the run method.

In practice, these two functions essentially follow a loop with (roughly) the following steps:

The interface makes a prompt with make_prompt.
1. If this is the first turn of the conversation, the interface will use the make_first_prompt method.
2. Otherwise, it will use the make_new_prompt method.
The interface calls the model with the prompt, and gets a response.
That response is sent to the should_continue method to see if the conversation should continue.
If it should, the interface creates new conversation messages with make_new_messages. This is particularly useful if you need to generate tool response messages. These new messages are added to the conversation history, and we are back to step 1.
If the conversation should not continue, the interface creates a payload with the response, the cost, and the finish reason. It uses the extract_result method to extract the structured output from the response, if any.

The methods to override¶

As you can see, as a developer, you can choose to override the following methods:

make_prompt: This method is tasked with assembling the prompt for the AI model.
make_first_prompt: This method is tasked with assembling the prompt for the first turn of the conversation.
make_new_prompt: This method is tasked with assembling the prompt for the new turn of the conversation.
make_new_messages: This method is tasked with assembling the new conversation messages.
should_continue: This method is tasked with deciding whether the conversation should continue.
extract_result: This method is tasked with extracting the structured output from the response.

Implement make_prompt or make_first_prompt + make_new_prompt

make_prompt , if not overridden, will call either make_first_prompt or make_new_prompt, depending on whether this is the first turn of the conversation or not. By default, these two methods raise a NotImplementedError .

This means you have a choice:

Implement make_prompt and leave the other methods empty. This means that you will have to determine whether this is the first turn of the conversation or not, and assemble the prompt accordingly.
Implement make_first_prompt and make_new_prompt and leave make_prompt empty.

The AI Interface Payload¶

The AIInterfacePayload class is the class that contains the response from the LLM, the cost, and the finish reason. Through its result attribute, it can also contain a specific structured output schema.

conatus.agents.ai_interfaces.base.AIInterfacePayload `dataclass` ¶

AIInterfacePayload(
    cost: float,
    finish_reason: AIInterfaceRunFinishReason | None,
    result: Result,
    response: AIResponse[Result] | AIResponse,
    state: RuntimeState,
)

Bases: Generic[Result]

The payload of a run of a simple AI interface.

cost `instance-attribute` ¶

cost: float

The cost of the run.

finish_reason `instance-attribute` ¶

finish_reason: AIInterfaceRunFinishReason | None

The reason the run finished.

result `instance-attribute` ¶

result: Result

The result of the run.

response `instance-attribute` ¶

response: AIResponse[Result] | AIResponse

The response from the AI interface.

state `instance-attribute` ¶

state: RuntimeState

The state of the runtime.

The AI Interface base class¶

conatus.agents.ai_interfaces.base.BaseAIInterface ¶

BaseAIInterface(
    *,
    task_config: ConsolidatedTaskConfig | None = None,
    run_writer: FileWriter | None = None,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
    interface_name: str | None = None,
    max_turns: int | None = None,
    runtime: Runtime | None = None,
    hide_runtime_from_ai: bool | None = None,
    actions: (
        Collection[Action]
        | Sequence[RawAction]
        | ActionStarterPack
        | None
    ) = None,
    computer_use_mode: bool | None = None,
    only_keep_one_computer_use_environment: (
        bool | None
    ) = None,
    stop_if_no_tool_calls: bool = True,
    **kwargs: ParamType
)

Bases: Generic[Result]

Base class for AI interfaces that are not linked to tasks or runtimes.

PARAMETER	DESCRIPTION
`task_config`	The task configuration of the agent. TYPE: `ConsolidatedTaskConfig \| None` DEFAULT: `None`
`run_writer`	The writer used to log run information. TYPE: `FileWriter \| None` DEFAULT: `None`
`model_config`	The configuration for the model. TYPE: `ModelConfig \| ModelConfigTD \| None` DEFAULT: `None`
`model_type`	The type of model to use for the AI interface. If `None`, we will attempt to retrieve it from a class attribute. TYPE: `ModelType \| None` DEFAULT: `None`
`interface_name`	The name of the AI interface. Will otherwise be the snake case of the class name. TYPE: `str \| None` DEFAULT: `None`
`max_turns`	The maximum number of turns the AI interface can take. Defaults to 25, or whatever is the class attribute. TYPE: `int \| None` DEFAULT: `None`
`runtime`	The runtime of the agent. TYPE: `Runtime \| None` DEFAULT: `None`
`hide_runtime_from_ai`	Whether to hide the runtime from the AI. Defaults to `True` for normal AI interfaces, and `False` for `BaseAIInterfaceWithTask`s. TYPE: `bool \| None` DEFAULT: `None`
`actions`	The actions of the agent. Note that this is only used if the runtime is not provided. TYPE: `Collection[Action] \| Sequence[RawAction] \| ActionStarterPack \| None` DEFAULT: `None`
`computer_use_mode`	Whether to use computer use mode. Defaults to `False`, unless a subclass overrides this. TYPE: `bool \| None` DEFAULT: `None`
`only_keep_one_computer_use_environment`	Whether to only keep one computer use environment. Defaults to `True`, unless a subclass overrides this. TYPE: `bool \| None` DEFAULT: `None`
`stop_if_no_tool_calls`	Whether to stop the run if no tool calls are made. Defaults to `True`. TYPE: `bool` DEFAULT: `True`
`kwargs`	Additional parameters for the AI interface. TYPE: `ParamType` DEFAULT: `{}`

Source code in conatus/agents/ai_interfaces/base.py

def __init__(  # noqa: PLR0913
    self,
    *,
    task_config: ConsolidatedTaskConfig | None = None,
    run_writer: FileWriter | None = None,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
    interface_name: str | None = None,
    max_turns: int | None = None,
    runtime: Runtime | None = None,
    hide_runtime_from_ai: bool | None = None,
    actions: Collection[Action]
    | Sequence[RawAction]
    | ActionStarterPack
    | None = None,
    computer_use_mode: bool | None = None,
    only_keep_one_computer_use_environment: bool | None = None,
    stop_if_no_tool_calls: bool = True,
    **kwargs: ParamType,
) -> None:
    """Initialize the AI interface.

    Args:
        task_config: The task configuration of the agent.
        run_writer: The writer used to log run information.
        model_config: The configuration for the model.
        model_type: The type of model to use for the AI interface. If
            `None`, we will attempt to retrieve it from a class attribute.
        interface_name: The name of the AI interface. Will otherwise be
            the snake case of the class name.
        max_turns: The maximum number of turns the AI interface can take.
            Defaults to 25, or whatever is the class attribute.
        runtime: The runtime of the agent.
        hide_runtime_from_ai: Whether to hide the runtime from the AI.
            Defaults to `True` for normal AI interfaces, and `False` for
            [`BaseAIInterfaceWithTask`][conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask]s.
        actions: The actions of the agent. Note that this is only used if
            the runtime is not provided.
        computer_use_mode: Whether to use computer use mode. Defaults to
            `False`, unless a subclass overrides this.
        only_keep_one_computer_use_environment: Whether to only keep one
            computer use environment. Defaults to `True`, unless a subclass
            overrides this.
        stop_if_no_tool_calls: Whether to stop the run if no tool calls are
            made. Defaults to `True`.
        kwargs: Additional parameters for the AI interface.
    """
    # We put kwargs in the function signature to allow subclasses to
    # define additional parameters as they see fit.
    _ = kwargs
    model_type = model_type or getattr(self, "model_type", None)
    task_config = task_config or ConsolidatedTaskConfig()
    cls_stream_val = getattr(self, "stream", None)
    self.stream = (
        cls_stream_val if cls_stream_val is not None else task_config.stream
    )
    self.model = self.retrieve_model(
        task_config=task_config,
        model_config=model_config,
        model_type=model_type,
    )
    self.run_writer = run_writer
    self.interface_name = (
        interface_name
        or getattr(self, "interface_name", None)
        or to_snake_case(self.__class__.__name__)
    )
    self.hide_runtime_from_ai = (
        hide_runtime_from_ai
        if hide_runtime_from_ai is not None
        else getattr(self, "hide_runtime_from_ai", True)
    )
    self.runtime = runtime or Runtime(
        actions=actions or [], hide_from_ai=self.hide_runtime_from_ai
    )
    self.prompt_history = []
    self.response_history = []
    self.turn_count = 0
    self.max_turns = max_turns or int(
        getattr(self, "max_turns", DEFAULT_MAX_TURNS)
    )
    self.model_type = model_type
    self.finish_reason = None
    self.conversation_history = []
    self.conversation_history_id = None
    self.system_message = None
    self.cost = 0
    self.computer_use_mode = (
        computer_use_mode
        if computer_use_mode is not None
        else getattr(self, "computer_use_mode", False)
    )
    self.only_keep_one_computer_use_environment = (
        only_keep_one_computer_use_environment
        if only_keep_one_computer_use_environment is not None
        else getattr(self, "only_keep_one_computer_use_environment", True)
    )
    self._tool_response_buffer = []
    self.stop_if_no_tool_calls = stop_if_no_tool_calls

stream `instance-attribute` ¶

stream: bool = (
    cls_stream_val if cls_stream_val is not None else stream
)

Whether the AI interface streams the response.

If True, the AI interface will stream the response to the user.

model `instance-attribute` ¶

model: BaseAIModel = retrieve_model(
    task_config=task_config,
    model_config=model_config,
    model_type=model_type,
)

The model to use for the AI interface.

run_writer `instance-attribute` ¶

run_writer: FileWriter | None = run_writer

The writer to use for the AI interface.

If None, no logging will be done.

interface_name `instance-attribute` ¶

interface_name: str = (
    interface_name
    or getattr(self, "interface_name", None)
    or to_snake_case(__name__)
)

The name of the AI interface as written in the log files.

hide_runtime_from_ai `instance-attribute` ¶

hide_runtime_from_ai: bool = (
    hide_runtime_from_ai
    if hide_runtime_from_ai is not None
    else getattr(self, "hide_runtime_from_ai", True)
)

Whether to hide the runtime from the AI.

If True, the runtime will not be shown to the AI.

runtime `instance-attribute` ¶

runtime: Runtime = runtime or Runtime(
    actions=actions or [], hide_from_ai=hide_runtime_from_ai
)

The runtime of the agent.

prompt_history `instance-attribute` ¶

prompt_history: list[AIPrompt[Result] | AIPrompt] = []

The history of prompts, which can be used to track the conversation.

response_history `instance-attribute` ¶

response_history: list[AIResponse[Result] | AIResponse] = []

The history of responses, which can be used to track the conversation.

turn_count `instance-attribute` ¶

turn_count: int = 0

The number of turns so far. Starts at 0.

max_turns `instance-attribute` ¶

max_turns: int = max_turns or int(
    getattr(self, "max_turns", DEFAULT_MAX_TURNS)
)

The maximum number of turns the AI interface can take.

Defaults to 25, but can be overridden by the subclass or the user in the constructor.

model_type `instance-attribute` ¶

model_type: ModelType | None = model_type

The type of model to use for the AI interface.

If None, the model type will be inferred from the model class.

finish_reason `instance-attribute` ¶

finish_reason: AIInterfaceRunFinishReason | None = None

The reason the run finished.

If None, the run is still ongoing.

conversation_history `instance-attribute` ¶

conversation_history: list[ConversationAIMessage] = []

The history of the conversation.

Note that this can include messages generated outside of the instance, if the user wants to pass them to the AI interface.

conversation_history_id `instance-attribute` ¶

conversation_history_id: str | None = None

The ID of the conversation history.

This is used to link the conversation history to the response. Useful when calling stateful AI interfaces.

system_message `instance-attribute` ¶

system_message: SystemAIMessage | None = None

The system message of the conversation.

If the user passes a system message with run , it will be stored here. At every turn of the conversation, the system message will be updated based on what's in the generated prompt.

cost `instance-attribute` ¶

cost: float = 0

The cost of the run so far.

computer_use_mode `instance-attribute` ¶

computer_use_mode: bool = (
    computer_use_mode
    if computer_use_mode is not None
    else getattr(self, "computer_use_mode", False)
)

Whether the AI interface uses computer use actions.

If this is True, the interface will do the following:

It will only show the tool calls that are compatible with the computer-use action.
It will use a computer use model (openai:computer-use-preview for now).
It will strive to keep the reasoning traces and pass it back to the conversation.

only_keep_one_computer_use_environment `instance-attribute` ¶

only_keep_one_computer_use_environment: bool = (
    only_keep_one_computer_use_environment
    if only_keep_one_computer_use_environment is not None
    else getattr(
        self, "only_keep_one_computer_use_environment", True
    )
)

Whether to only keep one computer use environment.

If this is True, the interface will only keep one computer use environment at a time. This means that, if for example, multiple browsers are open, the interface will only use the one that was created last.

stop_if_no_tool_calls `instance-attribute` ¶

stop_if_no_tool_calls: bool = stop_if_no_tool_calls

Whether to stop the run if no tool calls are made.

If True, the run will stop if no tool calls are made. Otherwise, we send an empty tool response to the AI.

latest_prompt `property` ¶

latest_prompt: AIPrompt[Result] | AIPrompt | None

The latest prompt.

latest_response `property` ¶

latest_response: AIResponse[Result] | AIResponse | None

The latest response.

retrieve_model `staticmethod` ¶

retrieve_model(
    task_config: ConsolidatedTaskConfig,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
) -> BaseAIModel

Retrieve the model for the AI interface.

But a user may want to override this in the TaskConfig object, so that they can specify a different model for the AI interface.

This method is tasked with retrieving the model, using the following order of precedence:

The model class retrieved from task_config.preferred_model .
The model class retrieved from task_config.preferred_provider .

It will then instantiate the model with the given configuration.

PARAMETER	DESCRIPTION
`task_config`	The task configuration of the agent. TYPE: `ConsolidatedTaskConfig`
`model_config`	The configuration for the model. TYPE: `ModelConfig \| ModelConfigTD \| None` DEFAULT: `None`
`model_type`	The type of model to use. This is useful if you need to differentiate between different types of models. Ignored if the user provides a preferred model. TYPE: `ModelType \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`BaseAIModel`	The model to use for the AI interface.

Source code in conatus/agents/ai_interfaces/base.py

@staticmethod
def retrieve_model(
    task_config: ConsolidatedTaskConfig,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
) -> BaseAIModel:
    """Retrieve the model for the AI interface.

    But a user may want to override this in the [`TaskConfig`
    ][conatus.tasks.config.TaskConfig] object, so that they can
    specify a different model for the AI interface.

    This method is tasked with retrieving the model, using the
    following order of precedence:

    - The model class retrieved from [`task_config.preferred_model`
      ][conatus.tasks.config.TaskConfig.preferred_model].
    - The model class retrieved from [`task_config.preferred_provider`
      ][conatus.tasks.config.TaskConfig.preferred_provider].

    It will then instantiate the model with the given configuration.

    Args:
        task_config: The task configuration of the agent.
        model_config: The configuration for the model.
        model_type: The type of model to use. This is useful if you need
            to differentiate between different types of models. Ignored
            if the user provides a preferred model.

    Returns:
        The model to use for the AI interface.
    """
    preferred_model = task_config.user_provided.get("preferred_model")
    if isinstance(preferred_model, str):
        return get_model_from_name(preferred_model, model_config)
    if isinstance(preferred_model, dict):
        return preferred_model.with_config(model_config)

    preferred_provider: str | None = task_config.user_provided.get(
        "preferred_provider"
    )
    if preferred_provider:
        return get_model_from_provider(
            provider_name=cast("ProviderName", preferred_provider),
            model_config=model_config,
            model_type=model_type,
        )

    attempt_model_name: str | None = None
    if isinstance(model_config, ModelConfig):
        attempt_model_name = getattr(model_config, "model_name", None)
    elif isinstance(model_config, dict):
        attempt_model_name = model_config.get("model_name")
    if attempt_model_name:
        return get_model_from_name(attempt_model_name, model_config)

    # If no model class is provided, we use the default model class.
    from conatus.models.open_ai import OpenAIModel

    return OpenAIModel(model_config=model_config, model_type=model_type)

make_prompt ¶

make_prompt(
    *,
    conversation_history: Collection[
        ConversationAIMessage
    ] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: (
        SystemAIMessage | None
    ) = None,
    previous_response: (
        AIResponse[Result] | AIResponse | None
    ) = None,
    new_messages: list[ConversationAIMessage] | None = None
) -> AIPrompt | AIPrompt[Result]

Make the prompt for the AI interface.

You should override this method. If you don't, you should then override the make_first_prompt and make_new_prompt methods.

PARAMETER	DESCRIPTION
`conversation_history`	The conversation history. TYPE: `Collection[ConversationAIMessage]` DEFAULT: `()`
`conversation_history_id`	The ID of the conversation history. TYPE: `str \| None` DEFAULT: `None`
`conversation_history_system_message`	The system message of the conversation history. TYPE: `SystemAIMessage \| None` DEFAULT: `None`
`previous_response`	The previous response from the AI. Normally, if you just care about the conversation history, you can ignore this. It's useful to retrieve things like structured outputs, cost, etc. TYPE: `AIResponse[Result] \| AIResponse \| None` DEFAULT: `None`
`new_messages`	The new messages to send to the AI. Similarly, they should be naturally be added to the conversation history, but if there is specific processing you need to do, that variable allows you to distinguish them specifically. TYPE: `list[ConversationAIMessage] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AIPrompt \| AIPrompt[Result]`	The prompt for the AI interface.

RAISES	DESCRIPTION
`ValueError`	If no previous response is found.

Source code in conatus/agents/ai_interfaces/base.py

def make_prompt(
    self,
    *,
    conversation_history: Collection[ConversationAIMessage] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: SystemAIMessage | None = None,
    previous_response: AIResponse[Result] | AIResponse | None = None,
    new_messages: list[ConversationAIMessage] | None = None,
) -> AIPrompt | AIPrompt[Result]:
    """Make the prompt for the AI interface.

    You should override this method. If you don't, you should then
    override the [`make_first_prompt`
    ][conatus.agents.ai_interfaces.base.BaseAIInterface.make_first_prompt]
    and [`make_new_prompt`
    ][conatus.agents.ai_interfaces.base.BaseAIInterface.make_new_prompt]
    methods.

    Args:
        conversation_history: The conversation history.
        conversation_history_id: The ID of the conversation history.
        conversation_history_system_message: The system message of the
            conversation history.
        previous_response: The previous response from the AI. Normally,
            if you just care about the conversation history, you can
            ignore this. It's useful to retrieve things like structured
            outputs, cost, etc.
        new_messages: The new messages to send to the AI. Similarly, they
            should be naturally be added to the conversation history, but
            if there is specific processing you need to do, that variable
            allows you to distinguish them specifically.

    Returns:
        The prompt for the AI interface.

    Raises:
        ValueError: If no previous response is found.
    """
    # IMPORTANT !!!
    #
    # NOTE: This code exists to call the two distinct methods below.
    # It is totally OK (in fact, for simpler cases, preferable) to
    # just implement scrap everything here and implement your own
    # logic.

    conversation_history_as_list: list[ConversationAIMessage] = (
        self.conversation_history or list(conversation_history)
    )

    if self.turn_count == 0:
        return self.make_first_prompt(
            conversation_history=conversation_history_as_list,
            conversation_history_id=conversation_history_id,
            conversation_history_system_message=conversation_history_system_message,
        )
    latest_response = self.latest_response or previous_response
    if latest_response is None:
        msg = "No previous response found."
        raise ValueError(msg)
    return self.make_new_prompt(
        conversation_history=conversation_history_as_list,
        conversation_history_id=conversation_history_id,
        conversation_history_system_message=latest_response.prompt.system_message,
        previous_response=latest_response,
        new_messages=new_messages or [],
    )

make_first_prompt ¶

make_first_prompt(
    *,
    conversation_history: list[ConversationAIMessage],
    conversation_history_id: str | None,
    conversation_history_system_message: (
        SystemAIMessage | None
    )
) -> AIPrompt | AIPrompt[Result]

Make the first prompt for the AI interface.

This method is used to make the first prompt for the AI interface. It is called by the BaseAIInterface.run method.

PARAMETER	DESCRIPTION
`conversation_history`	The conversation history. TYPE: `list[ConversationAIMessage]`
`conversation_history_id`	The ID of the conversation history. TYPE: `str \| None`
`conversation_history_system_message`	The system message of the conversation history. TYPE: `SystemAIMessage \| None`

RETURNS	DESCRIPTION
`AIPrompt \| AIPrompt[Result]`	The first prompt for the AI interface.

Source code in conatus/agents/ai_interfaces/base.py

def make_first_prompt(
    self,
    *,
    conversation_history: list[ConversationAIMessage],
    conversation_history_id: str | None,
    conversation_history_system_message: SystemAIMessage | None,
) -> AIPrompt | AIPrompt[Result]:  # pragma: no cover
    """Make the first prompt for the AI interface.

    This method is used to make the first prompt for the AI interface.
    It is called by the [`BaseAIInterface.run`
    ][conatus.agents.ai_interfaces.base.BaseAIInterface.run] method.

    Args:
        conversation_history: The conversation history.
        conversation_history_id: The ID of the conversation history.
        conversation_history_system_message: The system message of the
            conversation history.

    Returns:
        The first prompt for the AI interface.
    """
    _ = self, conversation_history, conversation_history_id
    _ = conversation_history_system_message
    msg = "You should implement this method."
    raise NotImplementedError(msg)

make_new_prompt ¶

make_new_prompt(
    *,
    conversation_history: list[ConversationAIMessage],
    conversation_history_id: str | None,
    conversation_history_system_message: (
        SystemAIMessage | None
    ),
    previous_response: AIResponse[Result] | AIResponse,
    new_messages: list[ConversationAIMessage]
) -> AIPrompt | AIPrompt[Result]

Make a new prompt for the AI interface.

This method is used to make a new prompt for the AI interface. It is called by the BaseAIInterface.run method.

PARAMETER	DESCRIPTION
`conversation_history`	The conversation history. TYPE: `list[ConversationAIMessage]`
`conversation_history_id`	The ID of the conversation history. TYPE: `str \| None`
`conversation_history_system_message`	The system message of the conversation history. TYPE: `SystemAIMessage \| None`
`previous_response`	The previous response received from the AI. Normally, if you just care about the conversation history, you can ignore this. It's useful to retrieve things like structured outputs, cost, etc. TYPE: `AIResponse[Result] \| AIResponse`
`new_messages`	The new messages to send to the AI. Similarly, they should be naturally be added to the conversation history, but if there is specific processing you need to do, that variable allows you to distinguish them specifically. TYPE: `list[ConversationAIMessage]`

RETURNS	DESCRIPTION
`AIPrompt \| AIPrompt[Result]`	The new prompt for the AI interface.

Source code in conatus/agents/ai_interfaces/base.py

def make_new_prompt(
    self,
    *,
    conversation_history: list[ConversationAIMessage],
    conversation_history_id: str | None,
    conversation_history_system_message: SystemAIMessage | None,
    previous_response: AIResponse[Result] | AIResponse,
    new_messages: list[ConversationAIMessage],
) -> AIPrompt | AIPrompt[Result]:  # pragma: no cover
    """Make a new prompt for the AI interface.

    This method is used to make a new prompt for the AI interface.
    It is called by the [`BaseAIInterface.run`
    ][conatus.agents.ai_interfaces.base.BaseAIInterface.run] method.

    Args:
        conversation_history: The conversation history.
        conversation_history_id: The ID of the conversation history.
        conversation_history_system_message: The system message of the
            conversation history.
        previous_response: The previous response received from the AI.
            Normally, if you just care about the conversation history, you
            can ignore this. It's useful to retrieve things like structured
            outputs, cost, etc.
        new_messages: The new messages to send to the AI. Similarly, they
            should be naturally be added to the conversation history, but
            if there is specific processing you need to do, that variable
            allows you to distinguish them specifically.

    Returns:
        The new prompt for the AI interface.
    """
    _ = self, conversation_history, conversation_history_id
    _ = conversation_history_system_message, previous_response
    _ = new_messages
    msg = "You should implement this method."
    raise NotImplementedError(msg)

make_new_messages `async` ¶

make_new_messages(
    response: AIResponse[Result] | AIResponse,
) -> list[ConversationAIMessage]

Make the new messages after a turn of the conversation.

This is particularly useful when dealing with a Runtime , as it allows the interface to generate tool response messages.

PARAMETER	DESCRIPTION
`response`	The response from the AI. TYPE: `AIResponse[Result] \| AIResponse`

RETURNS	DESCRIPTION
`list[ConversationAIMessage]`	The new messages after a turn of the conversation. (Do not include the response from the AI, as it will be automatically added to the conversation history.)

Source code in conatus/agents/ai_interfaces/base.py

async def make_new_messages(
    self, response: AIResponse[Result] | AIResponse
) -> list[ConversationAIMessage]:
    """Make the new messages after a turn of the conversation.

    This is particularly useful when dealing with a [`Runtime`
    ][conatus.runtime.Runtime], as it allows the interface to generate
    tool response messages.

    Args:
        response: The response from the AI.

    Returns:
        The new messages after a turn of the conversation. (Do not
            include the response from the AI, as it will be automatically
            added to the conversation history.)
    """
    _ = response
    tool_responses = self._tool_response_buffer
    self._tool_response_buffer = []
    return [*tool_responses]

extract_result ¶

extract_result(
    *,
    latest_response: AIResponse[Result] | AIResponse,
    latest_prompt: AIPrompt[Result] | AIPrompt,
    finish_reason: AIInterfaceRunFinishReason | None
) -> Result

Extract the result of the AI interface.

PARAMETER	DESCRIPTION
`latest_response`	The latest response received from the AI. TYPE: `AIResponse[Result] \| AIResponse`
`latest_prompt`	The latest prompt sent to the AI. TYPE: `AIPrompt[Result] \| AIPrompt`
`finish_reason`	The reason the run finished. Might be `None` if the run is not finished yet. TYPE: `AIInterfaceRunFinishReason \| None`

RETURNS	DESCRIPTION
`Result`	The processed result.

Source code in conatus/agents/ai_interfaces/base.py

def extract_result(
    self,
    *,
    latest_response: AIResponse[Result] | AIResponse,
    latest_prompt: AIPrompt[Result] | AIPrompt,
    finish_reason: AIInterfaceRunFinishReason | None,
) -> Result:
    """Extract the result of the AI interface.

    Args:
        latest_response: The latest response received from the AI.
        latest_prompt: The latest prompt sent to the AI.
        finish_reason: The reason the run finished. Might be `None` if the
            run is not finished yet.

    Returns:
        The processed result.
    """
    _ = self, latest_prompt, finish_reason
    return cast("Result", latest_response.result)

should_continue ¶

should_continue(
    response: AIResponse[Result] | AIResponse,
) -> bool

Whether the AI interface should continue.

By default, the AI interface will stop after twenty-five AI model calls, but this method can be overridden to change this behavior.

PARAMETER	DESCRIPTION
`response`	The response from the AI. TYPE: `AIResponse[Result] \| AIResponse`

RETURNS	DESCRIPTION
`bool`	Whether the AI interface should continue.

Source code in conatus/agents/ai_interfaces/base.py

def should_continue(
    self, response: AIResponse[Result] | AIResponse
) -> bool:
    """Whether the AI interface should continue.

    By default, the AI interface will stop after twenty-five AI model calls,
    but this method can be overridden to change this behavior.

    Args:
        response: The response from the AI.

    Returns:
        Whether the AI interface should continue.
    """
    _, tool_responses = self.runtime.run(
        response.code_snippets,
        response.tool_calls,
    )
    if not tool_responses and self.stop_if_no_tool_calls:
        self.finish_reason = "no_tool_calls"
        return False

    self._tool_response_buffer = [*tool_responses]
    if self.runtime.is_terminated:
        self.finish_reason = "runtime_terminated"
        return False

    if self.turn_count < self.max_turns:
        return True
    self.finish_reason = "max_turns"
    return False

run ¶

run(
    *,
    conversation_history: Collection[
        ConversationAIMessage
    ] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: (
        SystemAIMessage | None
    ) = None
) -> AIInterfacePayload[Result]

Run the AI interface.

This method abstracts away the logic of interacting with the AI models, and returns a AIInterfacePayload from which the Agent can extract the result, the cost of the interactions, etc.

If the user wishes to include a previous conversation, they can provide a list of ConversationAIMessage objects, and / or a string that will identify the conversation.

PARAMETER	DESCRIPTION
`conversation_history`	The conversation history, as a list of `ConversationAIMessage` objects. TYPE: `Collection[ConversationAIMessage]` DEFAULT: `()`
`conversation_history_id`	The ID of the conversation history. This should be used when dealing with stateful AI interfaces. TYPE: `str \| None` DEFAULT: `None`
`conversation_history_system_message`	The system message of the conversation history. TYPE: `SystemAIMessage \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AIInterfacePayload[Result]`	The payload of the run.

Source code in conatus/agents/ai_interfaces/base.py

def run(
    self,
    *,
    conversation_history: Collection[ConversationAIMessage] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: SystemAIMessage | None = None,
) -> AIInterfacePayload[Result]:
    """Run the AI interface.

    This method abstracts away the logic of interacting with the
    [AI models][conatus.models.base.BaseAIModel], and returns a
    [`AIInterfacePayload`][conatus.agents.ai_interfaces.base.AIInterfacePayload]
    from which the [`Agent`][conatus.agents.base.Agent] can extract
    the result, the cost of the interactions, etc.

    If the user wishes to include a previous conversation, they can
    provide a list of [`ConversationAIMessage`
    ][conatus.models.inputs_outputs.messages.ConversationAIMessage]
    objects, and / or a string that will identify the conversation.

    Args:
        conversation_history: The conversation history, as a list of
            [`ConversationAIMessage`
            ][conatus.models.inputs_outputs.messages.ConversationAIMessage]
            objects.
        conversation_history_id: The ID of the conversation history. This
            should be used when dealing with stateful AI interfaces.
        conversation_history_system_message: The system message of the
            conversation history.

    Returns:
        The payload of the run.
    """
    return run_async(
        self.arun(
            conversation_history,
            conversation_history_id,
            conversation_history_system_message,
        )
    )

arun `async` ¶

arun(
    conversation_history: Collection[
        ConversationAIMessage
    ] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: (
        SystemAIMessage | None
    ) = None,
) -> AIInterfacePayload[Result]

Run the AI interface asynchronously.

For more information, see BaseAIInterface.run .

PARAMETER	DESCRIPTION
`conversation_history`	The conversation history, as a list of `ConversationAIMessage` objects. TYPE: `Collection[ConversationAIMessage]` DEFAULT: `()`
`conversation_history_id`	The ID of the conversation history. This should be used when dealing with stateful AI interfaces. TYPE: `str \| None` DEFAULT: `None`
`conversation_history_system_message`	The system message of the conversation history. TYPE: `SystemAIMessage \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`AIInterfacePayload[Result]`	The response from the AI interface.

Source code in conatus/agents/ai_interfaces/base.py

async def arun(
    self,
    conversation_history: Collection[ConversationAIMessage] = (),
    conversation_history_id: str | None = None,
    conversation_history_system_message: SystemAIMessage | None = None,
) -> AIInterfacePayload[Result]:
    """Run the AI interface asynchronously.

    For more information, see [`BaseAIInterface.run`
    ][conatus.agents.ai_interfaces.base.BaseAIInterface.run].

    Args:
        conversation_history: The conversation history, as a list of
            [`ConversationAIMessage`
            ][conatus.models.inputs_outputs.messages.ConversationAIMessage]
            objects.
        conversation_history_id: The ID of the conversation history. This
            should be used when dealing with stateful AI interfaces.
        conversation_history_system_message: The system message of the
            conversation history.

    Returns:
        The response from the AI interface.
    """
    logger.debug("Running interface %s", self.interface_name)
    response: AIResponse[Result] | AIResponse | None = None
    prompt: AIPrompt[Result] | AIPrompt | None = None
    self.conversation_history = list(conversation_history)
    self.conversation_history_id = conversation_history_id
    self.system_message = conversation_history_system_message
    prompt, response = await self._run_first_turn(
        conversation_history=self.conversation_history,
        conversation_history_id=self.conversation_history_id,
        conversation_history_system_message=self.system_message,
    )
    while self.should_continue(response=response):
        new_messages = await self.make_new_messages(response=response)
        self.conversation_history.extend(new_messages)
        prompt, response = await self._run_new_turn(
            previous_response=response,
            new_messages=new_messages,
        )

    return AIInterfacePayload[Result](
        cost=self.cost,
        finish_reason=self.finish_reason,
        result=self.extract_result(
            latest_response=response,
            latest_prompt=prompt,
            finish_reason=self.finish_reason,
        ),
        response=response,
        state=self.runtime.state,
    )

get_tool_specifications ¶

get_tool_specifications(
    *, computer_use_mode: bool | None = None
) -> list[AIToolSpecification]

Get the tool specifications for the agent.

The tool specifications are extracted from the available actions in the runtime.

Note that if computer_use_mode is True, the tool specifications skip actions that conflict with the computer use mode, such as browser actions.

PARAMETER	DESCRIPTION
`computer_use_mode`	Whether to include the computer use actions. If `None`, the computer use mode will be determined by the `computer_use_mode` attribute of the AI interface. If this set to `True`, we will skip actions that are not compatible with the computer use mode (e.g. browser actions). TYPE: `bool \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[AIToolSpecification]`	A list of tool specifications.

Source code in conatus/agents/ai_interfaces/base.py

def get_tool_specifications(
    self, *, computer_use_mode: bool | None = None
) -> list[AIToolSpecification]:
    """Get the tool specifications for the agent.

    The tool specifications are extracted from the available actions in
    the runtime.

    Note that if `computer_use_mode` is `True`, the tool specifications
    skip actions that conflict with the computer use mode, such as
    browser actions.

    Args:
        computer_use_mode: Whether to include the computer use actions.
            If `None`, the computer use mode will be determined by the
            [`computer_use_mode`
            ][conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask.computer_use_mode]
            attribute of the AI interface. If this set to `True`, we will
            skip actions that are not compatible with the computer use mode
            (e.g. browser actions).

    Returns:
        A list of tool specifications.
    """
    if computer_use_mode is None:
        computer_use_mode = self.computer_use_mode
    return self.runtime.get_tool_specifications(
        computer_use_mode=computer_use_mode
    )

generate_prompt_response_callbacks ¶

generate_prompt_response_callbacks(
    *, response_callback_expects_chunks: bool = False
) -> tuple[Callable[[str], None], Callable[[str], None]]

Generate the callbacks for the prompt and response.

PARAMETER	DESCRIPTION
`response_callback_expects_chunks`	Whether the response callback expects chunks. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`Callable[[str], None]`	The callback for the prompt.
`Callable[[str], None]`	The callback for the response. If `response_callback_expects_chunks` is `True`, the callback will receive chunks of the response. Otherwise, it will receive the final response.

Source code in conatus/agents/ai_interfaces/base.py

def generate_prompt_response_callbacks(
    self,
    *,
    response_callback_expects_chunks: bool = False,
) -> tuple[Callable[[str], None], Callable[[str], None]]:
    """Generate the callbacks for the prompt and response.

    Args:
        response_callback_expects_chunks: Whether the response callback
            expects chunks.

    Returns:
        (Callable[[str], None]): The callback for the prompt.
        (Callable[[str], None]): The callback for the response. If
            `response_callback_expects_chunks` is `True`, the callback
            will receive chunks of the response. Otherwise, it will receive
            the final response.
    """

    def prompt_callback(raw_json_prompt: str) -> None:
        if self.run_writer:
            self.run_writer.write(
                data=json.dumps(json.loads(raw_json_prompt), indent=2),
                file_name=(
                    f"{self.interface_name}_{self.turn_count}_prompt.json"
                ),
                output_type=OutputType.DEBUG,
            )

    def response_callback(raw_json_response: str) -> None:
        ext = "jsonl" if response_callback_expects_chunks else "json"
        if self.run_writer:
            self.run_writer.write(
                data=raw_json_response + "\n",
                file_name=(
                    f"{self.interface_name}_{self.turn_count}_response.{ext}"
                ),
                output_type=OutputType.DEBUG,
                append=response_callback_expects_chunks,
            )

    return prompt_callback, response_callback

Hooking up the AI interface to a `Task`¶

So far, we have seen the BaseAIInterface class, which is the base class for all AI interfaces. The BaseAIInterfaceWithTask class is the base class for all AI interfaces that are linked to a Task.

There are a few differences between the two classes:

The BaseAIInterfaceWithTask expects a TaskDefinition and a TaskConfig object.
It exposes a long list of convenience methods to generate XML-like prompts. This is very useful to communicate with the LLM about the task and the variables. (See more below.)
While the BaseAIInterface will stop the run if no tool calls are made, the BaseAIInterfaceWithTask will not stop the run by default, but essentially coerce the AI to properly call the terminate action.

XML Convenience methods¶

The BaseAIInterfaceWithTask class provides a few convenience methods to generate XML-like prompts.

Not necessarily valid XML

These methods are not guaranteed to return valid XML. They are meant to be used as a guide to assemble the prompt.

These methods are:

get_docstrings_for_all_actions_xml: Get the docstrings for all actions in XML format.
get_task_inputs_xml: Get the task inputs in XML format.
get_task_outputs_xml: Get the task outputs in XML format.
get_task_description_xml: Get the task description in XML format.
get_task_definition_xml: Get the task definition in XML format (which is really just the task description, the task inputs, and the task outputs).
get_all_variables_xml: Get all variables in XML format.
get_steps_so_far_xml: Get the steps so far in XML format (including pseudocode).
get_last_step_xml: Get the last step in XML format.

You can use the custom_xml method to generate XML-like prompts for your own needs.

Other convenience methods¶

Other convenience methods are:

get_one_variable_repr: Get a string representation of a single variable.
get_tool_specifications: Get the tool specifications from the Runtime, with some filtering if computer_use_mode is enabled.

conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask ¶

BaseAIInterfaceWithTask(
    *,
    runtime: Runtime,
    task_definition: TaskDefinition,
    task_config: ConsolidatedTaskConfig,
    run_writer: FileWriter | None = None,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
    computer_use_mode: bool | None = None,
    only_keep_one_computer_use_environment: (
        bool | None
    ) = None,
    interface_name: str | None = None,
    max_turns: int | None = None,
    stop_if_no_tool_calls: bool = False,
    **kwargs: ParamType
)

Bases: BaseAIInterface[Result], ABC

Base class for AI interfaces that are linked to tasks.

PARAMETER	DESCRIPTION
`runtime`	The runtime state of the agent. TYPE: `Runtime`
`task_definition`	The task definition of the agent. TYPE: `TaskDefinition`
`task_config`	The task configuration of the agent. TYPE: `ConsolidatedTaskConfig`
`run_writer`	The writer used to log run information. If `None`, no logging will be done. TYPE: `FileWriter \| None` DEFAULT: `None`
`model_config`	The configuration for the model. TYPE: `ModelConfig \| ModelConfigTD \| None` DEFAULT: `None`
`model_type`	The type of model to use for the AI interface. If `None`, we will attempt to retrieve it from a class attribute. TYPE: `ModelType \| None` DEFAULT: `None`
`computer_use_mode`	Whether to use computer use mode. Defaults to `False`, unless a subclass overrides this. TYPE: `bool \| None` DEFAULT: `None`
`only_keep_one_computer_use_environment`	Whether to only keep one computer use environment. Defaults to `True`, unless a subclass overrides this. TYPE: `bool \| None` DEFAULT: `None`
`interface_name`	The name of the AI interface. Will otherwise be the snake case of the class name. TYPE: `str \| None` DEFAULT: `None`
`max_turns`	The maximum number of turns the AI interface can take. Defaults to 1, but can be overridden by the subclass or the user in the constructor. TYPE: `int \| None` DEFAULT: `None`
`stop_if_no_tool_calls`	Whether to stop the run if no tool calls are made. Defaults to `False`. TYPE: `bool` DEFAULT: `False`
`kwargs`	Additional parameters for the AI interface. TYPE: `ParamType` DEFAULT: `{}`

Source code in conatus/agents/ai_interfaces/base.py

def __init__(  # noqa: PLR0913
    self,
    *,
    runtime: Runtime,
    task_definition: TaskDefinition,
    task_config: ConsolidatedTaskConfig,
    run_writer: FileWriter | None = None,
    model_config: ModelConfig | ModelConfigTD | None = None,
    model_type: ModelType | None = None,
    computer_use_mode: bool | None = None,
    only_keep_one_computer_use_environment: bool | None = None,
    interface_name: str | None = None,
    max_turns: int | None = None,
    stop_if_no_tool_calls: bool = False,
    **kwargs: ParamType,
) -> None:
    """Initialize the AI interface.

    Args:
        runtime: The runtime state of the agent.
        task_definition: The task definition of the agent.
        task_config: The task configuration of the agent.
        run_writer: The writer used to log run information. If `None`,
            no logging will be done.
        model_config: The configuration for the model.
        model_type: The type of model to use for the AI interface. If
            `None`, we will attempt to retrieve it from a class attribute.
        computer_use_mode: Whether to use computer use mode. Defaults to
            `False`, unless a subclass overrides this.
        only_keep_one_computer_use_environment: Whether to only keep one
            computer use environment. Defaults to `True`, unless a subclass
            overrides this.
        interface_name: The name of the AI interface. Will otherwise be
            the snake case of the class name.
        max_turns: The maximum number of turns the AI interface can take.
            Defaults to 1, but can be overridden by the subclass or the
            user in the constructor.
        stop_if_no_tool_calls: Whether to stop the run if no tool calls are
            made. Defaults to `False`.
        kwargs: Additional parameters for the AI interface.
    """
    super().__init__(
        task_config=task_config,
        run_writer=run_writer,
        model_config=model_config,
        model_type=model_type,
        interface_name=interface_name,
        max_turns=max_turns,
        runtime=runtime,
        actions=None,
        computer_use_mode=computer_use_mode,
        only_keep_one_computer_use_environment=only_keep_one_computer_use_environment,
        stop_if_no_tool_calls=stop_if_no_tool_calls,
        hide_runtime_from_ai=False,
        **kwargs,
    )
    self.task_definition = task_definition
    self.task_config = task_config

task_definition `instance-attribute` ¶

task_definition: TaskDefinition = task_definition

The task definition of the agent.

task_config `instance-attribute` ¶

task_config: ConsolidatedTaskConfig = task_config

The task configuration of the agent.

filter_modified_variables ¶

filter_modified_variables(
    new_messages: list[ConversationAIMessage] | None = None,
) -> list[str]

From a list of messages, extract the names of the modified variables.

PARAMETER	DESCRIPTION
`new_messages`	The new messages. TYPE: `list[ConversationAIMessage] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[str]`	The names of the modified variables.

Source code in conatus/agents/ai_interfaces/base.py

def filter_modified_variables(
    self,
    new_messages: list[ConversationAIMessage] | None = None,
) -> list[str]:
    """From a list of messages, extract the names of the modified variables.

    Args:
        new_messages: The new messages.

    Returns:
        The names of the modified variables.
    """
    # We leave this here to offer the option to subclass this method
    # and use the `self` argument.
    _ = self
    modified_variables: list[str] = []
    if new_messages is None:
        return modified_variables
    for message in new_messages:
        if isinstance(message, ToolResponseAIMessage):
            modified_variables.extend(message.modified_variables or [])
    return modified_variables

get_docstrings_for_all_actions_xml ¶

get_docstrings_for_all_actions_xml(
    *, computer_use_mode: bool | None = None
) -> str

Get the docstrings for the task's actions in XML format.

You should retrieve something of the like:

<actions>
Here all the actions that can be used to perform the task.

[... More instructions ...]

Name: 'print_hello'

Description
-----------
Print hello.

Parameters
----------
user_name : str
    The name of the user to print hello to.

[... More actions ...]
</actions>

PARAMETER	DESCRIPTION
`computer_use_mode`	Whether to include the computer use actions. If `None`, the computer use mode will be determined by the `computer_use_mode` attribute of the AI interface. TYPE: `bool \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	The docstrings for the actions.

Source code in conatus/agents/ai_interfaces/base.py

def get_docstrings_for_all_actions_xml(
    self,
    *,
    computer_use_mode: bool | None = None,
) -> str:
    """Get the docstrings for the task's actions in XML format.

    You should retrieve something of the like:

    ```plaintext
    <actions>
    Here all the actions that can be used to perform the task.

    [... More instructions ...]

    Name: 'print_hello'

    Description
    -----------
    Print hello.

    Parameters
    ----------
    user_name : str
        The name of the user to print hello to.

    [... More actions ...]
    </actions>
    ```

    Args:
        computer_use_mode: Whether to include the computer use actions.
            If `None`, the computer use mode will be determined by the
            [`computer_use_mode`
            ][conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask.computer_use_mode]
            attribute of the AI interface.

    Returns:
        The docstrings for the actions.
    """
    if computer_use_mode is None:
        computer_use_mode = self.computer_use_mode
    docstrings = "\n\n".join(
        action.get_docstring(include_name=True)
        for action in self.task_definition.actions.values()
        if not (action.hide_if_computer_use_mode and computer_use_mode)
    )
    return (
        "<actions>\n"
        + TASK_DEFINITION_ACTIONS_EXPLANATION
        + f"{docstrings}\n</actions>"
    )

get_task_inputs_xml ¶

get_task_inputs_xml() -> str

Get the task definition inputs in XML format.

Note that this is not the inputs of the task, but the expected inputs of the task. In other words, we're not displaying the actual inputs of the task, but the inputs that the task expects. The actual inputs should be displayed in the variables section.

You should retrieve something of the like:

<task_inputs>
* 'user_name' of type '<class 'str'>' (The name of the user)
[... More inputs ...]
</task_inputs>

RETURNS	DESCRIPTION
`str`	The task inputs in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_task_inputs_xml(self) -> str:
    """Get the task definition inputs in XML format.

    Note that this is not the inputs of the task, but the expected
    inputs of the task. In other words, we're not displaying the actual
    inputs of the task, but the inputs that the task expects. The actual
    inputs should be displayed in the `variables` section.

    You should retrieve something of the like:

    ```plaintext
    <task_inputs>
    * 'user_name' of type '<class 'str'>' (The name of the user)
    [... More inputs ...]
    </task_inputs>
    ```

    Returns:
        The task inputs in XML format.
    """
    return (
        "<task_inputs>\n"
        + "\n".join(
            f"* '{name}' of type "
            f"'{process_typehint(type_hint, return_type='str')}'"
            f" ({description})"
            for name, (
                type_hint,
                _,
                _,
                description,
            ) in self.task_definition.inputs_with_positions.items()
        )
        + "\n</task_inputs>"
    )

get_starting_variables_xml ¶

get_starting_variables_xml(
    *, include_images: Literal[False] = False
) -> str

get_starting_variables_xml(
    *, include_images: Literal[True]
) -> list[UserAIMessageContentPart]

get_starting_variables_xml(
    *, include_images: bool = False
) -> str | list[UserAIMessageContentPart]

Get the task definition inputs in XML format.

Note that unlike get_task_inputs_xml this method will also include the starting variables in the representation.

RETURNS	DESCRIPTION
`str \| list[UserAIMessageContentPart]`	The task inputs in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_starting_variables_xml(
    self,
    *,
    include_images: bool = False,
) -> str | list[UserAIMessageContentPart]:
    """Get the task definition inputs in XML format.

    Note that unlike [`get_task_inputs_xml`
    ][conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask.get_task_inputs_xml]
    this method will also include the starting variables in the
    representation.

    Returns:
        The task inputs in XML format.
    """
    starting_variables = self.runtime.state.starting_variables
    return self._generate_variables_xml(
        include_images=include_images,
        include_variables=list(starting_variables.values()),
        tag_name="user_inputs",
        get_earliest_value=True,
    )

get_task_outputs_xml ¶

get_task_outputs_xml() -> str

Get the task definition outputs in XML format.

You should retrieve something of the like:

<task_outputs>
* 'result' of type '<class 'str'>' (The result of the task)
</task_outputs>

RETURNS	DESCRIPTION
`str`	The outputs for the task.

Source code in conatus/agents/ai_interfaces/base.py

def get_task_outputs_xml(self) -> str:
    """Get the task definition outputs in XML format.

    You should retrieve something of the like:

    ```plaintext
    <task_outputs>
    * 'result' of type '<class 'str'>' (The result of the task)
    </task_outputs>
    ```

    Returns:
        The outputs for the task.
    """
    return (
        "<task_outputs>\n"
        + "\n".join(
            f"* '{name}' of type "
            f"'{process_typehint(type_hint, return_type='str')}'"
            f" ({description})"
            for name, (
                type_hint,
                description,
            ) in self.task_definition.outputs_with_positions.items()
        )
        + "\n</task_outputs>"
    )

get_task_description_xml ¶

get_task_description_xml() -> str

Get the task description in XML format.

You should retrieve something of the like:

<task_description>
This is the original task definition, as passed by the user...
[... More preamble ...]
Here is the task definition:
[... Task definition ...]
</task_description>

RETURNS	DESCRIPTION
`str`	The task description in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_task_description_xml(self) -> str:
    """Get the task description in XML format.

    You should retrieve something of the like:

    ```plaintext
    <task_description>
    This is the original task definition, as passed by the user...
    [... More preamble ...]
    Here is the task definition:
    [... Task definition ...]
    </task_description>
    ```

    Returns:
        The task description in XML format.
    """
    return (
        "<task_description>\n"
        + TASK_DEFINITION_PROMPT_EXPLANATION
        + "\n"
        + self.task_definition.user_prompt
        + "\n</task_description>"
    )

get_task_definition_xml ¶

get_task_definition_xml() -> str

Get the task definition in XML format.

You should retrieve something of the like:

<task_definition>
<task_description>
[... Task description ...]
</task_description>
<task_inputs>
[... Task inputs ...]
</task_inputs>
<task_outputs>
[... Task outputs ...]
</task_outputs>
</task_definition>

RETURNS	DESCRIPTION
`str`	The task definition in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_task_definition_xml(self) -> str:
    """Get the task definition in XML format.

    You should retrieve something of the like:

    ```plaintext
    <task_definition>
    <task_description>
    [... Task description ...]
    </task_description>
    <task_inputs>
    [... Task inputs ...]
    </task_inputs>
    <task_outputs>
    [... Task outputs ...]
    </task_outputs>
    </task_definition>
    ```

    Returns:
        The task definition in XML format.
    """
    return (
        "<task_definition>\n"
        + self.get_task_description_xml()
        + "\n"
        + self.get_task_inputs_xml()
        + "\n"
        + self.get_task_outputs_xml()
        + "\n</task_definition>"
    )

get_one_variable_repr `staticmethod` ¶

get_one_variable_repr(
    variable: RuntimeVariable,
    *,
    get_earliest_value: bool = False
) -> str

Get the representation of a variable.

We only return the text representation of the variable.

You should retrieve something of the like:

* 'user_name' of type '<class 'str'>'
repr: 'John Doe'

PARAMETER	DESCRIPTION
`variable`	The variable to get the representation of. TYPE: `RuntimeVariable`
`get_earliest_value`	Whether to get the earliest value of the variable. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str`	The representation of the variable.

Source code in conatus/agents/ai_interfaces/base.py

@staticmethod
def get_one_variable_repr(
    variable: RuntimeVariable, *, get_earliest_value: bool = False
) -> str:
    """Get the representation of a variable.

    We only return the text representation of the variable.

    You should retrieve something of the like:

    ```plaintext
    * 'user_name' of type '<class 'str'>'
    repr: 'John Doe'
    ```

    Args:
        variable: The variable to get the representation of.
        get_earliest_value: Whether to get the earliest value of the
            variable.

    Returns:
        The representation of the variable.
    """
    value = (
        variable.value_repr_history[0][1][0]
        if get_earliest_value
        else variable.value_repr[0]
    )
    return (
        f"* '{variable.name}' of type "
        f"'{process_typehint(variable.type_hint, return_type='str')}'\n"
        f"repr: {value}"
    )

get_all_variables_xml ¶

get_all_variables_xml(
    *,
    include_images: Literal[False] = False,
    exclude_variables: list[RuntimeVariable] | None = None
) -> str

get_all_variables_xml(
    *,
    include_images: Literal[True],
    exclude_variables: list[RuntimeVariable] | None = None
) -> list[UserAIMessageContentPart]

get_all_variables_xml(
    *,
    include_images: bool = False,
    exclude_variables: list[RuntimeVariable] | None = None
) -> str | list[UserAIMessageContentPart]

Get the representation of all variables in XML format.

You should retrieve something of the like:

<variables>
* 'user_name' of type '<class 'str'>'
repr: 'John Doe'
[... More variables ...]
</variables>

PARAMETER	DESCRIPTION
`include_images`	Whether to include images in the representation. This means that the representation will be a list of `UserAIMessageContentPart` objects, each containing a text part and an image part. TYPE: `bool` DEFAULT: `False`
`exclude_variables`	The variables to exclude from the representation. TYPE: `list[RuntimeVariable] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`str \| list[UserAIMessageContentPart]`	The representation of all variables in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_all_variables_xml(
    self,
    *,
    include_images: bool = False,
    exclude_variables: list[RuntimeVariable] | None = None,
) -> str | list[UserAIMessageContentPart]:
    """Get the representation of all variables in XML format.

    You should retrieve something of the like:

    ```plaintext
    <variables>
    * 'user_name' of type '<class 'str'>'
    repr: 'John Doe'
    [... More variables ...]
    </variables>
    ```

    Args:
        include_images: Whether to include images in the representation.
            This means that the representation will be a list of
            [`UserAIMessageContentPart`][conatus.models.inputs_outputs.content_parts.UserAIMessageContentPart]
            objects, each containing a text part and an image part.
        exclude_variables: The variables to exclude from the representation.

    Returns:
        The representation of all variables in XML format.
    """
    return self._generate_variables_xml(
        include_images=include_images,
        exclude_variables=exclude_variables,
    )

get_modified_variables_xml ¶

get_modified_variables_xml(
    *,
    modified_variables: list[RuntimeVariable] | list[str],
    include_images: Literal[False] = False
) -> str

get_modified_variables_xml(
    *,
    modified_variables: list[RuntimeVariable] | list[str],
    include_images: Literal[True]
) -> list[UserAIMessageContentPart]

get_modified_variables_xml(
    *,
    modified_variables: list[RuntimeVariable] | list[str],
    include_images: bool = False
) -> str | list[UserAIMessageContentPart]

Get the modified variables in XML format.

You should retrieve something of the like:

<modified_variables>
* 'user_name' of type '<class 'str'>'
repr: 'John Doe'
[... More modified variables ...]
</modified_variables>

PARAMETER	DESCRIPTION
`modified_variables`	The variables to include in the representation. TYPE: `list[RuntimeVariable] \| list[str]`
`include_images`	Whether to include images in the representation. This means that the representation will be a list of `UserAIMessageContentPart` objects, each containing a text part and an image part. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str \| list[UserAIMessageContentPart]`	The modified variables in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_modified_variables_xml(
    self,
    *,
    modified_variables: list[RuntimeVariable] | list[str],
    include_images: bool = False,
) -> str | list[UserAIMessageContentPart]:
    """Get the modified variables in XML format.

    You should retrieve something of the like:

    ```plaintext
    <modified_variables>
    * 'user_name' of type '<class 'str'>'
    repr: 'John Doe'
    [... More modified variables ...]
    </modified_variables>
    ```

    Args:
        modified_variables: The variables to include in the representation.
        include_images: Whether to include images in the representation.
            This means that the representation will be a list of
            [`UserAIMessageContentPart`][conatus.models.inputs_outputs.content_parts.UserAIMessageContentPart]
            objects, each containing a text part and an image part.

    Returns:
        The modified variables in XML format.
    """
    return self._generate_variables_xml(
        include_images=include_images,
        include_variables=modified_variables,
        tag_name="modified_variables",
    )

get_steps_so_far_xml ¶

get_steps_so_far_xml(
    *, include_failed: bool = False
) -> str

Get the steps so far in XML format.

You should retrieve something of the like:

<steps_so_far>
These are the steps that have been executed so far.
[... More preamble ...]
Here are the steps so far:
[... Steps so far ...]
</steps_so_far>

PARAMETER	DESCRIPTION
`include_failed`	Whether to include failed steps. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str`	The steps so far in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_steps_so_far_xml(self, *, include_failed: bool = False) -> str:
    """Get the steps so far in XML format.

    You should retrieve something of the like:

    ```plaintext
    <steps_so_far>
    These are the steps that have been executed so far.
    [... More preamble ...]
    Here are the steps so far:
    [... Steps so far ...]
    </steps_so_far>
    ```

    Args:
        include_failed: Whether to include failed steps.

    Returns:
        The steps so far in XML format.
    """
    code = self.runtime.state.code(include_failed=include_failed)
    return (
        "<steps_so_far>\n"
        + STEPS_SO_FAR_EXPLANATION
        + "\n"
        + "```python\n"
        + code
        + "\n```\n"
        + "\n</steps_so_far>"
    )

get_last_step_xml ¶

get_last_step_xml() -> str

Get the last step in XML format.

You should retrieve something of the like:

<last_step>
This is the latest step of the task.
[... More preamble ...]
Code
-----
[... Code ...]
Messages
--------
stdout: [... stdout ...]
stderr: [... stderr ...]
</last_step>

RETURNS	DESCRIPTION
`str`	The last step in XML format.

Source code in conatus/agents/ai_interfaces/base.py

def get_last_step_xml(self) -> str:
    """Get the last step in XML format.

    You should retrieve something of the like:

    ```plaintext
    <last_step>
    This is the latest step of the task.
    [... More preamble ...]
    Code
    -----
    [... Code ...]
    Messages
    --------
    stdout: [... stdout ...]
    stderr: [... stderr ...]
    </last_step>
    ```

    Returns:
        The last step in XML format.
    """
    stdout = self.runtime.state.last_step.stdout
    stdout = (
        f"stdout: {stdout}" if stdout else "-- nothing printed to stdout --"
    )

    stderr = self.runtime.state.last_step.stderr
    stderr = (
        f"stderr: {stderr}" if stderr else "-- nothing printed to stderr --"
    )
    return (
        "<last_step>\n"
        + LAST_STEP_EXPLANATION
        + "\n"
        + "Code\n-----\n"
        + "```python\n"
        + (self.runtime.state.last_step.code(include_failed=True) or "")
        + "\n```\n"
        + "\nMessages\n-----"
        + f"\n{stdout}\n{stderr}\n</last_step>"
    )

custom_xml `staticmethod` ¶

custom_xml(text: str, tag: str) -> str

Get the text in XML format.

If you pass text="Hello, world!" and tag="greeting", you should retrieve something of the like:

<greeting>
Hello, world!
</greeting>

PARAMETER	DESCRIPTION
`text`	The text to get the XML representation of. TYPE: `str`
`tag`	The tag to use for the XML representation. TYPE: `str`

RETURNS	DESCRIPTION
`str`	The text in XML format.

Source code in conatus/agents/ai_interfaces/base.py

@staticmethod
def custom_xml(text: str, tag: str) -> str:
    """Get the text in XML format.

    If you pass `text="Hello, world!"` and `tag="greeting"`, you should
    retrieve something of the like:

    ```plaintext
    <greeting>
    Hello, world!
    </greeting>
    ```

    Args:
        text: The text to get the XML representation of.
        tag: The tag to use for the XML representation.

    Returns:
        The text in XML format.
    """
    return f"<{tag}>\n{text}\n</{tag}>"

Base AI interface¶

The Base AI Interface "contract"¶

Anatomy of the run method¶

The methods to override¶

The AI Interface Payload¶

conatus.agents.ai_interfaces.base.AIInterfacePayload dataclass ¶

cost instance-attribute ¶

finish_reason instance-attribute ¶

result instance-attribute ¶

response instance-attribute ¶

state instance-attribute ¶

The AI Interface base class¶

conatus.agents.ai_interfaces.base.BaseAIInterface ¶

stream instance-attribute ¶

model instance-attribute ¶

run_writer instance-attribute ¶

interface_name instance-attribute ¶

hide_runtime_from_ai instance-attribute ¶

runtime instance-attribute ¶

prompt_history instance-attribute ¶

response_history instance-attribute ¶

turn_count instance-attribute ¶

max_turns instance-attribute ¶

model_type instance-attribute ¶

finish_reason instance-attribute ¶

conversation_history instance-attribute ¶

conversation_history_id instance-attribute ¶

system_message instance-attribute ¶

cost instance-attribute ¶

computer_use_mode instance-attribute ¶

only_keep_one_computer_use_environment instance-attribute ¶

stop_if_no_tool_calls instance-attribute ¶

latest_prompt property ¶

latest_response property ¶

retrieve_model staticmethod ¶

make_prompt ¶

make_first_prompt ¶

make_new_prompt ¶

make_new_messages async ¶

extract_result ¶

should_continue ¶

run ¶

arun async ¶

get_tool_specifications ¶

generate_prompt_response_callbacks ¶

Hooking up the AI interface to a Task¶

XML Convenience methods¶

Other convenience methods¶

conatus.agents.ai_interfaces.base.BaseAIInterfaceWithTask ¶

task_definition instance-attribute ¶

task_config instance-attribute ¶

filter_modified_variables ¶

get_docstrings_for_all_actions_xml ¶

get_task_inputs_xml ¶

get_starting_variables_xml ¶

get_task_outputs_xml ¶

get_task_description_xml ¶

get_task_definition_xml ¶

get_one_variable_repr staticmethod ¶

get_all_variables_xml ¶

get_modified_variables_xml ¶

get_steps_so_far_xml ¶

get_last_step_xml ¶

custom_xml staticmethod ¶

Anatomy of the `run` method¶

conatus.agents.ai_interfaces.base.AIInterfacePayload `dataclass` ¶

cost `instance-attribute` ¶

finish_reason `instance-attribute` ¶

result `instance-attribute` ¶

response `instance-attribute` ¶

state `instance-attribute` ¶

stream `instance-attribute` ¶

model `instance-attribute` ¶

run_writer `instance-attribute` ¶

interface_name `instance-attribute` ¶

hide_runtime_from_ai `instance-attribute` ¶

runtime `instance-attribute` ¶

prompt_history `instance-attribute` ¶

response_history `instance-attribute` ¶

turn_count `instance-attribute` ¶

max_turns `instance-attribute` ¶

model_type `instance-attribute` ¶

finish_reason `instance-attribute` ¶

conversation_history `instance-attribute` ¶

conversation_history_id `instance-attribute` ¶

system_message `instance-attribute` ¶

cost `instance-attribute` ¶

computer_use_mode `instance-attribute` ¶

only_keep_one_computer_use_environment `instance-attribute` ¶

stop_if_no_tool_calls `instance-attribute` ¶

latest_prompt `property` ¶

latest_response `property` ¶

retrieve_model `staticmethod` ¶

make_new_messages `async` ¶

arun `async` ¶

Hooking up the AI interface to a `Task`¶

task_definition `instance-attribute` ¶

task_config `instance-attribute` ¶

get_one_variable_repr `staticmethod` ¶

custom_xml `staticmethod` ¶