The Self-Prompting Loop

1"""
2The Self-Prompting Loop
3
4┌─────────────────────────────────────────────────────┐
5│                                                      │
6│     ┌───────────────────────────────────────┐       │
7│     │                                       │       │
8│     ▼                                       │       │
9│  ┌──────────┐                               │       │
10│  │  THINK   │ ─── Generate thoughts about   │       │
11│  └────┬─────┘     current state             │       │
12│       │                                     │       │
13│       ▼                                     │       │
14│  ┌──────────┐                               │       │
15│  │  DECIDE  │ ─── Choose next action        │       │
16│  └────┬─────┘     based on thoughts         │       │
17│       │                                     │       │
18│       ▼                                     │       │
19│  ┌──────────┐                               │       │
20│  │  EXECUTE │ ─── Run the chosen action     │       │
21│  └────┬─────┘                               │       │
22│       │                                     │       │
23│       ▼                                     │       │
24│  ┌──────────┐                               │       │
25│  │ OBSERVE  │ ─── Capture and process       │       │
26│  └────┬─────┘     the result                │       │
27│       │                                     │       │
28│       ▼                                     │       │
29│  ┌──────────┐                               │       │
30│  │ REMEMBER │ ─── Update memory with        │       │
31│  └────┬─────┘     action and result         │       │
32│       │                                     │       │
33│       ▼                                     │       │
34│  ┌──────────┐     ┌─────────────┐          │       │
35│  │ EVALUATE │ ──▶ │ Goal met?   │ ─── No ───┘       │
36│  └──────────┘     └──────┬──────┘                   │
37│                          │                          │
38│                         Yes                         │
39│                          │                          │
40│                          ▼                          │
41│                    ┌──────────┐                     │
42│                    │  FINISH  │                     │
43│                    └──────────┘                     │
44└─────────────────────────────────────────────────────┘
45"""
46
47from typing import TypedDict, Optional
48from langchain_openai import ChatOpenAI
49from langchain_core.messages import SystemMessage, HumanMessage
50import json
51
52
53class LoopState(TypedDict):
54    """State maintained across loop iterations."""
55    goals: list[str]
56    thoughts: str
57    action: str
58    action_input: str
59    observation: str
60    memory: list[dict]
61    iteration: int
62    is_complete: bool
63
64
65class SelfPromptingLoop:
66    """Implementation of the self-prompting loop."""
67
68    def __init__(self, goals: list[str]):
69        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
70        self.state: LoopState = {
71            "goals": goals,
72            "thoughts": "",
73            "action": "",
74            "action_input": "",
75            "observation": "",
76            "memory": [],
77            "iteration": 0,
78            "is_complete": False
79        }
80
81    def run(self, max_iterations: int = 20) -> LoopState:
82        """Execute the self-prompting loop."""
83
84        while not self.state["is_complete"]:
85            if self.state["iteration"] >= max_iterations:
86                break
87
88            # 1. THINK - Generate thoughts
89            self._think()
90
91            # 2. DECIDE - Choose action
92            self._decide()
93
94            # 3. EXECUTE - Run action
95            self._execute()
96
97            # 4. OBSERVE - Process result
98            # (handled in execute)
99
100            # 5. REMEMBER - Update memory
101            self._remember()
102
103            # 6. EVALUATE - Check completion
104            self._evaluate()
105
106            self.state["iteration"] += 1
107
108        return self.state

1class SelfPromptingLoop:
2    # ... continued ...
3
4    def _think(self):
5        """Generate thoughts about current state."""
6        prompt = f"""
7Goals: {self.state['goals']}
8
9Recent actions: {self.state['memory'][-3:] if self.state['memory'] else 'None'}
10
11Current iteration: {self.state['iteration']}
12
13Think about:
141. What progress has been made?
152. What obstacles exist?
163. What should be done next?
174. Are the goals achievable?
18
19Output your thoughts as a stream of consciousness.
20"""
21        messages = [
22            SystemMessage(content="You are reasoning about your current state."),
23            HumanMessage(content=prompt)
24        ]
25
26        response = self.llm.invoke(messages)
27        self.state["thoughts"] = response.content
28
29    def _decide(self):
30        """Choose the next action based on thoughts."""
31        prompt = f"""
32Based on these thoughts:
33{self.state['thoughts']}
34
35Choose your next action. Output as JSON:
36{{
37    "action": "tool_name",
38    "action_input": "input for the tool",
39    "reasoning": "why this action"
40}}
41
42Available actions:
43- search: Search for information
44- write: Write content to a file
45- read: Read a file
46- code: Execute Python code
47- finish: Mark task as complete
48"""
49        messages = [
50            SystemMessage(content="Choose the next action."),
51            HumanMessage(content=prompt)
52        ]
53
54        response = self.llm.invoke(messages)
55
56        try:
57            decision = json.loads(response.content)
58            self.state["action"] = decision["action"]
59            self.state["action_input"] = decision["action_input"]
60        except json.JSONDecodeError:
61            self.state["action"] = "think"
62            self.state["action_input"] = "Failed to parse decision"
63
64    def _execute(self):
65        """Execute the chosen action."""
66        action = self.state["action"]
67        action_input = self.state["action_input"]
68
69        # Execute based on action type
70        if action == "search":
71            self.state["observation"] = self._search(action_input)
72        elif action == "write":
73            self.state["observation"] = self._write_file(action_input)
74        elif action == "read":
75            self.state["observation"] = self._read_file(action_input)
76        elif action == "code":
77            self.state["observation"] = self._execute_code(action_input)
78        elif action == "finish":
79            self.state["observation"] = "Task marked complete"
80            self.state["is_complete"] = True
81        else:
82            self.state["observation"] = f"Unknown action: {action}"
83
84    def _remember(self):
85        """Add current step to memory."""
86        memory_entry = {
87            "iteration": self.state["iteration"],
88            "thoughts": self.state["thoughts"][:200],  # Truncate
89            "action": self.state["action"],
90            "action_input": self.state["action_input"][:100],
91            "observation": self.state["observation"][:200]
92        }
93        self.state["memory"].append(memory_entry)
94
95    def _evaluate(self):
96        """Evaluate if goals are met."""
97        prompt = f"""
98Goals: {self.state['goals']}
99
100Completed actions: {len(self.state['memory'])}
101
102Last observation: {self.state['observation']}
103
104Are all goals achieved? Respond with just 'yes' or 'no'.
105"""
106        messages = [
107            SystemMessage(content="Evaluate goal completion."),
108            HumanMessage(content=prompt)
109        ]
110
111        response = self.llm.invoke(messages)
112        if "yes" in response.content.lower():
113            self.state["is_complete"] = True

1class PromptGenerator:
2    """Generates prompts for self-prompting agents."""
3
4    def __init__(self, persona: str = "autonomous agent"):
5        self.persona = persona
6
7    def build_system_prompt(self, tools: list[str]) -> str:
8        """Build the system prompt with available tools."""
9        tools_desc = "\n".join([f"- {tool}" for tool in tools])
10
11        return f"""You are an {self.persona} working autonomously toward goals.
12
13You operate in a loop:
141. Think about the current situation
152. Decide on the next action
163. Execute the action
174. Observe the result
185. Update your understanding
196. Repeat until goals are achieved
20
21Available tools:
22{tools_desc}
23
24Always output your response as valid JSON:
25{{
26    "thoughts": "Your reasoning about the situation",
27    "action": "The tool to use",
28    "action_input": "Input for the tool"
29}}
30
31Be methodical but efficient. Don't repeat failed actions."""
32
33    def build_context_prompt(
34        self,
35        goals: list[str],
36        memory: list[dict],
37        iteration: int,
38        max_iterations: int
39    ) -> str:
40        """Build prompt with current context."""
41
42        # Format recent memory
43        recent = memory[-5:] if memory else []
44        memory_str = ""
45        for m in recent:
46            memory_str += f"""
47Iteration {m['iteration']}:
48  Action: {m['action']}
49  Result: {m['observation'][:100]}...
50"""
51
52        return f"""
53=== CURRENT STATE ===
54
55Goals to achieve:
56{chr(10).join(f'- {g}' for g in goals)}
57
58Progress: {iteration}/{max_iterations} iterations
59
60Recent history:
61{memory_str if memory_str else 'No previous actions'}
62
63=== YOUR TURN ===
64
65Analyze the situation and decide your next action.
66"""
67
68    def build_reflection_prompt(self, state: dict) -> str:
69        """Build prompt for self-reflection."""
70        return f"""
71Reflect on your progress:
72
73Goals: {state['goals']}
74Actions taken: {len(state['memory'])}
75Last action: {state.get('action', 'None')}
76Last result: {state.get('observation', 'None')[:200]}
77
78Questions to consider:
791. Am I making progress toward the goals?
802. Have I made any mistakes?
813. Should I change my approach?
824. What have I learned?
83
84Provide a brief reflection.
85"""

1class OptimizedPromptGenerator(PromptGenerator):
2    """Optimized prompts for better agent performance."""
3
4    def __init__(self):
5        super().__init__()
6        self.token_budget = 4000  # Max tokens for context
7
8    def compress_memory(self, memory: list[dict]) -> str:
9        """Compress memory to fit token budget."""
10        if not memory:
11            return "No previous actions."
12
13        # Summarize older entries, keep recent detailed
14        recent_count = 3
15        older = memory[:-recent_count] if len(memory) > recent_count else []
16        recent = memory[-recent_count:]
17
18        compressed = ""
19
20        # Summarize older entries
21        if older:
22            actions = [m["action"] for m in older]
23            compressed += f"Earlier: {len(older)} actions ({', '.join(set(actions))}). "
24
25        # Keep recent entries detailed
26        for m in recent:
27            compressed += f"\n[{m['iteration']}] {m['action']}: {m['observation'][:50]}..."
28
29        return compressed
30
31    def add_examples(self, system_prompt: str) -> str:
32        """Add few-shot examples to system prompt."""
33        examples = """
34=== EXAMPLES ===
35
36Example 1 - Research task:
37{{
38    "thoughts": "I need to find information about AI agents. Let me search for recent articles.",
39    "action": "search",
40    "action_input": "AI agents 2025 best practices"
41}}
42
43Example 2 - Writing task:
44{{
45    "thoughts": "I have gathered enough research. Time to write the summary.",
46    "action": "write",
47    "action_input": {{"filename": "summary.md", "content": "# AI Agents Summary\n..."}}
48}}
49
50Example 3 - Completion:
51{{
52    "thoughts": "I have completed all the required tasks. The goals are achieved.",
53    "action": "finish",
54    "action_input": "All goals completed successfully"
55}}
56"""
57        return system_prompt + examples

1from pydantic import BaseModel, Field
2from typing import Literal
3
4
5class AgentAction(BaseModel):
6    """Structured action output."""
7    thoughts: str = Field(description="Reasoning about the situation")
8    action: Literal["search", "write", "read", "code", "think", "finish"]
9    action_input: str = Field(description="Input for the action")
10    confidence: float = Field(ge=0, le=1, description="Confidence in this action")
11
12
13class StructuredActionSelector:
14    """Select actions with structured output."""
15
16    def __init__(self):
17        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
18        self.structured_llm = self.llm.with_structured_output(AgentAction)
19
20    def select_action(
21        self,
22        goals: list[str],
23        memory: list[dict],
24        observation: str
25    ) -> AgentAction:
26        """Select next action with structured output."""
27
28        prompt = f"""
29Goals: {goals}
30
31Last observation: {observation}
32
33Recent actions: {[m['action'] for m in memory[-5:]]}
34
35Decide the next action to take.
36"""
37        messages = [
38            SystemMessage(content="You are selecting the next action."),
39            HumanMessage(content=prompt)
40        ]
41
42        return self.structured_llm.invoke(messages)
43
44
45# Usage
46selector = StructuredActionSelector()
47action = selector.select_action(
48    goals=["Research AI agents"],
49    memory=[{"action": "search", "observation": "Found 10 articles"}],
50    observation="Search completed successfully"
51)
52
53print(f"Action: {action.action}")
54print(f"Input: {action.action_input}")
55print(f"Confidence: {action.confidence}")

1class ActionValidator:
2    """Validate actions before execution."""
3
4    def __init__(self, allowed_actions: list[str]):
5        self.allowed_actions = allowed_actions
6        self.action_history: list[str] = []
7        self.max_repeats = 3
8
9    def validate(self, action: AgentAction) -> tuple[bool, str]:
10        """Validate an action before execution."""
11
12        # Check if action is allowed
13        if action.action not in self.allowed_actions:
14            return False, f"Unknown action: {action.action}"
15
16        # Check for repetition loops
17        if self._is_repeating(action):
18            return False, "Action appears to be in a loop"
19
20        # Check confidence threshold
21        if action.confidence < 0.3:
22            return False, "Low confidence action"
23
24        # Validate action input
25        if not self._validate_input(action):
26            return False, "Invalid action input"
27
28        return True, "Valid"
29
30    def _is_repeating(self, action: AgentAction) -> bool:
31        """Check if we're in a repetition loop."""
32        key = f"{action.action}:{action.action_input[:50]}"
33        self.action_history.append(key)
34
35        if len(self.action_history) > self.max_repeats:
36            recent = self.action_history[-self.max_repeats:]
37            if all(a == recent[0] for a in recent):
38                return True
39        return False
40
41    def _validate_input(self, action: AgentAction) -> bool:
42        """Validate action input."""
43        if action.action == "write":
44            # Check for required fields
45            try:
46                data = json.loads(action.action_input)
47                return "filename" in data and "content" in data
48            except json.JSONDecodeError:
49                return False
50
51        if action.action == "code":
52            # Basic safety check
53            dangerous = ["import os", "subprocess", "eval(", "exec("]
54            return not any(d in action.action_input for d in dangerous)
55
56        return True

1from dataclasses import dataclass
2from datetime import datetime, timedelta
3
4
5@dataclass
6class LoopController:
7    """Control loop execution and termination."""
8
9    max_iterations: int = 20
10    max_duration: timedelta = timedelta(minutes=10)
11    max_cost: float = 1.0  # USD
12    consecutive_failures: int = 3
13
14    def __post_init__(self):
15        self.start_time = datetime.now()
16        self.iteration = 0
17        self.total_cost = 0.0
18        self.failure_count = 0
19        self.termination_reason: str = ""
20
21    def should_continue(self) -> bool:
22        """Check if loop should continue."""
23
24        # Check iteration limit
25        if self.iteration >= self.max_iterations:
26            self.termination_reason = "Max iterations reached"
27            return False
28
29        # Check duration
30        elapsed = datetime.now() - self.start_time
31        if elapsed > self.max_duration:
32            self.termination_reason = "Timeout"
33            return False
34
35        # Check cost
36        if self.total_cost >= self.max_cost:
37            self.termination_reason = "Cost limit reached"
38            return False
39
40        # Check consecutive failures
41        if self.failure_count >= self.consecutive_failures:
42            self.termination_reason = "Too many failures"
43            return False
44
45        return True
46
47    def record_iteration(self, success: bool, cost: float = 0.0):
48        """Record an iteration."""
49        self.iteration += 1
50        self.total_cost += cost
51
52        if success:
53            self.failure_count = 0
54        else:
55            self.failure_count += 1
56
57    def get_status(self) -> dict:
58        """Get current status."""
59        return {
60            "iteration": self.iteration,
61            "max_iterations": self.max_iterations,
62            "elapsed": str(datetime.now() - self.start_time),
63            "cost": self.total_cost,
64            "failures": self.failure_count
65        }

1class GracefulShutdown:
2    """Handle graceful shutdown of autonomous agents."""
3
4    def __init__(self, agent):
5        self.agent = agent
6        self.shutdown_requested = False
7
8    def request_shutdown(self, reason: str):
9        """Request agent to shut down gracefully."""
10        self.shutdown_requested = True
11        self.shutdown_reason = reason
12
13    def should_shutdown(self) -> bool:
14        """Check if shutdown was requested."""
15        return self.shutdown_requested
16
17    def perform_shutdown(self) -> dict:
18        """Perform graceful shutdown."""
19
20        # Save current state
21        state_snapshot = {
22            "goals": self.agent.state["goals"],
23            "progress": self.agent.state["memory"],
24            "iteration": self.agent.state["iteration"],
25            "shutdown_reason": self.shutdown_reason
26        }
27
28        # Generate summary
29        summary = self.agent._generate_summary()
30
31        # Clean up resources
32        self._cleanup()
33
34        return {
35            "snapshot": state_snapshot,
36            "summary": summary,
37            "can_resume": True
38        }
39
40    def _cleanup(self):
41        """Clean up agent resources."""
42        # Close connections, save files, etc.
43        pass