0Pricing
AI Engineering Academy · Lekcja

Zapytanie, pobieranie i generowanie

Uczestnicy napiszą potok zapytań, który utworzy embedding pytania użytkownika, pobierze fragmenty top-k, sformatuje rozszerzony prompt, wywoła LLM i zwróci odpowiedź z cytowaniami.

Zapytanie, pobieranie i generowanie to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Potok zapytań: od początku do końca

Potok zapytań to część online systemu RAG — kod uruchamiany w czasie rzeczywistym, gdy użytkownik zadaje pytanie. Łączy wszystkie komponenty zbudowane podczas indeksowania: model embeddingowy, magazyn wektorowy, szablon promptu i LLM. Dobrze zaimplementowany potok zapytań kończy działanie w czasie krótszym niż 500 ms dla większości obciążeń i generuje odpowiedzi oparte na źródłach, wraz z cytowaniami. W tej lekcji zbudujemy od podstaw każdy jego etap.

Krok 1: Wygenerowanie embeddingu zapytania użytkownika

Pierwszym krokiem jest przekształcenie pytania użytkownika zapisanego w języku naturalnym w embedding wektorowy za pomocą tego samego modelu, którego użyto podczas indeksowania. Embedding koduje znaczenie semantyczne pytania i zostanie porównany z embeddingami fragmentów dokumentów w magazynie wektorowym. Ten krok powinien działać szybko — należy użyć lekkiego modelu, takiego jak text-embedding-3-small, oraz buforować embeddingi dla identycznych, powtarzających się zapytań.

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')

Krok 2: Pobranie fragmentów Top-K

Wyślij wektor zapytania do magazynu wektorowego, aby znaleźć K najbardziej podobnych semantycznie fragmentów. Zwrócone wyniki są uporządkowane według wyniku podobieństwa cosinusowego (zwykle od 0,0 do 1,0; im wyższy wynik, tym lepiej). Idealna wartość K równoważy bogactwo kontekstu z kosztem wykorzystania okna kontekstowego — K=5 to częsty punkt wyjścia. Można również zastosować tutaj filtry metadanych, aby ograniczyć wyszukiwanie do określonego działu, typu dokumentu lub zakresu dat.

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

Krok 3: Filtrowanie według progu wyniku

Nie wszystkie pobrane fragmenty są rzeczywiście istotne — niektóre mogą mieć niski wynik podobieństwa, ale nadal znajdować się wśród Top-K, ponieważ zapytanie wykracza poza zakres danych objętych indeksem. Zastosuj minimalny próg wyniku, aby odfiltrować dopasowania o niskiej wiarygodności. Jeśli wszystkie pobrane fragmenty znajdą się poniżej tego progu, zwróć odpowiedź „nie znaleziono informacji” zamiast przekazywać do LLM nieistotny kontekst, który doprowadziłby do gorszej odpowiedzi niż uprzejma odmowa.

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')

Krok 4: Sformatowanie bloku kontekstu

Zbierz pobrane fragmenty w ustrukturyzowany blok kontekstu, który odczyta LLM. Oznacz każdy fragment jego źródłem, aby model mógł je dokładnie cytować. Dla czytelności dodaj separator między fragmentami. Łączny kontekst powinien mieścić się w dostępnym budżecie tokenów — policz tokeny za pomocą tiktoken, a jeśli przekroczysz limit, skróć fragmenty lub usuń te o niższych wynikach. Blok kontekstu jest wstawiany do promptu między instrukcją systemową a pytaniem użytkownika.

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')

Krok 5: Zbudowanie rozszerzonego promptu

Połącz blok kontekstu, instrukcję systemową i pytanie użytkownika w końcowy prompt. Wiadomość systemowa nakazuje modelowi korzystać wyłącznie z dostarczonego kontekstu i cytować źródła. Wiadomość użytkownika zawiera sformatowany kontekst, a następnie pytanie. Taki wyraźny podział zapobiega mieszaniu przez model treści kontekstu z pytaniem i jednoznacznie wyznacza granicę między pobranymi danymi a danymi wejściowymi użytkownika.

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

Krok 6: Wywołanie LLM i pobranie odpowiedzi

Wyślij złożony prompt do LLM za pomocą Chat Completions API. W przypadku rzeczowych pytań i odpowiedzi użyj niskiej temperatury (od 0,0 do 0,3), aby uzyskać spójne odpowiedzi oparte na źródłach. Wyższa temperatura daje bardziej kreatywne odpowiedzi, ale zwiększa ryzyko dodania przez model informacji wykraczających poza kontekst. Przetwórz odpowiedź i zwróć zarówno jej treść, jak i pobrane źródła, aby aplikacja mogła wyświetlić użytkownikowi cytowania.

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
    }

Połączenie wszystkich elementów

Kompletny potok zapytań wykonuje te kroki sekwencyjnie. Każdy krok jest funkcją czystą, którą można niezależnie testować, a dane przepływają w uporządkowany sposób z jednego kroku do następnego. Dodanie logowania na każdym etapie zapewnia obserwowalność potoku — można dokładnie sprawdzić, które fragmenty pobrano, jakie miały wyniki, jak złożono kontekst i ile tokenów wykorzystano. Ta widoczność jest niezbędna do debugowania i poprawiania jakości wyszukiwania.

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)

Optymalizacja opóźnień

Potok zapytań zawiera dwa kroki zależne od operacji wejścia-wyjścia: wywołanie embeddingu i wywołanie LLM. Wykonuj je bez zbędnego oczekiwania: wywołanie embeddingu jest szybkie (<100 ms), natomiast wywołanie LLM trwa długo (500 ms–3 s). Aby zmniejszyć odczuwalne opóźnienie, strumieniuj odpowiedź LLM, dzięki czemu tokeny będą pojawiać się w miarę ich generowania, zamiast czekać na całą odpowiedź. Buforuj embeddingi identycznych, powtarzających się zapytań, aby uniknąć zbędnych wywołań 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

Logowanie na potrzeby obserwowalności

Produkcyjne potoki RAG wymagają ustrukturyzowanego logowania, aby można było diagnozować błędy wyszukiwania lub niepoprawne odpowiedzi LLM. Dla każdego żądania rejestruj zapytanie, identyfikatory i wyniki pobranych fragmentów, liczbę tokenów kontekstu, odpowiedź oraz opóźnienie. Przechowuj te logi w bazie danych lub na platformie obserwowalności. Gdy użytkownicy zgłoszą niepoprawne odpowiedzi, będzie można odtworzyć dokładne zapytanie i sprawdzić, które fragmenty pobrano oraz dlaczego okazały się niewystarczające.

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

Buforowanie embeddingów zapytań

Jeśli aplikacja otrzymuje wiele powtarzających się lub niemal identycznych zapytań — na przykład w botach FAQ, w których użytkownicy często zadają te same pytania — buforowanie embeddingów zapytań jest prostą i bardzo skuteczną optymalizacją. Oblicz skrót ciągu zapytania, sprawdź w pamięci podręcznej Redis odpowiadający mu embedding i wywołuj API embeddingów tylko w przypadku braku wpisu w pamięci podręcznej. W produkcyjnych botach FAQ i chatbotach wsparcia często osiąga się współczynnik trafień pamięci podręcznej na poziomie 30–60%, co eliminuje znaczną część kosztów API i zmniejsza opóźnienie o 50–100 ms dla każdego zapytania obsłużonego z pamięci podręcznej.

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

Szybki sprawdzian

Sprawdź swoją wiedzę na temat koncepcji inżynierii AI omówionych w tej lekcji.

Podsumowanie lekcji

W tej lekcji poznano: sześcioetapowy potok zapytań (wygenerowanie embeddingu zapytania, pobranie fragmentów, filtrowanie według wyniku, sformatowanie kontekstu, zbudowanie promptu, wygenerowanie odpowiedzi), filtrowanie według progu wyniku w celu obsługi zapytań wykraczających poza zakres indeksu oraz usprawnienia produkcyjne, w tym strumieniowanie odpowiedzi, ustrukturyzowane logowanie i optymalizację opóźnień. Następnie nauczymy się oceniać, czy kompletny system RAG rzeczywiście działa poprawnie.

Często zadawane pytania

Czy lekcja „Zapytanie, pobieranie i generowanie” jest bezpłatna?

Tak — pełny tekst „Zapytanie, pobieranie i generowanie” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „Zapytanie, pobieranie i generowanie”?

Uczestnicy napiszą potok zapytań, który utworzy embedding pytania użytkownika, pobierze fragmenty top-k, sformatuje rozszerzony prompt, wywoła LLM i zwróci odpowiedź z cytowaniami. Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?

Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.

Ile czasu zajmuje lekcja „Zapytanie, pobieranie i generowanie”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?

Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Wczytywanie dokumentów i ekstrakcja tekstu
  2. Strategie dzielenia tekstu: stały rozmiar, zdania i rekurencja
  3. Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów
  4. Zapytanie, pobieranie i generowanie
← Powrót do AI Engineering Academy