Skip to content

react

ipw.agents.react

React agent implementation using the Agno framework.

React

Bases: BaseAgent

React agent that uses the Agno Agent framework for tool-augmented reasoning.

Source code in intelligence-per-watt/src/ipw/agents/react.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@AgentRegistry.register("react")
class React(BaseAgent):
    """React agent that uses the Agno Agent framework for tool-augmented reasoning."""

    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[Callable] | None = None,
        instructions: str | None = None,
        event_recorder: Optional["EventRecorder"] = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the React agent.

        Args:
            model: The Agno Model instance to use.
            tools: Optional list of callable tools/functions for the agent to use.
            instructions: Optional custom instructions for the agent.
            event_recorder: Optional EventRecorder for per-action energy telemetry.
            **kwargs: Additional keyword arguments passed to the Agent constructor.
        """
        super().__init__(event_recorder=event_recorder)

        # Lazy import: agno is optional
        try:
            from agno.agent import Agent
        except ImportError:
            raise ImportError(
                "agno package is required for React agent. "
                "Install with: pip install agno"
            )

        self.model = model
        self._original_tools = tools or []
        self.instructions = instructions or self.DEFAULT_INSTRUCTIONS

        # Instrument tools if event_recorder is provided
        if event_recorder and self._original_tools:
            self.tools = self._instrument_tools(self._original_tools)
        else:
            self.tools = self._original_tools

        agent_kwargs: dict[str, Any] = {
            "model": self.model,
            "instructions": self.instructions,
            **kwargs,
        }
        if self.tools:
            agent_kwargs["tools"] = self.tools
            agent_kwargs["tool_choice"] = "auto"

        self.agent = Agent(**agent_kwargs)

    def _instrument_tools(self, tools: List[Callable]) -> List[Callable]:
        """Wrap tools to emit start/end events for energy tracking."""
        instrumented = []
        for tool in tools:
            tool_name = getattr(tool, "__name__", str(tool))

            @functools.wraps(tool)
            def wrapper(
                *args: Any,
                __tool: Callable = tool,
                __name: str = tool_name,
                **kwargs: Any,
            ) -> Any:
                self._record_event("tool_call_start", tool=__name)
                try:
                    return __tool(*args, **kwargs)
                finally:
                    self._record_event("tool_call_end", tool=__name)

            instrumented.append(wrapper)
        return instrumented

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

        Args:
            input: The input message or prompt for the agent.
            **kwargs: Additional keyword arguments passed to agent.run().

        Returns:
            AgentRunResult with the agent's response.
        """
        self._record_event("lm_inference_start", model=str(self.model))
        result = None
        try:
            result = self.agent.run(input, **kwargs)
            # Extract token metrics from the result if available
            end_metadata: dict[str, Any] = {"model": str(self.model)}
            input_tokens = 0
            output_tokens = 0
            if result is not None and hasattr(result, "metrics") and result.metrics is not None:
                metrics = result.metrics
                input_tokens = getattr(metrics, "input_tokens", 0)
                output_tokens = getattr(metrics, "output_tokens", 0)
                end_metadata["prompt_tokens"] = input_tokens
                end_metadata["completion_tokens"] = output_tokens
                end_metadata["total_tokens"] = getattr(metrics, "total_tokens", 0)
            self._record_event("lm_inference_end", **end_metadata)

            # Extract content from result
            content = ""
            if result is not None:
                if hasattr(result, "content"):
                    content = str(result.content)
                else:
                    content = str(result)

            return AgentRunResult(
                content=content,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
            )
        except Exception:
            self._record_event("lm_inference_end", model=str(self.model))
            raise

__init__(model, tools=None, instructions=None, event_recorder=None, **kwargs)

Initialize the React agent.

Parameters:

Name Type Description Default
model Any

The Agno Model instance to use.

 required
tools List[Callable] | None

Optional list of callable tools/functions for the agent to use.

 None
instructions str | None

Optional custom instructions for the agent.

 None
event_recorder Optional['EventRecorder']

Optional EventRecorder for per-action energy telemetry.

 None
**kwargs Any

Additional keyword arguments passed to the Agent constructor.

 {}
Source code in intelligence-per-watt/src/ipw/agents/react.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    model: Any,
    tools: List[Callable] | None = None,
    instructions: str | None = None,
    event_recorder: Optional["EventRecorder"] = None,
    **kwargs: Any,
) -> None:
    """Initialize the React agent.

    Args:
        model: The Agno Model instance to use.
        tools: Optional list of callable tools/functions for the agent to use.
        instructions: Optional custom instructions for the agent.
        event_recorder: Optional EventRecorder for per-action energy telemetry.
        **kwargs: Additional keyword arguments passed to the Agent constructor.
    """
    super().__init__(event_recorder=event_recorder)

    # Lazy import: agno is optional
    try:
        from agno.agent import Agent
    except ImportError:
        raise ImportError(
            "agno package is required for React agent. "
            "Install with: pip install agno"
        )

    self.model = model
    self._original_tools = tools or []
    self.instructions = instructions or self.DEFAULT_INSTRUCTIONS

    # Instrument tools if event_recorder is provided
    if event_recorder and self._original_tools:
        self.tools = self._instrument_tools(self._original_tools)
    else:
        self.tools = self._original_tools

    agent_kwargs: dict[str, Any] = {
        "model": self.model,
        "instructions": self.instructions,
        **kwargs,
    }
    if self.tools:
        agent_kwargs["tools"] = self.tools
        agent_kwargs["tool_choice"] = "auto"

    self.agent = Agent(**agent_kwargs)

run(input, **kwargs)

Run the React agent.

Parameters:

Name Type Description Default
input str

The input message or prompt for the agent.

 required
**kwargs Any

Additional keyword arguments passed to agent.run().

 {}

Returns:

Type Description
AgentRunResult

AgentRunResult with the agent's response.
Source code in intelligence-per-watt/src/ipw/agents/react.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def run(self, input: str, **kwargs: Any) -> AgentRunResult:
    """Run the React agent.

    Args:
        input: The input message or prompt for the agent.
        **kwargs: Additional keyword arguments passed to agent.run().

    Returns:
        AgentRunResult with the agent's response.
    """
    self._record_event("lm_inference_start", model=str(self.model))
    result = None
    try:
        result = self.agent.run(input, **kwargs)
        # Extract token metrics from the result if available
        end_metadata: dict[str, Any] = {"model": str(self.model)}
        input_tokens = 0
        output_tokens = 0
        if result is not None and hasattr(result, "metrics") and result.metrics is not None:
            metrics = result.metrics
            input_tokens = getattr(metrics, "input_tokens", 0)
            output_tokens = getattr(metrics, "output_tokens", 0)
            end_metadata["prompt_tokens"] = input_tokens
            end_metadata["completion_tokens"] = output_tokens
            end_metadata["total_tokens"] = getattr(metrics, "total_tokens", 0)
        self._record_event("lm_inference_end", **end_metadata)

        # Extract content from result
        content = ""
        if result is not None:
            if hasattr(result, "content"):
                content = str(result.content)
            else:
                content = str(result)

        return AgentRunResult(
            content=content,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
        )
    except Exception:
        self._record_event("lm_inference_end", model=str(self.model))
        raise