Spaces:
Running
Running
| """ | |
| "Built on Rock" — Pinecone Ingestion Script | |
| ────────────────────────────────────────────────────────────────────────────── | |
| Run this ONCE after setting up your Pinecone index to populate it with | |
| content from "Built on Rock" by Iyanuoluwa Joseph Akinnusi. | |
| Prerequisites: | |
| 1. Create a Pinecone index named 'venthrive-bff' with: | |
| - Dimensions: 1024 | |
| - Metric: Cosine | |
| - Cloud: AWS us-east-1 (Serverless, free tier) | |
| 2. Set PINECONE_API_KEY in backend/.env | |
| 3. Ensure the PDF is in the APEX folder (path configured below) | |
| Usage (from the VenThrive/backend/ directory): | |
| python scripts/ingest_book.py | |
| No local ML library needed — embeddings are generated by Pinecone's | |
| hosted multilingual-e5-large model (1024 dimensions). | |
| """ | |
| import os | |
| import sys | |
| import time | |
| # Load backend env | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from dotenv import load_dotenv | |
| load_dotenv(os.path.join(os.path.dirname(os.path.dirname(__file__)), ".env")) | |
| from pinecone import Pinecone | |
| import pypdf | |
| # ─── Config ─────────────────────────────────────────────────────────────────── | |
| # Absolute path — the PDF lives in the sibling APEX project folder | |
| _BASE = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) | |
| PDF_PATH = os.path.join(_BASE, "APEX", "Built On Rock by Iyanuoluwa Joseph Akinnusi-1.pdf") | |
| CHUNK_WORDS = 400 | |
| OVERLAP_WORDS = 50 | |
| BATCH_SIZE = 90 # Pinecone Inference allows up to 96 inputs per request | |
| EMBED_MODEL = "multilingual-e5-large" | |
| def extract_pdf_text(path: str) -> str: | |
| if not os.path.exists(path): | |
| raise FileNotFoundError( | |
| f"\n[ERR] PDF not found: {path}\n" | |
| "Ensure 'Built On Rock by Iyanuoluwa Joseph Akinnusi-1.pdf' " | |
| "is in the APEX folder." | |
| ) | |
| reader = pypdf.PdfReader(path) | |
| return "\n".join(page.extract_text() or "" for page in reader.pages) | |
| def chunk_text(text: str, chunk_words: int, overlap_words: int) -> list[dict]: | |
| words = text.split() | |
| chunks = [] | |
| i = 0 | |
| while i < len(words): | |
| chunk_text_str = " ".join(words[i : i + chunk_words]) | |
| chunks.append({ | |
| "id": f"chunk_{len(chunks):04d}", | |
| "text": chunk_text_str, | |
| "meta": { | |
| "source": "built_on_rock", | |
| "chunk_id": len(chunks), | |
| "word_start": i, | |
| "word_end": min(i + chunk_words, len(words)), | |
| }, | |
| }) | |
| i += chunk_words - overlap_words | |
| return chunks | |
| def main(): | |
| api_key = os.environ.get("PINECONE_API_KEY", "") | |
| index_name = os.environ.get("PINECONE_INDEX_NAME", "venthrive-bff") | |
| if not api_key: | |
| print("[ERR] PINECONE_API_KEY not set in backend/.env") | |
| return | |
| # ── Extract text ────────────────────────────────────────────────────────── | |
| print(f"[PDF] Reading: {os.path.basename(PDF_PATH)}") | |
| text = extract_pdf_text(PDF_PATH) | |
| words = len(text.split()) | |
| print(f" {words:,} words extracted from PDF") | |
| # ── Chunk ───────────────────────────────────────────────────────────────── | |
| chunks = chunk_text(text, CHUNK_WORDS, OVERLAP_WORDS) | |
| print(f" {len(chunks)} chunks ({CHUNK_WORDS} words each, {OVERLAP_WORDS} overlap)") | |
| # ── Connect to Pinecone ─────────────────────────────────────────────────── | |
| print(f"\n[PIN] Connecting to Pinecone index '{index_name}'...") | |
| pc = Pinecone(api_key=api_key) | |
| index = pc.Index(index_name) | |
| stats = index.describe_index_stats() | |
| existing = stats.get("total_vector_count", 0) | |
| print(f" Existing vectors: {existing}") | |
| if existing > 0: | |
| print(f" Index already has {existing} vectors.") | |
| ans = input(" Clear and re-ingest? (y/N): ").strip().lower() | |
| if ans == 'y': | |
| index.delete(delete_all=True) | |
| print(" ✓ Cleared") | |
| else: | |
| print(" Skipping ingestion.") | |
| return | |
| # ── Embed + upsert in batches ───────────────────────────────────────────── | |
| print(f"\n[AI] Embedding + uploading {len(chunks)} chunks via Pinecone Inference...") | |
| print(" (Model: multilingual-e5-large, 1024 dims — no local ML needed)") | |
| for i in range(0, len(chunks), BATCH_SIZE): | |
| batch = chunks[i : i + BATCH_SIZE] | |
| texts = [c["text"] for c in batch] | |
| # Generate embeddings via Pinecone Inference API | |
| embedding_result = pc.inference.embed( | |
| model=EMBED_MODEL, | |
| inputs=texts, | |
| parameters={"input_type": "passage", "truncate": "END"}, | |
| ) | |
| embeddings = [e.values for e in embedding_result] | |
| # Upsert vectors | |
| vectors = [ | |
| { | |
| "id": chunk["id"], | |
| "values": embeddings[j], | |
| "metadata": {**chunk["meta"], "text": chunk["text"]}, | |
| } | |
| for j, chunk in enumerate(batch) | |
| ] | |
| index.upsert(vectors=vectors) | |
| done = min(i + BATCH_SIZE, len(chunks)) | |
| print(f" {done}/{len(chunks)} chunks uploaded") | |
| # Small pause to stay within Pinecone rate limits | |
| if done < len(chunks): | |
| time.sleep(0.5) | |
| # ── Done ────────────────────────────────────────────────────────────────── | |
| final = index.describe_index_stats() | |
| total = final.get("total_vector_count", "?") | |
| print(f"\n[OK] Ingestion complete! Index '{index_name}' now has {total} vectors.") | |
| print(" VenThrive AI will now draw from 'Built on Rock' for every relevant query.") | |
| if __name__ == "__main__": | |
| main() | |