Statement Extractor Documentation

pip install corp-extractor

# Example for CUDA 12.1
pip install torch --index-url https://download.pytorch.org/whl/cu121
pip install corp-extractor

pip install corp-extractor

from statement_extractor import extract_statements

text = "Apple Inc. acquired Beats Electronics for $3 billion in May 2014."
statements = extract_statements(text)

for stmt in statements:
    print(f"{stmt.subject.text} ({stmt.subject.type}) -> {stmt.predicate} -> {stmt.object.text}")

Apple Inc. (ORG) -> acquired -> Beats Electronics
Apple Inc. (ORG) -> paid -> $3 billion
Beats Electronics (ORG) -> acquisition price -> $3 billion

for stmt in statements:
    print(f"{stmt.subject.text} -> {stmt.predicate} -> {stmt.object.text}")
    print(f"  method: {stmt.extraction_method}")  # hybrid, gliner, or model
    print(f"  confidence: {stmt.confidence_score:.2f}")

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()
ctx = pipeline.process("Amazon CEO Andy Jassy announced plans to hire workers.")

# Access fully qualified names (e.g., "Andy Jassy (CEO, Amazon)")
for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} --[{stmt.statement.predicate}]--> {stmt.object_fqn}")

    # Access labels (sentiment, etc.)
    for label in stmt.labels:
        print(f"  {label.label_type}: {label.label_value}")

# Full pipeline
corp-extractor pipeline "Amazon CEO Andy Jassy announced..."

# Run specific stages only
corp-extractor pipeline -f article.txt --stages 1-3

# Process documents and URLs (v0.7.0)
corp-extractor document process article.txt
corp-extractor document process https://example.com/article
corp-extractor document process report.pdf --use-ocr

from statement_extractor import extract_statements, PredicateTaxonomy, ExtractionOptions

# Define your domain's canonical predicates
taxonomy = PredicateTaxonomy(predicates=[
    "acquired", "founded", "works_for", "headquartered_in",
    "invested_in", "partnered_with", "announced"
])

options = ExtractionOptions(predicate_taxonomy=taxonomy)

text = "Google bought YouTube for $1.65 billion in 2006."
result = extract_statements(text, options)

for stmt in result:
    print(f"{stmt.predicate} -> {stmt.canonical_predicate}")
    # Output: bought -> acquired

# Using uv (recommended)
uv tool install "corp-extractor[embeddings]"

# Using pipx
pipx install "corp-extractor[embeddings]"

# Using pip
pip install "corp-extractor[embeddings]"

uvx corp-extractor split "Apple announced a new iPhone."

# Extract from text argument
corp-extractor split "Apple Inc. announced the iPhone 15."

# Extract from file
corp-extractor split -f article.txt

# Pipe from stdin
cat article.txt | corp-extractor split -

# Output as JSON
corp-extractor split "Tim Cook is CEO of Apple." --json

# Output as XML
corp-extractor split "Tim Cook is CEO of Apple." --xml

# Verbose output with confidence scores
corp-extractor split -f article.txt --verbose

# Use more beams for better quality
corp-extractor split -f article.txt --beams 8

# Run all 5 stages
corp-extractor pipeline "Amazon CEO Andy Jassy announced plans to hire workers."

# Run from file
corp-extractor pipeline -f article.txt

# Run specific stages
corp-extractor pipeline "..." --stages 1-3
corp-extractor pipeline "..." --stages 1,2,5

# Skip specific stages
corp-extractor pipeline "..." --skip-stages 4,5

# Enable specific plugins only
corp-extractor pipeline "..." --plugins gleif,companies_house

# Disable specific plugins
corp-extractor pipeline "..." --disable-plugins sec_edgar

# Output formats
corp-extractor pipeline "..." -o json
corp-extractor pipeline "..." -o yaml
corp-extractor pipeline "..." -o triples

# List all plugins
corp-extractor plugins list

# List plugins for a specific stage
corp-extractor plugins list --stage 3

# Get details about a plugin
corp-extractor plugins info gleif_qualifier
corp-extractor plugins info person_qualifier

Stage 1: Splitting
----------------------------------------
  t5_gemma_splitter  [priority: 100]

Stage 2: Extraction
----------------------------------------
  gliner2_extractor  [priority: 100]

Stage 3: Entity Qualification
----------------------------------------
  person_qualifier (PERSON)  [priority: 100]
  embedding_company_qualifier (ORG)  [priority: 5]

Stage 4: Labeling
----------------------------------------
  sentiment_labeler  [priority: 100]
  confidence_labeler  [priority: 100]
  relation_type_labeler  [priority: 100]

Stage 5: Taxonomy
----------------------------------------
  embedding_taxonomy_classifier  [priority: 100]

Extracted 2 statement(s):

--------------------------------------------------------------------------------
1. Andy Jassy (CEO, Amazon)
   --[announced]-->
   plans to hire workers
--------------------------------------------------------------------------------

{
  "statement_count": 2,
  "labeled_statements": [
    {
      "subject": {"text": "Andy Jassy", "type": "PERSON", "fqn": "Andy Jassy (CEO, Amazon)"},
      "predicate": "announced",
      "object": {"text": "plans to hire workers", "type": "EVENT"},
      "labels": {"sentiment": "positive"}
    }
  ]
}

Andy Jassy (CEO, Amazon)	announced	plans to hire workers
Amazon	has CEO	Andy Jassy (CEO, Amazon)

# Process all .txt files
for f in *.txt; do
  echo "=== $f ==="
  corp-extractor pipeline -f "$f" -o json > "${f%.txt}.json"
done

# Extract just predicates
corp-extractor split "Your text" --json | jq '.statements[].predicate'

# Filter high-confidence statements
corp-extractor split -f article.txt --json | jq '.statements[] | select(.confidence_score > 0.8)'

# Get FQNs from pipeline
corp-extractor pipeline "Your text" -o json | jq '.labeled_statements[].subject.fqn'

# Process local files
corp-extractor document process article.txt
corp-extractor document process report.txt --title "Annual Report" --year 2024

# Process URLs (web pages and PDFs)
corp-extractor document process https://example.com/article
corp-extractor document process https://example.com/report.pdf --use-ocr

# Configure chunking
corp-extractor document process article.txt --max-tokens 500 --overlap 50

# Preview chunking without extraction
corp-extractor document chunk article.txt --max-tokens 500

# Output formats
corp-extractor document process article.txt -o json
corp-extractor document process article.txt -o triples

# Show database status
corp-extractor db status

# Search for an organization
corp-extractor db search "Microsoft"
corp-extractor db search "Barclays" --source companies_house

# Search for a person (v0.9.0)
corp-extractor db search-people "Tim Cook"
corp-extractor db search-people "Elon Musk" --top-k 5

# Search for roles (v0.9.4)
corp-extractor db search-roles "CEO"
corp-extractor db search-roles "Chief Financial Officer"

# Search for locations (v0.9.4)
corp-extractor db search-locations "California"
corp-extractor db search-locations "Germany" --type country

# Import organizations from data sources
corp-extractor db import-gleif --download
corp-extractor db import-sec --download           # Bulk data (~100K+ filers)
corp-extractor db import-companies-house --download
corp-extractor db import-wikidata --limit 50000   # SPARQL-based

# Import notable people (v0.9.0)
corp-extractor db import-people --type executive --limit 5000
corp-extractor db import-people --all --limit 10000  # All person types

# Import from Wikidata dump (v0.9.1) - avoids SPARQL timeouts
corp-extractor db import-wikidata-dump --download --limit 50000
corp-extractor db import-wikidata-dump --dump /path/to/dump.bz2 --people --no-orgs

# Download/upload from HuggingFace Hub
corp-extractor db download                        # Lite version (default)
corp-extractor db download --full                 # Full version with metadata
corp-extractor db upload                          # Upload with all variants

# Migrate from old schema (companies.db → entities.db)
corp-extractor db migrate companies.db --rename-file

# Migrate to v2 normalized schema (v0.9.4)
corp-extractor db migrate-v2 entities.db entities-v2.db
corp-extractor db migrate-v2 entities.db entities-v2.db --resume  # Resume interrupted

# Generate int8 scalar embeddings (v0.9.4) - 75% smaller
corp-extractor db backfill-scalar
corp-extractor db backfill-scalar --skip-generate  # Only quantize existing

# Local database management
corp-extractor db create-lite entities.db         # Create lite version
corp-extractor db compress entities.db            # Compress with gzip

# Download and import (~100GB dump file)
corp-extractor db import-wikidata-dump --download --limit 50000

# Import only people
corp-extractor db import-wikidata-dump --download --people --no-orgs --limit 100000

# Import only organizations
corp-extractor db import-wikidata-dump --download --orgs --no-people --limit 100000

# Import only locations (v0.9.4)
corp-extractor db import-wikidata-dump --dump dump.bz2 --locations --no-people --no-orgs

# Use existing dump file
corp-extractor db import-wikidata-dump --dump /path/to/latest-all.json.bz2

brew install aria2   # macOS
apt install aria2    # Ubuntu/Debian

corp-extractor db import-people --type executive -v

from statement_extractor import extract_statements

# Use default beam search settings
result = extract_statements("Apple announced a new iPhone.")

# Customize beam search
result = extract_statements(
    "Apple announced a new iPhone.",
    num_beams=6,
    diversity_penalty=1.5
)

from statement_extractor import extract_statements

# Only return statements with confidence >= 0.7
result = extract_statements(
    "Apple CEO Tim Cook announced the iPhone 15.",
    min_confidence=0.7
)

# Access individual scores
for stmt in result:
    print(f"{stmt.subject.text} -> {stmt.predicate} -> {stmt.object.text}")
    print(f"  Confidence: {stmt.confidence_score:.2f}")

# Merge all beams (default)
result = extract_statements(text, beam_strategy="merge")

# Use only the best beam
result = extract_statements(text, beam_strategy="best")

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()
ctx = pipeline.process("Amazon CEO Andy Jassy announced plans to hire workers.")

# All matching relations are returned, sorted by confidence
for stmt in ctx.statements:
    print(f"{stmt.subject.text} --[{stmt.predicate}]--> {stmt.object.text}")
    print(f"  Category: {stmt.predicate_category}")
    print(f"  Confidence: {stmt.confidence_score:.2f}")

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

config = PipelineConfig(
    extractor_options={"predicates_file": "/path/to/custom_predicates.json"}
)
pipeline = ExtractionPipeline(config)

corp-extractor pipeline "..." --predicates-file custom_predicates.json

from statement_extractor.pipeline import PipelineConfig, ExtractionPipeline

# Run with specific plugins disabled
config = PipelineConfig(
    disabled_plugins={"mnli_taxonomy_classifier"}  # Use embedding classifier instead
)
pipeline = ExtractionPipeline(config)
ctx = pipeline.process(text)

from statement_extractor.document import DocumentPipeline

pipeline = DocumentPipeline()

# Process a web page
ctx = await pipeline.process_url("https://example.com/article")

# Process a PDF with OCR
from statement_extractor.document import URLLoaderConfig
config = URLLoaderConfig(use_ocr=True)
ctx = await pipeline.process_url("https://example.com/report.pdf", config)

# Download the pre-built database
corp-extractor db download

# Check what's in it
corp-extractor db status

# Search for organizations
corp-extractor db search "Microsoft"

# Search for people
corp-extractor db search-people "Tim Cook"

# Search for roles (v0.9.4)
corp-extractor db search-roles "CEO"

# Search for locations (v0.9.4)
corp-extractor db search-locations "California"

# Download lite version (default, smaller, faster)
corp-extractor db download

# Download full version (includes complete source metadata)
corp-extractor db download --full

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()
ctx = pipeline.process("Microsoft CEO Satya Nadella announced...")
# Database downloaded automatically if not present

-- Enum tables: source_types, people_types, organization_types, location_types
-- Organization: source_id (FK), entity_type_id (FK), region_id (FK to locations)
-- People: source_id (FK), person_type_id (FK), country_id (FK), known_for_role_id (FK)
-- Roles: qid, name, source_id (FK), canon_id
-- Locations: qid, name, source_id (FK), location_type_id (FK), parent_ids (hierarchy)

CREATE TABLE organizations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    qid INTEGER,                 -- Wikidata QID as integer (v0.9.4)
    name TEXT NOT NULL,
    name_normalized TEXT NOT NULL,
    source_id INTEGER NOT NULL,  -- FK to source_types(id)
    source_identifier TEXT NOT NULL,  -- LEI, CIK, Company Number
    region_id INTEGER,           -- FK to locations(id) (v0.9.4)
    entity_type_id INTEGER NOT NULL,  -- FK to organization_types(id)
    from_date TEXT,              -- ISO YYYY-MM-DD
    to_date TEXT,                -- ISO YYYY-MM-DD
    record TEXT NOT NULL,        -- JSON (empty in lite version)
    UNIQUE(source_identifier, source_id)
);

-- Both float32 and int8 embeddings supported (v0.9.4)
CREATE VIRTUAL TABLE organization_embeddings USING vec0(
    org_id INTEGER PRIMARY KEY, embedding float[768]
);
CREATE VIRTUAL TABLE organization_embeddings_scalar USING vec0(
    org_id INTEGER PRIMARY KEY, embedding int8[768]
);

CREATE TABLE people (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    qid INTEGER,                 -- Wikidata QID as integer (v0.9.4)
    name TEXT NOT NULL,
    name_normalized TEXT NOT NULL,
    source_id INTEGER NOT NULL,  -- FK to source_types(id)
    source_identifier TEXT NOT NULL,  -- QID, Owner CIK, Person number
    country_id INTEGER,          -- FK to locations(id) (v0.9.4)
    person_type_id INTEGER NOT NULL,  -- FK to people_types(id)
    known_for_role_id INTEGER,   -- FK to roles(id) (v0.9.4)
    known_for_org TEXT DEFAULT '',
    known_for_org_id INTEGER,    -- FK to organizations(id)
    from_date TEXT,              -- Role start date (ISO)
    to_date TEXT,                -- Role end date (ISO)
    birth_date TEXT,             -- ISO YYYY-MM-DD
    death_date TEXT,             -- ISO YYYY-MM-DD
    record TEXT NOT NULL,
    UNIQUE(source_identifier, source_id, known_for_role_id, known_for_org_id)
);

CREATE VIRTUAL TABLE person_embeddings USING vec0(
    person_id INTEGER PRIMARY KEY, embedding float[768]
);
CREATE VIRTUAL TABLE person_embeddings_scalar USING vec0(
    person_id INTEGER PRIMARY KEY, embedding int8[768]
);

-- Roles table for job titles
CREATE TABLE roles (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    qid INTEGER,                 -- Wikidata QID (e.g., 484876 for CEO)
    name TEXT NOT NULL,          -- "Chief Executive Officer"
    name_normalized TEXT NOT NULL,
    source_id INTEGER NOT NULL,  -- FK to source_types(id)
    canon_id INTEGER DEFAULT NULL,
    UNIQUE(name_normalized, source_id)
);

-- Locations table for geopolitical entities
CREATE TABLE locations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    qid INTEGER,                 -- Wikidata QID (e.g., 30 for USA)
    name TEXT NOT NULL,          -- "United States", "California"
    name_normalized TEXT NOT NULL,
    source_id INTEGER NOT NULL,  -- FK to source_types(id)
    source_identifier TEXT,      -- "US", "CA"
    parent_ids TEXT,             -- JSON array of parent location IDs
    location_type_id INTEGER NOT NULL,  -- FK to location_types(id)
    UNIQUE(source_identifier, source_id)
);

from statement_extractor.database import OrganizationDatabase

db = OrganizationDatabase()

# Search by name (hybrid: text + embedding)
matches = db.search_by_name("Microsoft Corporation", top_k=5)
for match in matches:
    print(f"{match.company.name} ({match.company.source}:{match.company.source_id})")
    print(f"  Similarity: {match.similarity_score:.3f}")
    print(f"  Type: {match.company.entity_type}")

# Search by embedding
from statement_extractor.database import CompanyEmbedder

embedder = CompanyEmbedder()
embedding = embedder.embed("Microsoft")
matches = db.search(embedding, top_k=10, min_similarity=0.7)

from statement_extractor.database import PersonDatabase

db = PersonDatabase()

# Search by name
matches = db.search_by_name("Tim Cook", top_k=5)
for match in matches:
    print(f"{match.person.name} - {match.person.known_for_role} at {match.person.known_for_org}")
    print(f"  Wikidata: {match.person.source_id}")
    print(f"  Type: {match.person.person_type}")

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()
ctx = pipeline.process("Microsoft CEO Satya Nadella announced new AI features.")

for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} --[{stmt.statement.predicate}]--> {stmt.object_fqn}")
    # e.g., "Satya Nadella (CEO, Microsoft) --[announced]--> new AI features"

from statement_extractor.database import OrganizationDatabase, CompanyRecord, EntityType

db = OrganizationDatabase()

record = CompanyRecord(
    name="My Company Inc",
    source="custom",
    source_id="CUSTOM001",
    region="US",
    entity_type=EntityType.business,
    record={"custom_field": "value"},
)
db.add_record(record)

# Companies House - UK companies (5.5M records)
corp-extractor db import-companies-house --download

# GLEIF - Global LEI data (2.6M records)
corp-extractor db import-gleif --download
corp-extractor db import-gleif /path/to/lei-data.json --limit 50000

# SEC Edgar - US public companies (73K filers)
corp-extractor db import-sec --download

# Wikidata organizations via SPARQL (1.5M records)
corp-extractor db import-wikidata --limit 50000

# Import by person type
corp-extractor db import-people --type executive --limit 5000
corp-extractor db import-people --type politician --limit 5000
corp-extractor db import-people --type athlete --limit 5000

# Import all person types
corp-extractor db import-people --all --limit 50000

# Skip existing records (faster for incremental updates)
corp-extractor db import-people --type executive --skip-existing

# Fetch role start/end dates (slower, queries per person)
corp-extractor db import-people --type executive --enrich-dates

# Download and import from Wikidata dump (~100GB compressed)
corp-extractor db import-wikidata-dump --download --limit 100000

# Import from local dump file
corp-extractor db import-wikidata-dump --dump /path/to/latest-all.json.bz2

# Import only people (no organizations)
corp-extractor db import-wikidata-dump --dump dump.bz2 --people --no-orgs

# Import only locations (countries, states, cities) - v0.9.4
corp-extractor db import-wikidata-dump --dump dump.bz2 --locations --no-people --no-orgs

# Resume interrupted import
corp-extractor db import-wikidata-dump --dump dump.bz2 --resume

# Skip records already in database
corp-extractor db import-wikidata-dump --dump dump.bz2 --skip-updates

brew install aria2   # macOS
apt install aria2    # Ubuntu/Debian

# 1. Import from all sources
corp-extractor db import-gleif --download
corp-extractor db import-sec --download
corp-extractor db import-companies-house --download
corp-extractor db import-wikidata --limit 100000
corp-extractor db import-wikidata-dump --download --people --no-orgs --limit 100000

# 2. Link equivalent records
corp-extractor db canonicalize

# 3. Generate scalar embeddings (75% storage reduction)
corp-extractor db backfill-scalar

# 4. Check status
corp-extractor db status

# 5. Upload to HuggingFace
export HF_TOKEN="hf_..."
corp-extractor db upload

# Create new v2 database (preserves original)
corp-extractor db migrate-v2 entities.db entities-v2.db

# Resume interrupted migration
corp-extractor db migrate-v2 entities.db entities-v2.db --resume

corp-extractor db canonicalize

class CompanyRecord(BaseModel):
    name: str                    # Organization name
    source: str                  # 'gleif', 'sec_edgar', 'companies_house', 'wikipedia'
    source_id: str               # LEI, CIK, UK Company Number, or QID
    region: str                  # Country/region code
    entity_type: EntityType      # Classification
    from_date: Optional[str]     # ISO YYYY-MM-DD
    to_date: Optional[str]       # ISO YYYY-MM-DD
    record: dict[str, Any]       # Full source record (empty in lite)

    @property
    def canonical_id(self) -> str:
        return f"{self.source}:{self.source_id}"

class PersonRecord(BaseModel):
    name: str                    # Display name
    source: str                  # 'wikidata'
    source_id: str               # Wikidata QID
    country: str                 # Country code
    person_type: PersonType      # Classification
    known_for_role: str          # Primary role
    known_for_org: str           # Primary organization name
    known_for_org_id: Optional[int]  # FK to organizations
    from_date: Optional[str]     # Role start (ISO)
    to_date: Optional[str]       # Role end (ISO)
    birth_date: Optional[str]    # Birth date (ISO)
    death_date: Optional[str]    # Death date (ISO)
    record: dict[str, Any]       # Full source record

    @property
    def is_historic(self) -> bool:
        return self.death_date is not None

class CompanyMatch(BaseModel):
    company: CompanyRecord
    similarity_score: float      # 0.0 to 1.0

class PersonMatch(BaseModel):
    person: PersonRecord
    similarity_score: float      # 0.0 to 1.0
    llm_confirmed: bool          # Whether LLM validated match

from statement_extractor.database import CompanyEmbedder

embedder = CompanyEmbedder()
embedding = embedder.embed("Apple Inc")  # Returns 768-dim numpy array

Error: Database not found at ~/.cache/corp-extractor/entities.db

Error: no such module: vec0

# Import in smaller batches
corp-extractor db import-wikidata-dump --dump dump.bz2 --limit 10000 --skip-updates
# Then resume for more
corp-extractor db import-wikidata-dump --dump dump.bz2 --limit 10000 --skip-updates --resume

corp-extractor db import-wikidata-dump --dump dump.bz2 --resume

def extract_statements(
    text: str,
    options: Optional[ExtractionOptions] = None,
    **kwargs
) -> ExtractionResult:
    """
    Extract structured statements from text.

    Args:
        text: Input text to extract statements from
        options: Extraction options (or pass individual options as kwargs)
        **kwargs: Individual option overrides (num_beams, diversity_penalty, etc.)

    Returns:
        ExtractionResult containing Statement objects
    """

def extract_statements_as_json(
    text: str,
    options: Optional[ExtractionOptions] = None,
    indent: Optional[int] = 2,
    **kwargs
) -> str:
    """Returns JSON string representation of the extraction result."""

def extract_statements_as_xml(
    text: str,
    options: Optional[ExtractionOptions] = None,
    **kwargs
) -> str:
    """Returns XML string with <statements> containing <stmt> elements."""

def extract_statements_as_dict(
    text: str,
    options: Optional[ExtractionOptions] = None,
    **kwargs
) -> dict:
    """Returns dictionary representation of the extraction result."""

from statement_extractor import extract_statements, extract_statements_as_json

# Basic extraction
result = extract_statements("Apple acquired Beats for $3 billion.")
for stmt in result:
    print(f"{stmt.subject.text} -> {stmt.predicate} -> {stmt.object.text}")

# With options via kwargs
result = extract_statements(
    "Tesla announced new factories.",
    num_beams=6,
    diversity_penalty=1.5
)

# JSON output
json_str = extract_statements_as_json("OpenAI released GPT-4.", indent=2)
print(json_str)

class StatementExtractor:
    def __init__(
        self,
        model_id: str = "Corp-o-Rate-Community/statement-extractor",
        device: Optional[str] = None,
        torch_dtype: Optional[torch.dtype] = None,
        predicate_taxonomy: Optional[PredicateTaxonomy] = None,
        predicate_config: Optional[PredicateComparisonConfig] = None,
        scoring_config: Optional[ScoringConfig] = None,
    ):
        """
        Initialize the statement extractor.

        Args:
            model_id: HuggingFace model ID or local path
            device: Device to use ('cuda', 'cpu', or None for auto-detect)
            torch_dtype: Torch dtype (default: bfloat16 on GPU, float32 on CPU)
            predicate_taxonomy: Optional taxonomy for predicate normalization
            predicate_config: Configuration for predicate comparison
            scoring_config: Configuration for quality scoring
        """

    def extract(
        self,
        text: str,
        options: Optional[ExtractionOptions] = None,
    ) -> ExtractionResult:
        """Extract statements from text."""

    def extract_as_xml(
        self,
        text: str,
        options: Optional[ExtractionOptions] = None,
    ) -> str:
        """Extract statements and return raw XML output."""

    def extract_as_json(
        self,
        text: str,
        options: Optional[ExtractionOptions] = None,
        indent: Optional[int] = 2,
    ) -> str:
        """Extract statements and return JSON string."""

    def extract_as_dict(
        self,
        text: str,
        options: Optional[ExtractionOptions] = None,
    ) -> dict:
        """Extract statements and return as dictionary."""

from statement_extractor import StatementExtractor, ExtractionOptions

# Force CPU usage
extractor = StatementExtractor(device="cpu")

# Extract with custom options
options = ExtractionOptions(num_beams=6, diversity_penalty=1.2)
result = extractor.extract("Microsoft partnered with OpenAI.", options)

class ExtractionOptions(BaseModel):
    # Beam search parameters
    num_beams: int = 4                    # 1-16, beams for diverse beam search
    diversity_penalty: float = 1.0        # >= 0.0, penalty for beam diversity
    max_new_tokens: int = 2048            # 128-8192, max tokens to generate
    min_statement_ratio: float = 1.0      # >= 0.0, min statements per sentence
    max_attempts: int = 3                 # 1-10, extraction retry attempts
    deduplicate: bool = True              # Remove duplicate statements

    # Predicate taxonomy & comparison
    predicate_taxonomy: Optional[PredicateTaxonomy] = None
    predicate_config: Optional[PredicateComparisonConfig] = None

    # Scoring configuration (v0.2.0)
    scoring_config: Optional[ScoringConfig] = None

    # Pluggable canonicalization
    entity_canonicalizer: Optional[Callable[[str], str]] = None

    # Mode flags
    merge_beams: bool = True              # Merge top-N beams vs select best
    embedding_dedup: bool = True          # Use embedding similarity for dedup

class ScoringConfig(BaseModel):
    quality_weight: float = 1.0           # >= 0.0, weight for confidence scores
    coverage_weight: float = 0.5          # >= 0.0, bonus for source text coverage
    redundancy_penalty: float = 0.3       # >= 0.0, penalty for duplicate triples
    length_penalty: float = 0.1           # >= 0.0, penalty for verbosity
    min_confidence: float = 0.0           # 0.0-1.0, minimum confidence threshold
    merge_top_n: int = 3                  # 1-10, beams to merge when merge_beams=True

class PredicateTaxonomy(BaseModel):
    predicates: list[str]                 # List of canonical predicate forms
    name: Optional[str] = None            # Optional taxonomy name

    @classmethod
    def from_file(cls, path: str | Path) -> "PredicateTaxonomy":
        """Load taxonomy from a file (one predicate per line)."""

    @classmethod
    def from_list(cls, predicates: list[str], name: Optional[str] = None) -> "PredicateTaxonomy":
        """Create taxonomy from a list of predicates."""

from statement_extractor import PredicateTaxonomy, ExtractionOptions, extract_statements

# Define canonical predicates
taxonomy = PredicateTaxonomy.from_list([
    "acquired", "founded", "works_for", "located_in", "partnered_with"
])

# Use in extraction
options = ExtractionOptions(predicate_taxonomy=taxonomy)
result = extract_statements("Google bought YouTube.", options)
# predicate "bought" maps to canonical "acquired"

class PredicateComparisonConfig(BaseModel):
    embedding_model: str = "sentence-transformers/paraphrase-MiniLM-L6-v2"
    similarity_threshold: float = 0.65    # 0.0-1.0, min similarity for taxonomy match
    dedup_threshold: float = 0.65         # 0.0-1.0, min similarity for duplicates
    normalize_text: bool = True           # Lowercase and strip before embedding

class Statement(BaseModel):
    subject: Entity                              # The subject entity
    predicate: str                               # The relationship/predicate
    object: Entity                               # The object entity
    source_text: Optional[str] = None            # Original text span

    # Quality scoring fields (v0.2.0)
    confidence_score: Optional[float] = None     # 0.0-1.0, quality score (semantic + entity)
    evidence_span: Optional[tuple[int, int]] = None  # Character offsets in source
    canonical_predicate: Optional[str] = None    # Canonical form if taxonomy used

    def as_triple(self) -> tuple[str, str, str]:
        """Return as (subject, predicate, object) tuple."""

    def __str__(self) -> str:
        """Format: 'subject -- predicate --> object'"""

stmt = result.statements[0]
print(stmt.subject.text)           # "Apple Inc."
print(stmt.predicate)              # "acquired"
print(stmt.object.text)            # "Beats Electronics"
print(stmt.confidence_score)       # 0.92
print(stmt.as_triple())            # ("Apple Inc.", "acquired", "Beats Electronics")

class Entity(BaseModel):
    text: str                        # The entity text
    type: EntityType = UNKNOWN       # The entity type

    def __str__(self) -> str:
        """Format: 'text (TYPE)'"""

class EntityType(str, Enum):
    ORG = "ORG"                 # Organization
    PERSON = "PERSON"           # Person
    GPE = "GPE"                 # Geopolitical entity (country, city, state)
    LOC = "LOC"                 # Non-GPE location
    PRODUCT = "PRODUCT"         # Product
    EVENT = "EVENT"             # Event
    WORK_OF_ART = "WORK_OF_ART" # Creative work
    LAW = "LAW"                 # Legal document
    DATE = "DATE"               # Date or time
    MONEY = "MONEY"             # Monetary value
    PERCENT = "PERCENT"         # Percentage
    QUANTITY = "QUANTITY"       # Quantity or measurement
    UNKNOWN = "UNKNOWN"         # Unknown type

class ExtractionResult(BaseModel):
    statements: list[Statement] = []     # List of extracted statements
    source_text: Optional[str] = None    # Original input text

    def __len__(self) -> int:
        """Number of statements."""

    def __iter__(self):
        """Iterate over statements."""

    def to_triples(self) -> list[tuple[str, str, str]]:
        """Return all statements as (subject, predicate, object) tuples."""

result = extract_statements(text)

# Iterate directly
for stmt in result:
    print(stmt)

# Check count
print(f"Found {len(result)} statements")

# Get as simple tuples
triples = result.to_triples()

class PredicateMatch(BaseModel):
    original: str                        # The original extracted predicate
    canonical: Optional[str] = None      # Matched canonical predicate, if any
    similarity: float = 0.0              # 0.0-1.0, cosine similarity score
    matched: bool = False                # Whether a match was found above threshold

from statement_extractor import PredicateComparer, PredicateTaxonomy

taxonomy = PredicateTaxonomy.from_list(["acquired", "founded", "works_for"])
comparer = PredicateComparer(taxonomy=taxonomy)

match = comparer.match_to_canonical("bought")
print(match.original)     # "bought"
print(match.canonical)    # "acquired"
print(match.similarity)   # ~0.82
print(match.matched)      # True

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

class ExtractionPipeline:
    def __init__(self, config: Optional[PipelineConfig] = None):
        """
        Initialize the extraction pipeline.

        Args:
            config: Pipeline configuration (default: all stages enabled)
        """

    def process(self, text: str, metadata: Optional[dict] = None) -> PipelineContext:
        """
        Process text through the pipeline stages.

        Args:
            text: Input text to process
            metadata: Optional source metadata (document ID, URL, etc.)

        Returns:
            PipelineContext with results from all stages
        """

pipeline = ExtractionPipeline()
ctx = pipeline.process("Amazon CEO Andy Jassy announced plans.")

print(f"Statements: {ctx.statement_count}")
for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

from statement_extractor.pipeline import PipelineConfig

class PipelineConfig(BaseModel):
    enabled_stages: set[int] = {1, 2, 3, 4, 5}  # Stages to run (1-5)
    enabled_plugins: Optional[set[str]] = None   # Plugins to enable (None = all)
    disabled_plugins: set[str] = set()           # Plugins to disable
    fail_fast: bool = False                       # Stop on first error
    parallel_processing: bool = False             # Enable parallel processing
    max_statements: Optional[int] = None          # Limit statements processed

    # Stage-specific options
    splitter_options: dict = {}
    extractor_options: dict = {}
    qualifier_options: dict = {}
    labeler_options: dict = {}
    taxonomy_options: dict = {}

    @classmethod
    def from_stage_string(cls, stages: str, **kwargs) -> "PipelineConfig":
        """Create config from stage string like '1-3' or '1,2,5'."""

    @classmethod
    def default(cls) -> "PipelineConfig":
        """All stages enabled."""

    @classmethod
    def minimal(cls) -> "PipelineConfig":
        """Only splitting and extraction (stages 1-2)."""

# Run only stages 1-3
config = PipelineConfig(enabled_stages={1, 2, 3})

# Disable specific plugins
config = PipelineConfig(disabled_plugins={"sec_edgar_qualifier"})

# From stage string
config = PipelineConfig.from_stage_string("1-3")

from statement_extractor.pipeline import PipelineContext

class PipelineContext(BaseModel):
    # Input
    source_text: str                                    # Original input text
    source_metadata: dict = {}                          # Document metadata

    # Stage outputs
    raw_triples: list[RawTriple] = []                   # Stage 1 output
    statements: list[PipelineStatement] = []           # Stage 2 output
    canonical_entities: dict[str, CanonicalEntity] = {} # Stage 3 output
    labeled_statements: list[LabeledStatement] = []    # Stage 4 output
    taxonomy_results: dict[tuple, list[TaxonomyResult]] = {}  # Stage 5 output (multiple labels per statement)

    # Processing metadata
    processing_errors: list[str] = []
    processing_warnings: list[str] = []
    stage_timings: dict[str, float] = {}

    @property
    def statement_count(self) -> int:
        """Number of statements in final output."""

    @property
    def has_errors(self) -> bool:
        """Check if any errors occurred."""

from statement_extractor.pipeline import PluginRegistry

class PluginRegistry:
    @classmethod
    def list_plugins(cls, stage: Optional[int] = None) -> list[dict]:
        """List all registered plugins, optionally filtered by stage."""

    @classmethod
    def get_plugin(cls, name: str) -> Optional[BasePlugin]:
        """Get a plugin by name."""

class RawTriple(BaseModel):
    subject_text: str                    # Raw subject text
    predicate_text: str                  # Raw predicate text
    object_text: str                     # Raw object text
    source_sentence: str                 # Source sentence
    confidence: float = 1.0              # Extraction confidence (0-1)

    def as_tuple(self) -> tuple[str, str, str]:
        """Return as (subject, predicate, object) tuple."""

class PipelineStatement(BaseModel):
    subject: ExtractedEntity             # Subject with type, span, confidence
    predicate: str                       # Predicate text
    predicate_category: Optional[str]    # Predicate category (e.g., "employment_leadership")
    object: ExtractedEntity              # Object with type, span, confidence
    source_text: str                     # Source text
    confidence_score: float = 1.0        # Overall confidence (from GLiNER2)
    extraction_method: Optional[str]     # Method: gliner_relation

from statement_extractor.plugins.extractors.gliner2 import GLiNER2Extractor

class GLiNER2Extractor(BaseExtractorPlugin):
    def __init__(
        self,
        predicates: Optional[list[str]] = None,
        predicates_file: Optional[str | Path] = None,
        entity_types: Optional[list[str]] = None,
        use_default_predicates: bool = True,
    ):
        """
        Initialize the GLiNER2 extractor.

        Args:
            predicates: Custom list of predicate names
            predicates_file: Path to custom predicates JSON file
            entity_types: Entity types to extract (default: all)
            use_default_predicates: Use 324 built-in predicates when no custom provided
        """

class EntityQualifiers(BaseModel):
    # Semantic qualifiers
    org: Optional[str] = None            # Organization/employer
    role: Optional[str] = None           # Job title/position

    # Location qualifiers
    region: Optional[str] = None         # State/province
    country: Optional[str] = None        # Country
    city: Optional[str] = None           # City
    jurisdiction: Optional[str] = None   # Legal jurisdiction

    # External identifiers
    identifiers: dict[str, str] = {}     # lei, ch_number, sec_cik, ticker, etc.

    def has_any_qualifier(self) -> bool:
        """Check if any qualifier is set."""

class CanonicalMatch(BaseModel):
    canonical_id: Optional[str]          # ID in canonical database
    canonical_name: Optional[str]        # Canonical name/label
    match_method: str                    # identifier, name_exact, name_fuzzy, embedding
    match_confidence: float = 1.0        # Confidence in match (0-1)
    match_details: Optional[dict]        # Additional match details

class CanonicalEntity(BaseModel):
    entity_ref: str                      # Reference to original entity
    original_text: str                   # Original entity text
    entity_type: EntityType              # Entity type
    qualifiers: EntityQualifiers         # Qualifiers and identifiers
    canonical_match: Optional[CanonicalMatch]  # Canonical match if found
    fqn: str                             # Fully Qualified Name
    qualification_sources: list[str]     # Plugins that contributed

class StatementLabel(BaseModel):
    label_type: str                      # sentiment, relation_type, confidence
    label_value: Union[str, float, bool] # The label value
    confidence: float = 1.0              # Confidence in label
    labeler: Optional[str]               # Plugin that produced the label

class LabeledStatement(BaseModel):
    statement: PipelineStatement         # Original statement
    subject_canonical: CanonicalEntity   # Canonicalized subject
    object_canonical: CanonicalEntity    # Canonicalized object
    labels: list[StatementLabel] = []    # Applied labels

    @property
    def subject_fqn(self) -> str:
        """Subject's fully qualified name."""

    @property
    def object_fqn(self) -> str:
        """Object's fully qualified name."""

    def get_label(self, label_type: str) -> Optional[StatementLabel]:
        """Get label by type."""

    def as_dict(self) -> dict:
        """Convert to simplified dictionary."""

for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

    # Access labels
    sentiment = stmt.get_label("sentiment")
    if sentiment:
        print(f"  Sentiment: {sentiment.label_value}")

    # Access qualifiers
    subject_quals = stmt.subject_canonical.qualified_entity.qualifiers
    if subject_quals.role:
        print(f"  Role: {subject_quals.role}")

class TaxonomyResult(BaseModel):
    taxonomy_name: str                   # e.g., "esg_topics"
    category: str                        # Top-level category
    label: str                           # Specific label
    label_id: Optional[int] = None       # Numeric ID if available
    confidence: float = 1.0              # Classification confidence (0-1)
    classifier: Optional[str] = None     # Plugin that produced this result
    metadata: dict = {}                  # Additional metadata

    @property
    def full_label(self) -> str:
        """Return category:label format."""

# Access taxonomy results from context
# Each statement may have multiple labels above the threshold
for (source_text, taxonomy_name), results in ctx.taxonomy_results.items():
    print(f"Statement: {source_text[:50]}...")
    print(f"  Taxonomy: {taxonomy_name}")
    print(f"  Labels ({len(results)}):")
    for result in results:
        print(f"    - {result.full_label} (confidence: {result.confidence:.2f})")

class ClassificationSchema(BaseModel):
    label_type: str                      # e.g., "sentiment"
    choices: list[str]                   # Available choices
    description: str = ""                # Description for the classifier
    scope: str = "statement"             # statement or entity

class TaxonomySchema(BaseModel):
    label_type: str                      # e.g., "taxonomy"
    values: list[str] | dict[str, list[str]]  # Flat list or category -> labels
    description: str = ""
    scope: str = "statement"
    label_descriptions: Optional[dict[str, str]] = None  # Descriptions for labels

from statement_extractor import ExtractionOptions, extract_statements

options = ExtractionOptions(
    num_beams=6,
    diversity_penalty=1.2,
    deduplicate=True
)

result = extract_statements("Apple acquired Beats for $3 billion.", options)

from statement_extractor import ScoringConfig, ExtractionOptions, extract_statements

# High precision mode - only keep confident extractions
precision_config = ScoringConfig(
    min_confidence=0.7,
    quality_weight=1.5,
    redundancy_penalty=0.5
)

# High recall mode - keep everything
recall_config = ScoringConfig(
    min_confidence=0.0,
    quality_weight=0.5,
    redundancy_penalty=0.1
)

# Use in extraction
options = ExtractionOptions(scoring_config=precision_config)
result = extract_statements(text, options)

from statement_extractor import (
    PredicateComparisonConfig,
    PredicateTaxonomy,
    ExtractionOptions,
    extract_statements
)

# Stricter matching for precision
config = PredicateComparisonConfig(
    similarity_threshold=0.75,
    dedup_threshold=0.80,
    normalize_text=True
)

taxonomy = PredicateTaxonomy.from_list([
    "acquired", "founded", "works_for", "located_in",
    "partnered_with", "invested_in", "announced"
])

options = ExtractionOptions(
    predicate_taxonomy=taxonomy,
    predicate_config=config
)

result = extract_statements("Google bought YouTube in 2006.", options)

from statement_extractor.pipeline import PipelineConfig, ExtractionPipeline

# Run only splitting and extraction (stages 1-2)
config = PipelineConfig(enabled_stages={1, 2})

# Run stages 1-3 (skip canonicalization and labeling)
config = PipelineConfig(enabled_stages={1, 2, 3})

# From stage string
config = PipelineConfig.from_stage_string("1-3")  # {1, 2, 3}
config = PipelineConfig.from_stage_string("1,2,5")  # {1, 2, 5}

# Use presets
config = PipelineConfig.default()   # All 5 stages
config = PipelineConfig.minimal()   # Stages 1-2 only

# Disable specific plugins
config = PipelineConfig(
    disabled_plugins={"sec_edgar_qualifier", "companies_house_qualifier"}
)

# Enable only specific plugins
config = PipelineConfig(
    enabled_plugins={"t5_gemma_splitter", "gliner2_extractor", "person_qualifier"}
)

config = PipelineConfig(
    splitter_options={
        "num_beams": 6,
        "diversity_penalty": 1.2,
    },
    extractor_options={
        "predicates_file": "/path/to/custom_predicates.json",  # Custom predicate file
    },
    qualifier_options={
        "timeout": 10.0,  # API timeout
    },
)

{
  "category_name": {
    "predicate_name": {
      "description": "Description for semantic matching",
      "threshold": 0.7
    }
  }
}

{
  "employment": {
    "works_for": {"description": "Employment relationship", "threshold": 0.75},
    "manages": {"description": "Management relationship", "threshold": 0.7}
  },
  "ownership": {
    "owns": {"description": "Ownership relationship", "threshold": 0.7},
    "acquired": {"description": "Acquisition of entity", "threshold": 0.75}
  }
}

# Fast extraction for high-volume processing
fast_config = PipelineConfig.minimal()

# Full resolution for knowledge graph building
full_config = PipelineConfig.default()

# Custom: qualifiers without external APIs
internal_config = PipelineConfig(
    enabled_stages={1, 2, 3, 4, 5},
    disabled_plugins={"gleif_qualifier", "companies_house_qualifier", "sec_edgar_qualifier"},
)

from statement_extractor import extract_statements

result = extract_statements("Apple CEO Tim Cook announced the iPhone 15.")

for stmt in result:
    print(f"Subject: {stmt.subject.text} ({stmt.subject.type})")
    print(f"Object: {stmt.object.text} ({stmt.object.type})")

Subject: Apple (ORG)
Object: Tim Cook (PERSON)
Subject: Tim Cook (PERSON)
Object: iPhone 15 (PRODUCT)

from statement_extractor import extract_statements, EntityType

result = extract_statements("Microsoft acquired Activision for $69 billion.")

for stmt in result:
    if stmt.subject.type == EntityType.ORG:
        print(f"Organization found: {stmt.subject.text}")
    if stmt.object.type == EntityType.MONEY:
        print(f"Monetary value: {stmt.object.text}")

from statement_extractor import extract_statements, EntityType

text = """
Apple announced revenue of $94.8 billion for Q3 2024.
CEO Tim Cook presented at the company's Cupertino headquarters.
The new iPhone 16 features improved battery life of 22 hours.
"""

result = extract_statements(text)

# Filter for statements where subject is an organization
org_statements = [
    stmt for stmt in result
    if stmt.subject.type == EntityType.ORG
]

# Filter for statements involving monetary values
money_statements = [
    stmt for stmt in result
    if stmt.subject.type == EntityType.MONEY or stmt.object.type == EntityType.MONEY
]

# Filter for statements about people
person_statements = [
    stmt for stmt in result
    if stmt.subject.type == EntityType.PERSON or stmt.object.type == EntityType.PERSON
]

print(f"Found {len(org_statements)} statements from organizations")
print(f"Found {len(money_statements)} statements with monetary values")
print(f"Found {len(person_statements)} statements about people")

from statement_extractor import extract_statements, EntityType

result = extract_statements("The synergy initiative improved operational efficiency.")

for stmt in result:
    if stmt.subject.type == EntityType.UNKNOWN:
        print(f"Unclassified entity: {stmt.subject.text}")
        # Consider manual review or domain-specific handling

from statement_extractor import extract_statements

text = """
Microsoft announced a partnership with OpenAI in 2019.
The deal was valued at $1 billion and aimed to develop
artificial general intelligence.
"""

result = extract_statements(text)

# Iterate over statements
for stmt in result:
    subject = f"{stmt.subject.text} ({stmt.subject.type})"
    object_ = f"{stmt.object.text} ({stmt.object.type})"
    print(f"{subject} -- {stmt.predicate} --> {object_}")

# Check confidence scores
for stmt in result:
    score = stmt.confidence_score or 0.0
    print(f"[{score:.2f}] {stmt}")

Microsoft (ORG) -- partnered with --> OpenAI
Microsoft (ORG) -- announced --> partnership
OpenAI (ORG) -- partnership valued at --> $1 billion
Microsoft (ORG) -- aims to develop --> artificial general intelligence

from statement_extractor import StatementExtractor

# Initialize extractor with GPU
extractor = StatementExtractor(device="cuda")

texts = [
    "Apple acquired Beats Electronics for $3 billion.",
    "Google was founded by Larry Page and Sergey Brin in 1998.",
    "Amazon announced a new fulfillment center in Texas."
]

# Process multiple texts
for text in texts:
    result = extractor.extract(text)
    print(f"Found {len(result)} statements in: {text[:40]}...")
    for stmt in result:
        print(f"  - {stmt}")
    print()

# Force CPU usage
extractor = StatementExtractor(device="cpu")

from statement_extractor import extract_statements, ScoringConfig, ExtractionOptions

text = "Elon Musk founded SpaceX in 2002 to reduce space transportation costs."

# High precision mode - only high-confidence statements
scoring = ScoringConfig(min_confidence=0.7)
options = ExtractionOptions(scoring_config=scoring)
result = extract_statements(text, options)

print("High-confidence statements:")
for stmt in result:
    print(f"  [{stmt.confidence_score:.2f}] {stmt}")

# Extract all statements first
result = extract_statements(text)

# Apply custom thresholds
high_confidence = [s for s in result if (s.confidence_score or 0) >= 0.8]
medium_confidence = [s for s in result if 0.5 <= (s.confidence_score or 0) < 0.8]
low_confidence = [s for s in result if (s.confidence_score or 0) < 0.5]

print(f"High: {len(high_confidence)}, Medium: {len(medium_confidence)}, Low: {len(low_confidence)}")

from statement_extractor import PredicateTaxonomy, ExtractionOptions, extract_statements

# Define your canonical predicates
taxonomy = PredicateTaxonomy(predicates=[
    "acquired", "founded", "works_for", "announced",
    "invested_in", "partnered_with", "committed_to"
])

options = ExtractionOptions(predicate_taxonomy=taxonomy)

text = "Google bought YouTube in 2006. Sequoia Capital backed the video platform."
result = extract_statements(text, options)

# View predicate normalization
for stmt in result:
    original = stmt.predicate
    canonical = stmt.canonical_predicate
    if canonical and canonical != original:
        print(f"'{original}' -> '{canonical}'")
    print(f"  {stmt.subject.text} -- {canonical or original} --> {stmt.object.text}")

'bought' -> 'acquired'
  Google -- acquired --> YouTube
'backed' -> 'invested_in'
  Sequoia Capital -- invested_in --> YouTube

# predicates.txt contains one predicate per line
taxonomy = PredicateTaxonomy.from_file("predicates.txt")

from statement_extractor import (
    extract_statements,
    extract_statements_as_json,
    extract_statements_as_xml,
    extract_statements_as_dict
)

text = "Netflix acquired Spry Fox, a game development studio, in 2022."

# JSON output (default 2-space indent)
json_str = extract_statements_as_json(text)
print(json_str)

# Compact JSON
json_compact = extract_statements_as_json(text, indent=None)

# XML output (raw model format)
xml_str = extract_statements_as_xml(text)
print(xml_str)

# Dictionary output (for programmatic use)
data = extract_statements_as_dict(text)
for stmt in data["statements"]:
    print(f"{stmt['subject']['text']} -> {stmt['predicate']} -> {stmt['object']['text']}")

{
  "statements": [
    {
      "subject": {"text": "Netflix", "type": "ORG"},
      "predicate": "acquired",
      "object": {"text": "Spry Fox", "type": "ORG"},
      "source_text": "Netflix acquired Spry Fox",
      "confidence_score": 0.94
    }
  ],
  "source_text": "Netflix acquired Spry Fox, a game development studio, in 2022."
}

from statement_extractor import ExtractionOptions, extract_statements

# Disable embedding-based deduplication
options = ExtractionOptions(
    embedding_dedup=False,      # Use exact string matching for dedup
    predicate_taxonomy=None     # No predicate normalization
)

result = extract_statements(text, options)

from statement_extractor import ExtractionOptions, extract_statements

# Define a canonicalization function
def canonicalize_entity(text: str) -> str:
    """Normalize entity names to canonical forms."""
    mappings = {
        "apple": "Apple Inc.",
        "apple inc": "Apple Inc.",
        "apple inc.": "Apple Inc.",
        "google": "Alphabet Inc.",
        "google llc": "Alphabet Inc.",
        "alphabet": "Alphabet Inc.",
        "msft": "Microsoft Corporation",
        "microsoft": "Microsoft Corporation",
    }
    return mappings.get(text.lower().strip(), text)

options = ExtractionOptions(entity_canonicalizer=canonicalize_entity)

text = "Apple and Google announced a partnership. Microsoft joined later."
result = extract_statements(text, options)

for stmt in result:
    # Entities are now canonicalized
    print(f"{stmt.subject.text} -- {stmt.predicate} --> {stmt.object.text}")

Apple Inc. -- partnered with --> Alphabet Inc.
Microsoft Corporation -- joined --> partnership

from statement_extractor import (
    StatementExtractor,
    ExtractionOptions,
    ScoringConfig,
    PredicateTaxonomy,
    PredicateComparisonConfig
)

# Configure scoring for high precision
scoring = ScoringConfig(
    min_confidence=0.6,
    quality_weight=1.0,
    redundancy_penalty=0.5
)

# Define canonical predicates
taxonomy = PredicateTaxonomy.from_list([
    "acquired", "founded", "invested_in", "partnered_with",
    "announced", "launched", "hired", "appointed"
])

# Configure predicate matching
predicate_config = PredicateComparisonConfig(
    similarity_threshold=0.7,
    dedup_threshold=0.8
)

# Initialize extractor
extractor = StatementExtractor(
    device="cuda",
    predicate_taxonomy=taxonomy,
    predicate_config=predicate_config,
    scoring_config=scoring
)

# Configure extraction options
options = ExtractionOptions(
    num_beams=6,
    diversity_penalty=1.2,
    deduplicate=True,
    merge_beams=True
)

# Process text
text = """
Amazon Web Services announced a strategic partnership with Anthropic,
investing up to $4 billion in the AI safety startup. The deal, announced
in September 2023, makes AWS Anthropic's primary cloud provider.
"""

result = extractor.extract(text, options)

print(f"Extracted {len(result)} high-confidence statements:\n")
for stmt in result:
    canonical = stmt.canonical_predicate or stmt.predicate
    score = stmt.confidence_score or 0.0
    print(f"[{score:.2f}] {stmt.subject.text} ({stmt.subject.type})")
    print(f"       -- {canonical} -->")
    print(f"       {stmt.object.text} ({stmt.object.type})")
    print()

Extracted 4 high-confidence statements:

[0.92] Amazon Web Services (ORG)
       -- partnered_with -->
       Anthropic (ORG)

[0.88] Amazon Web Services (ORG)
       -- invested_in -->
       Anthropic (ORG)

[0.85] Amazon Web Services (ORG)
       -- invested_in -->
       $4 billion (MONEY)

[0.78] AWS (ORG)
       -- is primary cloud provider for -->
       Anthropic (ORG)

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

pipeline = ExtractionPipeline()

text = """
Amazon CEO Andy Jassy announced plans to hire 10,000 workers in the UK.
The expansion will focus on Amazon Web Services operations in London.
"""

ctx = pipeline.process(text)

print(f"Extracted {ctx.statement_count} statements\n")

for stmt in ctx.labeled_statements:
    # FQN includes role and organization
    print(f"Subject: {stmt.subject_fqn}")
    print(f"Predicate: {stmt.statement.predicate}")
    print(f"Object: {stmt.object_fqn}")

    # Access labels
    for label in stmt.labels:
        print(f"  {label.label_type}: {label.label_value}")

    # Access qualifiers
    subject_quals = stmt.subject_canonical.qualified_entity.qualifiers
    if subject_quals.role:
        print(f"  Role: {subject_quals.role}")
    if subject_quals.org:
        print(f"  Organization: {subject_quals.org}")

    print("-" * 40)

Extracted 2 statements

Subject: Andy Jassy (CEO, Amazon)
Predicate: announced
Object: plans to hire 10,000 workers in the UK
  sentiment: positive
  Role: CEO
  Organization: Amazon
----------------------------------------
Subject: Amazon (AMZN)
Predicate: expanding operations in
Object: London (UK)
  sentiment: positive
----------------------------------------

from statement_extractor.pipeline import PipelineConfig, ExtractionPipeline

# Run only stages 1 and 2 (splitting + extraction)
config = PipelineConfig(enabled_stages={1, 2})
pipeline = ExtractionPipeline(config)

ctx = pipeline.process("Tim Cook is CEO of Apple Inc.")

# Access Stage 2 output (PipelineStatement)
for stmt in ctx.statements:
    print(f"{stmt.subject.text} ({stmt.subject.type.value})")
    print(f"  --[{stmt.predicate}]-->")
    print(f"  {stmt.object.text} ({stmt.object.type.value})")
    print(f"  Confidence: {stmt.confidence_score:.2f}")

from statement_extractor.pipeline import PipelineConfig, ExtractionPipeline

# Disable external API plugins
config = PipelineConfig(
    disabled_plugins={
        "gleif_qualifier",
        "companies_house_qualifier",
        "sec_edgar_qualifier",
    }
)

pipeline = ExtractionPipeline(config)
ctx = pipeline.process("OpenAI CEO Sam Altman announced GPT-5.")

# Will use person_qualifier (local LLM) but skip external lookups
for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

from statement_extractor.pipeline import PipelineConfig, ExtractionPipeline

# Use custom predicates file
config = PipelineConfig(
    extractor_options={
        "predicates_file": "/path/to/my_predicates.json"
    }
)

pipeline = ExtractionPipeline(config)
ctx = pipeline.process("John works for Apple Inc.")

# All matching relations are returned
for stmt in ctx.statements:
    print(f"{stmt.subject.text} --[{stmt.predicate}]--> {stmt.object.text}")
    print(f"  Category: {stmt.predicate_category}")
    print(f"  Confidence: {stmt.confidence_score:.2f}")

{
  "employment": {
    "works_for": {
      "description": "Employment relationship where person works for organization",
      "threshold": 0.75
    },
    "manages": {
      "description": "Management relationship where person manages entity",
      "threshold": 0.7
    }
  },
  "ownership": {
    "owns": {
      "description": "Ownership relationship",
      "threshold": 0.7
    },
    "acquired": {
      "description": "Acquisition of one entity by another",
      "threshold": 0.75
    }
  }
}

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()
ctx = pipeline.process("Microsoft CEO Satya Nadella announced Azure growth.")

# Stage 1: Raw triples
print("=== Stage 1: Raw Triples ===")
for triple in ctx.raw_triples:
    print(f"  {triple.subject_text} -> {triple.predicate_text} -> {triple.object_text}")

# Stage 2: Statements with types
print("\n=== Stage 2: Statements ===")
for stmt in ctx.statements:
    print(f"  {stmt.subject.text} ({stmt.subject.type.value}) -> {stmt.predicate}")

# Stage 3: Qualified entities
print("\n=== Stage 3: Qualified Entities ===")
for ref, qualified in ctx.qualified_entities.items():
    quals = qualified.qualifiers
    print(f"  {qualified.original_text}")
    if quals.role:
        print(f"    Role: {quals.role}")
    if quals.org:
        print(f"    Org: {quals.org}")
    for id_type, id_value in quals.identifiers.items():
        print(f"    {id_type}: {id_value}")

# Stage 4: Canonical entities
print("\n=== Stage 4: Canonical Entities ===")
for ref, canonical in ctx.canonical_entities.items():
    print(f"  {canonical.fqn}")
    if canonical.canonical_match:
        print(f"    Method: {canonical.canonical_match.match_method}")
        print(f"    Confidence: {canonical.canonical_match.match_confidence:.2f}")

# Stage 5: Labeled statements
print("\n=== Stage 5: Labeled Statements ===")
for stmt in ctx.labeled_statements:
    print(f"  {stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")
    for label in stmt.labels:
        print(f"    {label.label_type}: {label.label_value}")

# Stage 6: Taxonomy results (multiple labels per statement)
print("\n=== Stage 6: Taxonomy Results ===")
for (source_text, taxonomy_name), results in ctx.taxonomy_results.items():
    print(f"  Statement: {source_text[:40]}...")
    for result in results:
        print(f"    {result.full_label} (confidence: {result.confidence:.2f})")

# Timings
print("\n=== Stage Timings ===")
for stage, duration in ctx.stage_timings.items():
    print(f"  {stage}: {duration:.3f}s")

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

# Use minimal stages for speed
config = PipelineConfig.minimal()  # Stages 1-2 only
pipeline = ExtractionPipeline(config)

documents = [
    "Apple announced a new MacBook Pro.",
    "Google acquired Fitbit for $2.1 billion.",
    "Tesla CEO Elon Musk unveiled the Cybertruck.",
]

all_statements = []

for doc in documents:
    ctx = pipeline.process(doc)
    for stmt in ctx.statements:
        all_statements.append({
            "subject": stmt.subject.text,
            "subject_type": stmt.subject.type.value,
            "predicate": stmt.predicate,
            "object": stmt.object.text,
            "object_type": stmt.object.type.value,
            "confidence": stmt.confidence_score,
            "source": doc,
        })

print(f"Extracted {len(all_statements)} statements from {len(documents)} documents")

from statement_extractor.pipeline import ExtractionPipeline

pipeline = ExtractionPipeline()

text = """
Apple announced a commitment to carbon neutrality by 2030.
The company also reported reducing packaging waste by 75%.
"""

ctx = pipeline.process(text)

# Access taxonomy classifications (multiple labels per statement)
for (source_text, taxonomy_name), results in ctx.taxonomy_results.items():
    print(f"Statement: {source_text[:50]}...")
    print(f"  Taxonomy: {taxonomy_name}")
    print(f"  Labels:")
    for result in results:
        print(f"    - {result.full_label} (confidence: {result.confidence:.2f})")
    print()

Statement: Apple announced a commitment to carbon neutrality...
  Taxonomy: esg_topics
  Labels:
    - environment:carbon_emissions (confidence: 0.87)
    - environment_benefit:emissions_reduction (confidence: 0.72)
    - governance:sustainability_commitments (confidence: 0.45)

Statement: The company also reported reducing packaging waste...
  Taxonomy: esg_topics
  Labels:
    - environment:waste_management (confidence: 0.92)
    - environment_benefit:waste_reduction (confidence: 0.85)

from statement_extractor.pipeline import ExtractionPipeline, PipelineConfig

config = PipelineConfig(fail_fast=False)  # Continue on errors
pipeline = ExtractionPipeline(config)

ctx = pipeline.process("Some text that might cause issues...")

# Check for errors
if ctx.has_errors:
    print("Errors occurred:")
    for error in ctx.processing_errors:
        print(f"  - {error}")

# Check for warnings
if ctx.processing_warnings:
    print("Warnings:")
    for warning in ctx.processing_warnings:
        print(f"  - {warning}")

# Process results that succeeded
print(f"\nSuccessfully extracted {ctx.statement_count} statements")

# Install the library
pip install corp-extractor[embeddings]

# For GPU support, install PyTorch with CUDA first
pip install torch --index-url https://download.pytorch.org/whl/cu121

from statement_extractor import StatementExtractor

# Auto-detect GPU or fall back to CPU
extractor = StatementExtractor()

# Or explicitly set device
extractor = StatementExtractor(device="cuda")  # GPU
extractor = StatementExtractor(device="cpu")   # CPU

cd runpod
docker build --platform linux/amd64 -t your-username/statement-extractor .

docker push your-username/statement-extractor

curl -X POST https://api.runpod.ai/v2/YOUR_ENDPOINT/runsync \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"input": {"text": "<page>Your text here</page>"}}'