feat: implement swarm mode for multi-agent collaboration
- Add Swarm, SwarmAgent, SwarmMessage database models - Implement SwarmOrchestrator for task coordination - Add /api/swarm/* REST endpoints (create, status, results, stop, logs) - Extend K8sManager with swarm-specific methods - Support sequential, parallel, and hybrid orchestration strategies - Enable SSE streaming for real-time progress updates - Integrate with existing A2A agent framework
This commit is contained in:
elipitc
+208
@@ -1304,3 +1304,211 @@ class K8sManager:
|
||||
except ApiException as e:
|
||||
logger.error(f"列出Pod失败: {e}")
|
||||
raise Exception(f"列出Pod失败: {e.reason}")
|
||||
|
||||
# ============================================================================
|
||||
# Swarm Mode Methods (NEW)
|
||||
# ============================================================================
|
||||
|
||||
def create_swarm_namespace(self, swarm_id: str, role: str) -> str:
|
||||
"""
|
||||
Create namespace for swarm agent.
|
||||
|
||||
Args:
|
||||
swarm_id: Swarm ID
|
||||
role: Agent role
|
||||
|
||||
Returns:
|
||||
Namespace name
|
||||
"""
|
||||
namespace_name = sanitize_k8s_name(f"swarm-{swarm_id[:8]}-{role}")
|
||||
|
||||
try:
|
||||
# Check if namespace exists
|
||||
try:
|
||||
self.v1.read_namespace(name=namespace_name)
|
||||
logger.info(f"Namespace {namespace_name} already exists")
|
||||
return namespace_name
|
||||
except ApiException as e:
|
||||
if e.status != 404:
|
||||
raise
|
||||
|
||||
# Create namespace
|
||||
namespace = client.V1Namespace(
|
||||
metadata=client.V1ObjectMeta(
|
||||
name=namespace_name,
|
||||
labels={
|
||||
"managed-by": "agent-manager",
|
||||
"swarm-id": swarm_id[:8],
|
||||
"agent-role": role
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
self.v1.create_namespace(body=namespace)
|
||||
logger.info(f"✅ Created namespace: {namespace_name}")
|
||||
return namespace_name
|
||||
|
||||
except ApiException as e:
|
||||
logger.error(f"Failed to create namespace {namespace_name}: {e}")
|
||||
raise Exception(f"Failed to create namespace: {e.reason}")
|
||||
|
||||
def deploy_swarm_agent(
|
||||
self,
|
||||
swarm_id: str,
|
||||
agent_id: str,
|
||||
agent_config: Dict,
|
||||
namespace: str
|
||||
) -> Dict:
|
||||
"""
|
||||
Deploy swarm agent pod.
|
||||
|
||||
Args:
|
||||
swarm_id: Swarm ID
|
||||
agent_id: Agent ID
|
||||
agent_config: Agent configuration
|
||||
namespace: Namespace
|
||||
|
||||
Returns:
|
||||
Deployment info {pod_name, service_url, external_ip}
|
||||
"""
|
||||
try:
|
||||
pod_name = f"agent-{agent_id}"
|
||||
template = agent_config.get("template", "a2a_litellm_agent")
|
||||
role = agent_config.get("role", "worker")
|
||||
model = agent_config.get("model", "gpt-4")
|
||||
|
||||
# Get template image
|
||||
# TODO: Load from template database
|
||||
image_map = {
|
||||
"a2a_litellm_agent": "agnettaiji.azurecr.io/a2a_litellm_agent:latest",
|
||||
"code_manager_agent": "agnettaiji.azurecr.io/code_manager_agent:latest"
|
||||
}
|
||||
image = image_map.get(template, image_map["a2a_litellm_agent"])
|
||||
|
||||
# Environment variables
|
||||
env_vars = [
|
||||
client.V1EnvVar(name="SWARM_ID", value=swarm_id),
|
||||
client.V1EnvVar(name="AGENT_ID", value=agent_id),
|
||||
client.V1EnvVar(name="AGENT_ROLE", value=role),
|
||||
client.V1EnvVar(name="MODEL_NAME", value=model),
|
||||
client.V1EnvVar(name="POD_NAME", value=pod_name),
|
||||
client.V1EnvVar(name="NAMESPACE", value=namespace),
|
||||
]
|
||||
|
||||
# Add model API key from environment
|
||||
litellm_key = os.getenv("LITELLM_API_KEY")
|
||||
if litellm_key:
|
||||
env_vars.append(client.V1EnvVar(name="LITELLM_API_KEY", value=litellm_key))
|
||||
|
||||
# Create pod
|
||||
pod = client.V1Pod(
|
||||
metadata=client.V1ObjectMeta(
|
||||
name=pod_name,
|
||||
namespace=namespace,
|
||||
labels={
|
||||
"app": pod_name,
|
||||
"managed-by": "agent-manager",
|
||||
"swarm-id": swarm_id[:8],
|
||||
"agent-id": agent_id,
|
||||
"agent-role": role
|
||||
}
|
||||
),
|
||||
spec=client.V1PodSpec(
|
||||
containers=[
|
||||
client.V1Container(
|
||||
name="agent",
|
||||
image=image,
|
||||
ports=[client.V1ContainerPort(container_port=8000)],
|
||||
env=env_vars,
|
||||
resources=client.V1ResourceRequirements(
|
||||
requests={"cpu": "100m", "memory": "256Mi"},
|
||||
limits={"cpu": "500m", "memory": "512Mi"}
|
||||
)
|
||||
)
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
self.v1.create_namespaced_pod(namespace=namespace, body=pod)
|
||||
logger.info(f"✅ Created pod: {pod_name} in namespace {namespace}")
|
||||
|
||||
# Create service
|
||||
service = client.V1Service(
|
||||
metadata=client.V1ObjectMeta(
|
||||
name=pod_name,
|
||||
namespace=namespace,
|
||||
labels={"app": pod_name}
|
||||
),
|
||||
spec=client.V1ServiceSpec(
|
||||
selector={"app": pod_name},
|
||||
ports=[client.V1ServicePort(port=8000, target_port=8000)],
|
||||
type="ClusterIP"
|
||||
)
|
||||
)
|
||||
|
||||
self.v1.create_namespaced_service(namespace=namespace, body=service)
|
||||
logger.info(f"✅ Created service: {pod_name} in namespace {namespace}")
|
||||
|
||||
service_url = f"http://{pod_name}.{namespace}.svc.cluster.local:8000"
|
||||
|
||||
return {
|
||||
"pod_name": pod_name,
|
||||
"service_url": service_url,
|
||||
"external_ip": None,
|
||||
"namespace": namespace
|
||||
}
|
||||
|
||||
except ApiException as e:
|
||||
logger.error(f"Failed to deploy swarm agent: {e}")
|
||||
raise Exception(f"Failed to deploy swarm agent: {e.reason}")
|
||||
|
||||
def cleanup_swarm_resources(self, swarm_id: str):
|
||||
"""
|
||||
Cleanup all K8s resources for a swarm.
|
||||
|
||||
Args:
|
||||
swarm_id: Swarm ID
|
||||
"""
|
||||
try:
|
||||
# List all namespaces with swarm-id label
|
||||
namespaces = self.v1.list_namespace(
|
||||
label_selector=f"swarm-id={swarm_id[:8]}"
|
||||
)
|
||||
|
||||
for ns in namespaces.items:
|
||||
namespace_name = ns.metadata.name
|
||||
logger.info(f"Deleting namespace: {namespace_name}")
|
||||
|
||||
# Delete namespace (this will delete all resources in it)
|
||||
self.v1.delete_namespace(name=namespace_name)
|
||||
logger.info(f"✅ Deleted namespace: {namespace_name}")
|
||||
|
||||
logger.info(f"✅ Cleaned up all resources for swarm {swarm_id}")
|
||||
|
||||
except ApiException as e:
|
||||
logger.error(f"Failed to cleanup swarm resources: {e}")
|
||||
raise Exception(f"Failed to cleanup swarm resources: {e.reason}")
|
||||
|
||||
def get_swarm_agent_logs(self, namespace: str, pod_name: str, tail_lines: int = 100) -> str:
|
||||
"""
|
||||
Get logs from swarm agent pod.
|
||||
|
||||
Args:
|
||||
namespace: Namespace
|
||||
pod_name: Pod name
|
||||
tail_lines: Number of lines to tail
|
||||
|
||||
Returns:
|
||||
Pod logs
|
||||
"""
|
||||
try:
|
||||
logs = self.v1.read_namespaced_pod_log(
|
||||
name=pod_name,
|
||||
namespace=namespace,
|
||||
tail_lines=tail_lines
|
||||
)
|
||||
return logs
|
||||
|
||||
except ApiException as e:
|
||||
logger.error(f"Failed to get pod logs: {e}")
|
||||
raise Exception(f"Failed to get pod logs: {e.reason}")
|
||||
|
||||
Reference in New Issue
Block a user