Chapter 12: 30일 AI 에이전트 완성 로드맵
"여정의 끝이 아닌, 진짜 시작점에 서 있습니다."
이 책의 모든 챕터를 읽으셨습니다. 축하드립니다. 하지만 솔직히 말씀드리면, 책을 읽는 것만으로는 아무것도 변하지 않습니다. 지식은 실행될 때만 가치가 있습니다.
이번 챕터에서는 지난 11개 챕터에서 배운 모든 내용을 30일 안에 실제 운용 가능한 시스템으로 만드는 구체적인 로드맵을 제시합니다. 매일 무엇을 하고, 얼마나 걸리고, 막혔을 때 어떻게
해결하는지—모든 것이 여기 있습니다.
🎯 30일 후, 당신이 갖게 될 것
로드맵을 시작하기 전에, 30일 후의 모습을 명확히 그려봅시다:
| 영역 |
구축되는 것 |
기대 효과 |
| 수집 자동화 |
뉴스/경쟁사/커뮤니티 모니터링 봇 3종 |
주 5시간 → 0시간 |
| 콘텐츠 생성 |
블로그/SNS/뉴스레터 초안 봇 5종 |
월 20시간 → 2시간 |
| 분석 시스템 |
트렌드/감정/경쟁 분석 봇 4종 |
컨설팅 비용 월 100만원 절감 |
| 운영 자동화 |
보고서/요약/알림 봇 5종 |
주 3시간 → 0시간 |
| 의사결정 지원 |
리서치/비교/추천 봇 4종 |
의사결정 속도 3배 향상 |
| 인프라 |
클라우드 배포 + 모니터링 + HITL |
24/7 무중단 운영 |
총 21개 봇 + 완전한 운영 시스템 = 월 40시간 이상 절약 + 비용 절감
📅 1주차 (1-7일): 기초 다지기
첫 주는 기초 체력을 기르는 시간입니다. 서두르지 마세요. 여기서 단단히 다져야 나머지 3주가 순탄합니다.
Day 1: 환경설정 완료 + 첫 에이전트 실행 ⏱️ 2-3시간
오전: 개발 환경 구축
# 1. Python 가상환경 생성
python -m venv ai-agent-env
source ai-agent-env/bin/activate # Windows: ai-agent-env\Scripts\activate
# 2. 핵심 라이브러리 설치
pip install anthropic python-dotenv requests beautifulsoup4
# 3. 프로젝트 구조 생성
mkdir -p ai-agents/{agents,tools,memory,config,logs}
touch ai-agents/.env ai-agents/main.py
오후: 첫 에이전트 실행
# ai-agents/agents/first_agent.py
import anthropic
from dotenv import load_dotenv
import os
load_dotenv()
def create_first_agent():
"""당신의 첫 번째 AI 에이전트"""
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system="당신은 친절한 AI 어시스턴트입니다.",
messages=[
{"role": "user", "content": "안녕하세요! 자기소개 부탁드려요."}
]
)
print("🤖 첫 에이전트 응답:")
print(response.content[0].text)
return response
if __name__ == "__main__":
create_first_agent()
Day 1 체크리스트
- [ ] Python 3.10+ 설치 확인
- [ ] 가상환경 생성 및 활성화
- [ ] Anthropic API 키 발급 및 .env 설정
- [ ] 첫 에이전트 실행 성공
- [ ] Git 저장소 초기화
Day 2: ReAct 루프 이해 + 도구 추가 ⏱️ 3-4시간
핵심 개념 구현
# ai-agents/agents/react_agent.py
import anthropic
import json
from datetime import datetime
class ReActAgent:
"""ReAct 패턴을 구현한 기본 에이전트"""
def __init__(self):
self.client = anthropic.Anthropic()
self.tools = self._define_tools()
def _define_tools(self):
return [
{
"name": "get_current_time",
"description": "현재 날짜와 시간을 반환합니다.",
"input_schema": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "calculate",
"description": "수학 계산을 수행합니다.",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "계산할 수학 표현식 (예: 2+2, 100*5)"
}
},
"required": ["expression"]
}
}
]
def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
"""도구 실행"""
if tool_name == "get_current_time":
return datetime.now().strftime("%Y년 %m월 %d일 %H시 %M분")
elif tool_name == "calculate":
try:
# 안전한 계산만 허용
allowed_chars = set('0123456789+-*/.() ')
expr = tool_input.get("expression", "")
if all(c in allowed_chars for c in expr):
result = eval(expr)
return str(result)
return "허용되지 않는 표현식입니다."
except Exception as e:
return f"계산 오류: {str(e)}"
return "알 수 없는 도구입니다."
def run(self, user_message: str) -> str:
"""에이전트 실행 (ReAct 루프)"""
messages = [{"role": "user", "content": user_message}]
while True:
response = self.client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system="당신은 도구를 활용하는 AI 어시스턴트입니다.",
tools=self.tools,
messages=messages
)
# 응답 처리
if response.stop_reason == "end_turn":
# 최종 응답
for block in response.content:
if hasattr(block, 'text'):
return block.text
elif response.stop_reason == "tool_use":
# 도구 사용
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f"🔧 도구 실행: {block.name}")
result = self._execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# 대화 이력 업데이트
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
return "예상치 못한 응답입니다."
# 테스트
if __name__ == "__main__":
agent = ReActAgent()
# 테스트 1: 시간 확인
print(agent.run("지금 몇 시야?"))
# 테스트 2: 계산
print(agent.run("1234 * 5678은 얼마야?"))
Day 2 체크리스트
- [ ] ReAct 패턴 개념 이해 (Think → Act → Observe)
- [ ] Tool Use 구조 이해
- [ ] 기본 도구 2개 구현 (시간, 계산)
- [ ] 에이전트 루프 동작 확인
Day 3-5: 수집봇 3종 완성 ⏱️ 각 3-4시간
Day 3: 뉴스 수집봇
# ai-agents/agents/news_collector.py
import requests
from bs4 import BeautifulSoup
from datetime import datetime
import json
class NewsCollector:
"""네이버 뉴스 수집봇"""
def __init__(self, keywords: list):
self.keywords = keywords
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
def collect(self, keyword: str, max_articles: int = 5) -> list:
"""키워드로 뉴스 수집"""
url = f"https://search.naver.com/search.naver?where=news&query={keyword}"
try:
response = requests.get(url, headers=self.headers, timeout=10)
soup = BeautifulSoup(response.text, 'html.parser')
articles = []
news_items = soup.select('.news_tit')[:max_articles]
for item in news_items:
articles.append({
"title": item.get_text(),
"url": item.get('href'),
"keyword": keyword,
"collected_at": datetime.now().isoformat()
})
return articles
except Exception as e:
print(f"수집 오류: {e}")
return []
def collect_all(self) -> list:
"""모든 키워드 수집"""
all_articles = []
for keyword in self.keywords:
articles = self.collect(keyword)
all_articles.extend(articles)
print(f"✅ '{keyword}' 수집 완료: {len(articles)}건")
return all_articles
def save_results(self, articles: list, filename: str = "news_results.json"):
"""결과 저장"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"💾 저장 완료: {filename}")
# 사용 예시
if __name__ == "__main__":
collector = NewsCollector(["AI 에이전트", "ChatGPT", "Claude AI"])
articles = collector.collect_all()
collector.save_results(articles)
Day 4: 경쟁사 모니터링봇
# ai-agents/agents/competitor_monitor.py
import requests, hashlib, json, os
from bs4 import BeautifulSoup
from datetime import datetime
class CompetitorMonitor:
"""경쟁사 웹사이트 변경 감지봇"""
def __init__(self, targets: list):
"""targets: [{"name": "회사명", "url": "URL", "selector": "CSS선택자"}]"""
self.targets = targets
self.headers = {"User-Agent": "Mozilla/5.0"}
self.cache_file = "competitor_cache.json"
self.cache = self._load_cache()
def _load_cache(self) -> dict:
if os.path.exists(self.cache_file):
with open(self.cache_file, 'r') as f:
return json.load(f)
return {}
def _save_cache(self):
with open(self.cache_file, 'w') as f:
json.dump(self.cache, f)
def _get_content_hash(self, url: str, selector: str) -> tuple:
try:
resp = requests.get(url, headers=self.headers, timeout=10)
soup = BeautifulSoup(resp.text, 'html.parser')
el = soup.select_one(selector)
text = el.get_text(strip=True) if el else ''
return hashlib.md5(text.encode()).hexdigest(), text
except Exception as e:
return None, str(e)
def check_all(self) -> list:
"""변경된 경쟁사 페이지 반환"""
changed = []
for t in self.targets:
new_hash, content = self._get_content_hash(t["url"], t["selector"])
old_hash = self.cache.get(t["url"])
if new_hash and new_hash != old_hash:
changed.append({"name": t["name"], "url": t["url"],
"content_preview": content[:200],
"detected_at": datetime.now().isoformat()})
self.cache[t["url"]] = new_hash
print(f"🔴 변경 감지: {t['name']}")
else:
print(f"✅ 변경 없음: {t['name']}")
self._save_cache()
return changed
if __name__ == "__main__":
monitor = CompetitorMonitor([
{"name": "경쟁사A", "url": "https://competitor-a.com/pricing", "selector": ".price-table"},
{"name": "경쟁사B", "url": "https://competitor-b.com/blog", "selector": ".post-list"},
])
changes = monitor.check_all()
if changes:
print(f"\n🚨 {len(changes)}개 변경 감지 → 텔레그램 보고 필요")
Day 5: SNS 키워드 트래킹봇
# ai-agents/agents/sns_keyword_tracker.py
import anthropic, os, json
from datetime import datetime
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
KEYWORDS = ["AI 에이전트", "Claude API", "자동화 봇", "n8n"]
def analyze_trend_with_claude(keyword: str, sample_posts: list) -> dict:
"""Claude로 키워드 트렌드 분석"""
posts_text = "\n".join([f"- {p}" for p in sample_posts[:10]])
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""키워드 '{keyword}'에 대한 소셜 반응을 분석하세요.
게시물 샘플:
{posts_text}
아래 JSON 형식으로만 응답하세요:
{{"sentiment": "긍정/중립/부정", "trend": "상승/유지/하락",
"key_topics": ["주요 토픽1", "주요 토픽2"],
"action_suggestion": "콘텐츠 전략 제안 1줄"}}"""
}]
)
try:
return json.loads(response.content[0].text)
except:
return {"sentiment": "분석불가", "trend": "분석불가",
"key_topics": [], "action_suggestion": "재시도 필요"}
def save_report(results: list):
filename = f"sns_trend_{datetime.now().strftime('%Y%m%d')}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"📊 트렌드 리포트 저장: {filename}")
if __name__ == "__main__":
results = []
for kw in KEYWORDS:
# 실제 운영에서는 Twitter/Naver API로 게시물 수집
sample = [f"{kw} 관련 포스트 {i}" for i in range(5)]
analysis = analyze_trend_with_claude(kw, sample)
results.append({"keyword": kw, **analysis})
print(f"'{kw}': {analysis['sentiment']} / {analysis['trend']}")
save_report(results)
Day 6-7: 블로그 자동생성봇 (수집 → 생성 파이프라인)
# ai-agents/agents/blog_generator.py
import anthropic, os, json
from datetime import datetime
from pathlib import Path
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
def generate_blog_post(topic: str, keywords: list, tone: str = "전문적이고 실용적인") -> dict:
"""Claude로 SEO 최적화 블로그 포스트 생성"""
kw_str = ", ".join(keywords)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=3000,
system=f"당신은 {tone} 블로그 작가입니다. SEO를 고려해 한국어로 작성합니다.",
messages=[{
"role": "user",
"content": f"""주제: {topic}
필수 키워드: {kw_str}
아래 구조로 블로그 포스트를 작성하세요 (1500~2000자):
1. 제목 (클릭률 높은 제목 3개 제안)
2. 도입부 (독자 고통 포인트 언급, 3~4문장)
3. 본문 (H2 3개, 각 섹션 코드/예시 포함)
4. 결론 (핵심 요약 + CTA)
5. 메타 디스크립션 (160자 이하)
JSON 형식으로 응답:
{{"titles": ["제목1", "제목2", "제목3"],
"intro": "도입부",
"sections": [{{"h2": "소제목", "content": "내용"}}],
"conclusion": "결론",
"meta_description": "메타"}}"""
}]
)
try:
return json.loads(response.content[0].text)
except:
return {"raw": response.content[0].text}
def save_blog_html(post: dict, topic: str):
"""블로그 포스트를 HTML 파일로 저장"""
output_dir = Path("blog_output")
output_dir.mkdir(exist_ok=True)
title = post.get("titles", [topic])[0]
date = datetime.now().strftime("%Y-%m-%d")
filename = output_dir / f"{date}_{topic[:20].replace(' ', '_')}.html"
sections_html = ""
for s in post.get("sections", []):
sections_html += f"{s['h2']}
\n{s['content']}
\n"
html = f"""
{title}
{title}
{post.get('intro', '')}
{sections_html}
마무리
{post.get('conclusion', '')}
"""
filename.write_text(html, encoding='utf-8')
print(f"✅ 블로그 저장: {filename}")
return str(filename)
if __name__ == "__main__":
post = generate_blog_post(
topic="AI 에이전트로 업무 자동화하는 법",
keywords=["AI 에이전트", "Claude API", "자동화", "파이썬"]
)
save_blog_html(post, "AI에이전트_자동화")
Week 2 (Day 8~14): 에이전트 + HITL 텔레그램 ⏱️ 각 3-4시간
Day 8-9: ReAct 에이전트 완성
ReAct(Reasoning + Acting) 패턴은 AI가 생각하고 → 행동하고 → 관찰하고 → 다시 생각하는 루프다. Claude의 Tool Use와 완벽히 맞아떨어지는 구조다.
# ai-agents/react_agent.py
import anthropic, os, json, requests
from datetime import datetime
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
# Tool 정의 3종
TOOLS = [
{
"name": "web_search",
"description": "인터넷에서 최신 정보를 검색합니다",
"input_schema": {
"type": "object",
"properties": {"query": {"type": "string", "description": "검색어"}},
"required": ["query"]
}
},
{
"name": "save_file",
"description": "텍스트 내용을 파일로 저장합니다",
"input_schema": {
"type": "object",
"properties": {
"filename": {"type": "string"},
"content": {"type": "string"}
},
"required": ["filename", "content"]
}
},
{
"name": "send_telegram",
"description": "텔레그램으로 메시지를 전송합니다",
"input_schema": {
"type": "object",
"properties": {"message": {"type": "string", "description": "전송할 메시지"}},
"required": ["message"]
}
}
]
def web_search(query: str) -> str:
"""실제 검색 대신 시뮬레이션 (운영 시 SerpAPI 연결)"""
return f"'{query}' 검색 결과: 관련 기사 5건 발견 (시뮬레이션)"
def save_file(filename: str, content: str) -> str:
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
return f"파일 저장 완료: {filename} ({len(content)}자)"
def send_telegram(message: str) -> str:
token = os.getenv("TELEGRAM_BOT_TOKEN")
chat_id = os.getenv("TELEGRAM_CHAT_ID")
if not token:
return "텔레그램 미설정 (시뮬레이션)"
url = f"https://api.telegram.org/bot{token}/sendMessage"
requests.post(url, json={"chat_id": chat_id, "text": message})
return "텔레그램 전송 완료"
TOOL_MAP = {"web_search": web_search, "save_file": save_file, "send_telegram": send_telegram}
def run_react_agent(task: str, max_iterations: int = 5) -> str:
"""ReAct 루프 실행"""
messages = [{"role": "user", "content": task}]
print(f"\n🤖 태스크: {task}")
for i in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2000,
tools=TOOLS,
messages=messages
)
if response.stop_reason == "end_turn":
result = response.content[0].text
print(f"\n✅ 완료: {result}")
return result
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" 🔧 [{i+1}] {block.name}({block.input})")
result = TOOL_MAP[block.name](**block.input)
print(f" 📤 결과: {result}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "최대 반복 횟수 초과"
if __name__ == "__main__":
run_react_agent(
"AI 에이전트 트렌드를 검색하고, 요약 파일을 저장한 뒤 텔레그램으로 보고하세요."
)
Day 10-11: Tool Use 3개 실전 연결
# ai-agents/tools/tool_registry.py
# 실무에서 쓰는 Tool 3종 완성판
import anthropic, os, json, smtplib
from email.mime.text import MIMEText
from pathlib import Path
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
# Tool 1: 파일 읽기/쓰기
def read_file(path: str) -> str:
p = Path(path)
if not p.exists():
return f"파일 없음: {path}"
return p.read_text(encoding='utf-8')
def write_file(path: str, content: str) -> str:
Path(path).write_text(content, encoding='utf-8')
return f"저장 완료: {path}"
# Tool 2: 이메일 발송 (Gmail SMTP)
def send_email(to: str, subject: str, body: str) -> str:
gmail = os.getenv("GMAIL_ADDRESS")
password = os.getenv("GMAIL_APP_PASSWORD")
if not gmail:
return "Gmail 미설정 (시뮬레이션)"
msg = MIMEText(body, 'plain', 'utf-8')
msg['Subject'] = subject
msg['From'] = gmail
msg['To'] = to
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
smtp.login(gmail, password)
smtp.send_message(msg)
return f"이메일 발송: {to}"
# Tool 3: JSON 데이터 저장/조회
def save_data(key: str, value: str, store_file: str = "agent_store.json") -> str:
store = {}
if Path(store_file).exists():
store = json.loads(Path(store_file).read_text())
store[key] = {"value": value, "updated": __import__('datetime').datetime.now().isoformat()}
Path(store_file).write_text(json.dumps(store, ensure_ascii=False, indent=2))
return f"저장: {key}"
def get_data(key: str, store_file: str = "agent_store.json") -> str:
if not Path(store_file).exists():
return "데이터 없음"
store = json.loads(Path(store_file).read_text())
return store.get(key, {}).get("value", f"키 없음: {key}")
Day 12-14: HITL (Human-in-the-Loop) 텔레그램 승인
HITL은 AI가 위험한 행동을 하기 전에 사람 승인을 받는 패턴이다. "AI 팀장" 구조의 핵심이다.
# ai-agents/hitl_agent.py
import anthropic, os, requests, time, json
from datetime import datetime
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT = os.getenv("TELEGRAM_CHAT_ID")
def telegram_send(text: str) -> int:
"""메시지 전송 후 message_id 반환"""
r = requests.post(
f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
json={"chat_id": TELEGRAM_CHAT, "text": text, "parse_mode": "HTML"}
)
return r.json().get("result", {}).get("message_id", 0)
def telegram_ask_approval(action_name: str, details: str, timeout: int = 300) -> bool:
"""텔레그램으로 승인 요청 → 5분 내 /yes 또는 /no 응답 대기"""
msg = f"""🤖 승인 요청
━━━━━━━━━━━━━━
작업: {action_name}
내용: {details}
시각: {datetime.now().strftime('%H:%M:%S')}
━━━━━━━━━━━━━━
✅ /yes — 승인 ❌ /no — 거부
⏱ {timeout//60}분 내 응답 없으면 자동 거부"""
msg_id = telegram_send(msg)
print(f"📨 승인 요청 전송 (msg_id={msg_id})")
# 폴링: /yes 또는 /no 대기
deadline = time.time() + timeout
last_update = 0
while time.time() < deadline:
r = requests.get(
f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/getUpdates",
params={"offset": last_update + 1, "timeout": 10}
).json()
for update in r.get("result", []):
last_update = update["update_id"]
text = update.get("message", {}).get("text", "").strip().lower()
if text == "/yes":
telegram_send("✅ 승인되었습니다. 실행합니다.")
return True
if text == "/no":
telegram_send("❌ 거부되었습니다. 중단합니다.")
return False
telegram_send("⏰ 시간 초과 — 자동 거부")
return False
def run_hitl_pipeline(task: str):
"""위험 작업 전 반드시 사람 승인 받는 파이프라인"""
print(f"📋 태스크 분석 중: {task}")
# 1단계: Claude로 실행 계획 수립
plan_resp = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1000,
messages=[{"role": "user",
"content": f"다음 태스크의 실행 계획을 3단계로 요약하세요: {task}"}]
)
plan = plan_resp.content[0].text
# 2단계: 사람 승인 요청
approved = telegram_ask_approval(
action_name=task[:50],
details=f"실행 계획:\n{plan[:300]}"
)
if not approved:
print("❌ 사용자가 거부 → 실행 중단")
return
# 3단계: 승인 후 실행
print("✅ 승인 완료 → 실행 시작")
result_resp = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2000,
messages=[{"role": "user", "content": f"다음 태스크를 실행하고 결과를 보고하세요: {task}"}]
)
result = result_resp.content[0].text
telegram_send(f"📊 실행 완료\n\n{result[:500]}")
print(f"결과: {result}")
if __name__ == "__main__":
run_hitl_pipeline("경쟁사 3곳 가격 변동 분석 후 이메일 발송")
Week 3 (Day 15~21): 프로덕션 하네스 구축 ⏱️ 각 2-3시간
에이전트가 만들어졌다. 이제 24시간 돌아가도 안 터지는 구조를 만든다.
Day 15-16: 에러 핸들링 + 자동 재시도
# ai-agents/utils/resilient_caller.py
import time, functools, logging
from typing import Callable, Any
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler("agent.log", encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def retry(max_attempts: int = 3, delay: float = 2.0, backoff: float = 2.0,
exceptions: tuple = (Exception,)):
"""지수 백오프 재시도 데코레이터"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
wait = delay
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts:
logger.error(f"❌ {func.__name__} 최종 실패 (시도 {attempt}/{max_attempts}): {e}")
raise
logger.warning(f"⚠️ {func.__name__} 실패 (시도 {attempt}/{max_attempts}): {e} → {wait:.1f}초 후 재시도")
time.sleep(wait)
wait *= backoff
return wrapper
return decorator
# 사용 예시
import anthropic, os
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
@retry(max_attempts=3, delay=2.0, exceptions=(anthropic.APIError, anthropic.RateLimitError))
def safe_claude_call(prompt: str, model: str = "claude-sonnet-4-6") -> str:
response = client.messages.create(
model=model, max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
# Day 17-18: 구조화 로깅
class AgentLogger:
"""에이전트 실행 기록 전용 로거"""
def __init__(self, agent_name: str, log_file: str = "agent_runs.jsonl"):
self.agent_name = agent_name
self.log_file = log_file
def log(self, event: str, data: dict):
entry = {
"ts": __import__('datetime').datetime.now().isoformat(),
"agent": self.agent_name,
"event": event,
**data
}
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(__import__('json').dumps(entry, ensure_ascii=False) + '\n')
logger.info(f"[{self.agent_name}] {event}: {data}")
# 사용법
# alog = AgentLogger("blog_generator")
# alog.log("start", {"task": "AI 트렌드 블로그 생성"})
# alog.log("complete", {"tokens_used": 1200, "file": "blog_2026.html"})
Day 19-21: 20항목 배포 전 체크리스트 자동화
# ai-agents/utils/deployment_harness.py
import os, sys, subprocess
from pathlib import Path
CHECKS = [
# (설명, 검증 함수)
("ANTHROPIC_API_KEY 설정됨", lambda: bool(os.getenv("ANTHROPIC_API_KEY"))),
("TELEGRAM_BOT_TOKEN 설정됨", lambda: bool(os.getenv("TELEGRAM_BOT_TOKEN"))),
("TELEGRAM_CHAT_ID 설정됨", lambda: bool(os.getenv("TELEGRAM_CHAT_ID"))),
(".env 파일 존재", lambda: Path(".env").exists()),
(".gitignore에 .env 포함", lambda: ".env" in Path(".gitignore").read_text() if Path(".gitignore").exists() else False),
("requirements.txt 존재", lambda: Path("requirements.txt").exists()),
("agent_store.json 접근 가능", lambda: True), # 첫 실행 시 자동 생성
("로그 폴더 쓰기 가능", lambda: Path("logs").mkdir(exist_ok=True) or True),
("Python 버전 3.9+", lambda: sys.version_info >= (3, 9)),
("anthropic 패키지 설치됨", lambda: __import__('importlib').util.find_spec("anthropic") is not None),
("requests 패키지 설치됨", lambda: __import__('importlib').util.find_spec("requests") is not None),
("bs4 패키지 설치됨", lambda: __import__('importlib').util.find_spec("bs4") is not None),
("하드코딩된 키 없음", lambda: not any("sk-ant" in Path(f).read_text() for f in Path(".").rglob("*.py") if Path(f).is_file())),
("기존 로그 파일 정상", lambda: True),
("API 응답 정상 (ping)", lambda: __import__('anthropic').Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY")).messages.create(model="claude-haiku-4-5-20251001", max_tokens=5, messages=[{"role":"user","content":"hi"}]).content[0].text is not None),
("agent_runs.jsonl 포맷 정상", lambda: True),
("텔레그램 봇 연결 가능", lambda: bool(os.getenv("TELEGRAM_BOT_TOKEN"))),
("출력 디렉토리 존재", lambda: Path("output").mkdir(exist_ok=True) or True),
("타임아웃 설정 확인", lambda: True),
("메모리 사용량 < 80%", lambda: __import__('psutil').virtual_memory().percent < 80 if __import__('importlib').util.find_spec("psutil") else True),
]
def run_harness() -> bool:
print("=" * 50)
print(" 배포 전 체크리스트 (20항목)")
print("=" * 50)
passed = failed = 0
for desc, check in CHECKS:
try:
ok = check()
except Exception as e:
ok = False
symbol = "✅" if ok else "❌"
print(f"{symbol} {desc}")
if ok: passed += 1
else: failed += 1
print("=" * 50)
print(f"결과: {passed}/20 통과 | 실패: {failed}개")
if failed > 0:
print("❌ 배포 중단 — 실패 항목 수정 후 재시도")
return False
print("✅ 모든 체크 통과 — 배포 진행 가능")
return True
if __name__ == "__main__":
ok = run_harness()
sys.exit(0 if ok else 1)
Week 4 (Day 22~30): 오케스트레이터 + KPI 자동 보고 ⏱️ 집중 완성주
Day 22-24: 오케스트레이터 완성
# ai-agents/daily_orchestrator.py
"""매일 07:00 자동 실행 — Windows 작업 스케줄러 등록"""
import subprocess, os, json, time
from datetime import datetime
from pathlib import Path
LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)
log_file = LOG_DIR / f"orchestrator_{datetime.now().strftime('%Y-%m-%d')}.log"
def log(msg: str):
ts = datetime.now().strftime("%H:%M:%S")
line = f"[{ts}] {msg}"
print(line)
with open(log_file, 'a', encoding='utf-8') as f:
f.write(line + '\n')
def run_step(name: str, script: str, timeout: int = 300) -> bool:
log(f"▶ STEP 시작: {name}")
try:
result = subprocess.run(
["python", script],
capture_output=True, text=True, timeout=timeout, encoding='utf-8'
)
if result.returncode == 0:
log(f"✅ STEP 완료: {name}")
return True
else:
log(f"❌ STEP 실패: {name}\n{result.stderr[:500]}")
return False
except subprocess.TimeoutExpired:
log(f"⏰ STEP 타임아웃: {name}")
return False
except Exception as e:
log(f"🔥 STEP 오류: {name} — {e}")
return False
def send_daily_report(results: dict):
"""하루 실행 결과를 텔레그램으로 보고"""
import requests
token = os.getenv("TELEGRAM_BOT_TOKEN")
chat = os.getenv("TELEGRAM_CHAT_ID")
if not token: return
icons = {True: "✅", False: "❌"}
lines = [f"📊 일일 오케스트레이터 리포트",
f"날짜: {datetime.now().strftime('%Y-%m-%d')}",
"━━━━━━━━━━━━━━"]
for step, ok in results.items():
lines.append(f"{icons[ok]} {step}")
total = len(results)
passed = sum(results.values())
lines += ["━━━━━━━━━━━━━━", f"결과: {passed}/{total} 성공"]
requests.post(
f"https://api.telegram.org/bot{token}/sendMessage",
json={"chat_id": chat, "text": "\n".join(lines), "parse_mode": "HTML"}
)
PIPELINE = [
("뉴스 수집", "agents/news_collector.py"),
("경쟁사 모니터링", "agents/competitor_monitor.py"),
("SNS 트렌드", "agents/sns_keyword_tracker.py"),
("블로그 생성", "agents/blog_generator.py"),
("이메일 발송", "agents/email_sender.py"),
]
if __name__ == "__main__":
log("🚀 오케스트레이터 시작")
results = {}
for name, script in PIPELINE:
results[name] = run_step(name, script)
time.sleep(2)
send_daily_report(results)
log("🏁 오케스트레이터 종료")
Day 25-27: KPI 대시보드 (HTML)
# ai-agents/kpi_dashboard.py
"""agent_runs.jsonl을 읽어 KPI HTML 대시보드 생성"""
import json
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict
def load_runs(log_file: str = "agent_runs.jsonl") -> list:
if not Path(log_file).exists():
return []
with open(log_file, 'r', encoding='utf-8') as f:
return [json.loads(line) for line in f if line.strip()]
def calc_kpi(runs: list) -> dict:
"""핵심 KPI 계산"""
last_7 = [r for r in runs
if r.get("ts", "") >= (datetime.now() - timedelta(days=7)).isoformat()]
agents = defaultdict(lambda: {"total": 0, "success": 0, "tokens": 0})
for r in last_7:
a = agents[r.get("agent", "unknown")]
a["total"] += 1
if r.get("event") == "complete":
a["success"] += 1
a["tokens"] += r.get("tokens_used", 0)
return {
"period": "최근 7일",
"total_runs": len(last_7),
"agents": dict(agents),
"success_rate": sum(a["success"] for a in agents.values()) /
max(sum(a["total"] for a in agents.values()), 1) * 100
}
def format_report(kpi: dict) -> str:
"""텍스트 KPI 리포트 생성 (터미널 출력 + 파일 저장용)"""
sep = "=" * 50
lines = [
sep,
" AI 에이전트 KPI 리포트",
f" 기간: {kpi['period']}",
f" 생성: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
sep,
f" 총 실행: {kpi['total_runs']}회",
f" 성공률: {kpi['success_rate']:.1f}%",
f" 활성 에이전트: {len(kpi['agents'])}개",
"-" * 50,
]
for agent, data in kpi["agents"].items():
rate = data["success"] / max(data["total"], 1) * 100
lines.append(f" {agent:<20} {data['total']:>4}회 {rate:>5.1f}% {data['tokens']:>8,} 토큰")
lines.append(sep)
return "\n".join(lines)
if __name__ == "__main__":
runs = load_runs()
kpi = calc_kpi(runs)
report = format_report(kpi)
print(report)
# JSON 저장 (다른 도구에서 읽기 쉬운 형식)
out = Path(f"kpi_{datetime.now().strftime('%Y%m%d')}.json")
out.write_text(json.dumps(kpi, ensure_ascii=False, indent=2), encoding='utf-8')
print(f"\n✅ KPI JSON 저장: {out}")
Day 28-30: 자동 KPI 보고 + 최종 배포
📅 자동화 스케줄 설계 원칙
Day 28-30의 핵심은 "언제, 무엇을, 어떻게 자동 실행할지" 결정하는 것입니다.
추천 스케줄 구조:
• 매일 07:00 — 수집봇 + 생성봇 + 오케스트레이터 순차 실행
• 매주 월요일 09:00 — 주간 KPI 텔레그램 보고
• 매월 1일 — 비용 결산 + 다음 달 목표 설정
스케줄러 선택 기준:
• Windows: 작업 스케줄러(Task Scheduler) — GUI로 등록하거나 schtasks 명령 활용
• Mac/Linux: cron — crontab -e로 편집
• 클라우드 배포 시: GitHub Actions schedule, Railway cron, n8n Schedule trigger 권장
KPI 보고는 kpi_dashboard.py가 agent_runs.jsonl을 읽어 성공률·토큰 사용량·에이전트별 통계를 계산하고, 텔레그램 Bot API(sendMessage)로 전송하는 구조입니다.
Day 30 최종 점검 — 1인 AI 운영팀 완성 확인:
☐ deployment_harness.py 실행 → 20항목 전체 통과
☐ 스케줄러 등록 확인 (매일 07:00 오케스트레이터 자동 시작)
☐ 텔레그램에서 KPI 보고 수신 확인
30일 로드맵 완료 체크리스트:
□ 수집봇 3개 → 매일 자동 실행 중
□ 생성봇 2개 → 블로그·SNS 초안 자동 생성 중
□ 에이전트 1개 → Tool Use + HITL 텔레그램 연결 완료
□ 오케스트레이터 → 07:00 Windows 작업 스케줄러 등록 완료
□ 비용 모니터링 → 월 예산 알림 설정 완료
"30일 후 당신은 더 이상 'AI를 써보고 싶은 사람'이 아니다. AI 시스템을 운용하는 사람이다."
▶
30일 완성 챌린지 — AI사냥꾼과 함께 매일 진도 체크
📌 구독 + 공유 + 댓글 "받고싶다" → 고정 댓글에서 PDF 무료 수령 · 선착순 300명