LLM API 응답 가공

import json
from datetime import datetime
import uuid

# 1. 미들웨어 등에서 추출되어 보관 중인 Request Context
current_tenant_id = "tenant-marketing-01"
current_model = "gpt-4-turbo"

# 2. 외부 LLM API(OpenAI 등)로부터 받은 가상의 원시 응답 JSON
llm_raw_response = {
    "id": "chatcmpl-123",
    "choices": [{"message": {"content": "Hello! I am an AI."}}],
    "usage": {
        "prompt_tokens": 10,
        "completion_tokens": 7,
        "total_tokens": 17
    }
}

# 3. 토큰 추출 및 테넌트 맵핑을 통한 Chargeback 스키마 조립
def generate_chargeback_event(tenant_id: str, model_id: str, raw_response: dict) -> str:
    # 안전한 추출 (KeyError 방지)
    usage = raw_response.get("usage", {})
    
    chargeback_data = {
        "event_id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "tenant_id": tenant_id,
        "model_id": model_id,
        "metrics": {
            "input_tokens": usage.get("prompt_tokens", 0),
            "output_tokens": usage.get("completion_tokens", 0),
            "total_tokens": usage.get("total_tokens", 0)
        }
    }
    
    # 4. 다운스트림 시스템을 위한 JSON 포맷으로 직렬화
    return json.dumps(chargeback_data)

# 실행 결과
chargeback_json = generate_chargeback_event(current_tenant_id, current_model, llm_raw_response)
print(chargeback_json)

import uuid
from datetime import datetime, timezone
from pydantic import BaseModel, Field
from fastapi import FastAPI, Request, BackgroundTasks

app = FastAPI()

# 1. 중앙 과금 시스템(Billing DB/Kafka)과 약속된 엄격한 Pydantic 스키마 정의
class TokenMetrics(BaseModel):
    input_tokens: int = Field(..., ge=0, description="프롬프트 토큰 수")
    output_tokens: int = Field(..., ge=0, description="생성된 토큰 수")
    total_tokens: int = Field(..., ge=0, description="전체 토큰 수")

class ChargebackEvent(BaseModel):
    event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    tenant_id: str = Field(..., description="과금 대상 부서 또는 고객사 ID")
    model_id: str = Field(..., description="사용된 LLM 모델명 (단가 맵핑용)")
    metrics: TokenMetrics

# 2. 비동기 환경에서 과금 이벤트를 외부 시스템(예: Kafka, DB)으로 전송하는 가상의 워커
async def publish_chargeback_event(event_json: str):
    # 실제 환경에서는 aiokafka, boto3(Kinesis) 등을 사용하여 전송
    print(f"[Billing Pipeline] 전송 완료: {event_json}")

# 3. 메인 엔드포인트
@app.post("/v1/completions")
async def generate_text(request: Request, bg_tasks: BackgroundTasks):
    # A. 미들웨어가 request.state에 주입해둔 테넌트 식별자 확보
    # (실무에서는 Header나 JWT 파싱 미들웨어를 거친 상태)
    tenant_id = getattr(request.state, "tenant_id", "default_untracked_tenant")
    target_model = "claude-3-opus-20240229"
    
    # B. LLM 호출 및 응답 수신 모사 (실제는 Anthropic/OpenAI 비동기 클라이언트 호출)
    mock_llm_response = {
        "content": "실무 레벨의 과금 파이프라인 데이터입니다.",
        "usage": {"input_tokens": 105, "output_tokens": 45} # Anthropic 스타일
    }
    
    # C. 공급자 독립적인 파싱 및 스키마 인스턴스화
    try:
        # Pydantic 모델을 통해 누락된 필드가 없는지, 타입이 맞는지 런타임 검증
        chargeback_event = ChargebackEvent(
            tenant_id=tenant_id,
            model_id=target_model,
            metrics=TokenMetrics(
                input_tokens=mock_llm_response["usage"].get("input_tokens", 0),
                output_tokens=mock_llm_response["usage"].get("output_tokens", 0),
                total_tokens=(
                    mock_llm_response["usage"].get("input_tokens", 0) + 
                    mock_llm_response["usage"].get("output_tokens", 0)
                )
            )
        )
        
        # Pydantic 객체를 최종 JSON 문자열로 직렬화
        validated_json_payload = chargeback_event.model_dump_json()
        
        # D. 사용자 응답 속도를 저하시키지 않도록 BackgroundTasks에 넘겨 비동기 발행
        bg_tasks.add_task(publish_chargeback_event, validated_json_payload)
        
    except Exception as e:
        # 과금 객체 생성 실패 시(데이터 누락 등) 알람 발생 로직 (Sentry 등)
        print(f"[Error] Chargeback 생성 실패: {str(e)}")

    # E. 클라이언트에게는 텍스트만 깔끔하게 반환
    return {"message": mock_llm_response["content"]}