| from __future__ import annotations
|
| from abc import ABC, abstractmethod
|
| from typing import Dict, Any, Optional, List
|
|
|
| try:
|
| from src.services.llm_client import LLMClient, LLMConfig
|
| except ImportError:
|
| from services.llm_client import LLMClient, LLMConfig
|
|
|
|
|
| class BaseAgent(ABC):
|
| name: str = "base_agent"
|
| description: str = "Agent base class"
|
|
|
| def __init__(
|
| self,
|
| llm_client: Optional[LLMClient] = None,
|
| llm_config: Optional[LLMConfig] = None,
|
| ):
|
| if llm_client:
|
| self.llm = llm_client
|
| elif llm_config:
|
| self.llm = LLMClient(llm_config)
|
| else:
|
| self.llm = LLMClient.for_lmstudio()
|
|
|
| def think(self, prompt: str, system: Optional[str] = None) -> str:
|
| return self.llm.complete(prompt, system or self.get_system_prompt())
|
|
|
| def get_system_prompt(self) -> str:
|
| return f"""You are {self.name}. {self.description}
|
| You are an expert at your task. Think carefully and concisely.
|
| Always respond with actionable information."""
|
|
|
|
|
| class AgentResponse:
|
| def __init__(
|
| self, success: bool, data: Any = None, error: str = None, reasoning: str = ""
|
| ):
|
| self.success = success
|
| self.data = data
|
| self.error = error
|
| self.reasoning = reasoning
|
|
|
| def to_dict(self) -> Dict[str, Any]:
|
| return {
|
| "success": self.success,
|
| "data": self.data,
|
| "error": self.error,
|
| "reasoning": self.reasoning,
|
| }
|
|
|
|
|
| class LLMAgent(BaseAgent, ABC):
|
| def __init__(
|
| self,
|
| llm_client: Optional[LLMClient] = None,
|
| llm_config: Optional[LLMConfig] = None,
|
| **kwargs,
|
| ):
|
| super().__init__(llm_client, llm_config)
|
| self.provider = kwargs.get("provider", "lmstudio")
|
| self.model = kwargs.get("model", "local-model")
|
|
|
| def process(self, input_data: Any) -> AgentResponse:
|
| try:
|
| reasoning = self.think(self.build_prompt(input_data))
|
| result = self.execute(input_data, reasoning)
|
| return AgentResponse(success=True, data=result, reasoning=reasoning)
|
| except Exception as e:
|
| return AgentResponse(success=False, error=str(e))
|
|
|
| @abstractmethod
|
| def build_prompt(self, input_data: Any) -> str:
|
| pass
|
|
|
| @abstractmethod
|
| def execute(self, input_data: Any, reasoning: str) -> Any:
|
| pass
|
|
|
