Spaces:
Sleeping
Sleeping
| from .pdf_utils import extract_text_from_pdf, chunk_text | |
| import numpy as np | |
| import json | |
| import requests | |
| # Parâmetros Azure OpenAI | |
| ENDPOINT = "https://botdall3.openai.azure.com/openai/deployments/text-embedding-3-large/embeddings?api-version=2024-02-15-preview" | |
| KEY = "d154d781aaf34410b502bb9be357e174" | |
| def build_vector_index(pdf_path, index_path="my_vector_index.json"): | |
| text_chunks = chunk_text(extract_text_from_pdf(pdf_path), chunk_size=1500, overlap=200) | |
| embeddings = [] | |
| headers = { | |
| "Content-Type": "application/json", | |
| "api-key": KEY | |
| } | |
| for chunk in text_chunks: | |
| if len(chunk) > 2000: | |
| chunk = chunk[:2000] | |
| payload = {"input": [chunk]} | |
| try: | |
| response = requests.post(ENDPOINT, headers=headers, json=payload) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if "data" in data and data["data"] and "embedding" in data["data"][0]: | |
| embeddings.append(data["data"][0]["embedding"]) | |
| else: | |
| print(f"[ERRO] Resposta inesperada da API para chunk: {chunk[:50]}...") | |
| embeddings.append([0.0]*1536) | |
| else: | |
| print(f"[ERRO] Falha na requisição: {response.status_code} - {response.text}") | |
| embeddings.append([0.0]*1536) | |
| except Exception as e: | |
| print(f"[ERRO] Falha ao embedar chunk: {e}") | |
| embeddings.append([0.0]*1536) | |
| # Salva os embeddings e os textos em um arquivo JSON | |
| with open(index_path, 'w', encoding='utf-8') as f: | |
| json.dump({"chunks": text_chunks, "embeddings": embeddings}, f) | |
| return index_path | |
| def semantic_search(query, index_path="my_vector_index.json", top_k=3): | |
| headers = { | |
| "Content-Type": "application/json", | |
| "api-key": KEY | |
| } | |
| with open(index_path, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| chunks = data["chunks"] | |
| embeddings = np.array(data["embeddings"]) | |
| payload = {"input": [query]} | |
| response = requests.post(ENDPOINT, headers=headers, json=payload) | |
| if response.status_code == 200: | |
| data_emb = response.json() | |
| if "data" in data_emb and data_emb["data"] and "embedding" in data_emb["data"][0]: | |
| query_emb = np.array(data_emb["data"][0]["embedding"]) | |
| else: | |
| print("[ERRO] Resposta inesperada da API para query.") | |
| return [] | |
| else: | |
| print(f"[ERRO] Falha na requisição de embedding da query: {response.status_code} - {response.text}") | |
| return [] | |
| # Similaridade por produto escalar | |
| sims = np.dot(embeddings, query_emb) / (np.linalg.norm(embeddings, axis=1) * np.linalg.norm(query_emb) + 1e-8) | |
| top_idx = sims.argsort()[-top_k:][::-1] | |
| return [chunks[i] for i in top_idx] | |
| # Exemplo de uso direto (teste rápido) | |
| if __name__ == "__main__": | |
| # Gere o índice a partir de um PDF | |
| index_path = build_vector_index("data/exemplo.pdf") | |
| # Faça uma busca semântica | |
| resultados = semantic_search("Resumo do documento", index_path=index_path) | |
| print("Resultados mais relevantes:") | |
| for r in resultados: | |
| print(r) | |