"""
Pattern 112: Forward Planning

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class ForwardPlanningConfig:
    """Configuration for Forward Planning."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class ForwardPlanningState:
    """State information for Forward Planning."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class ForwardPlanningAgent:
    """Agent implementing Forward Planning."""
    
    def __init__(self, name: str, config: ForwardPlanningConfig = None):
        self.name = name
        self.config = config or ForwardPlanningConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = ForwardPlanningState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Forward Planning."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Forward Planning. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class ForwardPlanningPattern:
    """
    Forward Planning - Enhanced implementation
    
    This pattern provides comprehensive functionality for forward planning
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: ForwardPlanningConfig = None):
        self.config = config or ForwardPlanningConfig()
        self.agents: Dict[str, ForwardPlanningAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Forward Planning. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: ForwardPlanningAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Forward Planning",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Forward Planning",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Forward Planning Demo ===\n")
    
    # Create configuration
    config = ForwardPlanningConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = ForwardPlanningPattern(config)
    
    # Create and register agents
    agent1 = ForwardPlanningAgent("Agent1", config)
    agent2 = ForwardPlanningAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate forward planning functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Forward Planning

Pattern 112: 112 Forward Planning Pattern – Line-by-Line Explanation

Overview

Pattern 112: Forward Planning Pattern

Forward planning starts from the initial state and applies actions to reach the goal state. Plans by progressively building toward the goal through valid state transitions.

Code Structure Analysis

File Information

• Pattern Number: 112
• Pattern Name: 112 Forward Planning Pattern
• File Path: patterns\planning_strategy\112_forward_planning_pattern.py
• Total Lines: 281

Detailed Code Explanation

Module Docstring

"""
Pattern 112: Forward Planning Pattern

Forward planning starts from the initial state and applies actions to reach the goal state.
Plans by progressively building toward the goal through valid state transitions.
"""

Explanation: This module docstring describes the purpose and key features of the 112 Forward Planning Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
        from collections import deque  # 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 datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

State (Line 16)

Description: World state representation.

Methods:

__hash__ (Line 21)

def __hash__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

__eq__ (Line 24)

def __eq__(self, other):
    """No method docstring."""

Purpose: No method docstring.

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

satisfies (Line 27)

def satisfies(self, conditions):
    """Check if state satisfies conditions."""

Purpose: Check if state satisfies conditions.

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

Action (Line 33)

Description: Planning action.

Methods:

is_applicable (Line 41)

def is_applicable(self, state):
    """Check if action can be applied in given state."""

Purpose: Check if action can be applied in given state.

Parameters: – self: Instance reference – state: Current workflow state

apply (Line 45)

def apply(self, state):
    """Apply action to state, producing new state."""

Purpose: Apply action to state, producing new state.

Parameters: – self: Instance reference – state: Current workflow state

PlanStep (Line 52)

Description: Step in a plan.

Plan (Line 61)

Description: Complete plan from initial to goal state.

ForwardPlanningPattern (Line 69)

Description: Forward Planning Pattern – State-space forward search.

Methods:

__init__ (Line 74)

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”)

add_action (Line 79)

def add_action(self, action):
    """Register an action."""

Purpose: Register an action.

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

get_applicable_actions (Line 84)

def get_applicable_actions(self, state):
    """Get all actions applicable in current state."""

Purpose: Get all actions applicable in current state.

Parameters: – self: Instance reference – state: Current workflow state

forward_search (Line 88)

def forward_search(self, initial_state, goal_conditions, max_depth):
    """Forward search from initial state to goal.
Uses breadth-first search."""

Purpose: Forward search from initial state to goal. Uses breadth-first search.

Parameters: – self: Instance reference – initial_state: Parameter for initial_state – goal_conditions: Parameter for goal_conditions – max_depth: Parameter for max_depth

run (Line 149)

def run(self, initial_facts, goal_facts, max_depth):
    """Execute forward planning."""

Purpose: Execute forward planning.

Parameters: – self: Instance reference – initial_facts: Parameter for initial_facts – goal_facts: Parameter for goal_facts – max_depth: Parameter for max_depth

get_statistics (Line 205)

def get_statistics(self):
    """Get planning statistics."""

Purpose: Get planning statistics.

Parameters: – self: Instance reference

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 = State(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