"""
Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes.
Uses search algorithms and domain knowledge for goal-directed behavior.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ActionStatus(Enum):
    PLANNED = "planned"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Action:
    """An action in a plan."""
    name: str
    preconditions: List[str]
    effects: List[str]
    cost: float = 1.0
    status: ActionStatus = ActionStatus.PLANNED


@dataclass
class Plan:
    """A sequence of actions to achieve a goal."""
    goal: str
    actions: List[Action] = field(default_factory=list)
    current_state: Set[str] = field(default_factory=set)
    target_state: Set[str] = field(default_factory=set)


class PlanningAgentPattern:
    """
    Planning Agent Pattern - Goal-directed planning and execution.
    
    Features:
    - Goal decomposition
    - Precondition checking
    - Action sequencing
    - Plan execution
    - Replanning on failure
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.available_actions: Dict[str, Action] = {}
        self.plans: List[Plan] = []
    
    def register_action(self, action: Action):
        """Register an available action."""
        self.available_actions[action.name] = action
        print(f"✓ Registered action: {action.name}")
        print(f"  Preconditions: {', '.join(action.preconditions)}")
        print(f"  Effects: {', '.join(action.effects)}")
    
    def create_plan(self, goal: str, current_state: Set[str], 
                   target_conditions: List[str]) -> Plan:
        """Create a plan to achieve goal."""
        plan = Plan(
            goal=goal,
            current_state=current_state.copy(),
            target_state=set(target_conditions)
        )
        
        # Use LLM to generate plan
        state_str = ", ".join(current_state)
        target_str = ", ".join(target_conditions)
        actions_str = "\n".join([
            f"- {name}: requires [{', '.join(a.preconditions)}] → produces [{', '.join(a.effects)}]"
            for name, a in self.available_actions.items()
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a planning agent. Create a sequence of actions to achieve the goal. "
                      "List actions in order, one per line."),
            ("human", f"Goal: {goal}\n"
                     f"Current State: {state_str}\n"
                     f"Target Conditions: {target_str}\n\n"
                     f"Available Actions:\n{actions_str}\n\n"
                     f"Plan (list action names in order):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse action sequence
        lines = [line.strip() for line in response.content.split('\n') if line.strip()]
        for line in lines:
            # Extract action name
            clean_line = line.lstrip('0123456789.-•* ').strip()
            # Find matching action
            for action_name, action_template in self.available_actions.items():
                if action_name.lower() in clean_line.lower():
                    # Create action instance
                    action_instance = Action(
                        name=action_template.name,
                        preconditions=action_template.preconditions.copy(),
                        effects=action_template.effects.copy(),
                        cost=action_template.cost
                    )
                    plan.actions.append(action_instance)
                    break
        
        self.plans.append(plan)
        return plan
    
    def check_preconditions(self, action: Action, state: Set[str]) -> bool:
        """Check if action preconditions are met."""
        return all(precond in state for precond in action.preconditions)
    
    def execute_plan(self, plan: Plan) -> Dict[str, Any]:
        """Execute a plan step by step."""
        current_state = plan.current_state.copy()
        executed_actions = []
        failed_action = None
        
        for action in plan.actions:
            # Check preconditions
            if not self.check_preconditions(action, current_state):
                action.status = ActionStatus.FAILED
                failed_action = action.name
                print(f"✗ Action failed: {action.name} - preconditions not met")
                break
            
            # Execute action
            action.status = ActionStatus.EXECUTING
            print(f"→ Executing: {action.name}")
            
            # Apply effects
            for effect in action.effects:
                current_state.add(effect)
            
            action.status = ActionStatus.COMPLETED
            executed_actions.append(action.name)
            print(f"✓ Completed: {action.name}")
        
        # Check if goal achieved
        goal_achieved = plan.target_state.issubset(current_state)
        
        return {
            "goal": plan.goal,
            "executed_actions": executed_actions,
            "goal_achieved": goal_achieved,
            "failed_action": failed_action,
            "final_state": current_state
        }
    
    def replan(self, original_plan: Plan, current_state: Set[str]) -> Plan:
        """Create a new plan when original plan fails."""
        print(f"\n🔄 Replanning for goal: {original_plan.goal}")
        return self.create_plan(
            original_plan.goal,
            current_state,
            list(original_plan.target_state)
        )
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute planning scenario."""
        return {
            "pattern": "Planning Agent",
            "scenario": scenario,
            "available_actions": len(self.available_actions),
            "plans_created": len(self.plans)
        }


# Example usage
if __name__ == "__main__":
    print("=== Planning Agent Pattern Demo ===\n")
    
    agent = PlanningAgentPattern()
    
    # Register available actions
    print("--- Registering Actions ---")
    agent.register_action(Action(
        "GetIngredients",
        preconditions=["has_money"],
        effects=["has_ingredients"],
        cost=1.0
    ))
    agent.register_action(Action(
        "PrepareFood",
        preconditions=["has_ingredients", "has_kitchen"],
        effects=["has_prepared_food"],
        cost=2.0
    ))
    agent.register_action(Action(
        "CookFood",
        preconditions=["has_prepared_food"],
        effects=["has_cooked_food"],
        cost=1.5
    ))
    agent.register_action(Action(
        "ServeMeal",
        preconditions=["has_cooked_food", "has_table"],
        effects=["meal_ready"],
        cost=0.5
    ))
    
    # Create plan
    print("\n--- Creating Plan ---")
    initial_state = {"has_money", "has_kitchen", "has_table"}
    goal_conditions = ["meal_ready"]
    
    plan = agent.create_plan(
        "Prepare and serve a meal",
        initial_state,
        goal_conditions
    )
    
    print(f"\nPlan for: {plan.goal}")
    print(f"Actions: {', '.join([a.name for a in plan.actions])}")
    
    # Execute plan
    print("\n--- Executing Plan ---")
    result = agent.execute_plan(plan)
    
    print(f"\nExecution Result:")
    print(f"Goal: {result['goal']}")
    print(f"Executed: {', '.join(result['executed_actions'])}")
    print(f"Goal Achieved: {result['goal_achieved']}")
    if result['failed_action']:
        print(f"Failed At: {result['failed_action']}")
    
    # Summary
    print("\n--- Summary ---")
    summary = agent.run("Meal preparation planning")
    print(f"Pattern: {summary['pattern']}")
    print(f"Available Actions: {summary['available_actions']}")
    print(f"Plans Created: {summary['plans_created']}")

    

Agentic Architecture Pattern: Planning Agent

Pattern 42: 42 Planning Agent – Line-by-Line Explanation

Overview

Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes. Uses search algorithms and domain knowledge for goal-directed behavior.

Code Structure Analysis

File Information

• Pattern Number: 42
• Pattern Name: 42 Planning Agent
• File Path: patterns\decision_making\42_planning_agent.py
• Total Lines: 240

Detailed Code Explanation

Module Docstring

"""
Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes.
Uses search algorithms and domain knowledge for goal-directed behavior.
"""

Explanation: This module docstring describes the purpose and key features of the 42 Planning Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ActionStatus (Line 15)

class ActionStatus(Enum):
    PLANNED
    EXECUTING
    COMPLETED
    FAILED

Purpose: No enum docstring.

Values: – PLANNED: Represents planned – EXECUTING: Represents executing – COMPLETED: Successfully finished – FAILED: Execution failed

Classes

ActionStatus (Line 15)

Description: No class docstring.

Action (Line 23)

Description: An action in a plan.

Plan (Line 33)

Description: A sequence of actions to achieve a goal.

PlanningAgentPattern (Line 41)

Description: Planning Agent Pattern – Goal-directed planning and execution.

Features: – Goal decomposition – Precondition checking – Action sequencing – Plan execution – Replanning on failure

Methods:

__init__ (Line 53)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

register_action (Line 58)

def register_action(self, action):
    """Register an available action."""

Purpose: Register an available action.

Parameters: – self: Instance reference – action: Parameter for action

create_plan (Line 65)

def create_plan(self, goal, current_state, target_conditions):
    """Create a plan to achieve goal."""

Purpose: Create a plan to achieve goal.

Parameters: – self: Instance reference – goal: Parameter for goal – current_state: Parameter for current_state – target_conditions: Parameter for target_conditions

check_preconditions (Line 116)

def check_preconditions(self, action, state):
    """Check if action preconditions are met."""

Purpose: Check if action preconditions are met.

Parameters: – self: Instance reference – action: Parameter for action – state: Current workflow state

execute_plan (Line 120)

def execute_plan(self, plan):
    """Execute a plan step by step."""

Purpose: Execute a plan step by step.

Parameters: – self: Instance reference – plan: Parameter for plan

replan (Line 157)

def replan(self, original_plan, current_state):
    """Create a new plan when original plan fails."""

Purpose: Create a new plan when original plan fails.

Parameters: – self: Instance reference – original_plan: Parameter for original_plan – current_state: Parameter for current_state

run (Line 166)

def run(self, scenario):
    """Execute planning scenario."""

Purpose: Execute planning scenario.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = ActionStatus(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities