我为生产 AI 构建了两个高性能 Python 库：LLM 日志分析和向量相似搜索

Source: Dev.to

我的项目功能

llmlog_engine：面向LLM日志的列式分析

一种专用的嵌入式数据库，用于分析以 JSONL 存储的 LLM 应用日志。

核心功能

• 快速将 JSONL 导入为列式存储格式
• 在数值列和字符串列上进行高效过滤
• 分组聚合（COUNT、SUM、AVG、MIN、MAX）
• 对低基数字符串（模型名称、路由）进行字典编码
• SIMD 友好的内存布局以提升性能
• 与 pandas DataFrame 集成

性能

• 比纯 Python 快 6.8 倍，在 10 万行数据上
  • 基准：按模型+延迟过滤，按路由分组，计算 6 项指标
  • 纯 Python：0.82 s
  • C++ 引擎：0.12 s

mini_faiss：轻量级向量相似度搜索

一个专注且高性能的库，用于在稠密嵌入中进行相似度搜索。

核心功能

• SIMD 加速的距离计算（L2 与内积）
• NumPy 友好的 API，具有清晰的类型签名
• 约 1500 行可读的 C++ 代码
• 支持欧氏距离和余弦相似度
• 基于堆的 top‑k 选择

性能

• 比纯 NumPy 快约 7 倍，在典型工作负载下
  • 基准：10 万向量，768 维
  • 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||²

• 预先计算数据库向量的范数以提升效率
• 可向量化的循环支持 SIMD 自动向量化

Top‑k 选择

• 基于堆的算法：每次查询 O(N log k)
• k << N 时效率更高
• 针对最小（L2）和最大（内积）分别实现

行主序存储

data = [v_0[0], v_0[1], ..., v_0[d-1],
        v_1[0], v_1[1], ..., v_1[d-1],
        ...]

缓存友好，适合批量距离计算。

安装

两个库均使用标准的 Python 打包方式：

# llmlog_engine
git clone https://github.com/yuuichieguchi/llmlog_engine.git
cd llmlog_engine
pip install -e .

# mini_faiss
git clone https://github.com/yuuichieguchi/mini_faiss.git
cd mini_faiss
pip install .

要求

• Python 3.8+
• C++17 编译器（GCC、Clang、MSVC）
• CMake 3.15+
• pybind11（通过 pip 安装）

使用场景

llmlog_engine

• 监控生产环境中 LLM 应用的健康状态
• 按模型和端点分析延