ContractExtractor¶
Loads a document (or takes text), chunks it and makes schema-constrained model calls, returning a Contract. See Contract analysis.
ContractExtractor ¶
Main extraction orchestrator that coordinates document loading, chunking, LLM-based extraction, and result assembly.
__init__(llm_provider: LLMProvider | None = None, llm_provider_name: str | None = None, document_loader: DocumentLoader | None = None, chunking_strategy: ChunkingStrategy | None = None, confidence_threshold: float = 0.7, parallel_processing: bool = True) ¶
Initialize the contract extractor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
llm_provider | LLMProvider | None | Custom LLM provider instance | None |
llm_provider_name | str | None | Full model name: | None |
document_loader | DocumentLoader | None | Custom document loader | None |
chunking_strategy | ChunkingStrategy | None | Custom chunking strategy | None |
confidence_threshold | float | Minimum confidence score for extractions | 0.7 |
parallel_processing | bool | Whether to process chunks in parallel | True |
estimate_extraction_cost(document_path: str) -> dict[str, Any] ¶
Estimate the API cost of extracting a document without running extraction.
Loads and chunks the document, then uses the configured LLM provider's cost model to estimate the total token usage and USD cost across all extraction phases.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document_path | str | Path to the contract document. | required |
Returns:
| Type | Description |
|---|---|
dict[str, Any] | Dict with keys |
dict[str, Any] | (int, input tokens), |
dict[str, Any] | provider class name), |
dict[str, Any] | per-phase cost and token estimates). |
Example::
extractor = ContractExtractor(llm_provider=LocalProvider(model="llama3.1:8b"))
estimate = extractor.estimate_extraction_cost("contract.pdf")
print(f"Estimated cost: ${estimate['estimated_cost']:.4f}")
estimate_extraction_cost_from_text(text: str) -> dict[str, Any] ¶
Like estimate_extraction_cost but for already-loaded text.
extract(document_path: str, contract_type: ContractType | None = None, analyze_risks: bool = True, extract_financial: bool = True, known_parties: list[str] | None = None) -> Contract ¶
Extract all data from a contract document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document_path | str | Path to the contract document | required |
contract_type | ContractType | None | Optional contract type hint | None |
analyze_risks | bool | Whether to perform risk analysis | True |
extract_financial | bool | Whether to extract financial terms | True |
known_parties | list[str] | None | Optional list of known party names (hints) | None |
Returns:
| Type | Description |
|---|---|
Contract | Contract object with all extracted data |
Raises:
| Type | Description |
|---|---|
ExtractionError | If extraction fails |
extract_from_text(text: str, contract_type: ContractType | None = None, analyze_risks: bool = True, extract_financial: bool = True, known_parties: list[str] | None = None) -> Contract ¶
Like extract but for already-loaded text (no file metadata).
extract_async(document_path: str, **kwargs: Any) -> Contract async ¶
Async version of extract — runs the synchronous extractor in a thread pool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document_path | str | Path to the contract document | required |
**kwargs | Any | Additional arguments passed to extract() | {} |
extract_batch(document_paths: list[str], max_workers: int = 4, **kwargs: Any) -> list[Contract] ¶
Process multiple contracts in parallel.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document_paths | list[str] | List of document paths to process | required |
max_workers | int | Maximum number of parallel workers | 4 |
**kwargs | Any | Additional arguments passed to extract() | {} |
Returns:
| Type | Description |
|---|---|
list[Contract] | List of successfully extracted Contract objects (failed paths are skipped). |