1일 만에 완성한 Production-Ready AI 챗봇 시스템
Claude Code의 Multi-Agent 시스템을 활용하여 복잡한 프로젝트를 체계적으로 구현한 경험을 공유합니다.
SAP 데이터 웨어하우스 자연어 질의응답 챗봇을 구축했습니다. 사용자가 한국어나 영어로 질문하면, AI가 자동으로 SQL을 생성하고 Snowflake에서 데이터를 조회하여 결과를 보여주는 시스템입니다.
| 영역 | 기술 |
|---|---|
| Frontend | Next.js 14, React, TypeScript, Tailwind CSS |
| Backend | FastAPI, Python 3.11, Pydantic |
| AI/LLM | Google Gemini 2.0 Flash Experimental |
| Database | Snowflake (SAP.PUBLIC schema) |
| Container | Docker, Docker Compose |
| Development | Claude Code + Multi-Agent System |
Claude Code의 Sub Agent는 복잡한 프로젝트를 여러 전문 AI 에이전트들이 협업하여 구현하는 시스템입니다. 마치 실제 개발팀처럼 각 에이전트가 특정 역할을 맡아 작업합니다.
각 Phase별로 전담 에이전트를 만들어 집중력과 전문성을 높였습니다.
각 Phase의 context를 별도 파일로 관리하여 나중에 참조하기 쉽습니다.
독립적인 Phase는 동시에 여러 에이전트가 작업할 수 있습니다.
한 Phase에서 문제가 생겨도 다른 Phase에 영향을 주지 않습니다.
┌─────────────┐ ┌─────────────┐ ┌──────────────┐ ┌───────────┐
│ 사용자 │─────▶│ Frontend │─────▶│ Backend │─────▶│ Snowflake │
│ (Browser) │ │ (Next.js) │ │ (FastAPI) │ │ (SAP) │
└─────────────┘ └─────────────┘ └──────────────┘ └───────────┘
│
▼
┌─────────────┐
│ Gemini │
│ LLM API │
└─────────────┘
.claude/
└── agents/
├── sap-chatbot-supervisor.md # 🎯 프로젝트 총괄
├── sap-chatbot-phase1.md # Backend 구조
├── sap-chatbot-phase2.md # Gemini 연동
├── sap-chatbot-phase3.md # SQL 실행
├── sap-chatbot-phase4.md # Frontend UI
├── sap-chatbot-phase5.md # Docker화
└── sap-chatbot-phase6.md # 테스트 자동화
각 Agent는 YAML front matter를 포함한 Markdown 파일로 정의합니다:
---
name: sap-chatbot-phase1
description: Backend 기본 구조 + Snowflake JWT 인증
version: 1.0.0
model: sonnet
tags: [backend, snowflake, jwt]
---
# Role
Backend Express 서버 구축 및 Snowflake JWT 인증 구현
## Responsibilities
1. Express.js 서버 초기화
2. Snowflake JWT 토큰 생성
3. SQL API 클라이언트 구현
4. Winston 로거 설정
## Files to Create
- backend/app/main.py
- backend/app/config.py
- backend/app/utils/auth.py
...
## Success Criteria
- [ ] Express 서버 정상 실행
- [ ] JWT 토큰 생성 성공
- [ ] Snowflake 연결 성공
sap-chatbot-supervisor는 프로젝트 매니저 역할로:
Agent: sap-chatbot-phase1
# 1. Backend 디렉토리 구조 생성
backend/
├── app/
│ ├── main.py # FastAPI 진입점
│ ├── config.py # 환경 설정
│ ├── utils/
│ │ └── auth.py # JWT 인증
│ └── services/
│ └── snowflake.py # Snowflake 클라이언트
├── requirements.txt
└── Dockerfile
# backend/app/utils/auth.py
import jwt
import base64
from datetime import datetime, timedelta
from cryptography.hazmat.primitives import serialization
def get_snowflake_jwt_token() -> str:
"""Snowflake JWT 토큰 생성"""
# Private Key 로드
private_key_bytes = base64.b64decode(
os.getenv("SNOWFLAKE_PRIVATE_KEY")
)
private_key = serialization.load_pem_private_key(
private_key_bytes,
password=None
)
# JWT 페이로드
now = datetime.utcnow()
payload = {
"iss": f"{account}.{user}",
"sub": f"{account}.{user}",
"iat": now,
"exp": now + timedelta(hours=1)
}
# 토큰 서명
return jwt.encode(payload, private_key, algorithm="RS256")
# Agent 호출
/task subagent=sap-chatbot-phase1 "Phase 1 구현 시작"
# 테스트
curl http://localhost:3001/api/health
Agent: sap-chatbot-phase2
# backend/app/services/llm.py
import google.generativeai as genai
class GeminiService:
def __init__(self):
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
self.model = genai.GenerativeModel("gemini-2.0-flash-exp")
async def generate_sql(self, question: str, metadata: str) -> str:
"""자연어 질문을 SQL로 변환"""
prompt = f"""
당신은 SAP 데이터 분석 전문가입니다.
## 사용 가능한 테이블과 컬럼:
{metadata}
## 사용자 질문:
{question}
## 지침:
1. Snowflake SQL 문법 사용
2. SELECT 문만 생성 (DROP, DELETE, TRUNCATE 금지)
3. 테이블명은 SAP.PUBLIC.* 형식 사용
4. 결과는 SQL만 반환 (설명 불필요)
SQL:
"""
response = await self.model.generate_content_async(prompt)
return self._extract_sql(response.text)
# data_modeling/metadata.csv 활용
TABLE_NAME,COLUMN_NAME,DESCRIPTION
DIM_COMPANY,CLIENT_CD,클라이언트 코드
DIM_COMPANY,COMPANY_CD,회사 코드
DIM_COMPANY,COMPANY_NM,회사명
DIM_CUSTOMER,CUSTOMER_ID,고객 번호
...
Agent: sap-chatbot-phase3
# backend/app/services/query.py
import httpx
class SnowflakeQueryService:
async def execute_query(self, sql: str) -> dict:
"""Snowflake SQL API로 쿼리 실행"""
token = get_snowflake_jwt_token()
async with httpx.AsyncClient() as client:
# SQL 제출
submit_resp = await client.post(
f"https://{account}.snowflakecomputing.com/api/v2/statements",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
json={
"statement": sql,
"warehouse": warehouse,
"database": database,
"schema": schema,
"timeout": 60
}
)
# 결과 조회
statement_handle = submit_resp.json()["statementHandle"]
result = await self._poll_result(client, statement_handle, token)
return {
"results": result["data"],
"row_count": result["rowCount"],
"execution_time_ms": result["executionTime"]
}
# backend/app/utils/validation.py
DANGEROUS_KEYWORDS = ["DROP", "DELETE", "TRUNCATE", "ALTER", "CREATE"]
def validate_sql(sql: str) -> tuple[bool, str]:
"""SQL 안전성 검증"""
sql_upper = sql.upper()
for keyword in DANGEROUS_KEYWORDS:
if keyword in sql_upper:
return False, f"Dangerous keyword detected: {keyword}"
return True, ""
Agent: sap-chatbot-phase4
// frontend/components/ChatInterface.tsx
'use client';
import { useState } from 'react';
interface Message {
id: string;
role: 'user' | 'assistant';
content: string;
sql?: string;
results?: any[];
}
export default function ChatInterface() {
const [messages, setMessages] = useState<Message[]>([]);
const [input, setInput] = useState('');
const [loading, setLoading] = useState(false);
const handleSubmit = async (e: React.FormEvent) => {
e.preventDefault();
if (!input.trim()) return;
// 사용자 메시지 추가
const userMsg: Message = {
id: Date.now().toString(),
role: 'user',
content: input
};
setMessages(prev => [...prev, userMsg]);
setInput('');
setLoading(true);
try {
// Backend API 호출
const response = await fetch('http://localhost:3001/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ question: input })
});
const data = await response.json();
// AI 응답 추가
const assistantMsg: Message = {
id: (Date.now() + 1).toString(),
role: 'assistant',
content: `조회 결과: ${data.metadata.row_count}건`,
sql: data.sql_query,
results: data.results
};
setMessages(prev => [...prev, assistantMsg]);
} catch (error) {
console.error('Error:', error);
} finally {
setLoading(false);
}
};
return (
<div className="flex flex-col h-screen">
{/* 메시지 목록 */}
<div className="flex-1 overflow-y-auto p-4">
{messages.map(msg => (
<MessageBubble key={msg.id} message={msg} />
))}
</div>
{/* 입력창 */}
<form onSubmit={handleSubmit} className="p-4 border-t">
<input
type="text"
value={input}
onChange={(e) => setInput(e.target.value)}
placeholder="질문을 입력하세요..."
className="w-full p-3 border rounded-lg"
disabled={loading}
/>
</form>
</div>
);
}
// frontend/components/ResultTable.tsx
interface ResultTableProps {
results: any[];
}
export default function ResultTable({ results }: ResultTableProps) {
if (!results || results.length === 0) return null;
const columns = Object.keys(results[0]);
return (
<div className="overflow-x-auto mt-4">
<table className="min-w-full border">
<thead className="bg-gray-100">
<tr>
{columns.map(col => (
<th key={col} className="px-4 py-2 border">
{col}
</th>
))}
</tr>
</thead>
<tbody>
{results.map((row, idx) => (
<tr key={idx}>
{columns.map(col => (
<td key={col} className="px-4 py-2 border">
{row[col]}
</td>
))}
</tr>
))}
</tbody>
</table>
</div>
);
}
Agent: sap-chatbot-phase5
# docker-compose.yml
version: '3.8'
services:
backend:
build:
context: ./backend
dockerfile: Dockerfile
container_name: sap-chatbot-backend
ports:
- "3001:3001"
environment:
- SNOWFLAKE_ACCOUNT=${SNOWFLAKE_ACCOUNT}
- SNOWFLAKE_USER=${SNOWFLAKE_USER}
- SNOWFLAKE_PRIVATE_KEY=${SNOWFLAKE_PRIVATE_KEY}
- GEMINI_API_KEY=${GEMINI_API_KEY}
volumes:
- ./logs:/app/logs
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:3001/api/health"]
interval: 30s
timeout: 10s
retries: 3
restart: unless-stopped
frontend:
build:
context: ./frontend
dockerfile: Dockerfile
container_name: sap-chatbot-frontend
ports:
- "3000:3000"
environment:
- NEXT_PUBLIC_API_URL=http://localhost:3001
depends_on:
backend:
condition: service_healthy
restart: unless-stopped
volumes:
logs:
# backend/Dockerfile
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . .
EXPOSE 3001
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "3001"]
Agent: sap-chatbot-phase6
#!/bin/bash
# scripts/start.sh
set -e
# 색상 정의
GREEN='\033[0;32m'
RED='\033[0;31m'
NC='\033[0m'
# 사전 검증
check_prerequisites() {
echo "🔍 Checking prerequisites..."
# Docker 설치 확인
if ! command -v docker &> /dev/null; then
echo -e "${RED}❌ Docker not found${NC}"
exit 1
fi
# .env 파일 확인
if [ ! -f .env ]; then
echo -e "${RED}❌ .env file not found${NC}"
exit 1
fi
# 필수 환경변수 확인
source .env
required_vars=("SNOWFLAKE_ACCOUNT" "GEMINI_API_KEY")
for var in "${required_vars[@]}"; do
if [ -z "${!var}" ]; then
echo -e "${RED}❌ $var not set${NC}"
exit 1
fi
done
echo -e "${GREEN}✅ All prerequisites met${NC}"
}
# 포트 사용 확인
check_ports() {
for port in 3000 3001; do
if lsof -Pi :$port -sTCP:LISTEN -t >/dev/null ; then
echo -e "${RED}⚠️ Port $port is already in use${NC}"
fi
done
}
# 서비스 시작
start_services() {
local service=$1
local build_flag=$2
echo "🚀 Starting services..."
if [ "$build_flag" = "--build" ]; then
docker-compose up -d --build $service
else
docker-compose up -d $service
fi
echo -e "${GREEN}✅ Services started${NC}"
}
# 헬스체크
wait_for_health() {
echo "⏳ Waiting for services to be healthy..."
# Backend 헬스체크 (최대 60초)
for i in {1..60}; do
if curl -f http://localhost:3001/api/health &>/dev/null; then
echo -e "${GREEN}✅ Backend is healthy${NC}"
break
fi
if [ $i -eq 60 ]; then
echo -e "${RED}❌ Backend health check timeout${NC}"
exit 1
fi
sleep 1
done
# Frontend 헬스체크
for i in {1..60}; do
if curl -f http://localhost:3000 &>/dev/null; then
echo -e "${GREEN}✅ Frontend is healthy${NC}"
break
fi
if [ $i -eq 60 ]; then
echo -e "${RED}❌ Frontend health check timeout${NC}"
exit 1
fi
sleep 1
done
}
# 메인 실행
main() {
check_prerequisites
check_ports
start_services "$1" "$2"
if [ "$3" != "--no-health" ]; then
wait_for_health
fi
echo ""
echo "🎉 SAP Chatbot is ready!"
echo "Frontend: http://localhost:3000"
echo "Backend: http://localhost:3001"
echo "API Docs: http://localhost:3001/docs"
}
# 사용법
if [ "$1" = "--help" ]; then
cat << EOF
Usage: ./scripts/start.sh [SERVICE] [OPTIONS]
Services:
all Start all services (default)
backend Start backend only
frontend Start frontend only
Options:
--build Force rebuild images
--no-health Skip health checks
--verbose Show detailed output
--help Show this help
Examples:
./scripts/start.sh
./scripts/start.sh all --build
./scripts/start.sh backend --verbose
EOF
exit 0
fi
main "$@"
#!/bin/bash
# scripts/test.sh
set -e
# 테스트 결과 추적
TOTAL_TESTS=0
PASSED_TESTS=0
FAILED_TESTS=0
SKIPPED_TESTS=0
# 색상
GREEN='\033[0;32m'
RED='\033[0;31m'
YELLOW='\033[1;33m'
NC='\033[0m'
# 테스트 실행 함수
run_test() {
local test_name=$1
local test_script=$2
TOTAL_TESTS=$((TOTAL_TESTS + 1))
echo ""
echo "ℹ️ Test $TOTAL_TESTS: $test_name"
if bash "$test_script"; then
echo -e "${GREEN}✅ PASS: $test_name${NC}"
PASSED_TESTS=$((PASSED_TESTS + 1))
else
local exit_code=$?
if [ $exit_code -eq 2 ]; then
echo -e "${YELLOW}⚠️ SKIP: $test_name${NC}"
SKIPPED_TESTS=$((SKIPPED_TESTS + 1))
else
echo -e "${RED}❌ FAIL: $test_name${NC}"
FAILED_TESTS=$((FAILED_TESTS + 1))
fi
fi
}
# 메인 실행
echo "================================================"
echo " SAP Chatbot - Integration Tests"
echo "================================================"
# 사전 검증
run_test "Docker Prerequisites" "./tests/integration/test_prerequisites.sh"
run_test "Services Running" "./tests/integration/test_services.sh"
# Backend 테스트
run_test "Backend Health" "./tests/integration/test_backend_health.sh"
# E2E 테스트
run_test "End-to-End Query" "./tests/integration/test_e2e_query.sh"
# 에러 핸들링
run_test "Error Handling" "./tests/integration/test_error_handling.sh"
# 결과 출력
echo ""
echo "================================================"
echo " Test Summary"
echo "================================================"
echo ""
echo "ℹ️ Total Tests: $TOTAL_TESTS"
echo -e "${GREEN}✅ Passed: $PASSED_TESTS${NC}"
echo -e "${RED}❌ Failed: $FAILED_TESTS${NC}"
echo -e "${YELLOW}⚠️ Skipped: $SKIPPED_TESTS${NC}"
echo ""
if [ $FAILED_TESTS -eq 0 ]; then
echo -e "${GREEN}✅ All tests passed!${NC}"
exit 0
else
echo -e "${RED}❌ Some tests failed${NC}"
exit 1
fi
#!/bin/bash
# scripts/health.sh
set -e
# 서비스 헬스 체크
check_service_health() {
local service=$1
local url=$2
if curl -f "$url" &>/dev/null; then
echo -e "✅ $service: HEALTHY"
return 0
else
echo -e "❌ $service: UNHEALTHY"
return 1
fi
}
# JSON 출력
output_json() {
cat << EOF
{
"timestamp": "$(date -u +"%Y-%m-%dT%H:%M:%SZ")",
"overall_status": "$1",
"services": {
"backend": {
"status": "$2",
"endpoint": "http://localhost:3001"
},
"frontend": {
"status": "$3",
"endpoint": "http://localhost:3000"
}
}
}
EOF
}
# Watch 모드
watch_mode() {
while true; do
clear
check_health
sleep 5
done
}
# 메인 헬스 체크
check_health() {
echo "================================================"
echo " SAP Chatbot - Health Status"
echo "================================================"
echo "Last updated: $(date)"
echo ""
local backend_status="UNHEALTHY"
local frontend_status="UNHEALTHY"
local overall_status="UNHEALTHY"
# Backend 체크
if check_service_health "Backend" "http://localhost:3001/api/health"; then
backend_status="HEALTHY"
fi
# Frontend 체크
if check_service_health "Frontend" "http://localhost:3000"; then
frontend_status="HEALTHY"
fi
# 전체 상태
if [ "$backend_status" = "HEALTHY" ] && [ "$frontend_status" = "HEALTHY" ]; then
overall_status="HEALTHY"
echo ""
echo "================================================"
echo "✅ Overall Status: ALL SYSTEMS OPERATIONAL"
echo "================================================"
else
echo ""
echo "================================================"
echo "❌ Overall Status: SOME SYSTEMS DOWN"
echo "================================================"
fi
# JSON 출력 모드
if [ "$1" = "--json" ]; then
output_json "$overall_status" "$backend_status" "$frontend_status"
fi
}
# 메인 실행
if [ "$1" = "--watch" ]; then
watch_mode
elif [ "$1" = "--json" ]; then
check_health --json
else
check_health
fi
# tests/integration/test_e2e_query.sh
#!/bin/bash
set -e
BASE_URL="http://localhost:3001"
ENDPOINT="$BASE_URL/api/chat"
echo "Testing End-to-End Query Flow..."
# 테스트 1: 정상 쿼리
echo "Test 1: Valid query"
RESPONSE=$(curl -s -w "\n%{http_code}" -X POST "$ENDPOINT" \
-H "Content-Type: application/json" \
-d '{"question": "회사 목록을 보여줘"}')
HTTP_CODE=$(echo "$RESPONSE" | tail -n1)
BODY=$(echo "$RESPONSE" | head -n-1)
if [ "$HTTP_CODE" -eq 200 ]; then
echo "✅ PASS: Got 200 OK"
else
echo "❌ FAIL: Expected 200, got $HTTP_CODE"
exit 1
fi
# SQL 쿼리 생성 확인
if echo "$BODY" | jq -e '.sql_query' > /dev/null 2>&1; then
SQL=$(echo "$BODY" | jq -r '.sql_query')
echo "✅ PASS: SQL generated: $SQL"
else
echo "❌ FAIL: No SQL query in response"
exit 1
fi
# 테스트 2: 응답 시간 체크
START=$(date +%s%N)
curl -s -X POST "$ENDPOINT" \
-H "Content-Type: application/json" \
-d '{"question": "고객 수는?"}' > /dev/null
END=$(date +%s%N)
ELAPSED=$(( (END - START) / 1000000 )) # ms로 변환
if [ $ELAPSED -lt 10000 ]; then
echo "✅ PASS: Response time: ${ELAPSED}ms"
else
echo "❌ FAIL: Response too slow: ${ELAPSED}ms"
exit 1
fi
echo "✅ All E2E tests passed!"
# Claude Code에서 실행
각 Phase를 실제 agent에게 위임해줘.
Phase 1부터 시작: TaskCreate(subagent="sap-chatbot-phase1", prompt="agents/phase1/에 Backend 구현")
완료되면
Phase 2: TaskCreate(subagent="sap-chatbot-phase2", prompt="agents/phase2/에 Gemini API 구현")
이런 식으로 Phase 6까지 순차 실행
# Phase 1만 실행
sap-chatbot-phase1에게 agent를 위임해서 "Backend 구조 및 Snowflake JWT 구현" 해줘
# Phase 2 실행
sap-chatbot-phase2에게 agent를 위임해서 "Gemini API 연동 및 프롬프트 엔지니어링" 해줘
# 서비스 시작
./scripts/start.sh all
# 헬스 체크
./scripts/health.sh
# 통합 테스트
./scripts/test.sh
# 로그 확인
./scripts/logs.sh backend --follow
# 서비스 종료
./scripts/stop.sh all --archive-logs
================================================
SAP Chatbot - Integration Tests
================================================
✅ Test 1: Docker Prerequisites PASS
✅ Test 2: Services Running PASS
✅ Test 3: Backend Health Endpoint PASS (147ms)
✅ Test 4: Backend Response Time PASS (147ms)
✅ Test 5: Backend API Documentation PASS
✅ Test 6: Frontend Accessibility PASS
✅ Test 7: CORS Headers Present PASS
✅ Test 8: Error Handling (Missing Q) PASS
✅ Test 9: Log Files Generated PASS
⚠️ Test 10: Snowflake Connectivity SKIPPED
================================================
Test Summary
================================================
Total Tests: 10
✅ Passed: 9
❌ Failed: 0
⚠️ Skipped: 1
✅ All tests passed!
❌ Bad: "Backend 개발" (너무 광범위)
✅ Good: "Backend Express 서버 + Snowflake JWT 인증" (구체적)
## Dependencies
- Phase 1 완료 필수
- Phase 2 완료 필수
- .env 파일 존재
## Success Criteria
- [ ] Express 서버 정상 실행
- [ ] JWT 토큰 생성 성공
- [ ] Snowflake 연결 테스트 통과
docs/02-design/context/
├── phase1.md # Backend 구현 context
├── phase2.md # Gemini 연동 context
├── phase3.md # SQL 실행 context
├── phase4.md # Frontend context
├── phase5.md # Docker context
└── phase6.md # Test context
왜 중요한가?
## Phase Transition Logic
Phase 1 완료 체크:
- [ ] Express 서버 실행 확인
- [ ] JWT 토큰 생성 테스트 통과
- [ ] Context 파일 저장 완료
→ Phase 2로 전환:
- Phase 1의 context 읽기
- Phase 2 agent에게 전달
- Phase 2 시작
# Phase 3에서 에러 발생 시
# Phase 1, 2는 유지하고 Phase 3만 재실행
/task subagent=sap-chatbot-phase3 "Phase 3 재구현"
| Phase | Complexity | Model | 이유 |
|-------|-----------|-------|------|
| Phase 1 | High | Opus | JWT 암호화, 인증 로직 복잡 |
| Phase 2 | Medium | Sonnet | LLM API 연동 표준적 |
| Phase 4 | Medium | Sonnet | React 컴포넌트 구현 |
| Phase 6 | Low | Haiku | Shell script, 단순 테스트 |
Before (단일 Agent):
After (Multi-Agent):
Plan → Design → Phase 분할 → Agent 등록 → 순차 실행 → 검증 → 완료
각 단계가 명확하여 예측 가능한 개발 진행
1,715 lines의 Shell Scripts로: