Construindo um agente de briefing diário

O que é um agente de resumo diário?

Arquitetura do resumo

from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class DailyBriefingData:
    date: str
    calendar_events: List[Dict] = field(default_factory=list)
    unread_emails: List[Dict] = field(default_factory=list)
    news_headlines: List[Dict] = field(default_factory=list)
    weather: Dict = field(default_factory=dict)
    briefing_text: str = ''
    delivery_status: str = 'pending'

print('Daily briefing data structure defined')

Agendamento com APScheduler

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.triggers.cron import CronTrigger
import asyncio

scheduler = BackgroundScheduler(
    jobstores={'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')},
    timezone='America/New_York'
)

def run_briefing_job():
    print('Daily briefing job triggered')
    asyncio.run(generate_and_deliver_briefing('user-42'))

# 8am Monday-Friday
scheduler.add_job(
    run_briefing_job,
    CronTrigger(day_of_week='mon-fri', hour=8, minute=0),
    id='daily_briefing_user_42',
    replace_existing=True
)

scheduler.start()
print('Briefing scheduler started: runs Mon-Fri at 8am ET')

Busca de eventos do calendário

from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from datetime import datetime, timedelta
import pytz

def get_todays_calendar_events(token_path: str = 'token.json', timezone: str = 'America/New_York') -> list:
    creds = Credentials.from_authorized_user_file(token_path)
    service = build('calendar', 'v3', credentials=creds)
    
    tz = pytz.timezone(timezone)
    now = datetime.now(tz)
    start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0)
    end_of_day = now.replace(hour=23, minute=59, second=59)
    
    result = service.events().list(
        calendarId='primary',
        timeMin=start_of_day.isoformat(),
        timeMax=end_of_day.isoformat(),
        singleEvents=True,
        orderBy='startTime'
    ).execute()
    
    events = []
    for item in result.get('items', []):
        start = item.get('start', {}).get('dateTime', item.get('start', {}).get('date'))
        events.append({
            'title': item.get('summary', 'Untitled'),
            'start': start,
            'attendees': len(item.get('attendees', [])),
            'location': item.get('location', '')
        })
    return events

Busca de e-mails não lidos

from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import base64

def get_email_summaries(token_path: str = 'token.json', max_results: int = 10) -> list:
    creds = Credentials.from_authorized_user_file(token_path)
    service = build('gmail', 'v1', credentials=creds)
    
    results = service.users().messages().list(
        userId='me',
        q='is:unread newer_than:1d',
        maxResults=max_results
    ).execute()
    
    summaries = []
    for msg in results.get('messages', []):
        detail = service.users().messages().get(
            userId='me',
            id=msg['id'],
            format='metadata',
            metadataHeaders=['From', 'Subject', 'Date']
        ).execute()
        
        headers = {h['name']: h['value'] for h in detail['payload']['headers']}
        summaries.append({
            'from': headers.get('From', ''),
            'subject': headers.get('Subject', 'No Subject'),
            'date': headers.get('Date', '')
        })
    
    return summaries

Busca de manchetes de notícias

import httpx
import os

NEWS_API_KEY = os.environ.get('NEWS_API_KEY', 'your-key')

async def get_news_headlines(topics: list, max_per_topic: int = 3) -> list:
    headlines = []
    
    async with httpx.AsyncClient() as client:
        for topic in topics:
            try:
                response = await client.get(
                    'https://newsapi.org/v2/top-headlines',
                    params={
                        'q': topic,
                        'language': 'en',
                        'pageSize': max_per_topic,
                        'apiKey': NEWS_API_KEY
                    },
                    timeout=10.0
                )
                data = response.json()
                for article in data.get('articles', []):
                    headlines.append({
                        'topic': topic,
                        'title': article.get('title', ''),
                        'source': article.get('source', {}).get('name', ''),
                        'url': article.get('url', '')
                    })
            except Exception as e:
                print(f'News fetch failed for {topic}: {e}')
    
    return headlines[:10]  # Cap total

Busca paralela de dados

import asyncio
from datetime import datetime

async def fetch_all_briefing_data(user_prefs: dict) -> 'DailyBriefingData':
    data = DailyBriefingData(date=datetime.now().strftime('%A, %B %d, %Y'))
    
    async def safe_get_calendar():
        try:
            return get_todays_calendar_events()
        except Exception as e:
            print(f'Calendar fetch failed: {e}')
            return []
    
    async def safe_get_emails():
        try:
            return get_email_summaries()
        except Exception as e:
            print(f'Email fetch failed: {e}')
            return []
    
    async def safe_get_news():
        topics = user_prefs.get('news_topics', ['technology', 'business'])
        return await get_news_headlines(topics)
    
    # Run all fetches in parallel
    calendar_events, emails, news = await asyncio.gather(
        safe_get_calendar(),
        safe_get_emails(),
        safe_get_news(),
        return_exceptions=False
    )
    
    data.calendar_events = calendar_events
    data.unread_emails = emails
    data.news_headlines = news
    
    return data

Síntese pelo LLM

import openai
import json

client = openai.OpenAI(api_key='sk-...')

def synthesize_briefing(data: 'DailyBriefingData', user_prefs: dict) -> str:
    user_name = user_prefs.get('name', 'there')
    tone = user_prefs.get('briefing_tone', 'professional')  # 'casual', 'professional', 'concise'
    
    calendar_text = json.dumps(data.calendar_events[:5], indent=2) if data.calendar_events else 'No meetings today.'
    email_text = json.dumps(data.unread_emails[:5], indent=2) if data.unread_emails else 'No unread emails.'
    news_text = '\n'.join([f'- [{h["topic"]}] {h["title"]} ({h["source"]})' for h in data.news_headlines[:6]])
    
    prompt = (
        f'Create a {tone} morning briefing for {user_name} for {data.date}.\n\n'
        f'Today\'s calendar:\n{calendar_text}\n\n'
        f'Unread emails:\n{email_text}\n\n'
        f'News headlines:\n{news_text}\n\n'
        'Write a concise, actionable briefing in 3-4 paragraphs. '
        'Start with today\'s schedule, then email highlights, then relevant news. '
        'End with one suggested priority for the day.'
    )
    
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        temperature=0.3
    )
    return response.choices[0].message.content

Entrega por e-mail

def format_briefing_email(briefing_text: str, data: 'DailyBriefingData') -> str:
    paragraphs = briefing_text.split('\n\n')
    html_paragraphs = ''.join([f'<p>{p}</p>' for p in paragraphs if p.strip()])
    
    calendar_items = ''.join([
        f'<li><strong>{e["start"][:16]}</strong> - {e["title"]}</li>'
        for e in data.calendar_events[:5]
    ])
    
    return f'''
    <html><body style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto;">
    <h2 style="color: #333;">Morning Briefing — {data.date}</h2>
    {html_paragraphs}
    {'<h3>Today\'s Calendar</h3><ul>' + calendar_items + '</ul>' if calendar_items else ''}
    <hr>
    <p style="color: #666; font-size: 12px;">Generated by your Personal AI Assistant</p>
    </body></html>
    '''

def deliver_briefing_email(data: 'DailyBriefingData', user_prefs: dict):
    html = format_briefing_email(data.briefing_text, data)
    to_email = user_prefs.get('email', '')
    send_email_alert(
        to_email=to_email,
        subject=f'Morning Briefing - {data.date}',
        html_body=html
    )

print('Email delivery function defined')

Entrega por DM do Slack

from slack_sdk import WebClient
import os

slack_client = WebClient(token=os.environ.get('SLACK_BOT_TOKEN', 'xoxb-...'))

def deliver_briefing_slack(data: 'DailyBriefingData', user_prefs: dict):
    slack_user_id = user_prefs.get('slack_user_id')
    if not slack_user_id:
        print('No Slack user ID configured')
        return
    
    # Split briefing into sections for blocks
    paragraphs = [p for p in data.briefing_text.split('\n\n') if p.strip()]
    
    blocks = [
        {'type': 'header', 'text': {'type': 'plain_text', 'text': f'Morning Briefing - {data.date}'}},
        {'type': 'divider'}
    ]
    
    for para in paragraphs[:4]:  # Max 4 paragraphs
        blocks.append({
            'type': 'section',
            'text': {'type': 'mrkdwn', 'text': para}
        })
    
    if data.calendar_events:
        event_list = '\n'.join([f'• {e["start"][:16]} - {e["title"]}' for e in data.calendar_events[:3]])
        blocks.append({'type': 'section', 'text': {'type': 'mrkdwn', 'text': f'*Meetings*\n{event_list}'}})
    
    slack_client.chat_postMessage(
        channel=slack_user_id,
        text=f'Morning Briefing - {data.date}',
        blocks=blocks
    )
    print(f'Briefing delivered to Slack user {slack_user_id}')

Orquestrador principal do resumo

import asyncio
from datetime import datetime

async def generate_and_deliver_briefing(user_id: str):
    print(f'Generating briefing for {user_id} at {datetime.now()}')
    
    # Load user preferences
    conn = init_db()
    user_prefs = load_user_preferences(conn, user_id)
    
    # Step 1: Fetch all data in parallel
    data = await fetch_all_briefing_data(user_prefs)
    
    if not data.calendar_events and not data.unread_emails and not data.news_headlines:
        print('No data fetched, skipping briefing')
        return
    
    # Step 2: Synthesize with LLM
    data.briefing_text = synthesize_briefing(data, user_prefs)
    
    # Step 3: Deliver
    delivery_channel = user_prefs.get('delivery_channel', 'email')
    if delivery_channel == 'slack':
        deliver_briefing_slack(data, user_prefs)
    else:
        deliver_briefing_email(data, user_prefs)
    
    data.delivery_status = 'delivered'
    
    # Step 4: Log
    print(f'Briefing delivered via {delivery_channel}')
    print(f'Data: {len(data.calendar_events)} events, {len(data.unread_emails)} emails, {len(data.news_headlines)} headlines')

print('Daily briefing orchestrator defined')

Verificação de compreensão: agente de resumo diário

Resumo do agente de resumo diário