이벤트 기반 아키텍처로 AI API 연동하기: Kafka + Python 실전

Kafka Consumer + HolySheep AI 연동

# consumer.py
import json
import time
from datetime import datetime
from kafka import KafkaConsumer, KafkaProducer
from kafka.errors import KafkaError
from openai import OpenAI
from dotenv import dotenv_values
import logging

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

config = dotenv_values(".env")

class AIConsumer:
    def __init__(self):
        # HolySheep AI 클라이언트 초기화
        self.client = OpenAI(
            api_key=config['HOLYSHEEP_API_KEY'],
            base_url="https://api.holysheep.ai/v1",
            timeout=int(config['REQUEST_TIMEOUT'])
        )
        
        # Consumer 초기화
        self.consumer = KafkaConsumer(
            config['AI_REQUEST_TOPIC'],
            bootstrap_servers=config['KAFKA_BOOTSTRAP_SERVERS'],
            group_id='ai-api-consumers',
            auto_offset_reset='earliest',
            enable_auto_commit=False,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            max_poll_interval_ms=300000
        )
        
        # DLT(Death Letter Topic) Producer
        self.dlt_producer = KafkaProducer(
            bootstrap_servers=config['KAFKA_BOOTSTRAP_SERVERS'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        
        self.max_retries = int(config['MAX_RETRIES'])
    
    def call_ai_api(self, message: dict) -> dict:
        """HolySheep AI API 호출"""
        model = message.get('model', 'gpt-4.1')
        
        # 모델 매핑 (HolySheep AI 지원 모델)
        model_mapping = {
            'gpt-4.1': 'gpt-4.1',
            'claude-sonnet-4': 'claude-sonnet-4-20250514',
            'gemini-2.5-flash': 'gemini-2.5-flash',
            'deepseek-v3': 'deepseek-v3-0324'
        }
        
        api_model = model_mapping.get(model, 'gpt-4.1')
        
        response = self.client.chat.completions.create(
            model=api_model,
            messages=[
                {"role": "user", "content": message['prompt']}
            ],
            temperature=0.7,
            max_tokens=2000
        )
        
        return {
            "request_id": message['request_id'],
            "user_id": message['user_id'],
            "response": response.choices[0].message.content,
            "model_used": api_model,
            "tokens_used": response.usage.total_tokens,
            "completed_at": datetime.utcnow().isoformat()
        }
    
    def process_message(self, message: dict) -> bool:
        """메시지 처리 및 재시도 로직"""
        request_id = message['request_id']
        retry_count = message.get('retry_count', 0)
        
        try:
            logger.info(f"처리 중: {request_id} (재시도 {retry_count}회)")
            result = self.call_ai_api(message)
            
            # 결과 토픽으로 전송
            self.send_to_response_topic(result)
            return True
            
        except Exception as e:
            logger.error(f"처리 실패: {request_id}, 오류: {str(e)}")
            
            if retry_count < self.max_retries:
                # 재시도 큐로 전송
                message['retry_count'] = retry_count + 1
                self.retry_message(message)
                logger.info(f"재시도 예약: {request_id}")
            else:
                # DLT로 전송
                self.send_to_dlt(message, str(e))
                logger.warning(f"최대 재시도 초과, DLT 전송: {request_id}")
            
            return False
    
    def send_to_response_topic(self, result: dict):
        """응답을 Kafka 토픽으로 전송"""
        response_producer = KafkaProducer(
            bootstrap_servers=config['KAFKA_BOOTSTRAP_SERVERS'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        response_producer.send(
            config['AI_RESPONSE_TOPIC'],
            key=result['request_id'],
            value=result
        )
        response_producer.flush()
        response_producer.close()
    
    def retry_message(self, message: dict):
        """재시도 큐에 메시지 재전송 (지수 백오프)"""
        delay = 2 ** message['retry_count']
        time.sleep(delay)
        
        retry_producer = KafkaProducer(
            bootstrap_servers=config['KAFKA_BOOTSTRAP_SERVERS'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        retry_producer.send(config['AI_REQUEST_TOPIC'], value=message)
        retry_producer.flush()
        retry_producer.close()
    
    def send_to_dlt(self, message: dict, error: str):
        """Dead Letter Topic으로 전송"""
        dlt_message = {
            **message,
            "error": error,
            "failed_at": datetime.utcnow().isoformat()
        }
        self.dlt_producer.send(config['DLT_TOPIC'], value=dlt_message)
        self.dlt_producer.flush()
    
    def run(self):
        """메시지 소비 루프"""
        logger.info("AI Consumer 시작...")
        
        try:
            for message in self.consumer:
                success = self.process_message(message.value)
                
                if success:
                    self.consumer.commit()
                    
        except KeyboardInterrupt:
            logger.info("Consumer 종료 중...")
        finally:
            self.consumer.close()
            self.dlt_producer.close()


if __name__ == "__main__":
    consumer = AIConsumer()
    consumer.run()

# batch_processor.py
import json
from collections import defaultdict
from kafka import KafkaConsumer
from openai import OpenAI
from dotenv import dotenv_values
import threading
import time

config = dotenv_values(".env")

class BatchAIProcessor:
    """배치 처리를 통한 비용 최적화 프로세서"""
    
    def __init__(self, batch_size: int = 10, batch_timeout: int = 5):
        self.client = OpenAI(
            api_key=config['HOLYSHEEP_API_KEY'],
            base_url="https://api.holysheep.ai/v1"
        )
        
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self.buffer = defaultdict(list)
        self.lock = threading.Lock()
        
        # 배치 타이머 스레드
        self.running = True
        self.timer_thread = threading.Thread(target=self._batch_timer)
        self.timer_thread.daemon = True
        self.timer_thread.start()
    
    def add_request(self, message: dict):
        """요청을 버퍼에 추가"""
        model = message.get('model', 'gpt-4.1')
        
        with self.lock:
            self.buffer[model].append(message)
            
            # 배치 크기 도달 시 즉시 처리
            if len(self.buffer[model]) >= self.batch_size:
                self._process_batch(model)
    
    def _batch_timer(self):
        """타임아웃 기반 배치 처리"""
        while self.running:
            time.sleep(1)
            
            with self.lock:
                current_time = time.time()
                models_to_process = []
                
                for model, messages in self.buffer.items():
                    if messages:
                        # 배치 처리 실행
                        self._process_batch(model)
    
    def _process_batch(self, model: str):
        """배치로 AI API 호출"""
        with self.lock:
            messages = self.buffer.get(model, [])
            if not messages:
                return
            
            # 버퍼 비우기
            self.buffer[model] = []
        
        try:
            # HolySheep AI 배치 API 활용
            # 모델별 비용 최적화
            cost_per_1k_tokens = {
                'gpt-4.1': 8.0,
                'claude-sonnet-4': 15.0,
                'gemini-2.5-flash': 2.50,
                'deepseek-v3': 0.42
            }
            
            # 배치 요청 구성
            batch_requests = []
            for msg in messages:
                batch_requests.append({
                    "custom_id": msg['request_id'],
                    "method": "POST",
                    "url": "/v1/chat/completions",
                    "body": {
                        "model": model,
                        "messages": [{"role": "user", "content": msg['prompt']}]
                    }
                })
            
            # 배치 API 호출
            response = self.client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": msg['prompt']}],
                max_tokens=500
            )
            
            # 비용 계산 및 로깅
            estimated_cost = (response.usage.total_tokens / 1000) * cost_per_1k_tokens.get(model, 8.0)
            print(f"배치 처리 완료: {len(messages)}건, 예상 비용: ${estimated_cost:.4f}")
            
        except Exception as e:
            print(f"배치 처리 실패: {e}")
            # 개별 재시도 로직
    
    def shutdown(self):
        """ graceful shutdown"""
        self.running = False
        self.timer_thread.join(timeout=5)

# 문제: Kafka broker 연결 불가
kafka.errors.NoBrokersAvailable: No brokers available

해결 방법:
1. Kafka 서비스 확인
docker ps | grep kafka

2. 네트워크 연결 확인
nc -zv localhost 9092

3. Consumer 코드에 재연결 로직 추가
from kafka import KafkaConsumer
import time

def create_consumer_with_retry(max_retries=5):
    for attempt in range(max_retries):
        try:
            consumer = KafkaConsumer(
                'ai-requests',
                bootstrap_servers='localhost:9092',
                client_id=f'consumer-{attempt}'
            )
            return consumer
        except Exception as e:
            print(f"연결 시도 {attempt + 1} 실패: {e}")
            time.sleep(2 ** attempt)  # 지수 백오프
    raise ConnectionError("Kafka 연결 실패")

# 문제: HolySheep AI API 인증 실패
openai.AuthenticationError: Incorrect API key provided

해결 방법:
1. 환경변수 설정 확인
import os
print(f"API Key 설정 여부: {'HOLYSHEEP_API_KEY' in os.environ}")

2. HolySheep AI 키 검증
https://holysheep.ai/register 에서 API 키 확인

3. 올바른 base_url 사용 (중요!)
client = OpenAI(
    api_key=os.environ.get('HOLYSHEEP_API_KEY'),
    base_url="https://api.holysheep.ai/v1",  # 절대 api.openai.com 사용 금지
    timeout=60
)

4. 연결 테스트
try:
    models = client.models.list()
    print("연결 성공:", models
관련 리소스
📚 AI API 기술 문서
💰 요금제 보기
📖 개발자 문서
🚀 무료 가입
관련 문서
ko holysheep vs guijiliudong vs 302aiguoneizhongzhuan 2026 0
ko brazil developers accessing ai apis without usd pa 2026 0