Spaces:
Paused
Paused
| """ReAct reasoning pattern implementation.""" | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import Any | |
| from hermes.core.types import AgentStrategy | |
| logger = logging.getLogger(__name__) | |
| class ReActReasoner: | |
| """Implements the ReAct (Reasoning + Acting) pattern.""" | |
| def __init__(self) -> None: | |
| self.strategy = AgentStrategy.REACT | |
| def create_prompt( | |
| self, task: str, tools: list[dict[str, Any]], history: list[dict[str, str]] | |
| ) -> str: | |
| """Create ReAct prompt with tool descriptions.""" | |
| tool_descriptions = "\n".join( | |
| f"- {t['name']}: {t['description']}" for t in tools | |
| ) | |
| history_text = "" | |
| if history: | |
| history_text = "\nPrevious steps:\n" | |
| for h in history: | |
| if "thought" in h: | |
| history_text += f"Thought: {h['thought']}\n" | |
| if "action" in h: | |
| history_text += f"Action: {h['action']}\n" | |
| if "observation" in h: | |
| history_text += f"Observation: {h['observation']}\n" | |
| return f"""You are an AI agent that uses the ReAct pattern to solve tasks. | |
| Task: {task} | |
| Available tools: | |
| {tool_descriptions} | |
| {history_text} | |
| Think step by step. Use the following format: | |
| Thought: [your reasoning about what to do next] | |
| Action: [tool_name with arguments as JSON] | |
| Observation: [this will be provided after action execution] | |
| When you have enough information to answer, use: | |
| Thought: I now have enough information to provide a final answer. | |
| Final Answer: [your complete answer] | |
| Important: | |
| - Always start with a Thought | |
| - Use only the available tools | |
| - Provide clear reasoning in your thoughts | |
| - When done, provide a Final Answer""" | |
| def parse_response(self, response: str) -> dict[str, Any]: | |
| """Parse ReAct response into components.""" | |
| result: dict[str, Any] = {"thought": "", "action": None, "final_answer": None} | |
| lines = response.strip().split("\n") | |
| current_key = None | |
| current_value: list[str] = [] | |
| for line in lines: | |
| stripped = line.strip() | |
| if stripped.startswith("Thought:"): | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| current_key = "thought" | |
| current_value = [stripped[len("Thought:"):].strip()] | |
| elif stripped.startswith("Action:"): | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| current_key = "action" | |
| current_value = [stripped[len("Action:"):].strip()] | |
| elif stripped.startswith("Observation:"): | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| current_key = "observation" | |
| current_value = [stripped[len("Observation:"):].strip()] | |
| elif stripped.startswith("Final Answer:"): | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| current_key = "final_answer" | |
| current_value = [stripped[len("Final Answer:"):].strip()] | |
| elif current_key: | |
| current_value.append(stripped) | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| if result["action"]: | |
| try: | |
| action_str = result["action"] | |
| if "(" in action_str and action_str.endswith(")"): | |
| tool_name = action_str[: action_str.index("(")] | |
| args_str = action_str[action_str.index("(") + 1 : -1] | |
| try: | |
| args = json.loads(args_str) if args_str.strip() else {} | |
| except json.JSONDecodeError: | |
| args = {"input": args_str} | |
| result["action"] = {"tool": tool_name, "arguments": args} | |
| else: | |
| result["action"] = {"tool": action_str, "arguments": {}} | |
| except Exception: | |
| result["action"] = {"tool": result["action"], "arguments": {}} | |
| return result | |
| def should_continue(self, parsed: dict[str, Any], max_steps: int, current_step: int) -> bool: | |
| """Determine if the agent should continue.""" | |
| if parsed.get("final_answer"): | |
| return False | |
| if current_step >= max_steps: | |
| return False | |
| return bool(parsed.get("action")) | |