LLM
System Design

from openai import OpenAI
import json

client = OpenAI()

# Хорошо структурированный few-shot промпт для NER
SYSTEM = """Ты — система извлечения информации из деловых писем.
Извлекай: PERSON, ORGANIZATION, DATE, AMOUNT, LOCATION.
Отвечай ТОЛЬКО валидным JSON. Не добавляй пояснений."""

FEW_SHOT_EXAMPLES = [
    {
        "role": "user",
        "content": "Иван Петров из ООО Ромашка предлагает встретиться 15 мая в Москве."
    },
    {
        "role": "assistant",
        "content": """{"entities": [
  {"text": "Иван Петров", "type": "PERSON"},
  {"text": "ООО Ромашка", "type": "ORGANIZATION"},
  {"text": "15 мая", "type": "DATE"},
  {"text": "Москве", "type": "LOCATION"}
]}"""
    },
    {
        "role": "user",
        "content": "Платёж на 500 000 рублей от Сбербанка поступит 20 декабря."
    },
    {
        "role": "assistant",
        "content": """{"entities": [
  {"text": "500 000 рублей", "type": "AMOUNT"},
  {"text": "Сбербанка", "type": "ORGANIZATION"},
  {"text": "20 декабря", "type": "DATE"}
]}"""
    },
]

def extract_entities(text: str) -> dict:
    messages = [
        {"role": "system", "content": SYSTEM},
        *FEW_SHOT_EXAMPLES,
        {"role": "user", "content": text},
    ]
    resp = client.chat.completions.create(
        model="gpt-4o-mini",    # дешевле чем gpt-4o, часто достаточно
        messages=messages,
        temperature=0,           # детерминизм для структурированного вывода
        response_format={"type": "json_object"},  # JSON mode
    )
    return json.loads(resp.choices[0].message.content)

result = extract_entities("Директор Алексей Смирнов подписал договор с Газпромом на 2 млн.")
# {"entities": [
#   {"text": "Алексей Смирнов", "type": "PERSON"},
#   {"text": "Газпромом", "type": "ORGANIZATION"},
#   {"text": "2 млн", "type": "AMOUNT"}]}

from sentence_transformers import SentenceTransformer, CrossEncoder
import faiss, numpy as np
from openai import OpenAI

# ── Компоненты RAG-пайплайна ─────────────────────────────────
# 1. Bi-encoder: быстрый первичный retrieval (ANN search)
bi_encoder = SentenceTransformer("BAAI/bge-m3")  # поддерживает RU

# 2. Cross-encoder: медленный но точный реранкинг топ-K
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# 3. FAISS index (в проде: Qdrant, Weaviate, Pinecone)
def build_index(documents: list[str]):
    embeddings = bi_encoder.encode(documents, normalize_embeddings=True)
    index = faiss.IndexFlatIP(embeddings.shape[1])
    index.add(embeddings)
    return index, embeddings

def rag_pipeline(
    query: str,
    documents: list[str],
    index: faiss.Index,
    top_k_retrieve: int = 20,   # первичный retrieval: много
    top_k_rerank: int = 4,      # после реранкинга: мало
) -> str:
    # Шаг 1: Bi-encoder retrieval
    q_emb = bi_encoder.encode([query], normalize_embeddings=True)
    D, I = index.search(q_emb, top_k_retrieve)
    candidates = [documents[i] for i in I[0]]

    # Шаг 2: Cross-encoder reranking
    # Cross-encoder читает (query, doc) вместе → точнее bi-encoder
    pairs = [(query, doc) for doc in candidates]
    scores = cross_encoder.predict(pairs)
    ranked = sorted(zip(scores, candidates), reverse=True)
    top_docs = [doc for _, doc in ranked[:top_k_rerank]]

    # Шаг 3: Augmented generation
    context = "\n\n".join([f"[{i+1}] {d}" for i, d in enumerate(top_docs)])
    prompt = f"""Ответь на вопрос, используя ТОЛЬКО предоставленный контекст.
Если ответа нет в контексте — скажи "Не знаю".

Контекст:
{context}

Вопрос: {query}
Ответ:"""

    client = OpenAI()
    resp = client.chat.completions.create(
        model="gpt-4o-mini", temperature=0,
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

# Почему reranking важен:
# Bi-encoder: recall@20 ≈ 0.85, precision@4 ≈ 0.55
# + Cross-encoder: recall@4 ≈ 0.82, precision@4 ≈ 0.78
# Reranking добавляет +42% precision при том же recall!

from openai import OpenAI

client = OpenAI()

def summarize_with_faithfulness_check(document: str,
                                          max_sentences: int = 5) -> dict:
    """Суммаризация с автоматической проверкой фактической точности."""

    # Шаг 1: Суммаризация
    summary_prompt = f"""Создай краткое резюме документа в {max_sentences} предложениях.
Включай ТОЛЬКО факты из документа. Не добавляй внешних знаний.

Документ:
{document}

Резюме:"""

    summary_resp = client.chat.completions.create(
        model="gpt-4o-mini", temperature=0.3,
        messages=[{"role": "user", "content": summary_prompt}]
    )
    summary = summary_resp.choices[0].message.content

    # Шаг 2: Проверка faithfulness (NLI-подход через LLM)
    faithfulness_prompt = f"""Проверь каждое утверждение резюме на соответствие документу.
Отвечай JSON: {{"score": 0-1, "unsupported_claims": ["..."]}}

Документ: {document}
Резюме: {summary}"""

    faith_resp = client.chat.completions.create(
        model="gpt-4o", temperature=0,
        messages=[{"role": "user", "content": faithfulness_prompt}],
        response_format={"type": "json_object"},
    )
    import json
    faith_result = json.loads(faith_resp.choices[0].message.content)

    return {
        "summary": summary,
        "faithfulness_score": faith_result["score"],
        "unsupported_claims": faith_result.get("unsupported_claims", []),
        "is_reliable": faith_result["score"] > 0.9,
    }

# faithfulness_score < 0.9 → тригерим fallback (человеческий ревью)

import time
from openai import OpenAI
from dataclasses import dataclass

@dataclass
class LLMResponse:
    text: str
    confidence: float   # 0..1
    latency_ms: int
    tier_used: str
    needs_review: bool

def call_with_fallback(
    query: str,
    system_prompt: str,
    latency_limit_ms: int = 2000,
    confidence_threshold: float = 0.7,
) -> LLMResponse:

    client = OpenAI()

    # Tier 1: Primary (большая модель)
    try:
        t0 = time.time()
        resp = client.chat.completions.create(
            model="gpt-4o", temperature=0,
            messages=[{"role": "system", "content": system_prompt},
                      {"role": "user", "content": query}],
            timeout=latency_limit_ms / 1000,
        )
        latency = int((time.time() - t0) * 1000)
        text = resp.choices[0].message.content

        # Простая confidence heuristic через logprobs
        confidence = compute_confidence(resp)
        if confidence >= confidence_threshold:
            return LLMResponse(text, confidence, latency, "tier1", False)

    except Exception as e:
        log_error("tier1_failed", error=str(e), query=query)

    # Tier 2: Fallback (меньшая модель, быстрее)
    try:
        t0 = time.time()
        resp = client.chat.completions.create(
            model="gpt-4o-mini", temperature=0,
            messages=[{"role": "system", "content": system_prompt},
                      {"role": "user", "content": query}],
            timeout=1.0,
        )
        latency = int((time.time() - t0) * 1000)
        confidence = compute_confidence(resp)
        text = resp.choices[0].message.content
        return LLMResponse(text, confidence, latency, "tier2",
                             needs_review=confidence < 0.5)

    except Exception as e:
        log_error("tier2_failed", error=str(e))

    # Tier 3: Детерминированный fallback (всегда работает)
    template = get_template_response(query)  # шаблонные ответы по ключевым словам
    return LLMResponse(template, 0.3, 5, "tier3", needs_review=True)

from transformers import (
    AutoModelForCausalLM, AutoTokenizer,
    DataCollatorForLanguageModeling, TrainingArguments, Trainer
)
from datasets import load_dataset

# TAPT: продолжаем предобучение на доменном тексте (CLM)
# Цель: научить модель «говорить» на языке домена
# Пример: юридические договоры → NLP на договорах

tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B")
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2.5-7B")

# Доменный корпус: юридические тексты (unlabeled)
domain_corpus = load_dataset("text", data_files={"train": "legal_contracts.txt"})

def tokenize(examples):
    return tokenizer(examples["text"], truncation=True,
                      max_length=2048, return_special_tokens_mask=True)

tok_ds = domain_corpus.map(tokenize, batched=True, remove_columns=["text"])

# DataCollatorForLanguageModeling для CLM (не MLM)
collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)

# TAPT: LR ниже чем в SFT, больше epochs
args = TrainingArguments(
    output_dir="./qwen-legal-tapt",
    num_train_epochs=2,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,
    learning_rate=1e-4,    # выше чем SFT, ниже чем pretraining
    lr_scheduler_type="cosine",
    bf16=True,
    logging_steps=100,
)

trainer = Trainer(
    model=model, args=args,
    train_dataset=tok_ds["train"],
    data_collator=collator,
)
trainer.train()
# Результат: domain-adapted base model
# Следующий шаг: SFT на (instruction, output) парах для этого домена

from openai import OpenAI
client = OpenAI()

# Constitutional AI: модель сама критикует и исправляет ответы
# по набору принципов (конституции)

CONSTITUTION = [
    "Ответ не должен содержать личные данные пользователей.",
    "Ответ должен признавать неопределённость если она есть.",
    "Ответ не должен содержать юридических советов без оговорки.",
    "Ответ должен быть конструктивным, не осуждающим.",
]

def constitutional_generate(user_query: str, n_revisions: int = 2) -> str:
    """Генерируем ответ → критикуем по конституции → ревизируем."""

    # Шаг 1: Первоначальный ответ
    resp = client.chat.completions.create(
        model="gpt-4o-mini", temperature=0.7,
        messages=[{"role": "user", "content": user_query}]
    )
    draft = resp.choices[0].message.content

    for _ in range(n_revisions):
        principles = "\n".join([f"- {p}" for p in CONSTITUTION])

        # Шаг 2: Критика по принципам
        critique_prompt = f"""Оцени ответ по каждому принципу.
Принципы:
{principles}

Ответ:
{draft}

Укажи нарушения (если есть)."""

        critique_resp = client.chat.completions.create(
            model="gpt-4o", temperature=0,
            messages=[{"role": "user", "content": critique_prompt}]
        )
        critique = critique_resp.choices[0].message.content

        # Шаг 3: Ревизия с учётом критики
        revision_prompt = f"""Перепиши ответ, устранив нарушения.
Нарушения: {critique}
Исходный ответ: {draft}
Исправленный ответ:"""

        rev_resp = client.chat.completions.create(
            model="gpt-4o-mini", temperature=0.3,
            messages=[{"role": "user", "content": revision_prompt}]
        )
        draft = rev_resp.choices[0].message.content

    return draft   # финальный ответ после n_revisions итераций