Создаём RAG-систему: поиск контекста в векторной базе + генерация ответа через LLM. Интеграция с OpenAI GPT.
Создаём RAG-систему: поиск контекста в векторной базе + генерация ответа через LLM. Интеграция с OpenAI GPT.
За 60 минут вы:
RAG (Retrieval-Augmented Generation) — архитектура, сочетающая:
Вопрос → Поиск контекста в БД → Контекст + Вопрос → LLM → Ответ
pip install openai faiss-cpu sentence-transformersfrom openai import OpenAI
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
class RAGService:
"""RAG-сервис: поиск + генерация ответа."""
def __init__(self, index, documents, embeddings, api_key: str):
self.index = index
self.documents = documents
self.embeddings = embeddings
self.llm = OpenAI(api_key=api_key)
self.model = SentenceTransformer('all-MiniLM-L6-v2')
def search_context(self, query: str, k: int = 3) -> list[dict]:
"""Поиск релевантного контекста."""
query_embedding = self.model.encode([query])
faiss.normalize_L2(query_embedding)
distances, indices = self.index.search(query_embedding, k)
context = []
for idx, dist in zip(indices[0], distances[0]):
context.append({
"text": self.documents[idx],
"score": float(dist)
})
return context
def generate_answer(self, query: str, context: list[dict]) -> str:
"""Генерация ответа через LLM."""
context_text = "\n\n".join([c["text"] for c in context])
prompt = f"""Контекст:
{context_text}
Вопрос: {query}
Ответь на вопрос, используя только информацию из контекста."""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Ты полезный ассистент."},
{"role": "user", "content": prompt}
]
)
return response.choices[0].message.content
def query(self, query: str, k: int = 3) -> dict:
"""Полный RAG-запрос."""
context = self.search_context(query, k)
answer = self.generate_answer(query, context)
return {
"answer": answer,
"context": context
}
# Использование
if __name__ == "__main__":
import os
# Документы
documents = [
"Python — язык программирования общего назначения",
"Django — веб-фреймворк на Python для быстрой разработки",
"Flask — микрофреймворк для создания веб-приложений",
"FastAPI — современный фреймворк для создания API"
]
# Индекс
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(documents)
faiss.normalize_L2(embeddings)
index = faiss.IndexFlatIP(embeddings.shape[1])
index.add(embeddings)
# RAG сервис
rag = RAGService(
index=index,
documents=documents,
embeddings=embeddings,
api_key=os.getenv("OPENAI_API_KEY")
)
# Запрос
result = rag.query("Какой фреймворк выбрать для API?")
print(f"Вопрос: Какой фреймворк выбрать для API?\n")
print(f"Ответ: {result['answer']}\n")
print("Контекст:")
for c in result['context']:
print(f" [{c['score']:.3f}] {c['text']}")def generate_answer_with_sources(self, query: str, context: list[dict]) -> dict:
"""Генерация ответа с цитированием источников."""
context_text = "\n\n".join([
f"[{i+1}] {c['text']}"
for i, c in enumerate(context)
])
prompt = f"""Контекст:
{context_text}
Вопрос: {query}
Ответь на вопрос, используя информацию из контекста.
В конце укажи, какие источники использовал (номера [1], [2], etc.)."""
response = self.llm.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Ты полезный ассистент."},
{"role": "user", "content": prompt}
]
)
return {
"answer": response.choices[0].message.content,
"sources": context
}def query_with_relevance_check(self, query: str, k: int = 3, threshold: float = 0.5):
"""RAG с проверкой релевантности контекста."""
context = self.search_context(query, k)
# Проверка релевантности
relevant_context = [c for c in context if c["score"] >= threshold]
if not relevant_context:
return {
"answer": "Извините, у меня недостаточно информации для ответа.",
"context": [],
"relevant": False
}
answer = self.generate_answer(query, relevant_context)
return {
"answer": answer,
"context": relevant_context,
"relevant": True
}Следующая тема — Production-паттерны.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.