Spaces:
Running
Running
File size: 6,475 Bytes
e561008 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """
"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()
|