목차
- 에이전트를 한 덩어리로 짜면 안 되는 이유
- 작업 분해 — AI 에이전트 자동화의 시작점
- 툴 연동에서 부딪힌 것들
- 오류 복구 — 파이프라인이 뻗지 않으려면
- cron 등록과 비용 확인
경쟁사 기술 블로그 모니터링을 Python AI 에이전트로 자동화했다. 블로그 5곳을 읽고 키워드를 뽑아 슬랙에 요약을 올리는 작업인데, 수동으로 하면 주당 3~4시간이 들었다.
이틀 만에 파이프라인이 나왔는데, 절반은 설계를 잘못 잡아서 다시 짠 시간이었다. 코드보다 구조 설계가 핵심이다.
에이전트를 한 덩어리로 짜면 안 되는 이유
처음에는 단순하게 생각했다. main.py 하나에 Claude API 호출, 응답 파싱, 슬랙 전송을 몽땅 넣으면 된다고.
문제는 중간 단계에서 터졌을 때 드러났다. 블로그 5개 중 3개까지 수집·요약이 끝난 상태에서 4번째 URL이 403을 뱉으면 스크립트 전체가 죽는다. 이미 처리한 3개 결과도 날아간다. 처음부터 다시 돌려야 하고, API 비용도 이중으로 나간다.
데이터 파이프라인 설계에서 기본 중의 기본인 문제다. 에이전트를 처음 만들 때는 "LLM 호출 한 번이면 끝나겠지"라는 착각 때문에 이걸 간과하기 쉽다. LLM이 만능처럼 보여서 하나의 프롬프트에 모든 걸 욱여넣으려는 유혹이 있다. 실제로 "블로그 수집하고 요약하고 키워드 뽑아서 슬랙 메시지로 만들어줘"라고 한 번에 시켰더니, Claude 3.5 Sonnet 기준 입력 토큰만 8,000~12,000이 나왔고 출력도 2,000 토큰을 넘겼다. 비용 이전에, 한 번의 호출이 실패하면 전부 재실행해야 한다는 게 치명적이었다.
작업 분해 — AI 에이전트 자동화의 시작점
LangChain 0.3의 AgentExecutor를 먼저 시도해봤다 (출처: LangChain Python 공식 문서 v0.3, 2025년 하반기 릴리즈). tool calling 기능으로 에이전트가 알아서 필요한 도구를 골라 쓰게 하는 구조다. 의도대로 안 됐다. 에이전트가 tool을 호출하는 순서를 예측할 수 없었고, 같은 tool을 불필요하게 반복 호출하는 경우도 있었다. 비용이 2~3배로 뛰었다.
"에이전트가 알아서 판단하게" 대신 "내가 순서를 정하고 각 단계만 에이전트한테 맡기자"로 갈아탔다. 이걸 보통 오케스트레이터 패턴이라고 부른다. 전체 흐름은 Python 코드가 제어하고, LLM은 각 단계에서 도구로만 쓰는 방식이다. 작업을 이렇게 나눴다:
- 수집(Collect): 대상 URL에서 본문 텍스트 추출. LLM 불필요.
httpx+BeautifulSoup으로 처리. - 요약(Summarize): 각 본문을 300자 내외로 요약. Claude API 호출.
- 분석(Analyze): 요약 결과를 종합해서 키워드 추출 + 트렌드 코멘트 생성. Claude API 호출.
- 포맷(Format): 슬랙 마크다운으로 변환. 템플릿 기반이라 LLM 불필요.
- 전송(Send): 슬랙 Webhook으로 전송.
5단계 중 LLM이 개입하는 건 2번과 3번뿐이다. 나머지는 일반 Python 코드로 처리한다. 이렇게 나누면 각 단계의 입출력이 명확해지고, 중간 결과를 파일로 저장할 수 있다. 2번에서 실패하면 2번만 다시 돌리면 된다.
분리 기준은 하나다
실패했을 때 독립적으로 재시도할 수 있는가. 이게 안 되면 분리가 덜 된 거다.
# pipeline/state.py
from dataclasses import dataclass, field
import json
from pathlib import Path
@dataclass
class PipelineState:
"""파이프라인 중간 상태를 들고 다니는 객체"""
urls: list[str] = field(default_factory=list)
raw_texts: dict[str, str] = field(default_factory=dict) # url -> 본문
summaries: dict[str, str] = field(default_factory=dict) # url -> 요약
analysis: str = "" # 종합 분석
formatted_message: str = "" # 슬랙 메시지
def save_checkpoint(self, step_name: str):
"""단계 완료 후 중간 결과 저장"""
checkpoint_dir = Path("checkpoints")
checkpoint_dir.mkdir(exist_ok=True)
with open(checkpoint_dir / f"{step_name}.json", "w") as f:
json.dump({
"urls": self.urls,
"raw_texts": self.raw_texts,
"summaries": self.summaries,
"analysis": self.analysis,
}, f, ensure_ascii=False, indent=2)
PipelineState 하나가 파이프라인 전체를 관통한다. 각 단계 함수는 이 객체를 받아서 자기 필드만 채우고, 체크포인트를 저장한다. 3번 단계에서 API 에러가 나면 checkpoints/summarize.json에서 2번 결과를 불러와 3번부터 재시도하면 된다. 단순한 구조인데 이게 없었을 때와 있을 때의 차이가 크다.
LangChain Agent vs 직접 오케스트레이션
LangChain Agent가 나쁘다는 게 아니다. 사용자 입력이 매번 달라지고 어떤 도구를 써야 할지 동적으로 결정해야 하는 상황—챗봇이 대표적—에서는 Agent가 맞다.
내 경우처럼 "매주 같은 작업을 반복"하는 자동화에서는 Agent의 자율성이 오히려 리스크였다. 어떤 주에는 tool을 3번 호출하고, 어떤 주에는 7번 호출한다. 비용 예측이 안 된다. Anthropic API 기준으로 Claude Sonnet 계열의 입력 토큰 가격이 $3/MTok, 출력이 $15/MTok 수준이다 (출처: Anthropic 모델 가격 페이지, 2026년 3월 기준). 자율 에이전트가 토큰을 예측 불가능하게 소모하면 월말 청구서가 예상을 넘긴다. 정해진 워크플로를 반복하는 자동화에는 오케스트레이터 패턴이 낫다는 게 결론이다.
툴 연동에서 부딪힌 것들
각 단계를 함수로 분리한 뒤, 외부 서비스 연동에서 예상 못 한 문제가 터졌다.
웹 스크래핑: 403과 그에 대한 판단
httpx로 블로그 본문을 가져오는 건 간단하다. 근데 5개 URL을 asyncio로 동시에 요청하면 일부 사이트에서 403이 뜬다. 순차 요청 + 1초 딜레이로 바꿨다. 주 1회 배치 작업에 동시성 최적화는 의미 없다.
여기서 내린 판단 하나가 이후 안정성을 크게 바꿨다. 실패한 URL은 None으로 남기고 다음 단계에서 건너뛰게 한 거다. 처음에는 "하나라도 실패하면 전체 중단" 로직을 썼는데, 매주 1~2개는 403이 뜨는 바람에 파이프라인이 매번 멈췄다. 5개 중 4개만 성공해도 리포트는 충분히 쓸 만하다. 완벽한 결과보다 매주 돌아가는 게 먼저다.
프롬프트 설계: 구조화가 답이었다
요약 단계에서 처음엔 "다음 글을 300자로 요약해줘"라고만 시켰다. 결과가 들쭉날쭉했다. 어떤 건 200자, 어떤 건 500자. 핵심을 빠뜨리기도 했다. 프롬프트를 구조화하니 안정됐다:
SUMMARIZE_PROMPT = """아래 기술 블로그 글을 요약하라.
규칙:
- 250~350자 사이로 작성
- 핵심 기술 키워드를 반드시 포함
- 주관적 평가 제외, 사실만 서술
- JSON 형식: {{"summary": "...", "keywords": ["...", "..."]}}
본문:
{text}
"""
JSON 반환을 지정하면 파싱이 편하다. Anthropic Python SDK anthropic 패키지 v1.26 이상 기준으로 response.content[0].text에서 텍스트를 꺼내고 json.loads로 파싱한다. 간혹 Claude가 마크다운 코드블록으로 감싸서 반환하는 경우가 있어서, ```json 태그를 벗기는 전처리를 한 줄 추가했다. 이런 케이스는 공식 문서에 안 나온다. 직접 돌려봐야 아는 부분이다.
슬랙 Webhook
슬랙 Incoming Webhook의 페이로드 제한은 40,000자다 (출처: Slack API 메시지 페이로드 문서). 보통은 넘길 일이 없지만, 안전장치로 len(message) > 35000이면 요약을 잘라서 재포맷하는 로직을 넣었다. 이건 간단하다. 5분이면 붙인다.
오류 복구 — 파이프라인이 뻗지 않으려면
자동화에서 가장 중요한 건 "매주 돌아가는 것"이다. 완벽한 결과가 아니라, 빠지지 않고 실행되는 것. API 타임아웃은 정상 동작이다. anthropic.APITimeoutError: Request timed out. — 이 에러를 처음 봤을 때는 코드가 잘못된 줄 알았는데, 기본 타임아웃이 10분이라 웬만해선 안 뜬다. 뜬다면 네트워크 문제거나 서버 쪽 부하다. 이걸 전제로 설계해야 한다.
tenacity 라이브러리를 쓰면 재시도 로직을 데코레이터 하나로 붙일 수 있다.
from tenacity import retry, stop_after_attempt, wait_exponential
import anthropic
client = anthropic.Anthropic()
@retry(
stop=stop_after_attempt(3), # 최대 3회
wait=wait_exponential(min=2, max=30) # 2초 → 4초 → 8초 대기
)
def call_claude(prompt: str) -> str:
"""Claude API 호출. 실패 시 지수 백오프로 재시도."""
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
wait_exponential이 핵심이다. 첫 실패 후 2초, 두 번째 후 4초를 기다린다. API 서버가 과부하일 때 즉시 재요청하면 또 실패할 확률이 높다. 지수 백오프로 서버 회복 시간을 주는 거다. 처음에 tenacity 없이 for i in range(3): try: ... except: time.sleep(2) 형태로 직접 짰었는데, 예외 필터링이나 대기 시간 계산을 매번 수동으로 해야 해서 금방 tenacity로 갈아탔다.
fallback: 이전 결과 재사용
3번 재시도해도 실패하면 두 가지 선택이 있다. 파이프라인을 멈추거나, 이전에 성공한 결과를 쓰거나. 후자를 택했다. 주간 리포트가 한 주 통째로 빠지는 것보다 지난주 요약이라도 포함된 리포트가 나가는 게 낫다. 체크포인트 파일에서 직전 성공 결과를 불러와서 "(지난주 데이터)" 태그를 붙여 포함시킨다.
이 판단은 도메인에 따라 완전히 달라진다. 금융 데이터였으면 절대 이전 값을 재사용하면 안 된다. 내 경우는 경쟁사 블로그 요약이라 일주일 전 데이터가 섞여도 문제가 없었다.
로깅은 처음부터 structured로
파이프라인이 cron으로 새벽에 돌다 보니, 실패 원인을 나중에 추적할 수 있어야 한다. 처음엔 print문으로 로깅했다. 부끄럽지만 사실이다. 로그 파일에서 에러를 눈으로 찾는 데 시간을 너무 썼다. structlog로 바꾸니 JSON 포맷 로그가 나오고 jq로 필터링이 된다. log.error("api_timeout", url=url, attempt=3, fallback="previous_cache") 형태로 기록하면 나중에 jq 'select(.event == "api_timeout")'으로 타임아웃만 뽑아볼 수 있다. 이걸 나중에 바꾸려면 모든 함수를 건드려야 해서, 처음부터 잡는 게 맞다.
cron 등록과 비용 확인
완성된 파이프라인은 crontab에 0 7 * * 1로 등록했다. 매주 월요일 오전 7시 실행이다.
비용을 Anthropic 대시보드에서 확인해봤다. 주 1회 실행에 요약 5건 + 분석 1건, 입력 토큰 약 15,000 + 출력 토큰 약 3,000. Claude 3.5 Sonnet 기준 1회 실행 비용이 대략 $0.09. 월 $0.36이다. "API 비용이 많이 나오면 어쩌지"를 걱정했었는데, 정해진 워크플로를 반복하는 배치 작업은 비용이 아주 낮다. 자율 에이전트가 tool을 마음대로 호출하는 구조와는 차원이 다르다.
Python 3.12부터 표준 라이브러리에 들어온 tomllib으로 설정을 분리하고 있다. 대상 URL 목록, 프롬프트 템플릿, 슬랙 Webhook URL을 config.toml에 넣어두면 코드 수정 없이 설정만 바꿀 수 있다. 처음엔 전부 하드코딩했다가 "URL 하나 추가할 때마다 코드를 고쳐야 하나"라는 생각에 분리한 건데, 이건 진작에 했어야 했다.
현재 이 파이프라인은 3주째 매주 돌아가고 있다. 완전히 실패한 적은 한 번도 없고, 403 때문에 URL 1~2개가 빠진 주가 2번 있었다. 그 경우에도 나머지 글의 요약은 정상적으로 슬랙에 도착했다. 다음에는 수집 소스를 RSS 피드로 전환하고 요약 결과를 Notion DB에도 적재하는 걸 이 AI 에이전트 자동화 파이프라인에 붙여볼 생각이다.
