Sourcery Usage Guide for Python LLM Extraction#

What Sourcery Is#

Sourcery is both:

  1. A Python library you import (import sourcery) to run schema-first LLM extraction.
  2. A reference project with ingestion adapters, HTML reviewer UI, and runnable integration scripts.

Use it as a library inside your app, and use this repository as a production template for typed, source-grounded document extraction.

When To Use Sourcery#

Use Sourcery when you need:

  • typed extraction contracts (Pydantic models),
  • grounded spans (char_start, char_end) for every extraction,
  • deterministic chunking/alignment/merge behavior,
  • optional document-level reconciliation into canonical claims,
  • human review/export workflows.

Install#

Python requirement: >=3.12

Minimal runtime:

uv sync

With ingestion adapters (PDF/URL/HTML):

uv sync --extra ingest

With dev tooling:

uv sync --extra dev --extra ingest

Set provider credentials for the model route you use in RuntimeConfig.model (for example DEEPSEEK_API_KEY, OPENAI_API_KEY, ANTHROPIC_API_KEY, etc.).

Core Public API#

Import-level API (sourcery/__init__.py):

  1. extract(request: ExtractRequest, engine: SourceryEngine | None = None) -> ExtractResult
  2. aextract(request: ExtractRequest, engine: SourceryEngine | None = None) -> ExtractResult
  3. extract_from_sources(sources, *, task, runtime, options=None, engine=None) -> ExtractResult
  4. aextract_from_sources(...) -> ExtractResult
  5. SourceryEngine with .extract(...), .aextract(...), .extract_stream(...), .replay_run(...)

Streaming event contracts live in sourcery.contracts:

  • StreamExtractionAdded
  • StreamChunkDone
  • StreamPassDone

Data Contracts You Define#

1) EntitySpec#

  • name: str
  • attributes_model: type[BaseModel]

2) EntitySchemaSet#

  • entities: list[EntitySpec]

3) ExtractionTask#

  • instructions: str
  • schema: EntitySchemaSet
  • examples: list[ExtractionExample]
  • strict_example_alignment: bool = True

4) ExtractRequest#

  • documents: list[SourceDocument] | str
  • task: ExtractionTask
  • options: ExtractOptions = ExtractOptions()
  • runtime: RuntimeConfig

5) ExtractResult#

  • documents: list[DocumentResult]
  • run_trace: ExtractionRunTrace
  • metrics: RunMetrics
  • warnings: list[str]

DocumentResult includes:

  • extractions: list[AlignedExtraction]
  • canonical_claims: list[CanonicalClaim]

Runtime Config (RuntimeConfig)#

Required:

  • model: str

Core options:

  • temperature: float = 0.0
  • max_tokens: int | None = None
  • stream: bool = False
  • storage_dir: str = ".sourcery"
  • respect_context_window: bool = True

Reliability:

  • retry: RetryPolicy
  • max_attempts=3
  • initial_backoff_seconds=0.75
  • max_backoff_seconds=8.0
  • backoff_multiplier=2.0
  • retry_on_rate_limit=True
  • retry_on_transient_errors=True
  • auto_resume_paused_runs=True
  • max_pause_resumes=5

Session refinement (optional):

  • session_refinement: SessionRefinementConfig
  • enabled=False
  • max_turns=1
  • context_chars=320

Document-level reconciliation (optional):

  • reconciliation: ReconciliationConfig
  • enabled=False
  • use_workforce=True
  • min_mentions_for_claim=1
  • max_claims=200

RuntimeConfig.stream is passed to the underlying BlackGeorge runtime/provider. For Sourcery-level chunk progress, use SourceryEngine.extract_stream(...).

Extraction Options (ExtractOptions)#

  • max_chunk_chars=1200
  • context_window_chars=200
  • max_passes=2
  • batch_concurrency=16
  • enable_fuzzy_alignment=True
  • fuzzy_alignment_threshold=0.82
  • accept_partial_exact=False
  • stop_when_no_new_extractions=True
  • allow_unresolved=False

With accept_partial_exact=True, partial alignment requires a contiguous match of at least two word tokens covering more than half of the candidate's word tokens. Unresolved results with the same entity and text are deduplicated across passes even when their attributes differ. The normal merge priority retains the earlier pass, then the higher-confidence result within the same pass.

Minimal Example (Inline Text)#

from pydantic import BaseModel
import sourcery
from sourcery.contracts import (
    EntitySchemaSet,
    EntitySpec,
    ExtractRequest,
    ExtractionExample,
    ExtractionTask,
    ExampleExtraction,
    RuntimeConfig,
)

class PersonAttrs(BaseModel):
    role: str | None = None

request = ExtractRequest(
    documents="Alice is the CEO of Acme.",
    task=ExtractionTask(
        instructions="Extract person entities.",
        schema=EntitySchemaSet(
            entities=[EntitySpec(name="person", attributes_model=PersonAttrs)]
        ),
        examples=[
            ExtractionExample(
                text="Bob is the CTO.",
                extractions=[
                    ExampleExtraction(entity="person", text="Bob", attributes={"role": "CTO"})
                ],
            )
        ],
    ),
    runtime=RuntimeConfig(model="deepseek/deepseek-v4-pro"),
)

result = sourcery.extract(request)
print(result.metrics.model_dump(mode="json"))
for ext in result.documents[0].extractions:
    print(ext.entity, ext.text, ext.char_start, ext.char_end, ext.alignment_status)

Notebook equivalent: examples/notebooks/sourcery_quickstart.ipynb

Extract From Files / PDFs / URLs / Images#

Use the source-based helper:

result = sourcery.extract_from_sources(
    ["1706.03762v7.pdf", "https://example.com/article.html"],
    task=task,
    runtime=RuntimeConfig(model="deepseek/deepseek-v4-pro"),
)

Supported ingestion via load_source_document(s):

  1. Inline text
  2. Text files
  3. PDF files (pypdf)
  4. HTML files / raw HTML
  5. URLs
  6. Image files via VLM OCR (sourcery.ingest.load_vlm_ocr_document)

Notes:

  • PDF loader is text-extraction first (pypdf).
  • Image ingestion uses any vision-language model through blackgeorge multimodal support.

Notebook equivalent: examples/notebooks/sourcery_pdf_workflow.ipynb

Async Usage#

result = await sourcery.aextract(request)

Streaming Extraction#

extract_stream(...) processes up to batch_concurrency chunks concurrently. It emits each finished batch's extraction and chunk events in deterministic chunk order before starting the next batch. It is progress streaming, not completion-order or token streaming.

from sourcery.contracts import StreamChunkDone, StreamExtractionAdded, StreamPassDone
from sourcery.runtime import SourceryEngine

engine = SourceryEngine()

for event in engine.extract_stream(request):
    if isinstance(event, StreamExtractionAdded):
        print(event.document_id, event.extraction.entity, event.extraction.text)
    elif isinstance(event, StreamChunkDone):
        print("chunk done:", event.chunk_id, event.candidates_found)
    elif isinstance(event, StreamPassDone):
        print("pass done:", event.pass_id, event.additions_this_pass)

Advanced Engine Usage#

from sourcery.runtime import SourceryEngine

engine = SourceryEngine()
result = engine.extract(request)

raw_run_id = result.documents[0].extractions[0].provenance.raw_run_id
if raw_run_id:
    replay, events = engine.replay_run(request, raw_run_id)
    print(replay["status"] if replay else None, len(events))

Enabling Reconciliation + Session Refinement#

runtime = RuntimeConfig(
    model="deepseek/deepseek-v4-pro",
    session_refinement={"enabled": True, "max_turns": 1, "context_chars": 320},
    reconciliation={"enabled": True, "use_workforce": True, "max_claims": 100},
)

What this does:

  1. Session refinement adds multi-turn continuity hints per chunk.
  2. Reconciliation runs document-level resolver workflow and returns canonical_claims.

Outputs and Review#

Save JSONL#

from sourcery.io import save_extract_result_jsonl
save_extract_result_jsonl(result, "output/result.jsonl")

Generate HTML viewer#

from sourcery.io import write_document_html
write_document_html(result.documents[0], "output/document.viewer.html")

Generate reviewer UI#

from sourcery.io import write_reviewer_html
write_reviewer_html(result.documents[0], "output/document.reviewer.html")

Reviewer supports:

  • search,
  • entity/status filters,
  • approve/reject/reset,
  • export approved JSONL/CSV.

Scripted End-to-End Runs#

Benchmark comparison wrapper#

uv run benchmark_compare.py --text-types english

Error Model#

Important exception classes (sourcery/exceptions.py):

  • SourceryError
  • SourceryRuntimeError
  • SourceryProviderError
  • SourceryRateLimitError
  • SourceryRetryExhaustedError
  • SourceryPausedRunError
  • SourceryPipelineError
  • SourceryIngestionError
  • SourceryDependencyError

Validation Commands#

uv run --extra dev pytest -q
uv run --extra dev ruff check sourcery tests
uv run --extra dev mypy sourcery

Production Notes#

  1. Treat schemas as API contracts and version them.
  2. Start with strict examples and deterministic options.
  3. Enable reconciliation for long documents where alias/coreference matters.
  4. Keep reviewer approval in-the-loop for high-stakes workflows.
  5. Persist JSONL + run trace for audit and replay.