Запрос, поиск и генерация
Напишите конвейер обработки запроса, который создаёт эмбеддинг вопроса пользователя, находит k лучших фрагментов, формирует дополненный запрос, обращается к LLM и возвращает ответ с цитатами.
«Запрос, поиск и генерация» — бесплатный урок 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 — локальная установка не требуется.
Все уроки этого курса
- Загрузка документов и извлечение текста
- Стратегии разбиения: фиксированные фрагменты, предложения и рекурсия
- Индексирование: создание и сохранение фрагментов
- Запрос, поиск и генерация