Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/strategies/react.py: 19%
124 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""ReAct (Reason + Act) strategy for agent reasoning.
3Implements the Think → Act → Observe loop described in:
4 Yao et al., "ReAct: Synergizing Reasoning and Acting in Language Models" (2022)
6Each iteration:
7 1. **Think** — The LLM reasons about the current state and decides
8 whether to call a tool or produce a final answer.
9 2. **Act** — If the LLM requests a tool call, the tool is executed
10 with timeout and output truncation.
11 3. **Observe** — The tool result (or error) is fed back as the next
12 user message so the LLM can incorporate it.
14The loop terminates when:
15 - The LLM signals a final answer (``FINAL_ANSWER:`` prefix).
16 - The maximum iteration count is reached.
17 - A governance budget check fails.
18"""
20from __future__ import annotations
22import asyncio
23import time
24from typing import TYPE_CHECKING, Any, cast
26from lexigram.ai.agents.strategies.base import AbstractStrategy
27from lexigram.ai.agents.strategies.parsing import (
28 build_chat_messages_from_dict,
29 extract_final_answer,
30 extract_thought,
31 extract_tool_call,
32)
33from lexigram.ai.agents.strategies.token_utils import TokenAccumulator
34from lexigram.ai.agents.types import ReasoningStep, ToolExecutionRecord
35from lexigram.contracts.ai.agents import AgentError, AgentResponse
36from lexigram.contracts.ai.llm import ChatMessage, Role
37from lexigram.logging import (
38 get_logger,
39)
40from lexigram.result import Err, Ok, Result
42if TYPE_CHECKING:
43 from lexigram.contracts.ai.agents import MemoryProtocol, ToolProtocol
44 from lexigram.contracts.ai.llm import CompletionProtocol, LLMClientProtocol
46logger = get_logger(__name__)
49# ---------------------------------------------------------------------------
50# Prompt templates
51# ---------------------------------------------------------------------------
53_SYSTEM_SUFFIX = """
54You are a ReAct agent. You solve problems by iterating through Think, Act, and Observe steps.
56## Output Format
58On each turn you MUST produce EXACTLY ONE of two outputs:
60### Option A — Thinking + Tool Call
61```
62THOUGHT: <your step-by-step reasoning>
63ACTION: <tool_name>
64ACTION_INPUT: <JSON arguments for the tool>
65```
67### Option B — Final Answer
68```
69THOUGHT: <your final reasoning>
70FINAL_ANSWER: <your complete answer to the user>
71```
73## Rules
74- Always start with THOUGHT.
75- Use EXACTLY one tool per turn. Do not call multiple tools at once.
76- After receiving an OBSERVATION, think about what to do next.
77- When you have enough information, produce FINAL_ANSWER.
79## Available Tools
80{tool_descriptions}
81"""
83_OBSERVATION_TEMPLATE = "OBSERVATION: {observation}"
86class ReActStrategy(AbstractStrategy):
87 """ReAct (Reason + Act) reasoning strategy.
89 The ReAct strategy follows a Think → Act → Observe loop:
91 1. **Think** — LLM reasons about the current state.
92 2. **Act** — Parse and execute a tool call from the LLM response.
93 3. **Observe** — Feed tool result back as an observation.
94 4. **Decide** — Check if the LLM signals completion.
96 This is the default strategy for Lexigram agents.
98 Example::
100 from lexigram.ai.agents import Agent
101 from lexigram.ai.agents.strategies import ReActStrategy
103 agent = Agent(
104 llm=my_llm_client,
105 strategy=ReActStrategy(max_iterations=10),
106 )
107 response = await agent.run("What's the weather?")
108 """
110 def __init__(
111 self,
112 max_iterations: int = 10,
113 tool_timeout: float = 30.0,
114 observation_max_chars: int = 10_000,
115 timeout: float = 120.0,
116 tool_max_retries: int = 3,
117 ) -> None:
118 """Initialise the ReAct strategy.
120 Args:
121 max_iterations: Maximum number of Think→Act→Observe cycles.
122 tool_timeout: Per-tool execution timeout in seconds.
123 observation_max_chars: Maximum characters for tool output before
124 truncation.
125 timeout: Per-LLM-call timeout in seconds.
126 tool_max_retries: Retry attempts for transient tool errors
127 (``ConnectionError``, ``OSError``). Each retry waits
128 with exponential back-off (1s, 2s, 4s, …).
129 """
130 self.max_iterations = max_iterations
131 self.tool_timeout = tool_timeout
132 self.observation_max_chars = observation_max_chars
133 self.timeout = timeout
134 self.tool_max_retries = tool_max_retries
136 async def execute(
137 self,
138 message: str,
139 tools: list[ToolProtocol],
140 history: list[dict[str, Any]],
141 llm: LLMClientProtocol,
142 **kwargs: Any,
143 ) -> Result[AgentResponse, AgentError]:
144 """Execute the ReAct reasoning loop.
146 Args:
147 message: The user's input message.
148 tools: Tools available to the agent.
149 history: Conversation history as ChatMessage objects.
150 llm: LLM client implementing ``LLMClientProtocol``.
151 **kwargs: Additional parameters:
152 - ``system_prompt`` (str): Optional system prompt prefix.
153 - ``memory``: Optional memory backend for context retrieval.
155 Returns:
156 ``Ok(AgentResponse)`` with the final answer and full reasoning
157 trace. ``Err(AgentError)`` on unrecoverable failure.
158 """
159 system_prompt: str = kwargs.get("system_prompt", "")
160 memory = kwargs.get("memory")
161 guard_pipeline = kwargs.get("guard_pipeline")
163 steps: list[ReasoningStep] = []
164 tool_calls: list[ToolExecutionRecord] = []
165 usage = TokenAccumulator()
166 start_time = time.monotonic()
168 # Build tool lookup
169 tool_map: dict[str, ToolProtocol] = {t.name: t for t in tools}
171 # Retrieve memory context
172 memory_context = await self._get_memory_context(memory)
174 # Build system prompt with tool descriptions
175 tool_descriptions = "\n".join(
176 f"- **{t.name}**: {t.description}\n Parameters: {t.parameters_schema}"
177 for t in tools
178 )
179 full_system = (
180 system_prompt
181 + memory_context
182 + _SYSTEM_SUFFIX.format(tool_descriptions=tool_descriptions)
183 )
185 # Build messages from history
186 messages = build_chat_messages_from_dict(message, history, full_system)
188 for iteration in range(1, self.max_iterations + 1):
189 # --- THINK: Ask LLM to reason ---
190 completion = await self._call_llm(llm, messages)
191 if completion is None:
192 return Err(
193 AgentError(f"LLM returned empty response at iteration {iteration}")
194 )
195 usage.add(completion)
196 llm_text = (
197 completion.content
198 if hasattr(completion, "content")
199 else str(completion)
200 )
202 logger.debug(
203 "react_llm_response",
204 iteration=iteration,
205 length=len(llm_text),
206 )
208 # Parse the response
209 thought = extract_thought(llm_text)
210 final_answer = extract_final_answer(llm_text)
212 # --- DECIDE: Is this a final answer? ---
213 if final_answer is not None:
214 steps.append(
215 ReasoningStep(
216 step_number=iteration,
217 thought=thought,
218 action="final_answer",
219 observation=final_answer,
220 )
221 )
222 logger.info(
223 "react_final_answer",
224 iteration=iteration,
225 steps=len(steps),
226 tool_calls=len(tool_calls),
227 )
228 elapsed = (time.monotonic() - start_time) * 1000
229 return Ok(
230 AgentResponse(
231 message=final_answer,
232 steps=steps,
233 tool_calls=tool_calls,
234 total_tokens=usage.total_tokens,
235 prompt_tokens=usage.prompt_tokens,
236 completion_tokens=usage.completion_tokens,
237 duration_ms=elapsed,
238 metadata={
239 "strategy": "react",
240 "iterations": iteration,
241 },
242 )
243 )
245 # --- ACT: Parse tool call ---
246 tool_name, tool_args = extract_tool_call(llm_text)
248 if tool_name is None:
249 # No tool call and no final answer — nudge the LLM
250 steps.append(
251 ReasoningStep(
252 step_number=iteration,
253 thought=thought,
254 action=None,
255 observation="[No valid action detected — retrying]",
256 )
257 )
258 messages.append(ChatMessage(role=Role.ASSISTANT, content=llm_text))
259 messages.append(
260 ChatMessage(
261 role=Role.USER,
262 content=(
263 "Your response did not contain a valid ACTION or "
264 "FINAL_ANSWER. Please respond with either:\n"
265 "- ACTION: <tool_name> and ACTION_INPUT: <json_args>\n"
266 "- FINAL_ANSWER: <your answer>"
267 ),
268 )
269 )
270 continue
272 # --- EXECUTE: Run tool with timeout ---
273 logger.info(
274 "react_tool_call",
275 iteration=iteration,
276 tool=tool_name,
277 )
279 tool_record = await self._execute_tool(tool_name, tool_args, tool_map)
280 tool_calls.append(tool_record)
282 # Build observation
283 if tool_record.succeeded:
284 obs_text = str(tool_record.result)
285 else:
286 obs_text = f"Error: {tool_record.error}"
288 # Guard before truncation so detectors see the full content
289 from lexigram.ai.agents.strategies.guard_hook import guard_observation
291 obs_text = await guard_observation(
292 guard_pipeline, obs_text, tool_name=tool_name
293 )
295 # Truncate large outputs
296 if len(obs_text) > self.observation_max_chars:
297 obs_text = obs_text[: self.observation_max_chars] + "\n[TRUNCATED]"
299 steps.append(
300 ReasoningStep(
301 step_number=iteration,
302 thought=thought,
303 action=tool_name,
304 tool_call=tool_record,
305 observation=obs_text,
306 )
307 )
309 # --- OBSERVE: Feed result back ---
310 messages.append(ChatMessage(role=Role.ASSISTANT, content=llm_text))
311 messages.append(
312 ChatMessage(
313 role=Role.USER,
314 content=_OBSERVATION_TEMPLATE.format(observation=obs_text),
315 )
316 )
318 # Max iterations reached
319 elapsed = (time.monotonic() - start_time) * 1000
320 logger.warning(
321 "react_max_iterations",
322 max_iterations=self.max_iterations,
323 steps=len(steps),
324 )
325 # Return the best answer we have so far
326 last_obs = steps[-1].observation if steps else "No response generated"
327 return Ok(
328 AgentResponse(
329 message=f"[Max iterations reached] {last_obs}",
330 steps=steps,
331 tool_calls=tool_calls,
332 total_tokens=usage.total_tokens,
333 prompt_tokens=usage.prompt_tokens,
334 completion_tokens=usage.completion_tokens,
335 duration_ms=elapsed,
336 metadata={
337 "strategy": "react",
338 "iterations": self.max_iterations,
339 "max_iterations_reached": True,
340 },
341 )
342 )
344 # ------------------------------------------------------------------
345 # LLM Interaction
346 # ------------------------------------------------------------------
348 async def _call_llm(
349 self,
350 llm: LLMClientProtocol,
351 messages: list[ChatMessage],
352 **kwargs: Any,
353 ) -> CompletionProtocol | None:
354 """Call the LLM and return the completion, or ``None`` on failure."""
355 try:
356 result = await asyncio.wait_for(
357 llm.complete(cast("list[Any]", messages), **kwargs),
358 timeout=self.timeout,
359 )
360 except TimeoutError:
361 logger.warning("react_llm_timeout", timeout=self.timeout)
362 return None
363 except (OSError, ConnectionError, RuntimeError, ValueError) as exc:
364 logger.warning("react_llm_error", error=str(exc))
365 return None
367 if not result.is_ok():
368 logger.warning("react_llm_err_result", error=str(result.unwrap_err()))
369 return None
371 return result.unwrap()
373 # ------------------------------------------------------------------
374 # Tool Execution
375 # ------------------------------------------------------------------
377 async def _execute_tool(
378 self,
379 tool_name: str,
380 tool_args: dict[str, Any],
381 tool_map: dict[str, ToolProtocol],
382 ) -> ToolExecutionRecord:
383 """Execute a tool with timeout, retry on transient errors, and recovery.
385 Retries up to ``self.tool_max_retries`` times on transient infrastructure
386 errors (``ConnectionError``, ``OSError``) using exponential back-off
387 (1 s, 2 s, 4 s, …). ``TimeoutError`` and non-transient exceptions
388 short-circuit immediately without retrying.
390 Args:
391 tool_name: Name of the tool to execute.
392 tool_args: Arguments to pass to the tool.
393 tool_map: Mapping of tool names to tool instances.
395 Returns:
396 ``ToolExecutionRecord`` with the result or error details.
397 """
398 if tool_name not in tool_map:
399 return ToolExecutionRecord(
400 tool_name=tool_name,
401 arguments=tool_args,
402 error=f"Unknown tool: {tool_name}. Available: {list(tool_map)}",
403 )
405 tool = tool_map[tool_name]
406 start = time.monotonic()
407 last_error: BaseException | None = None
409 for attempt in range(self.tool_max_retries):
410 try:
411 output = await asyncio.wait_for(
412 tool.execute(**tool_args),
413 timeout=self.tool_timeout,
414 )
415 duration = (time.monotonic() - start) * 1000
416 if attempt > 0:
417 logger.info(
418 "react_tool_retry_succeeded",
419 tool=tool_name,
420 attempt=attempt + 1,
421 )
422 return ToolExecutionRecord(
423 tool_name=tool_name,
424 arguments=tool_args,
425 result=output,
426 duration_ms=duration,
427 )
428 except TimeoutError:
429 duration = (time.monotonic() - start) * 1000
430 return ToolExecutionRecord(
431 tool_name=tool_name,
432 arguments=tool_args,
433 error=f"Tool '{tool_name}' timed out after {self.tool_timeout}s",
434 duration_ms=duration,
435 )
436 except (ConnectionError, OSError) as exc:
437 last_error = exc
438 logger.warning(
439 "react_tool_transient_error",
440 tool=tool_name,
441 attempt=attempt + 1,
442 max_retries=self.tool_max_retries,
443 error=str(exc),
444 )
445 if attempt < self.tool_max_retries - 1:
446 await asyncio.sleep(1.0 * (2**attempt))
447 except (RuntimeError, TypeError, ValueError, LookupError) as exc:
448 duration = (time.monotonic() - start) * 1000
449 return ToolExecutionRecord(
450 tool_name=tool_name,
451 arguments=tool_args,
452 error=f"Tool '{tool_name}' failed: {exc}",
453 duration_ms=duration,
454 )
456 duration = (time.monotonic() - start) * 1000
457 return ToolExecutionRecord(
458 tool_name=tool_name,
459 arguments=tool_args,
460 error=(
461 f"Tool '{tool_name}' failed after {self.tool_max_retries} "
462 f"retries: {last_error}"
463 ),
464 duration_ms=duration,
465 )
467 # ------------------------------------------------------------------
468 # Message Building
469 # ------------------------------------------------------------------
471 @staticmethod
472 async def _get_memory_context(memory: MemoryProtocol | None) -> str:
473 """Retrieve context from memory backend if available."""
474 if memory is None:
475 return ""
476 try:
477 past_messages = await memory.get_messages()
478 if past_messages:
479 context_str = "\n".join(str(m) for m in past_messages[-5:])
480 return f"\n\nRelevant context from memory:\n{context_str}"
481 except (RuntimeError, TypeError, ValueError, OSError, AttributeError):
482 # Memory failures must not break the agent
483 pass
484 return ""
487__all__ = ["ReActStrategy"]