DOCSEARCHbyAlgolia
Submit Docs

LandingAI Agentic Document Extraction (ADE) Python Library

repository·main·Indexed 21 days ago

https://github.com/landing-ai/ade-python

The official Python library for the landingai-ade API, providing an interface to parse complex documents (PDFs, images) into structured Markdown and extract specific data fields using Pydantic models or JSON schemas. It supports synchronous and asynchronous operations, including a v2 client for the DPT-3 model family and a jobs API for processing large documents.

Tokens
17.3K
Snippets
51
Records
65
Agent score
76%

What's inside landingai-ade

  1. How V2 AI-Wiring handles spec changes

    main

    When the V2 spec drifts, the AI-wiring process attempts to automate the SDK update following these rules:

    • Field Changes: For existing operations, it adds new optional keyword parameters to existing methods (leveraging surface-lock for backward compatibility).
    • New Routes: It mirrors the pattern found in resources/v2/extract.py and parse.py. This includes:
      • Creating a resource module with a run() method.
      • Creating a *JobsResource for async routes (create/get/wait/list).
      • Adding a unified Job type via _normalize.py.
      • Defining a new response type in types/v2/.
      • Registering the resource in the resources/v2/v2.py container.
    • Exclusions: The /v2/workflow* routes are explicitly excluded from the AI-wiring logic to prevent accidental or unverified changes to complex workflow logic.

  2. How V2 Async Job envelopes are normalized

    main

    The ADE V2 client uses normalization functions (normalize_parse_job, normalize_extract_job, normalize_build_schema_job) to fold upstream API envelopes into a unified Job object. This ensures consistency despite upstream field drift.

    Normalization Logic:

    • Result Location: The response is found under the result key (previously data in older versions).
    • Error Handling: Failures are mapped to a structured error object {code, message}. Older flat failure_reason strings are also mapped to Job.error.
    • Timestamps: created_at and completed_at support both ISO-8601 strings and epoch seconds.
    • Status Resilience: Unknown or renamed status values default to pending rather than raising an error. The original raw envelope is always preserved in Job.raw.

  3. Understand the V2 Spec-Sync Architecture

    main

    The spec-sync pipeline is designed to track changes (drift) between the local SDK implementation and the live API specification. The V2 extension introduces a parallel loop to the existing V1 pipeline to monitor the V2 spec without affecting V1 stability.

    Key Components

    • Drift Detection: Uses check-drift.sh and fetch-normalize.sh to compare the current local snapshot against the live spec URL.
    • Mechanical Snapshot: Stores the normalized spec in specs/v2-aide.json and generated models in specs/_generated/v2_models.py.
    • AI-Wiring: A dormant step that triggers only when drift is detected. It uses a claude-code-action to draft code changes (new routes, field updates) to match the new spec. This step is designed to produce a reviewed draft that a human must finalize.
    • Contract Testing: Any drift detection PR is automatically validated against the live staging environment using contract-tests to ensure the proposed changes are valid.

  4. Syncing SDKs with Staging vs. Production environments

    main

    The SDK synchronization process decouples API deployment from SDK artifact release by targeting different environments at different stages:

    1. Staging Sync: The automation targets the staging spec (https://api.va.staging.landing.ai/v1/ade/openapi.json or https://aide.staging.landing.ai/openapi.json). When the staging API changes, a sync PR is opened. Developers can test the merged main branch against staging using: pip install git+... @main and setting LANDINGAI_ADE_ENVIRONMENT=staging.
    2. Production Release: The release gate is based on the production spec. An SDK release is only permitted if every path implemented in the SDK exists in the production spec. Once verified, a maintainer dispatches the Release workflow to publish the artifact to PyPI/npm.

    Key Rule: The surface-lock baseline is always the last release tag. This allows features to be reshaped or dropped in main (staging) before they ever reach production without breaking the compatibility guarantee provided to users.

  5. How the ADE SDK spec-sync automation works

    main

    The ADE Python SDK uses an automated pipeline to ensure the library stays in sync with the live API specifications (OpenAPI). The process is triggered by a cron job (~6h) or manual dispatch and follows a two-phase commit pattern to handle API drift:

    1. Mechanical Phase: The pipeline fetches the live OpenAPI specs, normalizes them, and compares them against committed snapshots in specs/*.json. If drift is detected, it performs a mechanical commit containing the new spec snapshots and regenerated types (using tools like datamodel-code-generator).
    2. AI Phase: An AI agent (anthropics/claude-code-action) runs in automation mode to update resource classes, method signatures, tests, and documentation (e.g., api.md, README.md) based on the spec changes. This results in a second, separately attributed commit.

    Safety Gates:

    • Surface-lock: A CI job (using griffe for Python) compares the new surface against the last release tag. Any change to the existing released public surface causes the build to fail, ensuring backward compatibility.
    • Contract Tests: Automated tests run against the staging environment to verify the new SDK code against the live staging API.
    • Human Review: All automated PRs require manual review and approval before merging to main.
    fetch v1 + v2 openapi.json ─ normalize (jq -S) ─ diff vs committed specs/*.json
  └─ on drift, ONE job, two phases, one PR branch:
       commit 1 (mechanical): spec snapshot + regenerated types
       commit 2 (AI):         Claude Code GitHub Action wires resources, methods, tests, docs from the spec diff

  6. Use the LandingAI ADE V2 Client

    main

    The client.v2 sub-client provides access to LandingAI's next-generation ADE gateway. It is an additive surface, meaning client.v2.* methods are separate from the standard client.* (V1) methods and do not affect V1 behavior.

    Key differences in V2:

    • Host: Uses api.ade.[env].landing.ai instead of the V1 host.
    • Unified Job Shape: Both client.v2.parse_jobs and client.v2.extract_jobs return a unified Job object. If a field is not explicitly in the typed model, it is available in Job.raw as a dictionary.

  7. Process large documents asynchronously with Jobs

    main

    For large documents that may timeout in a synchronous request, use the parse_jobs or extract_jobs interfaces. These follow a create -> get/wait/list pattern.

    Workflow:

    1. Create: Call .create() to start the job. This returns a Job object containing a job_id and status.
    2. Wait: Use the .wait() helper to block and poll the server with exponential backoff until the job reaches a terminal state.
    3. List/Get: Use .list() to find jobs by status or .get(job_id) to check a specific job's progress.

    Job Statuses: pending, processing, completed, failed, or cancelled.

    Errors:

    • JobWaitTimeoutError: Raised if the .wait() timeout is reached before the job finishes.
    • JobFailedError: Raised if the job fails on the server side.
    from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib.v2_errors import JobFailedError, JobWaitTimeoutError


client = LandingAIADE()

job = client.v2.parse_jobs.create(
    document=Path("path/to/large_file.pdf"),
    service_tier="standard",  # "standard" (default, lower cost) or "priority" (faster)
)
print(job.job_id, job.status)

# Block until the job finishes (polls with backoff)
try:
    done = client.v2.parse_jobs.wait(job.job_id, timeout=600, raise_on_failure=True)
    if done.result is not None:
        print(done.result.markdown[:200])
except JobWaitTimeoutError:
    print("Job did not finish in time; it is still running server-side.")
except JobFailedError as e:
    print(f"Job failed: {e}")

  8. Quickstart: Parse and Extract structured data

    main

    The standard workflow involves two steps: converting a document to Markdown using parse, and then pulling typed fields from that Markdown using extract with a Pydantic model.

    Note: Use client.v2 for all new projects to leverage the DPT-3 model family.

    from pathlib import Path
from pydantic import BaseModel, Field
from landingai_ade import LandingAIADE


class Invoice(BaseModel):
    invoice_number: str = Field(description="The invoice number")
    total: str = Field(description="Invoice grand total")


client = LandingAIADE()  # reads VISION_AGENT_API_KEY

# 1. Parse: convert the document to structured Markdown
parsed = client.v2.parse(document=Path("invoice.pdf"))
print(parsed.markdown)

# 2. Extract: pull typed fields out of the Markdown
result = client.v2.extract(schema=Invoice, markdown=parsed.markdown)
print(result.extraction)

  10. Configure the LandingAI ADE API Key

    main

    The client automatically reads your API key from the VISION_AGENT_API_KEY environment variable. Alternatively, you can pass the key directly to the LandingAIADE constructor.

    To use an environment variable:

    export VISION_AGENT_API_KEY=<your-api-key>

    To pass it in code:

    client = LandingAIADE(apikey="<your-api-key>")

  11. Configure the HTTP client and manage resources

    main

    You can override the underlying httpx client to configure proxies or custom transports using DefaultHttpxClient.

    Resource Management: The client closes connections when garbage collected, but it is recommended to use a context manager or call .close() explicitly to manage HTTP resources.

    import httpx
from landingai_ade import LandingAIADE, DefaultHttpxClient

# Custom HTTP client with proxy
client = LandingAIADE(
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

# Using context manager for resource management
with LandingAIADE() as client:
    client.v2.parse(...)

  12. Process large documents asynchronously with client.v2.parse_jobs

    main

    For long-running parsing tasks, use the asynchronous job API to avoid timeouts.

    Workflow:

    1. Create: Use client.v2.parse_jobs.create(...) to start a job. This returns a Job object containing a job_id.
    2. Check Status: Use client.v2.parse_jobs.get(job_id) to retrieve the current status.
    3. Wait/Poll: Use client.v2.parse_jobs.wait(job_id, ...) to block execution until the job reaches a terminal state (completed, failed, or cancelled).
    4. List Jobs: Use client.v2.parse_jobs.list(...) to view recent jobs. Returns a JobList which includes pagination metadata like .has_more.

    Wait Parameters:

    • timeout: Maximum seconds to wait (default 600).
    • poll_interval: Interval between polls.
    • raise_on_failure: If True, raises JobFailedError if the job ends in an error state.
    job = client.v2.parse_jobs.create(document=my_file, model='...')
# Block until finished
completed_job = client.v2.parse_jobs.wait(job.job_id, timeout=600)

if completed_job.status == 'completed':
    result = completed_job.result