Kafka + AI 실시간 파이프라인 - 스트리밍 데이터에 LLM 적용하기

from confluent_kafka import Consumer, Producer
import json
import requests
 
# Kafka Consumer 설정
consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'llm-processor',
    'auto.offset.reset': 'latest',
    'enable.auto.commit': False,
    'max.poll.interval.ms': 300000,
})
 
producer = Producer({'bootstrap.servers': 'kafka:9092'})
 
consumer.subscribe(['customer_reviews'])
 
def classify_with_llm(text: str) -> dict:
    """LLM으로 리뷰 분류"""
    response = requests.post(
        "http://vllm-server:8000/v1/chat/completions",
        json={
            "model": "meta-llama/Llama-3.1-8B-Instruct",
            "messages": [{"role": "user", "content": f'리뷰를 분석하세요. JSON 반환: {{"sentiment": "긍정/부정/중립", "category": "카테고리", "urgent": true/false}}\n\n리뷰: {text}'}],
            "temperature": 0.0,
            "max_tokens": 100
        },
        timeout=10
    )
    return json.loads(response.json()["choices"][0]["message"]["content"])
 
# 메시지 처리 루프
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        continue
    
    review = json.loads(msg.value().decode('utf-8'))
    
    try:
        result = classify_with_llm(review['text'])
        
        # 결과를 출력 토픽에 전송
        output = {**review, **result, "processed_at": datetime.now().isoformat()}
        producer.produce(
            'review_analysis',
            key=msg.key(),
            value=json.dumps(output).encode('utf-8')
        )
        
        # 긴급 리뷰는 알림
        if result.get("urgent"):
            producer.produce('urgent_alerts', value=json.dumps(output).encode('utf-8'))
        
        consumer.commit(msg)
        
    except Exception as e:
        # 실패 시 DLQ(Dead Letter Queue)로 전송
        producer.produce('review_dlq', value=msg.value())
        consumer.commit(msg)
 
    producer.flush()

import asyncio
import aiohttp
from collections import deque
 
class MicroBatchProcessor:
    def __init__(self, batch_size=10, batch_timeout=5.0):
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self.buffer = deque()
    
    async def process_stream(self, consumer, producer):
        """마이크로 배치 기반 스트림 처리"""
        while True:
            msg = consumer.poll(0.1)
            if msg and not msg.error():
                self.buffer.append(msg)
            
            # 배치 크기 또는 타임아웃 도달 시 처리
            if len(self.buffer) >= self.batch_size or \
               (self.buffer and self._timeout_reached()):
                batch = [self.buffer.popleft() for _ in range(min(len(self.buffer), self.batch_size))]
                
                # 배치 LLM 호출
                results = await self.batch_classify(batch)
                
                for msg, result in zip(batch, results):
                    producer.produce('output_topic', value=json.dumps(result).encode())
                    consumer.commit(msg)
                
                producer.flush()
    
    async def batch_classify(self, messages):
        """배치 LLM 호출"""
        texts = [json.loads(m.value())['text'] for m in messages]
        # vLLM 배치 API 호출
        async with aiohttp.ClientSession() as session:
            tasks = [self.call_llm(session, t) for t in texts]
            return await asyncio.gather(*tasks)

# 서버 로그 실시간 분석
def analyze_log(log_entry: str) -> dict:
    """로그 엔트리를 LLM으로 분석"""
    response = requests.post(
        "http://ollama:11434/api/generate",
        json={
            "model": "llama3.1:8b",
            "prompt": f"""다음 서버 로그를 분석하세요.
 
로그: {log_entry}
 
JSON 반환:
{{"severity": "critical/warning/info", "category": "memory/disk/network/application/security", "requires_action": true/false, "summary": "한줄 요약"}}""",
            "stream": False,
            "options": {"temperature": 0.0}
        }
    )
    return json.loads(response.json()["response"])