Практическое применение LLM в продакшене

# Пример работы с OpenAI API
import openai

client = openai.OpenAI(api_key="your-key")

def generate_response(prompt):
    response = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=500
    )
    return response.choices[0].message.content

class ContextManager:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.history = []

    def add_message(self, role, content):
        self.history.append({"role": role, "content": content})
        self.trim_history()

    def trim_history(self):
        # Удаляем старые сообщения если превышен лимит
        while self.count_tokens() > self.max_tokens:
            if len(self.history) > 2:
                self.history.pop(1)  # Сохраняем system prompt
            else:
                break

    def count_tokens(self):
        # Упрощенный подсчет
        return sum(len(m['content']) // 4 for m in self.history)

import redis
import hashlib
import json

class LLMCache:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379)

    def get_cache_key(self, prompt, model):
        data = f"{model}:{prompt}"
        return hashlib.md5(data.encode()).hexdigest()

    def get(self, prompt, model):
        key = self.get_cache_key(prompt, model)
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None

    def set(self, prompt, model, response, ttl=3600):
        key = self.get_cache_key(prompt, model)
        self.redis.setex(key, ttl, json.dumps(response))

# Использование
cache = LLMCache()

def get_response(prompt):
    cached = cache.get(prompt, "gpt-4")
    if cached:
        return cached

    response = generate_response(prompt)
    cache.set(prompt, "gpt-4", response)
    return response

import time
from openai import OpenAIError

def generate_with_retry(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return generate_response(prompt)
        except OpenAIError as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

from prometheus_client import Counter, Histogram

llm_requests = Counter('llm_requests_total', 'Total LLM requests')
llm_latency = Histogram('llm_request_duration_seconds', 'LLM request duration')
llm_tokens = Counter('llm_tokens_used', 'Total tokens used')

def monitored_generate(prompt):
    llm_requests.inc()

    with llm_latency.time():
        response = generate_response(prompt)

    # Подсчет токенов
    tokens_used = len(prompt) // 4 + len(response) // 4
    llm_tokens.inc(tokens_used)

    return response

# Цены на январь 2025
PRICING = {
    'gpt-4-turbo': {
        'input': 0.01,   # за 1K токенов
        'output': 0.03
    },
    'claude-3-sonnet': {
        'input': 0.003,
        'output': 0.015
    }
}

def calculate_cost(input_tokens, output_tokens, model='gpt-4-turbo'):
    prices = PRICING[model]
    cost = (input_tokens / 1000) * prices['input']
    cost += (output_tokens / 1000) * prices['output']
    return cost

# Пример
cost = calculate_cost(1000, 500, 'gpt-4-turbo')
print(f"Стоимость запроса: ${cost:.4f}")