Логирование запросов к AI API — это практика записи информации о каждом обращении к языковой модели: промпт, модель, время ответа, количество токенов, стоимость. Без логирования невозможно отлаживать промпты, контролировать расходы и находить узкие места. В этой статье покажем, как построить систему логирования для AI-приложений.
Что нужно логировать
Каждый запрос к AI API должен содержать следующие метрики:
- Timestamp — время запроса
- Model — используемая модель
- Prompt tokens — количество токенов во входных данных
- Completion tokens — количество токенов в ответе
- Total tokens — общее количество токенов
- Latency — время ответа (ms)
- Cost — стоимость запроса
- Status — успешен ли запрос
- User/session ID — кто сделал запрос
Обёртка для логирования на Python
import time
import json
import logging
from openai import OpenAI
from dataclasses import dataclass, asdict
from datetime import datetime
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ai_requests")
client = OpenAI(
base_url="https://api.modelswitch.ru/v1",
api_key="msk_ваш_ключ"
)
@dataclass
class AIRequestLog:
timestamp: str
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
latency_ms: int
status: str
user_id: str = ""
error: str = ""
def logged_completion(messages: list, model: str = "gpt-4o", user_id: str = "") -> str:
"""Запрос к AI API с автоматическим логированием."""
start_time = time.time()
try:
response = client.chat.completions.create(model=model, messages=messages)
latency_ms = int((time.time() - start_time) * 1000)
log_entry = AIRequestLog(
timestamp=datetime.utcnow().isoformat(),
model=model,
prompt_tokens=response.usage.prompt_tokens,
completion_tokens=response.usage.completion_tokens,
total_tokens=response.usage.total_tokens,
latency_ms=latency_ms,
status="success",
user_id=user_id
)
logger.info(json.dumps(asdict(log_entry)))
return response.choices[0].message.content
except Exception as e:
latency_ms = int((time.time() - start_time) * 1000)
log_entry = AIRequestLog(
timestamp=datetime.utcnow().isoformat(),
model=model,
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
latency_ms=latency_ms,
status="error",
user_id=user_id,
error=str(e)
)
logger.error(json.dumps(asdict(log_entry)))
raise
# Использование
result = logged_completion(
[{"role": "user", "content": "Что такое Kubernetes?"}],
model="gpt-4o",
user_id="user_123"
)Обёртка на TypeScript
import OpenAI from "openai";
const client = new OpenAI({
baseURL: "https://api.modelswitch.ru/v1",
apiKey: "msk_ваш_ключ",
});
interface AIRequestLog {
timestamp: string;
model: string;
promptTokens: number;
completionTokens: number;
totalTokens: number;
latencyMs: number;
status: "success" | "error";
userId?: string;
error?: string;
}
async function loggedCompletion(
messages: OpenAI.ChatCompletionMessageParam[],
model = "gpt-4o",
userId?: string
): Promise<string> {
const startTime = Date.now();
try {
const response = await client.chat.completions.create({ model, messages });
const latencyMs = Date.now() - startTime;
const log: AIRequestLog = {
timestamp: new Date().toISOString(),
model,
promptTokens: response.usage?.prompt_tokens ?? 0,
completionTokens: response.usage?.completion_tokens ?? 0,
totalTokens: response.usage?.total_tokens ?? 0,
latencyMs,
status: "success",
userId,
};
console.log(JSON.stringify(log));
return response.choices[0].message.content ?? "";
} catch (error) {
const log: AIRequestLog = {
timestamp: new Date().toISOString(),
model,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
latencyMs: Date.now() - startTime,
status: "error",
userId,
error: String(error),
};
console.error(JSON.stringify(log));
throw error;
}
}Анализ логов и метрик
Собранные логи позволяют строить аналитику:
import pandas as pd
# Загрузка логов из файла (JSON Lines)
logs = pd.read_json("ai_requests.jsonl", lines=True)
# Средний latency по моделям
print(logs.groupby("model")["latency_ms"].mean())
# Расход токенов по дням
logs["date"] = pd.to_datetime(logs["timestamp"]).dt.date
daily_tokens = logs.groupby("date")["total_tokens"].sum()
print(daily_tokens)
# Error rate
error_rate = logs[logs["status"] == "error"].shape[0] / logs.shape[0] * 100
print(f"Error rate: {error_rate:.2f}%")
# Самые дорогие запросы
top_expensive = logs.nlargest(10, "total_tokens")[["timestamp", "model", "total_tokens"]]
print(top_expensive)Интеграция с системами мониторинга
Для production рекомендуется отправлять метрики в специализированные системы:
- Prometheus + Grafana — для метрик latency, error rate, token usage
- ELK Stack — для полнотекстового поиска по логам запросов
- Sentry — для отслеживания ошибок с контекстом
- Дашборд ModelSwitch — встроенная аналитика по расходам, моделям и ключам
Логирование — это инвестиция, которая окупается при первом же инциденте. Начните с простой обёртки, постепенно добавляйте метрики и интеграции. Дашборд ModelSwitch предоставляет базовую аналитику из коробки.