AI Engineering Academy · Урок

Запрос, поиск и генерация

Напишите конвейер обработки запроса, который создаёт эмбеддинг вопроса пользователя, находит k лучших фрагментов, формирует дополненный запрос, обращается к LLM и возвращает ответ с цитатами.

Урок 4 из 413 шагов

«Запрос, поиск и генерация» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Engineering Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Engineering Academy содержит 4 уроков всего.

Конвейер запросов: от начала до конца

Конвейер запросов — это онлайн-часть RAG, то есть код, который выполняется в реальном времени, когда пользователь задаёт вопрос. Он связывает все компоненты, созданные во время индексации: модель эмбеддингов, векторное хранилище, шаблон запроса и LLM. Хорошо реализованный конвейер запросов завершается менее чем за 500 мс для большинства рабочих нагрузок и создаёт обоснованные ответами источников ответы с цитатами. В этом уроке мы с нуля создадим каждый этап.

Шаг 1: создание эмбеддинга запроса пользователя

На первом шаге естественно-языковой вопрос пользователя преобразуется в векторный эмбеддинг с помощью той же модели, которая использовалась при индексации. Этот эмбеддинг кодирует смысл вопроса и будет сравниваться с эмбеддингами фрагментов документов в векторном хранилище. Сделайте этот шаг быстрым: используйте лёгкую модель, например text-embedding-3-small, и кэшируйте эмбеддинги для повторяющихся идентичных запросов.

from openai import OpenAI

client = OpenAI()

def embed_query(question: str) -> list:
    response = client.embeddings.create(
        model='text-embedding-3-small',
        input=[question]
    )
    return response.data[0].embedding

user_question = 'What is our remote work policy?'
query_vector = embed_query(user_question)
print(f'Query embedded: {len(query_vector)}-dim vector')

Шаг 2: извлечение K лучших фрагментов

Отправьте вектор запроса в векторное хранилище, чтобы найти K наиболее близких по смыслу фрагментов. Возвращённые совпадения ранжируются по показателю косинусного сходства (обычно от 0,0 до 1,0; чем выше, тем лучше). Идеальное значение K обеспечивает баланс между информативностью контекста и расходом окна контекста: распространённая начальная настройка — K=5. Здесь также можно применить фильтры метаданных, чтобы ограничить поиск конкретным подразделением, типом документа или диапазоном дат.

def retrieve_chunks(query_vector, index, top_k=5, filters=None):
    query_params = {
        'vector': query_vector,
        'top_k': top_k,
        'include_metadata': True
    }
    if filters:
        query_params['filter'] = filters

    results = index.query(**query_params)

    chunks = []
    for match in results.matches:
        chunks.append({
            'score': match.score,
            'text': match.metadata['text'],
            'source': match.metadata.get('source', ''),
            'page': match.metadata.get('page', '')
        })
    return chunks

Шаг 3: фильтрация по порогу оценки

Не все извлечённые фрагменты действительно релевантны: некоторые могут иметь низкие показатели сходства, но всё равно попасть в число K лучших, если запрос выходит за пределы охвата индекса. Применяйте минимальный порог оценки, чтобы отфильтровать совпадения с низкой уверенностью. Если все извлечённые фрагменты ниже порога, верните ответ «информация не найдена», вместо того чтобы отправлять нерелевантный контекст в LLM: это привело бы к худшему ответу, чем корректный отказ.

MIN_SCORE_THRESHOLD = 0.75

def filter_by_score(chunks, threshold=MIN_SCORE_THRESHOLD):
    relevant = [c for c in chunks if c['score'] >= threshold]
    if not relevant:
        print(f'No chunks above threshold {threshold}. Scores: {[c["score"] for c in chunks]}')
    return relevant

retrieved = retrieve_chunks(query_vector, index, top_k=5)
filtered = filter_by_score(retrieved)
if not filtered:
    print('Responding: no relevant information found')

Шаг 4: форматирование блока контекста

Объедините извлечённые фрагменты в структурированный блок контекста, который прочитает LLM. Пометьте каждый фрагмент его источником, чтобы модель могла точно указать цитату. Для ясности добавьте разделитель между фрагментами. Удерживайте общий контекст в пределах бюджета токенов: подсчитывайте токены с помощью tiktoken и сокращайте контекст или удаляйте фрагменты с более низкими оценками, если превышен лимит. Блок контекста вставляется в запрос между системной инструкцией и вопросом пользователя.

def format_context(chunks):
    parts = []
    for i, chunk in enumerate(chunks, start=1):
        source_label = chunk['source']
        if chunk.get('page'):
            source_label += f", page {chunk['page']}"
        parts.append(
            f'[Document {i} | Source: {source_label}]\n{chunk["text"]}'
        )
    return '\n\n---\n\n'.join(parts)

context = format_context(filtered)
print(f'Context block: {len(context)} characters')

Шаг 5: создание расширенного запроса

Объедините блок контекста, системную инструкцию и вопрос пользователя в итоговый запрос. Системное сообщение сообщает модели, что нужно использовать только предоставленный контекст и указывать источники. Сообщение пользователя содержит отформатированный контекст, за которым следует вопрос. Такое чёткое разделение не позволяет модели смешивать содержимое контекста с вопросом и однозначно обозначает границу между извлечёнными данными и вводом пользователя.

def build_prompt(question, context):
    system_message = (
        'You are a helpful assistant. Answer the question using ONLY '
        'the information in the provided documents. '
        'Cite the document number(s) used, like [Doc 1]. '
        'If the documents do not contain the answer, say so.'
    )
    user_message = (
        f'Documents:\n\n{context}\n\n'
        f'Question: {question}'
    )
    return system_message, user_message

Шаг 6: вызов LLM и получение ответа

Отправьте подготовленный запрос в LLM с помощью API Chat Completions. Для фактических вопросов и ответов используйте низкую температуру (от 0,0 до 0,3), чтобы получать стабильные ответы, основанные на контексте. Более высокая температура делает ответы творческими, но повышает риск того, что модель добавит сведения, отсутствующие в контексте. Обработайте ответ и верните как текст ответа, так и извлечённые источники, чтобы приложение могло отобразить пользователю цитаты.

def generate_answer(question, context, sources):
    system_msg, user_msg = build_prompt(question, context)

    response = client.chat.completions.create(
        model='gpt-4o',
        temperature=0.1,   # low temperature for factual Q&A
        messages=[
            {'role': 'system', 'content': system_msg},
            {'role': 'user', 'content': user_msg}
        ]
    )
    answer = response.choices[0].message.content
    return {
        'answer': answer,
        'sources': sources,
        'tokens_used': response.usage.total_tokens
    }

Объединение всех компонентов

Полный конвейер запросов последовательно вызывает эти шаги. Каждый шаг является чистой функцией, которую можно тестировать независимо, а данные последовательно переходят от одного шага к следующему. Добавление ведения журналов на каждом шаге делает конвейер наблюдаемым: Вы можете точно увидеть, какие фрагменты были извлечены, какие оценки они получили, как был собран контекст и сколько токенов использовано. Такая видимость необходима для отладки и повышения качества извлечения.

def answer_question(user_question, vector_index):
    # Step 1: Embed query
    q_vector = embed_query(user_question)

    # Step 2: Retrieve
    chunks = retrieve_chunks(q_vector, vector_index, top_k=5)

    # Step 3: Filter low-confidence matches
    chunks = filter_by_score(chunks, threshold=0.70)
    if not chunks:
        return {'answer': 'I do not have information about that topic.', 'sources': []}

    # Step 4 & 5: Format and build prompt
    context = format_context(chunks)
    sources = [c['source'] for c in chunks]

    # Step 6: Generate
    return generate_answer(user_question, context, sources)

Оптимизация задержки

В конвейере запросов есть два шага, ограниченных операциями ввода-вывода: вызов для создания эмбеддинга и вызов LLM. Выполняйте их без лишних ожиданий: вызов для создания эмбеддинга быстрый (<100 мс), а вызов LLM медленный (500 мс–3 с). Чтобы уменьшить воспринимаемую задержку, передавайте ответ LLM потоково, чтобы токены появлялись по мере генерации, а не после ожидания полного ответа. Кэшируйте эмбеддинги повторяющихся идентичных запросов, чтобы избежать лишних вызовов API.

async def answer_question_streaming(question, index):
    q_vector = embed_query(question)
    chunks = retrieve_chunks(q_vector, index, top_k=5)
    chunks = filter_by_score(chunks)
    if not chunks:
        yield 'I do not have information about that topic.'
        return
    context = format_context(chunks)
    system_msg, user_msg = build_prompt(question, context)

    stream = await client.chat.completions.create(
        model='gpt-4o',
        stream=True,
        messages=[
            {'role': 'system', 'content': system_msg},
            {'role': 'user', 'content': user_msg}
        ]
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content or ''
        yield delta

Ведение журналов для наблюдаемости

Промышленные конвейеры RAG нуждаются в структурированном ведении журналов, чтобы Вы могли диагностировать сбои извлечения или плохие ответы LLM. Для каждого запроса записывайте запрос, ID и оценки извлечённых фрагментов, количество токенов контекста, ответ и задержку. Храните эти журналы в базе данных или на платформе наблюдаемости. Когда пользователи сообщают о плохих ответах, Вы сможете воспроизвести точный запрос и проверить, какие фрагменты были извлечены и почему их оказалось недостаточно.

import time
import logging
import json

def answer_question_with_logging(question, index):
    start = time.time()
    q_vector = embed_query(question)
    chunks = retrieve_chunks(q_vector, index, top_k=5)
    chunks = filter_by_score(chunks)
    context = format_context(chunks)
    result = generate_answer(question, context, [c['source'] for c in chunks])
    latency_ms = (time.time() - start) * 1000
    log_entry = {
        'question': question,
        'num_chunks_retrieved': len(chunks),
        'chunk_scores': [c['score'] for c in chunks],
        'tokens_used': result.get('tokens_used'),
        'latency_ms': round(latency_ms)
    }
    logging.info(json.dumps(log_entry))
    return result

Кэширование эмбеддингов запросов

Если Ваше приложение получает много повторяющихся или почти идентичных запросов, например в ботах FAQ, где пользователи часто задают одни и те же вопросы, кэширование эмбеддингов запросов становится простой и очень эффективной оптимизацией. Получите хеш строки запроса, проверьте в кэше Redis соответствующий эмбеддинг и вызывайте API эмбеддингов только при отсутствии записи в кэше. В промышленных FAQ и чат-ботах поддержки доля попаданий в кэш эмбеддингов часто составляет 30–60 %, что существенно снижает расходы на API и уменьшает задержку на 50–100 мс для каждого запроса, обработанного из кэша.

import hashlib
import json
import redis

r = redis.Redis(host='localhost', port=6379)
EMBED_CACHE_TTL = 86400  # 24 hours

def embed_query_cached(question):
    cache_key = 'embed:' + hashlib.sha256(question.encode()).hexdigest()
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)  # cache hit
    # Cache miss: call the API
    vector = embed_query(question)
    r.setex(cache_key, EMBED_CACHE_TTL, json.dumps(vector))
    return vector

Быстрая проверка

Проверьте, насколько хорошо Вы усвоили концепции AI Engineering из этого урока.

Итоги урока

В этом уроке Вы узнали о шестишаговом конвейере запросов (создание эмбеддинга запроса, извлечение фрагментов, фильтрация по оценке, форматирование контекста, создание запроса, генерация ответа), фильтрации по порогу оценки для обработки запросов, выходящих за пределы охвата индекса, а также об улучшениях для промышленной эксплуатации, включая потоковую передачу ответов, структурированное ведение журналов и оптимизацию задержки. Далее мы научимся оценивать, действительно ли Ваша завершённая система RAG работает правильно.

Можно начать бесплатно

Изучай Python с ИИ-репетитором — бесплатно

Пиши и запускай код прямо в браузере, получай мгновенную помощь от ИИ-репетитора 24/7 и продолжи учиться на сайте или в приложении.

Курсы
30
Уроки
120

Часто задаваемые вопросы

Урок «Запрос, поиск и генерация» бесплатный?

Да — полный текст урока «Запрос, поиск и генерация» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.

Чему я научусь в уроке «Запрос, поиск и генерация»?

Напишите конвейер обработки запроса, который создаёт эмбеддинг вопроса пользователя, находит k лучших фрагментов, формирует дополненный запрос, обращается к LLM и возвращает ответ с цитатами. Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Engineering Academy?

Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.

Сколько времени занимает урок «Запрос, поиск и генерация»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Engineering Academy?

Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. Загрузка документов и извлечение текста
  2. Стратегии разбиения: фиксированные фрагменты, предложения и рекурсия
  3. Индексирование: создание и сохранение фрагментов
  4. Запрос, поиск и генерация
← Назад к AI Engineering Academy