Spaces:
Paused
Paused
| """Reflexion reasoning pattern implementation. | |
| Reflexion: Act -> Evaluate -> Self-Reflect -> Repeat | |
| - Actor generates actions based on task context | |
| - Evaluator checks correctness of the result | |
| - Self-Reflection analyzes what went wrong and how to improve | |
| - Reflections persist across attempts for continuous improvement | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from typing import Any | |
| from hermes.core.types import AgentStrategy | |
| logger = logging.getLogger(__name__) | |
| class ReflexionReasoner: | |
| """Implements the Reflexion pattern (Act -> Evaluate -> Reflect -> Repeat).""" | |
| def __init__(self, max_attempts: int = 3) -> None: | |
| self.strategy = AgentStrategy.REFLEXION | |
| self.max_attempts = max_attempts | |
| def create_actor_prompt( | |
| self, task: str, tools: list[dict[str, Any]], reflections: list[str] | None = None | |
| ) -> str: | |
| """Create prompt for the actor to generate an action.""" | |
| tool_descriptions = "\n".join( | |
| f"- {t['name']}: {t['description']}" for t in tools | |
| ) | |
| reflections_section = "" | |
| if reflections: | |
| reflections_section = "\nReflections from previous attempts:\n" | |
| for i, r in enumerate(reflections, 1): | |
| reflections_section += f" {i}. {r}\n" | |
| return f"""You are an AI agent that uses the Reflexion pattern to solve tasks through iterative refinement. | |
| Task: {task} | |
| Available tools: | |
| {tool_descriptions} | |
| {reflections_section} | |
| Generate the next action to solve this task. Use the following format: | |
| Thought: [your reasoning about what to do] | |
| Action: [tool_name with arguments as JSON] | |
| Expected: [what you expect the result to be] | |
| If you have enough information to provide a final answer, use: | |
| Thought: I now have enough information. | |
| Final Answer: [your complete answer] | |
| Important: | |
| - Learn from past reflections and avoid repeating mistakes | |
| - Be precise in your tool arguments | |
| - Verify your assumptions""" | |
| def create_evaluator_prompt(self, task: str, result: str) -> str: | |
| """Create prompt for the evaluator to check result correctness.""" | |
| return f"""Evaluate whether the following result correctly addresses the task. | |
| Task: {task} | |
| Result to evaluate: | |
| {result} | |
| Determine if the result is correct and complete. | |
| Format: | |
| Status: [correct / incorrect / partial] | |
| Score: [0-100] | |
| Issues: | |
| - [issue 1] | |
| Missing: | |
| - [missing 1]""" | |
| def create_reflection_prompt( | |
| self, task: str, action: str, result: str, evaluation: str | |
| ) -> str: | |
| """Create prompt for self-reflection on what went wrong.""" | |
| return f"""Analyze what happened and generate a reflection to improve future attempts. | |
| Task: {task} | |
| Action taken: | |
| {action} | |
| Result obtained: | |
| {result} | |
| Evaluation: | |
| {evaluation} | |
| Generate a concise self-reflection that identifies: | |
| 1. What went wrong (if anything) | |
| 2. What could be improved | |
| 3. What to do differently next time | |
| Format: | |
| Reflection: [concise analysis of what happened] | |
| Errors: [specific mistakes made] | |
| Improvements: [specific changes for next attempt] | |
| Key Lesson: [single most important lesson]""" | |
| def parse_actor_response(self, response: str) -> dict[str, Any]: | |
| """Parse actor 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("Expected:"): | |
| if current_key and current_value: | |
| result[current_key] = "\n".join(current_value).strip() | |
| current_key = "expected" | |
| current_value = [stripped[len("Expected:"):].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 isinstance(action_str, str) and "(" 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} | |
| except Exception: | |
| pass | |
| return result | |
| def parse_evaluator_response(self, response: str) -> dict[str, Any]: | |
| """Parse evaluator response.""" | |
| result: dict[str, Any] = {"status": "incorrect", "score": 0, "issues": [], "missing": []} | |
| lines = response.strip().split("\n") | |
| current_section = None | |
| current_value: list[str] = [] | |
| for line in lines: | |
| stripped = line.strip() | |
| if stripped.startswith("Status:"): | |
| result["status"] = stripped[len("Status:"):].strip().lower() | |
| elif stripped.startswith("Score:"): | |
| try: | |
| result["score"] = int(stripped.split(":")[1].strip()) | |
| except ValueError: | |
| result["score"] = 0 | |
| elif stripped.startswith("Issues:"): | |
| current_section = "issues" | |
| current_value = [] | |
| elif stripped.startswith("Missing:"): | |
| if current_section: | |
| result[current_section] = current_value | |
| current_section = "missing" | |
| current_value = [] | |
| elif stripped.startswith("- ") and current_section: | |
| current_value.append(stripped[2:].strip()) | |
| if current_section: | |
| result[current_section] = current_value | |
| return result | |
| def parse_reflection_response(self, response: str) -> dict[str, Any]: | |
| """Parse reflection response.""" | |
| result: dict[str, Any] = { | |
| "reflection": "", "errors": [], "improvements": [], "key_lesson": "", | |
| } | |
| lines = response.strip().split("\n") | |
| current_section = None | |
| current_value: list[str] = [] | |
| for line in lines: | |
| stripped = line.strip() | |
| if stripped.startswith("Reflection:"): | |
| result["reflection"] = stripped[len("Reflection:"):].strip() | |
| elif stripped.startswith("Errors:"): | |
| current_section = "errors" | |
| current_value = [] | |
| elif stripped.startswith("Improvements:"): | |
| if current_section: | |
| result[current_section] = current_value | |
| current_section = "improvements" | |
| current_value = [] | |
| elif stripped.startswith("Key Lesson:"): | |
| if current_section: | |
| result[current_section] = current_value | |
| result["key_lesson"] = stripped[len("Key Lesson:"):].strip() | |
| current_section = None | |
| current_value = [] | |
| elif stripped.startswith("- ") and current_section: | |
| current_value.append(stripped[2:].strip()) | |
| if current_section: | |
| result[current_section] = current_value | |
| return result | |
| def should_continue( | |
| self, status: str, score: int, attempt: int | |
| ) -> bool: | |
| """Determine if another attempt should be made.""" | |
| if status == "correct" and score >= 80: | |
| return False | |
| return attempt < self.max_attempts | |