코드 리뷰 특화 모델을 데이터 수집부터 프로덕션 배포까지 전 과정을 실습하며, 시리즈에서 배운 모든 기법을 통합 적용합니다.
이 장에서는 시리즈 전체에서 배운 내용을 하나의 실전 프로젝트로 통합합니다. 목표는 Python 코드 리뷰에 특화된 LLM을 파인튜닝하여, 코드의 보안 취약점, 성능 문제, 스타일 개선점을 자동으로 분석하는 모델을 만드는 것입니다.
프로젝트 목표:
- 입력: Python 코드 스니펫
- 출력: 구조화된 코드 리뷰 (보안, 성능, 스타일, 개선 제안)
- 베이스 모델: Llama 3.1 8B Instruct
- 파인튜닝 방법: QLoRA
- 목표 성능: 베이스 모델 대비 리뷰 품질 30% 이상 향상
프로젝트 단계:
1. 데이터 수집 및 구축
2. 데이터 전처리 및 품질 검증
3. QLoRA 파인튜닝
4. 평가 및 벤치마킹
5. 모델 레지스트리 등록
6. 배포 및 서빙코드 리뷰 학습 데이터를 세 가지 소스에서 수집합니다.
GitHub의 공개 Pull Request 리뷰를 수집합니다.
# scripts/collect_github_reviews.py
import json
from pathlib import Path
from urllib.request import urlopen, Request
import os
def fetch_pr_reviews(
owner: str,
repo: str,
per_page: int = 30,
max_prs: int = 100
) -> list[dict]:
"""GitHub API에서 PR 리뷰 수집"""
reviews = []
token = os.environ.get("GITHUB_TOKEN", "")
headers = {}
if token:
headers["Authorization"] = "Bearer " + token
headers["Accept"] = "application/vnd.github.v3+json"
page = 1
collected = 0
while collected < max_prs:
url = (
"https://api.github.com/repos/"
+ owner + "/" + repo
+ "/pulls?state=closed&per_page="
+ str(per_page) + "&page=" + str(page)
)
req = Request(url, headers=headers)
resp = urlopen(req)
prs = json.loads(resp.read())
if not prs:
break
for pr in prs:
pr_number = pr["number"]
review_url = (
"https://api.github.com/repos/"
+ owner + "/" + repo
+ "/pulls/" + str(pr_number) + "/reviews"
)
rev_req = Request(review_url, headers=headers)
rev_resp = urlopen(rev_req)
pr_reviews = json.loads(rev_resp.read())
for review in pr_reviews:
if review["body"] and len(review["body"]) > 100:
reviews.append({
"repo": owner + "/" + repo,
"pr_number": pr_number,
"pr_title": pr["title"],
"review_body": review["body"],
"review_state": review["state"],
})
collected += 1
page += 1
return reviews
# 주요 Python 프로젝트에서 수집
repos = [
("python", "cpython"),
("django", "django"),
("pallets", "flask"),
("psf", "requests"),
]
all_reviews = []
for owner, repo in repos:
reviews = fetch_pr_reviews(owner, repo, max_prs=50)
all_reviews.extend(reviews)
print(owner + "/" + repo + ": " + str(len(reviews)) + "개")수집된 실제 데이터를 보완하기 위해 합성 데이터를 생성합니다.
# scripts/generate_synthetic_reviews.py
import anthropic
import json
client = anthropic.Anthropic()
CODE_REVIEW_CATEGORIES = [
"SQL 인젝션 취약점이 있는 코드",
"XSS 취약점이 있는 웹 코드",
"비효율적인 루프 처리",
"메모리 누수가 발생하는 코드",
"에러 처리가 부족한 코드",
"타입 힌트가 없는 코드",
"과도하게 복잡한 함수",
"테스트가 어려운 구조의 코드",
"하드코딩된 설정값",
"경쟁 조건이 발생하는 비동기 코드",
]
def generate_review_pair(category: str) -> dict:
"""코드와 리뷰 쌍 생성"""
prompt = (
"Python 코드 리뷰 학습 데이터를 생성해 주세요.\n\n"
"카테고리: " + category + "\n\n"
"다음 JSON 형식으로 응답해 주세요:\n"
"{\n"
' "code": "리뷰 대상 Python 코드 (20~50줄)",\n'
' "review": "구조화된 코드 리뷰"\n'
"}\n\n"
"리뷰 형식:\n"
"1. 요약: 전체적인 코드 평가 (1~2문장)\n"
"2. 보안: 보안 취약점 분석\n"
"3. 성능: 성능 개선 사항\n"
"4. 가독성: 코드 스타일 및 구조 개선\n"
"5. 개선된 코드: 수정된 코드 제시\n\n"
"리뷰는 전문적이고 구체적이어야 하며, "
"단순히 문제만 지적하는 것이 아니라 "
"해결 방법도 함께 제시해야 합니다."
)
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[{"role": "user", "content": prompt}]
)
text = response.content[0].text
# JSON 파싱
try:
start = text.index("{")
end = text.rindex("}") + 1
return json.loads(text[start:end])
except (ValueError, json.JSONDecodeError):
return None
def generate_dataset(
num_per_category: int = 50
) -> list[dict]:
"""전체 합성 데이터셋 생성"""
dataset = []
for category in CODE_REVIEW_CATEGORIES:
print("생성 중: " + category)
for i in range(num_per_category):
pair = generate_review_pair(category)
if pair:
dataset.append({
"messages": [
{
"role": "system",
"content": (
"당신은 시니어 Python 개발자입니다. "
"코드 리뷰를 수행할 때 보안, 성능, "
"가독성 관점에서 구체적인 개선 사항을 "
"제시합니다. 단순히 문제만 지적하지 않고, "
"개선된 코드도 함께 제안합니다."
)
},
{
"role": "user",
"content": (
"다음 Python 코드를 리뷰해 주세요.\n\n"
+ pair["code"]
)
},
{
"role": "assistant",
"content": pair["review"]
}
],
"category": category,
"source": "synthetic"
})
print(
" 완료: " + str(len(dataset)) + "개 누적"
)
return dataset# scripts/prepare_dataset.py
import json
from pathlib import Path
def prepare_final_dataset(
github_reviews: list[dict],
synthetic_data: list[dict],
output_dir: str
):
"""최종 학습 데이터셋 준비"""
all_data = []
# GitHub 리뷰 데이터를 대화 형식으로 변환
for review in github_reviews:
entry = {
"messages": [
{
"role": "system",
"content": (
"당신은 시니어 Python 개발자입니다. "
"코드 리뷰를 수행할 때 보안, 성능, "
"가독성 관점에서 구체적인 개선 사항을 "
"제시합니다."
)
},
{
"role": "user",
"content": (
"다음 PR의 코드를 리뷰해 주세요.\n\n"
"PR 제목: " + review["pr_title"]
)
},
{
"role": "assistant",
"content": review["review_body"]
}
],
"source": "github"
}
all_data.append(entry)
# 합성 데이터 추가
all_data.extend(synthetic_data)
print("전체 데이터: " + str(len(all_data)) + "개")
print(
" GitHub: " + str(len(github_reviews)) + "개, "
"합성: " + str(len(synthetic_data)) + "개"
)
# 저장
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
with open(output / "all_data.jsonl", "w") as f:
for d in all_data:
f.write(json.dumps(d, ensure_ascii=False) + "\n")
return all_data3장에서 구축한 전처리 파이프라인을 적용합니다.
# scripts/preprocess.py
from data_pipeline import DataPipeline
def run_preprocessing():
"""전처리 파이프라인 실행"""
pipeline = DataPipeline(
raw_dir="data/raw",
processed_dir="data/processed"
)
pipeline.run()
# 추가: 코드 리뷰 특화 검증
validate_code_review_quality("data/processed/train.jsonl")
def validate_code_review_quality(file_path: str):
"""코드 리뷰 데이터 품질 검증"""
import json
with open(file_path) as f:
data = [json.loads(line) for line in f]
issues = []
for i, example in enumerate(data):
assistant_msg = next(
(m for m in example["messages"]
if m["role"] == "assistant"),
None
)
if not assistant_msg:
issues.append(
"예제 " + str(i) + ": 어시스턴트 응답 없음"
)
continue
review = assistant_msg["content"]
# 최소 길이 확인
if len(review) < 200:
issues.append(
"예제 " + str(i) + ": 리뷰가 너무 짧음 ("
+ str(len(review)) + "자)"
)
# 구조 확인 (최소한 하나의 섹션 제목 포함)
has_structure = any(
marker in review
for marker in ["보안", "성능", "가독성", "개선"]
)
if not has_structure:
issues.append(
"예제 " + str(i) + ": 구조화되지 않은 리뷰"
)
if issues:
print("품질 이슈 " + str(len(issues)) + "건:")
for issue in issues[:10]:
print(" - " + issue)
else:
print("모든 데이터 품질 검증 통과")
if __name__ == "__main__":
run_preprocessing()5장과 6장의 내용을 통합한 학습 스크립트입니다.
# train_code_reviewer.py
import torch
from datasets import load_dataset
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
BitsAndBytesConfig,
TrainingArguments,
EarlyStoppingCallback,
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer
import wandb
import json
from pathlib import Path
def main():
# === 설정 ===
BASE_MODEL = "meta-llama/Llama-3.1-8B-Instruct"
OUTPUT_DIR = "./output/code-reviewer"
LORA_R = 32
LORA_ALPHA = 64
LEARNING_RATE = 2e-4
EPOCHS = 3
BATCH_SIZE = 2
GRAD_ACCUM = 8
MAX_SEQ_LEN = 2048
# === wandb ===
wandb.init(
project="code-reviewer",
name="qlora-v1",
config={
"base_model": BASE_MODEL,
"lora_r": LORA_R,
"lora_alpha": LORA_ALPHA,
"learning_rate": LEARNING_RATE,
"epochs": EPOCHS,
"effective_batch_size": BATCH_SIZE * GRAD_ACCUM,
}
)
# === 모델 로드 ===
print("1. 모델 로드 중...")
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
)
model = AutoModelForCausalLM.from_pretrained(
BASE_MODEL,
quantization_config=bnb_config,
device_map="auto",
attn_implementation="flash_attention_2",
)
model = prepare_model_for_kbit_training(
model, use_gradient_checkpointing=True
)
tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
# === LoRA ===
print("2. LoRA 설정 중...")
lora_config = LoraConfig(
r=LORA_R,
lora_alpha=LORA_ALPHA,
target_modules="all-linear",
lora_dropout=0.05,
task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# === 데이터 ===
print("3. 데이터 로드 중...")
dataset = load_dataset("json", data_files={
"train": "data/processed/train.jsonl",
"validation": "data/processed/val.jsonl",
})
def formatting_func(example):
return tokenizer.apply_chat_template(
example["messages"],
tokenize=False,
add_generation_prompt=False,
)
# === 학습 ===
print("4. 학습 시작...")
training_args = TrainingArguments(
output_dir=OUTPUT_DIR,
num_train_epochs=EPOCHS,
per_device_train_batch_size=BATCH_SIZE,
gradient_accumulation_steps=GRAD_ACCUM,
learning_rate=LEARNING_RATE,
lr_scheduler_type="cosine",
warmup_ratio=0.1,
weight_decay=0.01,
max_grad_norm=0.3,
logging_steps=10,
save_strategy="steps",
save_steps=100,
eval_strategy="steps",
eval_steps=100,
save_total_limit=3,
load_best_model_at_end=True,
metric_for_best_model="eval_loss",
bf16=True,
gradient_checkpointing=True,
gradient_checkpointing_kwargs={"use_reentrant": False},
optim="paged_adamw_8bit",
report_to="wandb",
seed=42,
)
trainer = SFTTrainer(
model=model,
args=training_args,
train_dataset=dataset["train"],
eval_dataset=dataset["validation"],
tokenizer=tokenizer,
formatting_func=formatting_func,
max_seq_length=MAX_SEQ_LEN,
packing=True,
callbacks=[
EarlyStoppingCallback(early_stopping_patience=3)
],
)
train_result = trainer.train()
# === 저장 ===
print("5. 모델 저장 중...")
best_adapter_path = OUTPUT_DIR + "/best-adapter"
trainer.save_model(best_adapter_path)
# 학습 결과 저장
results = {
"train_loss": train_result.training_loss,
"train_runtime": train_result.metrics["train_runtime"],
"train_samples_per_second": train_result.metrics[
"train_samples_per_second"
],
}
with open(OUTPUT_DIR + "/train_results.json", "w") as f:
json.dump(results, f, indent=2)
wandb.finish()
print("학습 완료")
if __name__ == "__main__":
main()7장의 평가 프레임워크를 적용합니다.
# evaluate_code_reviewer.py
import json
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import PeftModel
def load_model(adapter_path: str):
"""QLoRA 모델 로드"""
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
)
base_model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-3.1-8B-Instruct",
quantization_config=bnb_config,
device_map="auto",
)
model = PeftModel.from_pretrained(base_model, adapter_path)
model.eval()
tokenizer = AutoTokenizer.from_pretrained(
"meta-llama/Llama-3.1-8B-Instruct"
)
return model, tokenizer
def evaluate_code_review_quality(
model,
tokenizer,
test_data: list[dict]
) -> dict:
"""코드 리뷰 품질 평가"""
results = {
"total": len(test_data),
"has_security_analysis": 0,
"has_performance_analysis": 0,
"has_style_analysis": 0,
"has_improved_code": 0,
"avg_review_length": 0,
"samples": [],
}
total_length = 0
for example in test_data:
# 입력 준비 (어시스턴트 응답 제외)
messages = [
m for m in example["messages"]
if m["role"] != "assistant"
]
messages_for_gen = messages + []
inputs = tokenizer.apply_chat_template(
messages_for_gen,
return_tensors="pt",
add_generation_prompt=True,
)
# 응답 생성
with torch.no_grad():
outputs = model.generate(
inputs.to(model.device),
max_new_tokens=1024,
temperature=0.3,
do_sample=True,
top_p=0.9,
)
generated = tokenizer.decode(
outputs[0][inputs.shape[1]:],
skip_special_tokens=True,
)
# 품질 분석
if "보안" in generated or "취약점" in generated:
results["has_security_analysis"] += 1
if "성능" in generated or "최적화" in generated:
results["has_performance_analysis"] += 1
if "가독성" in generated or "스타일" in generated:
results["has_style_analysis"] += 1
if "개선" in generated or "수정" in generated:
results["has_improved_code"] += 1
total_length += len(generated)
# 샘플 저장 (처음 5개)
if len(results["samples"]) < 5:
expected = next(
(m["content"] for m in example["messages"]
if m["role"] == "assistant"),
""
)
user_input = next(
(m["content"] for m in example["messages"]
if m["role"] == "user"),
""
)
results["samples"].append({
"input": user_input[:200],
"expected": expected[:300],
"generated": generated[:300],
})
# 비율 계산
n = results["total"]
results["security_rate"] = results["has_security_analysis"] / n
results["performance_rate"] = results["has_performance_analysis"] / n
results["style_rate"] = results["has_style_analysis"] / n
results["improved_code_rate"] = results["has_improved_code"] / n
results["avg_review_length"] = total_length / n
return results
def run_evaluation(adapter_path: str, test_file: str):
"""전체 평가 실행"""
print("모델 로드 중...")
model, tokenizer = load_model(adapter_path)
print("테스트 데이터 로드 중...")
with open(test_file) as f:
test_data = [json.loads(line) for line in f]
# 최대 100개 평가
test_data = test_data[:100]
print("평가 중... (" + str(len(test_data)) + "개)")
results = evaluate_code_review_quality(model, tokenizer, test_data)
# 결과 출력
print("\n평가 결과:")
print("=" * 50)
print("보안 분석 포함율: " + str(round(results["security_rate"], 2)))
print("성능 분석 포함율: " + str(round(results["performance_rate"], 2)))
print("스타일 분석 포함율: " + str(round(results["style_rate"], 2)))
print("개선 코드 포함율: " + str(round(results["improved_code_rate"], 2)))
print("평균 리뷰 길이: " + str(round(results["avg_review_length"])) + "자")
# 결과 저장
with open("eval_results.json", "w") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print("\n결과 저장: eval_results.json")
return results
if __name__ == "__main__":
run_evaluation(
adapter_path="./output/code-reviewer/best-adapter",
test_file="data/processed/test.jsonl",
)8장의 레지스트리를 활용합니다.
# scripts/register.py
from model_registry import LocalModelRegistry
def register_code_reviewer():
"""코드 리뷰 모델을 레지스트리에 등록"""
import json
registry = LocalModelRegistry("./model-registry")
# 평가 결과 로드
with open("eval_results.json") as f:
eval_results = json.load(f)
# 학습 결과 로드
with open("output/code-reviewer/train_results.json") as f:
train_results = json.load(f)
# 모델 등록
version = registry.register_model(
model_name="code-reviewer",
adapter_path="./output/code-reviewer/best-adapter",
base_model="meta-llama/Llama-3.1-8B-Instruct",
method="QLoRA",
training_config={
"lora_r": 32,
"lora_alpha": 64,
"learning_rate": 2e-4,
"epochs": 3,
"batch_size": 16,
"max_seq_length": 2048,
},
evaluation_results={
"train_loss": train_results["train_loss"],
"security_rate": eval_results["security_rate"],
"performance_rate": eval_results["performance_rate"],
"style_rate": eval_results["style_rate"],
"improved_code_rate": eval_results["improved_code_rate"],
"avg_review_length": eval_results["avg_review_length"],
},
data_info={
"train_file": "data/processed/train.jsonl",
"sources": ["github", "synthetic"],
},
description="코드 리뷰 특화 QLoRA 모델 v1",
)
print("등록 완료: v" + str(version.version))
# 프로덕션 승격
registry.promote_to_production("code-reviewer", version.version)
if __name__ == "__main__":
register_code_reviewer()파인튜닝된 모델을 API 서버로 배포합니다.
# serve.py
import torch
from fastapi import FastAPI
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import PeftModel
app = FastAPI(title="Code Review API")
class ReviewRequest(BaseModel):
code: str
language: str = "python"
class ReviewResponse(BaseModel):
review: str
model_version: str
# 모델 로드 (서버 시작 시 1회)
print("모델 로드 중...")
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
)
base_model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-3.1-8B-Instruct",
quantization_config=bnb_config,
device_map="auto",
)
model = PeftModel.from_pretrained(
base_model,
"./output/code-reviewer/best-adapter"
)
model.eval()
tokenizer = AutoTokenizer.from_pretrained(
"meta-llama/Llama-3.1-8B-Instruct"
)
print("모델 로드 완료")
@app.post("/review", response_model=ReviewResponse)
async def review_code(request: ReviewRequest):
"""코드 리뷰 API"""
messages = [
{
"role": "system",
"content": (
"당신은 시니어 Python 개발자입니다. "
"코드 리뷰를 수행할 때 보안, 성능, "
"가독성 관점에서 구체적인 개선 사항을 "
"제시합니다. 단순히 문제만 지적하지 않고, "
"개선된 코드도 함께 제안합니다."
)
},
{
"role": "user",
"content": "다음 코드를 리뷰해 주세요.\n\n" + request.code
}
]
inputs = tokenizer.apply_chat_template(
messages,
return_tensors="pt",
add_generation_prompt=True,
)
with torch.no_grad():
outputs = model.generate(
inputs.to(model.device),
max_new_tokens=1024,
temperature=0.3,
do_sample=True,
top_p=0.9,
repetition_penalty=1.1,
)
review = tokenizer.decode(
outputs[0][inputs.shape[1]:],
skip_special_tokens=True,
)
return ReviewResponse(
review=review,
model_version="v1"
)
@app.get("/health")
async def health():
return {"status": "healthy", "model": "code-reviewer-v1"}# 서버 실행
uvicorn serve:app --host 0.0.0.0 --port 8000# Dockerfile
FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04
WORKDIR /app
# Python 설치
RUN apt-get update && apt-get install -y python3 python3-pip
RUN pip3 install torch transformers peft bitsandbytes fastapi uvicorn
# 모델 및 코드 복사
COPY serve.py .
COPY output/code-reviewer/best-adapter ./adapter/
ENV MODEL_PATH=/app/adapter
EXPOSE 8000
CMD ["uvicorn", "serve:app", "--host", "0.0.0.0", "--port", "8000"]import requests
response = requests.post(
"http://localhost:8000/review",
json={
"code": """
def get_user_data(user_id):
query = "SELECT * FROM users WHERE id = " + str(user_id)
result = db.execute(query)
password = result['password']
return {
'id': result['id'],
'name': result['name'],
'email': result['email'],
'password': password
}
"""
}
)
print(response.json()["review"])code-reviewer/
configs/
experiment_v1.yaml
data/
raw/
github_reviews.jsonl
synthetic_reviews.jsonl
processed/
train.jsonl
val.jsonl
test.jsonl
scripts/
collect_github_reviews.py
generate_synthetic_reviews.py
prepare_dataset.py
preprocess.py
validate_data.py
quality_gate.py
register.py
notify.py
model-registry/
registry.json
code-reviewer/
v1/
adapter/
output/
code-reviewer/
best-adapter/
train_results.json
train_code_reviewer.py
evaluate_code_reviewer.py
serve.py
Dockerfile
requirements.txt
.github/
workflows/
fine-tuning-pipeline.yml이 시리즈를 통해 LLM 파인튜닝의 전체 과정을 다루었습니다. 핵심 내용을 정리합니다.
시리즈 핵심 요약:
1장: 파인튜닝의 개념과 유형, 적용 시점 판단
2장: 학습 데이터 형식, 수집 전략, 다양성 확보
3장: 데이터 정제, 중복 제거, 토큰화, 패킹
4장: LoRA의 수학적 원리와 실전 적용
5장: QLoRA로 소비자 GPU에서 파인튜닝
6장: 학습 파이프라인과 하이퍼파라미터 최적화
7장: 자동 메트릭, LLM 평가, 벤치마킹
8장: 모델 레지스트리와 버전 관리
9장: CI/CD 기반 학습-평가-배포 자동화
10장: 실전 프로젝트로 전체 과정 통합파인튜닝은 단순한 기술적 작업이 아니라, 데이터 설계, 모델 공학, 평가 체계, 운영 자동화가 결합된 종합적인 엔지니어링 과정입니다. 이 시리즈에서 다룬 각 단계의 원칙과 실전 기법이 실제 프로젝트에서 의미 있는 결과를 만들어내는 데 도움이 되기를 바랍니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
파인튜닝의 학습, 평가, 배포 전체 과정을 CI/CD 파이프라인으로 자동화하고, 데이터 변경이나 코드 변경 시 자동으로 모델이 업데이트되는 체계를 구축합니다.
파인튜닝된 모델을 체계적으로 관리하기 위한 모델 레지스트리 구축, 버전 관리, 메타데이터 추적, 아티팩트 저장 전략을 다룹니다.
파인튜닝된 모델의 성능을 자동 메트릭, LLM 평가, 인간 평가를 통해 다각적으로 측정하고 벤치마킹하는 체계적인 방법을 다룹니다.