0Pricing
AI Agents · 课时

构建多应用自动化流程

通过智能体管理的工具调用串联 Gmail → Slack → Google Sheets

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

多应用流水线

多应用流水线会将多个服务连接成一个自动化工作流。本课将构建：Gmail 新邮件 → 代理读取并提取操作项 → 创建 Trello 卡片 → 发送 Slack 通知。

流水线架构

流水线包含四个阶段：

  • 触发：Gmail push 通知或轮询检测到新邮件
  • 提取：LLM 读取邮件并提取操作项
  • 创建：Trello API 为每个操作项创建一张卡片
  • 通知：Slack API 发布摘要消息

每个阶段都是一个具有明确输入和输出的独立函数。

from dataclasses import dataclass, field
from typing import List

@dataclass
class Email:
    id: str
    sender: str
    subject: str
    body: str

@dataclass
class ActionItem:
    title: str
    description: str
    due_date: str = None

@dataclass
class PipelineResult:
    email_id: str
    action_items: List[ActionItem] = field(default_factory=list)
    trello_card_ids: List[str] = field(default_factory=list)
    slack_message_ts: str = None
    error: str = None

if __name__ == '__main__':
    email = Email(id='e1', sender='alice@example.com', subject='Project update', body='See attached.')
    result = PipelineResult(email_id=email.id, action_items=[ActionItem(title='Review attachment', description='Check the doc')])
    print(f'Pipeline result for {result.email_id}: {len(result.action_items)} action item(s)')
    print(' -', result.action_items[0].title)

阶段 1：读取电子邮件

使用 Gmail API 获取新邮件。google-api-python-client 库负责处理身份验证和 API 调用。我们会轮询尚未查看的消息。

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

def get_gmail_service(token_path='token.json'):
    creds = Credentials.from_authorized_user_file(token_path)
    return build('gmail', 'v1', credentials=creds)

def fetch_unread_emails(service, max_results=10):
    results = service.users().messages().list(
        userId='me',
        q='is:unread',
        maxResults=max_results
    ).execute()
    messages = results.get('messages', [])
    emails = []
    for msg in messages:
        detail = service.users().messages().get(
            userId='me', id=msg['id'], format='full'
        ).execute()
        headers = {h['name']: h['value'] for h in detail['payload']['headers']}
        emails.append(Email(
            id=msg['id'],
            sender=headers.get('From', ''),
            subject=headers.get('Subject', ''),
            body=extract_body(detail)
        ))
    return emails

def extract_body(message_detail):
    payload = message_detail.get('payload', {})
    if 'data' in payload.get('body', {}):
        return base64.urlsafe_b64decode(payload['body']['data']).decode('utf-8')
    return ''

阶段 2：提取操作项

将邮件内容传递给 LLM，并要求它以结构化 JSON 格式提取操作项。使用 response_format 获取可靠的 JSON 输出。

import openai
import json

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

def extract_action_items(email: 'Email') -> list:
    prompt = (
        'Extract all action items from this email. '
        'Return JSON array with objects having fields: '
        'title (string), description (string), due_date (string or null).\n\n'
        f'From: {email.sender}\n'
        f'Subject: {email.subject}\n'
        f'Body:\n{email.body}'
    )
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    result = json.loads(response.choices[0].message.content)
    items = result.get('action_items', [])
    return [
        ActionItem(
            title=item['title'],
            description=item.get('description', ''),
            due_date=item.get('due_date')
        )
        for item in items
    ]

阶段 3：创建 Trello 卡片

Trello REST API 使用一个简单的 POST 请求创建卡片。您需要 API 密钥、令牌以及用于创建卡片的列表 ID。

import httpx

TRELLO_API_KEY = 'your-trello-api-key'
TRELLO_TOKEN = 'your-trello-token'
TRELLO_LIST_ID = 'your-list-id'

def create_trello_card(action_item: 'ActionItem') -> str:
    url = 'https://api.trello.com/1/cards'
    params = {
        'key': TRELLO_API_KEY,
        'token': TRELLO_TOKEN
    }
    data = {
        'name': action_item.title,
        'desc': action_item.description,
        'idList': TRELLO_LIST_ID,
        'due': action_item.due_date
    }
    response = httpx.post(url, params=params, json=data)
    response.raise_for_status()
    card = response.json()
    return card['id']

def create_trello_cards_for_items(action_items: list) -> list:
    card_ids = []
    for item in action_items:
        card_id = create_trello_card(item)
        print(f'Created Trello card: {item.title} (ID: {card_id})')
        card_ids.append(card_id)
    return card_ids

阶段 4：发送 Slack 通知

使用 Slack SDK 向频道发送格式化的摘要消息。消息应清楚地总结提取并创建了哪些操作项。

from slack_sdk import WebClient

slack_client = WebClient(token='xoxb-your-slack-bot-token')

def send_slack_summary(email: 'Email', action_items: list, card_ids: list, channel: str = '#automation'):
    if not action_items:
        return None
    
    items_text = '\n'.join([
        f'  - {item.title}'
        for item in action_items
    ])
    
    message = (
        f'*New email processed from {email.sender}*\n'
        f'*Subject:* {email.subject}\n\n'
        f'*Action items extracted ({len(action_items)}):*\n'
        f'{items_text}\n\n'
        f'Trello cards created: {len(card_ids)}'
    )
    response = slack_client.chat_postMessage(
        channel=channel,
        text=message,
        mrkdwn=True
    )
    return response['ts']

处理每个阶段中的错误

流水线的健壮程度取决于其错误处理能力。每个阶段都可能独立失败。请包装每次调用，记录错误，并决定是继续还是中止流水线。

import logging

logger = logging.getLogger('pipeline')

def run_pipeline_stage(stage_name, fn, *args, **kwargs):
    try:
        result = fn(*args, **kwargs)
        logger.info(f'Stage {stage_name}: success')
        return result, None
    except Exception as e:
        logger.error(f'Stage {stage_name}: FAILED - {e}')
        return None, str(e)

def process_email_pipeline(email):
    # Stage 2: Extract
    action_items, err = run_pipeline_stage('extract', extract_action_items, email)
    if err:
        return PipelineResult(email_id=email.id, error=f'Extract failed: {err}')
    if not action_items:
        logger.info(f'No action items found in email {email.id}')
        return PipelineResult(email_id=email.id, action_items=[])
    
    # Stage 3: Trello
    card_ids, err = run_pipeline_stage('trello', create_trello_cards_for_items, action_items)
    if err:
        card_ids = []  # Continue even if Trello fails
    
    # Stage 4: Slack
    ts, err = run_pipeline_stage('slack', send_slack_summary, email, action_items, card_ids or [])
    
    return PipelineResult(
        email_id=email.id,
        action_items=action_items,
        trello_card_ids=card_ids or [],
        slack_message_ts=ts
    )

流水线的结构化日志记录

使用结构化日志记录，以便查询流水线执行历史。将流水线启动、每个阶段完成情况和最终结果记录为 JSON 对象。

import logging
import json
from datetime import datetime

class PipelineLogger:
    def __init__(self, pipeline_name):
        self.pipeline_name = pipeline_name
        self.logger = logging.getLogger(pipeline_name)
        self.run_id = None
        self.start_time = None
    
    def start(self, email_id):
        self.run_id = f'{email_id}_{int(datetime.now().timestamp())}'
        self.start_time = datetime.now()
        self.logger.info(json.dumps({
            'event': 'pipeline_start',
            'run_id': self.run_id,
            'email_id': email_id
        }))
    
    def stage_done(self, stage, result_summary):
        self.logger.info(json.dumps({
            'event': 'stage_complete',
            'run_id': self.run_id,
            'stage': stage,
            'result': result_summary
        }))
    
    def finish(self, success, details):
        duration = (datetime.now() - self.start_time).total_seconds()
        self.logger.info(json.dumps({
            'event': 'pipeline_finish',
            'run_id': self.run_id,
            'success': success,
            'duration_seconds': duration,
            'details': details
        }))

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stdout)
    pl = PipelineLogger('demo_pipeline')
    pl.start('email_123')
    pl.stage_done('extract', {'items_found': 3})
    pl.finish(True, {'action_items': 2})

主流水线运行器

主运行器将所有部分串联起来。它会轮询新邮件，通过流水线处理每封邮件，并将其标记为已读，以避免重复处理。

import time

def mark_as_read(gmail_service, email_id):
    gmail_service.users().messages().modify(
        userId='me',
        id=email_id,
        body={'removeLabelIds': ['UNREAD']}
    ).execute()

def run_email_pipeline_loop(gmail_service, poll_interval=60):
    pipeline_logger = PipelineLogger('email_pipeline')
    print(f'Pipeline running. Polling every {poll_interval}s')
    
    while True:
        try:
            emails = fetch_unread_emails(gmail_service)
            print(f'Found {len(emails)} unread emails')
            
            for email in emails:
                pipeline_logger.start(email.id)
                result = process_email_pipeline(email)
                
                if result.error:
                    pipeline_logger.finish(False, {'error': result.error})
                else:
                    mark_as_read(gmail_service, email.id)
                    pipeline_logger.finish(True, {
                        'action_items': len(result.action_items),
                        'cards_created': len(result.trello_card_ids)
                    })
        
        except Exception as e:
            print(f'Pipeline loop error: {e}')
        
        time.sleep(poll_interval)

配置管理

请将所有 API 凭证和流水线设置保存在环境变量中，而不是代码里。在启动时加载这些变量，并验证必需的键是否存在。

import os
from dotenv import load_dotenv

load_dotenv()

class PipelineConfig:
    def __init__(self):
        self.openai_api_key = os.environ.get('OPENAI_API_KEY', '')
        self.slack_bot_token = os.environ.get('SLACK_BOT_TOKEN', '')
        self.trello_api_key = os.environ.get('TRELLO_API_KEY', '')
        self.trello_token = os.environ.get('TRELLO_TOKEN', '')
        self.trello_list_id = os.environ.get('TRELLO_LIST_ID', '')
        self.slack_channel = os.environ.get('SLACK_CHANNEL', '#automation')
        self.poll_interval = int(os.environ.get('POLL_INTERVAL_SECONDS', '60'))
    
    def validate(self):
        missing = []
        required = [
            ('OPENAI_API_KEY', self.openai_api_key),
            ('SLACK_BOT_TOKEN', self.slack_bot_token),
            ('TRELLO_API_KEY', self.trello_api_key),
            ('TRELLO_TOKEN', self.trello_token),
            ('TRELLO_LIST_ID', self.trello_list_id)
        ]
        for name, value in required:
            if not value:
                missing.append(name)
        if missing:
            raise ValueError(f'Missing required config: {missing}')
        return True

config = PipelineConfig()
config.validate()
print('Config validated successfully')

测试流水线

在运行端到端流程之前，请使用模拟数据独立测试流水线的每个阶段。这样，您无需消耗 API 配额或创建真实的 Trello 卡片，就可以验证逻辑是否正确。

from unittest.mock import MagicMock, patch

def test_extract_action_items_mock():
    mock_response = MagicMock()
    mock_response.choices[0].message.content = '{"action_items": [{"title": "Follow up with vendor", "description": "Call about invoice", "due_date": null}]}'
    
    with patch('openai.OpenAI') as mock_openai:
        mock_client = MagicMock()
        mock_client.chat.completions.create.return_value = mock_response
        mock_openai.return_value = mock_client
        
        test_email = Email(
            id='test123',
            sender='vendor@example.com',
            subject='Invoice Follow-up Needed',
            body='Please follow up with the vendor about the outstanding invoice.'
        )
        # Would call extract_action_items(test_email) with mocked OpenAI
        print('Test email:', test_email.subject)
        print('Mock response parsed successfully')

test_extract_action_items_mock()

知识检查：多应用流水线

请测试您对构建多应用自动化流水线的理解。

流水线回顾

您已经构建了一个完整的多应用自动化流水线：检测 Gmail 邮件、基于 LLM 提取信息、创建 Trello 卡片，以及发送 Slack 通知。关键设计原则包括：通过清晰的接口分离各个阶段，在每个步骤进行可靠的错误处理，进行结构化日志记录，以及通过环境变量进行配置。

常见问题解答

「构建多应用自动化流程」课时是免费的吗？

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

「构建多应用自动化流程」这节课中我会学到什么？

通过智能体管理的工具调用串联 Gmail → Slack → Google Sheets 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

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

「构建多应用自动化流程」课时需要多长时间？

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

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

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

此课程中的所有课时

  1. 触发器—操作代理模式
  2. 将代理连接到 Webhook
  3. 基于调度与 Cron 的代理
  4. 构建多应用自动化流程
← 返回 AI Agents