Implementando RAG em um aplicativo Django: um guia simples
O objetivo deste post é implementar alguns dos principais usos atuais de large language models (LLMs) em um aplicativo Django comum, especificamente a técnica chamada Retrieval-Augmented Generation (RAG), que permite que os usuários conversem com seus documentos. Antes de mergulhar na implementação, preciso explicar alguns dos blocos de construção básicos do RAG, ou seja, embeddings.
Embeddings
Embeddings são essencialmente vetores multidimensionais de números de ponto flutuante com um tamanho fixo que contêm informações de alguma forma representando o conteúdo e o contexto.
[[0.011513561010360718,
-0.02314218506217003,
-0.0171588733792305,
-0.03912165388464928,
-0.021027889102697372,
0.02629205398261547,
-0.03854633495211601,
...]]
Embedding truncado da palavra "Amazon"
Um aspecto interessante dos embeddings é que, ao armazená-los em um banco de dados, podemos realizar operações entre eles para medir a similaridade entre os conteúdos. Por exemplo, ao converter o termo 'Amazon' em um embedding e compará-lo com os embeddings de 'Brazil' e 'Food' no seu espaço vetorial, a distância entre 'Amazon' e 'Brazil' é menor do que entre 'Amazon' e 'Food', indicando que 'Amazon' e 'Brazil' são mais similares.
from langchain_openai import OpenAIEmbeddings
from pgvector.django import CosineDistance
from project.app.models import MyDocument
embeddings_function = OpenAIEmbeddings()
embedding1 = embeddings_function.embed_documents(["Brazil"])[0]
embedding2 = embeddings_function.embed_documents(["Food"])[0]
MyDocument.objects.annotate(distance=CosineDistance("embedding", embedding1)).order_by("distance")[0].distance
0.17931762111467797
MyDocument.objects.annotate(distance=CosineDistance("embedding", embedding2)).order_by("distance")[0].distance
0.1892012486575828
Isso é chamado de busca de similaridade semântica e será responsável pela parte de "retrieval" do RAG. Em essência, armazenamos cada documento como um embedding e então o recuperamos por meio de busca de similaridade mais tarde. No entanto, para armazenar esses embeddings e executar a recuperação, precisamos de um banco de dados vetorial; para isso, usaremos o pgvector.
Pgvector
Atualmente, há vários bancos de dados de vetores diferentes sendo usados, como Chroma ou LanceDB. Como um dos bancos de dados mais comuns usados em aplicativos Django é o Postgres, em vez de adicionar um novo banco de dados ao seu aplicativo atual, uma boa alternativa é usar o pgvector. O Pgvector é uma extensão do Postgres que fornece os recursos de um banco de dados de vetores, permitindo que você armazene embeddings e execute uma pesquisa de similaridade semântica. Para instalar o pgvector, você deve seguir estas instruções e, em seguida, executar este comando dentro da sua instância do Postgres:
CREATE EXTENSION vector;
e então, instale o módulo python pgvector
pip install pgvector
Depois disso, já é possível armazenar os embeddings. Porém, em vez de fazer isso manualmente, vamos salvá-los usando modelos do Django. Para salvar os embeddings no banco de dados, é simples; basta criar ou atualizar um modelo e adicionar um campo chamado VectorField com o número de dimensões que seu embedding terá, dessa maneira:
from Django.db import models
from pgvector.django import VectorField
class MyDocument(models.Model):
# ...
embedding = VectorField(dimensions=1536) # Supondo que estamos usando o modelo de embedding text-embedding-ada-002.
Depois disso, basta criar e executar a migração do banco de dados usando os seguintes comandos:
python manage.py makemigrations
python manage.py migrate
Agora, com a capacidade de armazenar embeddings usando o ORM do Django, podemos aproveitar o pgvector para armazenar e calcular a similaridade dos embeddings no banco de dados Postgres.
No entanto, antes de fazermos isso, precisamos converter o conteúdo dos documentos em embeddings em primeiro lugar. Para esta tarefa, usaremos o Langchain.
Langchain
Langchain é um framework que nos dá ferramentas para construir aplicações web que podem interagir com LLMs, incluindo módulos e classes que podem gerar embeddings. Um deles é OpenAIEmbeddings, que usa um dos modelos do OpenAI para gerar os embeddings. Para este exemplo, usaremos o modelo padrão (text-embedding-ada-002) que gera um vetor com 1536 dimensões.
pip install langchain langchain-openai
from langchain_openai import OpenAIEmbeddings
embeddings_function = OpenAIEmbeddings()
# Substitua 'example text' pelo seu conteúdo real
embeddings = embeddings_function.embed_documents(['example text'])
Agora que configuramos nosso pipeline de geração de embeddings usando o Langchain, precisamos carregar os documentos.
Para fazer isso, usaremos um dos carregadores de PDF fornecidos pela Langchain, que é o PyMuPDF. O carregador lerá o conteúdo do PDF e gerará uma lista de documents (instâncias da classe Document do langchain) que podemos então converter em embeddings.
pip install PyMuPDF
from langchain_community.document_loaders import PyMuPDFLoader
from project.app import MyDocument
file_path = "example.pdf"
loader = PyMuPDFLoader(file_path)
documents = loader.load()
O retorno de loader.load() vai ser esse:
[Document(metadata={'source': 'example.pdf', 'file_path': 'example.pdf', 'page': 0, 'total_pages': 1, 'format': 'PDF 2.0', 'title': '', 'author': '', 'subject': '', 'keywords': '', 'creator': '', 'producer': '', 'creationDate': 'D:20241009145551Z', 'modDate': 'D:20241009145551Z', 'trapped': ''}, page_content='Just an example \n')]
Agora, nós podemos iterar a lista de documents para extrair o page_content e gerar os embeddings:
documents_content = [document.page_content for document in documents]
embeddings = embeddings_function.embed_documents(documents_content)
E então podemos salvar os embeddings no modelo dessa maneira.
for embedding, document in zip(embeddings, documents):
MyDocument.objects.create(
embedding=embedding,
source=file_path,
content=document.page_content
)
Como podemos ver, adicionamos novos campos, source e content, ao modelo MyDocument. Isso nos permitirá recuperar a URL original e o conteúdo da fonte de cada documento mais tarde.
Após preencher o banco de dados com os embeddings, podemos começar a implementar o principal recurso deste aplicativo: o RAG.
RAG
A técnica RAG envolve injetar o conteúdo de documentos no contexto dos modelos LLM. No entanto, devido às limitações no número de tokens que podem ser incluídos no contexto, é necessário recuperar apenas os documentos mais relevantes em relação ao input do usuário — essa é a fase de recuperação do RAG.
Como estamos criando um aplicativo de chatbot, criaremos um RAG conversacional. Para fazer isso, usaremos os módulos langchain apropriados para desenvolver nossa função.
from langchain_openai import ChatOpenAI
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.chains.retrieval import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.pydantic_v1 import Field
from django.conf import settings
from project.app.retrievers import DocumentRetriever
class InMemoryHistory(BaseChatMessageHistory, BaseModel):
"""Implementação em memória do histórico de mensagens de chat."""
messages: list[BaseMessage] = Field(default_factory=list)
def add_message(self, message: BaseMessage) -> None:
self.messages.append(message)
def clear(self) -> None:
self.messages = []
store = {}
def conversational_rag(
condense_question_prompt_template,
system_prompt_template,
):
llm = ChatOpenAI(api_key=settings.OPENAI_API_KEY)
condense_question_prompt = ChatPromptTemplate.from_messages(
[
("system", condense_question_prompt_template),
("placeholder", "{chat_history}"),
("human", "{input}"),
],
)
history_aware_retriever = create_history_aware_retriever(
llm,
DocumentRetriever(max_results=4),
condense_question_prompt,
)
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt_template),
("placeholder", "{chat_history}"),
("human", "{input}"),
],
)
qa_chain = create_stuff_documents_chain(
llm=llm,
prompt=qa_prompt,
)
convo_qa_chain = create_retrieval_chain(history_aware_retriever, qa_chain)
def get_session_history(session_id):
if session_id not in store:
store[session_id] = InMemoryHistory()
return store[session_id]
return RunnableWithMessageHistory(
runnable=convo_qa_chain,
get_session_history=get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
output_messages_key="answer",
)
Aqui podemos ver que estamos recebendo dois parâmetros para prompting: condense_question_prompt_template e system_prompt_template. O primeiro é um prompt usado para condensar ou reformular a pergunta original do usuário, obtido por meio de uma chamada à LLM.
condense_question_prompt_template = """
Dado um histórico da conversa e a última pergunta do usuário
que pode fazer referência ao contexto no histórico da conversa,
formule uma pergunta independente que possa ser entendida
sem o histórico da conversa. NÃO responda à pergunta,
apenas reformule-a se necessário e, caso contrário, retorne-a como está.
"""
Já o segundo, define como o modelo deve responder as perguntas com base no contexto (os documentos recuperados).
system_prompt_template = """
Você é um assistente para responder perguntas.
Você deve responder com base no contexto fornecido e no histórico da conversa.
Se você não tiver nenhum contexto, apenas diga "Não sei".
Contexto: {context}
"""
O uso do condense_question_prompt_template é necessário em casos em que o input não tem um termo específico para um assunto, como por exemplo, "Dê-me mais informações sobre isso". Sem saber o contexto, é impossível determinar a que "isso" se refere, tornando mais difícil encontrar documentos semelhantes no banco de dados. Para resolver esse problema, o LLM usa o contexto da conversa atual para reformular o input do usuário e recuperar documentos relevantes.
Este prompt é usado na chamada de history_aware_retriever, que recebe uma instância de DocumentRetriever.
history_aware_retriever = create_history_aware_retriever(
llm,
DocumentRetriever(max_results=4),
condense_question_prompt,
)
O retriever é implementado desta forma:
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from pgvector.django import CosineDistance
from langchain_openai import OpenAIEmbeddings
from django.conf import settings
from project.app.models import MyDocument
class DocumentRetriever(BaseRetriever):
max_results: int
class Config:
arbitrary_types_allowed = True
def _get_relevant_documents(
self,
query: str,
*,
run_manager: CallbackManagerForRetrieverRun,
) -> list[Document]:
embeddings_function = OpenAIEmbeddings(api_key=settings.OPENAI_API_KEY)
embeddings = embeddings_function.embed_documents([query])
documents = MyDocument.objects.annotate(
distance=CosineDistance("embedding", embeddings[0])
).order_by("distance")
return [
Document(
page_content=document.content,
)
for document in documents[:self.max_results]
]
Esta classe define como recuperar documentos e define os parâmetros para o objeto Document, que é então chamado internamente pelo history_aware_retriever.
Nesta implementação, um queryset é criado utilizando a métrica CosineDistance para calcular a distância (ou similaridade) entre o embedding da pergunta e os embeddings armazenados no banco de dados. Os resultados são ordenados com base no valor da distância, onde as menores distâncias indicam maior similaridade entre os documentos. (Observação: para entender melhor por que a distância do cosseno é usada, consulte este artigo.)
O outro parâmetro, max_results, determina quantos documentos são recuperados da lista e, neste caso, estamos usando os 4 principais documentos.
Após definir o history_aware_retriever, precisamos definir a funcão responsável por de fato responder ao "input" do usuário, o callable qa_chain que utiliza o parâmetro system_prompt_template.
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt_template),
("placeholder", "{chat_history}"),
("human", "{input}"),
],
)
qa_chain = create_stuff_documents_chain(
llm=llm,
prompt=qa_prompt,
)
Em seguida, vamos combinar ambos os callables em um único, chamado convo_qa_chain, da seguinte forma:
convo_qa_chain = create_retrieval_chain(history_aware_retriever, qa_chain)
Na parte final, utilizamos convo_qa_chain como argumento para a classe RunnableWithMessageHistory. Esta classe é responsável por adicionar um histórico da conversa ao contexto do LLM.
RunnableWithMessageHistory(
runnable=convo_qa_chain,
get_session_history=get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
output_messages_key="answer",
)
O segundo parâmetro, get_session_history, recebe uma função responsável por recuperar o histórico da conversa com base no session_id, que será passado posteriormente quando a função conversation_rag for chamada.
def get_session_history(session_id):
if session_id not in store:
store[session_id] = InMemoryHistory()
return store[session_id]
Neste exemplo, estamos armazenando a conversa na memória, mas em aplicações de produção, é recomendado usar um banco de dados como o Redis.
O parâmetro history_messages_key, define o nome da variável que contém o histórico da conversa e será usada no template. Os parâmetros input_messages_key e output_messages_key correspondem à variável de "input" do template (a pergunta do usuário) e a chave do dict com a resposta gerada pelo LLM.
Com a instância pronta, agora podemos retornar e mostrar como utilizar tudo isso em uma view.
View
Na parte final, definimos uma view simples que recebe uma solicitação POST com a pergunta do usuário e passa o session_id como argumento.
def ask_ai(request):
user_question = request.POST.get("question")
chat_session_id = request.POST.get("chat_session_id")
if not user_question or not chat_session_id:
return HttpResponseBadRequest()
condense_question_prompt_template = """
Dado um histórico de bate-papo e a última pergunta do usuário
que pode fazer referência ao contexto no histórico de bate-papo,
formule uma pergunta independente que possa ser entendida
sem o histórico de bate-papo. NÃO responda à pergunta,
apenas reformule-a se necessário e, caso contrário, retorne-a como está.
"""
system_prompt_template = """
Você é um assistente para responder perguntas.
Você deve responder com base no contexto fornecido e no histórico da conversa.
Se você não tiver nenhum contexto, apenas diga "Não sei".
Contexto: {context}
"""
async def message_stream():
data = {
"input": user_question,
}
rag_chain = conversational_rag(
condense_question_system_template,
system_prompt_template,
)
async for chunk in rag_chain.astream(
data,
{"configurable": {"session_id": chat_session_id}},
):
yield chunk.get("answer", "")
response = StreamingHttpResponse(message_stream(), content_type="text/event-stream")
response["Cache-Control"] = "no-cache"
response["X-Accel-Buffering"] = "no"
return response
Nesta view, estamos retornando o resultado como um StreamingHttpResponse, que melhora a experiência do usuário ao permitir o envio de partes da resposta à medida que são geradas, em vez de aguardar a conclusão total do processamento. Esse comportamento é semelhante ao funcionamento do ChatGPT.
Para atingir esse efeito, precisamos passar uma função que retorna um gerador. A função message_stream é responsável por chamar a função conversational_rag, que retorna uma instância de RunnableWithMessageHistory que definimos anteriormente. Em seguida, chamamos o método astream nesta instância, que retorna a resposta em pedaços.
Com tudo isso definido, agora podemos escrever uma função JavaScript para consumir a resposta no frontend e atingir o efeito desejado.
async function askChatbot(question) {
const formData = new FormData();
formData.append("question", question);
formData.append("chat_session_id", `{{ chat_session_id }}`);
const response = await fetch(`{% url "app:ask_ai" %}`, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
'X-CSRFToken': `{{ csrf_token }}`
},
body: new URLSearchParams(formData)
});
if (!response.ok) {
throw new Error('Network response was not ok');
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let answer = '';
while (true) {
const {
done,
value
} = await reader.read();
if (done) break;
let messageChunk = decoder.decode(value, {
stream: true
});
// Concatena os pedaços da resposta
answer += messageChunk;
}
}
RAG usando o arquivo do livro "The pragmatic programmer"
Conclusão
A arquitetura do aplicativo que construímos é relativamente simples e fornece uma boa base para desenvolvimento futuro. No entanto, ainda há muito espaço para melhorias, e esta implementação não cobre a construção do frontend.
Dito isso, seguindo estas etapas, você deve ser capaz de adicionar alguns recursos interessantes a um aplicativo Django existente, incluindo interfaces de conversação, processamento de linguagem natural e recursos de interação do usuário mais avançados.
