Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Refactor action registration to use a decorator for user-friendly descriptions

Browse Source
This commit is contained in:
trojanhorse47
2025-10-31 13:55:59 +01:00
parent 2803263bed
commit 0598e58708
7 changed files with 48 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/app/agents/action_registry.py
View File

@@ -1,13 +1,6 @@
# Registro popolato da tutti i file Toolkit presenti all'avvio.
# Registro centrale popolato da tutti i file Toolkit all'avvio.
ACTION_DESCRIPTIONS: dict[str, str] = {}
def register_friendly_actions(actions: dict[str, str]):
"""
Aggiunge le descrizioni di un Toolkit al registro globale.
"""
global ACTION_DESCRIPTIONS
ACTION_DESCRIPTIONS.update(actions)
def get_user_friendly_action(tool_name: str) -> str:
"""
Restituisce un messaggio leggibile e descrittivo per l'utente
@@ -15,3 +8,27 @@ def get_user_friendly_action(tool_name: str) -> str:
"""
# Usa il dizionario ACTION_DESCRIPTIONS importato
return ACTION_DESCRIPTIONS.get(tool_name, f"⚙️ Eseguo l'operazione: {tool_name}...")
def friendly_action(description: str):
"""
Decoratore che registra automaticamente la descrizione "user-friendly"
di un metodo nel registro globale.
Questo decoratore viene eseguito all'avvio dell'app (quando i file
vengono importati) e popola il dizionario ACTION_DESCRIPTIONS.
Restituisce la funzione originale non modificata.
"""
def decorator(func):
# Registra l'azione
tool_name = func.__name__
if tool_name in ACTION_DESCRIPTIONS:
print(f"Attenzione: Azione '{tool_name}' registrata più volte.")
ACTION_DESCRIPTIONS[tool_name] = description
# Restituisce la funzione originale
return func
return decorator

10
src/app/agents/core.py
View File

@@ -146,12 +146,14 @@ class RunMessage:
- In esecuzione (➡️)
- Completato (✅)
Lo stato di esecuzione può essere assegnato solo ad uno step alla volta.
Lo stato di esecuzione può essere assegnato solo a uno step alla volta.
Args:
inputs (PipelineInputs): Input della pipeline per mostrare la configurazione.
prefix (str, optional): Prefisso del messaggio. Defaults to "".
suffix (str, optional): Suffisso del messaggio. Defaults to "".
inputs (PipelineInputs): Input della pipeline per mostrare la configurazione
prefix (str, optional): Prefisso del messaggio. Defaults to ""
suffix (str, optional): Suffisso del messaggio. Defaults to ""
"""
self.current = None
self.steps_total = None
self.base_message = f"Running configurations: \n{prefix}{inputs}{suffix}\n\n"
self.emojis = ['🔳', '➡️', '✅']
self.placeholder = '<<<>>>'

4
src/app/agents/pipeline.py
View File

@@ -125,8 +125,8 @@ class Pipeline:
"""
Esegue il workflow e restituisce gli eventi di stato e il risultato finale.
Args:
workflow: L'istanza di Workflow da eseguire.
query: Gli input della query.
workflow: L'istanza di Workflow da eseguire
query: Gli input della query
events: La lista di eventi e callback da gestire durante l'esecuzione.
Yields:
Aggiornamenti di stato e la risposta finale generata dal workflow.

15
src/app/api/tools/market_tool.py
View File

@@ -1,6 +1,6 @@
from agno.tools import Toolkit
from app.agents.action_registry import register_friendly_actions
from app.agents.action_registry import friendly_action
from app.api.wrapper_handler import WrapperHandler
from app.api.core.markets import MarketWrapper, Price, ProductInfo
from app.api.markets import BinanceWrapper, CoinBaseWrapper, CryptoCompareWrapper, YFinanceWrapper
@@ -40,6 +40,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
],
)
@friendly_action("🔍 Recupero le informazioni sul prodotto richiesto...")
def get_product(self, asset_id: str) -> ProductInfo:
"""
Gets product information for a *single* asset from the *first available* provider.
@@ -56,6 +57,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_product(asset_id))
@friendly_action("📦 Recupero i dati su più asset...")
def get_products(self, asset_ids: list[str]) -> list[ProductInfo]:
"""
Gets product information for a *list* of assets from the *first available* provider.
@@ -72,6 +74,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_products(asset_ids))
@friendly_action("📊 Recupero i dati storici dei prezzi...")
def get_historical_prices(self, asset_id: str, limit: int = 100) -> list[Price]:
"""
Gets historical price data for a *single* asset from the *first available* provider.
@@ -89,6 +92,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_historical_prices(asset_id, limit))
@friendly_action("🧩 Aggrego le informazioni da più fonti...")
def get_products_aggregated(self, asset_ids: list[str]) -> list[ProductInfo]:
"""
Gets product information for multiple assets from *all available providers* and *aggregates* the results.
@@ -109,6 +113,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
all_products = self.handler.try_call_all(lambda w: w.get_products(asset_ids))
return ProductInfo.aggregate(all_products)
@friendly_action("📈 Creo uno storico aggregato dei prezzi...")
def get_historical_prices_aggregated(self, asset_id: str = "BTC", limit: int = 100) -> list[Price]:
"""
Gets historical price data for a single asset from *all available providers* and *aggregates* the results.
@@ -129,11 +134,3 @@ class MarketAPIsTool(MarketWrapper, Toolkit):
"""
all_prices = self.handler.try_call_all(lambda w: w.get_historical_prices(asset_id, limit))
return Price.aggregate(all_prices)
register_friendly_actions({
"get_product": "🔍 Recupero le informazioni sul prodotto richiesto...",
"get_products": "📦 Recupero i dati su più asset...",
"get_historical_prices": "📊 Recupero i dati storici dei prezzi...",
"get_products_aggregated": "🧩 Aggrego le informazioni da più fonti...",
"get_historical_prices_aggregated": "📈 Creo uno storico aggregato dei prezzi...",
})

13
src/app/api/tools/news_tool.py
View File

@@ -1,6 +1,6 @@
from agno.tools import Toolkit
from app.agents.action_registry import register_friendly_actions
from app.agents.action_registry import friendly_action
from app.api.wrapper_handler import WrapperHandler
from app.api.core.news import NewsWrapper, Article
from app.api.news import NewsApiWrapper, GoogleNewsWrapper, CryptoPanicWrapper, DuckDuckGoWrapper
@@ -42,6 +42,7 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
],
)
@friendly_action("📰 Cerco le notizie principali...")
def get_top_headlines(self, limit: int = 100) -> list[Article]:
"""
Retrieves top headlines from the *first available* news provider.
@@ -58,6 +59,7 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_top_headlines(limit))
@friendly_action("🔎 Cerco notizie recenti sull'argomento...")
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
"""
Searches for the latest news on a specific topic from the *first available* provider.
@@ -75,6 +77,7 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_latest_news(query, limit))
@friendly_action("🗞️ Raccolgo le notizie principali da tutte le fonti...")
def get_top_headlines_aggregated(self, limit: int = 100) -> dict[str, list[Article]]:
"""
Retrieves top headlines from *all available providers* and aggregates the results.
@@ -94,6 +97,7 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
"""
return self.handler.try_call_all(lambda w: w.get_top_headlines(limit))
@friendly_action("📚 Raccolgo notizie specifiche da tutte le fonti...")
def get_latest_news_aggregated(self, query: str, limit: int = 100) -> dict[str, list[Article]]:
"""
Searches for news on a specific topic from *all available providers* and aggregates the results.
@@ -113,10 +117,3 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
Exception: If all providers fail to return results.
"""
return self.handler.try_call_all(lambda w: w.get_latest_news(query, limit))
register_friendly_actions({
"get_top_headlines": "📰 Cerco le notizie principali...",
"get_latest_news": "🔎 Cerco notizie recenti su un argomento...",
"get_top_headlines_aggregated": "🗞️ Raccolgo le notizie principali da tutte le fonti...",
"get_latest_news_aggregated": "📚 Raccolgo notizie specifiche da tutte le fonti...",
})

9
src/app/api/tools/social_tool.py
View File

@@ -1,6 +1,6 @@
from agno.tools import Toolkit
from app.agents.action_registry import register_friendly_actions
from app.agents.action_registry import friendly_action
from app.api.wrapper_handler import WrapperHandler
from app.api.core.social import SocialPost, SocialWrapper
from app.api.social import *
@@ -41,6 +41,7 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
],
)
@friendly_action("📱 Cerco i post più popolari sui social...")
def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]:
"""
Retrieves top cryptocurrency-related posts from the *first available* social media provider.
@@ -57,6 +58,7 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
"""
return self.handler.try_call(lambda w: w.get_top_crypto_posts(limit))
@friendly_action("🌐 Raccolgo i post da tutte le piattaforme social...")
def get_top_crypto_posts_aggregated(self, limit_per_wrapper: int = 5) -> dict[str, list[SocialPost]]:
"""
Retrieves top cryptocurrency-related posts from *all available providers* and aggregates the results.
@@ -75,8 +77,3 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
Exception: If all providers fail to return results.
"""
return self.handler.try_call_all(lambda w: w.get_top_crypto_posts(limit_per_wrapper))
register_friendly_actions({
"get_top_crypto_posts": "📱 Cerco i post più popolari sui social...",
"get_top_crypto_posts_aggregated": "🌐 Raccolgo i post da tutte le piattaforme social...",
})

2
src/app/interface/chat.py
View File

@@ -58,7 +58,7 @@ class ChatManager:
"""
self.inputs.user_query = message
pipeline = Pipeline(self.inputs)
listeners: list[tuple[PipelineEvent, Callable[[Any], str | None]]] = [
listeners: list[tuple[PipelineEvent, Callable[[Any], str | None]]] = [ # type: ignore
(PipelineEvent.QUERY_CHECK, lambda _: "🔍 Sto controllando la tua richiesta..."),
(PipelineEvent.INFO_RECOVERY, lambda _: "📊 Sto recuperando i dati (mercato, news, social)..."),
(PipelineEvent.REPORT_GENERATION, lambda _: "✍️ Sto scrivendo il report finale..."),