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, PDFs, and URLs (v0.7.0)
corp-extractor document process article.txt
corp-extractor document process report.pdf
corp-extractor document process report.pdf --pdf-parser glm_ocr_parser
corp-extractor document process https://example.com/article

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

# Use the default server URL (http://localhost:8111)
corp-extractor --server pipeline "Apple announced a new iPhone."

# Use a custom server URL
corp-extractor --server-url http://gpu-box:8111 pipeline "Apple announced a new iPhone."

# Or set via environment variable
export CORP_EXTRACTOR_SERVER=http://localhost:8111
corp-extractor pipeline "Apple announced a new iPhone."

# 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]

# Start the server (default: 0.0.0.0:8111)
corp-extractor serve

# Custom host and port
corp-extractor serve --host 127.0.0.1 --port 9000

# Skip model warmup (models load on first request)
corp-extractor serve --no-warmup

# Verbose logging
corp-extractor serve -v

# In another terminal
corp-extractor --server split "Apple announced a new iPhone."
corp-extractor --server pipeline "Amazon CEO Andy Jassy announced plans."
corp-extractor --server pipeline -f article.txt -o json
corp-extractor --server document process article.txt

from statement_extractor import extract_statements
result = extract_statements("text", server_url="http://localhost:8111")

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 local PDFs (auto-detected by .pdf extension)
corp-extractor document process report.pdf
corp-extractor document process report.pdf --pdf-parser glm_ocr_parser

# 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
corp-extractor document process https://example.com/report.pdf --pdf-parser glm_ocr_parser

# 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

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)

# Process a PDF with GLM-OCR parser
config = URLLoaderConfig(pdf_parser_plugin="glm_ocr_parser")
ctx = await pipeline.process_url("https://example.com/report.pdf", config)

# CLI: local PDF with default parser
corp-extractor document process report.pdf

# CLI: local PDF with GLM-OCR VLM parser
corp-extractor document process report.pdf --pdf-parser glm_ocr_parser

def extract_statements(
    text: str,
    options: Optional[ExtractionOptions] = None,
    server_url: Optional[str] = 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)
        server_url: URL of a running corp-extractor server to delegate to
                    (e.g. "http://localhost:8111"). When provided, no local
                    models are loaded.
        **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, server_url: Optional[str] = None):
        """
        Initialize the extraction pipeline.

        Args:
            config: Pipeline configuration (default: all stages enabled)
            server_url: URL of a running corp-extractor server to delegate to.
                        When provided, process() sends HTTP requests instead
                        of loading models locally.
        """

    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

curl http://localhost:8111/

{
  "status": "ok",
  "device": "cuda:0",
  "cuda_available": true,
  "mps_available": false,
  "models_loaded": {"extractor": true, "pipeline": true},
  "plugins": {
    "splitters": ["t5_gemma_splitter"],
    "extractors": ["gliner2_extractor"],
    "qualifiers": ["person_qualifier", "embedding_company_qualifier"],
    "labelers": ["sentiment_labeler", "confidence_labeler", "relation_type_labeler"],
    "taxonomy": ["embedding_taxonomy_classifier"]
  }
}

class PipelineRequest(BaseModel):
    text: str                              # Input text
    config: dict[str, Any] = {}            # Pipeline configuration

curl -X POST http://localhost:8111/pipeline \
  -H "Content-Type: application/json" \
  -d '{"text": "Apple CEO Tim Cook announced a new iPhone.", "config": {"enabled_stages": "1-3"}}'

class SplitRequest(BaseModel):
    text: str                              # Input text
    options: dict[str, Any] = {}           # ExtractionOptions fields

curl -X POST http://localhost:8111/split \
  -H "Content-Type: application/json" \
  -d '{"text": "Apple announced a new iPhone.", "options": {"num_beams": 6}}'

class DocumentRequest(BaseModel):
    text: str                              # Document text
    title: Optional[str] = None            # Document title
    stages: str = "1-6"                    # Pipeline stages
    max_tokens: int = 1000                 # Target tokens per chunk
    overlap: int = 100                     # Token overlap between chunks
    no_summary: bool = False               # Skip summarization
    no_dedup: bool = False                 # Skip cross-chunk deduplication

curl -X POST http://localhost:8111/document \
  -H "Content-Type: application/json" \
  -d '{"text": "Long document text...", "title": "Annual Report", "stages": "1-3"}'

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)

# Terminal 1: Start the server
corp-extractor serve

# Terminal 2: Use --server to delegate processing
corp-extractor --server pipeline "Amazon CEO Andy Jassy announced plans."
corp-extractor --server split -f article.txt --json
corp-extractor --server document process article.txt

# Or set the environment variable
export CORP_EXTRACTOR_SERVER=http://localhost:8111
corp-extractor pipeline "text"  # Automatically uses the server

from statement_extractor import extract_statements
from statement_extractor.pipeline import ExtractionPipeline

# Simple extraction via server
result = extract_statements("Apple announced iPhone.", server_url="http://localhost:8111")
for stmt in result:
    print(f"{stmt.subject.text} -> {stmt.predicate} -> {stmt.object.text}")

# Full pipeline via server
pipeline = ExtractionPipeline(server_url="http://localhost:8111")
ctx = pipeline.process("Apple CEO Tim Cook announced a new iPhone.")
for stmt in ctx.labeled_statements:
    print(f"{stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

import httpx

# Call the pipeline endpoint
resp = httpx.post("http://localhost:8111/pipeline", json={
    "text": "Apple CEO Tim Cook announced a new iPhone.",
    "config": {"enabled_stages": "1-3"},
}, timeout=300)

# Reconstruct full Pydantic model from response
from statement_extractor.pipeline.context import PipelineContext
ctx = PipelineContext.model_validate(resp.json())
for stmt in ctx.labeled_statements:
    print(f"  {stmt.subject_fqn} -> {stmt.statement.predicate} -> {stmt.object_fqn}")

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

# Start the persistent server
corp-extractor serve

# In another terminal, use --server to delegate to it
corp-extractor --server pipeline "Amazon CEO Andy Jassy announced..."
corp-extractor --server split -f article.txt --json
corp-extractor --server document process article.txt

export CORP_EXTRACTOR_SERVER=http://localhost:8111
corp-extractor pipeline "Your text"  # Automatically uses the server

from statement_extractor import extract_statements
from statement_extractor.pipeline import ExtractionPipeline
from statement_extractor.document import DocumentPipeline

# Delegate extraction to server
result = extract_statements("text", server_url="http://localhost:8111")

# Pipeline and document pipeline
pipeline = ExtractionPipeline(server_url="http://localhost:8111")
ctx = pipeline.process("Amazon CEO Andy Jassy announced...")

doc_pipeline = DocumentPipeline(server_url="http://localhost:8111")