0Pricing
AI Agents · 课时

常驻智能体设计模式

后台进程、守护智能体和持久连接管理

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

什么是常驻代理

常驻代理会作为后台服务持续运行，等待事件并主动采取行动。与请求-响应式代理不同，它会在交互之间持续存在，并随时间维护状态。

守护进程模式

守护进程会在后台运行，不依赖任何终端会话。请使用 Python 的守护线程或进程管理器，以便终端关闭后代理仍能继续运行。

import threading
import time
import signal
import sys

shutdown_flag = threading.Event()

def agent_main_loop():
    print('Agent daemon started')
    while not shutdown_flag.is_set():
        try:
            # Agent work: check for events, process tasks
            perform_agent_cycle()
            shutdown_flag.wait(timeout=60)  # Sleep 60s, wakes on shutdown
        except Exception as e:
            print(f'Agent loop error: {e}')
            shutdown_flag.wait(timeout=5)  # Brief pause on error
    print('Agent daemon stopped')

def perform_agent_cycle():
    print(f'Agent cycle at {time.strftime("%H:%M:%S")}')
    # Check emails, process queue, run scheduled tasks

def handle_signal(signum, frame):
    print(f'Signal {signum} received, shutting down...')
    shutdown_flag.set()

# Register signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, handle_signal)
signal.signal(signal.SIGINT, handle_signal)

# Start as daemon thread
thread = threading.Thread(target=agent_main_loop, daemon=True)
thread.start()
print('Agent running in background')

看门狗：崩溃时自动重启

看门狗会监控代理进程，并在进程崩溃时重新启动它。这对生产环境至关重要：代理不可避免地会遇到意外错误，因此必须能够自动恢复。

import subprocess
import time
import logging

logger = logging.getLogger('watchdog')

class AgentWatchdog:
    def __init__(self, agent_script: str, max_restarts: int = 10, restart_delay: float = 5.0):
        self.agent_script = agent_script
        self.max_restarts = max_restarts
        self.restart_delay = restart_delay
        self.restart_count = 0
        self.process = None
    
    def start(self):
        self.process = subprocess.Popen(
            ['python', self.agent_script],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT
        )
        logger.info(f'Agent started (PID: {self.process.pid})')
    
    def run_forever(self):
        self.start()
        while True:
            return_code = self.process.wait()
            logger.warning(f'Agent exited with code {return_code}')
            
            if self.restart_count >= self.max_restarts:
                logger.error(f'Max restarts ({self.max_restarts}) reached. Stopping watchdog.')
                break
            
            self.restart_count += 1
            logger.info(f'Restarting agent (attempt {self.restart_count})...')
            time.sleep(self.restart_delay * self.restart_count)  # Backoff
            self.start()

print('AgentWatchdog defined')

持久 WebSocket 连接

如需实时传送事件，请维护持久 WebSocket 连接。如果连接断开，请自动重新连接——这是常驻代理面临的关键挑战。

import asyncio
import websockets
import json

async def persistent_websocket_connection(uri: str, on_message):
    backoff = 1
    max_backoff = 60
    
    while True:  # Reconnect forever
        try:
            print(f'Connecting to {uri}')
            async with websockets.connect(uri, ping_interval=30, ping_timeout=10) as ws:
                print('WebSocket connected')
                backoff = 1  # Reset backoff on successful connection
                
                async for raw_message in ws:
                    try:
                        message = json.loads(raw_message)
                        await on_message(message)
                    except json.JSONDecodeError:
                        print(f'Invalid JSON message: {raw_message[:100]}')
        
        except websockets.exceptions.ConnectionClosed as e:
            print(f'WebSocket closed: {e}. Reconnecting in {backoff}s')
        except Exception as e:
            print(f'WebSocket error: {e}. Reconnecting in {backoff}s')
        
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, max_backoff)  # Exponential backoff

async def handle_ws_message(message: dict):
    print(f'Received: {message}')

print('Persistent WebSocket connection function defined')

心跳检查

心跳可以确认代理仍在运行并处理任务。请每隔 N 秒发送一次心跳信号；如果心跳停止，看门狗就会知道代理已卡住或停止运行。

import threading
import time
from datetime import datetime

class HeartbeatMonitor:
    def __init__(self, max_silence_seconds: int = 300):
        self.last_heartbeat = datetime.utcnow()
        self.max_silence = max_silence_seconds
        self.lock = threading.Lock()
    
    def beat(self):
        with self.lock:
            self.last_heartbeat = datetime.utcnow()
    
    def is_alive(self) -> bool:
        with self.lock:
            silence = (datetime.utcnow() - self.last_heartbeat).total_seconds()
            return silence < self.max_silence
    
    def silence_seconds(self) -> float:
        with self.lock:
            return (datetime.utcnow() - self.last_heartbeat).total_seconds()

monitor = HeartbeatMonitor(max_silence_seconds=60)

def agent_with_heartbeat():
    while not shutdown_flag.is_set():
        # Send heartbeat at start of each cycle
        monitor.beat()
        
        # Do agent work
        perform_agent_cycle()
        time.sleep(30)

# External watchdog checks the monitor
def watchdog_check():
    while True:
        if not monitor.is_alive():
            print(f'ALERT: Agent silent for {monitor.silence_seconds():.0f}s')
            # Restart agent here
        time.sleep(30)

print('Heartbeat monitor defined')

优雅关闭

优雅关闭会在停止前完成正在进行的工作。它接收停止信号，阻止新工作启动，完成当前任务，保存状态，然后干净地退出。

import signal
import threading
from contextlib import contextmanager

class GracefulShutdown:
    def __init__(self, timeout: float = 30.0):
        self.should_stop = threading.Event()
        self.active_tasks = 0
        self.lock = threading.Lock()
        self.timeout = timeout
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)
    
    def _handle_signal(self, signum, frame):
        print(f'Shutdown signal received. Waiting for {self.active_tasks} active tasks...')
        self.should_stop.set()
    
    @contextmanager
    def task(self):
        if self.should_stop.is_set():
            raise RuntimeError('Shutdown in progress, not accepting new tasks')
        with self.lock:
            self.active_tasks += 1
        try:
            yield
        finally:
            with self.lock:
                self.active_tasks -= 1
    
    def wait_for_all_tasks(self):
        self.should_stop.wait()
        deadline = time.time() + self.timeout
        while self.active_tasks > 0 and time.time() < deadline:
            time.sleep(0.1)
        if self.active_tasks > 0:
            print(f'WARNING: Forced shutdown with {self.active_tasks} tasks still active')

shutdown = GracefulShutdown(timeout=30)
print('Graceful shutdown manager created')

处理断开连接和重新连接

常驻代理需要制定处理服务断开连接的策略：在服务中断期间缓冲事件，重新连接后重放错过的事件，并避免丢失断开连接期间到达的事件。

import asyncio
import redis
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

class DisconnectHandler:
    def __init__(self, buffer_key: str = 'agent:offline_buffer'):
        self.buffer_key = buffer_key
        self.connected = True
    
    def on_disconnect(self):
        self.connected = False
        print(f'Disconnected at {datetime.utcnow()}')
    
    def on_reconnect(self):
        self.connected = True
        print(f'Reconnected at {datetime.utcnow()}')
        self.replay_buffered_events()
    
    def handle_event(self, event: dict):
        if not self.connected:
            # Buffer events for later replay
            import json
            r.lpush(self.buffer_key, json.dumps(event))
            print(f'Event buffered (offline): {event["type"]}')
            return
        self.process_event(event)
    
    def replay_buffered_events(self):
        import json
        replayed = 0
        while True:
            raw = r.rpop(self.buffer_key)
            if not raw:
                break
            event = json.loads(raw)
            self.process_event(event)
            replayed += 1
        if replayed:
            print(f'Replayed {replayed} buffered events')
    
    def process_event(self, event: dict):
        print(f'Processing event: {event["type"]}')

handler = DisconnectHandler()
print('Disconnect handler created')

使用 systemd 进行进程管理

对于 Linux 生产环境部署，请使用 systemd 管理代理进程。它可以处理自动重启，将日志记录到 journald，并在系统启动时启动代理。

# /etc/systemd/system/my-agent.service
SERVICE_FILE = '''
[Unit]
Description=My AI Agent Service
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/agent
ExecStart=/home/ubuntu/venv/bin/python agent.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

# Environment variables
EnvironmentFile=/home/ubuntu/agent/.env

# Resource limits
MemoryLimit=1G
CPUQuota=50%

[Install]
WantedBy=multi-user.target
'''

# Deploy commands:
# sudo cp my-agent.service /etc/systemd/system/
# sudo systemctl daemon-reload
# sudo systemctl enable my-agent
# sudo systemctl start my-agent
# sudo systemctl status my-agent
# sudo journalctl -u my-agent -f  # Follow logs

print('Systemd service configuration defined')
print('Enables: auto-start on boot, auto-restart on crash, centralized logging')

跨重启的状态持久化

常驻代理必须保存自身状态，以便重启后从中断处继续运行。请定期在磁盘或 Redis 中保存检查点数据，并在状态发生重大变化后保存。

import json
import os
from datetime import datetime

CHECKPOINT_FILE = '/tmp/agent_checkpoint.json'

def save_checkpoint(state: dict):
    state['last_saved'] = datetime.utcnow().isoformat()
    with open(CHECKPOINT_FILE, 'w') as f:
        json.dump(state, f, indent=2)
    print(f'Checkpoint saved at {state["last_saved"]}')

def load_checkpoint() -> dict:
    if not os.path.exists(CHECKPOINT_FILE):
        print('No checkpoint found, starting fresh')
        return {}
    with open(CHECKPOINT_FILE) as f:
        state = json.load(f)
    print(f'Checkpoint loaded from {state.get("last_saved", "unknown")}')
    return state

# Agent startup
agent_state = load_checkpoint()
last_processed_id = agent_state.get('last_processed_email_id', 0)
print(f'Resuming from email ID: {last_processed_id}')

# After processing each email
agent_state['last_processed_email_id'] = last_processed_id + 1
if agent_state['last_processed_email_id'] % 10 == 0:  # Checkpoint every 10 items
    save_checkpoint(agent_state)

监控常驻代理

请跟踪常驻代理的关键指标：运行时长、每小时处理的事件数、错误率、内存使用量以及最近活动时间。您可以通过健康检查端点公开这些指标，或将其推送到监控服务。

from fastapi import FastAPI
from datetime import datetime
import psutil
import os

app = FastAPI()
start_time = datetime.utcnow()
events_processed = 0
last_event_time = None

@app.get('/health')
def health_check():
    process = psutil.Process(os.getpid())
    uptime_seconds = (datetime.utcnow() - start_time).total_seconds()
    
    last_active = None
    if last_event_time:
        last_active = (datetime.utcnow() - last_event_time).total_seconds()
    
    return {
        'status': 'ok',
        'uptime_seconds': round(uptime_seconds),
        'events_processed': events_processed,
        'memory_mb': round(process.memory_info().rss / 1024 / 1024, 1),
        'cpu_percent': process.cpu_percent(interval=1),
        'last_event_seconds_ago': round(last_active) if last_active else None,
        'timestamp': datetime.utcnow().isoformat()
    }

针对外部依赖的熔断器

常驻代理会与可能发生故障的外部服务交互。熔断器会在一段时间内停止向发生故障的服务发起调用，从而避免级联故障。

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = 'closed'      # Normal operation
    OPEN = 'open'          # Service down, not calling
    HALF_OPEN = 'half_open'  # Testing if service recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
    
    def call(self, fn, *args):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise RuntimeError('Circuit open: service unavailable')
        
        try:
            result = fn(*args)
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print('Circuit closed: service recovered')
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.threshold:
                self.state = CircuitState.OPEN
                print(f'Circuit opened after {self.failure_count} failures')
            raise

circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
print('Circuit breaker created')

知识检查：常驻代理

请测试您对常驻代理设计模式的理解。

常驻代理设计模式总结

可靠的常驻代理通常结合以下机制：使用信号处理的守护进程，以实现干净关闭；在崩溃时自动重启的看门狗；支持指数退避重新连接的持久 WebSocket 连接；用于检测进程卡住的心跳检查；用于完成进行中工作的优雅关闭；以及用于在重启后恢复的状态检查点。生产环境中的进程管理请使用 systemd 或 supervisord。

常见问题解答

「常驻智能体设计模式」课时是免费的吗？

是的 — 「常驻智能体设计模式」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 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 反馈 — 无需本地设置。

此课程中的所有课时

  1. 常驻智能体设计模式
  2. 主动通知与提醒系统
  3. 跨会话持久化上下文
  4. 构建每日简报智能体
← 返回 AI Agents