"""
Pattern 283: Agent Registry Pattern
Purpose: Central registry for agent registration, lookup, and lifecycle management
"""
from typing import Dict, Any, List, Optional, Set, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import defaultdict
class RegistrationStatus(Enum):
"""Agent registration status."""
ACTIVE = "active"
INACTIVE = "inactive"
SUSPENDED = "suspended"
DEREGISTERED = "deregistered"
class AgentType(Enum):
"""Type of registered agent."""
SERVICE = "service"
WORKER = "worker"
COORDINATOR = "coordinator"
MONITOR = "monitor"
@dataclass
class AgentMetadata:
"""Metadata for registered agent."""
agent_id: str
name: str
agent_type: AgentType
capabilities: Set[str]
endpoint: str # Network location or identifier
version: str
status: RegistrationStatus = RegistrationStatus.ACTIVE
registered_at: datetime = field(default_factory=datetime.now)
last_heartbeat: Optional[datetime] = None
tags: Set[str] = field(default_factory=set)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class RegistryEvent:
"""Event in the registry."""
event_type: str # registered, deregistered, updated, heartbeat
agent_id: str
timestamp: datetime = field(default_factory=datetime.now)
details: Dict[str, Any] = field(default_factory=dict)
class AgentRegistry:
"""
Implements agent registry pattern.
Features:
- Centralized agent registration
- Agent lookup by ID, capability, type, tags
- Heartbeat monitoring
- Automatic deregistration on timeout
- Event tracking
- Query capabilities
"""
def __init__(self, heartbeat_timeout_seconds: int = 60):
self.agents: Dict[str, AgentMetadata] = {}
self.capability_index: Dict[str, Set[str]] = defaultdict(set)
self.type_index: Dict[AgentType, Set[str]] = defaultdict(set)
self.tag_index: Dict[str, Set[str]] = defaultdict(set)
self.events: List[RegistryEvent] = []
self.heartbeat_timeout = timedelta(seconds=heartbeat_timeout_seconds)
# Event listeners
self.listeners: List[Callable[[RegistryEvent], None]] = []
# Statistics
self.statistics = {
"total_registered": 0,
"total_deregistered": 0,
"active_agents": 0,
"suspended_agents": 0,
"total_heartbeats": 0,
"timeouts": 0
}
def register(
self,
agent_id: str,
name: str,
agent_type: AgentType,
capabilities: Set[str],
endpoint: str,
version: str = "1.0.0",
tags: Set[str] = None,
metadata: Dict[str, Any] = None
) -> Dict[str, Any]:
"""
Register an agent.
Args:
agent_id: Unique agent identifier
name: Agent name
agent_type: Type of agent
capabilities: Set of capabilities
endpoint: Network endpoint
version: Agent version
tags: Optional tags
metadata: Additional metadata
Returns:
Registration status
"""
if agent_id in self.agents:
existing = self.agents[agent_id]
if existing.status != RegistrationStatus.DEREGISTERED:
return {
"success": False,
"message": f"Agent '{agent_id}' already registered"
}
# Create agent metadata
agent = AgentMetadata(
agent_id=agent_id,
name=name,
agent_type=agent_type,
capabilities=capabilities,
endpoint=endpoint,
version=version,
tags=tags or set(),
metadata=metadata or {},
last_heartbeat=datetime.now()
)
self.agents[agent_id] = agent
# Update indices
for capability in capabilities:
self.capability_index[capability].add(agent_id)
self.type_index[agent_type].add(agent_id)
for tag in agent.tags:
self.tag_index[tag].add(agent_id)
# Record event
event = RegistryEvent(
event_type="registered",
agent_id=agent_id,
details={
"name": name,
"type": agent_type.value,
"capabilities": list(capabilities)
}
)
self.events.append(event)
self._notify_listeners(event)
self.statistics["total_registered"] += 1
self.statistics["active_agents"] += 1
return {
"success": True,
"agent_id": agent_id,
"name": name,
"type": agent_type.value,
"registered_at": agent.registered_at.isoformat(),
"message": f"Agent '{agent_id}' registered successfully"
}
def deregister(self, agent_id: str) -> Dict[str, Any]:
"""
Deregister an agent.
Args:
agent_id: Agent to deregister
Returns:
Deregistration status
"""
if agent_id not in self.agents:
return {
"success": False,
"message": f"Agent '{agent_id}' not found"
}
agent = self.agents[agent_id]
if agent.status == RegistrationStatus.DEREGISTERED:
return {
"success": False,
"message": f"Agent '{agent_id}' already deregistered"
}
# Update status
old_status = agent.status
agent.status = RegistrationStatus.DEREGISTERED
# Remove from indices
for capability in agent.capabilities:
self.capability_index[capability].discard(agent_id)
self.type_index[agent.agent_type].discard(agent_id)
for tag in agent.tags:
self.tag_index[tag].discard(agent_id)
# Record event
event = RegistryEvent(
event_type="deregistered",
agent_id=agent_id,
details={"previous_status": old_status.value}
)
self.events.append(event)
self._notify_listeners(event)
self.statistics["total_deregistered"] += 1
if old_status == RegistrationStatus.ACTIVE:
self.statistics["active_agents"] -= 1
elif old_status == RegistrationStatus.SUSPENDED:
self.statistics["suspended_agents"] -= 1
return {
"success": True,
"agent_id": agent_id,
"message": f"Agent '{agent_id}' deregistered"
}
def heartbeat(self, agent_id: str) -> Dict[str, Any]:
"""
Record agent heartbeat.
Args:
agent_id: Agent sending heartbeat
Returns:
Heartbeat status
"""
if agent_id not in self.agents:
return {
"success": False,
"message": f"Agent '{agent_id}' not registered"
}
agent = self.agents[agent_id]
if agent.status == RegistrationStatus.DEREGISTERED:
return {
"success": False,
"message": f"Agent '{agent_id}' is deregistered"
}
# Update heartbeat
agent.last_heartbeat = datetime.now()
# Reactivate if suspended
if agent.status == RegistrationStatus.SUSPENDED:
agent.status = RegistrationStatus.ACTIVE
self.statistics["suspended_agents"] -= 1
self.statistics["active_agents"] += 1
# Record event
event = RegistryEvent(
event_type="heartbeat",
agent_id=agent_id
)
self.events.append(event)
self.statistics["total_heartbeats"] += 1
return {
"success": True,
"agent_id": agent_id,
"status": agent.status.value,
"message": "Heartbeat recorded"
}
def check_timeouts(self) -> Dict[str, Any]:
"""
Check for agents that have timed out.
Returns:
List of timed out agents
"""
now = datetime.now()
timed_out = []
for agent_id, agent in self.agents.items():
if agent.status != RegistrationStatus.ACTIVE:
continue
if agent.last_heartbeat is None:
continue
time_since_heartbeat = now - agent.last_heartbeat
if time_since_heartbeat > self.heartbeat_timeout:
agent.status = RegistrationStatus.SUSPENDED
timed_out.append(agent_id)
event = RegistryEvent(
event_type="timeout",
agent_id=agent_id,
details={"timeout_seconds": time_since_heartbeat.total_seconds()}
)
self.events.append(event)
self._notify_listeners(event)
self.statistics["active_agents"] -= 1
self.statistics["suspended_agents"] += 1
self.statistics["timeouts"] += 1
return {
"timed_out_count": len(timed_out),
"agent_ids": timed_out,
"message": f"Found {len(timed_out)} timed out agents"
}
def lookup(self, agent_id: str) -> Dict[str, Any]:
"""
Lookup agent by ID.
Args:
agent_id: Agent identifier
Returns:
Agent information
"""
if agent_id not in self.agents:
return {
"success": False,
"message": f"Agent '{agent_id}' not found"
}
agent = self.agents[agent_id]
return {
"success": True,
"agent_id": agent_id,
"name": agent.name,
"type": agent.agent_type.value,
"capabilities": list(agent.capabilities),
"endpoint": agent.endpoint,
"version": agent.version,
"status": agent.status.value,
"registered_at": agent.registered_at.isoformat(),
"last_heartbeat": agent.last_heartbeat.isoformat() if agent.last_heartbeat else None,
"tags": list(agent.tags)
}
def find_by_capability(self, capability: str) -> Dict[str, Any]:
"""Find agents with specific capability."""
agent_ids = self.capability_index.get(capability, set())
# Filter active only
active_ids = [
aid for aid in agent_ids
if self.agents[aid].status == RegistrationStatus.ACTIVE
]
agents = [
{
"agent_id": aid,
"name": self.agents[aid].name,
"endpoint": self.agents[aid].endpoint
}
for aid in active_ids
]
return {
"capability": capability,
"count": len(agents),
"agents": agents
}
def find_by_type(self, agent_type: AgentType) -> Dict[str, Any]:
"""Find agents by type."""
agent_ids = self.type_index.get(agent_type, set())
active_ids = [
aid for aid in agent_ids
if self.agents[aid].status == RegistrationStatus.ACTIVE
]
agents = [
{
"agent_id": aid,
"name": self.agents[aid].name,
"capabilities": list(self.agents[aid].capabilities)
}
for aid in active_ids
]
return {
"type": agent_type.value,
"count": len(agents),
"agents": agents
}
def find_by_tag(self, tag: str) -> Dict[str, Any]:
"""Find agents by tag."""
agent_ids = self.tag_index.get(tag, set())
active_ids = [
aid for aid in agent_ids
if self.agents[aid].status == RegistrationStatus.ACTIVE
]
agents = [
{
"agent_id": aid,
"name": self.agents[aid].name,
"type": self.agents[aid].agent_type.value
}
for aid in active_ids
]
return {
"tag": tag,
"count": len(agents),
"agents": agents
}
def add_listener(self, listener: Callable[[RegistryEvent], None]) -> None:
"""Add event listener."""
self.listeners.append(listener)
def _notify_listeners(self, event: RegistryEvent) -> None:
"""Notify all listeners of event."""
for listener in self.listeners:
try:
listener(event)
except Exception:
pass # Don't let listener errors affect registry
def list_agents(
self,
status: Optional[RegistrationStatus] = None
) -> Dict[str, Any]:
"""List all agents with optional status filter."""
agents_list = []
for agent_id, agent in self.agents.items():
if status and agent.status != status:
continue
agents_list.append({
"agent_id": agent_id,
"name": agent.name,
"type": agent.agent_type.value,
"status": agent.status.value,
"capabilities": list(agent.capabilities)
})
return {
"total": len(agents_list),
"agents": agents_list
}
def get_statistics(self) -> Dict[str, Any]:
"""Get registry statistics."""
return {
"total_registered": self.statistics["total_registered"],
"total_deregistered": self.statistics["total_deregistered"],
"active_agents": self.statistics["active_agents"],
"suspended_agents": self.statistics["suspended_agents"],
"total_heartbeats": self.statistics["total_heartbeats"],
"timeouts": self.statistics["timeouts"],
"total_events": len(self.events),
"registered_capabilities": len(self.capability_index),
"registered_types": len([t for t, ids in self.type_index.items() if ids])
}
class AgentRegistryPattern:
"""
Coordinator for agent registry pattern.
Manages agent registration and lookup.
"""
def __init__(self, heartbeat_timeout_seconds: int = 60):
self.registry = AgentRegistry(heartbeat_timeout_seconds)
def register(self, agent_id: str, name: str, agent_type: AgentType, capabilities: Set[str], endpoint: str, **kwargs) -> Dict[str, Any]:
"""Register agent."""
return self.registry.register(agent_id, name, agent_type, capabilities, endpoint, **kwargs)
def deregister(self, agent_id: str) -> Dict[str, Any]:
"""Deregister agent."""
return self.registry.deregister(agent_id)
def heartbeat(self, agent_id: str) -> Dict[str, Any]:
"""Record heartbeat."""
return self.registry.heartbeat(agent_id)
def check_timeouts(self) -> Dict[str, Any]:
"""Check timeouts."""
return self.registry.check_timeouts()
def lookup(self, agent_id: str) -> Dict[str, Any]:
"""Lookup agent."""
return self.registry.lookup(agent_id)
def find_by_capability(self, capability: str) -> Dict[str, Any]:
"""Find by capability."""
return self.registry.find_by_capability(capability)
def list_agents(self) -> Dict[str, Any]:
"""List agents."""
return self.registry.list_agents()
def get_statistics(self) -> Dict[str, Any]:
"""Get statistics."""
return self.registry.get_statistics()
# Example usage
if __name__ == "__main__":
print("=== Agent Registry Pattern Demo ===\n")
# Create registry
pattern = AgentRegistryPattern(heartbeat_timeout_seconds=30)
print("1. Registering agents...")
pattern.register(
"agent-001",
"DataProcessor",
AgentType.WORKER,
{"data_processing", "etl"},
"http://localhost:8001",
tags={"production", "critical"}
)
pattern.register(
"agent-002",
"MLTrainer",
AgentType.SERVICE,
{"machine_learning", "training"},
"http://localhost:8002",
tags={"ml", "gpu"}
)
pattern.register(
"agent-003",
"Coordinator",
AgentType.COORDINATOR,
{"orchestration", "workflow"},
"http://localhost:8003"
)
agents = pattern.list_agents()
print(f" Registered {agents['total']} agents")
print("\n2. Looking up specific agent...")
info = pattern.lookup("agent-001")
if info['success']:
print(f" Agent: {info['name']}")
print(f" Type: {info['type']}")
print(f" Capabilities: {', '.join(info['capabilities'])}")
print(f" Endpoint: {info['endpoint']}")
print("\n3. Finding agents by capability...")
result = pattern.find_by_capability("data_processing")
print(f" Found {result['count']} agents with 'data_processing'")
for agent in result['agents']:
print(f" - {agent['agent_id']}: {agent['name']}")
print("\n4. Recording heartbeats...")
pattern.heartbeat("agent-001")
pattern.heartbeat("agent-002")
print(" Heartbeats recorded")
print("\n5. Checking for timeouts...")
import time
print(" (Simulating 2 second delay...)")
time.sleep(2)
pattern.heartbeat("agent-001") # Only agent-001 sends heartbeat
# agent-002 and agent-003 will timeout after 30 seconds in real scenario
print("\n6. Registry statistics:")
stats = pattern.get_statistics()
print(f" Total registered: {stats['total_registered']}")
print(f" Active agents: {stats['active_agents']}")
print(f" Total heartbeats: {stats['total_heartbeats']}")
print(f" Timeouts: {stats['timeouts']}")
print(f" Registered capabilities: {stats['registered_capabilities']}")
print("\nā Comprehensive agent registry pattern complete!")
self.state.current_step += 1
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, implementing Agent Registry. "
f"Provide clear, structured responses."),
("human", "Process this input: {input_data}")
])
chain = prompt | self.llm
response = chain.invoke({"input_data": str(input_data)})
result = {
"agent": self.name,
"step": self.state.current_step,
"input": input_data,
"output": str(response.content),
"timestamp": datetime.now()
}
self.execution_history.append(result)
self.state.results.append(result)
return result
def get_statistics(self) -> Dict[str, Any]:
"""Get agent statistics."""
return {
"name": self.name,
"total_executions": len(self.execution_history),
"current_step": self.state.current_step,
"results_count": len(self.state.results)
}
class AgentRegistryPattern:
"""
Agent Registry - Enhanced implementation
This pattern provides comprehensive functionality for agent registry
with full LangChain/LangGraph integration.
Features:
- Multi-agent coordination
- State management
- Execution tracking
- LangGraph workflow integration
"""
def __init__(self, config: AgentRegistryConfig = None):
self.config = config or AgentRegistryConfig()
self.agents: Dict[str, AgentRegistryAgent] = {}
self.workflow = self._build_workflow()
self.execution_log: List[Dict[str, Any]] = []
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow for pattern execution."""
class WorkflowState(TypedDict):
input: str
processed: str
output: str
metadata: Dict[str, Any]
def process_input(state: WorkflowState) -> WorkflowState:
"""Process input node."""
return {
**state,
"processed": f"Processed: {state['input']}",
"metadata": {"step": "process_input"}
}
def generate_output(state: WorkflowState) -> WorkflowState:
"""Generate output node."""
llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
prompt = ChatPromptTemplate.from_messages([
("system", "You are implementing Agent Registry. Generate comprehensive output."),
("human", "{input}")
])
chain = prompt | llm
response = chain.invoke({"input": state["processed"]})
return {
**state,
"output": str(response.content),
"metadata": {**state.get("metadata", {}), "step": "generate_output"}
}
# Build workflow graph
workflow = StateGraph(WorkflowState)
workflow.add_node("process", process_input)
workflow.add_node("generate", generate_output)
workflow.set_entry_point("process")
workflow.add_edge("process", "generate")
workflow.add_edge("generate", END)
return workflow.compile()
def register_agent(self, agent: AgentRegistryAgent):
"""Register an agent with the pattern."""
self.agents[agent.name] = agent
if self.config.enable_logging:
print(f"Registered agent: {agent.name}")
def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
"""Execute the pattern with given input."""
# Use workflow if no specific agent requested
if agent_name is None:
result = self.workflow.invoke({
"input": str(input_data),
"processed": "",
"output": "",
"metadata": {}
})
execution_record = {
"pattern": "Agent Registry",
"input": input_data,
"output": result.get("output", ""),
"metadata": result.get("metadata", {}),
"timestamp": datetime.now()
}
self.execution_log.append(execution_record)
return execution_record
# Execute with specific agent
agent = self.agents.get(agent_name)
if not agent:
raise ValueError(f"Agent {agent_name} not found")
result = agent.process(input_data)
self.execution_log.append(result)
return result
def run(self, input_data: Any) -> Dict[str, Any]:
"""Main execution method."""
return self.execute(input_data)
def get_statistics(self) -> Dict[str, Any]:
"""Get pattern statistics."""
return {
"pattern": "Agent Registry",
"total_executions": len(self.execution_log),
"registered_agents": len(self.agents),
"agents": [agent.get_statistics() for agent in self.agents.values()]
}
# Example usage
if __name__ == "__main__":
print("=== Agent Registry Demo ===\n")
# Create configuration
config = AgentRegistryConfig(
model_name="gpt-4",
temperature=0.7,
enable_logging=True
)
# Create pattern
pattern = AgentRegistryPattern(config)
# Create and register agents
agent1 = AgentRegistryAgent("Agent1", config)
agent2 = AgentRegistryAgent("Agent2", config)
pattern.register_agent(agent1)
pattern.register_agent(agent2)
print("\nExecuting pattern workflow...")
# Execute with workflow
result1 = pattern.run("Demonstrate agent registry functionality")
print(f"\nWorkflow Result:")
print(f" Output: {result1['output'][:150]}...")
# Execute with specific agents
result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
print(f"\nAgent1 Result:")
print(f" Output: {result2['output'][:150]}...")
result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
print(f"\nAgent2 Result:")
print(f" Output: {result3['output'][:150]}...")
# Show statistics
print("\n=== Statistics ===")
stats = pattern.get_statistics()
print(f"Pattern: {stats['pattern']}")
print(f"Total Executions: {stats['total_executions']}")
print(f"Registered Agents: {stats['registered_agents']}")
print(f"\nAgent Statistics:")
for agent_stats in stats['agents']:
print(f" {agent_stats['name']}: {agent_stats['total_executions']} executions")
Agentic Architecture Pattern: Agent Registry
Pattern 283: 283 Agent Registry – Line-by-Line Explanation
Overview
Pattern 283: Agent Registry Pattern
Purpose: Central registry for agent registration, lookup, and lifecycle management
Code Structure Analysis
File Information
- Pattern Number: 283
- Pattern Name: 283 Agent Registry
- File Path:
patterns\additional_advanced\283_agent_registry.py - Total Lines: 788
Detailed Code Explanation
Module Docstring
"""
Pattern 283: Agent Registry Pattern
Purpose: Central registry for agent registration, lookup, and lifecycle management
"""
Explanation: This module docstring describes the purpose and key features of the 283 Agent Registry pattern.
Imports Section
from typing import Dict, Any, List, Optional, Set, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from collections import defaultdict
import time # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Set, Callable: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime, timedelta: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
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
# Usage example not available
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