EDA + 서버리스로 구축하는 AI 파이프라인: AWS Lambda, EventBridge, RAG부터 실시간 이상 감지까지

외부 이벤트
    └─► 이벤트 브로커 (Kafka / EventBridge)
            ├─► Lambda: 전처리
            │       └─► 이벤트 발행 (fan-out)
            │               ├─► Lambda: AI 추론 A
            │               └─► Lambda: AI 추론 B
            │                       └─► 이벤트 발행
            │                               └─► Lambda: 후처리 / 액션
            └─► DLQ: 실패 이벤트 보관 및 재처리

# Lambda: 문서 파싱 & 청킹 (전처리 단계)
import boto3
import json
 
def handler(event, context):
    s3 = boto3.client('s3')
    eventbridge = boto3.client('events')
 
    record = event['Records'][0]['s3']
    bucket = record['bucket']['name']
    key = record['object']['key']
 
    try:
        obj = s3.get_object(Bucket=bucket, Key=key)
        text = obj['Body'].read().decode('utf-8')
        chunks = chunk_text(text, chunk_size=512, overlap=64)
 
        eventbridge.put_events(Entries=[{
            'Source': 'rag.preprocessing',
            'DetailType': 'DocumentChunked',
            'EventBusName': 'rag-pipeline-bus',  # 프로덕션에서는 커스텀 버스 지정 필수
            'Detail': json.dumps({
                'document_id': key,
                'chunks': chunks,
                'total_chunks': len(chunks)
            })
        }])
 
        return {'statusCode': 200, 'chunksCount': len(chunks)}
 
    except Exception as e:
        print(f"처리 실패: {e}")
        raise  # 예외를 다시 던져 SQS/EventBridge 재시도 트리거
 
def chunk_text(text: str, chunk_size: int, overlap: int) -> list[str]:
    # 실제 운영에서는 tiktoken 같은 토크나이저 기반 청킹을 권장합니다
    # 단어 기반 분할은 한국어 등 CJK 문자에서 토큰 수 예측이 어렵습니다
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)
    return chunks

# Lambda: 임베딩 생성 & 벡터 DB 저장
import boto3
import json
 
bedrock = boto3.client('bedrock-runtime')
 
def handler(event, context):
    detail = json.loads(event['detail'])
    document_id = detail['document_id']
    chunks = detail['chunks']
 
    try:
        embeddings = []
        for chunk in chunks:
            response = bedrock.invoke_model(
                modelId='amazon.titan-embed-text-v2:0',
                body=json.dumps({'inputText': chunk})
            )
            embedding = json.loads(response['body'].read())['embedding']
            embeddings.append({
                'chunk_text': chunk,
                'embedding': embedding,
                'document_id': document_id
            })
 
        # OpenSearch 배치 인덱싱 — opensearchpy bulk API로 구현
        batch_index_vectors(embeddings)
        return {'indexed': len(embeddings)}
 
    except Exception as e:
        print(f"임베딩 실패: {e}")
        raise
 
def batch_index_vectors(embeddings: list) -> None:
    # opensearchpy 클라이언트로 벡터 인덱싱 — 구현 생략
    pass

# Lambda: Kafka 트리거 — 특징 추출 및 추론
import boto3
import json
import base64
 
sagemaker = boto3.client('sagemaker-runtime')
eventbridge = boto3.client('events')
 
def handler(event, context):
    results = []
 
    for record in event['records']['transaction-topic-0']:
        try:
            payload = json.loads(base64.b64decode(record['value']))
            features = extract_features(payload)
 
            # 레코드 수가 많다면 배치 예측 엔드포인트나 asyncio 병렬 처리 검토를 권장합니다
            # 동기 루프는 레코드 수에 비례해 처리 시간이 선형 증가합니다
            response = sagemaker.invoke_endpoint(
                EndpointName='fraud-detection-serverless',
                ContentType='application/json',
                Body=json.dumps({'instances': [features]})
            )
            prediction = json.loads(response['Body'].read())
            fraud_score = prediction['predictions'][0]['score']
 
            if fraud_score > 0.85:
                eventbridge.put_events(Entries=[{
                    'Source': 'fraud.detection',
                    'DetailType': 'HighRiskTransaction',
                    'EventBusName': 'fraud-detection-bus',
                    'Detail': json.dumps({
                        'transaction_id': payload['id'],
                        'fraud_score': fraud_score,
                        'amount': payload['amount']
                    })
                }])
 
            results.append({'transaction_id': payload['id'], 'score': fraud_score})
 
        except Exception as e:
            print(f"레코드 처리 실패: {e}")
            continue  # 개별 레코드 실패가 전체 배치를 막지 않도록
 
    return results
 
def extract_features(transaction: dict) -> list[float]:
    return [
        transaction['amount'],
        transaction['merchant_category_code'],
        transaction['hour_of_day'],
        transaction['is_international'],
        transaction['velocity_1h'],
    ]

{
  "Comment": "병렬 에이전트 오케스트레이션 — Amazon States Language",
  "StartAt": "RouteRequest",
  "States": {
    "RouteRequest": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:routing-agent",
      "Next": "ParallelAgents"
    },
    "ParallelAgents": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "SearchAgent",
          "States": {
            "SearchAgent": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789012:function:search-agent",
              "End": true
            }
          }
        },
        {
          "StartAt": "CalculationAgent",
          "States": {
            "CalculationAgent": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789012:function:calc-agent",
              "End": true
            }
          }
        }
      ],
      "Next": "SynthesizeResults"
    },
    "SynthesizeResults": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:synthesis-agent",
      "End": true
    }
  }
}