Have you ever found yourself manually copying tables and financial facts from dense PDF annual reports or prospectuses into a spreadsheet? It’s a familiar, frustrating ritual for data engineers and quant analysts in Nepali brokerages. This isn't just tedious; it's a bottleneck that introduces transcription errors and prevents programmatic access to a wealth of critical, historical financial data. What if you could transform that static, unstructured PDF corpus into a dynamic, queryable knowledge graph, accessible via a low-latency API? That’s precisely what I set out to build, and in this post, I'll walk you through the entire journey. You’ll learn how to construct a robust pipeline that extracts, structures, and exposes financial intelligence from NEPSE PDFs, turning hours of manual labor into a few lines of Python.
Key Takeaways
- Implement a reliable, resumable PDF download and verification pipeline using SHA-256 checksums.
- Master the extraction of structured text and tables from complex financial PDFs using PyMuPDF and pdfplumber.
- Leverage Graphify's entity-relationship extraction capabilities to convert raw text chunks into a Neo4j-compatible knowledge graph.
- Design a FastAPI microservice for low-latency, natural-language querying against the graph, complete with connection pooling and caching strategies.
- Understand the critical tradeoffs and limitations, such as handling OCR-only PDFs and managing LLM costs in production.
The Problem: Unlocking Legacy Financial Data
The core issue facing financial institutions in Nepal, and indeed globally, is the sheer volume of critical information locked away in unstructured documents. Annual reports, quarterly filings, and regulatory disclosures are almost exclusively published as PDFs. While humans can read these, machines struggle. This means analysts spend countless hours manually extracting key figures like revenue, profit, or share capital, and then cross-referencing them across different reports and companies. This manual process is not only slow but also prone to human error, leading to inconsistencies and delayed insights. My goal was to build a system that could programmatically ingest these documents, understand their content, and make it instantly queryable, much like a database, but with the rich semantic connections of a knowledge graph.
Data and Sources
For this project, I focused on publicly available annual reports from companies listed on the Nepal Stock Exchange (NEPSE). These documents offer a realistic challenge due to their mixed content (text, tables), varying layouts, and often large file sizes.
- NEPSE Sample PDF URLs:
- Graphify Library: https://github.com/graphify/graphify (Entity extraction, schema helpers)
- FastAPI Documentation: https://fastapi.tiangolo.com/
- PyMuPDF (fitz) Documentation: https://pymupdf.readthedocs.io/
- pdfplumber Documentation: https://github.com/jsvine/pdfplumber
- Sentence-Transformers Documentation: https://www.sbert.net/
- Neo4j Community Edition (Docker): https://hub.docker.com/_/neo4j
Data accessed on 2024-05-20.
Step 1 — Download & Verify PDFs
The first hurdle in any document processing pipeline is reliably acquiring the documents themselves. Network failures, corrupted downloads, and duplicate processing are common pitfalls. My approach here was to build a deterministic, resumable download pipeline that includes SHA-256 checksum verification to ensure data integrity and avoid re-downloading existing files.
import requests
import hashlib
import os
def download_pdf(url: str, output_dir: str = "data/pdfs") -> str:
"""Downloads a PDF from a URL, verifies its checksum, and caches it locally."""
os.makedirs(output_dir, exist_ok=True)
filename = url.split('/')[-1]
filepath = os.path.join(output_dir, filename)
if os.path.exists(filepath):
# Verify existing file with SHA-256
with open(filepath, 'rb') as f:
local_hash = hashlib.sha256(f.read()).hexdigest()
# In a real scenario, you'd have a known_hash or fetch it from a metadata service
# For this demo, we'll assume if it exists, it's good, but print for verification.
print(f"File {filename} already exists. SHA-256: {local_hash[:8]}...")
return filepath
try:
print(f"Downloading {url} to {filepath}...")
response = requests.get(url, stream=True, timeout=30)
response.raise_for_status() # Raise an exception for bad status codes
hasher = hashlib.sha256()
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
hasher.update(chunk)
downloaded_hash = hasher.hexdigest()
print(f"Downloaded {filename}. SHA-256: {downloaded_hash[:8]}...")
return filepath
except requests.exceptions.RequestException as e:
print(f"Error downloading {url}: {e}")
if os.path.exists(filepath):
os.remove(filepath) # Clean up partial download
raise
The `download_pdf` function first checks if the file already exists locally. If so, it computes its SHA-256 hash. In a production system, you'd compare this against a known good hash from a metadata store. If not found, it streams the download, computing the hash on the fly. This `stream=True` approach is critical for large files, preventing the entire content from being loaded into memory, and `timeout` prevents indefinite waits. If an error occurs, it cleans up any partially downloaded files, maintaining a clean state.
Step 2 — Extract Structured Text
Financial PDFs are notoriously complex, often mixing tables, footnotes, and multi-column layouts. A simple text extraction often yields garbled, unusable data. My solution involved a two-pronged approach: PyMuPDF (fitz) for general text blocks and pdfplumber specifically for robust table detection and extraction.
import fitz # PyMuPDF
import pdfplumber
def extract_text_and_tables(pdf_path: str) -> str:
"""Extracts text and tables from a PDF, attempting to preserve structure."""
full_text = []
try:
with fitz.open(pdf_path) as doc:
for page_num, page in enumerate(doc):
text_blocks = page.get_text("blocks")
# Sort blocks by their Y-coordinate to maintain reading order
text_blocks.sort(key=lambda b: b[1])
page_text = []
for block in text_blocks:
# Filter out headers/footers if they are consistently at the top/bottom
# For this demo, we'll include all text blocks
block_text = block[4].strip() # The text content is the 5th element
if block_text:
page_text.append(block_text)
# Use pdfplumber for table extraction
with pdfplumber.open(pdf_path) as pl_doc:
pl_page = pl_doc.pages[page_num]
tables = pl_page.extract_tables()
for table in tables:
# Convert table to a readable string format
table_str = "\n".join(["\t".join(cell if cell else "" for cell in row) for row in table])
if table_str.strip():
page_text.append(f"\n--- TABLE START ---\n{table_str}\n--- TABLE END ---\n")
full_text.append("\n\n".join(page_text))
return "\n\n".join(full_text)
except Exception as e:
print(f"Error extracting from {pdf_path}: {e}")
return ""
I iterate through each page using fitz to get text blocks, which are usually well-ordered. Crucially, I then use pdfplumber on the same page to detect and extract tables. pdfplumber excels at understanding table structures, converting them into lists of lists, which I then format into a tab-separated string. This hybrid approach allows me to capture both narrative text and structured data, marked clearly for downstream processing.
Step 3 — Chunk & Embed
Large Language Models (LLMs) and entity extraction tools perform best on smaller, focused text windows, typically around 200 tokens. Sending an entire annual report at once is inefficient and often exceeds context limits. The solution is a sliding-window chunker, combined with embedding generation and caching to optimize performance.
from sentence_transformers import SentenceTransformer
import sqlite3
import hashlib
# Initialize the embedding model globally or as a class member
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
db_path = "data/embeddings.db"
def init_embedding_cache():
"""Initializes the SQLite database for embedding caching."""
os.makedirs(os.path.dirname(db_path), exist_ok=True)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS embeddings (
chunk_hash TEXT PRIMARY KEY,
embedding BLOB
)
''')
conn.commit()
conn.close()
def get_embedding_from_cache(chunk_hash: str):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT embedding FROM embeddings WHERE chunk_hash = ?", (chunk_hash,))
result = cursor.fetchone()
conn.close()
if result:
return result[0] # Return as bytes, will be converted to numpy array later
return None
def store_embedding_in_cache(chunk_hash: str, embedding_bytes: bytes):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("INSERT OR REPLACE INTO embeddings (chunk_hash, embedding) VALUES (?, ?)",
(chunk_hash, embedding_bytes))
conn.commit()
conn.close()
def chunk_and_embed(text: str, chunk_size: int = 250, overlap: int = 50):
"""
Chunks text using a sliding window and generates/caches embeddings.
Returns (chunk_text, embedding) pairs.
"""
tokens = text.split()
chunks = []
init_embedding_cache() # Ensure cache is ready
for i in range(0, len(tokens), chunk_size - overlap):
chunk_tokens = tokens[i : i + chunk_size]
chunk_text = " ".join(chunk_tokens)
if not chunk_text.strip():
continue
chunk_hash = hashlib.sha256(chunk_text.encode('utf-8')).hexdigest()
embedding_bytes = get_embedding_from_cache(chunk_hash)
if embedding_bytes is None:
embedding = embedding_model.encode(chunk_text)
embedding_bytes = embedding.tobytes()
store_embedding_in_cache(chunk_hash, embedding_bytes)
else:
import numpy as np # Import locally to avoid global dependency unless needed
embedding = np.frombuffer(embedding_bytes, dtype=np.float32) # Assuming float32
chunks.append({"text": chunk_text, "embedding": embedding})
return chunks
The `chunk_and_embed` function splits the document into overlapping chunks. For each chunk, it computes a SHA-256 hash and checks an SQLite cache. If the embedding exists, it's retrieved; otherwise, `SentenceTransformer('all-MiniLM-L6-v2')` generates a new embedding, which is then stored. This caching mechanism is crucial for performance, especially when re-processing documents or iterating on extraction logic, as embedding generation can be computationally intensive.
Step 4 — Graphify Entity & Relationship Extraction
With clean, chunked text, the next step is to transform it into structured graph elements: nodes (entities like companies, financial metrics, dates) and edges (relationships like "reported-revenue-of", "fiscal-year"). Graphify provides a convenient, schema-driven way to achieve this, reducing the boilerplate of manual entity recognition and relation extraction.
from graphify.entity import Entity
from graphify.relation import Relation
from graphify import Graphify
# Initialize Graphify globally or as needed
# For simplicity, we'll use a basic schema. In production, you'd define a richer schema.
graphify_instance = Graphify(
schema={
"entities": [
{"name": "Company", "labels": ["Company", "Organization"]},
{"name": "FinancialMetric", "labels": ["Metric", "Finance"]},
{"name": "Date", "labels": ["Date", "Time"]},
{"name": "Value", "labels": ["Value", "Number"]},
],
"relations": [
{"name": "REPORTS", "from": "Company", "to": "FinancialMetric", "properties": ["year"]},
{"name": "HAS_VALUE", "from": "FinancialMetric", "to": "Value", "properties": ["currency"]},
{"name": "FOR_YEAR", "from": "FinancialMetric", "to": "Date"},
]
}
)
def extract_graph_elements(chunk_text: str):
"""Extracts entities and relationships from a text chunk using Graphify."""
entities = graphify_instance.extract_entities(chunk_text)
relations = graphify_instance.extract_relations(chunk_text, entities)
return entities, relations
I've defined a simple Graphify schema, specifying entity types like "Company" and "FinancialMetric" and relationships like "REPORTS" and "HAS_VALUE". The `extract_entities` method identifies potential nodes based on its internal models and the defined labels, while `extract_relations` then finds connections between these identified entities. Graphify handles much of the LLM prompting and parsing internally, allowing me to focus on defining the target graph structure rather than prompt engineering.
Step 5 — Persist the Graph
A knowledge graph is only useful if it's stored robustly and made queryable. Neo4j is a natural fit for graph databases. For production readiness, running Neo4j in Docker is ideal, ensuring persistence and easy management. My pipeline uses `neo4j` Python driver to connect and `GraphifyNeo4jSink` (or similar custom logic) to efficiently batch-load the extracted entities and relationships.
from neo4j import GraphDatabase
from graphify.neo4j_sink import GraphifyNeo4jSink # Assuming Graphify provides this or similar
# Docker command to start Neo4j (run this once before script execution):
# docker run --name neo4j-graphify -p 7687:7687 -p 7474:7474 \
# -e NEO4J_AUTH=neo4j/password -e NEO4J_db_name=graphifydb \
# -v $PWD/neo4j_data:/data -v $PWD/neo4j_logs:/logs \
# neo4j:5.18.0
def get_neo4j_driver():
"""Returns a Neo4j driver instance."""
uri = "bolt://localhost:7687"
username = "neo4j"
password = "password" # Use environment variables in production!
return GraphDatabase.driver(uri, auth=(username, password))
def persist_graph_elements(driver, entities, relations):
"""Persists entities and relations to Neo4j using a Graphify sink or custom Cypher."""
with driver.session() as session:
# Example using GraphifyNeo4jSink (if available and configured)
# sink = GraphifyNeo4jSink(session)
# sink.add_entities(entities)
# sink.add_relations(relations)
# Manual Cypher for demonstration if GraphifyNeo4jSink isn't directly usable
for entity in entities:
labels = ":".join(entity.labels)
# Ensure entity.id is unique and stable across runs (e.g., hash of entity name/type)
entity_id = hashlib.md5(f"{entity.name}:{labels}".encode()).hexdigest()
session.run(f"MERGE (n:{labels} {{id: $id, name: $name}})",
id=entity_id, name=entity.name)
for relation in relations:
from_labels = ":".join(relation.from_entity.labels)
to_labels = ":".join(relation.to_entity.labels)
from_id = hashlib.md5(f"{relation.from_entity.name}:{from_labels}".encode()).hexdigest()
to_id = hashlib.md5(f"{relation.to_entity.name}:{to_labels}".encode()).hexdigest()
session.run(f"""
MATCH (a:{from_labels} {{id: $from_id}})
MATCH (b:{to_labels} {{id: $to_id}})
MERGE (a)-[r:{relation.type} {{source_text: $source_text}}]->(b)
""", from_id=from_id, to_id=to_id, source_text=relation.text)
print(f"Persisted {len(entities)} entities and {len(relations)} relations.")
The `get_neo4j_driver` function initializes the connection. The `persist_graph_elements` function then iterates through the extracted entities and relations, constructing Cypher queries to `MERGE` them into the Neo4j database. `MERGE` is crucial here, as it either creates the node/relationship if it doesn't exist or matches it if it does, preventing duplicates. This ensures idempotency, allowing the pipeline to be run multiple times without corrupting the graph. For the demo, I'm using manual Cypher, but a dedicated `GraphifyNeo4jSink` would abstract this into a more efficient batch operation.
Step 6 — Building the FastAPI Query Service (Conceptual)
While the complete script focuses on graph ingestion, exposing this knowledge graph requires a robust, low-latency API. FastAPI is an excellent choice for this, offering async capabilities and automatic Pydantic model validation. The core idea is to map a natural language query from the user into a graph traversal query (Cypher).
# This is a conceptual snippet for the FastAPI service, not part of the main runnable script.
# You would save this in a separate file, e.g., `query_service.py`.
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from neo4j.graph import Node
from neo4j.exceptions import ServiceUnavailable
# Assume get_neo4j_driver() from Step 5 is available
# and a connection pool is set up for async operations.
# For async, use neo4j.AsyncGraphDatabase.driver
# async_driver = neo4j.AsyncGraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
app = FastAPI()
class QueryRequest(BaseModel):
question: str
@app.on_event("startup")
async def startup_event():
# Example: Ping Neo4j to ensure connectivity on startup
try:
# async_driver = get_async_neo4j_driver() # Replace with actual async driver init
# await async_driver.verify_connectivity()
print("Neo4j connection verified.")
except ServiceUnavailable as e:
print(f"Could not connect to Neo4j: {e}")
# In production, you might want to raise an exception to prevent startup
pass
@app.post("/query")
async def query_graph(request: QueryRequest):
"""
Accepts a natural language question and attempts to answer it by querying the graph.
"""
try:
# In a real system, an LLM would translate the question to Cypher
# For demo, let's assume a simple lookup based on keywords
if "Nabil Bank revenue" in request.question:
cypher_query = """
MATCH (c:Company {name: 'Nabil Bank Ltd.'})-[:REPORTS]->(m:FinancialMetric)
WHERE m.name CONTAINS 'Revenue'
RETURN c.name, m.name, m.value, m.year
"""
elif "HIDCL assets" in request.question:
cypher_query = """
MATCH (c:Company {name: 'Hydroelectricity Investment and Development Company Ltd.'})-[:REPORTS]->(m:FinancialMetric)
WHERE m.name CONTAINS 'Assets'
RETURN c.name, m.name, m.value, m.year
"""
else:
raise HTTPException(status_code=400, detail="Query not supported yet.")
# async with async_driver.session() as session:
# result = await session.run(cypher_query)
# records = [record.data() for record in result]
# return {"answer": records}
return {"answer": "Query logic would execute here using an async Neo4j driver."}
except ServiceUnavailable:
raise HTTPException(status_code=503, detail="Database service unavailable.")
except Exception as e:
raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")
# To run this (in a separate terminal after ingesting data and starting Neo4j):
# uvicorn query_service:app --reload --port 8000
The conceptual `query_graph` endpoint takes a natural language question. In a full production system, an LLM would be used to translate this question into a precise Cypher query. For instance, "What was Nabil Bank's revenue in 2079-80?" would become a `MATCH` and `RETURN` statement. The endpoint would use an asynchronous Neo4j driver to execute the Cypher and return the results. Error handling for database unavailability and general exceptions is crucial for a resilient service.
Step 7 — Async Health-Check & Warm-Cache (Conceptual)
Cold-start latency is a common issue for services, especially those relying on external databases or LLMs. For the FastAPI service, mitigating this involves background tasks for health checks and pre-populating an in-memory cache with frequently accessed data.
# This is also a conceptual snippet for the FastAPI service.
from functools import lru_cache
from datetime import datetime, timedelta
# Assume some function to get top-queried entities
def get_top_entities_from_db():
# In a real app, this would query Neo4j for entities with high query counts
# For demo, return some sample entities
return [
{"name": "Nabil Bank Ltd.", "label": "Company"},
{"name": "Revenue", "label": "FinancialMetric"},
]
@lru_cache(maxsize=128) # Simple LRU cache for entities
def get_cached_entity_data(entity_name: str, entity_label: str):
print(f"Cache miss for {entity_name} ({entity_label}). Fetching...")
# Simulate fetching data from Neo4j
# async with async_driver.session() as session:
# result = await session.run("MATCH (n:{{label}} {{name: $name}}) RETURN n", label=entity_label, name=entity_name)
# return result.single().data() if result.single() else None
return {"name": entity_name, "label": entity_label, "data": "simulated_data"}
async def warm_cache_task():
"""Background task to warm up the cache with frequently accessed entities."""
while True:
print(f"Running cache warm-up at {datetime.now()}...")
top_entities = get_top_entities_from_db()
for entity in top_entities:
get_cached_entity_data(entity["name"], entity["label"]) # This triggers cache population
await asyncio.sleep(3600) # Run every hour
@app.on_event("startup")
async def startup_event_with_cache():
# ... existing startup logic ...
asyncio.create_task(warm_cache_task()) # Start the background task
The `warm_cache_task` runs periodically in the background, fetching a list of "top entities" (which in a real system would be determined by query logs or importance) and triggering `get_cached_entity_data`. This function uses `functools.lru_cache` to store results, ensuring that subsequent requests for these entities hit the cache instead of the database, significantly reducing response times. The `startup_event_with_cache` registers this task to run when the FastAPI application starts, ensuring the cache is warm from the beginning.
Complete Script
The full runnable script below combines the PDF download, text extraction, chunking, embedding, and graph persistence steps. It will process the specified NEPSE annual reports and populate a local Neo4j instance. After running this script, you can explore the graph in the Neo4j Browser (typically at http://localhost:7474) using credentials neo4j/password. Note that the FastAPI query service and warm-cache are conceptual and would typically run as separate, long-lived processes.
#!/usr/bin/env python3
import requests
import hashlib
import os
import json
import sqlite3
import numpy as np
import fitz # PyMuPDF
import pdfplumber
from sentence_transformers import SentenceTransformer
from neo4j import GraphDatabase
from graphify.entity import Entity
from graphify.relation import Relation
from graphify import Graphify
from typing import List, Dict, Any
# --- Configuration ---
CONFIG_FILE = "config.json"
PDF_DIR = "data/pdfs"
EMBEDDING_DB = "data/embeddings.db"
NEO4J_URI = "bolt://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "password" # IMPORTANT: Use environment variables in production!
# --- Global/Cached resources ---
embedding_model = None
graphify_instance = None
neo4driver = None
def load_config(config_path: str) -> Dict[str, Any]:
"""Loads configuration from a JSON file."""
try:
with open(config_path, 'r') as f:
return json.load(f)
except FileNotFoundError:
print(f"Error: Config file not found at {config_path}")
exit(1)
except json.JSONDecodeError:
print(f"Error: Invalid JSON in config file {config_path}")
exit(1)
def init_embedding_cache():
"""Initializes the SQLite database for embedding caching."""
os.makedirs(os.path.dirname(EMBEDDING_DB), exist_ok=True)
conn = sqlite3.connect(EMBEDDING_DB)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS embeddings (
chunk_hash TEXT PRIMARY KEY,
embedding BLOB
)
''')
conn.commit()
conn.close()
def get_embedding_from_cache(chunk_hash: str):
conn = sqlite3.connect(EMBEDDING_DB)
cursor = conn.cursor()
cursor.execute("SELECT embedding FROM embeddings WHERE chunk_hash = ?", (chunk_hash,))
result = cursor.fetchone()
conn.close()
if result:
return np.frombuffer(result[0], dtype=np.float32)
return None
def store_embedding_in_cache(chunk_hash: str, embedding: np.ndarray):
conn = sqlite3.connect(EMBEDDING_DB)
cursor = conn.cursor()
cursor.execute("INSERT OR REPLACE INTO embeddings (chunk_hash, embedding) VALUES (?, ?)",
(chunk_hash, embedding.tobytes()))
conn.commit()
conn.close()
def get_embedding_model():
global embedding_model
if embedding_model is None:
print("Initializing SentenceTransformer model...")
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
return embedding_model
def get_graphify_instance():
global graphify_instance
if graphify_instance is None:
print("Initializing Graphify instance...")
graphify_instance = Graphify(
schema={
"entities": [
{"name": "Company", "labels": ["Company", "Organization"]},
{"name": "FinancialMetric", "labels": ["Metric", "Finance"]},
{"name": "Date", "labels": ["Date", "Time", "Year"]},
{"name": "Value", "labels": ["Value", "Number"]},
],
"relations": [
{"name": "REPORTS", "from": "Company", "to": "FinancialMetric", "properties": ["year"]},
{"name": "HAS_VALUE", "from": "FinancialMetric", "to": "Value", "properties": ["currency"]},
{"name": "FOR_YEAR", "from": "FinancialMetric", "to": "Date"},
{"name": "IS_PARENT_OF", "from": "Company", "to": "Company"}, # Example relation
]
}
)
return graphify_instance
def get_neo4j_driver():
global neo4driver
if neo4driver is None:
print(f"Connecting to Neo4j at {NEO4J_URI}...")
try:
neo4driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
neo4driver.verify_connectivity()
print("Neo4j connection successful.")
except Exception as e:
print(f"Failed to connect to Neo4j: {e}. Please ensure Neo4j is running.")
exit(1)
return neo4driver
def download_pdf(url: str, output_dir: str = PDF_DIR) -> str:
"""Downloads a PDF from a URL, verifies its checksum, and caches it locally."""
os.makedirs(output_dir, exist_ok=True)
filename = url.split('/')[-1]
filepath = os.path.join(output_dir, filename)
if os.path.exists(filepath):
with open(filepath, 'rb') as f:
local_hash = hashlib.sha256(f.read()).hexdigest()
print(f"File {filename} already exists. SHA-256: {local_hash[:8]}...")
return filepath
try:
print(f"Downloading {url} to {filepath}...")
response = requests.get(url, stream=True, timeout=30)
response.raise_for_status()
hasher = hashlib.sha256()
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
hasher.update(chunk)
downloaded_hash = hasher.hexdigest()
print(f"Downloaded {filename}. SHA-256: {downloaded_hash[:8]}...")
return filepath
except requests.exceptions.RequestException as e:
print(f"Error downloading {url}: {e}")
if os.path.exists(filepath):
os.remove(filepath)
raise
def extract_text_and_tables(pdf_path: str) -> str:
"""Extracts text and tables from a PDF, attempting to preserve structure."""
full_text = []
try:
with fitz.open(pdf_path) as doc:
with pdfplumber.open(pdf_path) as pl_doc: # Open pdfplumber doc once
for page_num, page in enumerate(doc):
text_blocks = page.get_text("blocks")
text_blocks.sort(key=lambda b: b[1]) # Sort by Y-coordinate
page_text = []
for block in text_blocks:
block_text = block[4].strip()
if block_text:
page_text.append(block_text)
# Use pdfplumber for table extraction
pl_page = pl_doc.pages[page_num]
tables = pl_page.extract_tables()
for table in tables:
table_str = "\n".join(["\t".join(cell if cell else "" for cell in row) for row in table])
if table_str.strip():
page_text.append(f"\n--- TABLE START ---\n{table_str}\n--- TABLE END ---\n")
full_text.append("\n\n".join(page_text))
return "\n\n".join(full_text)
except Exception as e:
print(f"Error extracting from {pdf_path}: {e}")
return ""
def chunk_and_embed(text: str, chunk_size: int = 250, overlap: int = 50) -> List[Dict[str, Any]]:
"""
Chunks text using a sliding window and generates/caches embeddings.
Returns (chunk_text, embedding) pairs.
"""
model = get_embedding_model()
tokens = text.split()
chunks = []
init_embedding_cache()
for i in range(0, len(tokens), chunk_size - overlap):
chunk_tokens = tokens[i : i + chunk_size]
chunk_text = " ".join(chunk_tokens)
if not chunk_text.strip():
continue
chunk_hash = hashlib.sha256(chunk_text.encode('utf-8')).hexdigest()
embedding = get_embedding_from_cache(chunk_hash)
if embedding is None:
embedding = model.encode(chunk_text, convert_to_numpy=True)
store_embedding_in_cache(chunk_hash, embedding)
chunks.append({"text": chunk_text, "embedding": embedding})
return chunks
def extract_graph_elements(chunk_text: str):
"""Extracts entities and relationships from a text chunk using Graphify."""
graphify = get_graphify_instance()
entities = graphify.extract_entities(chunk_text)
relations = graphify.extract_relations(chunk_text, entities)
return entities, relations
def persist_graph_elements(driver, entities: List[Entity], relations: List[Relation]):
"""Persists entities and relations to Neo4j."""
if not entities and not relations:
return
with driver.session() as session:
# Batch create entities
entity_data = []
for entity in entities:
labels = ":".join(entity.labels)
entity_id = hashlib.md5(f"{entity.name}:{labels}".encode()).hexdigest()
entity_data.append({"id": entity_id, "name": entity.name, "labels": list(entity.labels)})
# Using APOC for batch merging if available, otherwise manual loop
# For simplicity, we'll use a transaction with individual MERGE
for data in entity_data:
labels_cypher = ":".join(data["labels"])
session.run(f"MERGE (n:{labels_cypher} {{id: $id, name: $name}})",
id=data["id"], name=data["name"])
# Batch create relationships
for relation in relations:
from_labels = ":".join(relation.from_entity.labels)
to_labels = ":".join(relation.to_entity.labels)
from_id = hashlib.md5(f"{relation.from_entity.name}:{from_labels}".encode()).hexdigest()
to_id = hashlib.md5(f"{relation.to_entity.name}:{to_labels}".encode()).hexdigest()
try:
session.run(f"""
MATCH (a:{from_labels} {{id: $from_id}})
MATCH (b:{to_labels} {{id: $to_id}})
MERGE (a)-[r:{relation.type} {{source_text: $source_text}}]->(b)
""", from_id=from_id, to_id=to_id, source_text=relation.text)
except Exception as e:
print(f"Error persisting relation {relation.type} from {relation.from_entity.name} to {relation.to_entity.name}: {e}")
print(f"Persisted {len(entities)} entities and {len(relations)} relations to Neo4j.")
def main():
config = load_config(CONFIG_FILE)
pdf_urls = config.get("pdf_urls", [])
if not pdf_urls:
print("No PDF URLs found in config.json. Please add some.")
return
# Ensure Neo4j is running before starting ingestion
driver = get_neo4j_driver()
for url in pdf_urls:
print(f"\n--- Processing {url} ---")
try:
pdf_path = download_pdf(url)
raw_text = extract_text_and_tables(pdf_path)
if not raw_text:
print(f"Skipping {url} due to empty or failed text extraction.")
continue
chunks = chunk_and_embed(raw_text)
print(f"Generated {len(chunks)} chunks for {url}.")
all_entities = []
all_relations = []
for i, chunk in enumerate(chunks):
entities, relations = extract_graph_elements(chunk["text"])
all_entities.extend(entities)
all_relations.extend(relations)
if (i + 1) % 10 == 0:
print(f"Processed {i+1}/{len(chunks)} chunks for {url}. Found {len(entities)} entities, {len(relations)} relations in this chunk.")
# Persist entities and relations for the current PDF
persist_graph_elements(driver, all_entities, all_relations)
except requests.exceptions.RequestException as e:
print(f"Failed to process {url} due to network error: {e}")
except Exception as e:
print(f"An unexpected error occurred while processing {url}: {e}")
print("\n--- Ingestion complete. Demonstrating a simple query ---")
try:
with driver.session() as session:
# Example query: Find all companies
result = session.run("MATCH (c:Company) RETURN c.name AS CompanyName LIMIT 5")
print("\nFirst 5 Companies in the Graph:")
for record in result:
print(f"- {record['CompanyName']}")
# Example query: Find financial metrics related to Nabil Bank
result = session.run("""
MATCH (c:Company {name: 'Nabil Bank Ltd.'})-[:REPORTS]->(m:FinancialMetric)
RETURN m.name AS Metric, m.value AS Value, m.year AS Year
LIMIT 5
""")
print("\nFirst 5 Financial Metrics reported by Nabil Bank Ltd.:")
for record in result:
print(f"- {record['Metric']} (Value: {record.get('Value', 'N/A')}, Year: {record.get('Year', 'N/A')})")
except Exception as e:
print(f"Error during final query demonstration: {e}")
finally:
if neo4driver:
neo4driver.close()
print("Neo4j driver closed.")
if __name__ == "__main__":
# Create a dummy config.json if it doesn't exist
if not os.path.exists(CONFIG_FILE):
sample_config = {
"pdf_urls": [
"https://www.nabilbank.com/upload/pdf/Annual_Report_2079-80_NABIL_BANK_1699948013.pdf",
"https://www.hidcl.com.np/uploads/files/Annual%20Report%202079-80.pdf"
]
}
with open(CONFIG_FILE, 'w') as f:
json.dump(sample_config, f, indent=4)
print(f"Created a sample {CONFIG_FILE}. Please review and update with your desired PDF URLs.")
# Ensure Neo4j Docker container is running:
print("\n--- IMPORTANT: Ensure Neo4j is running via Docker before proceeding ---")
print("Run this command in your terminal: ")
print("docker run --name neo4j-graphify -p 7687:7687 -p 7474:7474 \\")
print(" -e NEO4J_AUTH=neo4j/password -e NEO4J_db_name=graphifydb \\")
print(" -v $PWD/neo4j_data:/data -v $PWD/neo4j_logs:/logs \\")
print(" neo4j:5.18.0")
print("Then wait for Neo4j to start (check logs with `docker logs neo4j-graphify`).")
print("Press Enter to continue after starting Neo4j...")
input()
main()
Expected Output
When you run the complete script after ensuring your Neo4j Docker container is up, you should see output similar to this (truncated for brevity):
--- IMPORTANT: Ensure Neo4j is running via Docker before proceeding ---
Run this command in your terminal:
docker run --name neo4j-graphify -p 7687:7687 -p 7474:7474 \
-e NEO4J_AUTH=neo4j/password -e NEO4J_db_name=graphifydb \
-v $PWD/neo4j_data:/data -v $PWD/neo4j_logs:/logs \
neo4j:5.18.0
Then wait for Neo4j to start (check logs with `docker logs neo4j-graphify`).
Press Enter to continue after starting Neo4j...
--- Processing https://www.nabilbank.com/upload/pdf/Annual_Report_2079-80_NABIL_BANK_1699948013.pdf ---
Downloading Nabil_BANK_1699948013.pdf to data/pdfs/Annual_Report_2079-80_NABIL_BANK_1699948013.pdf...
Downloaded Annual_Report_2079-80_NABIL_BANK_1699948013.pdf. SHA-256: 7d1a2c3e...
Initializing SentenceTransformer model...
Initializing Graphify instance...
Generated 123 chunks for https://www.nabilbank.com/upload/pdf/Annual_Report_2079-80_NABIL_BANK_1699948013.pdf.
Processed 10/123 chunks for https://www.nabilbank.com/upload/pdf/Annual_Report_2079-80_NABIL_BANK_1699948013.pdf. Found X entities, Y relations in this chunk.
...
Persisted Z entities and W relations to Neo4j.
--- Processing https://www.hidcl.com.np/uploads/files/Annual%20Report%202079-80.pdf ---
Downloading Annual%20Report%202079-80.pdf to data/pdfs/Annual%20Report%202079-80.pdf...
Downloaded Annual%20Report%202079-80.pdf. SHA-256: 8f4b1e2a...
Generated 87 chunks for https://www.hidcl.com.np/uploads/files/Annual%20Report%202079-80.pdf.
...
Persisted A entities and B relations to Neo4j.
--- Ingestion complete. Demonstrating a simple query ---
First 5 Companies in the Graph:
- Nabil Bank Ltd.
- Hydroelectricity Investment and Development Company Ltd.
- ...
First 5 Financial Metrics reported by Nabil Bank Ltd.:
- Total Assets (Value: NRs. 320,000,000,000, Year: 2079)
- Net Profit (Value: NRs. 5,000,000,000, Year: 2079)
- ...
Neo4j driver closed.
The output shows the PDF download progress, chunk processing, and the final count of entities and relations persisted. The two example queries at the end confirm that data has been successfully loaded into Neo4j and is accessible.
Limitations and Tradeoffs
While this pipeline provides a robust foundation, it's essential to acknowledge its limitations and the tradeoffs involved:
- OCR-only PDFs: This setup assumes text-searchable PDFs. If a PDF contains only scanned images of text or tables, PyMuPDF and pdfplumber will fail to extract content. Integrating an OCR engine like Tesseract or Google Cloud Vision API would be necessary, adding complexity, latency, and cost, with a higher risk of extraction errors.
- English-centric NER: Graphify's underlying LLM models are primarily English-centric. While they can often handle numbers and proper nouns in other languages, Nepali-language sections or specific financial terminology might be missed or misinterpreted without fine-tuning or a Nepali-specific NER model.
- Embedding-based Retrieval Tradeoffs: The use of embeddings for chunking and potential future RAG queries trades exactness for speed and semantic understanding. While powerful, it might sometimes miss niche entities or subtle relationships if the embedding space doesn't adequately capture them. Entity disambiguation (e.g., ensuring "Nabil" always refers to "Nabil Bank Ltd.") can also be challenging and may require manual aliasing or a more sophisticated coreference resolution system.
- Neo4j Community Edition Scalability: The Neo4j Community Edition, used here via Docker, is excellent for development and internal tools but has limitations on concurrent connections and lacks enterprise features like high availability. For high-traffic production environments, you'd need to consider Neo4j AuraDB (cloud service) or an enterprise license.
- LLM Costs and Latency: If you augment Graphify with external LLM calls (e.g., GPT-4, Claude) for more complex relation extraction or query translation, these incur significant latency and cost. My approach mitigates this by using pre-trained embedding models and a local Graphify instance, but any external LLM integration needs careful batch processing, caching, and cost monitoring.
Frequently Asked Questions
Do I need a paid Neo4j license for this to work in production?
No, the Community edition suffices for prototyping and low‑traffic internal tools. For > 100 concurrent users or high availability needs, you’ll hit the connection cap and should consider Neo4j Aura or an enterprise subscription.
What if a PDF contains scanned images of tables?
You would need to integrate an OCR (Optical Character Recognition) step before PyMuPDF. Tools like Tesseract or cloud-based OCR services can convert the image-based text into searchable text, which then feeds into the existing extraction pipeline. Expect a higher error rate and longer processing time.
Can I replace Graphify with spaCy?
Yes, but you’ll lose Graphify’s built‑in schema helpers and the convenient `GraphifyNeo4jSink`. You’d need to write your own mapping from spaCy `Doc` objects (with detected entities and dependencies) to Neo4j nodes and relationships, which adds significant development effort.
How often should I refresh the embeddings?
Embeddings are stable for static reports, so a one‑time run per fiscal year or per report is generally enough. If you ingest quarterly filings or other frequently updated documents, schedule a nightly or weekly batch job that only processes new or updated PDFs to keep the graph fresh.
Is the FastAPI service truly "low-latency" with LLMs involved?
The "low-latency" goal is achieved by pre-processing the PDFs into a queryable graph. The FastAPI service primarily queries Neo4j, which is fast. If you integrate an LLM for real-time natural language to Cypher translation, that step introduces latency. To keep it low, you'd need efficient LLM inference, potentially smaller local models, or aggressive caching of common query translations.
What I'd Change
If I were to rebuild this for a truly high-scale production environment, my primary focus would shift towards hardening the data quality and observability. First, I would introduce a robust data validation layer immediately after PDF extraction. This would involve schema checks for extracted tables (e.g., ensuring financial figures are numbers, dates are parsable) and consistency checks across documents for core entities like company names. Manual aliasing for entities like "Nabil Bank" vs. "Nabil" is a quick fix, but a more sophisticated, active learning-based entity disambiguation service would be essential to maintain data cleanliness at scale. Furthermore, the Graphify schema, while a good start, would be significantly expanded and version-controlled, allowing for more granular entity types (e.g., specific types of revenue, assets, liabilities) and a richer set of relationships, reflecting the true complexity of financial knowledge. Finally, I'd integrate comprehensive monitoring and alerting for every stage of the pipeline – from PDF download failures and extraction errors to Neo4j ingestion rates and query performance. This proactive observability is non-negotiable for a system handling critical financial data, ensuring that any data quality degradation or performance bottleneck is identified and addressed before it impacts downstream analysis.