Introduction to LangGraph for Data Extraction
LangGraph is a powerful library built on top of LangChain that enables developers to create stateful, multi-actor applications using graph-based workflows. When it comes to data extraction, LangGraph shines by allowing you to orchestrate complex extraction pipelines with conditional routing, state management, and human-in-the-loop capabilities. This tutorial will walk you through building a complete data extraction pipeline from scratch.
What Is a Data Extraction Pipeline?
A data extraction pipeline is a systematic workflow that takes unstructured or semi-structured input data—such as emails, PDFs, web pages, or raw text—and transforms it into structured, machine-readable formats. The pipeline typically involves parsing, classification, entity recognition, validation, and storage stages. With LangGraph, each of these stages becomes a node in a directed graph, giving you fine-grained control over data flow and error handling.
Why LangGraph Matters for Extraction Workflows
Traditional extraction scripts often fail when data formats vary or when extraction requires multi-step reasoning. LangGraph addresses these challenges by offering:
- Stateful execution: Maintain context across extraction steps without manual state passing.
- Conditional routing: Dynamically route data based on content type, confidence scores, or validation results.
- Retry and error handling: Built-in support for re-running failed nodes with different strategies.
- Human-in-the-loop: Pause execution for human review when extraction confidence is low.
- Parallel execution: Process multiple documents or extraction tasks concurrently for better throughput.
- Observability: Built-in tracing and debugging through LangSmith integration.
Prerequisites and Setup
Before building the pipeline, ensure you have Python 3.9 or later installed. You will also need an OpenAI API key (or another LLM provider) since we will use an LLM for intelligent extraction. Let us start by installing the required packages.
pip install langgraph langchain langchain-openai pydantic python-dotenv
Create a .env file in your project root to store your API key securely:
OPENAI_API_KEY=your-api-key-here
Now create a pipeline.py file and add the initial imports and configuration:
import os
from typing import TypedDict, List, Optional, Dict, Any
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
load_dotenv()
llm = ChatOpenAI(
model="gpt-4o",
temperature=0,
api_key=os.getenv("OPENAI_API_KEY")
)
Defining the Pipeline State
The foundation of any LangGraph pipeline is its state. The state is a typed dictionary that flows through every node in the graph, carrying data from one step to the next. For a data extraction pipeline, the state should capture the raw input, intermediate processing results, extracted entities, validation status, and any error information.
class ExtractionState(TypedDict):
raw_input: str
source_type: str # "email", "invoice", "resume", "unknown"
preprocessed_text: str
extracted_data: Dict[str, Any]
validation_errors: List[str]
confidence_score: float
retry_count: int
final_output: Optional[Dict[str, Any]]
needs_human_review: bool
Each field serves a specific purpose. The raw_input holds the original text, source_type is determined by a classifier node, extracted_data stores the structured output, and validation_errors captures any issues found during validation. The retry_count and needs_human_review fields enable intelligent retry logic and human-in-the-loop functionality.
Building the Pipeline Nodes
Each node in the LangGraph pipeline is a function that takes the current state and returns a dictionary of state updates. Let us build each node step by step.
Node 1: Source Classification
The first node classifies the input data to determine what type of document we are dealing with. This classification drives downstream extraction logic.
def classify_source(state: ExtractionState) -> dict:
"""Classify the input text to determine its source type."""
raw_input = state["raw_input"]
classification_prompt = f"""
Analyze the following text and classify it into one of these categories:
- email: Personal or business correspondence
- invoice: Billing or payment documents
- resume: Professional resumes or CVs
- unknown: Does not fit any category above
Text to classify:
{raw_input[:2000]}
Respond with ONLY the category name, nothing else.
"""
response = llm.invoke(classification_prompt)
source_type = response.content.strip().lower()
valid_types = ["email", "invoice", "resume", "unknown"]
if source_type not in valid_types:
source_type = "unknown"
return {"source_type": source_type}
Node 2: Text Preprocessing
Before extraction, the raw text needs cleaning. This step removes unnecessary whitespace, normalizes encoding, and strips irrelevant content.
def preprocess_text(state: ExtractionState) -> dict:
"""Clean and normalize the input text for extraction."""
raw_input = state["raw_input"]
# Remove excessive whitespace
cleaned = " ".join(raw_input.split())
# Remove common noise patterns
noise_patterns = [
r"-----Original Message-----",
r"On \w{3}, \w{3} \d{1,2}, \d{4}.*?wrote:",
]
import re
for pattern in noise_patterns:
cleaned = re.sub(pattern, "", cleaned, flags=re.IGNORECASE)
# Truncate if extremely long to stay within token limits
max_chars = 8000
if len(cleaned) > max_chars:
cleaned = cleaned[:max_chars] + "... [truncated]"
return {"preprocessed_text": cleaned}
Node 3: Data Extraction
This is the core extraction node. It uses structured output parsing with Pydantic models to ensure the extracted data conforms to a defined schema. We define separate schemas for each source type.
class EmailData(BaseModel):
sender_name: str = Field(description="Name of the email sender")
sender_email: str = Field(description="Email address of the sender")
subject: str = Field(description="Subject line of the email")
key_points: List[str] = Field(description="Main points discussed in the email")
action_items: List[str] = Field(description="Tasks or actions requested")
sentiment: str = Field(description="Overall sentiment: positive, neutral, or negative")
class InvoiceData(BaseModel):
vendor_name: str = Field(description="Name of the vendor or company")
invoice_number: str = Field(description="Invoice identifier")
invoice_date: str = Field(description="Date of the invoice")
total_amount: str = Field(description="Total amount due")
currency: str = Field(description="Currency code, e.g., USD, EUR")
line_items: List[Dict[str, str]] = Field(description="List of items with description and amount")
due_date: str = Field(description="Payment due date")
class ResumeData(BaseModel):
candidate_name: str = Field(description="Full name of the candidate")
email: str = Field(description="Contact email address")
phone: str = Field(description="Contact phone number")
skills: List[str] = Field(description="List of technical or professional skills")
experience_years: str = Field(description="Years of professional experience")
education: List[str] = Field(description="Educational qualifications")
current_role: str = Field(description="Most recent or current job title")
def extract_data(state: ExtractionState) -> dict:
"""Extract structured data based on the classified source type."""
source_type = state["source_type"]
text = state["preprocessed_text"]
schema_map = {
"email": EmailData,
"invoice": InvoiceData,
"resume": ResumeData,
}
if source_type not in schema_map:
return {
"extracted_data": {},
"confidence_score": 0.0,
"validation_errors": [f"Unsupported source type: {source_type}"]
}
schema = schema_map[source_type]
structured_llm = llm.with_structured_output(schema)
extraction_prompt = f"""
Extract structured information from the following {source_type}.
Be precise and only include information that is explicitly stated.
If a field cannot be determined, use "Not specified" as the value.
Text:
{text}
"""
try:
result = structured_llm.invoke(extraction_prompt)
extracted = result.model_dump()
confidence = 0.85 # In production, compute this from model metadata
return {
"extracted_data": extracted,
"confidence_score": confidence,
"validation_errors": []
}
except Exception as e:
return {
"extracted_data": {},
"confidence_score": 0.0,
"validation_errors": [f"Extraction failed: {str(e)}"]
}
Node 4: Validation
After extraction, we validate the data to ensure quality. This node checks for missing fields, format issues, and logical inconsistencies.
def validate_extraction(state: ExtractionState) -> dict:
"""Validate the extracted data for completeness and correctness."""
extracted = state["extracted_data"]
source_type = state["source_type"]
errors = []
if not extracted:
errors.append("No data was extracted")
return {"validation_errors": errors}
# Define required fields per source type
required_fields = {
"email": ["sender_email", "subject"],
"invoice": ["vendor_name", "invoice_number", "total_amount"],
"resume": ["candidate_name", "email"],
}
required = required_fields.get(source_type, [])
for field in required:
value = extracted.get(field, "")
if not value or value == "Not specified":
errors.append(f"Required field '{field}' is missing or not specified")
# Validate email format if present
import re
email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
for email_field in ["sender_email", "email"]:
if email_field in extracted and extracted[email_field] != "Not specified":
if not re.match(email_pattern, extracted[email_field]):
errors.append(f"Invalid email format in field '{email_field}'")
# Check confidence threshold
if state["confidence_score"] < 0.7:
errors.append(f"Low confidence score: {state['confidence_score']}")
return {"validation_errors": errors}
Node 5: Retry Handler
When validation fails, the retry handler attempts re-extraction with adjusted parameters. It tracks retry attempts to prevent infinite loops.
def handle_retry(state: ExtractionState) -> dict:
"""Handle retry logic for failed extractions."""
retry_count = state.get("retry_count", 0)
max_retries = 2
if retry_count >= max_retries:
return {
"needs_human_review": True,
"retry_count": retry_count
}
# Adjust extraction strategy on retry
# For example, use a more powerful model or different prompt
return {"retry_count": retry_count + 1}
Node 6: Finalize Output
The final node packages the extracted and validated data into the output format.
def finalize_output(state: ExtractionState) -> dict:
"""Package the final output with metadata."""
return {
"final_output": {
"source_type": state["source_type"],
"data": state["extracted_data"],
"confidence": state["confidence_score"],
"validation_passed": len(state["validation_errors"]) == 0,
"metadata": {
"errors": state["validation_errors"],
"retries": state.get("retry_count", 0),
"human_review_required": state.get("needs_human_review", False)
}
}
}
Wiring the Graph Together
Now that all nodes are defined, we connect them into a directed graph. The graph uses conditional edges to route data based on validation results and retry counts.
def should_retry_or_finalize(state: ExtractionState) -> str:
"""Conditional edge: determine next step after validation."""
errors = state["validation_errors"]
retry_count = state.get("retry_count", 0)
if not errors:
return "finalize"
if retry_count < 2:
return "retry"
return "finalize"
def route_after_retry(state: ExtractionState) -> str:
"""Conditional edge: determine next step after retry handler."""
if state.get("needs_human_review", False):
return "finalize"
return "extract"
# Build the graph
workflow = StateGraph(ExtractionState)
# Add all nodes
workflow.add_node("classify", classify_source)
workflow.add_node("preprocess", preprocess_text)
workflow.add_node("extract", extract_data)
workflow.add_node("validate", validate_extraction)
workflow.add_node("retry", handle_retry)
workflow.add_node("finalize", finalize_output)
# Set the entry point
workflow.set_entry_point("classify")
# Add fixed edges
workflow.add_edge("classify", "preprocess")
workflow.add_edge("preprocess", "extract")
workflow.add_edge("extract", "validate")
# Add conditional edges
workflow.add_conditional_edges(
"validate",
should_retry_or_finalize,
{
"retry": "retry",
"finalize": "finalize"
}
)
workflow.add_conditional_edges(
"retry",
route_after_retry,
{
"extract": "extract",
"finalize": "finalize"
}
)
# Finalize is the terminal node
workflow.add_edge("finalize", END)
# Compile the graph
app = workflow.compile()
Running the Pipeline
With the graph compiled, you can now run the extraction pipeline on real data. Here is how to execute it with sample inputs.
def run_extraction(raw_text: str) -> dict:
"""Run the extraction pipeline on raw input text."""
initial_state = {
"raw_input": raw_text,
"source_type": "",
"preprocessed_text": "",
"extracted_data": {},
"validation_errors": [],
"confidence_score": 0.0,
"retry_count": 0,
"final_output": None,
"needs_human_review": False
}
result = app.invoke(initial_state)
return result["final_output"]
# Example 1: Extract data from an email
sample_email = """
From: Sarah Johnson <sarah.johnson@techcorp.com>
To: dev-team@techcorp.com
Subject: Q4 Project Update and Next Steps
Hi team,
I wanted to share a quick update on the Q4 migration project. We have
successfully completed the database migration and are now in the testing phase.
Action items:
1. John needs to complete the API integration tests by Friday
2. Maria should prepare the rollback documentation
3. Everyone should review the new deployment guide
Overall, things are progressing well. Let me know if you have any questions.
Best regards,
Sarah Johnson
Senior Project Manager
"""
result = run_extraction(sample_email)
print("=== Email Extraction Result ===")
print(f"Source Type: {result['source_type']}")
print(f"Confidence: {result['confidence']}")
print(f"Validation Passed: {result['validation_passed']}")
print(f"Extracted Data: {result['data']}")
# Example 2: Extract data from an invoice
sample_invoice = """
INVOICE
Vendor: Cloud Services Inc.
Invoice Number: INV-2024-0892
Date: March 15, 2024
Due Date: April 14, 2024
Line Items:
1. Enterprise Cloud Hosting - $2,400.00
2. SSL Certificate (Annual) - $199.00
3. Technical Support Package - $500.00
Total Amount Due: $3,099.00
Currency: USD
Payment Terms: Net 30
"""
result = run_extraction(sample_invoice)
print("\n=== Invoice Extraction Result ===")
print(f"Source Type: {result['source_type']}")
print(f"Confidence: {result['confidence']}")
print(f"Validation Passed: {result['validation_passed']}")
print(f"Extracted Data: {result['data']}")
Adding Human-in-the-Loop Review
For cases where extraction confidence is low or validation repeatedly fails, you can add a human review checkpoint. LangGraph supports interrupting execution at specific nodes, allowing a human to review and modify the state before resuming.
from langgraph.checkpoint.memory import MemorySaver
# Recompile with checkpointing for human-in-the-loop
workflow_with_checkpoint = StateGraph(ExtractionState)
workflow_with_checkpoint.add_node("classify", classify_source)
workflow_with_checkpoint.add_node("preprocess", preprocess_text)
workflow_with_checkpoint.add_node("extract", extract_data)
workflow_with_checkpoint.add_node("validate", validate_extraction)
workflow_with_checkpoint.add_node("retry", handle_retry)
workflow_with_checkpoint.add_node("finalize", finalize_output)
workflow_with_checkpoint.set_entry_point("classify")
workflow_with_checkpoint.add_edge("classify", "preprocess")
workflow_with_checkpoint.add_edge("preprocess", "extract")
workflow_with_checkpoint.add_edge("extract", "validate")
workflow_with_checkpoint.add_conditional_edges(
"validate",
should_retry_or_finalize,
{"retry": "retry", "finalize": "finalize"}
)
workflow_with_checkpoint.add_conditional_edges(
"retry",
route_after_retry,
{"extract": "extract", "finalize": "finalize"}
)
workflow_with_checkpoint.add_edge("finalize", END)
# Compile with interrupt before finalize when human review is needed
memory_saver = MemorySaver()
app_with_review = workflow_with_checkpoint.compile(
checkpointer=memory_saver,
interrupt_before=["finalize"]
)
def run_with_human_review(raw_text: str, thread_id: str = "default"):
"""Run pipeline with optional human review checkpoint."""
initial_state = {
"raw_input": raw_text,
"source_type": "",
"preprocessed_text": "",
"extracted_data": {},
"validation_errors": [],
"confidence_score": 0.0,
"retry_count": 0,
"final_output": None,
"needs_human_review": False
}
config = {"configurable": {"thread_id": thread_id}}
# Run until interruption or completion
result = app_with_review.invoke(initial_state, config=config)
# Check if human review is needed
if result.get("needs_human_review", False):
print("Human review required. Current state:")
print(f" Source type: {result['source_type']}")
print(f" Extracted data: {result['extracted_data']}")
print(f" Errors: {result['validation_errors']}")
# In a real app, you would present this to a human reviewer
# and let them modify the state before resuming
# Simulate human correction
human_correction = input("Enter corrected data (or press Enter to accept): ")
if human_correction:
# Update state with human input
app_with_review.update_state(
config,
{"extracted_data": eval(human_correction), "validation_errors": []}
)
# Resume execution
result = app_with_review.invoke(None, config=config)
return result["final_output"]
Batch Processing Multiple Documents
In production scenarios, you often need to process many documents. Here is a batch processing wrapper that handles multiple inputs efficiently.
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
def batch_extract(documents: List[str], max_workers: int = 4) -> List[dict]:
"""Process multiple documents in parallel."""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_doc = {
executor.submit(run_extraction, doc): idx
for idx, doc in enumerate(documents)
}
for future in as_completed(future_to_doc):
idx = future_to_doc[future]
try:
result = future.result()
results.append({"index": idx, "status": "success", "output": result})
except Exception as e:
results.append({"index": idx, "status": "error", "error": str(e)})
# Sort by original index
results.sort(key=lambda x: x["index"])
return results
def export_results(results: List[dict], output_file: str = "extraction_results.json"):
"""Export extraction results to a JSON file."""
with open(output_file, "w") as f:
json.dump(results, f, indent=2)
print(f"Results exported to {output_file}")
# Example batch processing
documents = [sample_email, sample_invoice]
batch_results = batch_extract(documents)
export_results(batch_results)
Best Practices for Production Pipelines
Building a robust extraction pipeline requires attention to several engineering and operational considerations. The following best practices will help you move from a prototype to a production-ready system.
Design for Schema Evolution
Document formats change over time. Design your Pydantic models with optional fields and version them explicitly. When schemas change, use migration logic to handle previously extracted data.
from pydantic import BaseModel, Field
from typing import Optional
class InvoiceDataV2(BaseModel):
vendor_name: str = Field(description="Name of the vendor")
invoice_number: str = Field(description="Invoice identifier")
invoice_date: str = Field(description="Date of the invoice")
total_amount: str = Field(description="Total amount due")
currency: str = Field(default="USD", description="Currency code")
line_items: List[Dict[str, str]] = Field(default_factory=list)
due_date: str = Field(default="Not specified")
tax_amount: Optional[str] = Field(default=None, description="Tax amount if listed")
payment_status: Optional[str] = Field(default=None, description="Paid, unpaid, or partial")
schema_version: str = Field(default="2.0", description="Schema version identifier")
Implement Comprehensive Logging
Every node should log its inputs, outputs, and any errors. This is critical for debugging extraction failures in production.
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("extraction_pipeline")
def extract_data_with_logging(state: ExtractionState) -> dict:
logger.info(f"Starting extraction for source type: {state['source_type']}")
logger.debug(f"Input text length: {len(state['preprocessed_text'])} chars")
try:
result = extract_data(state)
logger.info(
f"Extraction completed. Confidence: {result['confidence_score']}, "
f"Fields extracted: {list(result['extracted_data'].keys())}"
)
return result
except Exception as e:
logger.error(f"Extraction failed: {e}", exc_info=True)
raise
Use Cost-Aware Model Selection
Not every extraction requires the most powerful model. Use smaller, cheaper models for classification and simple fields, and reserve larger models for complex reasoning tasks.
def get_model_for_task(task: str):
"""Select an appropriate model based on the task complexity."""
model_map = {
"classify": "gpt-4o-mini", # Simple classification
"extract_simple": "gpt-4o-mini", # Straightforward field extraction
"extract_complex": "gpt-4o", # Complex reasoning required
"validate": "gpt-4o-mini", # Rule-based validation
}
model_name = model_map.get(task, "gpt-4o-mini")
return ChatOpenAI(model=model_name, temperature=0)
Cache Extraction Results
For documents that may be processed multiple times, implement caching to avoid redundant LLM calls and reduce costs.
import hashlib
def get_cache_key(text: str, source_type: str) -> str:
"""Generate a deterministic cache key for extraction results."""
content_hash = hashlib.sha256(f"{source_type}:{text}".encode()).hexdigest()
return f"extraction:{content_hash}"
# Use Redis or any cache backend in production
extraction_cache = {}
def extract_with_cache(state: ExtractionState) -> dict:
cache_key = get_cache_key(state["preprocessed_text"], state["source_type"])
if cache_key in extraction_cache:
logger.info(f"Cache hit for key: {cache_key}")
return extraction_cache[cache_key]
result = extract_data(state)
extraction_cache[cache_key] = result
return result
Monitor Pipeline Metrics
Track key metrics such as extraction success rate, average confidence scores, validation failure rates, and processing time per document. These metrics help identify bottlenecks and degradation over time.
import time
from collections import defaultdict
pipeline_metrics = defaultdict(list)
def track_metrics(node_name: str):
"""Decorator to track execution time and success rate per node."""
def decorator(func):
def wrapper(state):
start = time.time()
try:
result = func(state)
duration = time.time() - start
pipeline_metrics[f"{node_name}_duration"].append(duration)
pipeline_metrics[f"{node_name}_success"].append(1)
return result
except Exception as e:
duration = time.time() - start
pipeline_metrics[f"{node_name}_duration"].append(duration)
pipeline_metrics[f"{node_name}_success"].append(0)
raise
return wrapper
return decorator
def print_metrics():
"""Print aggregated pipeline metrics."""
for metric, values in pipeline_metrics.items():
avg = sum(values) / len(values)
print(f"{metric}: count={len(values)}, avg={avg:.4f}")
Handle Edge Cases Gracefully
Real-world data is messy. Your pipeline should handle empty inputs, extremely long documents, non-text content, and malformed data without crashing. Always wrap extraction logic in try-catch blocks and provide meaningful fallback values.
Conclusion
Building a data extraction pipeline with LangGraph gives you a powerful, flexible, and production-ready framework for transforming unstructured data into structured insights. By leveraging graph-based workflows, you gain fine-grained control over every stage of the extraction process—from classification and preprocessing through extraction, validation, and finalization. The conditional routing capabilities enable intelligent retry logic and human-in-the-loop review, while the stateful design ensures context is preserved across all steps. By following the best practices outlined in this tutorial—schema evolution, comprehensive logging, cost-aware model selection, caching, and metrics monitoring—you can build extraction pipelines that scale reliably in production environments. Start with the basic pipeline structure provided here, then iteratively add complexity as your extraction requirements grow, always keeping observability and error handling at the forefront of your design.