"""
Pattern 84: Service Agent Pattern
Purpose: Agent that provides services to other agents
"""
from typing import Dict, Any, List, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
class ServiceType(Enum):
"""Types of services."""
COMPUTATION = "computation"
INFORMATION = "information"
TRANSLATION = "translation"
VALIDATION = "validation"
TRANSFORMATION = "transformation"
@dataclass
class ServiceRequest:
"""A request for service."""
id: str
service_type: ServiceType
payload: Any
requester: str
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class ServiceResponse:
"""A response from service."""
request_id: str
result: Any
success: bool
error: str = None
completed_at: datetime = field(default_factory=datetime.now)
class ServiceAgent:
"""
Service Agent - Provides services to other agents
Offers specialized services that other agents can consume.
Maintains service catalog and handles requests.
"""
def __init__(self, name: str, services: List[ServiceType], model_name: str = "gpt-4"):
self.name = name
self.services = services
self.llm = ChatOpenAI(model=model_name, temperature=0.5)
self.request_history: List[ServiceRequest] = []
self.response_history: List[ServiceResponse] = []
def provides_service(self, service_type: ServiceType) -> bool:
"""Check if agent provides this service."""
return service_type in self.services
def process_request(self, request: ServiceRequest) -> ServiceResponse:
"""Process a service request."""
self.request_history.append(request)
if not self.provides_service(request.service_type):
response = ServiceResponse(
request_id=request.id,
result=None,
success=False,
error=f"Service {request.service_type.value} not provided"
)
self.response_history.append(response)
return response
try:
# Process using LLM
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, providing {request.service_type.value} service"),
("human", "Process this request: {payload}")
])
chain = prompt | self.llm
result = chain.invoke({"payload": str(request.payload)})
response = ServiceResponse(
request_id=request.id,
result=str(result.content),
success=True
)
except Exception as e:
response = ServiceResponse(
request_id=request.id,
result=None,
success=False,
error=str(e)
)
self.response_history.append(response)
return response
def get_service_catalog(self) -> Dict[str, Any]:
"""Get catalog of provided services."""
return {
"agent": self.name,
"services": [s.value for s in self.services],
"requests_handled": len(self.request_history),
"success_rate": sum(1 for r in self.response_history if r.success) / len(self.response_history) if self.response_history else 0
}
class ServiceAgentPattern:
"""
Service Agent Pattern - Manages service agents and routing
"""
def __init__(self):
self.agents: Dict[str, ServiceAgent] = {}
self.service_registry: Dict[ServiceType, List[str]] = {}
def register_agent(self, agent: ServiceAgent):
"""Register a service agent."""
self.agents[agent.name] = agent
# Update service registry
for service in agent.services:
if service not in self.service_registry:
self.service_registry[service] = []
self.service_registry[service].append(agent.name)
print(f"✓ Registered: {agent.name} - Services: {[s.value for s in agent.services]}")
def find_provider(self, service_type: ServiceType) -> ServiceAgent:
"""Find agent that provides service."""
if service_type in self.service_registry:
agent_name = self.service_registry[service_type][0]
return self.agents[agent_name]
return None
def request_service(self, service_type: ServiceType, payload: Any, requester: str = "client") -> ServiceResponse:
"""Request a service."""
provider = self.find_provider(service_type)
if not provider:
return ServiceResponse(
request_id="no_provider",
result=None,
success=False,
error=f"No provider for {service_type.value}"
)
request = ServiceRequest(
id=f"req_{len(provider.request_history) + 1}",
service_type=service_type,
payload=payload,
requester=requester
)
return provider.process_request(request)
def run(self, requests: List[tuple]) -> Dict[str, Any]:
"""Process multiple service requests."""
results = []
for service_type, payload in requests:
response = self.request_service(service_type, payload)
results.append({
"service": service_type.value,
"success": response.success,
"result": response.result[:100] if response.result else None,
"error": response.error
})
return {
"pattern": "ServiceAgent",
"total_requests": len(requests),
"successful": sum(1 for r in results if r['success']),
"results": results
}
if __name__ == "__main__":
print("=== Service Agent Pattern Demo ===\n")
pattern = ServiceAgentPattern()
# Register service providers
pattern.register_agent(ServiceAgent(
"ComputeService",
[ServiceType.COMPUTATION, ServiceType.VALIDATION]
))
pattern.register_agent(ServiceAgent(
"DataService",
[ServiceType.INFORMATION, ServiceType.TRANSFORMATION]
))
pattern.register_agent(ServiceAgent(
"LinguisticsService",
[ServiceType.TRANSLATION]
))
# Make service requests
print("\nRequesting services...\n")
requests = [
(ServiceType.COMPUTATION, "Calculate fibonacci(10)"),
(ServiceType.TRANSLATION, "Translate 'hello' to Spanish"),
(ServiceType.INFORMATION, "Get user data for ID 12345"),
(ServiceType.VALIDATION, "Validate email: user@example.com")
]
summary = pattern.run(requests)
print(f"\nPattern: {summary['pattern']}")
print(f"Total Requests: {summary['total_requests']}")
print(f"Successful: {summary['successful']}/{summary['total_requests']}")
print("\nService Results:")
for result in summary['results']:
status = "✓" if result['success'] else "✗"
print(f" {status} {result['service']}: {result['result'] or result['error']}")
Agentic Architecture Pattern: Service Agent
Pattern 84: 84 Service Agent – Line-by-Line Explanation
Overview
Pattern 84: Service Agent Pattern Purpose: Agent that provides services to other agents
Code Structure Analysis
File Information
- Pattern Number: 84
- Pattern Name: 84 Service Agent
- File Path:
patterns\specialized_agent\84_service_agent.py - Total Lines: 220
Detailed Code Explanation
Module Docstring
"""
Pattern 84: Service Agent Pattern
Purpose: Agent that provides services to other agents
"""
Explanation: This module docstring describes the purpose and key features of the 84 Service Agent pattern.
Imports Section
from typing import Dict, Any, List, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, 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_openai import ChatOpenAI: LangChain framework integration
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
Enumerations
ServiceType (Line 14)
class ServiceType(Enum):
COMPUTATION
INFORMATION
TRANSLATION
VALIDATION
TRANSFORMATION
Purpose: Types of services.
Values:
– COMPUTATION: Represents computation
– INFORMATION: Represents information
– TRANSLATION: Represents translation
– VALIDATION: Represents validation
– TRANSFORMATION: Represents transformation
Classes
ServiceType (Line 14)
Description: Types of services.
ServiceRequest (Line 24)
Description: A request for service.
ServiceResponse (Line 34)
Description: A response from service.
ServiceAgent (Line 43)
Description: Service Agent – Provides services to other agents
Offers specialized services that other agents can consume. Maintains service catalog and handles requests.
Methods:
__init__ (Line 51)
def __init__(self, name, services, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– services: Parameter for services
– model_name: LLM model to use (e.g., “gpt-4”)
provides_service (Line 58)
def provides_service(self, service_type):
"""Check if agent provides this service."""
Purpose: Check if agent provides this service.
Parameters:
– self: Instance reference
– service_type: Parameter for service_type
process_request (Line 62)
def process_request(self, request):
"""Process a service request."""
Purpose: Process a service request.
Parameters:
– self: Instance reference
– request: Parameter for request
get_service_catalog (Line 102)
def get_service_catalog(self):
"""Get catalog of provided services."""
Purpose: Get catalog of provided services.
Parameters:
– self: Instance reference
ServiceAgentPattern (Line 112)
Description: Service Agent Pattern – Manages service agents and routing
Methods:
__init__ (Line 117)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
register_agent (Line 121)
def register_agent(self, agent):
"""Register a service agent."""
Purpose: Register a service agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
find_provider (Line 133)
def find_provider(self, service_type):
"""Find agent that provides service."""
Purpose: Find agent that provides service.
Parameters:
– self: Instance reference
– service_type: Parameter for service_type
request_service (Line 140)
def request_service(self, service_type, payload, requester):
"""Request a service."""
Purpose: Request a service.
Parameters:
– self: Instance reference
– service_type: Parameter for service_type
– payload: Parameter for payload
– requester: Parameter for requester
run (Line 161)
def run(self, requests):
"""Process multiple service requests."""
Purpose: Process multiple service requests.
Parameters:
– self: Instance reference
– requests: Parameter for requests
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 = ServiceType(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