프로덕션 AI를 위한 고성능 Python 라이브러리 두 개를 만들었습니다: LLM 로그 애널리틱스 및 벡터 유사도 검색

Source: Dev.to

내 프로젝트가 하는 일

llmlog_engine: LLM 로그를 위한 컬럼형 분석

JSONL 형태로 저장된 LLM 애플리케이션 로그를 분석하기 위한 특화된 임베디드 데이터베이스.

핵심 기능

• JSONL을 컬럼형 저장 형식으로 빠르게 ingest
• 숫자 및 문자열 컬럼에 대한 효율적인 필터링
• Group‑by 집계 (COUNT, SUM, AVG, MIN, MAX)
• 낮은 카디널리티 문자열(모델명, 라우트)에 대한 사전 인코딩
• 성능을 위한 SIMD 친화적 메모리 레이아웃
• pandas DataFrame 통합

성능

• 6.8× faster than pure Python on 100 k rows
  • Benchmark: filter by model + latency, group by route, compute 6 metrics
  • Pure Python: 0.82 s
  • C++ 엔진: 0.12 s

mini_faiss: 경량 벡터 유사도 검색

밀집 임베딩에 대한 유사도 검색을 위한 집중적이고 고성능 라이브러리.

핵심 기능

• SIMD 가속 거리 계산 (L2 및 내적)
• NumPy 친화적 API와 깔끔한 타입 시그니처
• ~1500줄의 가독성 좋은 C++ 코드
• 유클리드와 코사인 유사도 모두 지원
• 힙 기반 top‑k 선택

성능

• ≈ 7× faster than pure NumPy on typical workloads
  • Benchmark: 100 k vectors, 768 dimensions
  • mini_faiss: 0.067 s
  • NumPy: 0.48 s

아키텍처 철학

두 라이브러리 모두 동일한 설계 패턴을 따릅니다:

• C++17 기반 핵심 로직 – 최신 C++를 이용한 성능 핵심 연산
• pybind11을 통한 Python 바인딩 – NumPy와의 제로 복사 데이터 전송
• 최소 의존성 – 무거운 프레임워크나 복잡한 빌드 체인 없음
• 컬럼형 / SIMD 친화적 레이아웃 – CPU 캐시 최적화 데이터 구조
• 타입 안전성 – Python/C++ 경계에서 엄격한 검증

이 접근 방식은 Python 개발자 경험을 유지하면서 네이티브에 가까운 성능을 제공합니다.

구문 예시

llmlog_engine

로그 로드 및 분석

from llmlog_engine import LogStore

# Load JSONL logs
store = LogStore.from_jsonl("production_logs.jsonl")

# Analyze slow responses by model
slow_by_model = (
    store.query()
    .filter(min_latency_ms=500)
    .aggregate(
        by=["model"],
        metrics={
            "count": "count",
            "avg_latency": "avg(latency_ms)",
            "max_latency": "max(latency_ms)",
        },
    )
)

print(slow_by_model)  # Returns pandas DataFrame

오류 분석

# Analyze error rates by model and route
errors = (
    store.query()
    .filter(status="error")
    .aggregate(
        by=["model", "route"],
        metrics={"count": "count"},
    )
)

복합 필터

# Filter by multiple conditions (AND logic)
result = (
    store.query()
    .filter(
        model="gpt-4.1",
        min_latency_ms=1000,
        route="chat",
    )
    .aggregate(
        by=["model"],
        metrics={"avg_tokens": "avg(tokens_output)"},
    )
)

기대되는 JSONL 형식

{"ts": "2024-01-01T12:00:00Z", "model": "gpt-4.1", "latency_ms": 423, "tokens_input": 100, "tokens_output": 921, "route": "chat", "status": "ok"}
{"ts": "2024-01-01T12:00:15Z", "model": "gpt-4.1-mini", "latency_ms": 152, "tokens_input": 50, "tokens_output": 214, "route": "rag", "status": "ok"}

mini_faiss

기본 유사도 검색

import numpy as np
from mini_faiss import IndexFlatL2

# Create index for 768‑dimensional vectors
d = 768
index = IndexFlatL2(d)

# Add vectors to index
xb = np.random.randn(10000, d).astype("float32")
index.add(xb)

# Search for nearest neighbors
xq = np.random.randn(5, d).astype("float32")
distances, indices = index.search(xq, k=10)

print(distances.shape)  # (5, 10) - 5 queries, 10 neighbors each
print(indices.shape)    # (5, 10)

코사인 유사도 검색

from mini_faiss import IndexFlatIP

# Create inner product index
index = IndexFlatIP(d=768)

# Normalize vectors for cosine similarity
xb = np.random.randn(10000, 768).astype("float32")
xb /= np.linalg.norm(xb, axis=1, keepdims=True)

index.add(xb)

# Assume xq_normalized is similarly normalized
distances, indices = index.search(xq_normalized, k=10)
# Higher distances = more similar

구현 하이라이트

llmlog_engine

사전 인코딩을 활용한 컬럼형 저장

• 문자열 컬럼(model, route, status)은 int32 ID로 매핑
• 숫자 컬럼은 연속 배열로 저장
• 필터링은 압축된 정수 표현을 기반으로 수행

쿼리 실행

• 필터 조건(AND 논리)으로 부울 마스크 생성
• 지정된 컬럼별로 일치하는 행을 그룹화
• 필터링된 행에만 집계 계산
• pandas DataFrame 반환

내부 표현 예시

Column: model       [0, 1, 0, 2, 0, ...] (int32 IDs)
Column: latency_ms  [423, 1203, 512, ...] (int32)
Dictionary: model   {0: "gpt-4.1-mini", 1: "gpt-4.1", 2: "gpt-4-turbo"}

mini_faiss

거리 계산 (L2)

||q - db||² = ||q||² - 2·q·db + ||db||²

• 효율성을 위해 데이터