aishastr.live

Agentic Architecture Pattern: Service Oriented

Written by

admin

in


    """
Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces.
Services can be discovered, composed, and orchestrated.
"""

from typing import Dict, Any, List, Optional, Callable
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 ServiceStatus(Enum):
    """Service status."""
    AVAILABLE = "available"
    BUSY = "busy"
    UNAVAILABLE = "unavailable"
    MAINTENANCE = "maintenance"


@dataclass
class ServiceContract:
    """Service contract defining interface."""
    service_name: str
    version: str
    description: str
    input_schema: Dict[str, str]
    output_schema: Dict[str, str]
    sla: Dict[str, Any] = field(default_factory=dict)  # Service Level Agreement


@dataclass
class ServiceCall:
    """Track service invocations."""
    service_name: str
    caller: str
    input_data: Any
    output_data: Any
    timestamp: datetime = field(default_factory=datetime.now)
    execution_time: float = 0.0
    success: bool = True


class Service:
    """Service with defined contract."""
    
    def __init__(self, contract: ServiceContract, implementation: Callable):
        self.contract = contract
        self.implementation = implementation
        self.status = ServiceStatus.AVAILABLE
        self.call_count = 0
        self.total_execution_time = 0.0
    
    def invoke(self, input_data: Dict[str, Any], caller: str = "unknown") -> Dict[str, Any]:
        """Invoke the service."""
        if self.status != ServiceStatus.AVAILABLE:
            return {
                "success": False,
                "error": f"Service {self.contract.service_name} is {self.status.value}"
            }
        
        try:
            start = datetime.now()
            result = self.implementation(input_data)
            execution_time = (datetime.now() - start).total_seconds()
            
            self.call_count += 1
            self.total_execution_time += execution_time
            
            return {
                "success": True,
                "service": self.contract.service_name,
                "version": self.contract.version,
                "result": result,
                "execution_time": execution_time
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def get_metrics(self) -> Dict[str, Any]:
        """Get service metrics."""
        avg_time = self.total_execution_time / self.call_count if self.call_count > 0 else 0
        return {
            "service": self.contract.service_name,
            "status": self.status.value,
            "call_count": self.call_count,
            "average_execution_time": f"{avg_time:.3f}s"
        }


class ServiceRegistry:
    """Registry for service discovery."""
    
    def __init__(self):
        self.services: Dict[str, Service] = {}
    
    def register(self, service: Service):
        """Register a service."""
        self.services[service.contract.service_name] = service
        print(f"Service registered: {service.contract.service_name} v{service.contract.version}")
    
    def unregister(self, service_name: str):
        """Unregister a service."""
        if service_name in self.services:
            del self.services[service_name]
            print(f"Service unregistered: {service_name}")
    
    def discover(self, query: str = "") -> List[str]:
        """Discover available services."""
        if not query:
            return list(self.services.keys())
        return [
            name for name in self.services.keys()
            if query.lower() in name.lower()
        ]
    
    def get_service(self, service_name: str) -> Optional[Service]:
        """Get a service by name."""
        return self.services.get(service_name)


class ServiceOrientedPattern:
    """
    Service-Oriented Pattern - SOA for agents.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.registry = ServiceRegistry()
        self.call_history: List[ServiceCall] = []
    
    def create_llm_service(self, service_name: str, prompt_template: str) -> Service:
        """Create an LLM-backed service."""
        
        def implementation(input_data: Dict[str, Any]) -> str:
            prompt = ChatPromptTemplate.from_messages([
                ("system", prompt_template),
                ("human", "{input}")
            ])
            chain = prompt | self.llm
            response = chain.invoke({"input": str(input_data)})
            return str(response.content)
        
        contract = ServiceContract(
            service_name=service_name,
            version="1.0",
            description=f"LLM service: {service_name}",
            input_schema={"input": "any"},
            output_schema={"output": "string"}
        )
        
        return Service(contract, implementation)
    
    def call_service(self, service_name: str, input_data: Dict[str, Any], 
                    caller: str = "client") -> Dict[str, Any]:
        """Call a service."""
        service = self.registry.get_service(service_name)
        
        if not service:
            return {
                "success": False,
                "error": f"Service {service_name} not found"
            }
        
        result = service.invoke(input_data, caller)
        
        # Track call
        call = ServiceCall(
            service_name=service_name,
            caller=caller,
            input_data=input_data,
            output_data=result.get("result"),
            execution_time=result.get("execution_time", 0.0),
            success=result.get("success", False)
        )
        self.call_history.append(call)
        
        return result
    
    def compose_services(self, pipeline: List[tuple], input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Compose multiple services in a pipeline."""
        results = []
        current_input = input_data
        
        for service_name, output_key in pipeline:
            result = self.call_service(service_name, current_input)
            
            if not result.get("success"):
                return {
                    "success": False,
                    "error": f"Service {service_name} failed: {result.get('error')}"
                }
            
            results.append({
                "service": service_name,
                "output": result.get("result")
            })
            
            # Pass output as input to next service
            current_input = {output_key: result.get("result")}
        
        return {
            "success": True,
            "pipeline": [s for s, _ in pipeline],
            "results": results
        }
    
    def run(self, service_name: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Execute service call."""
        result = self.call_service(service_name, input_data)
        
        return {
            "service": service_name,
            "input": input_data,
            "output": result,
            "available_services": self.registry.discover()
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get system statistics."""
        return {
            "total_services": len(self.registry.services),
            "total_calls": len(self.call_history),
            "service_metrics": [
                service.get_metrics()
                for service in self.registry.services.values()
            ]
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = ServiceOrientedPattern()
    
    # Create and register services
    text_analyzer = pattern.create_llm_service(
        "text_analyzer",
        "Analyze the text and provide insights about sentiment, tone, and key topics."
    )
    pattern.registry.register(text_analyzer)
    
    summarizer = pattern.create_llm_service(
        "summarizer",
        "Summarize the following text in 2-3 sentences."
    )
    pattern.registry.register(summarizer)
    
    translator = pattern.create_llm_service(
        "translator",
        "Translate the text to Spanish."
    )
    pattern.registry.register(translator)
    
    print("=== Service-Oriented Pattern Demo ===\n")
    
    # Service discovery
    print("Available services:", pattern.registry.discover())
    print()
    
    # Call individual service
    print("1. Text Analysis Service:")
    result1 = pattern.run(
        "text_analyzer",
        {"text": "This product is amazing! I love using it every day."}
    )
    print(f"   Success: {result1['output']['success']}")
    print(f"   Result: {result1['output']['result'][:150]}...\n")
    
    # Call another service
    print("2. Summarization Service:")
    result2 = pattern.run(
        "summarizer",
        {"text": "Artificial intelligence is transforming industries worldwide. From healthcare to finance, AI systems are helping humans make better decisions. Machine learning algorithms can process vast amounts of data quickly."}
    )
    print(f"   Success: {result2['output']['success']}")
    print(f"   Result: {result2['output']['result']}\n")
    
    # Service composition
    print("3. Service Composition (Analyze → Summarize):")
    composition_result = pattern.compose_services(
        pipeline=[
            ("text_analyzer", "analysis"),
            ("summarizer", "summary")
        ],
        input_data={"text": "The future of AI is bright with many opportunities."}
    )
    print(f"   Success: {composition_result['success']}")
    print(f"   Pipeline: {' → '.join(composition_result['pipeline'])}")
    for step in composition_result['results']:
        print(f"   {step['service']}: {step['output'][:100]}...")
    
    # Show statistics
    print("\n\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Total services: {stats['total_services']}")
    print(f"Total calls: {stats['total_calls']}")
    print("\nService Metrics:")
    for metrics in stats['service_metrics']:
        print(f"  {metrics['service']}: {metrics['call_count']} calls, avg time: {metrics['average_execution_time']}")

Agentic Architecture Pattern: Service Oriented

Pattern 100: 100 Service Oriented – Line-by-Line Explanation

Overview

Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces. Services can be discovered, composed, and orchestrated.

Code Structure Analysis

File Information

  • Pattern Number: 100
  • Pattern Name: 100 Service Oriented
  • File Path: patterns\tool_resource\100_service_oriented.py
  • Total Lines: 307

Detailed Code Explanation

Module Docstring

"""
Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces.
Services can be discovered, composed, and orchestrated.
"""

Explanation: This module docstring describes the purpose and key features of the 100 Service Oriented pattern.

Imports Section

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

ServiceStatus (Line 16)

class ServiceStatus(Enum):
    AVAILABLE
    BUSY
    UNAVAILABLE
    MAINTENANCE

Purpose: Service status.

Values: – AVAILABLE: Represents available – BUSY: Represents busy – UNAVAILABLE: Represents unavailable – MAINTENANCE: Represents maintenance

Classes

ServiceStatus (Line 16)

Description: Service status.

ServiceContract (Line 25)

Description: Service contract defining interface.

ServiceCall (Line 36)

Description: Track service invocations.

Service (Line 47)

Description: Service with defined contract.

Methods:

__init__ (Line 50)
def __init__(self, contract, implementation):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – contract: Parameter for contract – implementation: Parameter for implementation

invoke (Line 57)
def invoke(self, input_data, caller):
    """Invoke the service."""

Purpose: Invoke the service.

Parameters: – self: Instance reference – input_data: Parameter for input_data – caller: Parameter for caller

get_metrics (Line 86)
def get_metrics(self):
    """Get service metrics."""

Purpose: Get service metrics.

Parameters: – self: Instance reference

ServiceRegistry (Line 97)

Description: Registry for service discovery.

Methods:

__init__ (Line 100)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register (Line 103)
def register(self, service):
    """Register a service."""

Purpose: Register a service.

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

unregister (Line 108)
def unregister(self, service_name):
    """Unregister a service."""

Purpose: Unregister a service.

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

discover (Line 114)
def discover(self, query):
    """Discover available services."""

Purpose: Discover available services.

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

get_service (Line 123)
def get_service(self, service_name):
    """Get a service by name."""

Purpose: Get a service by name.

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

ServiceOrientedPattern (Line 128)

Description: Service-Oriented Pattern – SOA for agents.

Methods:

__init__ (Line 133)
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”)

create_llm_service (Line 138)
def create_llm_service(self, service_name, prompt_template):
    """Create an LLM-backed service."""

Purpose: Create an LLM-backed service.

Parameters: – self: Instance reference – service_name: Parameter for service_name – prompt_template: Parameter for prompt_template

call_service (Line 160)
def call_service(self, service_name, input_data, caller):
    """Call a service."""

Purpose: Call a service.

Parameters: – self: Instance reference – service_name: Parameter for service_name – input_data: Parameter for input_data – caller: Parameter for caller

compose_services (Line 186)
def compose_services(self, pipeline, input_data):
    """Compose multiple services in a pipeline."""

Purpose: Compose multiple services in a pipeline.

Parameters: – self: Instance reference – pipeline: Parameter for pipeline – input_data: Parameter for input_data

run (Line 214)
def run(self, service_name, input_data):
    """Execute service call."""

Purpose: Execute service call.

Parameters: – self: Instance reference – service_name: Parameter for service_name – input_data: Parameter for input_data

get_statistics (Line 225)
def get_statistics(self):
    """Get system statistics."""

Purpose: Get system 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 = ServiceStatus(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