Расскажи про свой опыт с генеративным ИИ

Мой опыт с генеративным ИИ и LLM

После 10+ лет в ML/DS я активно работаю с генеративными моделями — от их запуска до production deployment. Вот мой подробный опыт.

Эволюция: от GPT к современному состоянию

2022-2023: GPT-3 era Первый контакт был через OpenAI API. Я экспериментировал с:

• Text generation и классификацией
• Few-shot prompting для classification задач
• Fine-tuning на кастомных данных (дорого!)

import openai

openai.api_key = "sk-..."
response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Classify this text: ..."}
    ],
    temperature=0.7,
    max_tokens=100
)

Результат: GPT-3 очень дорого ($0.002 за 1K tokens), но качество хорошее. Не подходит для high-volume inference.

2023: Open-source boom Появились Llama, Mistral, Phi. Я начал self-hosting:

# Локальный запуск с ollama
ollama run mistral:latest
# Или с llama.cpp для CPU inference

# API через llama-cpp-python
from llama_cpp import Llama

llm = Llama(model_path="./mistral-7b.gguf", n_gpu_layers=-1)
response = llm("Что такое машинное обучение?", max_tokens=100)

2024: LLMOps и production systems Перейду к production-grade архитектуре.

Текущая работа с LLM

1. Использование LLM как сервиса

Основная работа на OpenAI API + вспомогательные:

from anthropic import Anthropic

client = Anthropic()

def extract_entities(text: str) -> dict:
    """Extract entities using Claude API"""
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system="You are an NER expert. Extract all entities.",
        messages=[
            {"role": "user", "content": f"Extract entities from: {text}"}
        ]
    )
    return response.content[0].text

# Использование
entities = extract_entities("Apple Inc. was founded by Steve Jobs in Cupertino.")
# Output: "ORGANIZATION: Apple Inc., PERSON: Steve Jobs, LOCATION: Cupertino"

Стоимость оптимизация:

• GPT-4o: $0.005 per 1K input tokens, $0.015 per 1K output
• Claude 3.5: $0.003 / $0.015
• Mistral: $0.0001 / $0.0003 (самый дешёвый)
• Llama (self-hosted): ~$0.00 (только инфра)

2. RAG (Retrieval-Augmented Generation)

Это главный use case в production. Вместо fine-tuning используем контекст:

from langchain.vectorstores import Pinecone
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 1. Индексируем документы в vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Pinecone.from_documents(
    documents=documents,
    embedding=embeddings,
    index_name="my-index"
)

# 2. Создаём RAG chain
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_store.as_retriever(k=3),
    return_source_documents=True
)

# 3. Отвечаем на вопросы на основе документов
answer = qa_chain({"query": "Какая политика возврата?"})
print(answer["result"])
print(answer["source_documents"])

Плюсы RAG:

• Не нужен fine-tuning
• Модель использует актуальные данные
• Можем контролировать, на чём обучается
• Дешевле, чем fine-tuning

3. Embeddings для semantic search

from openai import OpenAI

client = OpenAI(api_key="sk-...")

# 1. Генерируем embeddings для документов
texts = [
    "The cat sat on the mat",
    "A dog is playing in the garden",
    "Machine learning is powerful"
]

embeddings = []
for text in texts:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    embeddings.append(response.data[0].embedding)

# 2. Поиск похожих текстов
query = "animals"
query_embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input=query
).data[0].embedding

# Cosine similarity
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

similarities = cosine_similarity([query_embedding], embeddings)[0]
top_idx = np.argsort(similarities)[::-1][:3]
for idx in top_idx:
    print(f"{texts[idx]}: {similarities[idx]:.3f}")

Fine-tuning: когда и как

Fine-tuning стоит, если:

1. Много примеров (1000+) одного типа задачи
2. Качество базовой модели недостаточно
3. Нужна специфичная тонкая настройка

# Fine-tuning GPT-3.5
import openai
import json

# Подготовка данных
training_data = [
    {
        "messages": [
            {"role": "system", "content": "Classify sentiment"},
            {"role": "user", "content": "This movie was amazing!"},
            {"role": "assistant", "content": "POSITIVE"}
        ]
    },
    # ... ещё примеры
]

# Сохраняем в JSONL
with open("training.jsonl", "w") as f:
    for example in training_data:
        f.write(json.dumps(example) + "\n")

# Создаём файл для OpenAI
file_response = openai.File.create(
    file=open("training.jsonl", "rb"),
    purpose="fine-tune"
)

# Запускаем fine-tuning
ft_response = openai.FineTune.create(
    training_file=file_response.id,
    model="gpt-3.5-turbo",
    n_epochs=3
)

# Используем fine-tuned модель
response = openai.ChatCompletion.create(
    model=f"ft:gpt-3.5-turbo:{ft_response.fine_tuned_model}",
    messages=[...]
)

Стоимость fine-tuning: $0.0080 per 1K input tokens (в 4x дороже, чем базовая модель).

Prompt engineering & optimization

1. System prompts

SYSTEM_PROMPTS = {
    "analyst": """You are a data analyst expert. 
    Provide concise, data-driven insights.
    Use specific numbers and statistics.
    Format response as: [Key Finding] -> [Evidence] -> [Recommendation]""",
    
    "code_reviewer": """You are an expert code reviewer.
    Check for:
    1. Performance issues
    2. Security vulnerabilities
    3. Code clarity
    Format as: [Issue] - [Severity] - [Fix]""",
}

def analyze(text: str, role: str) -> str:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        system=SYSTEM_PROMPTS[role],
        messages=[{"role": "user", "content": text}],
        max_tokens=1024
    )
    return response.content[0].text

2. Chain-of-thought prompting

def solve_complex_problem(problem: str) -> str:
    prompt = f"""Solve this step by step.

Problem: {problem}

Think through:
1. What is the core issue?
2. What are the constraints?
3. What are possible approaches?
4. Which is optimal?
5. What's the final answer?

Answer:"""
    
    response = client.messages.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.5  # Более детерминированный результат
    )
    return response.content[0].text

3. Few-shot prompting

examples = """\nExample 1:
Input: "Great movie, loved it!"
Output: POSITIVE

Example 2:
Input: "Waste of time"
Output: NEGATIVE

Example 3:
Input: "It was okay"
Output: NEUTRAL
"""

def classify_text(text: str) -> str:
    response = client.messages.create(
        model="gpt-3.5-turbo",
        messages=[{
            "role": "user",
            "content": f"{examples}\n\nClassify: \"{text}\"\nOutput:"
        }],
        temperature=0
    )
    return response.content[0].text.strip()

Eval & Quality Assurance

Как мы проверяем качество генеративных моделей:

from langchain.evaluation import load_evaluator

# 1. Semantic similarity
similarity_evaluator = load_evaluator("labeled_score_string")

eval_result = similarity_evaluator.evaluate_strings(
    prediction="The capital of France is Paris",
    reference="Paris is the capital of France",
    input="What is the capital of France?"
)
print(f"Similarity score: {eval_result['score']}")  # 0.95

# 2. Custom evaluation
def check_factuality(output: str, ground_truth: list[str]) -> float:
    """Check if output contains any false claims"""
    facts = extract_facts(output)
    correct = sum(1 for f in facts if f in ground_truth)
    return correct / len(facts) if facts else 0

# 3. A/B тест в production
import random

def get_response(query: str) -> tuple[str, str]:
    if random.random() < 0.5:
        model = "gpt-4o"
    else:
        model = "gpt-3.5-turbo"
    
    response = client.messages.create(
        model=model,
        messages=[{"role": "user", "content": query}],
        metadata={"model": model}  # Для трекинга
    )
    return response.content[0].text, model

Challenges & Solutions

1. Hallucinations (галлюцинации)

# Solution: Constrain output format
def extract_with_verification(text: str, expected_fields: list[str]) -> dict:
    prompt = f"""Extract the following fields: {expected_fields}
    
If a field is not found, use "NOT_FOUND".
Return as JSON.

Text: {text}

JSON:"""
    
    response = client.messages.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0  # Deterministic
    )
    
    result = json.loads(response.content[0].text)
    # Verify structure
    for field in expected_fields:
        if field not in result:
            result[field] = "NOT_FOUND"
    return result

2. Latency

# Solution: Caching + parallel requests
import hashlib
import asyncio

cache = {}

async def get_embedding_cached(text: str) -> list[float]:
    hash_key = hashlib.sha256(text.encode()).hexdigest()
    if hash_key in cache:
        return cache[hash_key]
    
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    
    embedding = response.data[0].embedding
    cache[hash_key] = embedding
    return embedding

# Параллельно обработаем много запросов
async def process_batch(texts: list[str]):
    tasks = [get_embedding_cached(text) for text in texts]
    return await asyncio.gather(*tasks)

3. Cost control

# Monitoring usage
class TokenCounter:
    def __init__(self):
        self.input_tokens = 0
        self.output_tokens = 0
    
    def add_usage(self, input_tokens: int, output_tokens: int):
        self.input_tokens += input_tokens
        self.output_tokens += output_tokens
    
    def estimate_cost(self, model: str) -> float:
        prices = {
            "gpt-4o": {"input": 0.005, "output": 0.015},
            "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
        }
        p = prices[model]
        return (self.input_tokens * p["input"] + 
                self.output_tokens * p["output"]) / 1000

counter = TokenCounter()
# ... использование ...
cost = counter.estimate_cost("gpt-4o")
print(f"Estimated cost: ${cost:.2f}")

Мои рекомендации (2025)

Для разработки:

1. Начни с OpenAI API или Anthropic (best DX)
2. Используй RAG вместо fine-tuning (дешевле, лучше)
3. Оптимизируй промпты через A/B тесты
4. Кэшируй embeddings и frequently used outputs

Для production:

1. Используй self-hosted модели, если очень high volume
2. Реализуй fallback на smaller model при высокой нагрузке
3. Мониторь hallucinations в product
4. Трекируй стоимость per request

Инструменты, которые рекомендую:

• LangChain (orchestration)
• Pinecone/Weaviate (vector DB)
• MLflow (experiment tracking)
• Guardrails AI (safety checks)

Вывод

Генеративный ИИ — это мощный инструмент, но не серебряная пуля. Успех зависит от:

• Выбора правильной модели для задачи
• Отличного prompt engineering
• Хорошей evaluation и QA процесса
• Мониторинга cost vs quality trade-offs

После 2+ лет работы с LLM я вижу, что лучшие решения комбинируют LLM с traditional ML, а не полностью на них опираются.