Skip to content

DefaultAgent

DefaultAgent class

Full source code 
"""Basic agent class. See https://mini-swe-agent.com/latest/advanced/control_flow/ for visual explanation."""

import re
import subprocess
from collections.abc import Callable
from dataclasses import asdict, dataclass

from jinja2 import StrictUndefined, Template

from minisweagent import Environment, Model


@dataclass
class AgentConfig:
    # The default settings are the bare minimum to run the agent. Take a look at the config files for improved settings.
    system_template: str = "You are a helpful assistant that can do anything."
    instance_template: str = (
        "Your task: {{task}}. Please reply with a single shell command in triple backticks. "
        "To finish, the first line of the output of the shell command must be 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'."
    )
    timeout_template: str = (
        "The last command <command>{{action['action']}}</command> timed out and has been killed.\n"
        "The output of the command was:\n <output>\n{{output}}\n</output>\n"
        "Please try another command and make sure to avoid those requiring interactive input."
    )
    format_error_template: str = "Please always provide EXACTLY ONE action in triple backticks."
    action_observation_template: str = "Observation: {{output}}"
    step_limit: int = 0
    cost_limit: float = 3.0


class NonTerminatingException(Exception):
    """Raised for conditions that can be handled by the agent."""


class FormatError(NonTerminatingException):
    """Raised when the LM's output is not in the expected format."""


class ExecutionTimeoutError(NonTerminatingException):
    """Raised when the action execution timed out."""


class TerminatingException(Exception):
    """Raised for conditions that terminate the agent."""


class Submitted(TerminatingException):
    """Raised when the LM declares that the agent has finished its task."""


class LimitsExceeded(TerminatingException):
    """Raised when the agent has reached its cost or step limit."""


class DefaultAgent:
    def __init__(self, model: Model, env: Environment, *, config_class: Callable = AgentConfig, **kwargs):
        self.config = config_class(**kwargs)
        self.messages: list[dict] = []
        self.model = model
        self.env = env
        self.extra_template_vars = {}

    def render_template(self, template: str, **kwargs) -> str:
        template_vars = asdict(self.config) | self.env.get_template_vars() | self.model.get_template_vars()
        return Template(template, undefined=StrictUndefined).render(
            **kwargs, **template_vars, **self.extra_template_vars
        )

    def add_message(self, role: str, content: str, **kwargs):
        self.messages.append({"role": role, "content": content, **kwargs})

    def run(self, task: str, **kwargs) -> tuple[str, str]:
        """Run step() until agent is finished. Return exit status & message"""
        self.extra_template_vars |= {"task": task, **kwargs}
        self.messages = []
        self.add_message("system", self.render_template(self.config.system_template))
        self.add_message("user", self.render_template(self.config.instance_template))
        while True:
            try:
                self.step()
            except NonTerminatingException as e:
                self.add_message("user", str(e))
            except TerminatingException as e:
                self.add_message("user", str(e))
                return type(e).__name__, str(e)

    def step(self) -> dict:
        """Query the LM, execute the action, return the observation."""
        return self.get_observation(self.query())

    def query(self) -> dict:
        """Query the model and return the response."""
        if 0 < self.config.step_limit <= self.model.n_calls or 0 < self.config.cost_limit <= self.model.cost:
            raise LimitsExceeded()
        response = self.model.query(self.messages)
        self.add_message("assistant", **response)
        return response

    def get_observation(self, response: dict) -> dict:
        """Execute the action and return the observation."""
        output = self.execute_action(self.parse_action(response))
        observation = self.render_template(self.config.action_observation_template, output=output)
        self.add_message("user", observation)
        return output

    def parse_action(self, response: dict) -> dict:
        """Parse the action from the message. Returns the action."""
        actions = re.findall(r"```bash\s*\n(.*?)\n```", response["content"], re.DOTALL)
        if len(actions) == 1:
            return {"action": actions[0].strip(), **response}
        raise FormatError(self.render_template(self.config.format_error_template, actions=actions))

    def execute_action(self, action: dict) -> dict:
        try:
            output = self.env.execute(action["action"])
        except subprocess.TimeoutExpired as e:
            output = e.output.decode("utf-8", errors="replace") if e.output else ""
            raise ExecutionTimeoutError(
                self.render_template(self.config.timeout_template, action=action, output=output)
            )
        except TimeoutError:
            raise ExecutionTimeoutError(self.render_template(self.config.timeout_template, action=action, output=""))
        self.has_finished(output)
        return output

    def has_finished(self, output: dict[str, str]):
        """Raises Submitted exception with final output if the agent has finished its task."""
        lines = output.get("output", "").lstrip().splitlines(keepends=True)
        if lines and lines[0].strip() in ["MINI_SWE_AGENT_FINAL_OUTPUT", "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT"]:
            raise Submitted("".join(lines[1:]))

Understanding the control flow

Check out the control flow guide for a visual explanation of the agent's control flow following this picture:

Agent control flow

minisweagent.agents.default.AgentConfig dataclass 

AgentConfig(
    system_template: str = "You are a helpful assistant that can do anything.",
    instance_template: str = "Your task: {{task}}. Please reply with a single shell command in triple backticks. To finish, the first line of the output of the shell command must be 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'.",
    timeout_template: str = "The last command <command>{{action['action']}}</command> timed out and has been killed.\nThe output of the command was:\n <output>\n{{output}}\n</output>\nPlease try another command and make sure to avoid those requiring interactive input.",
    format_error_template: str = "Please always provide EXACTLY ONE action in triple backticks.",
    action_observation_template: str = "Observation: {{output}}",
    step_limit: int = 0,
    cost_limit: float = 3.0,
)

system_template class-attribute instance-attribute 

system_template: str = (
    "You are a helpful assistant that can do anything."
)

instance_template class-attribute instance-attribute 

instance_template: str = "Your task: {{task}}. Please reply with a single shell command in triple backticks. To finish, the first line of the output of the shell command must be 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'."

timeout_template class-attribute instance-attribute 

timeout_template: str = "The last command <command>{{action['action']}}</command> timed out and has been killed.\nThe output of the command was:\n <output>\n{{output}}\n</output>\nPlease try another command and make sure to avoid those requiring interactive input."

format_error_template class-attribute instance-attribute 

format_error_template: str = "Please always provide EXACTLY ONE action in triple backticks."

action_observation_template class-attribute instance-attribute 

action_observation_template: str = "Observation: {{output}}"

step_limit class-attribute instance-attribute 

step_limit: int = 0

cost_limit class-attribute instance-attribute 

cost_limit: float = 3.0

minisweagent.agents.default.DefaultAgent 

DefaultAgent(
    model: Model,
    env: Environment,
    *,
    config_class: Callable = AgentConfig,
    **kwargs,
)
Source code in src/minisweagent/agents/default.py 
57
58
59
60
61
62
def __init__(self, model: Model, env: Environment, *, config_class: Callable = AgentConfig, **kwargs):
    self.config = config_class(**kwargs)
    self.messages: list[dict] = []
    self.model = model
    self.env = env
    self.extra_template_vars = {}

config instance-attribute 

config = config_class(**kwargs)

messages instance-attribute 

messages: list[dict] = []

model instance-attribute 

model = model

env instance-attribute 

env = env

extra_template_vars instance-attribute 

extra_template_vars = {}

render_template 

render_template(template: str, **kwargs) -> str
Source code in src/minisweagent/agents/default.py 
64
65
66
67
68
def render_template(self, template: str, **kwargs) -> str:
    template_vars = asdict(self.config) | self.env.get_template_vars() | self.model.get_template_vars()
    return Template(template, undefined=StrictUndefined).render(
        **kwargs, **template_vars, **self.extra_template_vars
    )

add_message 

add_message(role: str, content: str, **kwargs)
Source code in src/minisweagent/agents/default.py 
70
71
def add_message(self, role: str, content: str, **kwargs):
    self.messages.append({"role": role, "content": content, **kwargs})

run 

run(task: str, **kwargs) -> tuple[str, str]

Run step() until agent is finished. Return exit status & message

Source code in src/minisweagent/agents/default.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def run(self, task: str, **kwargs) -> tuple[str, str]:
    """Run step() until agent is finished. Return exit status & message"""
    self.extra_template_vars |= {"task": task, **kwargs}
    self.messages = []
    self.add_message("system", self.render_template(self.config.system_template))
    self.add_message("user", self.render_template(self.config.instance_template))
    while True:
        try:
            self.step()
        except NonTerminatingException as e:
            self.add_message("user", str(e))
        except TerminatingException as e:
            self.add_message("user", str(e))
            return type(e).__name__, str(e)

step 

step() -> dict

Query the LM, execute the action, return the observation.

Source code in src/minisweagent/agents/default.py 
88
89
90
def step(self) -> dict:
    """Query the LM, execute the action, return the observation."""
    return self.get_observation(self.query())

query 

query() -> dict

Query the model and return the response.

Source code in src/minisweagent/agents/default.py 
92
93
94
95
96
97
98
def query(self) -> dict:
    """Query the model and return the response."""
    if 0 < self.config.step_limit <= self.model.n_calls or 0 < self.config.cost_limit <= self.model.cost:
        raise LimitsExceeded()
    response = self.model.query(self.messages)
    self.add_message("assistant", **response)
    return response

get_observation 

get_observation(response: dict) -> dict

Execute the action and return the observation.

Source code in src/minisweagent/agents/default.py 
100
101
102
103
104
105
def get_observation(self, response: dict) -> dict:
    """Execute the action and return the observation."""
    output = self.execute_action(self.parse_action(response))
    observation = self.render_template(self.config.action_observation_template, output=output)
    self.add_message("user", observation)
    return output

parse_action 

parse_action(response: dict) -> dict

Parse the action from the message. Returns the action.

Source code in src/minisweagent/agents/default.py 
107
108
109
110
111
112
def parse_action(self, response: dict) -> dict:
    """Parse the action from the message. Returns the action."""
    actions = re.findall(r"```bash\s*\n(.*?)\n```", response["content"], re.DOTALL)
    if len(actions) == 1:
        return {"action": actions[0].strip(), **response}
    raise FormatError(self.render_template(self.config.format_error_template, actions=actions))

execute_action 

execute_action(action: dict) -> dict
Source code in src/minisweagent/agents/default.py 
114
115
116
117
118
119
120
121
122
123
124
125
def execute_action(self, action: dict) -> dict:
    try:
        output = self.env.execute(action["action"])
    except subprocess.TimeoutExpired as e:
        output = e.output.decode("utf-8", errors="replace") if e.output else ""
        raise ExecutionTimeoutError(
            self.render_template(self.config.timeout_template, action=action, output=output)
        )
    except TimeoutError:
        raise ExecutionTimeoutError(self.render_template(self.config.timeout_template, action=action, output=""))
    self.has_finished(output)
    return output

has_finished 

has_finished(output: dict[str, str])

Raises Submitted exception with final output if the agent has finished its task.

Source code in src/minisweagent/agents/default.py 
127
128
129
130
131
def has_finished(self, output: dict[str, str]):
    """Raises Submitted exception with final output if the agent has finished its task."""
    lines = output.get("output", "").lstrip().splitlines(keepends=True)
    if lines and lines[0].strip() in ["MINI_SWE_AGENT_FINAL_OUTPUT", "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT"]:
        raise Submitted("".join(lines[1:]))

minisweagent.agents.default.NonTerminatingException

Bases: Exception

Raised for conditions that can be handled by the agent.

minisweagent.agents.default.TerminatingException

Bases: Exception

Raised for conditions that terminate the agent.

bug_report Something broken/unclear?

Open an issue on GitHub!
help Open-ended discussions

Join our Slack!
Our projects
SWE-agent SWE-agent SWE-rex SWE-ReX SWE-smith SWE-smith SWE-bench SWE-bench sb-cli sb-cli