How to Extract Structured Data from Documents with LangExtract and Together AI

mkdir langextract-project
cd langextract-project

python3 -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

pip install langextract

pip install langextract-litellm

response = litellm.completion(
    model=self.model_id,
    messages=messages,
    **self.provider_kwargs,
)

# Filter out non-serializable kwargs that LiteLLM doesn't understand
filtered_kwargs = {
    k: v for k, v in self.provider_kwargs.items()
    if k in ('temperature', 'max_tokens', 'top_p', 'frequency_penalty',
             'presence_penalty', 'timeout', 'api_base', 'api_key',
             'stop', 'n', 'stream', 'response_format')
}
 
response = litellm.completion(
    model=self.model_id,
    messages=messages,
    **filtered_kwargs,
)

export TOGETHER_AI_API_KEY="your-api-key-here"

$env:TOGETHER_AI_API_KEY = "your-api-key-here"

# Install Ollama (macOS)
brew install ollama
 
# Pull Llama 3.1 8B (runs on most laptops)
ollama pull llama3.1:8b
 
# Or pull a larger model if you have GPU
ollama pull llama3.1:70b

# Instead of Together AI:
# model_id="litellm/together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo"
 
# Use local Ollama:
model_id="litellm/ollama/llama3.1:8b"

import langextract as lx
import textwrap
 
# 1. Define what you want to extract
prompt = textwrap.dedent("""\
    Extract key information from this text.
    Identify methods, metrics, comparisons, and applications.
    Use exact text from the source. Do not paraphrase.
    Extract in order of appearance.""")
 
# 2. Provide an example to guide the model
examples = [
    lx.data.ExampleData(
        text="We propose FastNet, achieving 95% accuracy on ImageNet, 2x faster than ResNet. Applications include medical imaging.",
        extractions=[
            lx.data.Extraction(
                extraction_class="method",
                extraction_text="FastNet",
                attributes={"type": "model"}
            ),
            lx.data.Extraction(
                extraction_class="metric",
                extraction_text="95% accuracy on ImageNet",
                attributes={"benchmark": "ImageNet", "value": "95%"}
            ),
            lx.data.Extraction(
                extraction_class="comparison",
                extraction_text="2x faster than ResNet",
                attributes={"baseline": "ResNet", "improvement": "2x"}
            ),
            lx.data.Extraction(
                extraction_class="application",
                extraction_text="medical imaging",
                attributes={"domain": "healthcare"}
            ),
        ]
    )
]

# 3. Your input document
input_text = """
We introduce MAXS (Meta-Adaptive eXploration Strategy), a framework
that enables LLM agents to look ahead before committing to actions.
On five reasoning benchmarks, MAXS achieves 63.46% average accuracy
compared to 52.93% for Chain-of-Thought, while using 100x fewer
tokens than Monte Carlo Tree Search. The framework supports tool use
including code execution and web search.
"""
 
# 4. Run the extraction
result = lx.extract(
    text_or_documents=input_text,
    prompt_description=prompt,
    examples=examples,
    model_id="litellm/together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo",
)
 
# 5. Print results
print(f"Found {len(result.extractions)} extractions:\n")
for ext in result.extractions:
    print(f"[{ext.extraction_class}] {ext.extraction_text}")
    if ext.attributes:
        for key, value in ext.attributes.items():
            print(f"    {key}: {value}")
    print()

python extract.py

# Save results to JSONL
lx.io.save_annotated_documents([result],
    output_name="extractions.jsonl",
    output_dir=".")
 
# Generate visualization
html_content = lx.visualize("extractions.jsonl")
with open("visualization.html", "w") as f:
    f.write(html_content if isinstance(html_content, str) else html_content.data)
 
print("Open visualization.html in your browser")

prompt = """\
    Extract research findings from AI paper abstracts.
    - methods: The proposed technique or model name
    - metrics: Quantitative results with numbers and benchmarks
    - comparisons: How it compares to baselines
    - limitations: Any mentioned drawbacks or constraints"""
 
examples = [
    lx.data.ExampleData(
        text="Our model achieves 92% F1 on SQuAD, outperforming BERT by 3 points. Training requires 8 GPUs for 2 days.",
        extractions=[
            lx.data.Extraction(
                extraction_class="metric",
                extraction_text="92% F1 on SQuAD",
                attributes={"benchmark": "SQuAD", "metric_type": "F1", "value": "92%"}
            ),
            lx.data.Extraction(
                extraction_class="comparison",
                extraction_text="outperforming BERT by 3 points",
                attributes={"baseline": "BERT", "delta": "+3 points"}
            ),
            lx.data.Extraction(
                extraction_class="limitation",
                extraction_text="Training requires 8 GPUs for 2 days",
                attributes={"type": "compute_cost"}
            ),
        ]
    )
]

prompt = """\
    Extract key clauses from legal contracts.
    - party: Named entities (companies, individuals)
    - obligation: What someone must do
    - date: Deadlines and time periods
    - penalty: Consequences for non-compliance"""
 
examples = [
    lx.data.ExampleData(
        text="Acme Corp shall deliver the software by December 31, 2026. Late delivery incurs a 5% penalty per week.",
        extractions=[
            lx.data.Extraction(
                extraction_class="party",
                extraction_text="Acme Corp",
                attributes={"role": "obligor"}
            ),
            lx.data.Extraction(
                extraction_class="obligation",
                extraction_text="shall deliver the software",
                attributes={"type": "delivery"}
            ),
            lx.data.Extraction(
                extraction_class="date",
                extraction_text="December 31, 2026",
                attributes={"type": "deadline"}
            ),
            lx.data.Extraction(
                extraction_class="penalty",
                extraction_text="5% penalty per week",
                attributes={"trigger": "late delivery"}
            ),
        ]
    )
]

prompt = """\
    Extract medical information from clinical notes.
    - medication: Drug names
    - dosage: Amounts and frequencies
    - symptom: Patient complaints or findings
    - diagnosis: Identified conditions"""
 
examples = [
    lx.data.ExampleData(
        text="Patient reports persistent headache for 3 days. Prescribed ibuprofen 400mg twice daily. Suspected tension headache.",
        extractions=[
            lx.data.Extraction(
                extraction_class="symptom",
                extraction_text="persistent headache for 3 days",
                attributes={"duration": "3 days"}
            ),
            lx.data.Extraction(
                extraction_class="medication",
                extraction_text="ibuprofen",
                attributes={"type": "NSAID"}
            ),
            lx.data.Extraction(
                extraction_class="dosage",
                extraction_text="400mg twice daily",
                attributes={"amount": "400mg", "frequency": "twice daily"}
            ),
            lx.data.Extraction(
                extraction_class="diagnosis",
                extraction_text="tension headache",
                attributes={"certainty": "suspected"}
            ),
        ]
    )
]

for ext in result.extractions:
    print(f"Text: {ext.extraction_text}")
    print(f"Start: {ext.start_offset}, End: {ext.end_offset}")
 
    # Verify by slicing original
    original_slice = input_text[ext.start_offset:ext.end_offset]
    print(f"Verified: {original_slice == ext.extraction_text}")

examples = [
    lx.data.ExampleData(
        text="Revenue increased 23% to $4.2 billion in Q3 2026.",
        extractions=[
            lx.data.Extraction(
                extraction_class="metric",
                extraction_text="$4.2 billion",
                attributes={
                    "metric_type": "revenue",
                    "period": "Q3 2026",
                    "confidence": "high"  # Model self-reports confidence
                }
            ),
        ]
    )
]

def process_with_review(result, confidence_threshold=0.8):
    approved = []
    needs_review = []
 
    for ext in result.extractions:
        confidence = ext.attributes.get("confidence", "medium")
        confidence_score = {"high": 0.9, "medium": 0.7, "low": 0.5}.get(confidence, 0.5)
 
        if confidence_score >= confidence_threshold:
            approved.append(ext)
        else:
            needs_review.append({
                "extraction": ext,
                "source_text": input_text[ext.start_offset:ext.end_offset],
                "reason": f"Confidence: {confidence}"
            })
 
    return approved, needs_review
 
approved, flagged = process_with_review(result)
print(f"Auto-approved: {len(approved)}, Needs review: {len(flagged)}")

# Enable warnings for misaligned examples
import warnings
warnings.filterwarnings("default")
 
result = lx.extract(...)  # Warnings will show if examples have issues

import time
from functools import wraps
 
def retry_with_backoff(max_retries=3, base_delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
 
@retry_with_backoff(max_retries=3)
def extract_with_retry(text, prompt, examples, model_id):
    return lx.extract(
        text_or_documents=text,
        prompt_description=prompt,
        examples=examples,
        model_id=model_id,
    )

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((TimeoutError, ConnectionError))
)
def robust_extract(text, prompt, examples, model_id):
    return lx.extract(text_or_documents=text, prompt_description=prompt,
                      examples=examples, model_id=model_id)

result = lx.extract(
    text_or_documents=documents,  # List of texts or URLs
    prompt_description=prompt,
    examples=examples,
    model_id="litellm/together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo",
    max_workers=10,  # Parallel processing
    extraction_passes=2,  # Multiple passes for better recall
)

# Estimate tokens before running
from langextract.tokenizer import count_tokens
token_count = count_tokens(your_text)
estimated_cost = token_count * 0.0001  # Adjust for your model pricing

import json
import hashlib
 
def get_cached_or_extract(text, cache_dir="./cache"):
    text_hash = hashlib.md5(text.encode()).hexdigest()
    cache_file = f"{cache_dir}/{text_hash}.json"
 
    if os.path.exists(cache_file):
        with open(cache_file) as f:
            return json.load(f)
 
    result = lx.extract(text_or_documents=text, ...)
 
    with open(cache_file, "w") as f:
        json.dump(result.to_dict(), f)
 
    return result

from pydantic import BaseModel, field_validator
from datetime import datetime
from typing import Optional
 
class MetricExtraction(BaseModel):
    text: str
    benchmark: str
    value: float
    period: Optional[str] = None
 
    @field_validator('value', mode='before')
    @classmethod
    def parse_percentage(cls, v):
        if isinstance(v, str):
            return float(v.strip('%')) / 100
        return v
 
class ContractExtraction(BaseModel):
    party: str
    obligation: str
    deadline: datetime
 
    @field_validator('deadline', mode='before')
    @classmethod
    def parse_date(cls, v):
        if isinstance(v, str):
            # Handle common date formats
            for fmt in ['%Y-%m-%d', '%B %d, %Y', '%m/%d/%Y']:
                try:
                    return datetime.strptime(v, fmt)
                except ValueError:
                    continue
        return v
 
# Validate extractions
def validate_extraction(ext, model_class):
    try:
        return model_class(**ext.attributes, text=ext.extraction_text)
    except Exception as e:
        print(f"Validation failed: {e}")
        return None
 
validated = [validate_extraction(e, MetricExtraction)
             for e in result.extractions
             if e.extraction_class == "metric"]

EXTRACTION_SCHEMA_VERSION = "1.2"
 
# Schema changelog:
# v1.2: Added "limitation" class, improved metric attributes
# v1.1: Added confidence scores to all extractions
# v1.0: Initial schema with method, metric, comparison
 
examples = [
    lx.data.ExampleData(...)
]
 
# Store version with extractions for migration support
def save_extractions(result, version=EXTRACTION_SCHEMA_VERSION):
    return {
        "schema_version": version,
        "extracted_at": datetime.now().isoformat(),
        "extractions": [e.to_dict() for e in result.extractions]
    }

import pandas as pd
 
def extractions_to_dataframe(result):
    rows = []
    for ext in result.extractions:
        row = {
            "class": ext.extraction_class,
            "text": ext.extraction_text,
            "start": ext.start_offset,
            "end": ext.end_offset,
            **ext.attributes  # Flatten attributes into columns
        }
        rows.append(row)
    return pd.DataFrame(rows)
 
df = extractions_to_dataframe(result)
print(df[df["class"] == "metric"][["text", "benchmark", "value"]])

import sqlite3
 
def create_extractions_table(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS extractions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            document_id TEXT,
            extraction_class TEXT,
            extraction_text TEXT,
            start_offset INTEGER,
            end_offset INTEGER,
            attributes JSON,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
 
def insert_extractions(conn, doc_id, result):
    for ext in result.extractions:
        conn.execute("""
            INSERT INTO extractions (document_id, extraction_class, extraction_text,
                                     start_offset, end_offset, attributes)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (doc_id, ext.extraction_class, ext.extraction_text,
              ext.start_offset, ext.end_offset, json.dumps(ext.attributes)))
    conn.commit()
 
# Usage
conn = sqlite3.connect("extractions.db")
create_extractions_table(conn)
insert_extractions(conn, "paper_001", result)
 
# Query: Find all metrics above 90%
df = pd.read_sql("""
    SELECT * FROM extractions
    WHERE extraction_class = 'metric'
    AND json_extract(attributes, '$.value') > 90
""", conn)

# Export to JSON for dashboards
with open("extractions.json", "w") as f:
    json.dump([{
        "class": e.extraction_class,
        "text": e.extraction_text,
        **e.attributes
    } for e in result.extractions], f)
 
# Export to CSV for spreadsheets
df.to_csv("extractions.csv", index=False)