aishastr.live

Agentic Architecture Pattern: Capability Based

Written by

admin

in


    """
Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents
based on their declared capabilities, enabling flexible task distribution.
"""

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


class CapabilityLevel(Enum):
    """Proficiency level for capabilities."""
    BASIC = 1
    INTERMEDIATE = 2
    ADVANCED = 3
    EXPERT = 4


@dataclass
class Capability:
    """Agent capability definition."""
    name: str
    description: str
    level: CapabilityLevel
    parameters: Dict[str, Any] = field(default_factory=dict)
    
    def __hash__(self):
        return hash(self.name)
    
    def __eq__(self, other):
        return self.name == other.name


@dataclass
class Task:
    """Task requiring specific capabilities."""
    id: str
    description: str
    required_capabilities: Set[str]
    priority: int = 1
    assigned_to: Optional[str] = None


class CapableAgent:
    """Agent with declared capabilities."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.capabilities: Dict[str, Capability] = {}
        self.tasks_completed: int = 0
    
    def declare_capability(self, capability: Capability):
        """Declare a capability."""
        self.capabilities[capability.name] = capability
        print(f"{self.name} declared capability: {capability.name} ({capability.level.name})")
    
    def has_capability(self, capability_name: str) -> bool:
        """Check if agent has a capability."""
        return capability_name in self.capabilities
    
    def has_all_capabilities(self, capability_names: Set[str]) -> bool:
        """Check if agent has all required capabilities."""
        return all(self.has_capability(cap) for cap in capability_names)
    
    def get_capability_level(self, capability_name: str) -> Optional[CapabilityLevel]:
        """Get capability level."""
        if capability_name in self.capabilities:
            return self.capabilities[capability_name].level
        return None
    
    def execute_task(self, task: Task) -> str:
        """Execute a task using capabilities."""
        if not self.has_all_capabilities(task.required_capabilities):
            return f"Cannot execute - missing capabilities: {task.required_capabilities - set(self.capabilities.keys())}"
        
        # Use LLM to execute
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} with capabilities: {', '.join(self.capabilities.keys())}"),
            ("human", "Execute this task: {task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task.description})
        self.tasks_completed += 1
        
        return str(response.content)


class CapabilityBasedPattern:
    """
    Capability-Based Pattern - Match tasks to agents by capabilities.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, CapableAgent] = {}
        self.tasks: List[Task] = []
        self.assignments: List[Dict[str, Any]] = []
    
    def register_agent(self, agent: CapableAgent):
        """Register an agent."""
        self.agents[agent.name] = agent
        print(f"Agent registered: {agent.name}")
    
    def find_capable_agents(self, required_capabilities: Set[str]) -> List[CapableAgent]:
        """Find all agents with required capabilities."""
        return [
            agent for agent in self.agents.values()
            if agent.has_all_capabilities(required_capabilities)
        ]
    
    def find_best_agent(self, task: Task) -> Optional[CapableAgent]:
        """Find best agent for task based on capability levels."""
        capable_agents = self.find_capable_agents(task.required_capabilities)
        
        if not capable_agents:
            return None
        
        # Score agents by capability levels
        def score_agent(agent: CapableAgent) -> float:
            return sum(
                agent.get_capability_level(cap).value 
                for cap in task.required_capabilities
            )
        
        return max(capable_agents, key=score_agent)
    
    def assign_task(self, task: Task) -> Dict[str, Any]:
        """Assign task to best agent."""
        agent = self.find_best_agent(task)
        
        if not agent:
            return {
                "success": False,
                "task_id": task.id,
                "message": f"No agent found with capabilities: {task.required_capabilities}"
            }
        
        task.assigned_to = agent.name
        self.tasks.append(task)
        
        # Execute task
        result = agent.execute_task(task)
        
        assignment = {
            "task_id": task.id,
            "agent": agent.name,
            "capabilities_used": list(task.required_capabilities),
            "result": result,
            "timestamp": datetime.now()
        }
        self.assignments.append(assignment)
        
        return {
            "success": True,
            "task_id": task.id,
            "assigned_to": agent.name,
            "result": result
        }
    
    def get_capability_matrix(self) -> Dict[str, List[str]]:
        """Get matrix of agents and their capabilities."""
        return {
            agent_name: list(agent.capabilities.keys())
            for agent_name, agent in self.agents.items()
        }
    
    def run(self, task_description: str, 
            required_capabilities: Set[str]) -> Dict[str, Any]:
        """Execute capability-based task assignment."""
        task = Task(
            id=f"task_{len(self.tasks) + 1}",
            description=task_description,
            required_capabilities=required_capabilities
        )
        
        result = self.assign_task(task)
        
        return {
            "task": task_description,
            "required_capabilities": list(required_capabilities),
            "assignment": result,
            "capability_matrix": self.get_capability_matrix()
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_agents": len(self.agents),
            "total_tasks": len(self.tasks),
            "successful_assignments": sum(1 for a in self.assignments if a),
            "agent_performance": {
                name: agent.tasks_completed
                for name, agent in self.agents.items()
            }
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = CapabilityBasedPattern()
    
    # Create agents with capabilities
    agent1 = CapableAgent("DataScientist")
    agent1.declare_capability(Capability("data_analysis", "Analyze datasets", CapabilityLevel.EXPERT))
    agent1.declare_capability(Capability("visualization", "Create visualizations", CapabilityLevel.ADVANCED))
    agent1.declare_capability(Capability("python", "Python programming", CapabilityLevel.EXPERT))
    
    agent2 = CapableAgent("WebDeveloper")
    agent2.declare_capability(Capability("web_development", "Build web applications", CapabilityLevel.EXPERT))
    agent2.declare_capability(Capability("javascript", "JavaScript programming", CapabilityLevel.ADVANCED))
    agent2.declare_capability(Capability("api_design", "Design REST APIs", CapabilityLevel.INTERMEDIATE))
    
    agent3 = CapableAgent("DevOps")
    agent3.declare_capability(Capability("deployment", "Deploy applications", CapabilityLevel.EXPERT))
    agent3.declare_capability(Capability("monitoring", "Monitor systems", CapabilityLevel.ADVANCED))
    agent3.declare_capability(Capability("python", "Python programming", CapabilityLevel.INTERMEDIATE))
    
    # Register agents
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    pattern.register_agent(agent3)
    
    print("\n=== Capability-Based Pattern Demo ===\n")
    
    # Task 1: Data analysis
    print("1. Task: Analyze customer data")
    result1 = pattern.run(
        "Analyze customer purchase patterns and create visualizations",
        {"data_analysis", "visualization"}
    )
    print(f"   Assigned to: {result1['assignment']['assigned_to']}")
    print(f"   Result: {result1['assignment']['result'][:100]}...\n")
    
    # Task 2: Web development
    print("2. Task: Build dashboard API")
    result2 = pattern.run(
        "Create REST API for dashboard application",
        {"web_development", "api_design"}
    )
    print(f"   Assigned to: {result2['assignment']['assigned_to']}")
    print(f"   Result: {result2['assignment']['result'][:100]}...\n")
    
    # Task 3: Multi-capability
    print("3. Task: Deploy monitoring solution")
    result3 = pattern.run(
        "Deploy and monitor the application",
        {"deployment", "monitoring"}
    )
    print(f"   Assigned to: {result3['assignment']['assigned_to']}")
    print(f"   Result: {result3['assignment']['result'][:100]}...\n")
    
    # Show capability matrix
    print("\n=== Capability Matrix ===")
    for agent_name, capabilities in pattern.get_capability_matrix().items():
        print(f"{agent_name}: {', '.join(capabilities)}")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Total agents: {stats['total_agents']}")
    print(f"Total tasks: {stats['total_tasks']}")
    print(f"Agent performance: {stats['agent_performance']}")

Agentic Architecture Pattern: Capability Based

Pattern 99: 99 Capability Based – Line-by-Line Explanation

Overview

Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents based on their declared capabilities, enabling flexible task distribution.

Code Structure Analysis

File Information

  • Pattern Number: 99
  • Pattern Name: 99 Capability Based
  • File Path: patterns\tool_resource\99_capability_based.py
  • Total Lines: 271

Detailed Code Explanation

Module Docstring

"""
Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents
based on their declared capabilities, enabling flexible task distribution.
"""

Explanation: This module docstring describes the purpose and key features of the 99 Capability Based pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
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 datetime import datetime: Date and time utilities – 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

CapabilityLevel (Line 16)

class CapabilityLevel(Enum):
    BASIC
    INTERMEDIATE
    ADVANCED
    EXPERT

Purpose: Proficiency level for capabilities.

Values: – BASIC: Represents basic – INTERMEDIATE: Represents intermediate – ADVANCED: Represents advanced – EXPERT: Represents expert

Classes

CapabilityLevel (Line 16)

Description: Proficiency level for capabilities.

Capability (Line 25)

Description: Agent capability definition.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

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

Purpose: No method docstring.

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

Task (Line 40)

Description: Task requiring specific capabilities.

CapableAgent (Line 49)

Description: Agent with declared capabilities.

Methods:

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

Purpose: No method docstring.

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

declare_capability (Line 58)
def declare_capability(self, capability):
    """Declare a capability."""

Purpose: Declare a capability.

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

has_capability (Line 63)
def has_capability(self, capability_name):
    """Check if agent has a capability."""

Purpose: Check if agent has a capability.

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

has_all_capabilities (Line 67)
def has_all_capabilities(self, capability_names):
    """Check if agent has all required capabilities."""

Purpose: Check if agent has all required capabilities.

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

get_capability_level (Line 71)
def get_capability_level(self, capability_name):
    """Get capability level."""

Purpose: Get capability level.

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

execute_task (Line 77)
def execute_task(self, task):
    """Execute a task using capabilities."""

Purpose: Execute a task using capabilities.

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

CapabilityBasedPattern (Line 95)

Description: Capability-Based Pattern – Match tasks to agents by capabilities.

Methods:

__init__ (Line 100)
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_agent (Line 106)
def register_agent(self, agent):
    """Register an agent."""

Purpose: Register an agent.

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

find_capable_agents (Line 111)
def find_capable_agents(self, required_capabilities):
    """Find all agents with required capabilities."""

Purpose: Find all agents with required capabilities.

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

find_best_agent (Line 118)
def find_best_agent(self, task):
    """Find best agent for task based on capability levels."""

Purpose: Find best agent for task based on capability levels.

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

assign_task (Line 134)
def assign_task(self, task):
    """Assign task to best agent."""

Purpose: Assign task to best agent.

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

get_capability_matrix (Line 167)
def get_capability_matrix(self):
    """Get matrix of agents and their capabilities."""

Purpose: Get matrix of agents and their capabilities.

Parameters: – self: Instance reference

run (Line 174)
def run(self, task_description, required_capabilities):
    """Execute capability-based task assignment."""

Purpose: Execute capability-based task assignment.

Parameters: – self: Instance reference – task_description: Parameter for task_description – required_capabilities: Parameter for required_capabilities

get_statistics (Line 192)
def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern 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 = CapabilityLevel(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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts