Spaces:
Paused
Paused
File size: 4,745 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 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 | """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"))
|