ファインチューニング済みモデルの評価とデプロイ

デプロイ前に評価が必要な理由

保留テストセットの構築

import json
from typing import TypedDict

class TestCase(TypedDict):
    input: str                 # the user message
    expected_output: str       # the ideal response
    category: str              # e.g., 'format', 'accuracy', 'edge_case'
    evaluation_method: str     # 'exact_match', 'json_schema', 'llm_judge'

# Load test set (never used during training)
def load_test_set(path: str) -> list[TestCase]:
    cases = []
    with open(path) as f:
        for line in f:
            data = json.loads(line.strip())
            cases.append({
                'input': data['messages'][-2]['content'],  # user message
                'expected_output': data['messages'][-1]['content'],  # assistant response
                'category': data.get('metadata', {}).get('category', 'general'),
                'evaluation_method': data.get('metadata', {}).get('eval_method', 'llm_judge')
            })
    return cases

test_set = load_test_set('test.jsonl')
print(f'Test set loaded: {len(test_set)} examples')

タスク固有の評価に使う定量指標

import json

def evaluate_json_extraction(model_output: str, expected: str, schema: dict) -> dict:
    metrics = {'valid_json': False, 'schema_compliant': False, 'field_accuracy': 0.0}
    
    try:
        parsed = json.loads(model_output.strip())
        metrics['valid_json'] = True
        
        # Check schema compliance
        required_fields = schema.get('required', [])
        all_present = all(field in parsed for field in required_fields)
        correct_types = all(
            isinstance(parsed.get(field), schema['properties'][field]['expected_type'])
            for field in required_fields if field in parsed
        )
        metrics['schema_compliant'] = all_present and correct_types
        
        # Field-level accuracy against expected output
        expected_parsed = json.loads(expected)
        correct_fields = sum(1 for k in expected_parsed if parsed.get(k) == expected_parsed[k])
        metrics['field_accuracy'] = correct_fields / len(expected_parsed) if expected_parsed else 0.0
    
    except json.JSONDecodeError:
        pass  # valid_json stays False
    
    return metrics

LLM-as-Judgeによる評価

from openai import OpenAI

client = OpenAI()

def llm_judge_score(instruction: str, expected: str, actual: str) -> dict:
    judge_prompt = f'''Evaluate the quality of an AI assistant response.

Instruction given to assistant:
{instruction}

Expected ideal response:
{expected}

Actual response from model being evaluated:
{actual}

Rate the actual response on these criteria (1=poor, 5=excellent):
1. Correctness: Is the information accurate?
2. Format compliance: Does it follow the expected output format?
3. Completeness: Does it address all parts of the instruction?

Return JSON: {{"correctness": N, "format": N, "completeness": N, "overall": N, "reason": "brief explanation"}}'''
    
    response = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': judge_prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(response.choices[0].message.content)

比較評価の実行

def run_full_evaluation(test_set: list, models: dict, system_prompt: str) -> dict:
    results = {name: {'scores': [], 'errors': 0} for name in models}
    
    for i, test_case in enumerate(test_set):
        print(f'Evaluating test case {i+1}/{len(test_set)}')
        
        for model_name, model_fn in models.items():
            try:
                output = model_fn(test_case['input'], system_prompt)
                score = llm_judge_score(
                    test_case['input'],
                    test_case['expected_output'],
                    output
                )
                results[model_name]['scores'].append(score['overall'])
            except Exception as e:
                results[model_name]['errors'] += 1
                results[model_name]['scores'].append(0)
    
    # Summarize
    summary = {}
    for name, data in results.items():
        scores = data['scores']
        summary[name] = {
            'mean_score': sum(scores) / len(scores),
            'errors': data['errors']
        }
        print(f'{name}: mean={summary[name]["mean_score"]:.2f}, errors={data["errors"]}')
    return summary

壊滅的忘却の回帰テスト

REGRESSION_TEST_CASES = [
    # General QA
    {'input': 'What is the capital of France?', 'expected_substring': 'Paris'},
    {'input': 'What is 17 * 23?', 'expected_substring': '391'},
    # Instruction following
    {'input': 'List 3 planets. Format as: 1. Planet Name', 'expected_pattern': r'^1\. '},
    # Reasoning
    {'input': 'If all A are B and all B are C, are all A also C?', 'expected_substring': 'yes'},
]

def run_regression_tests(model_fn, test_cases: list) -> float:
    passed = 0
    for test in test_cases:
        output = model_fn(test['input'], '')
        if 'expected_substring' in test:
            if test['expected_substring'].lower() in output.lower():
                passed += 1
        elif 'expected_pattern' in test:
            import re
            if re.search(test['expected_pattern'], output):
                passed += 1
    
    rate = passed / len(test_cases)
    print(f'Regression test pass rate: {rate:.1%} ({passed}/{len(test_cases)})')
    return rate

ローカル推論用のGGUFへの変換

# Step 1: Convert merged HuggingFace model to GGUF
# git clone https://github.com/ggerganov/llama.cpp
# python llama.cpp/convert_hf_to_gguf.py ./merged-model --outtype f16 --outfile model-f16.gguf

# Step 2: Quantize to 4-bit
# ./llama.cpp/llama-quantize model-f16.gguf model-q4.gguf Q4_K_M

# Step 3: Run inference with llama.cpp Python bindings
# pip install llama-cpp-python
from llama_cpp import Llama

llm = Llama(
    model_path='./model-q4.gguf',
    n_ctx=4096,         # context window
    n_threads=8,        # CPU threads
    n_gpu_layers=0      # set > 0 to offload layers to GPU
)

output = llm.create_chat_completion(
    messages=[{'role': 'user', 'content': 'What is the capital of France?'}],
    temperature=0.1
)
print(output['choices'][0]['message']['content'])

本番環境でのvLLMによるサービング

# Start vLLM server (run from command line)
# pip install vllm
# python -m vllm.entrypoints.openai.api_server \
#     --model ./merged-model \
#     --host 0.0.0.0 \
#     --port 8000 \
#     --max-model-len 4096 \
#     --tensor-parallel-size 1

# Use with OpenAI client (drop-in replacement)
from openai import OpenAI

client = OpenAI(
    base_url='http://localhost:8000/v1',
    api_key='not-needed'  # vLLM doesn't require auth by default
)

response = client.chat.completions.create(
    model='merged-model',  # model name matches the path you passed to vLLM
    messages=[{'role': 'user', 'content': 'Extract JSON from: "Alice, 30, NYC"'}]
)
print(response.choices[0].message.content)

微調整済みモデルのA/Bテスト

import random

class ModelRouter:
    def __init__(self, fine_tuned_traffic_fraction=0.1):
        self.ft_fraction = fine_tuned_traffic_fraction
        self.metrics = {'base': {'count': 0, 'quality_sum': 0}, 'fine_tuned': {'count': 0, 'quality_sum': 0}}

    def route(self, user_id: str, request: str) -> dict:
        # Deterministic routing by user_id (same user always goes to same model)
        use_fine_tuned = (hash(user_id) % 100) < (self.ft_fraction * 100)
        model_group = 'fine_tuned' if use_fine_tuned else 'base'
        
        response = call_model(request, use_fine_tuned=use_fine_tuned)
        return {'response': response, 'model_group': model_group}

    def record_quality(self, model_group: str, quality_score: float):
        self.metrics[model_group]['count'] += 1
        self.metrics[model_group]['quality_sum'] += quality_score

    def ab_test_summary(self) -> dict:
        summary = {}
        for group, data in self.metrics.items():
            avg = data['quality_sum'] / data['count'] if data['count'] > 0 else 0
            summary[group] = {'avg_quality': avg, 'n': data['count']}
        return summary

微調整済みモデルの継続的なメンテナンス

デプロイ判断マトリックス

理解度チェック

レッスンのまとめ