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

Team Workflow aggiornato (#37)

Browse Source 
* Rimuovi la classe Predictor e aggiorna le importazioni in Pipeline e __init__.py
* Aggiungi modelli per l'analisi delle query e la generazione di report; aggiorna le configurazioni degli agenti
* Tests for report generation and team agent responses
* Aggiorna i prompt degli agenti
* Changed defaults models
* Aggiunta della classe PlanMemoryTool per la gestione dei task e aggiornamento della logica del team leader per un'esecuzione più dinamica del piano.
This commit was merged in pull request #37.
This commit is contained in:
Giacomo Bertolazzi
2025-10-20 22:05:58 +02:00
committed by GitHub
parent 885a70d748
commit 12339ccbff
19 changed files with 539 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
configs.yaml
View File

@@ -41,5 +41,6 @@ api:
agents:
strategy: Conservative
team_model: qwen3:1.7b
team_leader_model: qwen3:4b
predictor_model: qwen3:4b
team_leader_model: qwen3:8b
query_analyzer_model: qwen3:4b
report_generation_model: qwen3:8b

6
src/app/agents/__init__.py
View File

@@ -1,4 +1,4 @@
from app.agents.predictor import PredictorInput, PredictorOutput
from app.agents.pipeline import Pipeline, PipelineInputs, PipelineEvent
from app.agents.pipeline import Pipeline, PipelineEvent
from app.agents.core import PipelineInputs, QueryOutputs
__all__ = ["PredictorInput", "PredictorOutput", "Pipeline", "PipelineInputs", "PipelineEvent"]
__all__ = ["Pipeline", "PipelineInputs", "PipelineEvent", "QueryOutputs"]

121
src/app/agents/core.py Normal file
View File

@@ -0,0 +1,121 @@
from pydantic import BaseModel
from agno.agent import Agent
from agno.team import Team
from agno.tools.reasoning import ReasoningTools
from app.agents.plan_memory_tool import PlanMemoryTool
from app.api.tools import *
from app.configs import AppConfig
from app.agents.prompts import *
class QueryInputs(BaseModel):
user_query: str
strategy: str
class QueryOutputs(BaseModel):
response: str
is_crypto: bool
class PipelineInputs:
"""
Classe necessaria per passare gli input alla Pipeline.
Serve per raggruppare i parametri e semplificare l'inizializzazione.
"""
def __init__(self, configs: AppConfig | None = None) -> None:
"""
Inputs per la Pipeline di agenti.
Setta i valori di default se non specificati.
"""
self.configs = configs if configs else AppConfig()
agents = self.configs.agents
self.team_model = self.configs.get_model_by_name(agents.team_model)
self.team_leader_model = self.configs.get_model_by_name(agents.team_leader_model)
self.query_analyzer_model = self.configs.get_model_by_name(agents.query_analyzer_model)
self.report_generation_model = self.configs.get_model_by_name(agents.report_generation_model)
self.strategy = self.configs.get_strategy_by_name(agents.strategy)
self.user_query = ""
# ======================
# Dropdown handlers
# ======================
def choose_team_leader(self, index: int):
"""
Sceglie il modello LLM da usare per il Team Leader.
"""
assert index >= 0 and index < len(self.configs.models.all_models), "Index out of range for models list."
self.team_leader_model = self.configs.models.all_models[index]
def choose_team(self, index: int):
"""
Sceglie il modello LLM da usare per il Team.
"""
assert index >= 0 and index < len(self.configs.models.all_models), "Index out of range for models list."
self.team_model = self.configs.models.all_models[index]
def choose_strategy(self, index: int):
"""
Sceglie la strategia da usare per il Team.
"""
self.strategy = self.configs.strategies[index]
# ======================
# Helpers
# ======================
def list_models_names(self) -> list[str]:
"""
Restituisce la lista dei nomi dei modelli disponibili.
"""
return [model.label for model in self.configs.models.all_models]
def list_strategies_names(self) -> list[str]:
"""
Restituisce la lista delle strategie disponibili.
"""
return [strat.label for strat in self.configs.strategies]
def get_query_inputs(self) -> QueryInputs:
"""
Restituisce gli input per l'agente di verifica della query.
"""
return QueryInputs(
user_query=self.user_query,
strategy=self.strategy.label,
)
# ======================
# Agent getters
# ======================
def get_agent_team(self) -> Team:
market, news, social = self.get_tools()
market_agent = self.team_model.get_agent(MARKET_INSTRUCTIONS, "Market Agent", tools=[market])
news_agent = self.team_model.get_agent(NEWS_INSTRUCTIONS, "News Agent", tools=[news])
social_agent = self.team_model.get_agent(SOCIAL_INSTRUCTIONS, "Socials Agent", tools=[social])
return Team(
model=self.team_leader_model.get_model(TEAM_LEADER_INSTRUCTIONS),
name="CryptoAnalysisTeam",
tools=[ReasoningTools(), PlanMemoryTool()],
members=[market_agent, news_agent, social_agent],
)
def get_agent_query_checker(self) -> Agent:
return self.query_analyzer_model.get_agent(QUERY_CHECK_INSTRUCTIONS, "Query Check Agent", output_schema=QueryOutputs)
def get_agent_report_generator(self) -> Agent:
return self.report_generation_model.get_agent(REPORT_GENERATION_INSTRUCTIONS, "Report Generator Agent")
def get_tools(self) -> tuple[MarketAPIsTool, NewsAPIsTool, SocialAPIsTool]:
"""
Restituisce la lista di tools disponibili per gli agenti.
"""
api = self.configs.api
market_tool = MarketAPIsTool(currency=api.currency)
market_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
news_tool = NewsAPIsTool()
news_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
social_tool = SocialAPIsTool()
social_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
return market_tool, news_tool, social_tool

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

@@ -4,85 +4,38 @@ import logging
import random
from typing import Any, Callable
from agno.agent import RunEvent
from agno.team import Team, TeamRunEvent
from agno.tools.reasoning import ReasoningTools
from agno.run.workflow import WorkflowRunEvent
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
from app.api.tools import *
from app.agents.prompts import *
from app.configs import AppConfig
from app.agents.core import *
logging = logging.getLogger("pipeline")
class PipelineEvent(str, Enum):
PLANNER = "Planner"
QUERY_CHECK = "Query Check"
QUERY_ANALYZER = "Query Analyzer"
INFO_RECOVERY = "Info Recovery"
REPORT_GENERATION = "Report Generation"
REPORT_TRANSLATION = "Report Translation"
TOOL_USED = RunEvent.tool_call_completed
RUN_FINISHED = WorkflowRunEvent.workflow_completed.value
TOOL_USED = RunEvent.tool_call_completed.value
def check_event(self, event: str, step_name: str) -> bool:
return event == self.value or (WorkflowRunEvent.step_completed and step_name == self.value)
return event == self.value or (WorkflowRunEvent.step_completed == event and step_name == self.value)
class PipelineInputs:
"""
Classe necessaria per passare gli input alla Pipeline.
Serve per raggruppare i parametri e semplificare l'inizializzazione.
"""
def __init__(self, configs: AppConfig | None = None) -> None:
"""
Inputs per la Pipeline di agenti.
Setta i valori di default se non specificati.
"""
self.configs = configs if configs else AppConfig()
agents = self.configs.agents
self.team_model = self.configs.get_model_by_name(agents.team_model)
self.team_leader_model = self.configs.get_model_by_name(agents.team_leader_model)
self.predictor_model = self.configs.get_model_by_name(agents.predictor_model)
self.strategy = self.configs.get_strategy_by_name(agents.strategy)
self.user_query = ""
# ======================
# Dropdown handlers
# ======================
def choose_team_leader(self, index: int):
"""
Sceglie il modello LLM da usare per il Team Leader.
"""
self.leader_model = self.configs.models.all_models[index]
def choose_team(self, index: int):
"""
Sceglie il modello LLM da usare per il Team.
"""
self.team_model = self.configs.models.all_models[index]
def choose_strategy(self, index: int):
"""
Sceglie la strategia da usare per il Team.
"""
self.strategy = self.configs.strategies[index]
# ======================
# Helpers
# ======================
def list_models_names(self) -> list[str]:
"""
Restituisce la lista dei nomi dei modelli disponibili.
"""
return [model.label for model in self.configs.models.all_models]
def list_strategies_names(self) -> list[str]:
"""
Restituisce la lista delle strategie disponibili.
"""
return [strat.label for strat in self.configs.strategies]
@classmethod
def get_log_events(cls, run_id: int) -> list[tuple['PipelineEvent', Callable[[Any], None]]]:
return [
(PipelineEvent.QUERY_CHECK, lambda _: logging.info(f"[{run_id}] Query Check completed.")),
(PipelineEvent.QUERY_ANALYZER, lambda _: logging.info(f"[{run_id}] Query Analyzer completed.")),
(PipelineEvent.INFO_RECOVERY, lambda _: logging.info(f"[{run_id}] Info Recovery completed.")),
(PipelineEvent.REPORT_GENERATION, lambda _: logging.info(f"[{run_id}] Report Generation completed.")),
(PipelineEvent.TOOL_USED, lambda e: logging.info(f"[{run_id}] Tool used [{e.tool.tool_name}] by {e.agent_name}.")),
(PipelineEvent.RUN_FINISHED, lambda _: logging.info(f"[{run_id}] Run completed.")),
]
class Pipeline:
@@ -93,12 +46,14 @@ class Pipeline:
"""
def __init__(self, inputs: PipelineInputs):
"""
Inizializza la pipeline con gli input forniti.
Args:
inputs: istanza di PipelineInputs contenente le configurazioni e i parametri della pipeline.
"""
self.inputs = inputs
# ======================
# Core interaction
# ======================
def interact(self, listeners: dict[RunEvent | TeamRunEvent, Callable[[PipelineEvent], None]] = {}) -> str:
def interact(self, listeners: list[tuple[PipelineEvent, Callable[[Any], None]]] = []) -> str:
"""
Esegue la pipeline di agenti per rispondere alla query dell'utente.
Args:
@@ -108,7 +63,7 @@ class Pipeline:
"""
return asyncio.run(self.interact_async(listeners))
async def interact_async(self, listeners: dict[RunEvent | TeamRunEvent, Callable[[PipelineEvent], None]] = {}) -> str:
async def interact_async(self, listeners: list[tuple[PipelineEvent, Callable[[Any], None]]] = []) -> str:
"""
Versione asincrona che esegue la pipeline di agenti per rispondere alla query dell'utente.
Args:
@@ -119,61 +74,47 @@ class Pipeline:
run_id = random.randint(1000, 9999) # Per tracciare i log
logging.info(f"[{run_id}] Pipeline query: {self.inputs.user_query}")
# Step 1: Crea gli agenti e il team
market_tool, news_tool, social_tool = self.get_tools()
market_agent = self.inputs.team_model.get_agent(instructions=MARKET_INSTRUCTIONS, name="MarketAgent", tools=[market_tool])
news_agent = self.inputs.team_model.get_agent(instructions=NEWS_INSTRUCTIONS, name="NewsAgent", tools=[news_tool])
social_agent = self.inputs.team_model.get_agent(instructions=SOCIAL_INSTRUCTIONS, name="SocialAgent", tools=[social_tool])
team = Team(
model=self.inputs.team_leader_model.get_model(COORDINATOR_INSTRUCTIONS),
name="CryptoAnalysisTeam",
tools=[ReasoningTools()],
members=[market_agent, news_agent, social_agent],
events = [*PipelineEvent.get_log_events(run_id), *listeners]
query = QueryInputs(
user_query=self.inputs.user_query,
strategy=self.inputs.strategy.description
)
# Step 3: Crea il workflow
#query_planner = Step(name=PipelineEvent.PLANNER, agent=Agent())
info_recovery = Step(name=PipelineEvent.INFO_RECOVERY, team=team)
#report_generation = Step(name=PipelineEvent.REPORT_GENERATION, agent=Agent())
#report_translate = Step(name=AppEvent.REPORT_TRANSLATION, agent=Agent())
workflow = Workflow(
name="App Workflow",
steps=[
#query_planner,
info_recovery,
#report_generation,
#report_translate
]
)
# Step 4: Fai partire il workflow e prendi l'output
query = f"The user query is: {self.inputs.user_query}\n\n They requested a {self.inputs.strategy.label} investment strategy."
result = await self.run(workflow, query, events={})
logging.info(f"[{run_id}] Run finished")
workflow = self.build_workflow()
result = await self.run(workflow, query, events=events)
return result
# ======================
# Helpers
# =====================
def get_tools(self) -> tuple[MarketAPIsTool, NewsAPIsTool, SocialAPIsTool]:
"""
Restituisce la lista di tools disponibili per gli agenti.
"""
api = self.inputs.configs.api
market_tool = MarketAPIsTool(currency=api.currency)
market_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
news_tool = NewsAPIsTool()
news_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
social_tool = SocialAPIsTool()
social_tool.handler.set_retries(api.retry_attempts, api.retry_delay_seconds)
def build_workflow(self) -> Workflow:
"""
Costruisce il workflow della pipeline di agenti.
Returns:
L'istanza di Workflow costruita.
"""
# Step 1: Crea gli agenti e il team
team = self.inputs.get_agent_team()
query_check = self.inputs.get_agent_query_checker()
report = self.inputs.get_agent_report_generator()
return (market_tool, news_tool, social_tool)
# Step 2: Crea gli steps
def condition_query_ok(step_input: StepInput) -> StepOutput:
val = step_input.previous_step_content
return StepOutput(stop=not val.is_crypto) if isinstance(val, QueryOutputs) else StepOutput(stop=True)
query_check = Step(name=PipelineEvent.QUERY_CHECK, agent=query_check)
info_recovery = Step(name=PipelineEvent.INFO_RECOVERY, team=team)
report_generation = Step(name=PipelineEvent.REPORT_GENERATION, agent=report)
# Step 3: Ritorna il workflow completo
return Workflow(name="App Workflow", steps=[
query_check,
condition_query_ok,
info_recovery,
report_generation
])
@classmethod
async def run(cls, workflow: Workflow, query: str, events: dict[PipelineEvent, Callable[[Any], None]]) -> str:
async def run(cls, workflow: Workflow, query: QueryInputs, events: list[tuple[PipelineEvent, Callable[[Any], None]]]) -> str:
"""
Esegue il workflow e gestisce gli eventi tramite le callback fornite.
Args:
@@ -188,16 +129,18 @@ class Pipeline:
content = None
async for event in iterator:
step_name = getattr(event, 'step_name', '')
for app_event, listener in events.items():
for app_event, listener in events:
if app_event.check_event(event.event, step_name):
listener(event)
if event.event == WorkflowRunEvent.workflow_completed:
if event.event == WorkflowRunEvent.step_completed:
content = getattr(event, 'content', '')
if isinstance(content, str):
think_str = "</think>"
think = content.rfind(think_str)
content = content[(think + len(think_str)):] if think != -1 else content
return content if content else "No output from workflow, something went wrong."
if content and isinstance(content, str):
think_str = "</think>"
think = content.rfind(think_str)
return content[(think + len(think_str)):] if think != -1 else content
if content and isinstance(content, QueryOutputs):
return content.response
logging.error(f"No output from workflow: {content}")
return "No output from workflow, something went wrong."

55
src/app/agents/plan_memory_tool.py Normal file
View File

@@ -0,0 +1,55 @@
from agno.tools.toolkit import Toolkit
from typing import TypedDict, Literal
class Task(TypedDict):
name: str
status: Literal["pending", "completed", "failed"]
result: str | None
class PlanMemoryTool(Toolkit):
def __init__(self):
self.tasks: list[Task] = []
Toolkit.__init__(self, # type: ignore[call-arg]
instructions="This tool manages an execution plan. Add tasks, get the next pending task, update a task's status (completed, failed) and result, or list all tasks.",
tools=[
self.add_tasks,
self.get_next_pending_task,
self.update_task_status,
self.list_all_tasks,
]
)
def add_tasks(self, task_names: list[str]) -> str:
"""Adds multiple new tasks to the plan with 'pending' status."""
count = 0
for name in task_names:
if not any(t['name'] == name for t in self.tasks):
self.tasks.append({"name": name, "status": "pending", "result": None})
count += 1
return f"Added {count} new tasks."
def get_next_pending_task(self) -> Task | None:
"""Retrieves the first task that is still 'pending'."""
for task in self.tasks:
if task["status"] == "pending":
return task
return None
def update_task_status(self, task_name: str, status: Literal["completed", "failed"], result: str | None = None) -> str:
"""Updates the status and result of a specific task by its name."""
for task in self.tasks:
if task["name"] == task_name:
task["status"] = status
if result is not None:
task["result"] = result
return f"Task '{task_name}' updated to {status}."
return f"Error: Task '{task_name}' not found."
def list_all_tasks(self) -> list[str]:
"""Lists all tasks in the plan with their status and result."""
if not self.tasks:
return ["No tasks in the plan."]
return [f"- {t['name']}: {t['status']} (Result: {t.get('result', 'N/A')})" for t in self.tasks]

16
src/app/agents/predictor.py
View File

@@ -1,16 +0,0 @@
from pydantic import BaseModel, Field
from app.api.core.markets import ProductInfo
class PredictorInput(BaseModel):
data: list[ProductInfo] = Field(..., description="Market data as a list of ProductInfo")
style: str = Field(..., description="Prediction style")
sentiment: str = Field(..., description="Aggregated sentiment from news and social analysis")
class ItemPortfolio(BaseModel):
asset: str = Field(..., description="Name of the asset")
percentage: float = Field(..., description="Percentage allocation to the asset")
motivation: str = Field(..., description="Motivation for the allocation")
class PredictorOutput(BaseModel):
strategy: str = Field(..., description="Concise operational strategy in Italian")
portfolio: list[ItemPortfolio] = Field(..., description="List of portfolio items with allocations")

10
src/app/agents/prompts/__init__.py
View File

@@ -6,16 +6,18 @@ def __load_prompt(file_name: str) -> str:
file_path = __PROMPTS_PATH / file_name
return file_path.read_text(encoding='utf-8').strip()
COORDINATOR_INSTRUCTIONS = __load_prompt("team_leader.txt")
TEAM_LEADER_INSTRUCTIONS = __load_prompt("team_leader.txt")
MARKET_INSTRUCTIONS = __load_prompt("team_market.txt")
NEWS_INSTRUCTIONS = __load_prompt("team_news.txt")
SOCIAL_INSTRUCTIONS = __load_prompt("team_social.txt")
PREDICTOR_INSTRUCTIONS = __load_prompt("predictor.txt")
QUERY_CHECK_INSTRUCTIONS = __load_prompt("query_check.txt")
REPORT_GENERATION_INSTRUCTIONS = __load_prompt("report_generation.txt")
__all__ = [
"COORDINATOR_INSTRUCTIONS",
"TEAM_LEADER_INSTRUCTIONS",
"MARKET_INSTRUCTIONS",
"NEWS_INSTRUCTIONS",
"SOCIAL_INSTRUCTIONS",
"PREDICTOR_INSTRUCTIONS",
"QUERY_CHECK_INSTRUCTIONS",
"REPORT_GENERATION_INSTRUCTIONS",
]

27
src/app/agents/prompts/predictor.txt
View File

@@ -1,27 +0,0 @@
You are an **Allocation Algorithm (Crypto-Algo)** specialized in analyzing market data and sentiment to generate an investment strategy and a target portfolio.
Your sole objective is to process the user_input data and generate the strictly structured output as required by the response format. **You MUST NOT provide introductions, preambles, explanations, conclusions, or any additional comments that are not strictly required.**
## Processing Instructions (Absolute Rule)
The allocation strategy must be **derived exclusively from the "Allocation Logic" corresponding to the requested *style*** and the provided market/sentiment data. **DO NOT** use external or historical knowledge.
## Allocation Logic
### "Aggressivo" Style (Aggressive)
* **Priority:** Maximizing return (high volatility accepted).
* **Focus:** Higher allocation to **non-BTC/ETH assets** with high momentum potential (Altcoins, mid/low-cap assets).
* **BTC/ETH:** Must serve as a base (anchor), but their allocation **must not exceed 50%** of the total portfolio.
* **Sentiment:** Use positive sentiment to increase exposure to high-risk assets.
### "Conservativo" Style (Conservative)
* **Priority:** Capital preservation (volatility minimized).
* **Focus:** Major allocation to **BTC and/or ETH (Large-Cap Assets)**.
* **BTC/ETH:** Their allocation **must be at least 70%** of the total portfolio.
* **Altcoins:** Any allocations to non-BTC/ETH assets must be minimal (max 30% combined) and for assets that minimize speculative risk.
* **Sentiment:** Use positive sentiment only as confirmation for exposure, avoiding reactions to excessive "FOMO" signals.
## Output Requirements (Content MUST be in Italian)
1. **Strategy (strategy):** Must be a concise operational description **in Italian ("in Italiano")**, with a maximum of 5 sentences.
2. **Portfolio (portfolio):** The sum of all percentages must be **exactly 100%**. The justification (motivation) for each asset must be a single clear sentence **in Italian ("in Italiano")**.

18
src/app/agents/prompts/query_check.txt Normal file
View File

@@ -0,0 +1,18 @@
GOAL: check if the query is crypto-related
1) Determine the language of the query:
- This will help you understand better the intention of the user
- Focus on the query of the user
- DO NOT answer the query
2) Determine if the query is crypto or investment-related:
- Crypto-related if it mentions cryptocurrencies, tokens, NFTs, blockchain, exchanges, wallets, DeFi, oracles, smart contracts, on-chain, off-chain, staking, yield, liquidity, tokenomics, coins, ticker symbols, etc.
- Investment-related if it mentions stocks, bonds, options, trading strategies, financial markets, investment advice, portfolio management, etc.
- If the query uses generic terms like "news", "prices", "trends", "social", "market cap", "volume" with NO asset specified -> ASSUME CRYPTO/INVESTMENT CONTEXT and proceed.
- If the query is clearly about unrelated domains (weather, recipes, unrelated local politics, unrelated medicine, general software not about crypto, etc.) -> return NOT_CRYPTO error.
- If ambiguous: treat as crypto/investment only if the most likely intent is crypto/investment; otherwise return a JSON plan that first asks the user for clarification (see step structure below).
3) Ouput the result:
- if is crypto related then output the query
- if is not crypto related, then output why is not related in a brief message

61
src/app/agents/prompts/report_generation.txt Normal file
View File

@@ -0,0 +1,61 @@
**TASK:** You are a specialized **Markdown Reporting Assistant**. Your task is to receive a structured analysis report from a "Team Leader" and re-format it into a single, cohesive, and well-structured final report in Markdown for the end-user.
**INPUT:** The input will be a structured block containing an `Overall Summary` and *zero or more* data sections (e.g., `Market`, `News`, `Social`, `Assumptions`). Each section will contain a `Summary` and `Full Data`.
**CORE RULES:**
1. **Strict Conditional Rendering (CRUCIAL):** Your primary job is to format *only* the data you receive. You MUST check each data section from the input (e.g., `Market & Price Data`, `News & Market Sentiment`).
2. **Omit Empty Sections (CRUCIAL):** If a data section is **not present** in the input, or if its `Full Data` field is empty, null, or marked as 'Data not available', you **MUST** completely omit that entire section from the final report. **DO NOT** print the Markdown header (e.g., `## 1. Market & Price Data`), the summary, or any placeholder text for that missing section.
3. **Omit Report Notes:** This same rule applies to the `## 4. Report Notes` section. Render it *only* if an `Assumptions` or `Execution Log` field is present in the input.
4. **Present All Data:** For sections that *are* present and contain data, your report's text MUST be based on the `Summary` provided, and you MUST include the `Full Data` (e.g., Markdown tables for prices).
5. **Do Not Invent:**
* **Do NOT** invent new hypotheses, metrics, or conclusions.
* **Do NOT** print internal field names (like 'Full Data') or agent names.
6. **No Extraneous Output:**
* Your entire response must be **only the Markdown report**.
* Do not include any pre-amble (e.g., "Here is the report:").
---
**MANDATORY REPORT STRUCTURE:**
(Follow the CORE RULES to conditionally render these sections. If no data sections are present, you will only render the Title and Executive Summary.)
# [Report Title - e.g., "Crypto Analysis Report: Bitcoin"]
## Executive Summary
[Use the `Overall Summary` from the input here.]
---
## 1. Market & Price Data
[Use the `Summary` from the input's Market section here.]
**Detailed Price Data:**
[Present the `Full Data` from the Market section here.]
---
## 2. News & Market Sentiment
[Use the `Summary` from the input's News section here.]
**Key Topics Discussed:**
[List the main topics identified in the News summary.]
**Supporting News/Data:**
[Present the `Full Data` from the News section here.]
---
## 3. Social Sentiment
[Use the `Summary` from the input's Social section here.]
**Trending Narratives:**
[List the main narratives identified in the Social summary.]
**Supporting Social/Data:**
[Present the `Full Data` from the Social section here.]
---
## 4. Report Notes
[Use this section to report any `Assumptions` or `Execution Log` data provided in the input.]

57
src/app/agents/prompts/team_leader.txt
View File

@@ -1,15 +1,48 @@
You are the expert coordinator of a financial analysis team specializing in cryptocurrencies.
**TASK:** You are the **Crypto Analysis Team Leader**, an expert coordinator of a financial analysis team.
Your team consists of three agents:
- **MarketAgent**: Provides quantitative market data, price analysis, and technical indicators.
- **NewsAgent**: Scans and analyzes the latest news, articles, and official announcements.
- **SocialAgent**: Gauges public sentiment, trends, and discussions on social media.
**INPUT:** You will receive a user query. Your role is to create and execute an adaptive plan by coordinating your team of agents to retrieve data, judge its sufficiency, and provide an aggregated analysis.
Your primary objective is to answer the user's query by orchestrating the work of your team members.
**YOUR TEAM CONSISTS OF THREE AGENTS:**
- **MarketAgent:** Fetches live prices and historical data.
- **NewsAgent:** Analyzes news sentiment and top topics.
- **SocialAgent:** Gauges public sentiment and trending narratives.
Your workflow is as follows:
1. **Deconstruct the user's query** to identify the required information.
2. **Delegate specific tasks** to the most appropriate agent(s) to gather the necessary data and initial analysis.
3. **Analyze the information** returned by the agents.
4. If the initial data is insufficient or the query is complex, **iteratively re-engage the agents** with follow-up questions to build a comprehensive picture.
5. **Synthesize all the gathered information** into a final, coherent, and complete analysis that fills all the required output fields.
**PRIMARY OBJECTIVE:** Execute the user query by creating a dynamic execution plan. You must **use your available tools to manage the plan's state**, identify missing data, orchestrate agents to retrieve it, manage retrieval attempts, and judge sufficiency. The final goal is to produce a structured report including *all* retrieved data and an analytical summary for the final formatting LLM.
**WORKFLOW (Execution Logic):**
1. **Analyze Query & Scope Plan:** Analyze the user's query. Create an execution plan identifying the *target data* needed. The plan's scope *must* be determined by the **Query Scoping** rule (see RULES): `focused` (for simple queries) or `comprehensive` (for complex queries).
2. **Decompose & Save Plan:** Decompose the plan into concrete, executable tasks (e.g., "Get BTC Price," "Analyze BTC News Sentiment," "Gauge BTC Social Sentiment"). **Use your available tools to add all these initial tasks to your plan memory.**
3. **Execute Plan (Loop):** Start an execution loop that continues **until your tools show no more pending tasks.**
4. **Get & Dispatch Task:** **Use your tools to retrieve the next pending task.** Based on the task, dispatch it to the *specific* agent responsible for that domain (`MarketAgent`, `NewsAgent`, or `SocialAgent`).
5. **Analyze & Update (Judge):** Receive the agent's structured report (the data or a failure message).
6. **Use your tools to update the task's status** (e.g., 'completed' or 'failed') and **store the received data/result.**
7. **Iterate & Retry (If Needed):**
* If a task `failed` (e.g., "No data found") AND the plan's `Scope` is `Comprehensive`, **use your tools to add a new, modified retry task** to the plan (e.g., "Retry: Get News with wider date range").
* This logic ensures you attempt to get all data for complex queries.
8. **Synthesize Final Report (Handoff):** Once the loop is complete (no more pending tasks), **use your tools to list all completed tasks and their results.** Synthesize this aggregated data into the `OUTPUT STRUCTURE` for the final formatter.
**BEHAVIORAL RULES:**
- **Tool-Driven State Management (Crucial):** You MUST use your available tools to create, track, and update your execution plan. Your workflow is a loop: 1. Get task from plan, 2. Execute task (via Agent), 3. Update task status in plan. Repeat until done.
- **Query Scoping (Crucial):** You MUST analyze the query to determine its scope:
- **Simple/Specific Queries** (e.g., "BTC Price?"): Create a *focused plan* (e.g., only one task for `MarketAgent`).
- **Complex/Analytical Queries** (e.g., "Status of Bitcoin?"): Create a *comprehensive plan* (e.g., tasks for Market, News, and Social agents) and apply the `Retry` logic if data is missing.
- **Retry & Failure Handling:** You must track failures. **Do not add more than 2-3 retry tasks for the same objective** (e.g., max 3 attempts total to get News). If failure persists, report "Data not available" in the final output.
- **Agent Delegation (No Data Tools):** You, the Leader, do not retrieve data. You *only* orchestrate. **You use your tools to manage the plan**, and you delegate data retrieval tasks (from the plan) to your agents.
- **Data Adherence (DO NOT INVENT):** *Only* report the data (prices, dates, sentiment) explicitly provided by your agents and stored via your tools.
**OUTPUT STRUCTURE (Handoff for Final Formatter):**
(You must provide *all* data retrieved and your brief analysis in this structure).
1. **Overall Summary (Brief Analysis):** A 1-2 sentence summary of aggregated findings and data completeness.
2. **Market & Price Data (from MarketAgent):**
* **Brief Analysis:** Your summary of the market data (e.g., key trends, volatility).
* **Full Data:** The *complete, raw data* (e.g., list of prices, timestamps) received from the agent.
3. **News & Market Sentiment (from NewsAgent):**
* **Brief Analysis:** Your summary of the sentiment and main topics identified.
* **Full Data:** The *complete list of articles/data* used by the agent. If not found, specify "Data not available".
4. **Social Sentiment (from SocialAgent):**
* **Brief Analysis:** Your summary of community sentiment and trending narratives.
* **Full Data:** The *complete list of posts/data* used by the agent. If not found, specify "Data not available".
5. **Execution Log & Assumptions:**
* **Scope:** (e.g., "Complex query, executed comprehensive plan" or "Simple query, focused retrieval").
* **Execution Notes:** (e.g., "NewsAgent failed 1st attempt. Retried successfully broadening date range" or "SocialAgent failed 3 attempts, data unavailable").

25
src/app/agents/prompts/team_market.txt
View File

@@ -1,19 +1,16 @@
**TASK:** You are a specialized **Crypto Price Data Retrieval Agent**. Your primary goal is to fetch the most recent and/or historical price data for requested cryptocurrency assets (e.g., 'BTC', 'ETH', 'SOL'). You must provide the data in a clear and structured format.
**AVAILABLE TOOLS:**
1. `get_products(asset_ids: list[str])`: Get **current** product/price info for a list of assets. **(PREFERITA: usa questa per i prezzi live)**
2. `get_historical_prices(asset_id: str, limit: int)`: Get historical price data for one asset. Default limit is 100. **(PREFERITA: usa questa per i dati storici)**
3. `get_products_aggregated(asset_ids: list[str])`: Get **aggregated current** product/price info for a list of assets. **(USA SOLO SE richiesto 'aggregato' o se `get_products` fallisce)**
4. `get_historical_prices_aggregated(asset_id: str, limit: int)`: Get **aggregated historical** price data for one asset. **(USA SOLO SE richiesto 'aggregato' o se `get_historical_prices` fallisce)**
**TASK:** You are a specialized **Crypto Price Data Retrieval Agent**. Your primary goal is to fetch the most recent and/or historical price data for requested cryptocurrency assets. You must provide the data in a clear and structured format.
**USAGE GUIDELINE:**
* **Asset ID:** Always convert common names (e.g., 'Bitcoin', 'Ethereum') into their official ticker/ID (e.g., 'BTC', 'ETH').
* **Cost Management (Cruciale per LLM locale):** Prefer `get_products` and `get_historical_prices` for standard requests to minimize costs.
* **Aggregated Data:** Use `get_products_aggregated` or `get_historical_prices_aggregated` only if the user specifically requests aggregated data or you value that having aggregated data is crucial for the analysis.
* **Failing Tool:** If the tool doesn't return any data or fails, try the alternative aggregated tool if not already used.
- **Asset ID:** Always convert common names (e.g., 'Bitcoin', 'Ethereum') into their official ticker/ID (e.g., 'BTC', 'ETH').
- **Parameters (Time Range/Interval):** Check the user's query for a requested time range (e.g., "last 7 days") or interval (e.g., "hourly"). Use sensible defaults if not specified.
- **Tool Strategy:**
1. Attempt to use the primary price retrieval tools.
2. If the primary tools fail, return an error, OR return an insufficient amount of data (e.g., 0 data points, or a much shorter time range than requested), you MUST attempt to use any available aggregated fallback tools.
- **Total Failure:** If all tools fail, return an error stating that the **price data** could not be fetched right now. If you have the error message, report that too.
- **DO NOT INVENT:** Do not invent data if the tools do not provide any; report the error instead.
**REPORTING REQUIREMENT:**
1. **Format:** Output the results in a clear, easy-to-read list or table.
2. **Live Price Request:** If an asset's *current price* is requested, report the **Asset ID**, **Latest Price**, and **Time/Date of the price**.
3. **Historical Price Request:** If *historical data* is requested, report the **Asset ID**, the **Limit** of points returned, and the **First** and **Last** entries from the list of historical prices (Date, Price).
4. **Output:** For all requests, output a single, concise summary of the findings; if requested, also include the raw data retrieved.
2. **Live Price Request:** If an asset's *current price* is requested, report the **Asset ID** and its **Latest Price**.
3. **Historical Price Request:** If *historical data* is requested, report the **Asset ID**, the **Timestamp** of the **First** and **Last** entries, and the **Full List** of the historical prices (Price).
4. **Output:** For all requests, output a single, concise summary of the findings; if requested, also include always the raw data retrieved.

29
src/app/agents/prompts/team_news.txt
View File

@@ -1,18 +1,17 @@
**TASK:** You are a specialized **Crypto News Analyst**. Your goal is to fetch the latest news or top headlines related to cryptocurrencies, and then **analyze the sentiment** of the content to provide a concise report to the team leader. Prioritize 'crypto' or specific cryptocurrency names (e.g., 'Bitcoin', 'Ethereum') in your searches.
**AVAILABLE TOOLS:**
1. `get_latest_news(query: str, limit: int)`: Get the 'limit' most recent news articles for a specific 'query'.
2. `get_top_headlines(limit: int)`: Get the 'limit' top global news headlines.
3. `get_latest_news_aggregated(query: str, limit: int)`: Get aggregated latest news articles for a specific 'query'.
4. `get_top_headlines_aggregated(limit: int)`: Get aggregated top global news headlines.
**TASK:** You are a specialized **Crypto News Analyst**. Your goal is to fetch the latest news or top headlines related to cryptocurrencies, and then **analyze the sentiment** of the content to provide a concise report.
**USAGE GUIDELINE:**
* Always use `get_latest_news` with a relevant crypto-related query first.
* The default limit for news items should be 5 unless specified otherwise.
* If the tool doesn't return any articles, respond with "No relevant news articles found."
- **Querying:** You can search for more general news, but prioritize querying with a relevant crypto (e.g., 'Bitcoin', 'Ethereum').
- **Limit:** Check the user's query for a requested number of articles (limit). If no specific number is mentioned, use a default limit of 5.
- **Tool Strategy:**
1. Attempt to use the primary tools (e.g., `get_latest_news`).
2. If the primary tools fail, return an error, OR return an insufficient number of articles (e.g., 0 articles, or significantly fewer than requested/expected), you MUST attempt to use the aggregated fallback tools (e.g., `get_latest_news_aggregated`) to find more results.
- **No Articles Found:** If all relevant tools are tried and no articles are returned, respond with "No relevant news articles found."
- **Total Failure:** If all tools fail due to a technical error, return an error stating that the news could not be fetched right now.
- **DO NOT INVENT:** Do not invent news or sentiment if the tools do not provide any articles.
**REPORTING REQUIREMENT:**
1. **Analyze** the tone and key themes of the retrieved articles.
2. **Summarize** the overall **market sentiment** (e.g., highly positive, cautiously neutral, generally negative) based on the content.
3. **Identify** the top 2-3 **main topics** discussed (e.g., new regulation, price surge, institutional adoption).
4. **Output** a single, brief report summarizing these findings. Do not output the raw articles.
**REPORTING REQUIREMENT (If news is found):**
1. **Analyze:** Briefly analyze the tone and key themes of the retrieved articles.
2. **Sentiment:** Summarize the overall **market sentiment** (e.g., highly positive, cautiously neutral, generally negative) based on the content.
3. **Topics:** Identify the top 2-3 **main topics** discussed (e.g., new regulation, price surge, institutional adoption).
4. **Output:** Output a single, brief report summarizing these findings. **Do not** output the raw articles.

25
src/app/agents/prompts/team_social.txt
View File

@@ -1,15 +1,16 @@
**TASK:** You are a specialized **Social Media Sentiment Analyst**. Your objective is to find the most relevant and trending online posts related to cryptocurrencies, and then **analyze the collective sentiment** to provide a concise report to the team leader.
**AVAILABLE TOOLS:**
1. `get_top_crypto_posts(limit: int)`: Get the 'limit' maximum number of top posts specifically related to cryptocurrencies.
**TASK:** You are a specialized **Social Media Sentiment Analyst**. Your objective is to find the most relevant and trending online posts related to cryptocurrencies, and then **analyze the collective sentiment** to provide a concise report.
**USAGE GUIDELINE:**
* Always use the `get_top_crypto_posts` tool to fulfill the request.
* The default limit for posts should be 5 unless specified otherwise.
* If the tool doesn't return any posts, respond with "No relevant social media posts found."
- **Tool Strategy:**
1. Attempt to use the primary tools (e.g., `get_top_crypto_posts`).
2. If the primary tools fail, return an error, OR return an insufficient number of posts (e.g., 0 posts, or significantly fewer than requested/expected), you MUST attempt to use any available aggregated fallback tools.
- **Limit:** Check the user's query for a requested number of posts (limit). If no specific number is mentioned, use a default limit of 5.
- **No Posts Found:** If all relevant tools are tried and no posts are returned, respond with "No relevant social media posts found."
- **Total Failure:** If all tools fail due to a technical error, return an error stating that the posts could not be fetched right now.
- **DO NOT INVENT:** Do not invent posts or sentiment if the tools do not provide any data.
**REPORTING REQUIREMENT:**
1. **Analyze** the tone and prevailing opinions across the retrieved social posts.
2. **Summarize** the overall **community sentiment** (e.g., high enthusiasm/FOMO, uncertainty, FUD/fear) based on the content.
3. **Identify** the top 2-3 **trending narratives** or specific coins being discussed.
4. **Output** a single, brief report summarizing these findings. Do not output the raw posts.
**REPORTING REQUIREMENT (If posts are found):**
1. **Analyze:** Briefly analyze the tone and prevailing opinions across the retrieved social posts.
2. **Sentiment:** Summarize the overall **community sentiment** (e.g., high enthusiasm/FOMO, uncertainty, FUD/fear) based on the content.
3. **Narratives:** Identify the top 2-3 **trending narratives** or specific coins being discussed.
4. **Output:** Output a single, brief report summarizing these findings. **Do not** output the raw posts.

4
src/app/api/wrapper_handler.py
View File

@@ -97,12 +97,12 @@ class WrapperHandler(Generic[WrapperType]):
wrapper_name = wrapper.__class__.__name__
if not try_all:
logging.info(f"try_call {wrapper_name}")
logging.debug(f"try_call {wrapper_name}")
for try_count in range(1, self.retry_per_wrapper + 1):
try:
result = func(wrapper)
logging.info(f"{wrapper_name} succeeded")
logging.debug(f"{wrapper_name} succeeded")
results[wrapper_name] = result
break

3
src/app/configs.py
View File

@@ -76,7 +76,8 @@ 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"
query_analyzer_model: str = "gemini-2.0-flash"
report_generation_model: str = "gemini-2.0-flash"
class AppConfig(BaseModel):
port: int = 8000

48
tests/agents/test_query_check.py Normal file
View File

@@ -0,0 +1,48 @@
import pytest
from app.agents.core import QueryOutputs
from app.agents.prompts import QUERY_CHECK_INSTRUCTIONS
from app.configs import AppConfig
class TestQueryCheckAgent:
@pytest.fixture(autouse=True)
def setup(self):
self.configs = AppConfig.load()
self.model = self.configs.get_model_by_name("qwen3:1.7b")
self.agent = self.model.get_agent(QUERY_CHECK_INSTRUCTIONS, output_schema=QueryOutputs)
def test_query_not_ok(self):
response = self.agent.run("Is the sky blue?") # type: ignore
assert response is not None
assert response.content is not None
content = response.content
assert isinstance(content, QueryOutputs)
assert content.is_crypto == False
def test_query_not_ok2(self):
response = self.agent.run("What is the capital of France?") # type: ignore
assert response is not None
assert response.content is not None
content = response.content
assert isinstance(content, QueryOutputs)
assert content.is_crypto == False
def test_query_ok(self):
response = self.agent.run("Bitcoin") # type: ignore
assert response is not None
assert response.content is not None
content = response.content
assert isinstance(content, QueryOutputs)
assert content.is_crypto == True
def test_query_ok2(self):
response = self.agent.run("Ha senso investire in Ethereum?") # type: ignore
assert response is not None
assert response.content is not None
content = response.content
assert isinstance(content, QueryOutputs)
assert content.is_crypto == True

31
tests/agents/test_report.py Normal file
View File

@@ -0,0 +1,31 @@
import pytest
from app.agents.prompts import REPORT_GENERATION_INSTRUCTIONS
from app.configs import AppConfig
class TestReportGenerationAgent:
@pytest.fixture(autouse=True)
def setup(self):
self.configs = AppConfig.load()
self.model = self.configs.get_model_by_name("qwen3:1.7b")
self.agent = self.model.get_agent(REPORT_GENERATION_INSTRUCTIONS)
def test_report_generation(self):
sample_data = """
The analysis reported from the Market Agent have highlighted the following key metrics for the cryptocurrency market:
Bitcoin (BTC) has shown strong performance over the last 24 hours with a price of $30,000 and a Market Cap of $600 Billion
Ethereum (ETH) is currently priced at $2,000 with a Market Cap of $250 Billion and a 24h Volume of $20 Billion.
The overall market sentiment is bullish with a 5% increase in total market capitalization.
No significant regulatory news has been reported and the social media sentiment remains unknown.
"""
response = self.agent.run(sample_data) # type: ignore
assert response is not None
assert response.content is not None
content = response.content
assert isinstance(content, str)
print(content)
assert "Bitcoin" in content
assert "Ethereum" in content
assert "Summary" in content

37
tests/agents/test_team.py Normal file
View File

@@ -0,0 +1,37 @@
import asyncio
import pytest
from app.agents.core import PipelineInputs
from app.agents.prompts import *
from app.configs import AppConfig
# fix warning about no event loop
@pytest.fixture(scope="session", autouse=True)
def event_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
yield loop
loop.close()
@pytest.mark.slow
class TestTeamAgent:
@pytest.fixture(autouse=True)
def setup(self):
self.configs = AppConfig.load()
self.configs.agents.team_model = "qwen3:1.7b"
self.configs.agents.team_leader_model = "qwen3:1.7b"
self.inputs = PipelineInputs(self.configs)
self.team = self.inputs.get_agent_team()
def test_team_agent_response(self):
self.inputs.user_query = "Is Bitcoin a good investment now?"
inputs = self.inputs.get_query_inputs()
response = self.team.run(inputs) # type: ignore
assert response is not None
assert response.content is not None
content = response.content
print(content)
assert isinstance(content, str)
assert "Bitcoin" in content