AI: Geração Aumentada por Recuperação (RAG)

from openai import OpenAI

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

text = "O subsídio de viagem da empresa é de R$ 45 por dia."
vector = get_embedding(text)

print(f"Dimensão: {len(vector)}")
print(f"Primeiros 5 valores: {vector[:5]}")

Dimensão: 1536
Primeiros 5 valores: [0.0123, -0.0045, 0.0378, -0.0211, 0.0056]

import numpy as np

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr = np.array(a)
    b_arr = np.array(b)
    dot = np.dot(a_arr, b_arr)
    norm_a = np.linalg.norm(a_arr)
    norm_b = np.linalg.norm(b_arr)
    return float(dot / (norm_a * norm_b))

sentences = {
    "travel_allowance": get_embedding("A diária de viagem da empresa é de US$ 45 por dia."),
    "reimbursement": get_embedding("Como solicito o reembolso de viagem?"),
    "unrelated": get_embedding("O tempo está ensolarado hoje."),
}

sim1 = cosine_similarity(sentences["travel_allowance"], sentences["reimbursement"])
sim2 = cosine_similarity(sentences["travel_allowance"], sentences["unrelated"])

print(f"Frases relacionadas:   {sim1:.4f}")
print(f"Frases não relacionadas: {sim2:.4f}")

Frases relacionadas:   0.8234
Frases não relacionadas: 0.3012

def top_k_search(
    query_embedding: list[float],
    doc_embeddings: list[list[float]],
    doc_texts: list[str],
    k: int = 3,
) -> list[tuple[str, float]]:
    similarities = [
        (doc_texts[i], cosine_similarity(query_embedding, doc_embeddings[i]))
        for i in range(len(doc_texts))
    ]
    similarities.sort(key=lambda x: x[1], reverse=True)
    return similarities[:k]

documents = [
    "A diária de viagem é de US$ 45 para viagens nacionais.",
    "Viagens internacionais requerem aprovação antecipada do VP.",
    "Funcionários devem apresentar recibos dentro de 30 dias.",
    "A política de trabalho remoto permite até 2 dias por semana em casa.",
    "A cozinha do escritório é abastecida com café e lanches gratuitos.",
]

doc_embeddings = [get_embedding(doc) for doc in documents]
query = "Quanto posso gastar com viagem por dia?"
query_embedding = get_embedding(query)

results = top_k_search(query_embedding, doc_embeddings, documents, k=3)

for text, score in results:
    print(f"[{score:.4f}] {text}")

[0.8912] A diária de viagem é de US$ 45 para viagens nacionais.
[0.7634] Funcionários devem apresentar recibos dentro de 30 dias.
[0.7102] Viagens internacionais requerem aprovação antecipada do VP.

from openai import OpenAI

client = OpenAI()

def rag_query(question: str, documents: list[str], k: int = 3) -> str:
    query_emb = get_embedding(question)
    doc_embs = [get_embedding(doc) for doc in documents]
    top_docs = top_k_search(query_emb, doc_embs, documents, k=k)

    context = "\n\n".join([f"[Doc {i+1}] {text}" for i, (text, _) in enumerate(top_docs)])

    prompt = f"""Responda à pergunta com base SOMENTE no seguinte contexto.
Se o contexto não contiver a resposta, diga "Eu não sei."

Contexto:
{context}

Pergunta: {question}
Resposta:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return response.choices[0].message.content

docs = [
    "A diária para viagens nacionais é de $45.",
    "Viagens internacionais requerem aprovação do VP com pelo menos 2 semanas de antecedência.",
    "Os funcionários devem enviar relatórios de despesas dentro de 30 dias após a viagem.",
    "O reembolso máximo de hotel é de $200 por noite.",
    "As passagens devem ser reservadas no portal de viagens da empresa.",
]

answer = rag_query("Qual é a diária para viagens?", docs)
print(answer)

A diária para viagens nacionais é de $45.

def ask_without_rag(question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        temperature=0,
    )
    return response.choices[0].message.content

question = "Qual é a diária de viagem da Acme Corp?"

print("=== Sem RAG ===")
print(ask_without_rag(question))

print("\n=== Com RAG ===")
print(rag_query(question, docs))

=== Sem RAG ===
Eu não tenho informações específicas sobre a diária de viagem da Acme Corp.
As diárias de viagem das empresas variam, mas uma faixa comum é de $50-$100 por dia.
Por favor, consulte a política de viagens da sua empresa para o valor exato.

=== Com RAG ===
A diária para viagens nacionais é de $45.

from langchain.text_splitter import RecursiveCharacterTextSplitter

text = """Acme Corp Travel Policy

1. Domestic Travel
The daily travel allowance is $45. This covers meals and incidental expenses.
Hotel reimbursement is capped at $200 per night.

2. International Travel
All international trips require VP approval at least 2 weeks in advance.
The daily allowance for international travel is $75.
Hotel reimbursement is capped at $300 per night.

3. Expense Reporting
All expense reports must be submitted within 30 days of trip completion.
Receipts are required for all expenses over $25."""

splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} ---")
    print(chunk)
    print()

--- Chunk 1 ---
Acme Corp Travel Policy

1. Domestic Travel
The daily travel allowance is $45. This covers meals and incidental expenses.
Hotel reimbursement is capped at $200 per night.

--- Chunk 2 ---
2. International Travel
All international trips require VP approval at least 2 weeks in advance.
The daily allowance for international travel is $75.

--- Chunk 3 ---
international travel is $75.
Hotel reimbursement is capped at $300 per night.

3. Expense Reporting
All expense reports must be submitted within 30 days of trip completion.

import numpy as np
from openai import OpenAI

client = OpenAI()

# --- Etapa 1: Preparar documentos ---
documents = [
    {
        "title": "Política de Viagens",
        "content": (
            "Política de Viagens da Acme Corp\n\n"
            "1. Viagens Domésticas\n"
            "O subsídio diário de viagem é de $45 para viagens domésticas. "
            "Isso cobre refeições e despesas incidentais. "
            "O reembolso de hotel é limitado a $200 por noite.\n\n"
            "2. Viagens Internacionais\n"
            "Todas as viagens internacionais requerem aprovação do VP com pelo menos 2 semanas de antecedência. "
            "O subsídio diário para viagens internacionais é de $75. "
            "O reembolso de hotel é limitado a $300 por noite."
        ),
    },
    {
        "title": "Política de Licenças",
        "content": (
            "Política de Licenças da Acme Corp\n\n"
            "1. Licença Anual\n"
            "Funcionários em tempo integral recebem 15 dias de licença anual remunerada por ano. "
            "Licenças não utilizadas podem ser transportadas para o próximo ano, até um máximo de 5 dias.\n\n"
            "2. Licença Médica\n"
            "Os funcionários recebem 10 dias de licença médica remunerada por ano. "
            "Um atestado médico é necessário para ausências que excedam 3 dias consecutivos."
        ),
    },
    {
        "title": "Política de Segurança de TI",
        "content": (
            "Política de Segurança de TI da Acme Corp\n\n"
            "1. Requisitos de Senha\n"
            "Todas as senhas devem ter pelo menos 12 caracteres e incluir maiúsculas, "
            "minúsculas, números e caracteres especiais. "
            "As senhas devem ser alteradas a cada 90 dias.\n\n"
            "2. Política de Dispositivos\n"
            "Dispositivos pessoais não podem se conectar à rede corporativa. "
            "Todos os dispositivos da empresa devem ter um software antivírus aprovado instalado."
        ),
    },
]

# --- Etapa 2: Segmentar documentos ---
def chunk_text(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        start += chunk_size - overlap
    return chunks

all_chunks: list[str] = []
chunk_sources: list[str] = []

for doc in documents:
    chunks = chunk_text(doc["content"])
    for chunk in chunks:
        all_chunks.append(chunk)
        chunk_sources.append(doc["title"])

print(f"Total de chunks: {len(all_chunks)}")

# --- Etapa 3: Gerar embeddings ---
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

chunk_embeddings = [get_embedding(chunk) for chunk in all_chunks]
print(f"Dimensão do embedding: {len(chunk_embeddings[0])}")

# --- Etapa 4: Busca vetorial ---
def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr = np.array(a)
    b_arr = np.array(b)
    return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

def search(query: str, k: int = 3) -> list[tuple[str, str, float]]:
    query_emb = get_embedding(query)
    scores = [
        (all_chunks[i], chunk_sources[i], cosine_similarity(query_emb, chunk_embeddings[i]))
        for i in range(len(all_chunks))
    ]
    scores.sort(key=lambda x: x[2], reverse=True)
    return scores[:k]

# --- Etapa 5: Consulta RAG ---
def rag_answer(question: str, k: int = 3) -> str:
    results = search(question, k=k)

    context_parts = []
    for i, (chunk, source, score) in enumerate(results):
        context_parts.append(f"[Fonte: {source}, Relevância: {score:.2f}]\n{chunk}")
    context = "\n\n---\n\n".join(context_parts)

    prompt = f"""Responda à pergunta APENAS com base no contexto a seguir.
Se o contexto não contiver informações suficientes, diga "Não tenho informações suficientes."
Sempre cite de qual documento fonte sua resposta vem.

Contexto:
{context}

Pergunta: {question}
Resposta:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return response.choices[0].message.content

# --- Etapa 6: Teste ---
questions = [
    "Qual é o subsídio diário de viagem para viagens domésticas?",
    "Quantos dias de licença anual os funcionários recebem?",
    "Quais são os requisitos de senha?",
    "Qual é a política de trabalho remoto da empresa?",
]

for q in questions:
    print(f"P: {q}")
    print(f"R: {rag_answer(q)}")
    print()

Total de chunks: 9
Dimensão do embedding: 1536

P: Qual é o subsídio diário de viagem para viagens domésticas?
R: O subsídio diário de viagem para viagens domésticas é de $45, de acordo com a Política de Viagens.

P: Quantos dias de licença anual os funcionários recebem?
R: Funcionários em tempo integral recebem 15 dias de licença anual remunerada por ano, de acordo com a Política de Licenças.

P: Quais são os requisitos de senha?
R: As senhas devem ter pelo menos 12 caracteres e incluir maiúsculas, minúsculas, números e caracteres especiais. Devem ser alteradas a cada 90 dias, de acordo com a Política de Segurança de TI.

P: Qual é a política de trabalho remoto da empresa?
R: Não tenho informações suficientes. O contexto fornecido não contém nenhuma política de trabalho remoto.

texts = [
    "Cats are popular pets known for their independence.",
    "Dogs are loyal companions that love to play fetch.",
    "The stock market rose 3% today on strong earnings.",
    "Felines enjoy climbing and sleeping in sunny spots.",
    "Interest rates were raised by the central bank.",
]

def vector_search(
    query: str,
    corpus: list[str],
    k: int = 5,
    embedding_model: str = "text-embedding-3-small",
) -> list[dict]:
    """
    Returns top-k most similar documents with scores.
    Each result is a dict: {"text": ..., "score": ..., "rank": ...}
    """
    # Sua implementação aqui
    pass