Spaces:
Runtime error
Runtime error
| import requests | |
| import json | |
| from typing import Dict, Any, Optional, List | |
| from config import SWARMS_API_KEY, SWARMS_BASE_URL, DEFAULT_MODEL, DEFAULT_MAX_TOKENS, DEFAULT_TEMPERATURE, DEFAULT_MAX_LOOPS | |
| class BaseAgent: | |
| def __init__(self, name: str, system_prompt: str, model_name: str = DEFAULT_MODEL): | |
| self.name = name | |
| self.system_prompt = system_prompt | |
| self.model_name = model_name | |
| self.max_tokens = DEFAULT_MAX_TOKENS | |
| self.temperature = DEFAULT_TEMPERATURE | |
| self.max_loops = DEFAULT_MAX_LOOPS | |
| self.headers = { | |
| "x-api-key": SWARMS_API_KEY, | |
| "Content-Type": "application/json" | |
| } | |
| def analyze(self, task: str, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: | |
| """Analyze the given task using the agent's expertise""" | |
| if context: | |
| task = f"{task}\n\nContext: {json.dumps(context)}" | |
| agent_config = { | |
| "agent_name": self.name, | |
| "system_prompt": self.system_prompt, | |
| "model_name": self.model_name, | |
| "role": "worker", | |
| "max_loops": self.max_loops, | |
| "max_tokens": self.max_tokens, | |
| "temperature": self.temperature, | |
| "task": task | |
| } | |
| try: | |
| if SWARMS_API_KEY: | |
| response = requests.post( | |
| f"{SWARMS_BASE_URL}/v1/agent/completions", | |
| headers=self.headers, | |
| json=agent_config, | |
| timeout=120 | |
| ) | |
| return response.json() | |
| else: | |
| # Mock response for development without API key | |
| return { | |
| "status": "success", | |
| "output": { | |
| "agent_name": self.name, | |
| "content": f"Mock analysis for {self.name} based on task: {task[:100]}..." | |
| } | |
| } | |
| except Exception as e: | |
| return {"error": f"Agent analysis failed: {str(e)}"} |