AI Agents · 课时

构建每日简报智能体

Q: 「构建每日简报智能体」这节课中我会学到什么？

晨间摘要：自动发送新闻 + 日历 + 电子邮件汇总 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

晨间摘要：自动发送新闻 + 日历 + 电子邮件汇总

构建每日简报智能体 是 CoddyKit 上的免费 AI Agents 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

什么是每日简报代理

每日简报代理会在每天早上的预定时间运行，从多个来源（日历、电子邮件、新闻）收集信息，使用 LLM 对其进行综合，并发送个性化摘要。这是定时个人代理的经典示例。

简报架构

简报流程包含五个阶段：

触发：上午 8 点的定时任务启动代理
获取：并行获取日历事件、未读电子邮件和新闻标题
综合：LLM 根据所有数据创建连贯的简报
个性化：根据用户偏好调整语气和内容
发送：通过电子邮件或 Slack 私信发送

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')

使用 APScheduler 进行调度

请将简报安排在每个工作日早上 8 点运行。使用 APScheduler 和 SQLite 持久化，以便任务在服务器重启后仍然存在。

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')

获取日历事件

请使用 Google Calendar API 获取今天的日历事件。代理需要知道安排了哪些会议，才能将它们加入简报。

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

获取未读电子邮件

请获取最近的未读电子邮件。简报代理只需要摘要——标题和发件人——而不需要完整正文，以便控制 LLM 上下文的规模。

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

获取新闻标题

请使用新闻 API 获取相关新闻标题。根据用户配置的兴趣主题进行筛选，以生成个性化的新闻部分。

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

并行获取数据

请同时获取日历事件、电子邮件、新闻和天气。由于这些操作彼此独立，并行获取可以将总获取时间从约 4 秒缩短到约 1 秒。

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

LLM 综合

将所有获取的数据传递给 LLM，并要求它撰写自然且个性化的简报。提示词结构很重要：请清晰地组织数据，以便 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

通过电子邮件发送

请将简报格式化为 HTML 电子邮件并发送。良好的 HTML 电子邮件格式可以让简报更易于在移动设备上阅读，因为大多数人会在移动设备上阅读早间摘要。

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')

通过 Slack 私信发送

对于主要使用 Slack 的用户，将格式化后的简报作为 Slack 私信发送通常比电子邮件更受欢迎。请使用 Block Kit 实现丰富的格式。

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}')

简报主编排器

编排器会串联所有阶段：并行获取数据，使用 LLM 进行综合，发送到用户偏好的渠道，并记录结果以便监控。

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')

知识检查：每日简报代理

请测试您对构建每日简报代理的理解。

每日简报代理总结

完整的每日简报代理包含：用于在工作日早上 8 点执行定时任务的 APScheduler；从日历、电子邮件和新闻 API 并行获取数据；使用 LLM 创建个性化且连贯的简报；由用户偏好驱动的个性化；以及通过电子邮件或 Slack 私信进行多渠道发送。这是经典的常驻个人代理模式，展示了本课程中的所有概念。

常见问题解答

「构建每日简报智能体」课时是免费的吗？

是的 — 「构建每日简报智能体」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「构建每日简报智能体」这节课中我会学到什么？

晨间摘要：自动发送新闻 + 日历 + 电子邮件汇总你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「构建每日简报智能体」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents