File size: 2,483 Bytes
734b5b4 | 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 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | 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
|