Introduction to Data Extraction with llama.cpp
Data extraction — the process of pulling structured information from unstructured text — has traditionally required brittle regex patterns or expensive cloud APIs. With the rise of local large language models (LLMs), developers can now build powerful extraction pipelines that run entirely on their own hardware. llama.cpp, the lightweight C++ inference engine created by Georgi Gerganov, has become the go-to choice for running quantized models locally with minimal overhead.
This tutorial walks through building a complete, production-ready data extraction pipeline using llama.cpp. We'll cover everything from compiling the engine and selecting a model, to writing extraction prompts, parsing structured output, and wiring everything into a reusable Python pipeline.
What Is llama.cpp and Why It Matters
llama.cpp is an open-source C/C++ library for efficient LLM inference on consumer hardware. It supports GGUF-formatted quantized models, allowing you to run capable models like Llama 3, Mistral, Phi, or Qwen on a laptop CPU or a modest GPU. Unlike Python-centric frameworks that wrap PyTorch, llama.cpp is purpose-built for inference speed and memory efficiency.
Why Use It for Data Extraction?
- Privacy: Sensitive documents never leave your machine — critical for legal, medical, and financial use cases.
- Cost: No per-token API charges. Once you have the model, inference is free.
- Latency: No network round-trips. Local inference can be faster for small extractions.
- Flexibility: Swap models freely to balance speed, accuracy, and memory footprint.
- Offline capability: Works in air-gapped environments.
Prerequisites and Setup
Before building the pipeline, you need a working llama.cpp installation and a suitable model. The setup below assumes a Linux or macOS environment, but Windows users can follow equivalent steps using CMake and MSVC.
Compiling llama.cpp
Clone the repository and build the shared library. If you have an NVIDIA GPU, enable CUDA support for dramatically faster inference.
# Clone the repository
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
# CPU-only build
cmake -B build
cmake --build build --config Release -j
# CUDA build (optional, for NVIDIA GPUs)
cmake -B build -DGGML_CUDA=ON
cmake --build build --config Release -j
After building, you'll find the llama-server binary and the shared library in the build/bin/ directory. The server exposes an OpenAI-compatible HTTP API, which is the easiest integration point for most pipelines.
Downloading a Quantized Model
Quantized GGUF models are available on Hugging Face. For data extraction, a small instruct-tuned model in the 7B–8B parameter range offers a good balance. The Q4_K_M quantization provides near-full precision quality at roughly 4GB of memory.
# Install huggingface-cli
pip install huggingface-hub
# Download a Llama 3.1 8B model, Q4_K_M quantization
huggingface-cli download \
bartowski/Meta-Llama-3.1-8B-Instruct-GGUF \
Meta-Llama-3.1-8B-Instruct-Q4_K_M.gguf \
--local-dir ./models
Starting the Server
./build/bin/llama-server \
-m ./models/Meta-Llama-3.1-8B-Instruct-Q4_K_M.gguf \
--host 0.0.0.0 --port 8080 \
-c 8192 -ngl 33
The -ngl 33 flag offloads all 33 layers to the GPU if available. The -c 8192 flag sets the context window to 8192 tokens, which is usually sufficient for document extraction tasks.
Designing the Extraction Pipeline
A robust extraction pipeline has four stages: ingestion, prompting, generation, and parsing. Each stage should be isolated so you can test and tune them independently.
Stage 1: Document Ingestion
Real-world inputs come in many formats — PDFs, emails, HTML, plain text. Normalize everything to clean text before sending it to the model. Here's a simple ingestion module that handles common formats.
import re
from pathlib import Path
def load_document(path: str) -> str:
"""Load a document and return clean plain text."""
p = Path(path)
suffix = p.suffix.lower()
if suffix == ".txt":
text = p.read_text(encoding="utf-8", errors="ignore")
elif suffix == ".html":
import html2text
text = html2text.html2text(p.read_text(encoding="utf-8", errors="ignore"))
elif suffix == ".pdf":
import pdfplumber
text = ""
with pdfplumber.open(p) as pdf:
for page in pdf.pages:
text += page.extract_text() or ""
else:
text = p.read_text(encoding="utf-8", errors="ignore")
# Collapse excessive whitespace
text = re.sub(r"\n{3,}", "\n\n", text)
text = re.sub(r"[ \t]+", " ", text)
return text.strip()
Stage 2: Prompt Engineering for Extraction
The prompt is the heart of the pipeline. A good extraction prompt does three things: defines the schema explicitly, provides a clear example, and instructs the model to output only structured data with no prose.
EXTRACTION_SYSTEM_PROMPT = """You are a data extraction engine.
Extract structured information from the user's document and return ONLY
valid JSON matching the requested schema. Do not include explanations,
markdown fences, or any text outside the JSON object.
If a field is not present in the document, use null."""
def build_extraction_prompt(document: str, schema: dict) -> str:
import json
schema_str = json.dumps(schema, indent=2)
return f"""Extract the following fields from the document.
Schema:
{schema_str}
Document:
\"\"\"
{document}
\"\"\"
Return a single JSON object matching the schema exactly."""
Defining the schema as a JSON object with field descriptions helps the model understand what each field should contain. For example:
invoice_schema = {
"invoice_number": "string - the invoice ID",
"issue_date": "string - ISO 8601 date (YYYY-MM-DD)",
"due_date": "string - ISO 8601 date (YYYY-MM-DD)",
"vendor_name": "string - name of the issuing company",
"vendor_address": "string - full postal address",
"customer_name": "string - name of the billed customer",
"line_items": [
{
"description": "string",
"quantity": "number",
"unit_price": "number",
"total": "number"
}
],
"subtotal": "number",
"tax_amount": "number",
"total": "number",
"currency": "string - ISO 4217 code"
}
Stage 3: Calling the Model
Since llama-server exposes an OpenAI-compatible API, you can use the standard openai Python library pointed at your local server.
import json
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:8080/v1",
api_key="not-needed" # llama-server ignores the key
)
def extract(document: str, schema: dict, model: str = "local-model") -> dict:
prompt = build_extraction_prompt(document, schema)
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": EXTRACTION_SYSTEM_PROMPT},
{"role": "user", "content": prompt}
],
temperature=0.0,
max_tokens=4096,
response_format={"type": "json_object"}
)
raw = response.choices[0].message.content
return json.loads(raw)
Setting temperature=0.0 makes output deterministic, which is essential for reproducible extraction. The response_format parameter instructs the server to constrain output to valid JSON when the underlying model supports it (Llama 3.1 and many newer models do).
Stage 4: Robust Output Parsing
Even with JSON mode, models occasionally wrap output in markdown fences or include stray text. A defensive parser recovers valid JSON from imperfect output.
import re
import json
def parse_json_response(raw: str) -> dict:
"""Extract and parse JSON from a model response, tolerating noise."""
# Strip markdown code fences if present
cleaned = re.sub(r"^(?:json)?\s*", "", raw.strip(), flags=re.MULTILINE)
cleaned = re.sub(r"\s*$", "", cleaned.strip())
# Try direct parse first
try:
return json.loads(cleaned)
except json.JSONDecodeError:
pass
# Find the outermost JSON object
match = re.search(r"\{.*\}", cleaned, re.DOTALL)
if match:
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
pass
raise ValueError(f"Could not parse JSON from response:\n{raw[:500]}")
Putting It All Together
Now we combine all stages into a single pipeline class with validation and retry logic.
import json
import logging
from pathlib import Path
from typing import Optional
from openai import OpenAI
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("extraction-pipeline")
class ExtractionPipeline:
def __init__(
self,
base_url: str = "http://localhost:8080/v1",
model: str = "local-model",
max_retries: int = 2
):
self.client = OpenAI(base_url=base_url, api_key="not-needed")
self.model = model
self.max_retries = max_retries
def extract_from_file(self, file_path: str, schema: dict) -> dict:
document = load_document(file_path)
return self.extract_from_text(document, schema)
def extract_from_text(self, document: str, schema: dict) -> dict:
prompt = build_extraction_prompt(document, schema)
for attempt in range(self.max_retries + 1):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": EXTRACTION_SYSTEM_PROMPT},
{"role": "user", "content": prompt}
],
temperature=0.0,
max_tokens=4096,
response_format={"type": "json_object"}
)
raw = response.choices[0].message.content
result = parse_json_response(raw)
self._validate(result, schema)
logger.info(f"Extraction succeeded on attempt {attempt + 1}")
return result
except (json.JSONDecodeError, ValueError) as e:
logger.warning(f"Attempt {attempt + 1} failed: {e}")
if attempt == self.max_retries:
raise
raise RuntimeError("Unreachable")
def _validate(self, result: dict, schema: dict) -> None:
"""Check that all top-level keys from the schema are present."""
missing = [k for k in schema if k not in result]
if missing:
raise ValueError(f"Missing required fields: {missing}")
# Usage example
if __name__ == "__main__":
pipeline = ExtractionPipeline()
schema = {
"company_name": "string",
"job_title": "string",
"location": "string",
"salary_range": "string or null",
"required_skills": "array of strings",
"description": "string"
}
result = pipeline.extract_from_file("job_posting.html", schema)
print(json.dumps(result, indent=2))
Batch Processing Multiple Documents
In production, you'll often process hundreds or thousands of documents. Use concurrent requests to keep the GPU saturated. The llama-server handles request queuing internally, so you can fire multiple requests in parallel.
import concurrent.futures
from pathlib import Path
def batch_extract(
pipeline: ExtractionPipeline,
file_paths: list[str],
schema: dict,
max_workers: int = 4
) -> list[dict]:
results = []
errors = []
def process(path):
try:
return {"file": path, "data": pipeline.extract_from_file(path, schema)}
except Exception as e:
return {"file": path, "error": str(e)}
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [executor.submit(process, p) for p in file_paths]
for future in concurrent.futures.as_completed(futures):
outcome = future.result()
if "error" in outcome:
errors.append(outcome)
else:
results.append(outcome)
logger.info(f"Extracted {len(results)} documents, {len(errors)} errors")
return results
Tune max_workers based on your hardware. A single GPU typically handles 2–4 concurrent requests well before memory pressure degrades throughput.
Best Practices
Choose the Right Model Size
Don't default to the largest model available. For structured extraction tasks, a 7B–8B model often matches the accuracy of much larger models while running 3–5× faster. Benchmark a few models on your specific document type before committing.
Use Grammar-Constrained Generation
For maximum reliability, use llama.cpp's grammar-constrained generation feature. You can pass a GBNF (GGML BNF) grammar that forces the model to produce valid JSON matching your exact schema. This eliminates parsing failures entirely.
# example_grammar.gbnf
root ::= "{" ws "\"invoice_number\":" ws string "," ws \
"\"issue_date\":" ws string "," ws \
"\"total\":" ws number "," ws \
"\"currency\":" ws string ws "}"
ws ::= [ \t\n]*
string ::= "\"" ([^"\\] | "\\" .)* "\""
number ::= "-"? ([0-9] | [1-9][0-9]*) ("." [0-9]+)?
Pass the grammar to the server via the --grammar flag or include it in the API request body as the grammar field.
Chunk Long Documents
Documents longer than the context window need chunking. Split on natural boundaries (paragraphs, sections) and extract from each chunk, then merge results. For invoices or forms that fit in context, this is unnecessary.
def chunk_text(text: str, max_chars: int = 6000) -> list[str]:
"""Split text into chunks at paragraph boundaries."""
paragraphs = text.split("\n\n")
chunks, current = [], ""
for para in paragraphs:
if len(current) + len(para) > max_chars and current:
chunks.append(current)
current = para
else:
current = current + "\n\n" + para if current else para
if current:
chunks.append(current)
return chunks
Log Everything
Record the raw model output alongside the parsed result. When extraction fails in production, you'll need the original response to debug whether the issue was the prompt, the model, or the document.
Validate with Pydantic
Replace the simple _validate method with Pydantic models for type checking, coercion, and clear error messages.
from pydantic import BaseModel, Field
from typing import Optional
class LineItem(BaseModel):
description: str
quantity: float
unit_price: float
total: float
class Invoice(BaseModel):
invoice_number: str
issue_date: str
due_date: Optional[str] = None
vendor_name: str
line_items: list[LineItem] = Field(default_factory=list)
subtotal: float
tax_amount: float
total: float
currency: str
# In the pipeline:
validated = Invoice(**result)
Performance Tuning
Extraction throughput depends on several knobs. Here are the most impactful ones:
- Quantization level: Q4_K_M is the sweet spot. Q3 is faster but loses accuracy on complex schemas; Q8 is more accurate but 2× slower.
- Context length: Set
-cto the minimum needed. Larger contexts consume more memory and slow prompt processing. - GPU layers: Offload as many layers as GPU memory allows with
-ngl. - Batch size: Increase
-band-ubfor faster prompt processing on long inputs. - Flash attention: Enable with
-fafor a meaningful speed boost on supported models.
Conclusion
Building a data extraction pipeline with llama.cpp gives you a private, cost-effective, and highly controllable alternative to cloud APIs. By combining a well-quantized local model with disciplined prompt engineering, defensive JSON parsing, and grammar-constrained generation, you can achieve production-grade extraction accuracy on commodity hardware. Start with a small model and a focused schema, validate outputs with Pydantic, and scale up model size or add GPU offloading only when accuracy demands it. The pipeline architecture presented here — ingestion, prompting, generation, parsing — is modular by design, so you can swap models, schemas, or even backends without rewriting the system. With these building blocks in place, you're ready to tackle extraction tasks across invoices, resumes, contracts, medical records, or any other document domain your applications require.