ค้นคำค้น คืนข้อมูล และสร้างคำตอบ

ขั้นตอนการค้นหา: ตั้งแต่ต้นจนจบ

ขั้นตอนที่ 1: สร้างเวกเตอร์ฝังตัวจากคำค้นของผู้ใช้

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 อันดับแรก

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: กรองตามเกณฑ์คะแนน

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: จัดรูปแบบบล็อกบริบท

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 และรับคำตอบ

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)

การเพิ่มประสิทธิภาพเวลาแฝง

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

การบันทึกเหตุการณ์เพื่อการสังเกตการทำงาน

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

การแคชเวกเตอร์ฝังตัวของคำค้น

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

ตรวจสอบความเข้าใจ

สรุปบทเรียน