import json import os from collections import deque import openhands import openhands.agenthub.codeact_agent.function_calling as codeact_function_calling from openhands.controller.agent import Agent from openhands.controller.state.state import State from openhands.core.config import AgentConfig from openhands.core.logger import openhands_logger as logger from openhands.core.message import Message, TextContent from openhands.core.message_utils import ( apply_prompt_caching, events_to_messages, ) from openhands.events.action import ( Action, AgentFinishAction, ) from openhands.llm.llm import LLM from openhands.memory.condenser import Condenser from openhands.runtime.plugins import ( AgentSkillsRequirement, JupyterRequirement, PluginRequirement, ) from openhands.utils.prompt import PromptManager class CodeActAgent(Agent): VERSION = '2.2' """ The Code Act Agent is a minimalist agent. The agent works by passing the model a list of action-observation pairs and prompting the model to take the next step. ### Overview This agent implements the CodeAct idea ([paper](https://arxiv.org/abs/2402.01030), [tweet](https://twitter.com/xingyaow_/status/1754556835703751087)) that consolidates LLM agents' **act**ions into a unified **code** action space for both *simplicity* and *performance* (see paper for more details). The conceptual idea is illustrated below. At each turn, the agent can: 1. **Converse**: Communicate with humans in natural language to ask for clarification, confirmation, etc. 2. **CodeAct**: Choose to perform the task by executing code - Execute any valid Linux `bash` command - Execute any valid `Python` code with [an interactive Python interpreter](https://ipython.org/). This is simulated through `bash` command, see plugin system below for more details. ![image](https://github.com/All-Hands-AI/OpenHands/assets/38853559/92b622e3-72ad-4a61-8f41-8c040b6d5fb3) """ sandbox_plugins: list[PluginRequirement] = [ # NOTE: AgentSkillsRequirement need to go before JupyterRequirement, since # AgentSkillsRequirement provides a lot of Python functions, # and it needs to be initialized before Jupyter for Jupyter to use those functions. AgentSkillsRequirement(), JupyterRequirement(), ] def __init__( self, llm: LLM, config: AgentConfig, ) -> None: """Initializes a new instance of the CodeActAgent class. Parameters: - llm (LLM): The llm to be used by this agent """ super().__init__(llm, config) self.pending_actions: deque[Action] = deque() self.reset() # Retrieve the enabled tools self.tools = codeact_function_calling.get_tools( codeact_enable_browsing=self.config.codeact_enable_browsing, codeact_enable_jupyter=self.config.codeact_enable_jupyter, codeact_enable_llm_editor=self.config.codeact_enable_llm_editor, ) logger.debug( f'TOOLS loaded for CodeActAgent: {json.dumps(self.tools, indent=2, ensure_ascii=False).replace("\\n", "\n")}' ) self.prompt_manager = PromptManager( microagent_dir=os.path.join( os.path.dirname(os.path.dirname(openhands.__file__)), 'microagents', ) if self.config.enable_prompt_extensions else None, prompt_dir=os.path.join(os.path.dirname(__file__), 'prompts'), disabled_microagents=self.config.disabled_microagents, ) self.condenser = Condenser.from_config(self.config.condenser) logger.debug(f'Using condenser: {self.condenser}') def reset(self) -> None: """Resets the CodeAct Agent.""" super().reset() self.pending_actions.clear() def step(self, state: State) -> Action: """Performs one step using the CodeAct Agent. This includes gathering info on previous steps and prompting the model to make a command to execute. Parameters: - state (State): used to get updated info Returns: - CmdRunAction(command) - bash command to run - IPythonRunCellAction(code) - IPython code to run - AgentDelegateAction(agent, inputs) - delegate action for (sub)task - MessageAction(content) - Message action to run (e.g. ask for clarification) - AgentFinishAction() - end the interaction """ # Continue with pending actions if any if self.pending_actions: return self.pending_actions.popleft() # if we're done, go back latest_user_message = state.get_last_user_message() if latest_user_message and latest_user_message.content.strip() == '/exit': return AgentFinishAction() # prepare what we want to send to the LLM messages = self._get_messages(state) params: dict = { 'messages': self.llm.format_messages_for_llm(messages), } params['tools'] = self.tools response = self.llm.completion(**params) actions = codeact_function_calling.response_to_actions(response) for action in actions: self.pending_actions.append(action) return self.pending_actions.popleft() def _get_messages(self, state: State) -> list[Message]: """Constructs the message history for the LLM conversation. This method builds a structured conversation history by processing events from the state and formatting them into messages that the LLM can understand. It handles both regular message flow and function-calling scenarios. The method performs the following steps: 1. Initializes with system prompt and optional initial user message 2. Processes events (Actions and Observations) into messages 3. Handles tool calls and their responses in function-calling mode 4. Manages message role alternation (user/assistant/tool) 5. Applies caching for specific LLM providers (e.g., Anthropic) 6. Adds environment reminders for non-function-calling mode Args: state (State): The current state object containing conversation history and other metadata Returns: list[Message]: A list of formatted messages ready for LLM consumption, including: - System message with prompt - Initial user message (if configured) - Action messages (from both user and assistant) - Observation messages (including tool responses) - Environment reminders (in non-function-calling mode) Note: - In function-calling mode, tool calls and their responses are carefully tracked to maintain proper conversation flow - Messages from the same role are combined to prevent consecutive same-role messages - For Anthropic models, specific messages are cached according to their documentation """ if not self.prompt_manager: raise Exception('Prompt Manager not instantiated.') messages: list[Message] = self._initial_messages() # Condense the events from the state. events = self.condenser.condensed_history(state) messages += events_to_messages( events, max_message_chars=self.llm.config.max_message_chars, vision_is_active=self.llm.vision_is_active(), enable_som_visual_browsing=self.config.enable_som_visual_browsing, ) messages = self._enhance_messages(messages) if self.llm.is_caching_prompt_active(): apply_prompt_caching(messages) return messages def _initial_messages(self) -> list[Message]: """Creates the initial messages (including the system prompt) for the LLM conversation.""" assert self.prompt_manager, 'Prompt Manager not instantiated.' return [ Message( role='system', content=[ TextContent( text=self.prompt_manager.get_system_message(), cache_prompt=self.llm.is_caching_prompt_active(), ) ], ) ] def _enhance_messages(self, messages: list[Message]) -> list[Message]: """Enhances the user message with additional context based on keywords matched. Args: messages (list[Message]): The list of messages to enhance Returns: list[Message]: The enhanced list of messages """ assert self.prompt_manager, 'Prompt Manager not instantiated.' results: list[Message] = [] is_first_message_handled = False prev_role = None for msg in messages: if msg.role == 'user' and not is_first_message_handled: is_first_message_handled = True # compose the first user message with examples self.prompt_manager.add_examples_to_initial_message(msg) # and/or repo/runtime info if self.config.enable_prompt_extensions: self.prompt_manager.add_info_to_initial_message(msg) # enhance the user message with additional context based on keywords matched if msg.role == 'user': self.prompt_manager.enhance_message(msg) # Add double newline between consecutive user messages if prev_role == 'user' and len(msg.content) > 0: # Find the first TextContent in the message to add newlines for content_item in msg.content: if isinstance(content_item, TextContent): # If the previous message was also from a user, prepend two newlines to ensure separation content_item.text = '\n\n' + content_item.text break results.append(msg) prev_role = msg.role return results