openhands

ipw.agents.openhands

OpenHands agent implementation with per-tool energy tracking.

OpenHands

Bases: BaseAgent

OpenHands agent using the OpenHands SDK with energy telemetry.

Source code in intelligence-per-watt/src/ipw/agents/openhands.py

@AgentRegistry.register("openhands")
class OpenHands(BaseAgent):
    """OpenHands agent using the OpenHands SDK with energy telemetry."""

    DEFAULT_INSTRUCTIONS = (
        "You are a helpful assistant that can answer questions "
        "and use the tools provided to you if necessary."
    )

    def __init__(
        self,
        model: Any,
        tools: list | None = None,
        mcp_tools: Optional[Dict[str, "BaseMCPServer"]] = None,
        event_recorder: Optional["EventRecorder"] = None,
        max_turns: int = 20,
        **kwargs: Any,
    ) -> None:
        """Initialize the OpenHands agent.

        Args:
            model: The LLM model instance to use.
            tools: List of OpenHands Tool specs.
            mcp_tools: Optional dict mapping tool name to BaseMCPServer instance.
            event_recorder: Optional EventRecorder for per-action energy telemetry.
            max_turns: Maximum iterations per run (default 20).
            **kwargs: Additional keyword arguments passed to the Agent constructor.
        """
        super().__init__(event_recorder=event_recorder)

        # Lazy imports: openhands-sdk is optional
        try:
            from openhands.sdk import Agent, Event, LLMConvertibleEvent, LLMSummarizingCondenser, LocalConversation
            from openhands.sdk.event.llm_convertible.action import ActionEvent
            from openhands.sdk.event.llm_convertible.observation import ObservationEvent
        except ImportError:
            raise ImportError(
                "openhands-sdk package is required for OpenHands agent. "
                "Install with: pip install openhands-sdk"
            )

        self.model = model
        self.tools = tools
        self._pending_tool: Optional[str] = None
        self._tool_names_used: List[str] = []
        self._num_turns: int = 0

        # Store references for use in callbacks
        self._ActionEvent = ActionEvent
        self._ObservationEvent = ObservationEvent
        self._LLMConvertibleEvent = LLMConvertibleEvent

        # Context condenser
        condenser = LLMSummarizingCondenser(
            llm=model,
            max_tokens=24000,
            keep_first=2,
        )

        agent_kwargs = {"llm": model, "condenser": condenser}

        if tools:
            agent_kwargs["tools"] = tools
        elif mcp_tools:
            extra_tool_specs = _register_mcp_tools(mcp_tools)
            agent_kwargs["tools"] = extra_tool_specs

        self.agent = Agent(**agent_kwargs)

        self.conversation = LocalConversation(
            agent=self.agent,
            callbacks=[self._instrumented_callback],
            workspace=os.getcwd(),
            max_iteration_per_run=max_turns,
        )
        self.current_result = ""

    def _instrumented_callback(self, event: Any) -> None:
        """Instrumented callback that emits telemetry events for tool calls."""
        if isinstance(event, self._ActionEvent):
            tool_name = event.tool_name
            self._pending_tool = tool_name
            self._tool_names_used.append(tool_name)
            self._num_turns += 1
            self._record_event("tool_call_start", tool=tool_name)
        elif isinstance(event, self._ObservationEvent):
            tool_name = self._pending_tool or event.tool_name
            self._record_event("tool_call_end", tool=tool_name)
            self._pending_tool = None

        if isinstance(event, self._LLMConvertibleEvent):
            self.current_result = event.to_llm_message()

    @staticmethod
    def _extract_text(message: Any) -> str:
        """Extract plain text from an OpenHands Message or fallback to str()."""
        if hasattr(message, "content") and isinstance(message.content, (list, tuple)):
            try:
                from openhands.sdk.llm.message import TextContent
                parts = [
                    item.text for item in message.content if isinstance(item, TextContent)
                ]
                if parts:
                    text = "\n".join(parts)
                else:
                    text = str(message)
            except ImportError:
                text = str(message)
        elif isinstance(message, str):
            text = message
        else:
            text = str(message)

        # Strip <think>...</think> blocks (extended thinking output)
        text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL)
        text = re.sub(r".*</think>", "", text, flags=re.DOTALL)
        return text.strip()

    def run(self, input: str, **kwargs: Any) -> AgentRunResult:
        """Run the OpenHands agent with telemetry.

        Args:
            input: The input message or prompt for the agent.
            **kwargs: Additional keyword arguments.

        Returns:
            AgentRunResult with content, tool_names_used, and num_turns.
        """
        from openhands.sdk.conversation.response_utils import get_agent_final_response

        # Reset per-run tracking
        self._tool_names_used = []
        self._num_turns = 0

        self._record_event("lm_inference_start", model=str(self.model))
        try:
            self.conversation.send_message(input)
            self.conversation.run()

            result = get_agent_final_response(self.conversation.state.events)
            if not result:
                # Agent hit the turn cap without calling FinishTool.
                logger.info("No FinishTool call detected, sending synthesis nudge (2 turns)")
                saved_limit = self.conversation.max_iteration_per_run
                self.conversation.max_iteration_per_run = 2
                self.conversation.send_message(
                    "You have run out of turns. Please provide your final answer now. "
                    "Use the finish tool to submit your answer."
                )
                self.conversation.run()
                self.conversation.max_iteration_per_run = saved_limit
                result = get_agent_final_response(self.conversation.state.events)

            if not result:
                result = self._extract_text(self.current_result)
                logger.warning("get_agent_final_response() returned empty after nudge, using callback fallback")

            result = self._extract_text(result)
            self.current_result = ""
            return AgentRunResult(
                content=result,
                tool_calls_attempted=len(self._tool_names_used),
                tool_calls_succeeded=len(self._tool_names_used),
                tool_names_used=list(self._tool_names_used),
                num_turns=self._num_turns,
            )
        finally:
            self._record_event("lm_inference_end", model=str(self.model))
            try:
                self.conversation.close()
            except Exception as e:
                logger.warning(f"Error closing conversation: {e}")

__init__(model, tools=None, mcp_tools=None, event_recorder=None, max_turns=20, **kwargs)

Initialize the OpenHands agent.

Parameters:

Name	Type	Description	Default
model	Any	The LLM model instance to use.	required
tools	list | None	List of OpenHands Tool specs.	None
mcp_tools	Optional[Dict[str, 'BaseMCPServer']]	Optional dict mapping tool name to BaseMCPServer instance.	None
event_recorder	Optional['EventRecorder']	Optional EventRecorder for per-action energy telemetry.	None
max_turns	int	Maximum iterations per run (default 20).	20
**kwargs	Any	Additional keyword arguments passed to the Agent constructor.	{}

Source code in intelligence-per-watt/src/ipw/agents/openhands.py

def __init__(
    self,
    model: Any,
    tools: list | None = None,
    mcp_tools: Optional[Dict[str, "BaseMCPServer"]] = None,
    event_recorder: Optional["EventRecorder"] = None,
    max_turns: int = 20,
    **kwargs: Any,
) -> None:
    """Initialize the OpenHands agent.

    Args:
        model: The LLM model instance to use.
        tools: List of OpenHands Tool specs.
        mcp_tools: Optional dict mapping tool name to BaseMCPServer instance.
        event_recorder: Optional EventRecorder for per-action energy telemetry.
        max_turns: Maximum iterations per run (default 20).
        **kwargs: Additional keyword arguments passed to the Agent constructor.
    """
    super().__init__(event_recorder=event_recorder)

    # Lazy imports: openhands-sdk is optional
    try:
        from openhands.sdk import Agent, Event, LLMConvertibleEvent, LLMSummarizingCondenser, LocalConversation
        from openhands.sdk.event.llm_convertible.action import ActionEvent
        from openhands.sdk.event.llm_convertible.observation import ObservationEvent
    except ImportError:
        raise ImportError(
            "openhands-sdk package is required for OpenHands agent. "
            "Install with: pip install openhands-sdk"
        )

    self.model = model
    self.tools = tools
    self._pending_tool: Optional[str] = None
    self._tool_names_used: List[str] = []
    self._num_turns: int = 0

    # Store references for use in callbacks
    self._ActionEvent = ActionEvent
    self._ObservationEvent = ObservationEvent
    self._LLMConvertibleEvent = LLMConvertibleEvent

    # Context condenser
    condenser = LLMSummarizingCondenser(
        llm=model,
        max_tokens=24000,
        keep_first=2,
    )

    agent_kwargs = {"llm": model, "condenser": condenser}

    if tools:
        agent_kwargs["tools"] = tools
    elif mcp_tools:
        extra_tool_specs = _register_mcp_tools(mcp_tools)
        agent_kwargs["tools"] = extra_tool_specs

    self.agent = Agent(**agent_kwargs)

    self.conversation = LocalConversation(
        agent=self.agent,
        callbacks=[self._instrumented_callback],
        workspace=os.getcwd(),
        max_iteration_per_run=max_turns,
    )
    self.current_result = ""

run(input, **kwargs)

Run the OpenHands agent with telemetry.

Parameters:

Name	Type	Description	Default
input	str	The input message or prompt for the agent.	required
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
AgentRunResult	AgentRunResult with content, tool_names_used, and num_turns.

Source code in intelligence-per-watt/src/ipw/agents/openhands.py

def run(self, input: str, **kwargs: Any) -> AgentRunResult:
    """Run the OpenHands agent with telemetry.

    Args:
        input: The input message or prompt for the agent.
        **kwargs: Additional keyword arguments.

    Returns:
        AgentRunResult with content, tool_names_used, and num_turns.
    """
    from openhands.sdk.conversation.response_utils import get_agent_final_response

    # Reset per-run tracking
    self._tool_names_used = []
    self._num_turns = 0

    self._record_event("lm_inference_start", model=str(self.model))
    try:
        self.conversation.send_message(input)
        self.conversation.run()

        result = get_agent_final_response(self.conversation.state.events)
        if not result:
            # Agent hit the turn cap without calling FinishTool.
            logger.info("No FinishTool call detected, sending synthesis nudge (2 turns)")
            saved_limit = self.conversation.max_iteration_per_run
            self.conversation.max_iteration_per_run = 2
            self.conversation.send_message(
                "You have run out of turns. Please provide your final answer now. "
                "Use the finish tool to submit your answer."
            )
            self.conversation.run()
            self.conversation.max_iteration_per_run = saved_limit
            result = get_agent_final_response(self.conversation.state.events)

        if not result:
            result = self._extract_text(self.current_result)
            logger.warning("get_agent_final_response() returned empty after nudge, using callback fallback")

        result = self._extract_text(result)
        self.current_result = ""
        return AgentRunResult(
            content=result,
            tool_calls_attempted=len(self._tool_names_used),
            tool_calls_succeeded=len(self._tool_names_used),
            tool_names_used=list(self._tool_names_used),
            num_turns=self._num_turns,
        )
    finally:
        self._record_event("lm_inference_end", model=str(self.model))
        try:
            self.conversation.close()
        except Exception as e:
            logger.warning(f"Error closing conversation: {e}")