상시 실행 에이전트 설계 패턴
백그라운드 프로세스, 데몬 에이전트, 지속 연결 관리를 학습합니다.
상시 실행 에이전트 설계 패턴은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 1번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 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를 사용하여 에이전트 프로세스를 관리하십시오. 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를 사용하십시오.
자주 묻는 질문
“상시 실행 에이전트 설계 패턴” 강의는 무료인가요?
네 — “상시 실행 에이전트 설계 패턴” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“상시 실행 에이전트 설계 패턴”에서 뭘 배우나요?
백그라운드 프로세스, 데몬 에이전트, 지속 연결 관리를 학습합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 1번째 강의입니다.
“상시 실행 에이전트 설계 패턴” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 상시 실행 에이전트 설계 패턴
- 선제적 알림 및 경고 시스템
- 세션 간 맥락 유지
- 일일 브리핑 에이전트 구축