"""
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