diff --git a/src/app/api/tools/instructions/__init__.py b/src/app/api/tools/instructions/__init__.py new file mode 100644 index 0000000..6ed0d5a --- /dev/null +++ b/src/app/api/tools/instructions/__init__.py @@ -0,0 +1,21 @@ +from pathlib import Path + +__INSTRUCTIONS_PATH = Path(__file__).parent + +def __load_tool_instruction(file_name: str) -> str: + file_path = __INSTRUCTIONS_PATH / file_name + return file_path.read_text(encoding='utf-8').strip() + +MARKET_TOOL_INSTRUCTIONS = __load_tool_instruction("market_instructions.md") +NEWS_TOOL_INSTRUCTIONS = __load_tool_instruction("news_instructions.md") +SOCIAL_TOOL_INSTRUCTIONS = __load_tool_instruction("social_instructions.md") +PLAN_MEMORY_TOOL_INSTRUCTIONS = __load_tool_instruction("plan_memory_instructions.md") +SYMBOLS_TOOL_INSTRUCTIONS = __load_tool_instruction("symbols_instructions.md") + +__all__ = [ + "MARKET_TOOL_INSTRUCTIONS", + "NEWS_TOOL_INSTRUCTIONS", + "SOCIAL_TOOL_INSTRUCTIONS", + "PLAN_MEMORY_TOOL_INSTRUCTIONS", + "SYMBOLS_TOOL_INSTRUCTIONS", +] \ No newline at end of file diff --git a/src/app/api/tools/market_tool.py b/src/app/api/tools/market_tool.py index 72c3973..5aa11b1 100644 --- a/src/app/api/tools/market_tool.py +++ b/src/app/api/tools/market_tool.py @@ -1,5 +1,5 @@ -from pathlib import Path from agno.tools import Toolkit +from app.api.tools.instructions import MARKET_TOOL_INSTRUCTIONS 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 @@ -13,17 +13,6 @@ class MarketAPIsTool(MarketWrapper, Toolkit): Providers can be configured in configs.yaml under api.market_providers. """ - @staticmethod - def _load_instructions() -> str: - """ - Load the toolkit instructions from the external text file. - - Returns: - str: The content of the instructions file. - """ - instructions_path = Path(__file__).parent / "instructions" / "market_instructions.md" - return instructions_path.read_text(encoding="utf-8") - def __init__(self): """ Initialize the MarketAPIsTool with market API wrappers configured in configs.yaml. @@ -41,6 +30,7 @@ class MarketAPIsTool(MarketWrapper, Toolkit): Toolkit.__init__( # type: ignore self, name="Market APIs Toolkit", + instructions=MARKET_TOOL_INSTRUCTIONS, tools=[ self.get_product, self.get_products, @@ -48,7 +38,6 @@ class MarketAPIsTool(MarketWrapper, Toolkit): self.get_products_aggregated, self.get_historical_prices_aggregated, ], - instructions=self._load_instructions(), ) def get_product(self, asset_id: str) -> ProductInfo: diff --git a/src/app/api/tools/news_tool.py b/src/app/api/tools/news_tool.py index 2d53244..1f834c5 100644 --- a/src/app/api/tools/news_tool.py +++ b/src/app/api/tools/news_tool.py @@ -1,5 +1,5 @@ -from pathlib import Path from agno.tools import Toolkit +from app.api.tools.instructions import NEWS_TOOL_INSTRUCTIONS from app.api.wrapper_handler import WrapperHandler from app.api.core.news import NewsWrapper, Article from app.api.news import NewsApiWrapper, GoogleNewsWrapper, CryptoPanicWrapper, DuckDuckGoWrapper @@ -16,17 +16,6 @@ class NewsAPIsTool(NewsWrapper, Toolkit): If no wrapper succeeds, an exception is raised. """ - @staticmethod - def _load_instructions() -> str: - """ - Load the toolkit instructions from the external text file. - - Returns: - str: The content of the instructions file. - """ - instructions_path = Path(__file__).parent / "instructions" / "news_instructions.md" - return instructions_path.read_text(encoding="utf-8") - def __init__(self): """ Initialize the NewsAPIsTool with news API wrappers configured in configs.yaml. @@ -44,13 +33,13 @@ class NewsAPIsTool(NewsWrapper, Toolkit): Toolkit.__init__( # type: ignore self, name="News APIs Toolkit", + instructions=NEWS_TOOL_INSTRUCTIONS, tools=[ self.get_top_headlines, self.get_latest_news, self.get_top_headlines_aggregated, self.get_latest_news_aggregated, ], - instructions=self._load_instructions(), ) def get_top_headlines(self, limit: int = 100) -> list[Article]: diff --git a/src/app/api/tools/plan_memory_tool.py b/src/app/api/tools/plan_memory_tool.py index 2c0ca72..646d118 100644 --- a/src/app/api/tools/plan_memory_tool.py +++ b/src/app/api/tools/plan_memory_tool.py @@ -1,6 +1,6 @@ -from pathlib import Path from agno.tools.toolkit import Toolkit from typing import TypedDict, Literal +from app.api.tools.instructions import PLAN_MEMORY_TOOL_INSTRUCTIONS @@ -11,21 +11,12 @@ class Task(TypedDict): class PlanMemoryTool(Toolkit): - @staticmethod - def _load_instructions() -> str: - """ - Load the toolkit instructions from the external markdown file. - - Returns: - str: The content of the instructions file. - """ - instructions_path = Path(__file__).parent / "instructions" / "plan_memory_instructions.md" - return instructions_path.read_text(encoding="utf-8") def __init__(self): self.tasks: list[Task] = [] Toolkit.__init__(self, # type: ignore[call-arg] - instructions=self._load_instructions(), + name="Plan Memory Toolkit", + instructions=PLAN_MEMORY_TOOL_INSTRUCTIONS, tools=[ self.add_tasks, self.get_next_pending_task, diff --git a/src/app/api/tools/social_tool.py b/src/app/api/tools/social_tool.py index ef367cc..5d78c12 100644 --- a/src/app/api/tools/social_tool.py +++ b/src/app/api/tools/social_tool.py @@ -1,5 +1,5 @@ -from pathlib import Path from agno.tools import Toolkit +from app.api.tools.instructions import SOCIAL_TOOL_INSTRUCTIONS from app.api.wrapper_handler import WrapperHandler from app.api.core.social import SocialPost, SocialWrapper from app.api.social import * @@ -17,17 +17,6 @@ class SocialAPIsTool(SocialWrapper, Toolkit): If no wrapper succeeds, an exception is raised. """ - @staticmethod - def _load_instructions() -> str: - """ - Load the toolkit instructions from the external text file. - - Returns: - str: The content of the instructions file. - """ - instructions_path = Path(__file__).parent / "instructions" / "social_instructions.md" - return instructions_path.read_text(encoding="utf-8") - def __init__(self): """ Initialize the SocialAPIsTool with social media API wrappers configured in configs.yaml. @@ -44,12 +33,12 @@ class SocialAPIsTool(SocialWrapper, Toolkit): Toolkit.__init__( # type: ignore self, - name="Socials Toolkit", + name="Socials APIs Toolkit", + instructions=SOCIAL_TOOL_INSTRUCTIONS, tools=[ self.get_top_crypto_posts, self.get_top_crypto_posts_aggregated, ], - instructions=self._load_instructions(), ) def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]: diff --git a/src/app/api/tools/symbols_tool.py b/src/app/api/tools/symbols_tool.py index 3a44fbc..b49d879 100644 --- a/src/app/api/tools/symbols_tool.py +++ b/src/app/api/tools/symbols_tool.py @@ -4,8 +4,8 @@ import asyncio import logging import pandas as pd from io import StringIO -from pathlib import Path from agno.tools.toolkit import Toolkit +from app.api.tools.instructions import SYMBOLS_TOOL_INSTRUCTIONS logging.basicConfig(level=logging.INFO) logging = logging.getLogger("crypto_symbols") @@ -19,23 +19,12 @@ class CryptoSymbolsTools(Toolkit): Classe per ottenere i simboli delle criptovalute tramite Yahoo Finance. """ - @staticmethod - def _load_instructions() -> str: - """ - Load the toolkit instructions from the external markdown file. - - Returns: - str: The content of the instructions file. - """ - instructions_path = Path(__file__).parent / "instructions" / "symbols_instructions.md" - return instructions_path.read_text(encoding="utf-8") - def __init__(self, cache_file: str = 'resources/cryptos.csv'): self.cache_file = cache_file self.final_table = pd.read_csv(self.cache_file) if os.path.exists(self.cache_file) else pd.DataFrame() # type: ignore Toolkit.__init__(self, # type: ignore name="Crypto Symbols Tool", - instructions=self._load_instructions(), + instructions=SYMBOLS_TOOL_INSTRUCTIONS, tools=[ self.get_all_symbols, self.get_symbols_by_name,