Skip to content

Graffold Docs

Pipeline Architecture

graffold/graffold-api

Pipeline Architecture¶

Graffold's ingestion pipeline transforms scientific literature and structured data into a queryable knowledge graph. The pipeline is modular — each stage is independently usable and testable.

Pipeline Overview¶

┌─────────────────────────────────────────────────────────────────────┐
│                         Source Fetchers                               │
│  PubMed (Entrez) · bioRxiv/medRxiv · PDF Upload · CSV/Excel         │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                      Content Extraction                               │
│  Docling · PyMuPDF · MarkItDown · pdftotext · Regex fallback         │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                       Text Processing                                 │
│  Clean (Unicode, boilerplate) → Section Parse → Token Chunk (tiktoken)│
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                     KG Extraction (LLM)                               │
│  Entity/Relationship Extraction → Gleaning (multi-pass) → Dedup      │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                      Post-Processing                                  │
│  Entity Resolution · Relationship Consolidation · Evidence Scoring    │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                       Enrichment                                      │
│  UniProt · Open Targets · DisGeNET · MONDO Hierarchy · Communities   │
└──────────────────────────────┬──────────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────────┐
│                      Vector Embeddings                                │
│  Entity embeddings · Chunk embeddings · Vector index creation         │
└─────────────────────────────────────────────────────────────────────┘

Module Reference¶

Ingest (`pipeline/ingest/`)¶

Module	Purpose
`pubmed_fetcher.py`	NCBI Entrez API: search + batch abstract retrieval with PMID dedup
`preprint_fetcher.py`	Unified bioRxiv/medRxiv fetcher with pagination and retry
`content_extractor.py`	Multi-strategy extraction orchestrator (tries strategies in priority order)
`extraction_strategies.py`	Pluggable backends: Docling, PyMuPDF, pdftotext, MarkItDown, Regex
`extraction_models.py`	Data models: ExtractionSource, ExtractionResult, QualityMetrics
`cleaner.py`	Unicode normalization, boilerplate removal, hyphenation fix
`section_parser.py`	Splits papers into labeled sections (Abstract, Methods, Results, etc.)
`token_chunker.py`	tiktoken-based chunking with sentence boundary preservation
`kg_pipeline.py`	Entity/relationship extraction via LLM with dedup and graph storage
`embedding_pipeline.py`	Batch vector embedding generation + Neo4j vector index management

Processors (`pipeline/processors/`)¶

Module	Purpose
`evidence_scorer.py`	Study design classification + composite evidence scoring
`gleaning_extractor.py`	Multi-pass LLM extraction for improved entity recall
`entity_resolver.py`	3-tier dedup: canonical ID → synonym → fuzzy matching
`relationship_consolidator.py`	Merges parallel edges, aggregates all evidence metadata
`community_detector.py`	Louvain clustering + LLM community summarization
`opentargets_integrator.py`	Open Targets GraphQL API → KG enrichment
`disgenet_integrator.py`	DisGeNET gene-disease associations → KG enrichment
`uniprot_integrator.py`	UniProt REST API → protein functional annotations
`disease_hierarchy_enricher.py`	MONDO ontology → IS_SUBTYPE_OF relationships
`refrag_processor.py`	16x context compression via micro-chunk embeddings
`vision_processor.py`	Bedrock vision model for biomedical figure analysis
`preprint_version_tracker.py`	bioRxiv/medRxiv version monitoring and publication detection

Orchestration (`pipeline/`)¶

Module	Purpose
`orchestrator.py`	Top-level coordinator: document → all stages → embeddings
`parallel_ingest.py`	Concurrent multi-source ingestion (PubMed + bioRxiv + medRxiv)
`progress_tracker.py`	Structured JSON progress events with ETA
`cost_tracker.py`	LLM token usage and estimated cost aggregation

Using the Orchestrator¶

from pipeline.orchestrator import PipelineOrchestrator, OrchestratorConfig

orchestrator = PipelineOrchestrator(
    config=OrchestratorConfig(
        database="mydb",
        llm_service="bedrock",
        enable_gleaning=True,
        generate_embeddings=True,
    )
)

# Ingest a PDF
result = await orchestrator.ingest_pdf("paper.pdf", metadata={"doi": "10.1234/..."})

# Ingest raw text (e.g., PubMed abstract)
result = await orchestrator.ingest_text(abstract_text, metadata={"pmid": "12345678"})

# Batch ingest
result = await orchestrator.ingest_batch([
    {"file": "paper1.pdf", "metadata": {"doi": "10.1234/a"}},
    {"text": abstract_text, "metadata": {"pmid": "99999"}},
])

print(f"Extracted {result.extraction_stats.entities_extracted} entities")

Multi-Strategy Content Extraction¶

The content extractor tries backends in priority order until one succeeds:

from pipeline.ingest.content_extractor import ContentExtractor
from pipeline.ingest.extraction_models import ExtractionSource

extractor = ContentExtractor()
print(f"Available: {extractor.get_available_strategies()}")
# ['docling', 'pymupdf', 'markitdown', 'regex']

source = ExtractionSource.from_path("paper.pdf")
result = extractor.extract(source)

if result.success:
    print(f"Extracted {result.quality.word_count} words via {result.method}")

Strategy Priority¶

Docling — IBM's document AI. Best for complex layouts, tables, OCR. Requires docling>=2.88.0.
PyMuPDF — Fast, reliable text extraction. Requires pymupdf>=1.24.0.
pdftotext — Alternative via poppler-utils CLI. Requires system package.
MarkItDown — Microsoft's multi-format converter (PDF, DOCX, Excel, HTML).
Regex — Always-available fallback for HTML/text.

Token-Based Chunking¶

from pipeline.ingest.token_chunker import TokenChunker

chunker = TokenChunker(chunk_size=512, overlap=64)
chunks = chunker.chunk(text)

for chunk in chunks:
    print(f"Chunk {chunk.index}: {chunk.token_count} tokens")

The chunker uses tiktoken (cl100k_base encoding) and preserves sentence boundaries. Falls back to word-count approximation if tiktoken is unavailable.

Parallel Ingestion¶

For bulk knowledge graph building:

from pipeline.parallel_ingest import ParallelIngestOrchestrator, SourceConfig

orchestrator = ParallelIngestOrchestrator(
    sources=[
        SourceConfig(source="pubmed", queries=["kidney biomarker", "cardiac protein"], max_results_per_query=500),
        SourceConfig(source="biorxiv", queries=["protein biomarker"], max_results_per_query=100),
    ]
)

result = await orchestrator.run()
print(f"Total articles: {result.total_articles}")

Or from a queries file:

orchestrator = ParallelIngestOrchestrator.from_queries_file(
    queries_file="data/queries.txt",
    pubmed_max=500,
    biorxiv_max=100,
)