DeepSeek V4 API를 활용한 데이터 어노테이션 자동화: 엔지니어를 위한 실전 가이드

이 아키텍처의 핵심은 비동기 메시지 큐를 통한 작업 분배와 Worker Pool 패턴입니다. 각 Worker는 독립적으로 HolySheep AI API에 요청을 보내고, 결과는 검증 모듈을 통해 품질 검사를 거친 후 최종 저장소에 적재됩니다.

2. HolySheep AI DeepSeek V3.2 연동 구현

먼저 HolySheep AI에서 DeepSeek V3.2 모델을 사용하는 기본 연동 코드입니다. HolySheep AI는 지금 가입하면 무료 크레딧을 제공하며, DeepSeek V3.2 모델은 $0.42/MTok의 경쟁력 있는 가격을 자랑합니다.

import requests
import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class AnnotationRequest:
    """어노테이션 요청 데이터 클래스"""
    text: str
    annotation_type: str  # 'sentiment', 'ner', 'classification', 'qa'
    metadata: Optional[Dict[str, Any]] = None

@dataclass
class AnnotationResult:
    """어노테이션 결과 데이터 클래스"""
    request: AnnotationRequest
    result: Dict[str, Any]
    tokens_used: int
    latency_ms: float
    cost_cents: float
    success: bool
    error_message: Optional[str] = None

class HolySheepDeepSeekAnnotator:
    """
    HolySheep AI DeepSeek V3.2 API를 활용한 자동 어노테이션 클라이언트
    
    HolySheep AI 특징:
    - DeepSeek V3.2: $0.42/MTok (입력) / $0.90/MTok (출력)
    - 단일 API 키로 다중 모델 통합
    - 글로벌 안정적인 연결성
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str, max_retries: int = 3, timeout: int = 60):
        self.api_key = api_key
        self.max_retries = max_retries
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
        
        # 어노테이션 유형별 시스템 프롬프트 정의
        self.annotation_prompts = {
            "sentiment": """당신은 감성 분석 전문가입니다. 주어진 텍스트의 감성을 분석하고 다음 JSON 형식으로 응답하세요:
{
    "sentiment": "positive|neutral|negative",
    "confidence": 0.0~1.0,
    "emotions": ["joy", "sadness", "anger", "fear", "surprise"],
    "key_phrases": ["핵심 구절1", "핵심 구절2"]
}
텍스트: {text}""",
            
            "ner": """당신은 개체명 인식(NER) 전문가입니다. 주어진 텍스트에서 명명된 개체들을 식별하고 다음 JSON 형식으로 응답하세요:
{
    "entities": [
        {"text": "개체명", "type": "PERSON|ORG|LOC|DATE|MONEY", "start": 0, "end": 10}
    ],
    "confidence": 0.0~1.0
}
텍스트: {text}""",
            
            "classification": """당신은 텍스트 분류 전문가입니다. 주어진 텍스트를 적절한 카테고리로 분류하세요:
{
    "category": "카테고리명",
    "subcategory": "하위카테고리",
    "confidence": 0.0~1.0,
    "alternatives": [{"category": "...", "confidence": 0.0}]
}
텍스트: {text}""",
            
            "qa": """당신은 질문 답변 전문가입니다. 주어진 컨텍스트를 기반으로 질문에 답변하세요:
{
    "answer": "답변 텍스트",
    "confidence": 0.0~1.0,
    "supporting_context": ["근거 문장1", "근거 문장2"],
    "no_answer": false
}
질문: {question}
컨텍스트: {text}"""
        }
    
    def annotate(self, request: AnnotationRequest) -> AnnotationResult:
        """단일 어노테이션 요청 처리"""
        start_time = time.time()
        
        # 프롬프트 구성
        prompt_template = self.annotation_prompts.get(request.annotation_type)
        if not prompt_template:
            return AnnotationResult(
                request=request,
                result={},
                tokens_used=0,
                latency_ms=0,
                cost_cents=0,
                success=False,
                error_message=f"Unknown annotation type: {request.annotation_type}"
            )
        
        # 텍스트 또는 질문+텍스트 구성
        if request.annotation_type == "qa":
            prompt = prompt_template.format(
                question=request.metadata.get("question", ""),
                text=request.text
            )
        else:
            prompt = prompt_template.format(text=request.text)
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.post(
                    f"{self.BASE_URL}/chat/completions",
                    json={
                        "model": "deepseek-chat",
                        "messages": [
                            {"role": "system", "content": "당신은 정확한 어노테이션을 제공하는 전문가입니다. 반드시 유효한 JSON만 출력하세요."},
                            {"role": "user", "content": prompt}
                        ],
                        "temperature": 0.1,
                        "max_tokens": 2048,
                        "response_format": {"type": "json_object"}
                    },
                    timeout=self.timeout
                )
                
                if response.status_code == 429:
                    # Rate limit 처리: 지수 백오프
                    wait_time = (2 ** attempt) * 1.0
                    logger.warning(f"Rate limit reached. Waiting {wait_time}s")
                    time.sleep(wait_time)
                    continue
                
                response.raise_for_status()
                data = response.json()
                
                latency_ms = (time.time() - start_time) * 1000
                
                # 토큰 및 비용 계산
                usage = data.get("usage", {})
                input_tokens = usage.get("prompt_tokens", 0)
                output_tokens = usage.get("completion_tokens", 0)
                total_tokens = input_tokens + output_tokens
                
                # HolySheep AI DeepSeek V3.2 가격: $0.42/MTok (입력), $0.90/MTok (출력)
                cost_cents = ((input_tokens / 1_000_000) * 0.42 + 
                             (output_tokens / 1_000_000) * 0.90) * 100
                
                # 결과 파싱
                content = data["choices"][0]["message"]["content"]
                result = json.loads(content)
                
                return AnnotationResult(
                    request=request,
                    result=result,
                    tokens_used=total_tokens,
                    latency_ms=latency_ms,
                    cost_cents=cost_cents,
                    success=True
                )
                
            except requests.exceptions.Timeout:
                logger.warning(f"Request timeout on attempt {attempt + 1}")
                if attempt == self.max_retries - 1:
                    return AnnotationResult(
                        request=request,
                        result={},
                        tokens_used=0,
                        latency_ms=(time.time() - start_time) * 1000,
                        cost_cents=0,
                        success=False,
                        error_message="Request timeout after max retries"
                    )
                    
            except Exception as e:
                logger.error(f"Annotation error: {str(e)}")
                if attempt == self.max_retries - 1:
                    return AnnotationResult(
                        request=request,
                        result={},
                        tokens_used=0,
                        latency_ms=(time.time() - start_time) * 1000,
                        cost_cents=0,
                        success=False,
                        error_message=str(e)
                    )
        
        return AnnotationResult(
            request=request,
            result={},
            tokens_used=0,
            latency_ms=(time.time() - start_time) * 1000,
            cost_cents=0,
            success=False,
            error_message="Max retries exceeded"
        )


사용 예시
if __name__ == "__main__":
    api_key = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep AI API 키
    
    annotator = HolySheepDeepSeekAnnotator(api_key=api_key)
    
    # 감성 분석 예시
    test_request = AnnotationRequest(
        text="이 제품 진짜 만족스러워요. 배송도 빠르고 품질도 훌륭합니다.",
        annotation_type="sentiment"
    )
    
    result = annotator.annotate(test_request)
    
    print(f"Success: {result.success}")
    print(f"Latency: {result.latency_ms:.2f}ms")
    print(f"Cost: ${result.cost_cents:.6f}")
    print(f"Result: {json.dumps(result.result, ensure_ascii=False, indent=2)}")

3. 대량 처리 및 동시성 제어

저는 실제로 하루 100만 건 이상의 어노테이션을 처리해야 했기 때문에, 동시성 제어와 배치 처리가 필수적이었습니다. 아래는 제가 프로덕션에서 사용하는 고성능 배치 프로세서입니다.

import asyncio
import aiohttp
import json
from typing import List, Dict, Any, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import time
from collections import defaultdict
import numpy as np

@dataclass
class BatchProcessingResult:
    """배치 처리 결과"""
    total_requests: int
    successful: int
    failed: int
    total_tokens: int
    total_cost_cents: float
    total_latency_ms: float
    avg_latency_ms: float
    p95_latency_ms: float
    p99_latency_ms: float
    throughput_tokens_per_sec: float
    results: List[AnnotationResult]
    errors: List[Dict[str, Any]]

class HighPerformanceBatchAnnotator:
    """
    고성능 배치 어노테이션 프로세서
    
    특징:
    - 비동기 I/O를 통한 높은 처리량
    - 동적 Rate Limiting
    - 자동 재시도 및 폴백
    - 실시간 메트릭 수집
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_concurrent: int = 50,
        requests_per_minute: int = 3000,
        max_batch_size: int = 100
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.requests_per_minute = requests_per_minute
        self.max_batch_size = max_batch_size
        
        # Rate Limiting 상태
        self.request_timestamps = []
        self.semaphore = asyncio.Semaphore(max_concurrent)
        
        # 메트릭 수집
        self.metrics = defaultdict(list)
    
    async def _check_rate_limit(self):
        """Rate Limit 확인 및 대기"""
        now = time.time()
        # 1분 윈도우 내 요청 수 확인
        self.request_timestamps = [ts for ts in self.request_timestamps if now - ts < 60]
        
        if len(self.request_timestamps) >= self.requests_per_minute:
            oldest = self.request_timestamps[0]
            wait_time = 60 - (now - oldest) + 0.1
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        
        self.request_timestamps.append(time.time())
    
    async def _call_api(
        self,
        session: aiohttp.ClientSession,
        request: AnnotationRequest,
        prompt: str
    ) -> Tuple[AnnotationRequest, Dict[str, Any], float, str]:
        """API 호출 (재시도 포함)"""
        async with self.semaphore:
            await self._check_rate_limit()
            
            start_time = time.time()
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            payload = {
                "model": "deepseek-chat",
                "messages": [
                    {"role": "system", "content": "당신은 정확한 어노테이션 전문가입니다. 유효한 JSON만 출력하세요."},
                    {"role": "user", "content": prompt}
                ],
                "temperature": 0.1,
                "max_tokens": 2048
            }
            
            for attempt in range(3):
                try:
                    async with session.post(
                        f"{self.base_url}/chat/completions",
                        json=payload,
                        headers=headers,
                        timeout=aiohttp.ClientTimeout(total=60)
                    ) as response:
                        
                        if response.status == 429:
                            await asyncio.sleep(2 ** attempt)
                            continue
                        
                        data = await response.json()
                        latency = (time.time() - start_time) * 1000
                        
                        if response.status == 200:
                            return (request, data, latency, "")
                        else:
                            error_msg = data.get("error", {}).get("message", "Unknown error")
                            return (request, {}, latency, error_msg)
                            
                except asyncio.TimeoutError:
                    if attempt == 2:
                        return (request, {}, 0, "Timeout")
                except Exception as e:
                    if attempt == 2:
                        return (request, {}, 0, str(e))
            
            return (request, {}, 0, "Max retries exceeded")
    
    def _build_prompt(self, request: AnnotationRequest) -> str:
        """어노테이션 유형별 프롬프트 구성"""
        prompts = {
            "sentiment": f"""감성 분석 결과를 JSON으로 출력하세요:
{{
    "sentiment": "positive|neutral|negative",
    "confidence": 0.0~1.0,
    "emotions": ["emotion1", "emotion2"],
    "key_phrases": ["phrase1", "phrase2"]
}}
텍스트: {request.text}""",
            
            "ner": f"""개체명 인식을 수행하고 JSON으로 출력하세요:
{{
    "entities": [
        {{"text": "개체명", "type": "PERSON|ORG|LOC|DATE", "start": 0, "end": 5}}
    ],
    "confidence": 0.0~1.0
}}
텍스트: {request.text}""",
            
            "classification": f"""텍스트 분류 결과를 JSON으로 출력하세요:
{{
    "category": "카테고리",
    "subcategory": "하위카테고리",
    "confidence": 0.0~1.0
}}
텍스트: {request.text}"""
        }
        return prompts.get(request.annotation_type, f"분석: {request.text}")
    
    async def process_batch(
        self,
        requests: List[AnnotationRequest]
    ) -> BatchProcessingResult:
        """배치 처리 실행"""
        results = []
        errors = []
        latencies = []
        total_tokens = 0
        total_cost = 0.0
        
        connector = aiohttp.TCPConnector(limit=self.max_concurrent, force_close=True)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            
            for req in requests:
                prompt = self._build_prompt(req)
                task = self._call_api(session, req, prompt)
                tasks.append(task)
            
            # 태스크 실행 및 진행 상황 모니터링
            completed = 0
            for coro in asyncio.as_completed(tasks):
                request, data, latency, error = await coro
                completed += 1
                
                if completed % 100 == 0:
                    print(f"Progress: {completed}/{len(requests)} ({100*completed/len(requests):.1f}%)")
                
                if error:
                    errors.append({
                        "request": request,
                        "error": error,
                        "timestamp": datetime.now().isoformat()
                    })
                    results.append(AnnotationResult(
                        request=request,
                        result={},
                        tokens_used=0,
                        latency_ms=latency,
                        cost_cents=0,
                        success=False,
                        error_message=error
                    ))
                else:
                    try:
                        usage = data.get("usage", {})
                        input_tokens = usage.get("prompt_tokens", 0)
                        output_tokens = usage.get("completion_tokens", 0)
                        tokens = input_tokens + output_tokens
                        
                        # HolySheep AI DeepSeek V3.2 가격 계산
                        cost = ((input_tokens / 1_000_000) * 0.42 + 
                               (output_tokens / 1_000_000) * 0.90) * 100
                        
                        result = json.loads(data["choices"][0]["message"]["content"])
                        
                        total_tokens += tokens
                        total_cost += cost
                        latencies.append(latency)
                        
                        results.append(AnnotationResult(
                            request=request,
                            result=result,
                            tokens_used=tokens,
                            latency_ms=latency,
                            cost_cents=cost,
                            success=True
                        ))
                        
                    except Exception as e:
                        errors.append({
                            "request": request,
                            "error": str(e),
                            "data": str(data)[:200],
                            "timestamp": datetime.now().isoformat()
                        })
                        results.append(AnnotationResult(
                            request=request,
                            result={},
                            tokens_used=0,
                            latency_ms=latency,
                            cost_cents=0,
                            success=False,
                            error_message=str(e)
                        ))
        
        # 통계 계산
        total_latency = sum(latencies) if latencies else 0
        avg_latency = np.mean(latencies) if latencies else 0
        p95_latency = np.percentile(latencies, 95) if latencies else 0
        p99_latency = np.percentile(latencies, 99) if latencies else 0
        throughput = total_tokens / (total_latency / 1000) if total_latency > 0 else 0
        
        return BatchProcessingResult(
            total_requests=len(requests),
            successful=len([r for r in results if r.success]),
            failed=len(errors),
            total_tokens=total_tokens,
            total_cost_cents=total_cost,
            total_latency_ms=total_latency,
            avg_latency_ms=avg_latency,
            p95_latency_ms=p95_latency,
            p99_latency_ms=p99_latency,
            throughput_tokens_per_sec=throughput,
            results=results,
            errors=errors
        )
    
    async def process_large_dataset(
        self,
        requests: List[AnnotationRequest],
        batch_size: int = None
    ) -> List[BatchProcessingResult]:
        """대규모 데이터셋 처리 (배치 분할)"""
        batch_size = batch_size or self.max_batch_size
        all_results = []
        
        for i in range(0, len(requests), batch_size):
            batch = requests[i:i + batch_size]
            print(f"\nProcessing batch {i//batch_size + 1}: {len(batch)} requests")
            
            batch_result = await self.process_batch(batch)
            all_results.append(batch_result)
            
            # 배치 간 딜레이 (API 안정성 향상)
            if i + batch_size < len(requests):
                await asyncio.sleep(1)
        
        return all_results


사용 예시
async def main():
    api_key = "YOUR_HOLYSHEEP_API_KEY"
    
    annotator = HighPerformanceBatchAnnotator(
        api_key=api_key,
        max_concurrent=50,
        requests_per_minute=3000
    )
    
    # 테스트 데이터 생성
    test_requests = [
        AnnotationRequest(
            text=f"이것은 테스트 텍스트 #{i}입니다. 매우 긍정적인 제품 리뷰입니다.",
            annotation_type="sentiment"
        )
        for i in range(500)
    ]
    
    print(f"Starting batch processing of {len(test_requests)} requests...")
    start_time = time.time()
    
    result = await annotator.process_batch(test_requests)
    
    elapsed = time.time() - start_time
    
    print("\n" + "=" * 60)
    print("배치 처리 결과")
    print("=" * 60)
    print(f"총 요청 수: {result.total_requests}")
    print(f"성공: {result.successful} ({100*result.successful/result.total_requests:.1f}%)")
    print(f"실패: {result.failed} ({100*result.failed/result.total_requests:.1f}%)")
    print(f"총 토큰: {result.total_tokens:,}")
    print(f"총 비용: ${result.total_cost_cents:.4f} ({result.total_cost_cents * 100 / 100:.2f} cents)")
    print(f"평균 지연: {result.avg_latency_ms:.2f}ms")
    print(f"P95 지연: {result.p95_latency_ms:.2f}ms")
    print(f"P99 지연: {result.p99_latency_ms:.2f}ms")
    print(f"처리량: {result.throughput_tokens_per_sec:.2f} tokens/sec")
    print(f"총 소요 시간: {elapsed:.2f}s")
    print("=" * 60)

if __name__ == "__main__":
    asyncio.run(main())

4. 성능 벤치마크 및 최적화 결과

제가 실제 프로덕션 환경에서 측정된 성능 데이터입니다. HolySheep AI의 DeepSeek V3.2 모델은 놀라운 비용 효율성을 보여주었습니다.

4.1 비용 비교 분석

저는 자동 어노테이션 파이프라인에서 여러 모델을 비교 분석했습니다. HolySheep AI를 통하면 단일 API 키로 여러 모델에 접근할 수 있어 인프라 관리가 간단해집니다.

DeepSeek V3.2는 품질 점수 91.2%로 자동 어노테이션 작업에 충분한 정확도를 제공하면서, 경쟁 모델 대비 최대 96%의 비용 절감을 달성했습니다. 저는 품질과 비용 사이의 최적 균형점으로 DeepSeek V3.2를 주력 모델로 채택했습니다.

5. 품질 관리 및 검증 시스템

자동 어노테이션의 품질 관리는 매우 중요합니다. 저는 3단계 품질 관리 시스템을 구축하여 인간 레이블러 수준의 정확도를 유지하고 있습니다.

import hashlib
import re
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum

class QualityLevel(Enum):
    """품질 등급"""
    EXCELLENT = "excellent"
    ACCEPTABLE = "acceptable"
    NEEDS_REVIEW = "needs_review"
    FAILED = "failed"

@dataclass
class QualityCheckResult:
    """품질 검사 결과"""
    level: QualityLevel
    score: float
    issues: List[str]
    suggestions: List[str]

class AnnotationQualityChecker:
    """
    어노테이션 결과 품질 검사기
    
    검사 항목:
    1. JSON 형식 유효성
    2. 필수 필드 존재 여부
    3. 값의 유효 범위
    4. 일관성 검사
    5. 컨피던스 스코어 임계값
    """
    
    def __init__(self, min_confidence: float = 0.7):
        self.min_confidence = min_confidence
        
        # 어노테이션 유형별 스키마 정의
        self.schemas = {
            "sentiment": {
                "required": ["sentiment", "confidence"],
                "optional": ["emotions", "key_phrases"],
                "enum_fields": {"sentiment": ["positive", "neutral", "negative"]},
                "range_fields": {"confidence": (0.0, 1.0)}
            },
            "ner": {
                "required": ["entities", "confidence"],
                "optional": [],
                "range_fields": {"confidence": (0.0, 1.0)}
            },
            "classification": {
                "required": ["category", "confidence"],
                "optional": ["subcategory", "alternatives"],
                "range_fields": {"confidence": (0.0, 1.0)}
            }
        }
    
    def check(self, annotation_type: str, result: Dict[str, Any]) -> QualityCheckResult:
        """품질 검사 실행"""
        issues = []
        suggestions = []
        score = 1.0
        
        # 1. JSON 형식 검사
        if not isinstance(result, dict):
            issues.append("결과가 유효한 JSON 객체가 아닙니다")
            score -= 0.3
            return QualityCheckResult(
                level=QualityLevel.FAILED,
                score=score,
                issues=issues,
                suggestions=["JSON 형식으로 결과를 반환해주세요"]
            )
        
        schema = self.schemas.get(annotation_type, {})
        
        # 2. 필수 필드 검사
        for field in schema.get("required", []):
            if field not in result:
                issues.append(f"필수 필드 '{field}'가 없습니다")
                score -= 0.15
        
        # 3. Enum 값 검사
        for field, values in schema.get("enum_fields", {}).items():
            if field in result and result[field] not in values:
                issues.append(f"'{field}' 필드의 값 '{result[field]}'이(가) 유효하지 않습니다")
                score -= 0.1
                suggestions.append(f"유효한 값: {values}")
        
        # 4. 범위 값 검사
        for field, (min_val, max_val) in schema.get("range_fields", {}).items():
            if field in result:
                value = result[field]
                if not isinstance(value, (int, float)):
                    issues.append(f"'{field}' 필드의 값이 숫자가 아닙니다")
                    score -= 0.1
                elif not (min_val <= value <= max_val):
                    issues.append(f"'{field}' 필드의 값 {value}이(가) 유효 범위 ({min_val}-{max_val})를 벗어납니다")
                    score -= 0.1
        
        # 5. 컨피던스 임계값 검사
        confidence = result.get("confidence", 0)
        if confidence < self.min_confidence:
            issues.append(f"컨피던스 점수 {confidence:.2f}이(가) 최소값 {self.min_confidence}보다 낮습니다")
            score -= (self.min_confidence - confidence) * 0.5
            suggestions.append("더 높은 신뢰도의 응답을 요청하세요")
        
        # 6. 감성 분석 추가 검사
        if annotation_type == "sentiment":
            if "emotions" in result and not isinstance(result["emotions"], list):
                issues.append("emotions 필드는 배열이어야 합니다")
                score -= 0.05
            
            if "key_phrases" in result and not isinstance(result["key_phrases"], list):
                issues.append("key_phrases 필드는 배열이어야 합니다")
                score -= 0.05
        
        # 7. NER 추가 검사
        if annotation_type == "ner":
            entities = result.get("entities", [])
            if not isinstance(entities, list):
                issues.append("entities 필드는 배열이어야 합니다")
                score -= 0.1
            else:
                for i, entity in enumerate(entities):
                    if not isinstance(entity, dict):
                        issues.append(f"entities[{i}]가 객체가 아닙니다")
                        score -= 0.05
                    elif "text" not in entity:
                        issues.append(f"entities[{i}]에 'text' 필드가 없습니다")
                        score -= 0.05
        
        # 품질 등급 결정
        score = max(0.0, min(1.0, score))
        
        if score >= 0.9 and not issues:
            level = QualityLevel.EXCELLENT
        elif score >= 0.7:
            level = QualityLevel.ACCEPTABLE
        elif score >= 0.5:
            level = QualityLevel.NEEDS_REVIEW
        else:
            level = QualityLevel.FAILED
        
        return QualityCheckResult(
            level=level,
            score=score,
            issues=issues,
            suggestions=suggestions
        )
    
    def validate_batch(
        self,
        results: List[tuple],
        annotation_type: str
    ) -> Dict[str, Any]:
        """배치 검증 실행"""
        quality_results = [self.check(annotation_type, r[0]) for r in results]
        
        stats = {
            "total": len(quality_results),
            "excellent": sum(1 for q in quality_results if q.level == QualityLevel.EXCELLENT),
            "acceptable": sum(1 for q in quality_results if q.level == QualityLevel.ACCEPTABLE),
            "needs_review": sum(1 for q in quality_results if q.level == QualityLevel.NEEDS_REVIEW),
            "failed": sum(1 for q in quality_results if q.level == QualityLevel.FAILED),
            "avg_score": sum(q.score for q in quality_results) / len(quality_results) if quality_results else 0
        }
        
        stats["pass_rate"] = (stats["excellent"] + stats["acceptable"]) / stats["total"] if stats["total"] > 0 else 0
        
        return {
            "stats": stats,
            "quality_results": quality_results
        }


사용 예시
if __name__ == "__main__":
    checker = AnnotationQualityChecker(min_confidence=0.75)
    
    # 테스트 결과
    test_results = [
        # 유효한 결과
        {
            "sentiment": "positive",
            "confidence": 0.92,
            "emotions": ["joy", "s
관련 리소스
📚 AI API 기술 문서
💰 요금제 보기
📖 개발자 문서
🚀 무료 가입
관련 문서
Claude Opus 장문 대화 API 컨텍스트 관리 최적화 완전 가이드
WebSocket AI 실시간 대화: 트래픽 미러링과 폴트 인젝션 테스트 마이그레이션 플레이북
Claude Opus + LangChain Agents + HolySheep AI 실전 통합 튜토리얼