| 1 | """Parse actions & format observations with toolcalls"""
|
| 2 |
|
| 3 | import json
|
| 4 | import time
|
| 5 |
|
| 6 | from jinja2 import StrictUndefined, Template
|
| 7 |
|
| 8 | from minisweagent.exceptions import FormatError
|
| 9 | from minisweagent.models.utils.openai_multimodal import expand_multimodal_content
|
| 10 |
|
| 11 | BASH_TOOL = {
|
| 12 | "type": "function",
|
| 13 | "function": {
|
| 14 | "name": "bash",
|
| 15 | "description": "Execute a bash command",
|
| 16 | "parameters": {
|
| 17 | "type": "object",
|
| 18 | "properties": {
|
| 19 | "command": {
|
| 20 | "type": "string",
|
| 21 | "description": "The bash command to execute",
|
| 22 | }
|
| 23 | },
|
| 24 | "required": ["command"],
|
| 25 | },
|
| 26 | },
|
| 27 | }
|
| 28 |
|
| 29 |
|
| 30 | def parse_toolcall_actions(tool_calls: list, *, format_error_template: str) -> list[dict]:
|
| 31 | """Parse tool calls from the response. Raises FormatError if unknown tool or invalid args."""
|
| 32 | if not tool_calls:
|
| 33 | raise FormatError(
|
| 34 | {
|
| 35 | "role": "user",
|
| 36 | "content": Template(format_error_template, undefined=StrictUndefined).render(
|
| 37 | error="No tool calls found in the response. Every response MUST include at least one tool call."
|
| 38 | ),
|
| 39 | "extra": {"interrupt_type": "FormatError"},
|
| 40 | }
|
| 41 | )
|
| 42 | actions = []
|
| 43 | for tool_call in tool_calls:
|
| 44 | error_msg = ""
|
| 45 | args = {}
|
| 46 | try:
|
| 47 | args = json.loads(tool_call.function.arguments)
|
| 48 | except Exception as e:
|
| 49 | error_msg = f"Error parsing tool call arguments: {e}. "
|
| 50 | if tool_call.function.name != "bash":
|
| 51 | error_msg += f"Unknown tool '{tool_call.function.name}'."
|
| 52 | if "command" not in args:
|
| 53 | error_msg += "Missing 'command' argument in bash tool call."
|
| 54 | if error_msg:
|
| 55 | raise FormatError(
|
| 56 | {
|
| 57 | "role": "user",
|
| 58 | "content": Template(format_error_template, undefined=StrictUndefined).render(
|
| 59 | error=error_msg.strip()
|
| 60 | ),
|
| 61 | "extra": {"interrupt_type": "FormatError"},
|
| 62 | }
|
| 63 | )
|
| 64 | actions.append({"command": args["command"], "tool_call_id": tool_call.id})
|
| 65 | return actions
|
| 66 |
|
| 67 |
|
| 68 | def format_toolcall_observation_messages(
|
| 69 | *,
|
| 70 | actions: list[dict],
|
| 71 | outputs: list[dict],
|
| 72 | observation_template: str,
|
| 73 | template_vars: dict | None = None,
|
| 74 | multimodal_regex: str = "",
|
| 75 | ) -> list[dict]:
|
| 76 | """Format execution outputs into tool result messages."""
|
| 77 | not_executed = {"output": "", "returncode": -1, "exception_info": "action was not executed"}
|
| 78 | padded_outputs = outputs + [not_executed] * (len(actions) - len(outputs))
|
| 79 | results = []
|
| 80 | for action, output in zip(actions, padded_outputs):
|
| 81 | content = Template(observation_template, undefined=StrictUndefined).render(
|
| 82 | output=output, **(template_vars or {})
|
| 83 | )
|
| 84 | msg = {
|
| 85 | "content": content,
|
| 86 | "extra": {
|
| 87 | "raw_output": output.get("output", ""),
|
| 88 | "returncode": output.get("returncode"),
|
| 89 | "timestamp": time.time(),
|
| 90 | "exception_info": output.get("exception_info"),
|
| 91 | **output.get("extra", {}),
|
| 92 | },
|
| 93 | }
|
| 94 | if "tool_call_id" in action:
|
| 95 | msg["tool_call_id"] = action["tool_call_id"]
|
| 96 | msg["role"] = "tool"
|
| 97 | else:
|
| 98 | msg["role"] = "user" # human issued commands
|
| 99 | if multimodal_regex:
|
| 100 | msg = expand_multimodal_content(msg, pattern=multimodal_regex)
|
| 101 | results.append(msg)
|
| 102 | return results
|
| 103 |
|