AI Agents · 课时

触发器—操作代理模式

Q: 「触发器—操作代理模式」这节课中我会学到什么？

事件检测 → 决策 → 操作：自动化循环的核心。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

事件检测 → 决策 → 操作：自动化循环的核心。

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

什么是触发操作代理

触发操作代理会监视事件并执行相应操作。这个三部分循环是：检测事件 → LLM 决定操作 → 执行操作。

示例：收到电子邮件 → 进行摘要并回复；上传文件 → 验证并处理；每天上午 9 点 → 生成每日报告。

触发器类型

触发器分为三类：

基于事件：电子邮件到达或文件上传时触发网络钩子
基于时间：按照 cron 调度以固定间隔运行代理
基于轮询：代理反复检查 API 以获取新数据

选择正确的触发器类型会决定代理的延迟和资源使用情况。

检测阶段

检测意味着接收或识别事件。对于网络钩子，您的服务器会接收 POST 请求。对于轮询，您的代理会查询 API，并将结果与上次看到的状态进行比较。

import json

def detect_new_email(current_emails, last_seen_id):
    new_emails = [
        e for e in current_emails
        if e['id'] > last_seen_id
    ]
    return new_emails

# Simulate detection
current = [{'id': 3, 'subject': 'Meeting'}, {'id': 4, 'subject': 'Invoice'}]
new = detect_new_email(current, last_seen_id=2)
print('New emails:', [e['subject'] for e in new])

决策阶段

检测到事件后，代理会将上下文发送给 LLM，并询问应执行什么操作。LLM 要么选择一个工具，要么返回直接响应。

import openai

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

def decide_action(event_description):
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': 'You are an automation agent. Decide what action to take for the event.'},
            {'role': 'user', 'content': f'Event: {event_description}'}
        ],
        tools=[
            {'type': 'function', 'function': {'name': 'send_reply', 'description': 'Reply to email', 'parameters': {'type': 'object', 'properties': {'message': {'type': 'string'}}, 'required': ['message']}}}
        ]
    )
    return response.choices[0].message

result = decide_action('New email: Invoice for $500 from supplier')
print(result)

执行阶段

执行阶段会运行选定的操作。这可能包括调用 API、写入文件、发送消息或触发另一个工作流。请始终处理错误并记录结果。

import logging
import sys

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger('agent')

def execute_action(action_name, params):
    try:
        if action_name == 'send_reply':
            # In real code, call Gmail API here
            logger.info(f'Sending reply: {params["message"]}')
            return {'status': 'success'}
        elif action_name == 'create_task':
            logger.info(f'Creating task: {params["title"]}')
            return {'status': 'success'}
        else:
            raise ValueError(f'Unknown action: {action_name}')
    except Exception as e:
        logger.error(f'Action failed: {e}')
        return {'status': 'error', 'message': str(e)}

if __name__ == '__main__':
    result = execute_action('send_reply', {'message': 'Thanks for reaching out!'})
    print('Result:', result)

状态机模型

状态机是自动化代理的强大模型。状态可能包括：IDLE、DETECTING、DECIDING、EXECUTING、ERROR。状态转换会在事件或条件触发时发生。

状态机能让代理行为更可预测，也更容易调试。

from enum import Enum

class AgentState(Enum):
    IDLE = 'idle'
    DETECTING = 'detecting'
    DECIDING = 'deciding'
    EXECUTING = 'executing'
    ERROR = 'error'

class AutomationAgent:
    def __init__(self):
        self.state = AgentState.IDLE
    
    def transition(self, new_state):
        print(f'State: {self.state.value} -> {new_state.value}')
        self.state = new_state
    
    def run_cycle(self, event=None):
        self.transition(AgentState.DETECTING)
        if event:
            self.transition(AgentState.DECIDING)
            self.transition(AgentState.EXECUTING)
        self.transition(AgentState.IDLE)

agent = AutomationAgent()
agent.run_cycle(event={'type': 'email', 'subject': 'Test'})

收到电子邮件的触发模式

Gmail push 通知使用 Pub/Sub。收到新邮件时，Google 会向您的主题发布消息。您的代理接收网络钩子，获取并处理邮件。

from fastapi import FastAPI, Request
import base64, json

app = FastAPI()

@app.post('/gmail-push')
async def gmail_push(request: Request):
    body = await request.json()
    # Decode Pub/Sub message
    message = body.get('message', {})
    data = base64.b64decode(message.get('data', '')).decode('utf-8')
    notification = json.loads(data)
    
    email_address = notification.get('emailAddress')
    history_id = notification.get('historyId')
    
    print(f'New email for {email_address}, historyId: {history_id}')
    # Fetch email details and run agent here
    return {'status': 'ok'}

文件上传的触发模式

S3 事件通知或本地文件系统监视器可以在文件出现时触发代理。watchdog 库会监视目录中的新文件。

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class UploadHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f'New file detected: {event.src_path}')
        self.process_file(event.src_path)
    
    def process_file(self, filepath):
        # Run agent logic on new file
        print(f'Processing: {filepath}')

observer = Observer()
handler = UploadHandler()
observer.schedule(handler, path='/tmp/uploads/', recursive=False)
observer.start()

try:
    time.sleep(30)  # Watch for 30 seconds
finally:
    observer.stop()
    observer.join()

基于时间的触发模式

基于时间的触发器会按照调度运行代理。使用 APScheduler 进行进程内调度，或使用系统 cron 任务进行进程级调度。

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

scheduler = BlockingScheduler()

def daily_report_agent():
    print(f'Daily report running at {datetime.now()}')
    # Fetch data, call LLM, send report
    pass

def hourly_check_agent():
    print(f'Hourly check at {datetime.now()}')
    pass

# Run at 8am every day
scheduler.add_job(daily_report_agent, 'cron', hour=8, minute=0)

# Run every 30 minutes
scheduler.add_job(hourly_check_agent, 'interval', minutes=30)

print('Scheduler started')
scheduler.start()

幂等操作执行

自动化代理必须具备幂等性：同一个操作运行两次不应产生重复影响。请使用幂等键和执行前检查模式。

import hashlib

processed_events = set()  # In production, use Redis or DB

def compute_event_id(event):
    content = f"{event['type']}:{event['source_id']}:{event['timestamp']}"
    return hashlib.sha256(content.encode()).hexdigest()[:16]

def handle_event_idempotent(event):
    event_id = compute_event_id(event)
    
    if event_id in processed_events:
        print(f'Skipping duplicate event: {event_id}')
        return {'status': 'duplicate', 'event_id': event_id}
    
    # Process event
    print(f'Processing event: {event_id}')
    processed_events.add(event_id)
    return {'status': 'processed', 'event_id': event_id}

# Simulate duplicate event
event = {'type': 'email', 'source_id': 'abc123', 'timestamp': '2024-01-01T09:00:00'}
print(handle_event_idempotent(event))
print(handle_event_idempotent(event))  # Duplicate - skipped

错误状态与恢复

健壮的代理能够优雅地处理故障。执行失败时，代理可以使用指数退避进行重试、提醒人工处理，或将任务移入死信队列以供人工审核。

import time

def execute_with_retry(action_fn, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = action_fn()
            print(f'Success on attempt {attempt + 1}')
            return result
        except Exception as e:
            wait = 2 ** attempt  # Exponential backoff: 1s, 2s, 4s
            print(f'Attempt {attempt + 1} failed: {e}. Retrying in {wait}s')
            if attempt < max_retries - 1:
                time.sleep(wait)
            else:
                print('All retries exhausted. Moving to dead-letter queue.')
                raise

# Example usage
call_count = [0]

def flaky_action():
    call_count[0] += 1
    if call_count[0] < 3:
        raise ConnectionError('Service unavailable')
    return 'Done'

execute_with_retry(flaky_action)

知识检查：触发操作模式

请检验您对触发操作代理模式的理解。

整合所有部分

完整的触发操作代理会组合所有部分：触发源（电子邮件、文件、计时器）、检测层、基于 LLM 的决策、幂等执行、重试逻辑和状态跟踪。

从简单开始：一种触发器，一个操作。随着您逐渐熟悉代理的行为，再逐步增加复杂性。

常见问题解答

「触发器—操作代理模式」课时是免费的吗？

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

「触发器—操作代理模式」这节课中我会学到什么？

事件检测 → 决策 → 操作：自动化循环的核心。你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

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

「触发器—操作代理模式」课时需要多长时间？

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

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

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

此课程中的所有课时

← 返回 AI Agents