Spaces:
Paused
Paused
File size: 1,683 Bytes
4a2ab42 4ae946d 4a2ab42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 | import logging
from typing import Any
from core.plugin_system import PluginContext, PluginInterface, PluginMetadata
logger = logging.getLogger(__name__)
class RoundRobinAssignerPlugin(PluginInterface):
"""
Assigns cases to agents in a round-robin fashion.
"""
@property
def metadata(self) -> PluginMetadata:
return PluginMetadata(
name="round_robin_assigner",
version="1.0.0",
namespace="zenith/workflow/round_robin_assigner",
author="Zenith Team",
description="Assigns cases to agents in a round-robin fashion",
dependencies={},
capabilities=["workflow", "assignment"],
security_level="official",
api_version="v1",
)
async def initialize(self, context: PluginContext) -> bool:
self.context = context
self.counter = 0
return True
async def execute(self, inputs: dict[str, Any]) -> dict[str, Any]:
"""
Assigns a case.
Inputs: {"case_id": "123", "agents": ["Alice", "Bob", "Charlie"]}
"""
agents = inputs.get("agents", [])
if not agents:
return {"error": "No agents provided"}
# Simplistic Round Robin
assigned_agent = agents[self.counter % len(agents)]
self.counter += 1
logger.info(
f"🔄 [AssignmentPlugin] Assigned case {inputs.get('case_id')} to {assigned_agent}"
)
return {"assigned_agent": assigned_agent, "strategy": "round_robin"}
async def cleanup(self) -> None:
pass
def validate_config(self, config: dict[str, Any]) -> list[str]:
return []
|