Трейсинг и оценка качества в LangSmith | Курс по LangChain урок 10

Цель урока

Вы научитесь подключать LangSmith, читать трейсы, чтобы понять, что происходит внутри цепочки. Создавать тестовые датасеты и автоматически оценивать качество RAG системы с помощью LLM-as-a-judge.

Необходимые знания:

Уроки 1–9 (вся предыдущая база)
Базовое понимание метрик качества

Ключевые концепции:

Зачем нужен трейсинг в LLM приложениях
LangSmith: runs, traces, projects
Подключение трейсинга
Датасеты и примеры
Оценщики: LLM-as-a-judge
Метрики RAG: faithfulness, answer relevance, context precision
Отладка с помощью трейсов

Зачем нужен трейсинг

В обычном приложении отладка - это логи и стек вызовов. В LLM приложении этого недостаточно:

Цепочка делает 5 вызовов к модели, какой из них дал плохой результат?
Агент потратил 30 секунд, на каком шаге?
RAG вернул неверный ответ, проблема в поиске или в генерации?
Какой промпт был передан модели на самом деле?

Трейсинг записывает каждый вызов в цепочке: входные данные, выходные данные, время, количество токенов, стоимость. Это делает LLM приложения отлаживаемыми.

LangSmith - это платформа для трейсинга, оценки и мониторинга LangChain приложений. Бесплатный версия покрывает разработку и тестирование.

Подключение LangSmith

Зарегистрируйтесь на smith.langchain.com
В левом меню нажмите Settings → API Keys → Create API Key
Скопируйте ключ (показывается один раз)

Проект в LangSmith это папка для трейсов. Удобно создавать отдельный проект для каждого приложения или эксперимента. Название проекта задаётся в LANGCHAIN_PROJECT. Создать проект: Home → + New Project.

Добавляем в .env:

LANGCHAIN_API_KEY=ls__...      # API ключ LangSmith
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=my-project   # название проекта в LangSmith

Это всё. Никаких изменений в коде? Трейсинг включается автоматически для всех вызовов LangChain.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv() 

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"))

chain = ChatPromptTemplate.from_messages([
    ("system", "Объясняй кратко."),
    ("human", "{question}"),
]) | model | StrOutputParser()

result = chain.invoke({"question": "Что такое event loop?"})
print(result)
# Этот вызов автоматически записан в LangSmith

После запуска откройте LangSmith → Projects → my-project, увидите трейс со всеми деталями.

Что видно в трейсе

Каждый трейс показывает дерево вызовов:

RunnableSequence                     [2.3s, $0.002]
├── ChatPromptTemplate                [0.0s]
│   └── input: {question: "Что такое event loop?"}
│   └── output: [SystemMessage, HumanMessage]
├── ChatOpenAI                        [2.2s, 150 tokens]
│   └── input: [SystemMessage, HumanMessage]
│   └── output: AIMessage("Event loop — это...")
└── StrOutputParser                   [0.0s]
    └── output: "Event loop — это..."

Для каждого узла видно:

Точные входные данные (промпт как он был передан модели)
Выходные данные
Время выполнения
Количество токенов и стоимость
Ошибки с полным стектрейсом

Именованные трейсы

По умолчанию трейсы называются по типу Runnable. Можно задать понятное имя.

# Продолжение предыдущего примера, chain уже определён
from langchain_core.tracers.context import tracing_v2_enabled

# Способ 1: именованный запуск через контекстный менеджер
with tracing_v2_enabled(project_name="my-project"):
    result = chain.invoke({"question": "Что такое GIL?"})

# Способ 2: через config — имя, метаданные и теги прямо в вызове
result = chain.invoke(
    {"question": "Что такое GIL?"},
    config={
        "run_name": "explain_concept",
        "metadata": {"user_id": "user_42", "version": "v2"},
        "tags": ["production", "explain"],
    }
)
print(result)

metadata и tags позволяют фильтровать трейсы в LangSmith, удобно для A/B тестирования промптов или анализа по пользователям.

Ручной трейсинг

Иногда нужно трейсить код, который не использует LangChain напрямую.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langsmith import traceable

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"))
chain = ChatPromptTemplate.from_messages([
    ("system", "Объясняй кратко."),
    ("human", "{question}"),
]) | model | StrOutputParser()


@traceable(name="fetch_user_context")
def fetch_user_context(user_id: str) -> dict:
    """Эта функция будет отображаться в трейсе как отдельный шаг."""
    # Обращение к БД, кешу и т.д.
    return {"user_id": user_id, "preferences": ["python", "asyncio"]}


@traceable(name="format_response")
def format_response(answer: str, sources: list) -> str:
    return f"{answer}\n\nИсточники: {', '.join(sources)}"


# Обе функции будут видны в трейсе LangSmith
context = fetch_user_context("user_42")
answer = chain.invoke({"question": "Как работает GIL?"})
formatted = format_response(answer, ["doc_1", "doc_2"])
print(formatted)

В LangSmith вы увидите трейс с деревом вызовов:

fetch_user_context          [0.0s]
  └── input:  user_id = "user_42"
  └── output: {"user_id": "user_42", "preferences": [...]}

RunnableSequence            [1.8s]
  ├── ChatPromptTemplate
  ├── ChatOpenAI              [1.8s, ~80 tokens]
  └── StrOutputParser

format_response             [0.0s]
  └── input:  answer = "GIL (Global Interpreter Lock)...", sources = [...]
  └── output: "GIL (Global Interpreter Lock)...\n\nИсточники: doc_1, doc_2"

Без @traceable в трейсе был бы виден только RunnableSequence. С декоратором все три шага отображаются как отдельные узлы, и для каждого видны входные/выходные данные и время выполнения.

Датасеты и оценка качества

Трейсинг помогает с отладкой. Для оценки качества нужен другой инструмент, датасеты с ожидаемыми ответами и evaluators.

Создание датасета

from dotenv import load_dotenv
from langsmith import Client


load_dotenv()

client = Client()

# Создаём датасет
dataset = client.create_dataset(
    dataset_name="rag-qa-dataset",
    description="Вопросы и ответы для оценки RAG-системы",
)

# Добавляем примеры: вход + ожидаемый выход
examples = [
    {
        "inputs":  {"question": "Как запустить несколько корутин параллельно?"},
        "outputs": {"answer": "Используй asyncio.gather() для параллельного запуска корутин."},
    },
    {
        "inputs":  {"question": "Что такое event loop в asyncio?"},
        "outputs": {"answer": "Event loop — менеджер выполнения корутин в asyncio."},
    },
    {
        "inputs":  {"question": "Как сделать HTTP-запрос асинхронно?"},
        "outputs": {"answer": "Используй aiohttp.ClientSession или httpx.AsyncClient."},
    },
]

client.create_examples(
    inputs=[e["inputs"] for e in examples],
    outputs=[e["outputs"] for e in examples],
    dataset_id=dataset.id,
)

print(f"Датасет создан: {dataset.id}")

Запуск оценки

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import JsonOutputParser
from langsmith.evaluation import evaluate
from langsmith import Client

load_dotenv()
client = Client()

# Пример RAG цепочки
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
texts = [
    "asyncio.gather() запускает корутины параллельно.",
    "asyncio.sleep() приостанавливает корутину.",
]
docs = [Document(page_content=t) for t in texts]
vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})


def format_docs(docs): return "\n\n".join(d.page_content for d in docs)


rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "Отвечай только на основе контекста:\n{context}"),
    ("human", "{question}"),
])

rag_chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(retriever.invoke(x["question"])))
    | rag_prompt | model | StrOutputParser()
)


# Функция, которую оцениваем — принимает пример, возвращает ответ
def rag_target(inputs: dict) -> dict:
    answer = rag_chain.invoke({"question": inputs["question"]})
    return {"answer": answer}


# LLM-as-a-judge: оцениваем корректность ответа
judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


correctness_prompt = ChatPromptTemplate.from_messages([
    ("system", """Оцени корректность ответа относительно эталонного.
Верни JSON: {{"score": 0.0-1.0, "reasoning": "краткое объяснение"}}

1.0 — ответ полностью корректен
0.5 — ответ частично верен
0.0 — ответ неверен или не по теме"""),
    ("human", """Вопрос: {input}
Эталонный ответ: {reference}
Полученный ответ: {output}"""),
])


judge_chain = correctness_prompt | judge_model | JsonOutputParser()


def correctness_evaluator(run, example) -> dict:
    """Оценивает корректность ответа. Возвращает score от 0 до 1."""
    result = judge_chain.invoke({
        "input": example.inputs["question"],
        "reference": example.outputs["answer"],
        "output": run.outputs["answer"],
    })
    return {
        "key": "correctness",
        "score": result["score"],
        "comment": result["reasoning"],
    }


# Запускаем оценку
results = evaluate(
    rag_target,
    data="rag-qa-dataset",
    evaluators=[correctness_evaluator],
    experiment_prefix="rag-v1",
    metadata={"model": "gpt-4o", "chunk_size": 1000},
)


for r in results:
    print(r)

Результаты появляются в LangSmith → Datasets → rag-qa-dataset → Experiments. Можно сравнивать разные версии системы.

Метрики для RAG

Для RAG важны три отдельных метрики. Каждая оценивается отдельным LLM-judge.

Функции принимают run и example, объекты LangSmith, которые передаются автоматически внутри evaluate(...). Напрямую вызвать их нельзя, но ниже для каждой функции показан пример прямого вызова judge для проверки логики.

1. Faithfulness

Ответ должен основываться на контексте, а не выдумываться.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

load_dotenv()
judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


def check_faithfulness(context: str, answer: str) -> dict:
    """Прямой вызов judge. Можно запустить без LangSmith."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Определи, подтверждается ли ответ предоставленным контекстом.
Верни JSON: {{"score": 0.0-1.0, "reasoning": "объяснение"}}
1.0 — каждое утверждение в ответе есть в контексте
0.0 — ответ содержит информацию, которой нет в контексте"""),
        ("human", "Контекст:\n{context}\n\nОтвет:\n{answer}"),
    ])
    return (prompt | judge_model | JsonOutputParser()).invoke({
        "context": context, "answer": answer,
    })


def faithfulness_evaluator(run, example) -> dict:
    """Тонкая обёртка для evaluate(...) из LangSmith."""
    result = check_faithfulness(
        context=run.outputs.get("context", ""),
        answer=run.outputs.get("answer", ""),
    )
    return {"key": "faithfulness", "score": result["score"]}

# Проверка логики напрямую:
result = check_faithfulness(
    context="asyncio.gather() запускает корутины параллельно.",
    answer="asyncio.gather() используется для параллельного запуска корутин.",
)
print(result)  # {"score": 1.0, "reasoning": "Ответ полностью подтверждается контекстом."}

2. Answer Relevance (релевантность ответа)

Ответ должен отвечать именно на заданный вопрос:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

load_dotenv()
judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


def check_answer_relevance(question: str, answer: str) -> dict:
    """Прямой вызов judge. Можно запустить без LangSmith."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Оцени, насколько ответ релевантен вопросу.
Верни JSON: {{"score": 0.0-1.0, "reasoning": "объяснение"}}"""),
        ("human", "Вопрос: {question}\n\nОтвет: {answer}"),
    ])
    return (prompt | judge_model | JsonOutputParser()).invoke({
        "question": question, "answer": answer,
    })

def answer_relevance_evaluator(run, example) -> dict:
    """Тонкая обёртка для evaluate(...) из LangSmith."""
    result = check_answer_relevance(
        question=example.inputs["question"],
        answer=run.outputs.get("answer", ""),
    )
    return {"key": "answer_relevance", "score": result["score"]}

# Проверка логики напрямую:
result = check_answer_relevance(
    question="Как запустить несколько корутин?",
    answer="asyncio.gather() запускает несколько корутин параллельно.",
)
print(result) # {"score": 0.7, ...}

3. Context Precision (точность поиска)

Найденные документы действительно нужны для ответа. Здесь реализована упрощённая версия, LLM-judge оценивает, насколько контекст был полезен для генерации ответа. Context Precision в академическом смысле требует пословной разметки релевантности каждого чанка как в RAGAS.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

load_dotenv()
judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


def check_context_precision(question: str, context: str, answer: str) -> dict:
    """Прямой вызов judge. Можно запустить без LangSmith."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Оцени, насколько контекст необходим и достаточен для ответа.
Верни JSON: {{"score": 0.0-1.0, "reasoning": "объяснение"}}
1.0 — весь контекст релевантен и использован
0.0 — контекст не нужен для ответа или содержит лишнее"""),
        ("human", "Вопрос: {question}\nКонтекст:\n{context}\nОтвет:\n{answer}"),
    ])
    return (prompt | judge_model | JsonOutputParser()).invoke({
        "question": question, "context": context, "answer": answer,
    })


def context_precision_evaluator(run, example) -> dict:
    """Тонкая обёртка для evaluate(...) из LangSmith."""
    result = check_context_precision(
        question=example.inputs["question"],
        context=run.outputs.get("context", ""),
        answer=run.outputs.get("answer", ""),
    )
    return {"key": "context_precision", "score": result["score"]}

# Проверка логики напрямую:
result = check_context_precision(
    question="Как запустить несколько корутин?",
    context="asyncio.gather() запускает корутины параллельно.\nasyncio.sleep() приостанавливает корутину.",
    answer="asyncio.gather() запускает несколько корутин параллельно.",
)
print(result)  # {"score": 0.7, ...}

Оффлайн оценка без LangSmith

Если не хотите использовать LangSmith, можно оценивать локально.

import os
import statistics
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

# Пример RAG цепочки для тестирования
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
texts = [
    "asyncio.gather() запускает корутины параллельно.",
    "asyncio.sleep() приостанавливает корутину без блокировки event loop.",
    "aiohttp.ClientSession используется для асинхронных HTTP запросов.",
]
docs = [Document(page_content=t) for t in texts]
vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})


def format_docs(docs): return "\n\n".join(d.page_content for d in docs)


rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "Отвечай только на основе контекста:\n{context}"),
    ("human", "{question}"),
])

rag_chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(retriever.invoke(x["question"])))
    | rag_prompt | model | StrOutputParser()
)


def evaluate_locally(rag_fn, test_cases: list[dict]) -> dict:
    """
    rag_fn: функция, принимает вопрос, возвращает {"answer": str, "context": str}
    test_cases: [{"question": str, "expected": str}, ...]
    """
    correctness_scores = []

    prompt = ChatPromptTemplate.from_messages([
        ("system", "Оцени корректность ответа (0.0-1.0). JSON: {{\"score\": float}}"),
        ("human", "Вопрос: {q}\nЭталон: {ref}\nОтвет: {ans}"),
    ])
    scorer = prompt | judge_model | JsonOutputParser()

    for case in test_cases:
        result = rag_fn(case["question"])
        score = scorer.invoke({
            "q":   case["question"],
            "ref": case["expected"],
            "ans": result["answer"],
        })
        correctness_scores.append(score["score"])
        print(f"Q: {case['question'][:50]}... → score: {score['score']:.2f}")

    return {
        "mean_correctness": statistics.mean(correctness_scores),
        "min_correctness":  min(correctness_scores),
        "max_correctness":  max(correctness_scores),
    }


# Использование
test_cases = [
    {"question": "Как запустить несколько корутин?", "expected": "asyncio.gather()"},
    {"question": "Что делает asyncio.sleep()?", "expected": "Приостанавливает корутину без блокировки"},
    {"question": "Какая альтернатива requests для async?", "expected": "aiohttp или httpx"},
]

metrics = evaluate_locally(lambda q: {"answer": rag_chain.invoke({"question": q})}, test_cases)
print(f"\nРезультаты: {metrics}")

Отладка

Сценарий 1: Модель даёт неверный ответ

Открываем трейс в LangSmith
Смотрим узел ChatOpenAI → inputs → точный промпт
Проверяем: правильный ли контекст передан? Есть ли нужная информация?
Если контекст неверный, проблема в retriever. Если верный, проблема в промпте.

Сценарий 2: Агент зациклился

Находим трейс с большим временем выполнения
Разворачиваем дерево вызовов и видим, какие инструменты вызывались
Определяем шаг, после которого агент начал повторяться
Правим системный промпт или добавляем ограничение итераций

Сценарий 3: Высокие затраты

LangSmith → Tracing → выбираем проект → выбираем период
Сортируем runs по стоимости
Находим неожиданно дорогие вызовы
Смотрим, не слишком ли большой контекст передаётся? Не дублируются ли вызовы?

LangSmith агрегирует стоимость по проекту, пользователю (metadata.user_id), тегам и версии модели, удобно для мониторинга реальных затрат в production.

Polly или AI-ассистент для отладки

Встроенный ассистент в интерфейсе LangSmith. Вместо ручного просмотра дерева вызовов можно спросить прямо: "Почему агент зациклился?" или "Где потеряли контекст диалога?". Polly анализирует полный трейс и предлагает гипотезы.

LangSmith Fetch CLI отладка из терминала

pip install langsmith-fetch

# Последние 20 трейсов проекта
langsmith-fetch traces --project-uuid <uuid> --limit 20

# Трейсы за последние 30 минут
langsmith-fetch traces --project-uuid <uuid> --last-n-minutes 30

# Экспорт трейсов в файлы для офлайн-анализа
langsmith-fetch traces ./traces --project-uuid <uuid>

UUID проекта найдёте в LangSmith → Tracing → выбрать проект → Settings.

Сравнение версий системы

Одна из главных возможностей LangSmith, сравнивать, как изменение в системе влияет на качество.

# Этот код — концептуальный шаблон.
# basic_rag_chain и advanced_rag_chain, RAG цепочки из уроков 8 и 9.
# correctness_evaluator, faithfulness_evaluator, функции из предыдущего раздела.
# Датасет "rag-qa-dataset" создан через LangSmith Client (см. выше).
from langsmith.evaluation import evaluate

# Версия 1: базовый RAG
results_v1 = evaluate(
    lambda inputs: {"answer": basic_rag_chain.invoke(inputs["question"])},
    data="rag-qa-dataset",
    evaluators=[correctness_evaluator, faithfulness_evaluator],
    experiment_prefix="rag-baseline",
)

# Версия 2: RAG с гибридным поиском и re-ranking
results_v2 = evaluate(
    lambda inputs: {"answer": advanced_rag_chain.invoke(inputs["question"])},
    data="rag-qa-dataset",
    evaluators=[correctness_evaluator, faithfulness_evaluator],
    experiment_prefix="rag-advanced",
)

# В LangSmith можно поставить эксперименты рядом и сравнить по метрикам

Это основа итеративного улучшения: меняем, измеряем, сравниваем.

Pairwise Annotation Queues, A/B оценка ответов

Вместо оценки каждого ответа по отдельности разметчик видит два ответа рядом и выбирает лучший. Удобно для финального выбора между промптом v1 и v2 или между моделями.

Настраивается через LangSmith UI: Datasets → Annotation Queues → New Pairwise Queue.

Распространённые ошибки

1. Нет LANGCHAIN_PROJECT все трейсы в default

# Плохо: все трейсы мешаются в одном проекте
LANGCHAIN_TRACING_V2=true
# Хорошо: разные фичи/версии, разные проекты
LANGCHAIN_PROJECT=rag-v2-hybrid-search

2. Оценка на одних и тех же примерах, которые использовались при разработке

Тестовый датасет должен состоять из примеров, которые система не видела при создании. Иначе метрики завышены.

3. LLM-judge без валидации

Модель судья тоже ошибается. Проверьте вручную несколько примеров с разными оценками и убедитесь, что судья адекватен.

4. Оценивать только финальный ответ

Проблема может быть на уровне retrieval (нашли не то) или generation (не использовали то, что нашли). Оценивай оба уровня.

5. Не сохранять контекст в outputs

# Плохо: evaluator не видит, что нашёл retriever
def rag_target(inputs):
    return {"answer": rag_chain.invoke(inputs["question"])}

# Хорошо: сохраняем и ответ, и контекст для оценки
def rag_target(inputs):
    result = rag_with_sources.invoke(inputs["question"])
    return {
        "answer":  result["answer"],
        "context": "\n".join(d.page_content for d in result["sources"]),
    }

Полный пример кода

import os
import statistics
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

load_dotenv()

# --- RAG-система ---

model  = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")

texts = [
    "asyncio.gather() запускает корутины параллельно, возвращает список результатов.",
    "asyncio.create_task() создаёт фоновую задачу из корутины.",
    "asyncio.sleep(n) приостанавливает корутину без блокировки event loop.",
    "aiohttp.ClientSession используется для асинхронных HTTP запросов.",
    "httpx.AsyncClient — современная альтернатива aiohttp с похожим API на requests.",
    "pytest-asyncio позволяет тестировать async def через @pytest.mark.asyncio.",
]

docs = [Document(page_content=t, metadata={"id": i}) for i, t in enumerate(texts)]
vectorstore = FAISS.from_documents(docs, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """Отвечай только на основе контекста. Если ответа нет, скажи об этом.

Контекст:
{context}"""),
    ("human", "{question}"),
])


def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

rag_chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(retriever.invoke(x["question"])))
    | rag_prompt
    | model
    | StrOutputParser()
)

rag_with_context = RunnableParallel(
    answer=rag_chain,
    context=lambda x: format_docs(retriever.invoke(x["question"])),
)

# --- Оценочный датасет ---

test_cases = [
    {
        "question": "Как запустить несколько корутин одновременно?",
        "expected": "asyncio.gather() запускает корутины параллельно.",
    },
    {
        "question": "Как приостановить корутину?",
        "expected": "asyncio.sleep() приостанавливает корутину без блокировки.",
    },
    {
        "question": "Какую библиотеку использовать для асинхронных HTTP запросов?",
        "expected": "aiohttp или httpx для асинхронных запросов.",
    },
    {
        "question": "Как тестировать асинхронный код?",
        "expected": "pytest-asyncio с декоратором @pytest.mark.asyncio.",
    },
    {
        "question": "Как развернуть кластер Kubernetes?",
        "expected": None,  # нет в базе знаний
    },
]


# --- LLM-as-a-judge ---

judge_model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

eval_prompt = ChatPromptTemplate.from_messages([
    ("system", """Оцени качество ответа RAG-системы по трём критериям.
Верни JSON строго в формате:
{{
  "correctness": {{"score": 0.0-1.0, "comment": "..."}},
  "faithfulness": {{"score": 0.0-1.0, "comment": "..."}},
  "answer_relevance": {{"score": 0.0-1.0, "comment": "..."}}
}}"""),
    ("human", """Вопрос: {question}
Эталонный ответ: {expected}
Полученный ответ: {answer}
Контекст: {context}"""),
])

evaluator = eval_prompt | judge_model | JsonOutputParser()


# --- Запуск оценки ---

def run_evaluation(test_cases: list) -> dict:
    all_scores = {"correctness": [], "faithfulness": [], "answer_relevance": []}

    print("=" * 60)
    for i, case in enumerate(test_cases, 1):
        result = rag_with_context.invoke({"question": case["question"]})

        scores = evaluator.invoke({
            "question": case["question"],
            "expected": case["expected"] or "Ответа нет в базе знаний",
            "answer":   result["answer"],
            "context":  result["context"],
        })

        print(f"\n[{i}/{len(test_cases)}] {case['question'][:55]}...")
        print(f"  Ответ: {result['answer'][:80]}...")
        for metric, data in scores.items():
            s = data["score"]
            all_scores[metric].append(s)
            emoji = "✅" if s >= 0.7 else ("⚠️" if s >= 0.4 else "❌")
            print(f"  {emoji} {metric}: {s:.2f} — {data['comment'][:60]}")

    print("\n" + "=" * 60)
    print("ИТОГОВЫЕ МЕТРИКИ:")
    aggregated = {}
    for metric, scores in all_scores.items():
        mean = statistics.mean(scores)
        aggregated[metric] = mean
        print(f"  {metric}: {mean:.2f} (min: {min(scores):.2f}, max: {max(scores):.2f})")

    return aggregated

metrics = run_evaluation(test_cases)

Практическое задание

Создайте систему оценки качества для RAG из урока 8 или 9.

Требования:

1) Подключите LangSmith трейсинг или реализуйте локальную оценку без него

2) Создайте датасет минимум из 8 тест-кейсов, покрывающих:

Вопросы с ответом в базе знаний
Вопросы без ответа в базе знаний
Вопросы, требующие объединения информации из нескольких чанков

3) Реализуйте три функции evaluator:

correctness - насколько правильный ответ
faithfulness - основывается ли ответ на контексте
no_answer_detection - правильно ли система говорит "не знаю" (precision для случаев без ответа)

4) Запустите оценку для двух версий системы:

Базовый RAG (урок 8)
Продвинутый RAG с гибридным поиском (урок 9)

5) Выведи сравнительную таблицу метрик:

Метрика             | Базовый RAG | Продвинутый RAG
correctness         |    0.72     |     0.85
faithfulness        |    0.81     |     0.89
no_answer_detection |    0.60     |     0.75

Итоги урока

Мы научились делать, улучшать и измерять качество LLM приложений. Теперь разберём финальный практический навык, как правильно собрать всё вместе в продакшн приложение. В следующем уроке создадим полноценного чат-бота с RAG и API. Бэкенд на FastAPI, стриминг ответов, версионирование промптов и деплой.

<< Урок 9

Урок 11 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия