Fine-tuning Claude trên AWS Bedrock — Hướng dẫn từng bước

Prompt engineering và RAG giải quyết được 90% use cases. Nhưng khi bạn cần Claude hoàn toàn thích nghi với domain-specific language, style, hoặc format của tổ chức — fine-tuning là bước tiếp theo. AWS Bedrock là platform chính thức để fine-tune Claude một cách managed và secure.

Fine-tuning vs Prompt Engineering — Khi nào nên fine-tune?

Fine-tuning KHÔNG phải lúc nào cũng là câu trả lời đúng. Cân nhắc kỹ trước khi invest vào quá trình phức tạp này:

Rule of thumb: Nếu có thể đạt kết quả tốt với < 10 examples trong prompt, dùng few-shot prompting thay vì fine-tune.

Bước 1: Chuẩn bị AWS Bedrock

import boto3
import json
import os

# Setup AWS credentials
session = boto3.Session(
    aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
    region_name="us-east-1"  # Bedrock fine-tuning available in us-east-1, us-west-2
)

bedrock_client = session.client('bedrock')
bedrock_runtime = session.client('bedrock-runtime')

# Kiểm tra models có sẵn để fine-tune
def list_available_base_models():
    """Liệt kê Claude models có thể fine-tune"""
    response = bedrock_client.list_foundation_models(
        byProvider="Anthropic",
        byOutputModality="TEXT",
        byCustomizationType="FINE_TUNING"
    )
    for model in response['modelSummaries']:
        print(f"Model: {model['modelId']}")
        print(f"  Name: {model['modelName']}")
        print(f"  Status: {model['modelLifecycle']['status']}")
    return response['modelSummaries']

available_models = list_available_base_models()

Bước 2: Chuẩn bị Training Data

Fine-tuning yêu cầu data theo format JSONL cụ thể. Mỗi dòng là một training example:

# Format chuẩn cho Claude fine-tuning trên Bedrock
training_examples = [
    {
        "prompt": "Viết email từ chối ứng viên nhẹ nhàng cho vị trí Senior Developer",
        "completion": """Kính gửi [Tên ứng viên],

Cảm ơn bạn đã dành thời gian tham gia quá trình tuyển dụng cho vị trí Senior Developer tại [Công ty].

Sau khi xem xét kỹ lưỡng, chúng tôi quyết định tiếp tục với ứng viên có profile phù hợp hơn với nhu cầu hiện tại của team. Quyết định này không hề dễ dàng vì bạn có nhiều điểm mạnh đáng ghi nhận.

Chúng tôi sẽ lưu CV của bạn để xem xét cho các cơ hội phù hợp trong tương lai. Chúc bạn thành công trong sự nghiệp.

Trân trọng,
[Tên HR]"""
    },
    {
        "prompt": "Viết email xác nhận offer letter cho ứng viên đã được nhận",
        "completion": """Kính gửi [Tên ứng viên],

Thay mặt Ban Giám đốc [Công ty], tôi vui mừng thông báo bạn đã được chính thức chấp nhận vào vị trí [Chức vụ].

Chi tiết offer:
- Vị trí: [Chức vụ]
- Mức lương: [Số tiền] VND/tháng
- Ngày bắt đầu: [Ngày]
- Địa điểm: [Văn phòng]

Vui lòng xác nhận nhận email này và phản hồi trong vòng 3 ngày làm việc. Chúng tôi sẽ gửi hợp đồng chính thức sau khi nhận được xác nhận của bạn.

Chào mừng bạn đến với đội ngũ [Công ty]!

Trân trọng,
[Tên HR]"""
    }
    # ... minimum 100-1000 examples cho fine-tuning hiệu quả
]

def prepare_training_data(examples: list, output_path: str):
    """Chuẩn bị file JSONL cho Bedrock fine-tuning"""
    with open(output_path, 'w', encoding='utf-8') as f:
        for example in examples:
            # Bedrock format: anthropic_version + messages
            formatted = {
                "anthropic_version": "bedrock-2023-05-31",
                "messages": [
                    {"role": "user", "content": example["prompt"]},
                    {"role": "assistant", "content": example["completion"]}
                ]
            }
            f.write(json.dumps(formatted, ensure_ascii=False) + "
")

    print(f"Prepared {len(examples)} training examples -> {output_path}")
    return output_path

training_file = prepare_training_data(training_examples, "/tmp/training_data.jsonl")

Bước 3: Kiểm tra chất lượng Training Data

def validate_training_data(file_path: str) -> dict:
    """Validate training data trước khi upload"""
    issues = []
    stats = {
        "total_examples": 0,
        "avg_prompt_length": 0,
        "avg_completion_length": 0,
        "min_completion_tokens": float('inf'),
        "max_completion_tokens": 0
    }

    prompt_lengths = []
    completion_lengths = []

    with open(file_path, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            try:
                example = json.loads(line.strip())
                messages = example.get("messages", [])

                if len(messages) < 2:
                    issues.append(f"Line {line_num}: Need at least 2 messages (user + assistant)")
                    continue

                user_msg = next((m for m in messages if m["role"] == "user"), None)
                assistant_msg = next((m for m in messages if m["role"] == "assistant"), None)

                if not user_msg:
                    issues.append(f"Line {line_num}: Missing user message")
                if not assistant_msg:
                    issues.append(f"Line {line_num}: Missing assistant message")

                # Check lengths
                user_len = len(user_msg["content"]) if user_msg else 0
                asst_len = len(assistant_msg["content"]) if assistant_msg else 0

                if user_len < 10:
                    issues.append(f"Line {line_num}: Prompt too short ({user_len} chars)")
                if asst_len < 20:
                    issues.append(f"Line {line_num}: Completion too short ({asst_len} chars)")

                prompt_lengths.append(user_len)
                completion_lengths.append(asst_len)
                stats["total_examples"] += 1

            except json.JSONDecodeError as e:
                issues.append(f"Line {line_num}: JSON parse error: {e}")

    if prompt_lengths:
        stats["avg_prompt_length"] = sum(prompt_lengths) / len(prompt_lengths)
    if completion_lengths:
        stats["avg_completion_length"] = sum(completion_lengths) / len(completion_lengths)
        stats["min_completion_tokens"] = min(completion_lengths)
        stats["max_completion_tokens"] = max(completion_lengths)

    print(f"Validation Results:")
    print(f"  Total examples: {stats['total_examples']}")
    print(f"  Avg prompt: {stats['avg_prompt_length']:.0f} chars")
    print(f"  Avg completion: {stats['avg_completion_length']:.0f} chars")
    print(f"  Issues found: {len(issues)}")
    for issue in issues[:5]:  # Show first 5 issues
        print(f"  - {issue}")

    return {"stats": stats, "issues": issues, "valid": len(issues) == 0}

validation = validate_training_data(training_file)

Bước 4: Upload Data lên S3

import boto3
from datetime import datetime

s3_client = session.client('s3')

def upload_training_data(local_path: str, bucket: str, prefix: str = "bedrock-ft") -> str:
    """Upload training data lên S3"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    s3_key = f"{prefix}/training/{timestamp}/training_data.jsonl"

    s3_client.upload_file(local_path, bucket, s3_key)
    s3_uri = f"s3://{bucket}/{s3_key}"
    print(f"Uploaded to: {s3_uri}")
    return s3_uri

S3_BUCKET = "my-bedrock-finetuning-bucket"
training_s3_uri = upload_training_data(training_file, S3_BUCKET)

Bước 5: Tạo Fine-tuning Job

def create_finetuning_job(
    training_data_uri: str,
    output_s3_uri: str,
    base_model_id: str = "anthropic.claude-haiku-20240307-v1:0",
    job_name: str = None,
    num_epochs: int = 2,
    batch_size: int = 8,
    learning_rate: float = 0.00001
) -> str:
    """Tạo và start fine-tuning job"""

    if not job_name:
        job_name = f"claude-ft-{datetime.now().strftime('%Y%m%d-%H%M%S')}"

    response = bedrock_client.create_model_customization_job(
        jobName=job_name,
        baseModelIdentifier=base_model_id,
        customizationType="FINE_TUNING",
        roleArn=os.environ.get("BEDROCK_ROLE_ARN"),  # IAM role với Bedrock + S3 access
        trainingDataConfig={
            "s3Uri": training_data_uri
        },
        outputDataConfig={
            "s3Uri": output_s3_uri
        },
        hyperParameters={
            "epochCount": str(num_epochs),
            "batchSize": str(batch_size),
            "learningRate": str(learning_rate)
        }
    )

    job_arn = response['jobArn']
    print(f"Fine-tuning job created: {job_arn}")
    return job_arn

job_arn = create_finetuning_job(
    training_data_uri=training_s3_uri,
    output_s3_uri=f"s3://{S3_BUCKET}/bedrock-ft/output/",
    num_epochs=3,
    batch_size=8,
    learning_rate=0.00001
)

Bước 6: Monitor Training Progress

import time

def monitor_training_job(job_arn: str, poll_interval: int = 60) -> dict:
    """Monitor fine-tuning job cho đến khi hoàn thành"""
    print(f"Monitoring job: {job_arn}")
    start_time = datetime.now()

    while True:
        response = bedrock_client.get_model_customization_job(
            jobIdentifier=job_arn
        )

        status = response['status']
        elapsed = (datetime.now() - start_time).total_seconds() / 60

        print(f"[{elapsed:.0f}min] Status: {status}")

        if status == "Completed":
            print(f"Training completed!")
            print(f"Output model ARN: {response.get('outputModelArn')}")
            return {
                "status": "completed",
                "model_arn": response.get('outputModelArn'),
                "duration_minutes": elapsed
            }

        elif status in ["Failed", "Stopped"]:
            print(f"Training {status}")
            print(f"Reason: {response.get('failureMessage', 'Unknown')}")
            return {"status": status.lower(), "error": response.get('failureMessage')}

        elif status in ["InProgress", "Starting"]:
            # Log training metrics nếu có
            metrics = response.get('trainingMetrics', {})
            if metrics:
                print(f"  Training loss: {metrics.get('trainingLoss', 'N/A')}")

        time.sleep(poll_interval)

result = monitor_training_job(job_arn)

Bước 7: Evaluate Fine-tuned Model

def evaluate_model(
    finetuned_model_arn: str,
    test_prompts: list[dict],
    base_model_id: str = "anthropic.claude-haiku-20240307-v1:0"
) -> dict:
    """So sánh fine-tuned model vs base model"""

    results = []

    for test in test_prompts:
        # Call fine-tuned model
        ft_response = bedrock_runtime.invoke_model(
            modelId=finetuned_model_arn,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1000,
                "messages": [{"role": "user", "content": test["prompt"]}]
            })
        )
        ft_result = json.loads(ft_response['body'].read())
        ft_output = ft_result['content'][0]['text']

        # Call base model
        base_response = bedrock_runtime.invoke_model(
            modelId=base_model_id,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1000,
                "messages": [{"role": "user", "content": test["prompt"]}]
            })
        )
        base_result = json.loads(base_response['body'].read())
        base_output = base_result['content'][0]['text']

        results.append({
            "prompt": test["prompt"],
            "expected": test.get("expected", ""),
            "finetuned_output": ft_output,
            "base_output": base_output
        })

    # Dùng Claude để score results
    scored_results = score_with_claude(results)
    return scored_results

def score_with_claude(results: list) -> dict:
    """Dùng Claude làm judge để evaluate kết quả"""
    import anthropic
    client = anthropic.Anthropic()

    scores = {"finetuned": [], "base": []}

    for result in results[:5]:  # Sample 5 for evaluation
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Compare these two responses to the prompt. Rate each 1-10 for quality, style-match, and accuracy.

Prompt: {result['prompt']}
Expected style: {result.get('expected', 'Professional Vietnamese business email')}

Response A (Fine-tuned): {result['finetuned_output'][:300]}
Response B (Base model): {result['base_output'][:300]}

Score format: A:X B:Y (just numbers)"""
            }]
        )
        text = response.content[0].text
        # Parse scores
        try:
            parts = text.split()
            a_score = float([p for p in parts if p.startswith("A:")][0].split(":")[1])
            b_score = float([p for p in parts if p.startswith("B:")][0].split(":")[1])
            scores["finetuned"].append(a_score)
            scores["base"].append(b_score)
        except Exception:
            pass

    avg_ft = sum(scores["finetuned"]) / len(scores["finetuned"]) if scores["finetuned"] else 0
    avg_base = sum(scores["base"]) / len(scores["base"]) if scores["base"] else 0

    print(f"Fine-tuned model score: {avg_ft:.1f}/10")
    print(f"Base model score: {avg_base:.1f}/10")
    print(f"Improvement: {avg_ft - avg_base:+.1f} points")

    return {"finetuned_score": avg_ft, "base_score": avg_base, "improvement": avg_ft - avg_base}

Chi phí Fine-tuning

Typical training cost: 1000 examples x ~500 tokens avg = 500K tokens = khoảng $4 USD. Nhỏ so với ongoing prompt engineering time.

Tổng kết

Fine-tuning Claude trên AWS Bedrock là quy trình có cấu trúc rõ ràng: prepare data → validate → upload S3 → create job → monitor → evaluate. Thành công phụ thuộc vào chất lượng training data hơn là số lượng.

Chỉ nên fine-tune khi: có 500+ high-quality examples, task có consistent style/format mà prompt engineering không đạt được, và ROI rõ ràng so với cost.

Xem thêm: Tool Evaluation trong Agent Systems để build testing frameworks cho cả fine-tuned và base models.

Bài viết liên quan

• Batch Processing — Xử lý hàng loạt request với Claude API
• RAG với Pinecone + Claude — Vector database cho AI
• Multi-Document Agent — Truy vấn nhiều tài liệu với LlamaIndex
• Claude cho Data: Validation và data quality
• Claude cho Engineering: Debug và xử lý lỗi