"""
Pattern 111: Hierarchical Task Network (HTN) Pattern

Decomposes high-level tasks into hierarchical subtasks until primitive actions are reached.
Implements HTN planning with LangChain for task decomposition.
"""

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


class TaskType(Enum):
    """Type of task in HTN."""
    PRIMITIVE = "primitive"  # Directly executable action
    COMPOUND = "compound"    # Needs decomposition


@dataclass
class Task:
    """Represents a task in the HTN."""
    task_id: str
    name: str
    task_type: TaskType
    parameters: Dict[str, Any] = field(default_factory=dict)
    subtasks: List['Task'] = field(default_factory=list)
    preconditions: List[str] = field(default_factory=list)
    effects: List[str] = field(default_factory=list)
    status: str = "pending"  # pending, executing, completed, failed
    result: Optional[Any] = None


@dataclass
class Method:
    """Decomposition method for compound tasks."""
    method_id: str
    name: str
    task_name: str  # Which compound task this decomposes
    preconditions: List[str]
    subtasks_template: List[Dict[str, Any]]  # Template for subtasks


@dataclass
class Operator:
    """Primitive operator that can be executed."""
    operator_id: str
    name: str
    parameters: List[str]
    preconditions: List[str]
    effects: List[str]
    execute_func: Optional[callable] = None


class HTNPlanner:
    """
    Hierarchical Task Network planner that decomposes tasks using methods.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.methods: Dict[str, List[Method]] = {}  # task_name -> methods
        self.operators: Dict[str, Operator] = {}
        self.world_state: Dict[str, bool] = {}
        
    def add_method(self, method: Method):
        """Add a decomposition method."""
        if method.task_name not in self.methods:
            self.methods[method.task_name] = []
        self.methods[method.task_name].append(method)
        
    def add_operator(self, operator: Operator):
        """Add a primitive operator."""
        self.operators[operator.name] = operator
        
    def set_world_state(self, state: Dict[str, bool]):
        """Set the current world state."""
        self.world_state = state.copy()
        
    def check_preconditions(self, preconditions: List[str]) -> bool:
        """Check if preconditions are satisfied."""
        for precond in preconditions:
            if precond.startswith("not "):
                pred = precond[4:]
                if self.world_state.get(pred, False):
                    return False
            else:
                if not self.world_state.get(precond, False):
                    return False
        return True
    
    def apply_effects(self, effects: List[str]):
        """Apply effects to world state."""
        for effect in effects:
            if effect.startswith("not "):
                pred = effect[4:]
                self.world_state[pred] = False
            else:
                self.world_state[effect] = True
    
    def decompose_task(self, task: Task) -> Optional[List[Task]]:
        """Decompose a compound task using applicable methods."""
        if task.task_type == TaskType.PRIMITIVE:
            return [task]
        
        # Find applicable methods
        methods = self.methods.get(task.name, [])
        applicable_methods = [
            m for m in methods
            if self.check_preconditions(m.preconditions)
        ]
        
        if not applicable_methods:
            # Use LLM to suggest decomposition
            return self._llm_decompose(task)
        
        # Choose first applicable method
        method = applicable_methods[0]
        
        # Create subtasks from template
        subtasks = []
        for subtask_template in method.subtasks_template:
            subtask = Task(
                task_id=f"{task.task_id}-{len(subtasks)+1}",
                name=subtask_template["name"],
                task_type=TaskType(subtask_template.get("type", "primitive")),
                parameters=subtask_template.get("parameters", {}),
                preconditions=subtask_template.get("preconditions", []),
                effects=subtask_template.get("effects", [])
            )
            subtasks.append(subtask)
        
        task.subtasks = subtasks
        return subtasks
    
    def _llm_decompose(self, task: Task) -> List[Task]:
        """Use LLM to decompose a task when no methods match."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an HTN planner. Decompose tasks into subtasks."),
            ("human", """Task: {task_name}
Parameters: {parameters}
Current State: {state}

Decompose this task into 2-5 concrete subtasks.
Format each as: TaskName|type|description

Where type is either 'primitive' or 'compound'""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "task_name": task.name,
            "parameters": task.parameters,
            "state": self.world_state
        })
        
        # Parse response
        lines = str(response.content).strip().split('\n')
        subtasks = []
        
        for i, line in enumerate(lines[:5]):
            parts = line.split('|')
            if len(parts) >= 2:
                subtask = Task(
                    task_id=f"{task.task_id}-{i+1}",
                    name=parts[0].strip(),
                    task_type=TaskType.PRIMITIVE if "primitive" in parts[1].lower() else TaskType.COMPOUND,
                    parameters=task.parameters
                )
                subtasks.append(subtask)
        
        task.subtasks = subtasks
        return subtasks
    
    def plan(self, root_task: Task) -> List[Task]:
        """Create a plan by decomposing the root task."""
        plan = []
        queue = [root_task]
        
        while queue:
            task = queue.pop(0)
            
            if task.task_type == TaskType.PRIMITIVE:
                plan.append(task)
            else:
                # Decompose compound task
                subtasks = self.decompose_task(task)
                if subtasks:
                    # Add subtasks to front of queue
                    queue = subtasks + queue
        
        return plan
    
    def execute_plan(self, plan: List[Task]) -> List[Dict[str, Any]]:
        """Execute a plan of primitive tasks."""
        results = []
        
        for task in plan:
            if not self.check_preconditions(task.preconditions):
                results.append({
                    "task_id": task.task_id,
                    "name": task.name,
                    "status": "failed",
                    "reason": "Preconditions not met"
                })
                break
            
            # Execute task
            operator = self.operators.get(task.name)
            
            if operator and operator.execute_func:
                try:
                    result = operator.execute_func(task.parameters)
                    task.status = "completed"
                    task.result = result
                    
                    # Apply effects
                    self.apply_effects(task.effects)
                    
                    results.append({
                        "task_id": task.task_id,
                        "name": task.name,
                        "status": "completed",
                        "result": result
                    })
                except Exception as e:
                    results.append({
                        "task_id": task.task_id,
                        "name": task.name,
                        "status": "failed",
                        "error": str(e)
                    })
                    break
            else:
                # Simulate execution with LLM
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are executing a primitive task."),
                    ("human", "Execute: {task_name} with parameters {parameters}")
                ])
                
                chain = prompt | self.llm
                response = chain.invoke({
                    "task_name": task.name,
                    "parameters": task.parameters
                })
                
                task.status = "completed"
                task.result = str(response.content)
                self.apply_effects(task.effects)
                
                results.append({
                    "task_id": task.task_id,
                    "name": task.name,
                    "status": "completed",
                    "result": task.result
                })
        
        return results


class HTNPattern:
    """
    Complete HTN planning pattern with method library.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.planner = HTNPlanner(model_name=model_name)
        self._setup_default_methods()
        
    def _setup_default_methods(self):
        """Setup default methods for common tasks."""
        # Example: Travel method
        self.planner.add_method(Method(
            method_id="travel-1",
            name="Travel by Air",
            task_name="travel",
            preconditions=["has_airport"],
            subtasks_template=[
                {"name": "book_flight", "type": "primitive", "effects": ["flight_booked"]},
                {"name": "go_to_airport", "type": "primitive", "preconditions": ["flight_booked"]},
                {"name": "board_plane", "type": "primitive", "preconditions": ["at_airport"]},
                {"name": "fly", "type": "primitive", "effects": ["at_destination"]}
            ]
        ))
        
    def run(self, task: Task, initial_state: Dict[str, bool]) -> Dict[str, Any]:
        """Run HTN planning and execution."""
        self.planner.set_world_state(initial_state)
        
        # Create plan
        plan = self.planner.plan(task)
        
        # Execute plan
        results = self.planner.execute_plan(plan)
        
        return {
            "task": task.name,
            "plan_length": len(plan),
            "plan": [{"id": t.task_id, "name": t.name} for t in plan],
            "execution_results": results,
            "final_state": self.planner.world_state
        }


def main():
    """Example usage of HTN Pattern."""
    print("=" * 80)
    print("Hierarchical Task Network (HTN) Pattern Demo")
    print("=" * 80)
    
    # Create HTN pattern
    pattern = HTNPattern()
    planner = pattern.planner
    
    # Add methods for "prepare_meal"
    planner.add_method(Method(
        method_id="meal-1",
        name="Prepare Pasta Meal",
        task_name="prepare_meal",
        preconditions=["has_ingredients"],
        subtasks_template=[
            {"name": "boil_water", "type": "primitive", "effects": ["water_boiling"]},
            {"name": "cook_pasta", "type": "primitive", "preconditions": ["water_boiling"], "effects": ["pasta_cooked"]},
            {"name": "prepare_sauce", "type": "primitive", "effects": ["sauce_ready"]},
            {"name": "combine", "type": "primitive", "preconditions": ["pasta_cooked", "sauce_ready"], "effects": ["meal_ready"]}
        ]
    ))
    
    # Add operators
    planner.add_operator(Operator(
        operator_id="op-1",
        name="boil_water",
        parameters=["amount"],
        preconditions=[],
        effects=["water_boiling"]
    ))
    
    # Define root task
    root_task = Task(
        task_id="task-1",
        name="prepare_meal",
        task_type=TaskType.COMPOUND,
        parameters={"meal_type": "pasta", "servings": 2}
    )
    
    # Initial world state
    initial_state = {
        "has_ingredients": True,
        "has_kitchen": True
    }
    
    # Run HTN planning
    print("\n🎯 Root Task:")
    print(f"  {root_task.name} (type: {root_task.task_type.value})")
    print(f"  Parameters: {root_task.parameters}")
    
    print("\n📍 Initial State:")
    for key, value in initial_state.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 80)
    print("🔄 Planning and Execution")
    print("=" * 80)
    
    result = pattern.run(root_task, initial_state)
    
    # Display plan
    print("\n📋 Generated Plan:")
    for i, step in enumerate(result["plan"], 1):
        print(f"  {i}. {step['name']} ({step['id']})")
    
    # Display execution results
    print("\n✅ Execution Results:")
    for exec_result in result["execution_results"]:
        print(f"\n  {exec_result['name']} ({exec_result['task_id']})")
        print(f"    Status: {exec_result['status']}")
        if exec_result.get('result'):
            print(f"    Result: {str(exec_result['result'])[:100]}...")
    
    # Display final state
    print("\n📍 Final State:")
    for key, value in result["final_state"].items():
        print(f"  {key}: {value}")
    
    print(f"\n✨ Plan completed with {len(result['plan'])} steps")


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Hierarchical Task Network

Pattern 111: 111 Hierarchical Task Network Htn Pattern – Line-by-Line Explanation

Overview

Pattern 111: Hierarchical Task Network (HTN) Pattern

Planning approach that decomposes high-level tasks into subtasks recursively until primitive actions are reached. Uses methods to define decompositions.

Code Structure Analysis

File Information

• Pattern Number: 111
• Pattern Name: 111 Hierarchical Task Network Htn Pattern
• File Path: patterns\planning_strategy\111_hierarchical_task_network_htn_pattern.py
• Total Lines: 249

Detailed Code Explanation

Module Docstring

"""
Pattern 111: Hierarchical Task Network (HTN) Pattern

Planning approach that decomposes high-level tasks into subtasks recursively
until primitive actions are reached. Uses methods to define decompositions.
"""

Explanation: This module docstring describes the purpose and key features of the 111 Hierarchical Task Network Htn Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
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: 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

TaskType (Line 15)

class TaskType(Enum):
    COMPOUND
    PRIMITIVE

Purpose: Type of task in HTN.

Values: – COMPOUND: Represents compound – PRIMITIVE: Represents primitive

Classes

TaskType (Line 15)

Description: Type of task in HTN.

Task (Line 22)

Description: Task in HTN.

Method (Line 32)

Description: Method for decomposing compound tasks.

TaskNetwork (Line 41)

Description: Network of tasks to execute.

HTNPattern (Line 47)

Description: Hierarchical Task Network Pattern – Recursive task decomposition.

Methods:

__init__ (Line 52)

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_task (Line 58)

def add_task(self, task):
    """Register a task."""

Purpose: Register a task.

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

add_method (Line 64)

def add_method(self, method):
    """Register a decomposition method."""

Purpose: Register a decomposition method.

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

decompose (Line 70)

def decompose(self, task_name, state):
    """Decompose a compound task into subtasks."""

Purpose: Decompose a compound task into subtasks.

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

_check_preconditions (Line 96)

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

Purpose: Check if preconditions are met.

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

plan (Line 103)

def plan(self, goal_task, initial_state):
    """Create a plan by decomposing the goal task."""

Purpose: Create a plan by decomposing the goal task.

Parameters: – self: Instance reference – goal_task: Parameter for goal_task – initial_state: Parameter for initial_state

execute_plan (Line 118)

def execute_plan(self, plan):
    """Execute the primitive tasks in the plan."""

Purpose: Execute the primitive tasks in the plan.

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

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