From 86277fdbdcc15cea12c277612ffda7ef795b2ca7 Mon Sep 17 00:00:00 2001 From: Berack96 Date: Sun, 12 Oct 2025 15:58:27 +0200 Subject: [PATCH] Aggiunto configurazione dell'applicazione --- configs.yaml | 41 +++++++ src/app/__main__.py | 8 +- src/app/agents/__init__.py | 5 +- src/app/agents/models.py | 107 ----------------- src/app/agents/pipeline.py | 37 +++--- src/app/agents/predictor.py | 8 +- src/app/agents/team.py | 12 +- src/app/configs.py | 222 ++++++++++++++++++++++++++++++++++++ 8 files changed, 298 insertions(+), 142 deletions(-) create mode 100644 configs.yaml delete mode 100644 src/app/agents/models.py create mode 100644 src/app/configs.py diff --git a/configs.yaml b/configs.yaml new file mode 100644 index 0000000..7dd259d --- /dev/null +++ b/configs.yaml @@ -0,0 +1,41 @@ +port: 8000 +gradio_share: false +logging_level: INFO + +strategies: + - name: Conservative + label: Conservative + description: Focus on stable and low-risk investments. + - name: Balanced + label: Balanced + description: A mix of growth and stability. + - name: Aggressive + label: Aggressive + description: High-risk, high-reward investments. + +models: + gemini: + - name: gemini-2.0-flash + label: Gemini + - name: gemini-2.0-pro + label: Gemini Pro + ollama: + - name: gpt-oss:latest + label: Ollama GPT + - name: qwen3:8b + label: Qwen 3 (8B) + - name: qwen3:4b + label: Qwen 3 (4B) + - name: qwen3:1.7b + label: Qwen 3 (1.7B) + +api: + retry_attempts: 3 + retry_delay_seconds: 2 + currency: EUR + +agents: + strategy: Conservative + team_model: qwen3:1.7b + team_leader_model: qwen3:4b + predictor_model: qwen3:4b diff --git a/src/app/__main__.py b/src/app/__main__.py index 578ef35..d924e9a 100644 --- a/src/app/__main__.py +++ b/src/app/__main__.py @@ -1,6 +1,7 @@ import gradio as gr from dotenv import load_dotenv from agno.utils.log import log_info #type: ignore +from app.configs import AppConfig from app.utils import ChatManager from app.agents import Pipeline @@ -8,7 +9,10 @@ from app.agents import Pipeline if __name__ == "__main__": # Inizializzazioni load_dotenv() - pipeline = Pipeline() + + configs = AppConfig.load() + pipeline = Pipeline(configs) + chat = ChatManager() ######################################## @@ -57,7 +61,7 @@ if __name__ == "__main__": type="index", label="Stile di investimento" ) - style.change(fn=pipeline.choose_style, inputs=style, outputs=None) + style.change(fn=pipeline.choose_strategy, inputs=style, outputs=None) chatbot = gr.Chatbot(label="Conversazione", height=500, type="messages") msg = gr.Textbox(label="Scrivi la tua richiesta", placeholder="Es: Quali sono le crypto interessanti oggi?") diff --git a/src/app/agents/__init__.py b/src/app/agents/__init__.py index e23f31d..7d4287b 100644 --- a/src/app/agents/__init__.py +++ b/src/app/agents/__init__.py @@ -1,6 +1,5 @@ -from app.agents.models import AppModels -from app.agents.predictor import PredictorInput, PredictorOutput, PredictorStyle +from app.agents.predictor import PredictorInput, PredictorOutput from app.agents.team import create_team_with from app.agents.pipeline import Pipeline -__all__ = ["AppModels", "PredictorInput", "PredictorOutput", "PredictorStyle", "create_team_with", "Pipeline"] +__all__ = ["PredictorInput", "PredictorOutput", "create_team_with", "Pipeline"] diff --git a/src/app/agents/models.py b/src/app/agents/models.py deleted file mode 100644 index 79d4a26..0000000 --- a/src/app/agents/models.py +++ /dev/null @@ -1,107 +0,0 @@ -import os -import ollama -from enum import Enum -from agno.agent import Agent -from agno.models.base import Model -from agno.models.google import Gemini -from agno.models.ollama import Ollama -from agno.tools import Toolkit -from agno.utils.log import log_warning #type: ignore -from pydantic import BaseModel - - -class AppModels(Enum): - """ - Enum per i modelli supportati. - Aggiungere nuovi modelli qui se necessario. - Per quanto riguarda Ollama, i modelli dovranno essere scaricati e installati - localmente seguendo le istruzioni di https://ollama.com/docs/guide/install-models - """ - GEMINI = "gemini-2.0-flash" # API online - GEMINI_PRO = "gemini-2.0-pro" # API online, più costoso ma migliore - OLLAMA_GPT = "gpt-oss:latest" # + good - slow (13b) - OLLAMA_QWEN = "qwen3:latest" # + good + fast (8b) - OLLAMA_QWEN_4B = "qwen3:4b" # + fast + decent (4b) - OLLAMA_QWEN_1B = "qwen3:1.7b" # + very fast + decent (1.7b) - - @staticmethod - def availables_local() -> list['AppModels']: - """ - Controlla quali provider di modelli LLM locali sono disponibili. - Ritorna una lista di provider disponibili. - """ - try: - models_list = ollama.list() - availables = [model['model'] for model in models_list['models']] - app_models = [model for model in AppModels if model.name.startswith("OLLAMA")] - return [model for model in app_models if model.value in availables] - except Exception as e: - log_warning(f"Ollama is not running or not reachable: {e}") - return [] - - @staticmethod - def availables_online() -> list['AppModels']: - """ - Controlla quali provider di modelli LLM online hanno le loro API keys disponibili - come variabili d'ambiente e ritorna una lista di provider disponibili. - """ - if not os.getenv("GOOGLE_API_KEY"): - log_warning("No GOOGLE_API_KEY set in environment variables.") - return [] - availables = [AppModels.GEMINI, AppModels.GEMINI_PRO] - return availables - - @staticmethod - def availables() -> list['AppModels']: - """ - Controlla quali provider di modelli LLM locali sono disponibili e quali - provider di modelli LLM online hanno le loro API keys disponibili come variabili - d'ambiente e ritorna una lista di provider disponibili. - L'ordine di preferenza è: - 1. Gemini (Google) - 2. Ollama (locale) - """ - availables = [ - *AppModels.availables_online(), - *AppModels.availables_local() - ] - assert availables, "No valid model API keys set in environment variables." - return availables - - def get_model(self, instructions:str) -> Model: - """ - Restituisce un'istanza del modello specificato. - Args: - instructions: istruzioni da passare al modello (system prompt). - Returns: - Un'istanza di BaseModel o una sua sottoclasse. - Raise: - ValueError se il modello non è supportato. - """ - name = self.value - if self in {model for model in AppModels if model.name.startswith("GEMINI")}: - return Gemini(name, instructions=[instructions]) - elif self in {model for model in AppModels if model.name.startswith("OLLAMA")}: - return Ollama(name, instructions=[instructions]) - - raise ValueError(f"Modello non supportato: {self}") - - def get_agent(self, instructions: str, name: str = "", output_schema: type[BaseModel] | None = None, tools: list[Toolkit] | None = None) -> Agent: - """ - Costruisce un agente con il modello e le istruzioni specificate. - Args: - instructions: istruzioni da passare al modello (system prompt) - name: nome dell'agente (opzionale) - output: schema di output opzionale (Pydantic BaseModel) - tools: lista opzionale di strumenti (tools) da fornire all'agente - Returns: - Un'istanza di Agent. - """ - return Agent( - model=self.get_model(instructions), - name=name, - retries=2, - tools=tools, - delay_between_retries=5, # seconds - output_schema=output_schema - ) diff --git a/src/app/agents/pipeline.py b/src/app/agents/pipeline.py index f1d2147..fa260f1 100644 --- a/src/app/agents/pipeline.py +++ b/src/app/agents/pipeline.py @@ -1,9 +1,9 @@ from agno.run.agent import RunOutput -from app.agents.models import AppModels from app.agents.team import create_team_with -from app.agents.predictor import PredictorInput, PredictorOutput, PredictorStyle +from app.agents.predictor import PredictorInput, PredictorOutput from app.agents.prompts import * from app.api.base.markets import ProductInfo +from app.configs import AppConfig class Pipeline: @@ -13,13 +13,12 @@ class Pipeline: e scelto dall'utente tramite i dropdown dell'interfaccia grafica. """ - def __init__(self): - self.available_models = AppModels.availables() - self.all_styles = list(PredictorStyle) + def __init__(self, configs: AppConfig): + self.configs = configs - self.style = self.all_styles[0] - self.team = create_team_with(AppModels.OLLAMA_QWEN_1B) - self.choose_predictor(0) # Modello di default + # Stato iniziale + self.choose_strategy(0) + self.choose_predictor(0) # ====================== # Dropdown handlers @@ -28,17 +27,17 @@ class Pipeline: """ Sceglie il modello LLM da usare per il Predictor. """ - model = self.available_models[index] + model = self.configs.models.all_models[index] self.predictor = model.get_agent( PREDICTOR_INSTRUCTIONS, output_schema=PredictorOutput, ) - def choose_style(self, index: int): + def choose_strategy(self, index: int): """ - Sceglie lo stile (conservativo/aggressivo) da usare per il Predictor. + Sceglie la strategia da usare per il Predictor. """ - self.style = self.all_styles[index] + self.strat = self.configs.strategies[index].description # ====================== # Helpers @@ -47,13 +46,13 @@ class Pipeline: """ Restituisce la lista dei nomi dei modelli disponibili. """ - return [model.name for model in self.available_models] + return [model.label for model in self.configs.models.all_models] def list_styles(self) -> list[str]: """ Restituisce la lista degli stili di previsione disponibili. """ - return [style.value for style in self.all_styles] + return [strat.label for strat in self.configs.strategies] # ====================== # Core interaction @@ -66,7 +65,11 @@ class Pipeline: 4. Restituisce la strategia finale """ # Step 1: raccolta output dai membri del Team - team_outputs = self.team.run(query) # type: ignore + team_model = self.configs.get_model_by_name(self.configs.agents.team_model) + leader_model = self.configs.get_model_by_name(self.configs.agents.team_leader_model) + + team = create_team_with(team_model, leader_model) + team_outputs = team.run(query) # type: ignore # Step 2: aggregazione output strutturati all_products: list[ProductInfo] = [] @@ -87,7 +90,7 @@ class Pipeline: # Step 3: invocazione Predictor predictor_input = PredictorInput( data=all_products, - style=self.style, + style=self.strat, sentiment=aggregated_sentiment ) @@ -101,6 +104,6 @@ class Pipeline: [f"{item.asset} ({item.percentage}%): {item.motivation}" for item in prediction.portfolio] ) return ( - f"📊 Strategia ({self.style.value}): {prediction.strategy}\n\n" + f"📊 Strategia ({self.strat}): {prediction.strategy}\n\n" f"💼 Portafoglio consigliato:\n{portfolio_lines}" ) diff --git a/src/app/agents/predictor.py b/src/app/agents/predictor.py index 2ef709c..da79e7e 100644 --- a/src/app/agents/predictor.py +++ b/src/app/agents/predictor.py @@ -1,15 +1,9 @@ -from enum import Enum from pydantic import BaseModel, Field from app.api.base.markets import ProductInfo - -class PredictorStyle(Enum): - CONSERVATIVE = "Conservativo" - AGGRESSIVE = "Aggressivo" - class PredictorInput(BaseModel): data: list[ProductInfo] = Field(..., description="Market data as a list of ProductInfo") - style: PredictorStyle = Field(..., description="Prediction style") + style: str = Field(..., description="Prediction style") sentiment: str = Field(..., description="Aggregated sentiment from news and social analysis") class ItemPortfolio(BaseModel): diff --git a/src/app/agents/team.py b/src/app/agents/team.py index 8b12db6..076bfb3 100644 --- a/src/app/agents/team.py +++ b/src/app/agents/team.py @@ -1,29 +1,29 @@ from agno.team import Team -from app.agents import AppModels from app.api.markets import MarketAPIsTool from app.api.news import NewsAPIsTool from app.api.social import SocialAPIsTool from app.agents.prompts import * +from app.configs import AppModel -def create_team_with(models: AppModels, coordinator: AppModels | None = None) -> Team: - market_agent = models.get_agent( +def create_team_with(model: AppModel, coordinator: AppModel | None = None) -> Team: + market_agent = model.get_agent( instructions=MARKET_INSTRUCTIONS, name="MarketAgent", tools=[MarketAPIsTool()] ) - news_agent = models.get_agent( + news_agent = model.get_agent( instructions=NEWS_INSTRUCTIONS, name="NewsAgent", tools=[NewsAPIsTool()] ) - social_agent = models.get_agent( + social_agent = model.get_agent( instructions=SOCIAL_INSTRUCTIONS, name="SocialAgent", tools=[SocialAPIsTool()] ) - coordinator = coordinator or models + coordinator = coordinator or model return Team( model=coordinator.get_model(COORDINATOR_INSTRUCTIONS), name="CryptoAnalysisTeam", diff --git a/src/app/configs.py b/src/app/configs.py new file mode 100644 index 0000000..2687671 --- /dev/null +++ b/src/app/configs.py @@ -0,0 +1,222 @@ +import os +import ollama +import yaml +import logging.config +from pydantic import BaseModel +from agno.agent import Agent +from agno.tools import Toolkit +from agno.models.base import Model +from agno.models.google import Gemini +from agno.models.ollama import Ollama + +log = logging.getLogger(__name__) + + + +class AppModel(BaseModel): + name: str = "gemini-2.0-flash" + label: str = "Gemini" + model: type[Model] | None = None + + def get_model(self, instructions: str) -> Model: + """ + Restituisce un'istanza del modello specificato. + Args: + instructions: istruzioni da passare al modello (system prompt). + Returns: + Un'istanza di BaseModel o una sua sottoclasse. + Raise: + ValueError se il modello non è supportato. + """ + if self.model is None: + raise ValueError(f"Model class for '{self.name}' is not set.") + return self.model(id=self.name, instructions=[instructions]) + + def get_agent(self, instructions: str, name: str = "", output_schema: type[BaseModel] | None = None, tools: list[Toolkit] | None = None) -> Agent: + """ + Costruisce un agente con il modello e le istruzioni specificate. + Args: + instructions: istruzioni da passare al modello (system prompt) + name: nome dell'agente (opzionale) + output: schema di output opzionale (Pydantic BaseModel) + tools: lista opzionale di strumenti (tools) da fornire all'agente + Returns: + Un'istanza di Agent. + """ + return Agent( + model=self.get_model(instructions), + name=name, + retries=2, + tools=tools, + delay_between_retries=5, # seconds + output_schema=output_schema + ) + +class APIConfig(BaseModel): + retry_attempts: int = 3 + retry_delay_seconds: int = 2 + currency: str = "USD" + +class Strategy(BaseModel): + name: str = "Conservative" + label: str = "Conservative" + description: str = "Focus on low-risk investments with steady returns." + +class ModelsConfig(BaseModel): + gemini: list[AppModel] = [AppModel()] + ollama: list[AppModel] = [] + + @property + def all_models(self) -> list[AppModel]: + return self.gemini + self.ollama + +class AgentsConfigs(BaseModel): + strategy: str = "Conservative" + team_model: str = "gemini-2.0-flash" + team_leader_model: str = "gemini-2.0-flash" + predictor_model: str = "gemini-2.0-flash" + +class AppConfig(BaseModel): + port: int = 8000 + gradio_share: bool = False + logging_level: str = "INFO" + api: APIConfig = APIConfig() + strategies: list[Strategy] = [Strategy()] + models: ModelsConfig = ModelsConfig() + agents: AgentsConfigs = AgentsConfigs() + + @classmethod + def load(cls, file_path: str = "configs.yaml") -> 'AppConfig': + """ + Load the application configuration from a YAML file. + Be sure to call load_dotenv() before if you use environment variables. + Args: + file_path: path to the YAML configuration file. + Returns: + An instance of AppConfig with the loaded settings. + """ + with open(file_path, 'r') as f: + data = yaml.safe_load(f) + + configs = cls(**data) + configs.set_logging_level() + configs.validate_models() + log.info(f"Loaded configuration from {file_path}") + return configs + + def get_model_by_name(self, name: str) -> AppModel: + """ + Retrieve a model configuration by its name. + Args: + name: the name of the model to retrieve. + Returns: + The AppModel instance if found. + Raises: + ValueError if no model with the specified name is found. + """ + for model in self.models.all_models: + if model.name == name: + return model + raise ValueError(f"Model with name '{name}' not found.") + + def get_strategy_by_name(self, name: str) -> Strategy: + """ + Retrieve a strategy configuration by its name. + Args: + name: the name of the strategy to retrieve. + Returns: + The Strategy instance if found. + Raises: + ValueError if no strategy with the specified name is found. + """ + for strat in self.strategies: + if strat.name == name: + return strat + raise ValueError(f"Strategy with name '{name}' not found.") + + def set_logging_level(self) -> None: + """ + Set the logging level based on the configuration. + """ + logging.config.dictConfig({ + 'version': 1, + 'disable_existing_loggers': False, # Mantiene i logger esistenti (es. di terze parti) + 'formatters': { + 'colored': { + '()': 'colorlog.ColoredFormatter', + 'format': '%(log_color)s%(levelname)s%(reset)s [%(asctime)s] (%(name)s) - %(message)s' + }, + }, + 'handlers': { + 'console': { + 'class': 'logging.StreamHandler', + 'formatter': 'colored', + 'level': self.logging_level, + }, + }, + 'root': { # Configura il logger root + 'handlers': ['console'], + 'level': self.logging_level, + }, + 'loggers': { + 'httpx': {'level': 'WARNING'}, # Troppo spam per INFO + } + }) + + # Modifichiamo i logger di agno + import agno.utils.log # type: ignore + agno_logger_names = ["agno", "agno-team", "agno-workflow"] + for logger_name in agno_logger_names: + logger = logging.getLogger(logger_name) + logger.handlers.clear() + logger.propagate = True + + def validate_models(self) -> None: + """ + Validate the configured models for each provider. + """ + self.__validate_online_models("gemini", clazz=Gemini, key="GOOGLE_API_KEY") + self.__validate_ollama_models() + + def __validate_online_models(self, provider: str, clazz: type[Model], key: str | None = None) -> None: + """ + Validate models for online providers like Gemini. + Args: + provider: name of the provider (e.g. "gemini") + clazz: class of the model (e.g. Gemini) + key: API key required for the provider (optional) + """ + if getattr(self.models, provider) is None: + log.warning(f"No models configured for provider '{provider}'.") + + models: list[AppModel] = getattr(self.models, provider) + if key and os.getenv(key) is None: + log.warning(f"No {key} set in environment variables for {provider}.") + models.clear() + return + + for model in models: + model.model = clazz + + def __validate_ollama_models(self) -> None: + """ + Validate models for the Ollama provider. + """ + try: + models_list = ollama.list() + availables = {model['model'] for model in models_list['models']} + not_availables: list[str] = [] + + for model in self.models.ollama: + if model.name in availables: + model.model = Ollama + else: + not_availables.append(model.name) + if not_availables: + log.warning(f"Ollama models not available: {not_availables}") + + self.models.ollama = [model for model in self.models.ollama if model.model] + + except Exception as e: + log.warning(f"Ollama is not running or not reachable: {e}") +