Skip to content

DefaultAgent

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 'MINI_SWE_AGENT_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 'MINI_SWE_AGENT_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 
59
60
61
62
63
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

config instance-attribute 

config = config_class(**kwargs)

messages instance-attribute 

messages: list[dict] = []

model instance-attribute 

model = model

env instance-attribute 

env = env

render_template 

render_template(template: str, **kwargs) -> str
Source code in src/minisweagent/agents/default.py 
65
66
67
def render_template(self, template: str, **kwargs) -> str:
    cs = asdict(self.config) | asdict(self.env.config) | asdict(self.model.config) | platform.uname()._asdict()
    return Template(template).render(**kwargs, **cs, **os.environ)

add_message 

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

run 

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

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

Source code in src/minisweagent/agents/default.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
def run(self, task: str) -> tuple[str, str]:
    """Run step() until agent is finished. Return exit status & message"""
    self.messages = []
    self.add_message("system", self.config.system_template)
    self.add_message("user", self.render_template(self.config.instance_template, task=task))
    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 
86
87
88
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 
90
91
92
93
94
95
96
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["content"])
    return response

get_observation 

get_observation(response: dict) -> dict

Execute the action and return the observation.

Source code in src/minisweagent/agents/default.py 
 98
 99
100
101
102
103
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 
105
106
107
108
109
110
def parse_action(self, response: dict) -> dict:
    """Parse the action from the message. Returns the action."""
    actions = re.findall(r"```bash\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 
112
113
114
115
116
117
118
119
120
121
122
123
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 
125
126
127
128
129
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()
    if lines and lines[0].strip() == "MINI_SWE_AGENT_FINAL_OUTPUT":
        raise Submitted("\n".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