- Home
- iThink
- Vibe Coding Professionisti
- Un assistente di ricerca AI fatto in casa
Un assistente di ricerca AI fatto in casa
di Rocco Balzamà- Pubblicato il
- Aggiornato il
- 20 min di lettura
Prepara le ricerche senza perdere il filo
Per preparare una chiamata con un potenziale cliente cerchi informazioni sull’azienda, sui concorrenti e sulle notizie del settore. Nell’esempio considero 30-60 minuti di ricerca per chiamata e 5 chiamate a settimana: il tempo si accumula in fretta.
Lo stesso succede quando segui le norme sull’AI, il mercato o le pubblicazioni scientifiche: apri molte schede, leggi i riassunti e salvi link, ma devi ancora mettere insieme ciò che hai trovato.
In questo capitolo costruisci un assistente che raccoglie le informazioni, le sintetizza e prepara un riepilogo in Markdown ogni mattina alle 7:00. Puoi leggerlo in circa 5 minuti e tornare alle fonti quando devi verificare un dettaglio.
Come funziona l’assistente di ricerca
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐ ┌─────────────────┐
│ SOURCES │────▶│ COLLECTOR │────▶│ PROCESSOR │────▶│ SYNTHESIZER │
│ - Web (RSS, │ │ - requests/BS4 │ │ - newspaper4k │ │ - LLM │
│ search, │ │ - Playwright │ │ (article │ │ (summarize, │
│ specific │ │ (JS-heavy) │ │ extraction) │ │ key findings,│
│ URLs) │ │ - arXiv/PubMed │ │ - PDF parsing │ │ insights) │
│ - Academic │ │ API │ │ - Deduplication │ │ - Structured │
│ - Internal │ │ - Rate limiting │ │ - Relevance │ │ output │
│ (Notion, │ │ - Robots.txt │ │ scoring │ │ (markdown) │
│ Drive) │ │ compliance │ │ - Lang detection │ └─────────────────┘
└─────────────────┘ └──────────────────┘ └────────────────────┘ │
▼
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐ ┌─────────────────┐
│ DELIVERY │◀────│ FORMATTER │◀────│ QUALITY GATE │◀────│ │
│ - Markdown │ │ - Markdown │ │ - Source verify │ │ │
│ briefing │ │ template │ │ - Hallucination │ │ │
│ - Email │ │ - Citations │ │ check │ │ │
│ - Slack │ │ - Tags/categories│ │ - Relevance │ │ │
│ - Notion │ │ - Date/topic │ │ threshold │ │ │
│ - n8n webhook │ └──────────────────┘ └────────────────────┘ └─────────────────┘
└─────────────────┘Regole da implementare per raccogliere e usare le informazioni
| Principio | Implementazione | | ----------- | ----------------- | | Robots.txt compliance | Controlla prima di ogni crawl, rispetta Crawl-delay | | Rate limiting | Min 2-5 sec tra una richiesta e l'altra sullo stesso dominio, max 10 req/min | | User-Agent identitario | VibeResearchBot/1.0 (+https://tuosito.com/bot; email@tuo.com) | | No PII extraction | Non salvare email, telefoni, indirizzi personali | | Copyright awareness | Solo excerpt < 200 char, link to source, no full-text republish | | Data minimization | Salva solo: title, url, summary, key_findings, date, source | | Retention policy | Auto-delete > 90 giorni (configurabile) | | Opt-out respect | Se il sito ha noarchive o nosnippet → skip |
Il codice di questo capitolo applica solo robots.txt e una pausa fissa tra le richieste: Crawl-delay, limite per dominio, cancellazione dopo 90 giorni, controllo di noarchive/nosnippet ed estratti brevi vanno aggiunti prima dell'uso. Il collector salva fino a 10.000 caratteri di testo per la sintesi: per il testo integrale serve l'eccezione TDM (artt. 70-ter e 70-quater L. 633/1941), da usare solo rispettando l'opt-out dei siti.
I moduli dell’assistente
pip install aiohttp playwright trafilatura newspaper4k feedparser arxiv pymed anthropic python-dotenv && playwright install chromium1. Raccogli le pagine web (web_collector.py)
# web_collector.py — Ethical web collection
import asyncio
import aiohttp
import os
import re
from urllib.parse import urlparse
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from datetime import datetime
import json
from pathlib import Path
import hashlib
from playwright.async_api import async_playwright
import trafilatura # Ottima estrazione contenuto
import newspaper
from newspaper import Article # newspaper4k (stesso nome di modulo)
from urllib.robotparser import RobotFileParser
@dataclass
class SourceConfig:
name: str
base_url: str
type: str # "rss", "search", "direct", "sitemap"
selectors: Dict = field(default_factory=dict) # CSS selectors per estrazione
rate_limit_sec: float = 3.0
max_pages: int = 10
requires_js: bool = False
enabled: bool = True
category: str = "General"
@dataclass
class CollectedItem:
url: str
title: str
content: str
summary: str
source: str
category: str
published_at: Optional[str]
collected_at: str
content_hash: str
metadata: Dict = field(default_factory=dict)
class EthicalCollector:
def __init__(self, config: Dict, cache_dir: str = "research_cache"):
self.config = config
self.cache = Path(cache_dir)
self.cache.mkdir(parents=True, exist_ok=True)
self.session: Optional[aiohttp.ClientSession] = None
self.playwright_browser = None
self.visited: Set[str] = set()
self.robots_cache: Dict[str, any] = {}
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30)
self.session = aiohttp.ClientSession(
timeout=timeout,
headers={"User-Agent": "VibeResearchBot/1.0 (+https://tuosito.it/bot; contatto@tuosito.it)"}
)
self.playwright = await async_playwright().start()
self.playwright_browser = await self.playwright.chromium.launch(headless=True)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session: await self.session.close()
if self.playwright_browser: await self.playwright_browser.close()
if self.playwright: await self.playwright.stop()
def _get_robots_parser(self, domain: str):
if domain in self.robots_cache:
return self.robots_cache[domain]
try:
parser = RobotFileParser(f"https://{domain}/robots.txt")
parser.read()
self.robots_cache[domain] = parser
return parser
except Exception:
self.robots_cache[domain] = None
return None
def _can_fetch(self, url: str) -> bool:
parsed = urlparse(url)
parser = self._get_robots_parser(parsed.netloc)
if parser:
return parser.can_fetch("VibeResearchBot", url)
return True # Default allow se robots.txt non accessibile
def _get_cache_path(self, url: str) -> Path:
h = hashlib.md5(url.encode()).hexdigest()
return self.cache / f"{h}.json"
def _is_cached_fresh(self, url: str, max_age_hours: int = 24) -> Optional[CollectedItem]:
cache_path = self._get_cache_path(url)
if cache_path.exists():
data = json.loads(cache_path.read_text())
cached_at = datetime.fromisoformat(data["collected_at"])
if (datetime.now() - cached_at).total_seconds() < max_age_hours * 3600:
return CollectedItem(**data)
return None
async def collect_from_rss(self, feed_url: str, category: str, max_items: int = 20) -> List[CollectedItem]:
import feedparser
feed = feedparser.parse(feed_url)
items = []
for entry in feed.entries[:max_items]:
url = entry.get("link", "")
if not url or not self._can_fetch(url):
continue
# Prova cache
cached = self._is_cached_fresh(url)
if cached:
items.append(cached)
continue
# Estrai con trafilatura (veloce, no JS)
try:
downloaded = trafilatura.fetch_url(url)
if downloaded:
content = trafilatura.extract(downloaded, include_comments=False, include_tables=False)
if content and len(content) > 200:
item = CollectedItem(
url=url,
title=entry.get("title", ""),
content=content[:10000],
summary=content[:500],
source=feed.feed.get("title", feed_url),
category=category,
published_at=entry.get("published", datetime.now().isoformat()),
collected_at=datetime.now().isoformat(),
content_hash=hashlib.md5(content.encode()).hexdigest()
)
self._save_cache(item)
items.append(item)
except Exception as e:
print(f"⚠️ Errore RSS {url}: {e}")
await asyncio.sleep(self.config.get("default_rate_limit", 2.0))
return items
async def collect_from_search(self, query: str, category: str, max_results: int = 10) -> List[CollectedItem]:
# API di ricerca con licenza (Brave Search API; SerpAPI è un'alternativa): chiave nel .env
api_key = os.getenv("BRAVE_API_KEY")
if not api_key:
print("⚠️ BRAVE_API_KEY mancante: ricerca saltata")
return []
async with self.session.get(
"https://api.search.brave.com/res/v1/web/search",
params={"q": query, "count": max_results},
headers={"Accept": "application/json", "X-Subscription-Token": api_key},
) as resp:
data = await resp.json()
urls = [r["url"] for r in data.get("web", {}).get("results", [])][:max_results]
items = []
for url in urls:
if not self._can_fetch(url):
continue
cached = self._is_cached_fresh(url)
if cached:
items.append(cached)
continue
item = await self._extract_article(url, category)
if item:
items.append(item)
await asyncio.sleep(self.config.get("default_rate_limit", 3.0))
return items
async def _extract_article(self, url: str, category: str) -> Optional[CollectedItem]:
# Prova newspaper4k prima (veloce)
try:
article = Article(url)
article.download()
article.parse()
if article.text and len(article.text) > 300:
return CollectedItem(
url=url, title=article.title, content=article.text[:10000],
summary=article.text[:500], source=urlparse(url).netloc,
category=category, published_at=article.publish_date.isoformat() if article.publish_date else datetime.now().isoformat(),
collected_at=datetime.now().isoformat(),
content_hash=hashlib.md5(article.text.encode()).hexdigest()
)
except Exception:
pass
# Fallback: trafilatura
try:
downloaded = trafilatura.fetch_url(url)
if downloaded:
content = trafilatura.extract(downloaded)
if content and len(content) > 300:
return CollectedItem(url=url, title="", content=content[:10000], summary=content[:500],
source=urlparse(url).netloc, category=category, published_at=datetime.now().isoformat(),
collected_at=datetime.now().isoformat(), content_hash=hashlib.md5(content.encode()).hexdigest())
except Exception:
pass
return None
async def collect_js_heavy(self, url: str, category: str, wait_selector: str = None) -> Optional[CollectedItem]:
"""Per siti che richiedono JS rendering (Playwright)"""
if not self.playwright_browser:
return None
page = await self.playwright_browser.new_page()
try:
await page.goto(url, wait_until="networkidle", timeout=30000)
if wait_selector:
await page.wait_for_selector(wait_selector, timeout=10000)
content = await page.inner_text("body")
title = await page.title()
await page.close()
if len(content) > 500:
return CollectedItem(url=url, title=title, content=content[:10000], summary=content[:500],
source=urlparse(url).netloc, category=category, published_at=datetime.now().isoformat(),
collected_at=datetime.now().isoformat(), content_hash=hashlib.md5(content.encode()).hexdigest())
except Exception as e:
print(f"⚠️ Playwright error {url}: {e}")
await page.close()
return None
def _save_cache(self, item: CollectedItem):
cache_path = self._get_cache_path(item.url)
cache_path.write_text(json.dumps(item.__dict__, ensure_ascii=False, indent=2), encoding="utf-8")2. Raccogli le pubblicazioni (academic_collector.py)
# academic_collector.py — arXiv & PubMed API
import arxiv
from pymed import PubMed
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import json
from pathlib import Path
@dataclass
class Paper:
arxiv_id: str
title: str
authors: List[str]
summary: str
published: datetime
updated: datetime
categories: List[str]
pdf_url: str
primary_category: str
comment: str
journal_ref: str
doi: str
class AcademicCollector:
def __init__(self, cache_dir: str = "research_cache/academic"):
self.cache = Path(cache_dir)
self.cache.mkdir(parents=True, exist_ok=True)
self.arxiv_client = arxiv.Client()
self.pubmed = PubMed(tool="TuoBot", email="tua@email.it")
def search_arxiv(self, query: str, max_results: int = 20,
categories: List[str] = None, days_back: int = 30) -> List[Paper]:
if categories:
query = f"({query}) AND (" + " OR ".join(f"cat:{c}" for c in categories) + ")"
search = arxiv.Search(
query=query,
max_results=max_results,
sort_by=arxiv.SortCriterion.SubmittedDate,
sort_order=arxiv.SortOrder.Descending
)
papers = []
cutoff = datetime.now() - timedelta(days=days_back)
for result in self.arxiv_client.results(search):
if result.published.replace(tzinfo=None) < cutoff:
break
paper = Paper(
arxiv_id=result.entry_id.split("/")[-1],
title=result.title,
authors=[a.name for a in result.authors],
summary=result.summary,
published=result.published,
updated=result.updated,
categories=result.categories,
pdf_url=result.pdf_url,
primary_category=result.primary_category,
comment=result.comment or "",
journal_ref=result.journal_ref or "",
doi=result.doi or ""
)
papers.append(paper)
return papers
def search_pubmed(self, query: str, max_results: int = 20, days_back: int = 30) -> List[Dict]:
cutoff = (datetime.now() - timedelta(days=days_back)).strftime("%Y/%m/%d")
# Sintassi del filtro per data: verificala nella documentazione di PubMed
full_query = f"{query} AND ({cutoff}[Date - Publication] : 3000[Date - Publication])"
results = self.pubmed.query(full_query, max_results=max_results)
papers = []
for article in results:
papers.append({
"pubmed_id": article.pubmed_id,
"title": article.title,
"authors": [str(a) for a in article.authors] if article.authors else [],
"abstract": article.abstract,
"publication_date": article.publication_date,
"journal": article.journal,
"doi": article.doi,
"url": f"https://pubmed.ncbi.nlm.nih.gov/{article.pubmed_id}/"
})
return papers
def get_paper_summary(self, paper: Paper, use_llm: bool = True) -> Dict:
"""Genera summary strutturato per briefing"""
base = {
"title": paper.title,
"authors": paper.authors[:3],
"published": paper.published.strftime("%Y-%m-%d"),
"category": paper.primary_category,
"url": f"https://arxiv.org/abs/{paper.arxiv_id}",
"pdf": paper.pdf_url,
"key_points": [], # Sarà riempito da LLM
"relevance": "TBD",
"actionable": False
}
if use_llm:
# Chiamata LLM per estrarre key points (vedi synthesizer)
pass
return base3. Sintetizza con il modello (synthesizer.py)
# synthesizer.py — LLM-powered synthesis & briefing generation
import anthropic
import json
from typing import List, Dict, Any
from dataclasses import dataclass
import re
from datetime import datetime
from email.utils import parsedate_to_datetime
from pathlib import Path
import os
from dotenv import load_dotenv
load_dotenv()
def _iso_date(value: str) -> str:
"""Data ISO da RFC 822 (feed RSS) o da una stringa già ISO."""
try:
return parsedate_to_datetime(value).date().isoformat()
except (TypeError, ValueError):
return value[:10]
BRIEFING_PROMPT = """Sei un Senior Research Analyst. Sintetizza i materiali sotto in un **Executive Briefing** per un {role}.
OBIETTIVO: {objective}
AUDIENCE: {audience}
FOCUS AREAS: {focus_areas}
MATERIALI RACCOLTI ({count} items):
{items_json}
STRUTTURA OBBLIGATORIA OUTPUT (JSON):
{{
"briefing_date": "YYYY-MM-DD",
"executive_summary": "3-5 frasi: cosa è successo, perché importa, cosa fare",
"key_themes": [
{{"theme": "nome tema", "items_count": 3, "summary": "sintesi 2 frasi", "impact": "HIGH/MEDIUM/LOW", "sources": ["url1", "url2"]}}
],
"critical_findings": [
{{"finding": "descrizione", "evidence": "citazione/date specifiche", "source_url": "url", "confidence": 0.0-1.0, "action": "cosa fare"}}
],
"trending_topics": [
{{"topic": "nome", "mention_count": 5, "sentiment": "POSITIVE/NEGATIVE/NEUTRAL", "key_sources": ["url"]}}
],
"papers_summary": [
{{"title": "titolo", "authors": ["autore1"], "venue": "arXiv/PubMed", "key_contribution": "1 frase", "relevance": "HIGH/MEDIUM/LOW", "url": "url"}}
],
"competitor_moves": [
{{"company": "nome", "action": "cosa ha fatto", "source": "url", "date": "YYYY-MM-DD", "threat_level": "HIGH/MEDIUM/LOW"}}
],
"action_items": [
{{"priority": "HIGH/MEDIUM/LOW", "action": "azione concreta", "owner": "ruolo", "deadline": "YYYY-MM-DD", "context": "perché ora"}}
],
"sources_index": [
{{"url": "url", "title": "titolo", "category": "categoria", "date": "YYYY-MM-DD", "relevance": "HIGH/MEDIUM/LOW"}}
]
}}
REGOLE:
- Max 5 key_themes, 5 critical_findings, 5 trending, 5 papers, 5 competitors, 5 actions
- Ogni claim deve avere source_url tracciabile
- Niente hallucination: se info non nei materiali → non inventare
- Linguaggio: italiano, business-professional, actionable
- Priorità: impatto revenue/rischio/opportunità > curiosità
"""
class Synthesizer:
def __init__(self, api_key: str = None, model: str = "claude-sonnet-5-5"):
self.client = anthropic.Anthropic(api_key=api_key or os.getenv("ANTHROPIC_API_KEY"))
self.model = model
def generate_briefing(self, items: List[Dict], role: str = "consulente AI automation per PMI",
objective: str = "Identificare opportunità automazione, trend mercato, mosse competitor",
audience: str = "founder/consulenti/freelance",
focus_areas: List[str] = None) -> Dict:
focus = focus_areas or ["AI automation", "produttività", "SaaS B2B", "freelance economy", "no-code tools"]
# Prepara items per LLM (trunca per token limit)
items_for_llm = []
for item in items[:30]: # Max 30 items
items_for_llm.append({
"title": item.get("title", "")[:200],
"summary": item.get("summary", "")[:300],
"source": item.get("source", ""),
"url": item.get("url", ""),
"category": item.get("category", ""),
"date": _iso_date(item.get("published_at") or item.get("collected_at", ""))
})
prompt = BRIEFING_PROMPT.format(
role=role, objective=objective, audience=audience,
focus_areas=", ".join(focus), count=len(items_for_llm),
items_json=json.dumps(items_for_llm, ensure_ascii=False, indent=2)
)
response = self.client.messages.create(
model=self.model, max_tokens=4000,
messages=[{"role": "user", "content": prompt}]
)
text = next(b.text for b in response.content if b.type == "text")
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text.strip()) # toglie i recinti ```json
try:
briefing = json.loads(text)
briefing["generated_at"] = datetime.now().isoformat()
briefing["items_processed"] = len(items_for_llm)
return briefing
except json.JSONDecodeError as e:
print(f"⚠️ Parse error: {e}")
return {"error": "Parse failed", "raw": text[:500]}
def format_markdown_briefing(self, briefing: Dict) -> str:
"""Converte briefing JSON in Markdown leggibile"""
md = f"""# 📋 Daily Research Briefing — {briefing.get('briefing_date', 'Today')}
*Generato: {briefing.get('generated_at', '')[:16].replace('T', ' ')} | Items analizzati: {briefing.get('items_processed', 0)}*
## 🎯 Executive Summary
{briefing.get('executive_summary', 'N/A')}
## 🔑 Key Themes
"""
for theme in briefing.get('key_themes', []):
impact_emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}.get(theme.get('impact'), "⚪")
md += f"### {impact_emoji} {theme.get('theme', '')} ({theme.get('items_count', 0)} items)\n"
md += f"{theme.get('summary', '')}\n\n"
if theme.get('sources'):
md += f"*Fonti: {', '.join(theme['sources'][:3])}*\n\n"
md += "## ⚡ Critical Findings\n"
for finding in briefing.get('critical_findings', []):
conf = finding.get('confidence', 0)
conf_bar = "█" * int(conf * 10) + "░" * (10 - int(conf * 10))
md += f"- **{finding.get('finding', '')}** [{conf_bar} {conf:.0%}]\n"
md += f" - *Evidence:* {finding.get('evidence', '')}\n"
md += f" - *Source:* {finding.get('source_url', '')}\n"
md += f" - *Action:* {finding.get('action', '')}\n\n"
md += "## 📈 Trending Topics\n"
for trend in briefing.get('trending_topics', []):
sent_emoji = {"POSITIVE": "📈", "NEGATIVE": "📉", "NEUTRAL": "➡️"}.get(trend.get('sentiment'), "❓")
md += f"- {sent_emoji} **{trend.get('topic', '')}** ({trend.get('mention_count', 0)} mentions)\n"
md += "\n## 📚 Relevant Papers\n"
for paper in briefing.get('papers_summary', []):
rel_emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}.get(paper.get('relevance'), "⚪")
md += f"- {rel_emoji} **{paper.get('title', '')}**\n"
md += f" - Authors: {', '.join(paper.get('authors', []))} | {paper.get('venue', '')}\n"
md += f" - Key: {paper.get('key_contribution', '')}\n"
md += f" - [Link]({paper.get('url', '')})\n\n"
md += "## 🏢 Competitor Intelligence\n"
for comp in briefing.get('competitor_moves', []):
threat_emoji = {"HIGH": "🚨", "MEDIUM": "⚠️", "LOW": "👀"}.get(comp.get('threat_level'), "❓")
md += f"- {threat_emoji} **{comp.get('company', '')}**: {comp.get('action', '')} ({comp.get('date', '')})\n"
md += f" - Source: {comp.get('source', '')}\n\n"
md += "## ✅ Action Items\n"
for action in briefing.get('action_items', []):
pri_emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}.get(action.get('priority'), "⚪")
md += f"- {pri_emoji} **{action.get('action', '')}** (Owner: {action.get('owner', '')}, By: {action.get('deadline', '')})\n"
md += f" - *Context:* {action.get('context', '')}\n\n"
md += "---\n## 📚 Sources Index\n"
for src in briefing.get('sources_index', []):
rel_emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}.get(src.get('relevance'), "⚪")
md += f"- {rel_emoji} [{src.get('title', '')}]({src.get('url', '')}) — {src.get('category', '')} — {src.get('date', '')}\n"
md += "\n---\n*Generated by Vibe Research Agent*"
return md4. Collega i moduli con research_agent.py
# research_agent.py — Main orchestrator
import asyncio
import json
from datetime import datetime
from pathlib import Path
from typing import List, Dict
import os
from dotenv import load_dotenv
from web_collector import EthicalCollector, SourceConfig
from academic_collector import AcademicCollector
from synthesizer import Synthesizer
load_dotenv()
# Configurazione fonti default (feed di esempio: verifica che siano ancora attivi)
DEFAULT_SOURCES = [
SourceConfig("TechCrunch AI", "https://techcrunch.com/tag/artificial-intelligence/feed/", "rss", category="AI News", rate_limit_sec=2),
SourceConfig("The Verge AI", "https://www.theverge.com/ai-artificial-intelligence/rss/index.xml", "rss", category="AI News"),
SourceConfig("HBR Technology", "https://hbr.org/feed?topic=technology", "rss", category="Strategy"),
SourceConfig("a16z", "https://a16z.com/feed/", "rss", category="VC/Trends"),
SourceConfig("ArXiv CS.AI", "https://rss.arxiv.org/rss/cs.AI", "rss", category="Academic"),
SourceConfig("Google AI Blog", "https://blog.google/products/ai/rss/", "rss", category="Big Tech"),
]
SEARCH_QUERIES = [
("AI automation small business", "Market Trends"),
("no-code tools 2027", "Tools"),
("freelance AI productivity", "Freelance"),
("SMB digital transformation Italy", "Local Market"),
]
class ResearchAgent:
def __init__(self, output_dir: str = "briefings", config: Dict = None):
self.output = Path(output_dir)
self.output.mkdir(parents=True, exist_ok=True)
self.config = config or {}
self.sources = [SourceConfig(**s) if isinstance(s, dict) else s for s in (self.config.get("sources") or DEFAULT_SOURCES)]
self.search_queries = self.config.get("search_queries") or SEARCH_QUERIES
async def run_daily_briefing(self, date: str = None) -> Dict:
date = date or datetime.now().strftime("%Y-%m-%d")
print(f"🚀 Avvio briefing {date}")
all_items = []
# 1. Collect da RSS/Direct
async with EthicalCollector(self.config) as collector:
for source in self.sources:
if not source.enabled:
continue
print(f" 📥 Collecting: {source.name} ({source.type})")
try:
if source.type == "rss":
items = await collector.collect_from_rss(source.base_url, source.category, source.max_pages)
elif source.type == "search":
items = await collector.collect_from_search(source.base_url, source.category, source.max_pages)
else:
continue
all_items.extend([i.__dict__ for i in items])
print(f" ✅ {len(items)} items")
except Exception as e:
print(f" ❌ Error: {e}")
# 2. Collect da Search Queries
async with EthicalCollector(self.config) as collector:
for query, category in self.search_queries:
print(f" 🔍 Search: {query}")
try:
items = await collector.collect_from_search(query, category, max_results=5)
all_items.extend([i.__dict__ for i in items])
except Exception as e:
print(f" ❌ Error: {e}")
# 3. Academic papers
print(" 📚 Collecting academic papers...")
academic = AcademicCollector()
try:
arxiv_papers = academic.search_arxiv("AI automation OR LLM applications OR productivity", max_results=10, days_back=7)
for p in arxiv_papers:
all_items.append({
"title": p.title, "summary": p.summary, "source": "arXiv",
"url": f"https://arxiv.org/abs/{p.arxiv_id}", "category": "Academic",
"published_at": p.published.isoformat(), "collected_at": datetime.now().isoformat()
})
except Exception as e:
print(f" ❌ ArXiv error: {e}")
# 4. Deduplicazione
seen = set()
unique_items = []
for item in all_items:
h = item.get("content_hash") or hash(item.get("url", ""))
if h not in seen:
seen.add(h)
unique_items.append(item)
print(f" 📊 Totale unique items: {len(unique_items)}")
# 5. Synthesis
print(" 🧠 Synthesizing...")
synthesizer = Synthesizer()
briefing = synthesizer.generate_briefing(unique_items)
# 6. Save outputs
briefing["briefing_date"] = date
json_path = self.output / f"briefing_{date}.json"
json_path.write_text(json.dumps(briefing, ensure_ascii=False, indent=2), encoding="utf-8")
md = synthesizer.format_markdown_briefing(briefing)
md_path = self.output / f"briefing_{date}.md"
md_path.write_text(md, encoding="utf-8")
print(f"✅ Briefing salvato: {md_path}")
return briefing
# CLI Entry
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Vibe Research Agent - Daily Briefing")
parser.add_argument("--date", help="Date YYYY-MM-DD (default: today)")
parser.add_argument("--output", default="briefings", help="Output directory")
parser.add_argument("--config", help="Config JSON file")
args = parser.parse_args()
config = {}
if args.config:
config = json.loads(Path(args.config).read_text())
agent = ResearchAgent(output_dir=args.output, config=config)
asyncio.run(agent.run_daily_briefing(args.date))Prepara il riepilogo giornaliero con n8n
Il file si legge con un percorso assoluto, il nodo "Extract From File" ne ricava il testo e il nodo Markdown lo converte in HTML per l'email. Nel database Notion crea prima le proprietà del titolo e della data. Per Execute Command vale la nota del Capitolo 2. Lo schema è indicativo: costruisci il flusso nell'editor di n8n ed esportane il JSON.
{
"name": "Research Agent - Daily Briefing",
"nodes": [
{"name": "Cron Daily 06:30", "type": "n8n-nodes-base.cron", "typeVersion": 1, "position": [200, 200], "parameters": {"triggerTimes": {"item": [{"mode": "everyDay", "hour": 6, "minute": 30}]}}},
{"name": "Run Research Agent", "type": "n8n-nodes-base.executeCommand", "typeVersion": 1, "position": [400, 200], "parameters": {"command": "cd ~/vibe-coding/research && python research_agent.py"}},
{"name": "Read Markdown", "type": "n8n-nodes-base.readWriteFile", "typeVersion": 1, "position": [600, 200], "parameters": {"operation": "read", "fileSelector": "=/home/tuoutente/vibe-coding/research/briefings/briefing_{{ $now.toFormat('yyyy-MM-dd') }}.md"}},
{"name": "Extract Text", "type": "n8n-nodes-base.extractFromFile", "typeVersion": 1, "position": [800, 200], "parameters": {"operation": "text"}},
{"name": "Markdown to HTML", "type": "n8n-nodes-base.markdown", "typeVersion": 1, "position": [1000, 100], "parameters": {"mode": "markdownToHtml", "markdown": "={{ $json.data }}", "destinationKey": "html"}},
{"name": "Send Email", "type": "n8n-nodes-base.gmail", "typeVersion": 1, "position": [1200, 100], "parameters": {"fromEmail": "research@tuodominio.com", "toEmail": "tuo@email.com", "subject": "=📋 Daily Research Briefing - {{ $now.toFormat('dd/MM/yyyy') }}", "html": "={{ $json.html }}", "options": {}}},
{"name": "Post to Slack", "type": "n8n-nodes-base.slack", "typeVersion": 1, "position": [800, 400], "parameters": {"resource": "file", "operation": "upload", "binaryData": true, "channelIds": ["#research-briefings"], "options": {"initialComment": "📋 Daily Briefing pronto"}}},
{"name": "Save to Notion", "type": "n8n-nodes-base.notion", "typeVersion": 1, "position": [1000, 300], "parameters": {"resource": "databasePage", "operation": "create", "databaseId": "ID_DEL_TUO_DATABASE", "title": "=Daily Briefing {{ $now.toFormat('dd/MM/yyyy') }}", "blockUi": {"blockValues": [{"type": "paragraph", "textContent": "={{ $json.data }}"}]}}}
],
"connections": {
"Cron Daily 06:30": {"main": [[{"node": "Run Research Agent", "type": "main", "index": 0}]]},
"Run Research Agent": {"main": [[{"node": "Read Markdown", "type": "main", "index": 0}]]},
"Read Markdown": {"main": [[{"node": "Extract Text", "type": "main", "index": 0}, {"node": "Post to Slack", "type": "main", "index": 0}]]},
"Extract Text": {"main": [[{"node": "Markdown to HTML", "type": "main", "index": 0}, {"node": "Save to Notion", "type": "main", "index": 0}]]},
"Markdown to HTML": {"main": [[{"node": "Send Email", "type": "main", "index": 0}]]}
}
}Un esempio fittizio di riepilogo: briefing_2027-03-09.md
L'esempio mostra solo il formato: nomi, numeri e link sono segnaposto. Proprio perché il modello può inventare fatti e fonti, il quality gate deve verificare ogni link prima della consegna.
# 📋 Daily Research Briefing — 2027-03-09
*Generato: 2027-03-09 06:31 | Items analizzati: 47*
## 🎯 Executive Summary
[Azienda A] lancia [Modello X] con reasoning migliorato; [Azienda B] rilascia [Modello Y]. EU AI Act: si avvicinano gli obblighi per i sistemi ad alto rischio. Un nuovo paper arXiv misura l'effetto dell'AI pair programming sulla produttività. Competitor X lancia "AI Operations Suite" a [prezzo].
## 🔑 Key Themes
### 🔴 Foundation Model Updates (3 items)
[Azienda A] e [Azienda B] rilasciano modelli con reasoning migliorato. Impatto da verificare su costi e qualità per coding.
*Fonti: [link fonte], [link fonte], [link fonte]*
### 🟡 EU AI Act Compliance (2 items)
Obblighi per i sistemi ad alto rischio: dal 2 dicembre 2027 (allegato III) e dal 2 agosto 2028 (prodotti dell'allegato I, art. 6 par. 1). Opportunità: consulting compliance per PMI.
*Fonti: eur-lex.europa.eu, [link fonte]*
## ⚡ Critical Findings
- **[Modello X]: prezzo API invariato, migliore su coding** [████████░░ 80%]
- *Evidence:* "[citazione dalla fonte]" ([link fonte], [data])
- *Action:* Test migration su pipeline content/coding entro questa settimana
## 📈 Trending Topics
- 📈 **Agentic workflows** (12 mentions)
- 📉 **Prompt engineering roles declining** (8 mentions)
- ➡️ **Local LLM adoption** (6 mentions)
## 📚 Relevant Papers
- 🔴 **"[Titolo del paper]"** ([ID arXiv])
- Authors: [Autore] et al. | arXiv
- Key: [contributo principale in una frase]
- [Link]([link fonte])
## 🏢 Competitor Intelligence
- 🚨 **Competitor X**: Lancia "AI Operations Suite" a [prezzo] (include n8n-hosted + LLM proxy)
- Source: [link fonte] ([data])
## ✅ Action Items
- 🔴 **Test [Modello X] su content pipeline** (Owner: Tech Lead, By: 2027-03-12)
- *Context:* Potenziale quality boost per draft generation
- 🟡 **Draft AI Act compliance checklist per clienti PMI** (Owner: Consulting Lead, By: 2027-03-20)
- *Context:* Nuovo revenue stream consulting
## 📚 Sources Index
- 🔴 [[Modello X]: note di rilascio]([link fonte]) — Big Tech — [data]
- 🔴 [[Modello Y]: annuncio]([link fonte]) — Big Tech — [data]
- 🟡 [EU AI Act Implementation](https://eur-lex.europa.eu/...) — Regulation — 2027-03-07Costi e tempi di esecuzione
| Componente | Costo mensile stimato | Note | | ------------ | ---------------------- | ------ | | API LLM (synthesis) | €3-5 | ~15k token/giorno | | ArXiv/PubMed API | Gratuito | Rate limited | | RSS/News collection | Gratuito | Solo bandwidth | | Playwright (JS-heavy) | CPU locale | Solo se necessario | | n8n Cloud | €20/mese | O self-hosted | | Totale | ~€25/mese | vs €2.000+/mese analista junior |
Prossimo capitolo
Capitolo 6: automazioni con n8n. Collega il modulo del sito al CRM, alle email, alle attività in Asana e alle notifiche Slack. Vai al capitolo 6
Serie Vibe Coding per Professionisti — Capitolo 5 di 8. Il codice completo è in questo articolo. Nel prossimo capitolo colleghiamo più servizi con n8n.
Per approfondire
- Agenti AI: cosa sono e come usarli in azienda
- Agente digitale, capitolo 1: cos'è un agente AI
- Shadow AI: il rischio invisibile
- AI Act per le imprese: gli obblighi
- Chatbot e agenti AI
