|
1
|
"""Skill interface for the PlanOpticon planning agent.""" |
|
2
|
|
|
3
|
from abc import ABC, abstractmethod |
|
4
|
from dataclasses import dataclass, field |
|
5
|
from typing import Any, Dict, List, Optional |
|
6
|
|
|
7
|
|
|
8
|
@dataclass |
|
9
|
class Artifact: |
|
10
|
"""Output from a skill execution.""" |
|
11
|
|
|
12
|
name: str |
|
13
|
content: str # The generated content (markdown, json, etc.) |
|
14
|
artifact_type: str # "project_plan", "prd", "roadmap", "task_list", "document", "issues" |
|
15
|
format: str = "markdown" # "markdown", "json", "mermaid" |
|
16
|
metadata: Dict[str, Any] = field(default_factory=dict) |
|
17
|
|
|
18
|
|
|
19
|
@dataclass |
|
20
|
class AgentContext: |
|
21
|
"""Shared context for agent skills.""" |
|
22
|
|
|
23
|
knowledge_graph: Any = None # KnowledgeGraph instance |
|
24
|
query_engine: Any = None # GraphQueryEngine instance |
|
25
|
provider_manager: Any = None # ProviderManager instance |
|
26
|
planning_entities: List[Any] = field(default_factory=list) |
|
27
|
user_requirements: Dict[str, Any] = field(default_factory=dict) |
|
28
|
conversation_history: List[Dict[str, str]] = field(default_factory=list) |
|
29
|
artifacts: List[Artifact] = field(default_factory=list) |
|
30
|
config: Dict[str, Any] = field(default_factory=dict) |
|
31
|
|
|
32
|
|
|
33
|
class Skill(ABC): |
|
34
|
"""Base class for agent skills.""" |
|
35
|
|
|
36
|
name: str = "" |
|
37
|
description: str = "" |
|
38
|
|
|
39
|
@abstractmethod |
|
40
|
def execute(self, context: AgentContext, **kwargs) -> Artifact: |
|
41
|
"""Execute this skill and return an artifact.""" |
|
42
|
... |
|
43
|
|
|
44
|
def can_execute(self, context: AgentContext) -> bool: |
|
45
|
"""Check if this skill can execute given the current context.""" |
|
46
|
return context.knowledge_graph is not None and context.provider_manager is not None |
|
47
|
|
|
48
|
|
|
49
|
# Skill registry |
|
50
|
_skills: Dict[str, "Skill"] = {} |
|
51
|
|
|
52
|
|
|
53
|
def register_skill(skill: "Skill") -> None: |
|
54
|
"""Register a skill instance in the global registry.""" |
|
55
|
_skills[skill.name] = skill |
|
56
|
|
|
57
|
|
|
58
|
def get_skill(name: str) -> Optional["Skill"]: |
|
59
|
"""Look up a skill by name.""" |
|
60
|
return _skills.get(name) |
|
61
|
|
|
62
|
|
|
63
|
def list_skills() -> List["Skill"]: |
|
64
|
"""Return all registered skills.""" |
|
65
|
return list(_skills.values()) |
|
66
|
|