finance-buddy / agents /base_agent.py
lucifer7210's picture
Upload 40 files
8809c2b verified
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)}"}