Add purpose_agent/actor.py

purpose_agent/actor.py

@@ -0,0 +1,317 @@
+"""
+Actor Module — The agent that executes actions in the environment.
+
+Implements a ReAct-style (Reason + Act) loop where each step produces:
+  1. Thought: Chain-of-thought reasoning about the current state
+  2. Action: What to do next (name + params)
+  3. Expected Delta: What the actor predicts will change
+
+The Actor's system prompt is dynamically composed from:
+  - Base instructions (static)
+  - Strategic memory heuristics (updated after each task — from MUSE)
+  - Retrieved procedural SOPs (fetched on demand — from MUSE)
+  - Tool-level "muscle memory" (returned with each observation — from MUSE)
+
+This module is intentionally stateless between tasks — all learning happens
+via the memory system that feeds into the prompt.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from purpose_agent.types import Action, Heuristic, MemoryTier, State
+from purpose_agent.llm_backend import ChatMessage, LLMBackend
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# System Prompt Templates
+# ---------------------------------------------------------------------------
+
+ACTOR_SYSTEM_PROMPT = """\
+You are a goal-directed agent. Your purpose is to achieve the stated goal
+by taking incremental actions that each move the state closer to the goal.
+
+## Your Decision Process
+For each step, you MUST:
+1. THINK: Analyze the current state. What has been achieved? What remains?
+2. ACT: Choose the single best next action from available actions.
+3. PREDICT: State specifically what you expect to change after this action.
+
+## Rules
+- Take ONE action per step. Never skip ahead or combine actions.
+- Be specific in your predictions — name exact state fields you expect to change.
+- If a previous action didn't produce the expected result, adapt your strategy.
+- If you believe the goal is achieved, use action "DONE" with no parameters.
+
+## Available Actions
+{available_actions}
+
+## Learned Strategies (from past experience)
+{strategic_memory}
+
+## Relevant Procedures
+{procedural_memory}
+"""
+
+ACTOR_STEP_PROMPT = """\
+## Current Goal
+{purpose}
+
+## Current State
+{state}
+
+## Action History (last {history_window} steps)
+{history}
+
+## Tool Tips
+{tool_memory}
+
+Based on the current state and your goal, decide your next action.
+
+Respond in this exact JSON format:
+```json
+{{
+  "thought": "Your reasoning about the current state and what to do next",
+  "action": {{
+    "name": "action_name",
+    "params": {{"param1": "value1"}}
+  }},
+  "expected_delta": "Specific prediction of what will change in the state"
+}}
+```
+"""
+
+
+class Actor:
+    """
+    The Actor agent — executes actions in an environment.
+    
+    The Actor does NOT evaluate its own performance. That's the Purpose
+    Function's job. The Actor just reasons, acts, and predicts.
+    
+    Architecture notes (from MUSE arxiv:2510.08002):
+    - System prompt is composed dynamically from 3-tier memory
+    - Strategic memory is always present (global dilemmas → strategies)
+    - Procedural memory is lazy-loaded (index in prompt, details on demand)
+    - Tool memory is returned per-step (dynamic instructions with observations)
+    
+    Args:
+        llm: The LLM backend to use for reasoning
+        available_actions: Dict of {action_name: description} the agent can take
+        history_window: How many past steps to include in the prompt
+        strategic_memory: List of strategic heuristics (loaded at task start)
+        procedural_memory: List of procedural SOPs (indexed, fetched on demand)
+        tool_memory: Dict of {action_name: dynamic_tip} (updated per-step)
+    """
+
+    def __init__(
+        self,
+        llm: LLMBackend,
+        available_actions: dict[str, str] | None = None,
+        history_window: int = 5,
+        strategic_memory: list[Heuristic] | None = None,
+        procedural_memory: list[Heuristic] | None = None,
+        tool_memory: dict[str, str] | None = None,
+    ):
+        self.llm = llm
+        self.available_actions = available_actions or {"DONE": "Signal that the goal is achieved"}
+        self.history_window = history_window
+        self.strategic_memory = strategic_memory or []
+        self.procedural_memory = procedural_memory or []
+        self.tool_memory = tool_memory or {}
+
+    # ------------------------------------------------------------------
+    # Prompt Composition
+    # ------------------------------------------------------------------
+
+    def _format_actions(self) -> str:
+        if not self.available_actions:
+            return "No specific action constraints. You may take any action."
+        lines = []
+        for name, desc in self.available_actions.items():
+            lines.append(f"- **{name}**: {desc}")
+        return "\n".join(lines)
+
+    def _format_strategic_memory(self) -> str:
+        if not self.strategic_memory:
+            return "None yet — this is your first task."
+        lines = []
+        for h in sorted(self.strategic_memory, key=lambda x: -x.q_value):
+            lines.append(f"- When: {h.pattern}\n  Do: {h.strategy} (confidence: {h.q_value:.2f})")
+        return "\n".join(lines)
+
+    def _format_procedural_memory(self) -> str:
+        if not self.procedural_memory:
+            return "No standard operating procedures available."
+        lines = ["Available SOPs (ask for details if relevant):"]
+        for h in self.procedural_memory:
+            lines.append(f"- [{h.id}] {h.pattern}: {h.strategy}")
+        return "\n".join(lines)
+
+    def _format_tool_memory(self) -> str:
+        if not self.tool_memory:
+            return "No tool-specific tips available."
+        lines = []
+        for action_name, tip in self.tool_memory.items():
+            lines.append(f"- **{action_name}**: {tip}")
+        return "\n".join(lines)
+
+    def _format_history(self, history: list[dict[str, Any]]) -> str:
+        if not history:
+            return "No actions taken yet."
+        recent = history[-self.history_window:]
+        lines = []
+        for i, entry in enumerate(recent):
+            step_num = len(history) - len(recent) + i + 1
+            lines.append(
+                f"Step {step_num}: Action={entry.get('action', 'N/A')}, "
+                f"Result={entry.get('result', 'N/A')[:200]}"
+            )
+        return "\n".join(lines)
+
+    def _build_system_prompt(self) -> str:
+        return ACTOR_SYSTEM_PROMPT.format(
+            available_actions=self._format_actions(),
+            strategic_memory=self._format_strategic_memory(),
+            procedural_memory=self._format_procedural_memory(),
+        )
+
+    def _build_step_prompt(
+        self, purpose: str, state: State, history: list[dict[str, Any]]
+    ) -> str:
+        return ACTOR_STEP_PROMPT.format(
+            purpose=purpose,
+            state=state.describe(),
+            history=self._format_history(history),
+            tool_memory=self._format_tool_memory(),
+            history_window=self.history_window,
+        )
+
+    # ------------------------------------------------------------------
+    # Core Action Generation
+    # ------------------------------------------------------------------
+
+    def decide(
+        self,
+        purpose: str,
+        current_state: State,
+        history: list[dict[str, Any]] | None = None,
+    ) -> Action:
+        """
+        Given the current state and purpose, decide the next action.
+        
+        Returns an Action with thought, name, params, and expected_delta.
+        """
+        history = history or []
+
+        messages = [
+            ChatMessage(role="system", content=self._build_system_prompt()),
+            ChatMessage(role="user", content=self._build_step_prompt(
+                purpose=purpose,
+                state=current_state,
+                history=history,
+            )),
+        ]
+
+        # Try structured output first, fall back to text parsing
+        schema = {
+            "type": "object",
+            "properties": {
+                "thought": {"type": "string"},
+                "action": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "params": {"type": "object"},
+                    },
+                    "required": ["name"],
+                },
+                "expected_delta": {"type": "string"},
+            },
+            "required": ["thought", "action", "expected_delta"],
+        }
+
+        try:
+            result = self.llm.generate_structured(messages, schema=schema)
+        except Exception as e:
+            logger.warning(f"Structured output failed ({e}), falling back to text parse")
+            raw = self.llm.generate(messages, temperature=0.7)
+            result = self._parse_action_text(raw)
+
+        action_data = result.get("action", {})
+        return Action(
+            name=action_data.get("name", "UNKNOWN"),
+            params=action_data.get("params", {}),
+            thought=result.get("thought", ""),
+            expected_delta=result.get("expected_delta", ""),
+        )
+
+    # ------------------------------------------------------------------
+    # Memory Updates (called by Orchestrator between tasks)
+    # ------------------------------------------------------------------
+
+    def update_strategic_memory(self, heuristics: list[Heuristic]) -> None:
+        """Replace strategic memory with updated heuristics."""
+        self.strategic_memory = [
+            h for h in heuristics if h.tier == MemoryTier.STRATEGIC
+        ]
+        logger.info(f"Actor strategic memory updated: {len(self.strategic_memory)} heuristics")
+
+    def update_procedural_memory(self, heuristics: list[Heuristic]) -> None:
+        """Update the procedural SOP index."""
+        self.procedural_memory = [
+            h for h in heuristics if h.tier == MemoryTier.PROCEDURAL
+        ]
+        logger.info(f"Actor procedural memory updated: {len(self.procedural_memory)} SOPs")
+
+    def update_tool_memory(self, tips: dict[str, str]) -> None:
+        """Update per-action tool tips."""
+        self.tool_memory.update(tips)
+        logger.info(f"Actor tool memory updated: {list(tips.keys())}")
+
+    # ------------------------------------------------------------------
+    # Fallback Text Parser
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _parse_action_text(raw: str) -> dict[str, Any]:
+        """Best-effort extraction of action JSON from free-form text."""
+        import re
+
+        # Try to find JSON block
+        json_match = re.search(r'\{[^{}]*"thought"[^{}]*\}', raw, re.DOTALL)
+        if json_match:
+            try:
+                return json.loads(json_match.group())
+            except json.JSONDecodeError:
+                pass
+
+        # Try to find JSON in code blocks
+        code_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', raw, re.DOTALL)
+        if code_match:
+            try:
+                return json.loads(code_match.group(1))
+            except json.JSONDecodeError:
+                pass
+
+        # Last resort: extract key-value pairs
+        thought = ""
+        thought_match = re.search(r'"thought"\s*:\s*"([^"]*)"', raw)
+        if thought_match:
+            thought = thought_match.group(1)
+
+        action_name = "UNKNOWN"
+        name_match = re.search(r'"name"\s*:\s*"([^"]*)"', raw)
+        if name_match:
+            action_name = name_match.group(1)
+
+        return {
+            "thought": thought or raw[:200],
+            "action": {"name": action_name, "params": {}},
+            "expected_delta": "Unable to parse prediction",
+        }