이벤트 기반 AI 에이전트: LLM에 귀를 달아라
Source: Dev.to
왜 “프롬프트 → 응답”이 잘못된 사고 모델인가
LangChain과 ChatGPT가 폭발적으로 인기를 끌면서, 모두가 챗봇을 만들었습니다. 상호작용 모델은 동기식이었습니다: 사람이 메시지를 보내고, 모델이 답변하고, 루프가 끝납니다. 깔끔하고 이해하기 쉬우나 자동화에는 완전히 잘못된 방식입니다.
문제는 가장 가치 있는 작업이 대화 사이에서 일어난다는 점입니다.
- 당신의 CRM이 새 리드를 오전 2시에 받는다.
- 결제가 실패하면 웹훅이 트리거된다.
- 크론 작업이 사이트에서 깨진 링크를 찾는다.
이러한 경우는 인간이 무언가를 타이핑하는 것으로 시작되지 않습니다.
이벤트‑드리븐 에이전트는 모델을 뒤집습니다: 요청받기를 기다리는 대신, 신호를 청취하고 자율적으로 행동합니다. LLM은 더 큰 반응형 시스템 안의 추론 엔진이 되며, 진입점이 아닙니다.
이렇게 생각해 보세요 — 당신의 에이전트는 로그인 상태를 유지하는 팀원과 같습니다. 그들은 대기열을 감시하고, 들어오는 작업을 처리하며, 정말 막혔을 때만 에스컬레이션합니다.
실제로 필요한 세 가지 기본 요소
시작하기 위해 전체 오케스트레이션 프레임워크가 필요하지 않습니다. 다음 세 가지만 있으면 됩니다:
- 트리거 레이어 – 이벤트를 생성합니다 (웹훅 엔드포인트, 크론 작업, 파일 감시자, 큐 컨슈머 등).
- 라우팅 레이어 – 어떤 에이전트(또는 어떤 도구)가 어떤 이벤트를 처리할지 결정합니다. 이것은 마법이 아니라 단순한 디스패처입니다.
- 액션 레이어 – 실제 에이전트: LLM 호출 + 도구 + 선택적 출력(API 호출, DB 쓰기, Slack 메시지 등).
최소 이벤트‑드리븐 에이전트 스켈레톤
# app.py
from fastapi import FastAPI, Request
import asyncio
app = FastAPI()
async def run_agent(event: dict):
event_type = event.get("type")
if event_type == "new_lead":
await enrich_and_score(event["data"])
elif event_type == "payment_failed":
await draft_recovery_email(event["data"])
@app.post("/webhook")
async def handle_webhook(request: Request):
payload = await request.json()
# fire‑and‑forget – return 200 immediately
asyncio.create_task(run_agent(payload))
return {"status": "accepted"}Note:
create_taskmakes the handler non‑blocking. If your webhook receiver blocks, upstream services will time out and retry, which can cause duplicate agent runs at odd hours.
프레임워크 비용 없이 멀티‑에이전트 라우팅
에이전트가 두 개 이상일 때는 라우팅이 필요합니다. 대부분의 프레임워크는 유연성을 해치는 추상화를 추가합니다. 여기 가볍고 벤더에 종속되지 않는 접근법이 있습니다:
# router.py
AGENT_REGISTRY = {
"new_lead": LeadEnrichmentAgent(),
"payment_failed": RecoveryAgent(),
"mention": MonitoringAgent(),
"form_submit": OnboardingAgent(),
}
async def dispatch(event: dict):
agent = AGENT_REGISTRY.get(event["type"])
if agent:
await agent.run(event["data"])
else:
await fallback_agent.run(event)각 에이전트는 run() 메서드를 가진 단순 클래스입니다. 상속 체인이나 마법 데코레이터가 없습니다. pytest와 모의 이벤트 딕셔너리를 사용해 각각을 독립적으로 테스트할 수 있습니다.
트래픽이 많아지면, 딕셔너리 디스패치를 적절한 큐(Redis Streams, SQS, 심지어 폴링을 이용한 SQLite)로 교체하세요. 에이전트 코드는 그대로 유지되고 — 피더만 변경됩니다.
프로덕션 시스템처럼 실패를 처리하기
튜토리얼에서는 항상 이 부분을 건너뛰죠: 에이전트가 실패한다. API가 타임아웃된다. LLM이 도구 호출을 환각한다. 하위 웹훅이 다운된다.
첫날부터 실패에 대비해 구축하세요.
멱등성 키
처리된 이벤트 ID를 저장하세요. 동일한 웹훅이 두 번 발생하면(반드시 발생합니다), 중복을 건너뛰세요.
# idempotency.py
import sqlite3
DB_PATH = "agent_state.db"
def already_processed(event_id: str) -> bool:
conn = sqlite3.connect(DB_PATH)
row = conn.execute(
"SELECT 1 FROM processed_events WHERE id = ?", (event_id,)
).fetchone()
conn.close()
return row is not None
def mark_processed(event_id: str):
conn = sqlite3.connect(DB_PATH)
conn.execute(
"INSERT OR IGNORE INTO processed_events (id) VALUES (?)", (event_id,)
)
conn.commit()
conn.close()데드레터 큐
이벤트가 세 번 실패하면 failed_events 테이블로 라우팅하여 사람의 검토를 받게 하세요. 오류를 조용히 무시하지 마세요.
LLM 호출에 대한 타임아웃
client.messages.create()(또는 동등한 함수)를 타임아웃과 함께 감싸세요. 영원히 멈춰 있는 에이전트는 전체 큐를 차단합니다.
이 세 가지 패턴만으로도 주말 해커톤에서 만든 어떤 것보다 에이전트를 10배 더 신뢰할 수 있게 만들 것입니다.
수동 모니터링: 요청 없이 감시하는 에이전트
가장 강력한 활용 사례는 모니터링입니다 — 데이터 소스를 지속적으로 관찰하고 조건이 충족되면 행동하는 에이전트입니다.
Example: 귀사의 제품 서브레딧을 감시하고 브랜드가 부정적인 감정으로 언급될 때마다 답변 초안을 작성하는 에이전트.
import asyncio
async def reddit_monitor_loop():
seen = set()
while True:
mentions = fetch_recent_mentions(subreddit="yourproduct")
for post in mentions:
if post.id not in seen:
seen.add(post.id)
if is_negative_sentiment(post.title):
await draft_reply(post)
await asyncio.sleep(300) # poll every 5 minutesfetch_recent_mentions– Reddit에 대한 API 호출.is_negative_sentiment– LLM 또는 감정 분석 모델.draft_reply– LLM이 생성한 초안으로, 자동으로 게시하거나 검토를 위해 Slack 채널에 보낼 수 있습니다.
마무리
- Event‑driven은 프로덕션 급 AI 에이전트를 위한 올바른 사고 모델입니다.
- 필요한 기본 요소는 세 가지뿐입니다: trigger → router → action.
- 라우팅은 단순하고 테스트 가능하도록 유지하세요; 정말 필요하지 않은 한 무거운 프레임워크는 피하세요.
- 처음부터 멱등성, 데드레터 처리, 타임아웃을 구축하세요.
이러한 빌딩 블록을 사용하면 “프롬프트 → 응답” 챗봇을 실제 스택에 존재하는 자율적인 어시스턴트로 바꿔 매일 실질적인 가치를 제공할 수 있습니다.
if post["id"] not in seen:
seen.add(post["id"])
sentiment = await classify_sentiment(post["text"])
if sentiment == "negative":
draft = await draft_response(post)
await notify_slack(draft, post["url"])
await asyncio.sleep(300) # poll every 5 minutes이 코드는 백그라운드 작업으로 영원히 실행됩니다. 인간이 트리거하지 않습니다. 그냥… 작동합니다. 아침에 일어나면 모든 부정적인 언급을 보여주는 Slack 메시지와 바로 게시할 수 있는 응답 초안이 표시됩니다.
그것이 event‑driven agents의 약속입니다 — 질문에 답하는 AI가 아니라 일을 수행하는 AI.
Over‑engineering 없이 배포하기
Kubernetes는 필요 없습니다. 대부분의 인디 프로젝트와 초기 스타트업에게는 nohup과 프로세스 감시자를 갖춘 단일 VPS면 충분합니다:
nohup python3 -u agent_worker.py > logs/agent.log 2>&1 &프로세스가 충돌하면 재시작하도록 감시자(또는 cron 작업)를 실행하세요. 모든 로그를 tail로 확인할 수 있는 파일에 기록합니다. 마지막으로 처리된 이벤트의 타임스탬프를 반환하는 /health 엔드포인트를 추가하세요 — 타임스탬프가 오래되면 루프가 멈춘 것을 알 수 있습니다.
이 설정의 한계에 실제로 도달했을 때만 인프라 복잡성을 추가하세요. 대부분의 팀은 절대 그렇게 하지 않습니다.
함정은 당신이 가지고 있지 않은 규모에 맞춰 설계하는 것입니다. 월 $6 VPS에서 실행되며 하루에 1,000개의 이벤트를 안정적으로 처리하는 이벤트‑드리븐 에이전트는 세 번의 스프린트가 걸리는 오케스트레이션 거대 시스템보다 훨씬 더 큰 가치를 가집니다.
시작은 지루하게. 트리거를 자동화하고, 에이전트를 배포하세요. 문제가 요구할 때만 확장하면 됩니다.
모든 내용을 실용적인 가이드로 정리했습니다 — 패턴, 운영 시 흔히 겪는 함정, 완전한 코드 예제 포함:
Event‑Driven AI Agents: Give Your LLM Ears