이 글은 누구를 위한 것인가
- AI 제품을 출시했지만 이후 개선 방법이 없는 팀
- 사용자 피드백을 모델 개선에 연결하는 파이프라인이 없는 팀
- 데이터 플라이휠 개념은 알지만 구체적 구현이 필요한 ML 엔지니어
들어가며
아마존, 구글이 AI에서 독보적인 이유는 데이터 플라이휠이다. 더 많은 사용자 → 더 많은 데이터 → 더 좋은 모델 → 더 많은 사용자. 이 선순환을 어떻게 작은 팀도 구현할 수 있을까?
이 글은 bluefoxdev.kr의 AI 제품 성장 전략 를 참고하여 작성했습니다.
1. 데이터 플라이휠 구조
[AI 데이터 플라이휠 단계]
1. 사용자 인터랙션 수집:
AI 응답 → 사용자 행동 추적
좋아요/싫어요 → 명시적 피드백
수정/재생성 요청 → 암묵적 피드백
체류 시간, 클릭율 → 간접 피드백
2. 데이터 품질 필터링:
명확한 선호 신호만 선별
에지 케이스 식별
인간 검토가 필요한 것 분류
3. 자동 레이블링:
LLM-as-Judge로 품질 평가
기존 데이터와 유사도 확인
클러스터링으로 중복 제거
4. 모델 업데이트:
소량 고품질 데이터로 파인튜닝
또는 프롬프트 개선
또는 RAG 지식 베이스 업데이트
5. 평가 및 배포:
오프라인 벤치마크 테스트
온라인 A/B 테스트
점진적 롤아웃 (5% → 20% → 100%)
6. 다시 1단계로:
개선된 모델 → 더 많은 사용자
→ 더 많은 데이터
2. 피드백 수집 및 레이블링 자동화
import anthropic
from datetime import datetime
client = anthropic.Anthropic()
async def collect_implicit_feedback(
query: str,
ai_response: str,
user_action: str, # 'accepted', 'edited', 'regenerated', 'discarded'
edited_response: str | None = None,
) -> dict:
"""사용자 액션으로 암묵적 피드백 수집"""
quality_signal = {
"accepted": 1.0,
"edited": 0.6, # 편집 = 어느 정도 만족
"regenerated": 0.2, # 재생성 = 불만족
"discarded": 0.0,
}.get(user_action, 0.5)
feedback = {
"query": query,
"response": ai_response,
"quality_signal": quality_signal,
"user_action": user_action,
"edited_response": edited_response,
"timestamp": datetime.utcnow().isoformat(),
}
# 편집된 응답이 있으면 선호 쌍으로 저장
if user_action == "edited" and edited_response:
feedback["preference_pair"] = {
"chosen": edited_response, # 사용자가 선호하는 것
"rejected": ai_response, # AI가 생성한 것
}
await save_feedback(feedback)
return feedback
async def auto_label_with_llm(samples: list[dict]) -> list[dict]:
"""LLM-as-Judge로 데이터 자동 레이블링"""
labeled = []
for sample in samples:
response = client.messages.create(
model="claude-opus-4-7",
max_tokens=300,
messages=[{
"role": "user",
"content": f"""다음 AI 응답의 품질을 평가하세요.
질문: {sample['query']}
응답: {sample['response']}
JSON으로 평가:
{{
"overall_quality": 1-10,
"helpfulness": 1-10,
"accuracy": 1-10,
"conciseness": 1-10,
"should_include_in_training": true/false,
"improvement_notes": "개선 포인트 (선택)"
}}"""
}]
)
import json
label = json.loads(response.content[0].text)
if label["should_include_in_training"] and label["overall_quality"] >= 7:
labeled.append({
**sample,
"label": label,
"training_priority": label["overall_quality"],
})
return labeled
async def trigger_fine_tuning_pipeline(
labeled_data: list[dict],
min_samples: int = 100,
):
"""충분한 데이터 수집 시 파인튜닝 파이프라인 트리거"""
if len(labeled_data) < min_samples:
return {"status": "waiting", "current": len(labeled_data), "needed": min_samples}
# 학습 데이터 포맷 변환 (SFT 형식)
training_data = []
for sample in labeled_data:
if "preference_pair" in sample:
# DPO (Direct Preference Optimization) 형식
training_data.append({
"prompt": sample["query"],
"chosen": sample["preference_pair"]["chosen"],
"rejected": sample["preference_pair"]["rejected"],
})
else:
# SFT 형식
training_data.append({
"messages": [
{"role": "user", "content": sample["query"]},
{"role": "assistant", "content": sample["response"]},
]
})
# 파인튜닝 잡 시작 (예: OpenAI, Anthropic, 자체 서버)
job = await start_fine_tuning_job(training_data)
return {"status": "started", "job_id": job["id"], "samples": len(training_data)}
마무리
데이터 플라이휠의 핵심은 "사용자 행동이 곧 학습 데이터"라는 인식이다. 사용자가 AI 응답을 편집하면 선호 데이터(DPO)가 자동 생성된다. 처음에는 피드백 버튼 하나(👍/👎)만 추가해도 데이터 수집이 시작된다. 월 1,000개 피드백이 쌓이면 의미 있는 파인튜닝이 가능하다.