diff --git a/openhands/core/exceptions.py b/openhands/core/exceptions.py index c33297a0d245..1a95dc2c44b5 100644 --- a/openhands/core/exceptions.py +++ b/openhands/core/exceptions.py @@ -94,3 +94,21 @@ class CloudFlareBlockageError(Exception): """Exception raised when a request is blocked by CloudFlare.""" pass + + +class FunctionCallConversionError(Exception): + """Exception raised when FunctionCallingConverter failed to convert a non-function call message to a function call message. + + This typically happens when there's a malformed message (e.g., missing tags). But not due to LLM output. + """ + + pass + + +class FunctionCallValidationError(Exception): + """Exception raised when FunctionCallingConverter failed to validate a function call message. + + This typically happens when the LLM outputs unrecognized function call / parameter names / values. + """ + + pass diff --git a/openhands/llm/fn_call_converter.py b/openhands/llm/fn_call_converter.py new file mode 100644 index 000000000000..134db199f995 --- /dev/null +++ b/openhands/llm/fn_call_converter.py @@ -0,0 +1,430 @@ +"""Convert function calling messages to non-function calling messages and vice versa. + +This will inject prompts so that models that doesn't support function calling +can still be used with function calling agents. + +We follow format from: https://docs.litellm.ai/docs/completion/function_call +""" + +import copy +import json +import re +from typing import Iterable + +from litellm import ChatCompletionToolParam + +from openhands.core.exceptions import ( + FunctionCallConversionError, + FunctionCallValidationError, +) + +# Inspired by: https://docs.together.ai/docs/llama-3-function-calling#function-calling-w-llama-31-70b +SYSTEM_PROMPT_SUFFIX_TEMPLATE = """ +You have access to the following functions: + +{description} + +If you choose to call a function ONLY reply in the following format with NO suffix: + + +value_1 + +This is the value for the second parameter +that can span +multiple lines + + + + +Reminder: +- Function calls MUST follow the specified format, start with +- Required parameters MUST be specified +- Only call one function at a time +- You may provide optional reasoning for your function call in natural language BEFORE the function call, but NOT after. +- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls +""" + +# Regex patterns for function call parsing +FN_REGEX_PATTERN = r']+)>\n(.*?)' +FN_PARAM_REGEX_PATTERN = r']+)>(.*?)' + +# Add new regex pattern for tool execution results +TOOL_RESULT_REGEX_PATTERN = r'EXECUTION RESULT of \[(.*?)\]:\n(.*)' + + +def convert_tool_calls_to_string(tool_calls: list[dict]) -> str: + """Convert tool calls to content in string format.""" + if not isinstance(tool_calls, list) or len(tool_calls) != 1: + raise FunctionCallConversionError( + 'Only one tool call is supported for FunctionCallingConverter.' + ) + tool_call = tool_calls[0] + if 'function' not in tool_call: + raise FunctionCallConversionError("Tool call must contain 'function' key.") + if 'index' not in tool_call: + raise FunctionCallConversionError("Tool call must contain 'index' key.") + if 'id' not in tool_call: + raise FunctionCallConversionError("Tool call must contain 'id' key.") + if 'type' not in tool_call: + raise FunctionCallConversionError("Tool call must contain 'type' key.") + if tool_call['type'] != 'function': + raise FunctionCallConversionError("Tool call type must be 'function'.") + + ret = f"\n" + try: + args = json.loads(tool_call['function']['arguments']) + except json.JSONDecodeError as e: + raise FunctionCallConversionError( + f"Failed to parse arguments as JSON. Arguments: {tool_call['function']['arguments']}" + ) from e + for param_name, param_value in args.items(): + is_multiline = '\n' in param_value + ret += f'' + if is_multiline: + ret += '\n' + ret += f'{param_value}' + if is_multiline: + ret += '\n' + ret += '\n' + ret += '' + return ret + + +def convert_tools_to_description(tools: list[ChatCompletionToolParam]) -> str: + ret = '' + for i, tool in enumerate(tools): + assert tool['type'] == 'function' + fn = tool['function'] + if i > 0: + ret += '\n' + ret += f"---- BEGIN FUNCTION #{i+1}: {fn['name']} ----\n" + ret += f"Description: {fn['description']}\n" + if 'parameters' in fn: + ret += f"Parameters: {json.dumps(fn['parameters'], indent=2)}\n" + else: + ret += 'No parameters are required for this function.\n' + ret += f'---- END FUNCTION #{i+1} ----\n' + return ret + + +def convert_fncall_messages_to_non_fncall_messages( + messages: list[dict], + tools: list[ChatCompletionToolParam], +) -> list[dict]: + """Convert function calling messages to non-function calling messages.""" + messages = copy.deepcopy(messages) + + formatted_tools = convert_tools_to_description(tools) + system_prompt_suffix = SYSTEM_PROMPT_SUFFIX_TEMPLATE.format( + description=formatted_tools + ) + + converted_messages = [] + for message in messages: + role, content = message['role'], message['content'] + # 1. SYSTEM MESSAGES + # append system prompt suffix to content + if role == 'system': + if isinstance(content, str): + content += system_prompt_suffix + elif isinstance(content, list): + if content and content[-1]['type'] == 'text': + content[-1]['text'] += system_prompt_suffix + else: + content.append({'type': 'text', 'text': system_prompt_suffix}) + else: + raise FunctionCallConversionError( + f'Unexpected content type {type(content)}. Expected str or list. Content: {content}' + ) + converted_messages.append({'role': 'system', 'content': content}) + # 2. USER MESSAGES (no change) + elif role == 'user': + converted_messages.append(message) + + # 3. ASSISTANT MESSAGES + # - 3.1 no change if no function call + # - 3.2 change if function call + elif role == 'assistant': + if 'tool_calls' in message: + # change content to function call + tool_content = convert_tool_calls_to_string(message['tool_calls']) + if isinstance(content, str): + content += '\n\n' + tool_content + elif isinstance(content, list): + if content and content[-1]['type'] == 'text': + content[-1]['text'] += '\n\n' + tool_content + else: + content.append({'type': 'text', 'text': tool_content}) + else: + raise FunctionCallConversionError( + f'Unexpected content type {type(content)}. Expected str or list. Content: {content}' + ) + converted_messages.append({'role': 'assistant', 'content': content}) + # 4. TOOL MESSAGES (tool outputs) + elif role == 'tool': + # Convert tool result as assistant message + prefix = f'EXECUTION RESULT of [{message["name"]}]:\n' + # and omit "tool_call_id" AND "name" + if isinstance(content, str): + content = prefix + content + elif isinstance(content, list): + if content and content[-1]['type'] == 'text': + content[-1]['text'] = prefix + content[-1]['text'] + else: + content = [{'type': 'text', 'text': prefix}] + content + else: + raise FunctionCallConversionError( + f'Unexpected content type {type(content)}. Expected str or list. Content: {content}' + ) + converted_messages.append({'role': 'user', 'content': content}) + else: + raise FunctionCallConversionError( + f'Unexpected role {role}. Expected system, user, assistant or tool.' + ) + return converted_messages + + +def _extract_and_validate_params( + matching_tool: dict, param_matches: Iterable[re.Match], fn_name: str +) -> dict: + params = {} + # Parse and validate parameters + required_params = set() + if 'parameters' in matching_tool and 'required' in matching_tool['parameters']: + required_params = set(matching_tool['parameters'].get('required', [])) + + allowed_params = set() + if 'parameters' in matching_tool and 'properties' in matching_tool['parameters']: + allowed_params = set(matching_tool['parameters']['properties'].keys()) + + param_name_to_type = {} + if 'parameters' in matching_tool and 'properties' in matching_tool['parameters']: + param_name_to_type = { + name: val.get('type', 'string') + for name, val in matching_tool['parameters']['properties'].items() + } + + # Collect parameters + found_params = set() + for param_match in param_matches: + param_name = param_match.group(1) + param_value = param_match.group(2).strip() + + # Validate parameter is allowed + if allowed_params and param_name not in allowed_params: + raise FunctionCallValidationError( + f"Parameter '{param_name}' is not allowed for function '{fn_name}'. " + f'Allowed parameters: {allowed_params}' + ) + + # Validate and convert parameter type + # supported: string, integer, array + if param_name in param_name_to_type: + if param_name_to_type[param_name] == 'integer': + try: + param_value = int(param_value) + except ValueError: + raise FunctionCallValidationError( + f"Parameter '{param_name}' is expected to be an integer." + ) + elif param_name_to_type[param_name] == 'array': + try: + param_value = json.loads(param_value) + except json.JSONDecodeError: + raise FunctionCallValidationError( + f"Parameter '{param_name}' is expected to be an array." + ) + else: + # string + pass + + # Enum check + if 'enum' in matching_tool['parameters']['properties'][param_name]: + if ( + param_value + not in matching_tool['parameters']['properties'][param_name]['enum'] + ): + raise FunctionCallValidationError( + f"Parameter '{param_name}' is expected to be one of {matching_tool['parameters']['properties'][param_name]['enum']}." + ) + + params[param_name] = param_value + found_params.add(param_name) + + # Check all required parameters are present + missing_params = required_params - found_params + if missing_params: + raise FunctionCallValidationError( + f"Missing required parameters for function '{fn_name}': {missing_params}" + ) + return params + + +def convert_non_fncall_messages_to_fncall_messages( + messages: list[dict], + tools: list[ChatCompletionToolParam], +) -> list[dict]: + """Convert non-function calling messages back to function calling messages.""" + messages = copy.deepcopy(messages) + formatted_tools = convert_tools_to_description(tools) + system_prompt_suffix = SYSTEM_PROMPT_SUFFIX_TEMPLATE.format( + description=formatted_tools + ) + + converted_messages = [] + tool_call_counter = 1 # Counter for tool calls + + for message in messages: + role, content = message['role'], message['content'] + + # For system messages, remove the added suffix + if role == 'system': + if isinstance(content, str): + # Remove the suffix if present + content = content.split(system_prompt_suffix)[0] + elif isinstance(content, list): + if content and content[-1]['type'] == 'text': + # Remove the suffix from the last text item + content[-1]['text'] = content[-1]['text'].split( + system_prompt_suffix + )[0] + converted_messages.append({'role': 'system', 'content': content}) + # Skip user messages (no conversion needed) + elif role == 'user': + # Check for tool execution result pattern + if isinstance(content, str): + tool_result_match = re.search( + TOOL_RESULT_REGEX_PATTERN, content, re.DOTALL + ) + elif isinstance(content, list): + tool_result_match = next( + ( + _match + for item in content + if item.get('type') == 'text' + and ( + _match := re.search( + TOOL_RESULT_REGEX_PATTERN, item['text'], re.DOTALL + ) + ) + ), + None, + ) + else: + raise FunctionCallConversionError( + f'Unexpected content type {type(content)}. Expected str or list. Content: {content}' + ) + + if tool_result_match: + if not ( + isinstance(content, str) + or ( + isinstance(content, list) + and len(content) == 1 + and content[0].get('type') == 'text' + ) + ): + raise FunctionCallConversionError( + f'Expected str or list with one text item when tool result is present in the message. Content: {content}' + ) + tool_name = tool_result_match.group(1) + tool_result = tool_result_match.group(2).strip() + + # Convert to tool message format + converted_messages.append( + { + 'role': 'tool', + 'name': tool_name, + 'content': [{'type': 'text', 'text': tool_result}] + if isinstance(content, list) + else tool_result, + 'tool_call_id': f'toolu_{tool_call_counter-1:02d}', # Use last generated ID + } + ) + else: + converted_messages.append(message) + + # Handle assistant messages + elif role == 'assistant': + if isinstance(content, str): + fn_match = re.search(FN_REGEX_PATTERN, content, re.DOTALL) + elif isinstance(content, list): + if content and content[-1]['type'] == 'text': + fn_match = re.search( + FN_REGEX_PATTERN, content[-1]['text'], re.DOTALL + ) + else: + fn_match = None + fn_match_exists = any( + item.get('type') == 'text' + and re.search(FN_REGEX_PATTERN, item['text'], re.DOTALL) + for item in content + ) + if fn_match_exists and not fn_match: + raise FunctionCallConversionError( + f'Expecting function call in the LAST index of content list. But got content={content}' + ) + else: + raise FunctionCallConversionError( + f'Unexpected content type {type(content)}. Expected str or list. Content: {content}' + ) + + if fn_match: + fn_name = fn_match.group(1) + fn_body = fn_match.group(2) + matching_tool = next( + ( + tool['function'] + for tool in tools + if tool['type'] == 'function' + and tool['function']['name'] == fn_name + ), + None, + ) + # Validate function exists in tools + if not matching_tool: + raise FunctionCallValidationError( + f"Function '{fn_name}' not found in available tools: {[tool['function']['name'] for tool in tools if tool['type'] == 'function']}" + ) + + # Parse parameters + param_matches = re.finditer(FN_PARAM_REGEX_PATTERN, fn_body, re.DOTALL) + params = _extract_and_validate_params( + matching_tool, param_matches, fn_name + ) + + # Create tool call with unique ID + tool_call_id = f'toolu_{tool_call_counter:02d}' + tool_call = { + 'index': 1, # always 1 because we only support **one tool call per message** + 'id': tool_call_id, + 'type': 'function', + 'function': {'name': fn_name, 'arguments': json.dumps(params)}, + } + tool_call_counter += 1 # Increment counter + + # Remove the function call part from content + if isinstance(content, list): + assert content and content[-1]['type'] == 'text' + content[-1]['text'] = ( + content[-1]['text'].split(' server.log 2>&1 &`.\n* Interactive: If a bash command returns exit code `-1`, this means the process is not yet finished. The assistant must then send a second call to terminal with an empty `command` (which will retrieve any additional logs), or it can send additional text (set `command` to the text) to STDIN of the running process, or it can send command=`ctrl+c` to interrupt the process.\n* Timeout: If a command execution result says "Command timed out. Sending SIGINT to the process", the assistant should retry running the command in the background.\n', + 'parameters': { + 'type': 'object', + 'properties': { + 'command': { + 'type': 'string', + 'description': 'The bash command to execute. Can be empty to view additional logs when previous exit code is `-1`. Can be `ctrl+c` to interrupt the currently running process.', + } + }, + 'required': ['command'], + }, + }, + }, + { + 'type': 'function', + 'function': { + 'name': 'finish', + 'description': 'Finish the interaction when the task is complete OR if the assistant cannot proceed further with the task.', + }, + }, + { + 'type': 'function', + 'function': { + 'name': 'str_replace_editor', + 'description': 'Custom editing tool for viewing, creating and editing files\n* State is persistent across command calls and discussions with the user\n* If `path` is a file, `view` displays the result of applying `cat -n`. If `path` is a directory, `view` lists non-hidden files and directories up to 2 levels deep\n* The `create` command cannot be used if the specified `path` already exists as a file\n* If a `command` generates a long output, it will be truncated and marked with ``\n* The `undo_edit` command will revert the last edit made to the file at `path`\n\nNotes for using the `str_replace` command:\n* The `old_str` parameter should match EXACTLY one or more consecutive lines from the original file. Be mindful of whitespaces!\n* If the `old_str` parameter is not unique in the file, the replacement will not be performed. Make sure to include enough context in `old_str` to make it unique\n* The `new_str` parameter should contain the edited lines that should replace the `old_str`\n', + 'parameters': { + 'type': 'object', + 'properties': { + 'command': { + 'description': 'The commands to run. Allowed options are: `view`, `create`, `str_replace`, `insert`, `undo_edit`.', + 'enum': [ + 'view', + 'create', + 'str_replace', + 'insert', + 'undo_edit', + ], + 'type': 'string', + }, + 'path': { + 'description': 'Absolute path to file or directory, e.g. `/repo/file.py` or `/repo`.', + 'type': 'string', + }, + 'file_text': { + 'description': 'Required parameter of `create` command, with the content of the file to be created.', + 'type': 'string', + }, + 'old_str': { + 'description': 'Required parameter of `str_replace` command containing the string in `path` to replace.', + 'type': 'string', + }, + 'new_str': { + 'description': 'Optional parameter of `str_replace` command containing the new string (if not given, no string will be added). Required parameter of `insert` command containing the string to insert.', + 'type': 'string', + }, + 'insert_line': { + 'description': 'Required parameter of `insert` command. The `new_str` will be inserted AFTER the line `insert_line` of `path`.', + 'type': 'integer', + }, + 'view_range': { + 'description': 'Optional parameter of `view` command when `path` points to a file. If none is given, the full file is shown. If provided, the file will be shown in the indicated line number range, e.g. [11, 12] will show lines 11 and 12. Indexing at 1 to start. Setting `[start_line, -1]` shows all lines from `start_line` to the end of the file.', + 'items': {'type': 'integer'}, + 'type': 'array', + }, + }, + 'required': ['command', 'path'], + }, + }, + }, +] + + +def test_convert_tools_to_description(): + formatted_tools = convert_tools_to_description(FNCALL_TOOLS) + assert ( + formatted_tools.strip() + == """---- BEGIN FUNCTION #1: execute_bash ---- +Description: Execute a bash command in the terminal. +* Long running commands: For commands that may run indefinitely, it should be run in the background and the output should be redirected to a file, e.g. command = `python3 app.py > server.log 2>&1 &`. +* Interactive: If a bash command returns exit code `-1`, this means the process is not yet finished. The assistant must then send a second call to terminal with an empty `command` (which will retrieve any additional logs), or it can send additional text (set `command` to the text) to STDIN of the running process, or it can send command=`ctrl+c` to interrupt the process. +* Timeout: If a command execution result says "Command timed out. Sending SIGINT to the process", the assistant should retry running the command in the background. + +Parameters: { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "The bash command to execute. Can be empty to view additional logs when previous exit code is `-1`. Can be `ctrl+c` to interrupt the currently running process." + } + }, + "required": [ + "command" + ] +} +---- END FUNCTION #1 ---- + +---- BEGIN FUNCTION #2: finish ---- +Description: Finish the interaction when the task is complete OR if the assistant cannot proceed further with the task. +No parameters are required for this function. +---- END FUNCTION #2 ---- + +---- BEGIN FUNCTION #3: str_replace_editor ---- +Description: Custom editing tool for viewing, creating and editing files +* State is persistent across command calls and discussions with the user +* If `path` is a file, `view` displays the result of applying `cat -n`. If `path` is a directory, `view` lists non-hidden files and directories up to 2 levels deep +* The `create` command cannot be used if the specified `path` already exists as a file +* If a `command` generates a long output, it will be truncated and marked with `` +* The `undo_edit` command will revert the last edit made to the file at `path` + +Notes for using the `str_replace` command: +* The `old_str` parameter should match EXACTLY one or more consecutive lines from the original file. Be mindful of whitespaces! +* If the `old_str` parameter is not unique in the file, the replacement will not be performed. Make sure to include enough context in `old_str` to make it unique +* The `new_str` parameter should contain the edited lines that should replace the `old_str` + +Parameters: { + "type": "object", + "properties": { + "command": { + "description": "The commands to run. Allowed options are: `view`, `create`, `str_replace`, `insert`, `undo_edit`.", + "enum": [ + "view", + "create", + "str_replace", + "insert", + "undo_edit" + ], + "type": "string" + }, + "path": { + "description": "Absolute path to file or directory, e.g. `/repo/file.py` or `/repo`.", + "type": "string" + }, + "file_text": { + "description": "Required parameter of `create` command, with the content of the file to be created.", + "type": "string" + }, + "old_str": { + "description": "Required parameter of `str_replace` command containing the string in `path` to replace.", + "type": "string" + }, + "new_str": { + "description": "Optional parameter of `str_replace` command containing the new string (if not given, no string will be added). Required parameter of `insert` command containing the string to insert.", + "type": "string" + }, + "insert_line": { + "description": "Required parameter of `insert` command. The `new_str` will be inserted AFTER the line `insert_line` of `path`.", + "type": "integer" + }, + "view_range": { + "description": "Optional parameter of `view` command when `path` points to a file. If none is given, the full file is shown. If provided, the file will be shown in the indicated line number range, e.g. [11, 12] will show lines 11 and 12. Indexing at 1 to start. Setting `[start_line, -1]` shows all lines from `start_line` to the end of the file.", + "items": { + "type": "integer" + }, + "type": "array" + } + }, + "required": [ + "command", + "path" + ] +} +---- END FUNCTION #3 ----""".strip() + ) + + +FNCALL_MESSAGES = [ + { + 'content': [ + { + 'type': 'text', + 'text': "You are a helpful assistant that can interact with a computer to solve tasks.\n\n* If user provides a path, you should NOT assume it's relative to the current working directory. Instead, you should explore the file system to find the file before working on it.\n\n\n", + 'cache_control': {'type': 'ephemeral'}, + } + ], + 'role': 'system', + }, + { + 'content': [ + { + 'type': 'text', + 'text': "\n/workspace/astropy__astropy__5.1\n\nI've uploaded a python code repository in the directory astropy__astropy__5.1. LONG DESCRIPTION:\n\n", + } + ], + 'role': 'user', + }, + { + 'content': [ + { + 'type': 'text', + 'text': "I'll help you implement the necessary changes to meet the requirements. Let's follow the steps:\n\n1. First, let's explore the repository structure:", + } + ], + 'role': 'assistant', + 'tool_calls': [ + { + 'index': 1, + 'function': { + 'arguments': '{"command": "ls -la /workspace/astropy__astropy__5.1"}', + 'name': 'execute_bash', + }, + 'id': 'toolu_01', + 'type': 'function', + } + ], + }, + { + 'content': [ + { + 'type': 'text', + 'text': 'ls -la /workspace/astropy__astropy__5.1\r\nls: /workspace/astropy__astropy__5.1: Bad file descriptor\r\nlrwxrwxrwx 1 root root 8 Oct 28 21:58 /workspace/astropy__astropy__5.1 -> /testbed[Python Interpreter: /opt/miniconda3/envs/testbed/bin/python]\nroot@openhands-workspace:/workspace/astropy__astropy__5.1 # \n[Command finished with exit code 0]', + } + ], + 'role': 'tool', + 'tool_call_id': 'toolu_01', + 'name': 'execute_bash', + }, + { + 'content': [ + { + 'type': 'text', + 'text': "I see there's a symlink. Let's explore the actual directory:", + } + ], + 'role': 'assistant', + 'tool_calls': [ + { + 'index': 1, + 'function': { + 'arguments': '{"command": "ls -la /testbed"}', + 'name': 'execute_bash', + }, + 'id': 'toolu_02', + 'type': 'function', + } + ], + }, + { + 'content': [ + { + 'type': 'text', + 'text': 'SOME OBSERVATION', + } + ], + 'role': 'tool', + 'tool_call_id': 'toolu_02', + 'name': 'execute_bash', + }, + { + 'content': [ + { + 'type': 'text', + 'text': "Let's look at the source code file mentioned in the PR description:", + } + ], + 'role': 'assistant', + 'tool_calls': [ + { + 'index': 1, + 'function': { + 'arguments': '{"command": "view", "path": "/testbed/astropy/io/fits/card.py"}', + 'name': 'str_replace_editor', + }, + 'id': 'toolu_03', + 'type': 'function', + } + ], + }, + { + 'content': [ + { + 'type': 'text', + 'text': "Here's the result of running `cat -n` on /testbed/astropy/io/fits/card.py:\n 1\t# Licensed under a 3-clause BSD style license - see PYFITS.rst...VERY LONG TEXT", + } + ], + 'role': 'tool', + 'tool_call_id': 'toolu_03', + 'name': 'str_replace_editor', + }, +] + +NON_FNCALL_MESSAGES = [ + { + 'role': 'system', + 'content': [ + { + 'type': 'text', + 'text': 'You are a helpful assistant that can interact with a computer to solve tasks.\n\n* If user provides a path, you should NOT assume it\'s relative to the current working directory. Instead, you should explore the file system to find the file before working on it.\n\n\n\nYou have access to the following functions:\n\n---- BEGIN FUNCTION #1: execute_bash ----\nDescription: Execute a bash command in the terminal.\n* Long running commands: For commands that may run indefinitely, it should be run in the background and the output should be redirected to a file, e.g. command = `python3 app.py > server.log 2>&1 &`.\n* Interactive: If a bash command returns exit code `-1`, this means the process is not yet finished. The assistant must then send a second call to terminal with an empty `command` (which will retrieve any additional logs), or it can send additional text (set `command` to the text) to STDIN of the running process, or it can send command=`ctrl+c` to interrupt the process.\n* Timeout: If a command execution result says "Command timed out. Sending SIGINT to the process", the assistant should retry running the command in the background.\n\nParameters: {\n "type": "object",\n "properties": {\n "command": {\n "type": "string",\n "description": "The bash command to execute. Can be empty to view additional logs when previous exit code is `-1`. Can be `ctrl+c` to interrupt the currently running process."\n }\n },\n "required": [\n "command"\n ]\n}\n---- END FUNCTION #1 ----\n\n---- BEGIN FUNCTION #2: finish ----\nDescription: Finish the interaction when the task is complete OR if the assistant cannot proceed further with the task.\nNo parameters are required for this function.\n---- END FUNCTION #2 ----\n\n---- BEGIN FUNCTION #3: str_replace_editor ----\nDescription: Custom editing tool for viewing, creating and editing files\n* State is persistent across command calls and discussions with the user\n* If `path` is a file, `view` displays the result of applying `cat -n`. If `path` is a directory, `view` lists non-hidden files and directories up to 2 levels deep\n* The `create` command cannot be used if the specified `path` already exists as a file\n* If a `command` generates a long output, it will be truncated and marked with ``\n* The `undo_edit` command will revert the last edit made to the file at `path`\n\nNotes for using the `str_replace` command:\n* The `old_str` parameter should match EXACTLY one or more consecutive lines from the original file. Be mindful of whitespaces!\n* If the `old_str` parameter is not unique in the file, the replacement will not be performed. Make sure to include enough context in `old_str` to make it unique\n* The `new_str` parameter should contain the edited lines that should replace the `old_str`\n\nParameters: {\n "type": "object",\n "properties": {\n "command": {\n "description": "The commands to run. Allowed options are: `view`, `create`, `str_replace`, `insert`, `undo_edit`.",\n "enum": [\n "view",\n "create",\n "str_replace",\n "insert",\n "undo_edit"\n ],\n "type": "string"\n },\n "path": {\n "description": "Absolute path to file or directory, e.g. `/repo/file.py` or `/repo`.",\n "type": "string"\n },\n "file_text": {\n "description": "Required parameter of `create` command, with the content of the file to be created.",\n "type": "string"\n },\n "old_str": {\n "description": "Required parameter of `str_replace` command containing the string in `path` to replace.",\n "type": "string"\n },\n "new_str": {\n "description": "Optional parameter of `str_replace` command containing the new string (if not given, no string will be added). Required parameter of `insert` command containing the string to insert.",\n "type": "string"\n },\n "insert_line": {\n "description": "Required parameter of `insert` command. The `new_str` will be inserted AFTER the line `insert_line` of `path`.",\n "type": "integer"\n },\n "view_range": {\n "description": "Optional parameter of `view` command when `path` points to a file. If none is given, the full file is shown. If provided, the file will be shown in the indicated line number range, e.g. [11, 12] will show lines 11 and 12. Indexing at 1 to start. Setting `[start_line, -1]` shows all lines from `start_line` to the end of the file.",\n "items": {\n "type": "integer"\n },\n "type": "array"\n }\n },\n "required": [\n "command",\n "path"\n ]\n}\n---- END FUNCTION #3 ----\n\n\nIf you choose to call a function ONLY reply in the following format with NO suffix:\n\n\nvalue_1\n\nThis is the value for the second parameter\nthat can span\nmultiple lines\n\n\n\n\nReminder:\n- Function calls MUST follow the specified format, start with \n- Required parameters MUST be specified\n- Only call one function at a time\n- You may provide optional reasoning for your function call in natural language BEFORE the function call, but NOT after.\n- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls\n', + 'cache_control': {'type': 'ephemeral'}, + } + ], + }, + { + 'content': [ + { + 'type': 'text', + 'text': "\n/workspace/astropy__astropy__5.1\n\nI've uploaded a python code repository in the directory astropy__astropy__5.1. LONG DESCRIPTION:\n\n", + } + ], + 'role': 'user', + }, + { + 'role': 'assistant', + 'content': [ + { + 'type': 'text', + 'text': "I'll help you implement the necessary changes to meet the requirements. Let's follow the steps:\n\n1. First, let's explore the repository structure:\n\n\nls -la /workspace/astropy__astropy__5.1\n", + } + ], + }, + { + 'role': 'user', + 'content': [ + { + 'type': 'text', + 'text': 'EXECUTION RESULT of [execute_bash]:\nls -la /workspace/astropy__astropy__5.1\r\nls: /workspace/astropy__astropy__5.1: Bad file descriptor\r\nlrwxrwxrwx 1 root root 8 Oct 28 21:58 /workspace/astropy__astropy__5.1 -> /testbed[Python Interpreter: /opt/miniconda3/envs/testbed/bin/python]\nroot@openhands-workspace:/workspace/astropy__astropy__5.1 # \n[Command finished with exit code 0]', + } + ], + }, + { + 'role': 'assistant', + 'content': [ + { + 'type': 'text', + 'text': "I see there's a symlink. Let's explore the actual directory:\n\n\nls -la /testbed\n", + } + ], + }, + { + 'role': 'user', + 'content': [ + { + 'type': 'text', + 'text': 'EXECUTION RESULT of [execute_bash]:\nSOME OBSERVATION', + } + ], + }, + { + 'role': 'assistant', + 'content': [ + { + 'type': 'text', + 'text': "Let's look at the source code file mentioned in the PR description:\n\n\nview\n/testbed/astropy/io/fits/card.py\n", + } + ], + }, + { + 'role': 'user', + 'content': [ + { + 'type': 'text', + 'text': "EXECUTION RESULT of [str_replace_editor]:\nHere's the result of running `cat -n` on /testbed/astropy/io/fits/card.py:\n 1\t# Licensed under a 3-clause BSD style license - see PYFITS.rst...VERY LONG TEXT", + } + ], + }, +] + +FNCALL_RESPONSE = { + 'content': [ + { + 'type': 'text', + 'text': 'Let me search for the `_format_float` method mentioned in the PR description:', + } + ], + 'role': 'assistant', + 'tool_calls': [ + { + 'index': 1, + 'function': { + 'arguments': '{"command": "grep -n \\"_format_float\\" /testbed/astropy/io/fits/card.py"}', + 'name': 'execute_bash', + }, + 'id': 'toolu_01KmxSLyHu4HSEBkuxUE2SCr', + 'type': 'function', + } + ], +} + +NON_FNCALL_RESPONSE = { + 'content': [ + { + 'type': 'text', + 'text': 'Let me search for the `_format_float` method mentioned in the PR description:\n\n\ngrep -n "_format_float" /testbed/astropy/io/fits/card.py\n', + } + ], + 'role': 'assistant', +} + + +@pytest.mark.parametrize( + 'tool_calls, expected', + [ + # Original test case + ( + FNCALL_RESPONSE['tool_calls'], + """ +grep -n "_format_float" /testbed/astropy/io/fits/card.py +""", + ), + # Test case with multiple parameters + ( + [ + { + 'index': 1, + 'function': { + 'arguments': '{"command": "view", "path": "/test/file.py", "view_range": [1, 10]}', + 'name': 'str_replace_editor', + }, + 'id': 'test_id', + 'type': 'function', + } + ], + """ +view +/test/file.py +[1, 10] +""", + ), + ], +) +def test_convert_tool_calls_to_string(tool_calls, expected): + converted = convert_tool_calls_to_string(tool_calls) + print(converted) + assert converted == expected + + +def test_convert_fncall_messages_to_non_fncall_messages(): + converted_non_fncall = convert_fncall_messages_to_non_fncall_messages( + FNCALL_MESSAGES, FNCALL_TOOLS + ) + assert converted_non_fncall == NON_FNCALL_MESSAGES + + +def test_convert_non_fncall_messages_to_fncall_messages(): + converted = convert_non_fncall_messages_to_fncall_messages( + NON_FNCALL_MESSAGES, FNCALL_TOOLS + ) + print(json.dumps(converted, indent=2)) + assert converted == FNCALL_MESSAGES + + +def test_two_way_conversion_nonfn_to_fn_to_nonfn(): + non_fncall_copy = copy.deepcopy(NON_FNCALL_MESSAGES) + converted_fncall = convert_non_fncall_messages_to_fncall_messages( + NON_FNCALL_MESSAGES, FNCALL_TOOLS + ) + assert ( + non_fncall_copy == NON_FNCALL_MESSAGES + ) # make sure original messages are not modified + assert converted_fncall == FNCALL_MESSAGES + + fncall_copy = copy.deepcopy(FNCALL_MESSAGES) + converted_non_fncall = convert_fncall_messages_to_non_fncall_messages( + FNCALL_MESSAGES, FNCALL_TOOLS + ) + assert ( + fncall_copy == FNCALL_MESSAGES + ) # make sure original messages are not modified + assert converted_non_fncall == NON_FNCALL_MESSAGES + + +def test_two_way_conversion_fn_to_nonfn_to_fn(): + fncall_copy = copy.deepcopy(FNCALL_MESSAGES) + converted_non_fncall = convert_fncall_messages_to_non_fncall_messages( + FNCALL_MESSAGES, FNCALL_TOOLS + ) + assert ( + fncall_copy == FNCALL_MESSAGES + ) # make sure original messages are not modified + assert converted_non_fncall == NON_FNCALL_MESSAGES + + non_fncall_copy = copy.deepcopy(NON_FNCALL_MESSAGES) + converted_fncall = convert_non_fncall_messages_to_fncall_messages( + NON_FNCALL_MESSAGES, FNCALL_TOOLS + ) + assert ( + non_fncall_copy == NON_FNCALL_MESSAGES + ) # make sure original messages are not modified + assert converted_fncall == FNCALL_MESSAGES