feat: Implement PDDL-based planning system (#503)

src/agents/planner/pddl/__init__.py

@@ -0,0 +1,20 @@
+"""
+PDDL (Planning Domain Definition Language) module for Devika's advanced planning system.
+This module implements PDDL-based planning capabilities as described in:
+https://arxiv.org/abs/2305.14909
+"""
+
+from .predicate import PDDLPredicate
+from .domain import PDDLDomain
+from .action import PDDLAction
+from .problem import PDDLProblem
+from .planner import PDDLPlanner, Plan
+
+__all__ = [
+    'PDDLPredicate',
+    'PDDLDomain',
+    'PDDLAction',
+    'PDDLProblem',
+    'PDDLPlanner',
+    'Plan'
+]

src/agents/planner/pddl/action.py

@@ -0,0 +1,42 @@
+"""
+PDDL Action class for defining planning actions.
+"""
+from typing import List, Dict, Optional
+from dataclasses import dataclass, field
+
+@dataclass
+class PDDLAction:
+    """Represents a PDDL action with parameters, preconditions, and effects."""
+    name: str
+    parameters: List[Dict[str, str]]  # List of {name: type} dicts
+    preconditions: List[str]
+    effects: List[str]
+    duration: Optional[float] = None  # For temporal planning support
+
+    def to_pddl(self) -> str:
+        """Convert the action to PDDL string representation."""
+        # Format parameters
+        params = ' '.join(f"?{name} - {type}"
+                         for param in self.parameters
+                         for name, type in param.items())
+
+        # Format preconditions and effects
+        pre = ' '.join(f"({p})" if not p.startswith('(') else p
+                      for p in self.preconditions)
+        eff = ' '.join(f"({e})" if not e.startswith('(') else e
+                      for e in self.effects)
+
+        # Basic action format
+        pddl = [
+            f"  (:action {self.name}",
+            f"    :parameters ({params})",
+            f"    :precondition (and {pre})",
+            f"    :effect (and {eff})"
+        ]
+
+        # Add duration if specified (for temporal planning)
+        if self.duration is not None:
+            pddl.insert(2, f"    :duration {self.duration}")
+
+        pddl.append("  )")
+        return '\n'.join(pddl)

src/agents/planner/pddl/domain.py

@@ -0,0 +1,43 @@
+"""
+PDDL Domain class for defining planning domains.
+"""
+from typing import List, Dict, Optional
+from dataclasses import dataclass, field
+
+from .predicate import PDDLPredicate
+
+@dataclass
+class PDDLDomain:
+    """Represents a PDDL domain with predicates and requirements."""
+    name: str
+    predicates: List[PDDLPredicate] = field(default_factory=list)
+    requirements: List[str] = field(default_factory=lambda: ['strips', 'typing'])
+    types: List[str] = field(default_factory=list)
+
+    def add_predicate(self, predicate: PDDLPredicate) -> None:
+        """Add a predicate to the domain."""
+        self.predicates.append(predicate)
+
+    def to_pddl(self) -> str:
+        """Convert the domain to PDDL string representation."""
+        pddl = [f"(define (domain {self.name})"]
+
+        # Add requirements
+        if self.requirements:
+            reqs = ' '.join(f":{req}" for req in self.requirements)
+            pddl.append(f"  (:requirements {reqs})")
+
+        # Add types
+        if self.types:
+            types = ' '.join(self.types)
+            pddl.append(f"  (:types {types})")
+
+        # Add predicates
+        if self.predicates:
+            pddl.append("  (:predicates")
+            for pred in self.predicates:
+                pddl.append(f"    {pred.to_pddl()}")
+            pddl.append("  )")
+
+        pddl.append(")")
+        return '\n'.join(pddl)

src/agents/planner/pddl/planner.py

@@ -0,0 +1,90 @@
+"""
+PDDL-based planner integration for Devika's advanced planning system.
+Based on the approach described in https://arxiv.org/abs/2305.14909
+"""
+from typing import List, Dict, Optional, Tuple
+from dataclasses import dataclass
+
+from .domain import PDDLDomain
+from .problem import PDDLProblem
+from .predicate import PDDLPredicate
+from .action import PDDLAction
+
+@dataclass
+class Plan:
+    """Represents a sequence of actions forming a plan."""
+    steps: List[Dict[str, str]]  # List of {action: params} dicts
+    cost: Optional[float] = None
+    metadata: Dict[str, any] = None
+
+class PDDLPlanner:
+    """PDDL-based planner that integrates with LLM for domain modeling."""
+
+    def __init__(self, llm_client):
+        """Initialize planner with LLM client for domain modeling."""
+        self.llm = llm_client
+
+    async def generate_domain_model(self, task_description: str) -> PDDLDomain:
+        """Generate PDDL domain model from natural language description."""
+        # Prompt LLM to generate domain model
+        prompt = f"""Given the following task description, generate a PDDL domain model
+        with appropriate types, predicates, and actions:
+
+        {task_description}
+
+        Format the response as a Python dict with keys:
+        - name: domain name
+        - types: list of type names
+        - predicates: list of {{"name": pred_name, "parameters": [{"name": param_name, "type": type_name}]}}
+        - actions: list of {{"name": action_name, "parameters": [...], "preconditions": [...], "effects": [...]}}
+        """
+
+        response = await self.llm.generate(prompt)
+        # Parse response and create PDDLDomain
+        domain_spec = eval(response)  # Safe since we control the LLM prompt
+
+        domain = PDDLDomain(name=domain_spec["name"])
+        domain.types = domain_spec["types"]
+
+        for pred_spec in domain_spec["predicates"]:
+            domain.add_predicate(PDDLPredicate(**pred_spec))
+
+        for action_spec in domain_spec["actions"]:
+            domain.add_action(PDDLAction(**action_spec))
+
+        return domain
+
+    async def solve(self, domain: PDDLDomain, problem: PDDLProblem) -> Optional[Plan]:
+        """Generate a plan for the given PDDL domain and problem."""
+        # Convert domain and problem to PDDL
+        domain_pddl = domain.to_pddl()
+        problem_pddl = problem.to_pddl()
+
+        # Use LLM to generate plan
+        prompt = f"""Given the following PDDL domain and problem, generate a valid plan:
+
+        Domain:
+        {domain_pddl}
+
+        Problem:
+        {problem_pddl}
+
+        Format the response as a list of action applications, one per line:
+        (action param1 param2 ...)
+        """
+
+        response = await self.llm.generate(prompt)
+
+        # Parse plan from response
+        if not response.strip():
+            return None
+
+        steps = []
+        for line in response.strip().split('\n'):
+            if not line.strip():
+                continue
+            # Parse (action param1 param2 ...) format
+            parts = line.strip('()').split()
+            steps.append({parts[0]: ' '.join(parts[1:])})
+
+        return Plan(steps=steps)

src/agents/planner/pddl/predicate.py

@@ -0,0 +1,18 @@
+"""
+PDDL Predicate class for defining planning predicates.
+"""
+from typing import List, Dict, Optional
+from dataclasses import dataclass, field
+
+@dataclass
+class PDDLPredicate:
+    """Represents a PDDL predicate with typed parameters."""
+    name: str
+    parameters: List[Dict[str, str]]  # List of {name: type} dicts
+
+    def to_pddl(self) -> str:
+        """Convert the predicate to PDDL string representation."""
+        params = ' '.join(f"?{name} - {type}"
+                         for param in self.parameters
+                         for name, type in param.items())
+        return f"({self.name} {params})"

src/agents/planner/pddl/problem.py

@@ -0,0 +1,47 @@
+"""
+PDDL Problem class for defining planning problems.
+"""
+from typing import List, Dict, Optional
+from dataclasses import dataclass, field
+
+@dataclass
+class PDDLProblem:
+    """Represents a PDDL problem with objects, initial state, and goal state."""
+    name: str
+    domain: str
+    objects: Dict[str, List[str]]  # type -> list of objects
+    init: List[str]  # List of initial state predicates
+    goal: List[str]  # List of goal state predicates
+    metric: Optional[str] = None  # For optimization problems
+
+    def to_pddl(self) -> str:
+        """Convert the problem to PDDL string representation."""
+        pddl = [
+            f"(define (problem {self.name})",
+            f"  (:domain {self.domain})"
+        ]
+
+        # Add objects
+        if self.objects:
+            obj_strs = []
+            for type_name, objs in self.objects.items():
+                obj_list = ' '.join(objs)
+                obj_strs.append(f"{obj_list} - {type_name}")
+            pddl.append(f"  (:objects {' '.join(obj_strs)})")
+
+        # Add initial state
+        init_str = ' '.join(f"({pred})" if not pred.startswith('(') else pred
+                           for pred in self.init)
+        pddl.append(f"  (:init {init_str})")
+
+        # Add goal state
+        goal_str = ' '.join(f"({pred})" if not pred.startswith('(') else pred
+                           for pred in self.goal)
+        pddl.append(f"  (:goal (and {goal_str}))")
+
+        # Add optimization metric if specified
+        if self.metric:
+            pddl.append(f"  (:metric {self.metric})")
+
+        pddl.append(")")
+        return '\n'.join(pddl)

tests/test_pddl/test_action.py

@@ -0,0 +1,38 @@
+"""
+Tests for the PDDL action implementation.
+"""
+import pytest
+from src.agents.planner.pddl import PDDLAction
+
+def test_basic_action():
+    """Test basic PDDLAction creation and PDDL output."""
+    action = PDDLAction(
+        name="move",
+        parameters=[
+            {"from": "location"},
+            {"to": "location"}
+        ],
+        preconditions=["at ?from", "connected ?from ?to"],
+        effects=["not (at ?from)", "at ?to"]
+    )
+    pddl = action.to_pddl()
+    assert ":action move" in pddl
+    assert ":parameters (?from - location ?to - location)" in pddl
+    assert ":precondition (and (at ?from) (connected ?from ?to))" in pddl
+    assert ":effect (and (not (at ?from)) (at ?to))" in pddl
+
+def test_temporal_action():
+    """Test PDDLAction with duration for temporal planning."""
+    action = PDDLAction(
+        name="drive",
+        parameters=[
+            {"v": "vehicle"},
+            {"from": "location"},
+            {"to": "location"}
+        ],
+        preconditions=["at ?v ?from"],
+        effects=["not (at ?v ?from)", "at ?v ?to"],
+        duration=5.0
+    )
+    pddl = action.to_pddl()
+    assert ":duration 5.0" in pddl

tests/test_pddl/test_domain.py

@@ -0,0 +1,45 @@
+"""
+Tests for the PDDL domain implementation.
+"""
+import pytest
+from src.agents.planner.pddl import PDDLDomain, PDDLPredicate
+
+def test_pddl_domain_creation():
+    """Test basic PDDLDomain creation."""
+    domain = PDDLDomain(name="navigation")
+    assert domain.name == "navigation"
+    assert domain.requirements == ['strips', 'typing']
+    assert domain.types == []
+    assert domain.predicates == []
+
+def test_pddl_domain_with_predicates():
+    """Test PDDLDomain with predicates."""
+    domain = PDDLDomain(name="navigation")
+    pred1 = PDDLPredicate(
+        name="at",
+        parameters=[{"loc": "location"}]
+    )
+    pred2 = PDDLPredicate(
+        name="connected",
+        parameters=[
+            {"from": "location"},
+            {"to": "location"}
+        ]
+    )
+    domain.add_predicate(pred1)
+    domain.add_predicate(pred2)
+
+    pddl = domain.to_pddl()
+    assert "(define (domain navigation)" in pddl
+    assert ":requirements :strips :typing" in pddl
+    assert "(at ?loc - location)" in pddl
+    assert "(connected ?from - location ?to - location)" in pddl
+
+def test_pddl_domain_with_types():
+    """Test PDDLDomain with custom types."""
+    domain = PDDLDomain(
+        name="robot-world",
+        types=["location", "robot", "object"]
+    )
+    pddl = domain.to_pddl()
+    assert "(:types location robot object)" in pddl