> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.eyelevel.ai/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.eyelevel.ai/_mcp/server.

Use an extraction workflow when a document should return structured JSON. This
guide creates a workflow from YAML, assigns it to a bucket, ingests a document,
and reads the result with the GroundX Python SDK.

GroundX validates the YAML and creates the workflow settings. Send the YAML
directly through the SDK instead of preparing workflow settings in your
application.

## 1. Start With The JSON You Want

This guide creates the following output:

```json
{
  "statement": {
    "account_number": "123456789",
    "due_date": "2026-06-30",
    "total_amount_due": 128.55
  },
  "service": {
    "service_address": "100 Main St, Denver, CO 80202"
  }
}
```

Group names and `workflow_output_key` values become names in the returned JSON.

## 2. Write The Workflow YAML

An `extraction_policy_version: v1` workflow uses these settings:

* `workflow.custom_steps` names the extraction passes. `level` selects the
  document unit each pass processes: `chunk`, `section`, or `document`.
  `kind: instruct` is for non-repeating fields, `keys` for repeating rows such
  as charges, and `summary` for repeating records such as meters. A
  document-level step cannot use `instruct`.
* `workflow.agent_chain` assigns each group a built-in processing sequence. Use
  the sequence whose task names match the group's role. The `reconcile`, `qa`,
  and `save` tasks combine candidate values, check them, and write the result.
* `role` selects processing behavior, not the group's output name or subject.
  Use `statement` for non-repeating values, `meters` for repeating parent
  records, and `charges` for related child records.
* `workflow_step` connects a group to a named custom step.
* `workflow_output_key` names a field in the returned JSON.

Declare a `role` and `workflow_step` for every routed group, and a
`workflow_output_key` for every routed field. GroundX does not infer a role from
a group name or data shape.

Each field prompt has four required settings:

| Setting        | Purpose                                                                                                                              |
| -------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
| `description`  | Defines the field's meaning and scope.                                                                                               |
| `identifiers`  | Lists one to three visible labels or stable source cues. These help locate evidence but do not prove that a nearby value is correct. |
| `instructions` | States how to choose, reject, format, or return a value.                                                                             |
| `type`         | Declares the returned JSON value type.                                                                                               |

Each chunk-level custom step runs once per chunk. For long documents, estimate
the request count before ingest. Use `workflow.section_strategy: page` with
`level: section` for broad statement passes when the estimate approaches 2,000
requests.

**`statement.yaml`**

```yaml title="statement.yaml"
extraction_policy_version: v1

workflow:
  custom_steps:
    - name: statement_fields
      level: chunk
      kind: instruct
    - name: service_fields
      level: chunk
      kind: instruct
  agent_chain:
    - parallel:
        - group: statement
          chain: [reconcile_statement, qa_statement, save_statement]
        - group: service
          chain: [reconcile_statement, qa_statement, save_statement]

statement:
  role: statement
  workflow_step: statement_fields
  fields:
    account_number:
      workflow_output_key: account_number
      prompt:
        description: The utility account number printed on the statement.
        identifiers: [Account Number, "Account #"]
        instructions: Return the account number exactly as printed.
        type: str
    due_date:
      workflow_output_key: due_date
      prompt:
        description: The payment due date.
        identifiers: [Due Date, Payment Due]
        instructions: Return the date as YYYY-MM-DD.
        type: str
    total_amount_due:
      workflow_output_key: total_amount_due
      prompt:
        description: The final amount due for the statement.
        identifiers: [Total Amount Due, Amount Due]
        instructions: Return only the numeric amount.
        type: float

service:
  role: statement
  workflow_step: service_fields
  fields:
    service_address:
      workflow_output_key: service_address
      prompt:
        description: The billed service address.
        identifiers: [Service Address, Service Location]
        instructions: Return the address exactly as printed.
        type: str
```

## 3. Create And Assign The Workflow

Install the extraction helpers:

```sh
pip install "groundx[extract]"
```

Create the workflow from the YAML file, then assign it to a bucket:

**`Python`**

```python title="Python"
import os

from groundx import GroundX

client = GroundX(api_key=os.environ["GROUNDX_API_KEY"])
bucket_id = 1234

response = client.create_extraction_workflow(
    path="statement.yaml",
    name="statement extraction",
)
workflow_id = response.workflow.workflow_id
if workflow_id is None:
    raise RuntimeError("GroundX did not return a workflow ID")

client.workflows.add_to_id(id=bucket_id, workflow_id=workflow_id)
```

Pass exactly one of `path` or `yaml_text`. The SDK sends the file contents
directly to GroundX. Use
`client.workflows.add_to_account(...)` when the workflow should be the account
default.

Update a workflow by sending the full YAML again:

**`Python`**

```python title="Python"
client.update_extraction_workflow(
    "workflow-id",
    path="statement.yaml",
    name="statement extraction",
)
```

Load the stored workflow definition when you need to inspect it:

**`Python`**

```python title="Python"
definition = client.load_extraction_definition(workflow_id="workflow-id")
```

## 4. Ingest A Document

Upload the document to the assigned bucket with full processing:

**`Python`**

```python title="Python"
import time

from groundx import Document

ingest_response = client.ingest(
    documents=[
        Document(
            bucket_id=bucket_id,
            file_name="statement.pdf",
            file_type="pdf",
            file_path="https://example.com/statement.pdf",
            process_level="full",
        )
    ],
)

process_id = ingest_response.ingest.process_id
while True:
    status_response = client.documents.get_processing_status_by_id(
        process_id=process_id,
    )
    status = status_response.ingest.status
    if status in {"complete", "error", "cancelled"}:
        break
    time.sleep(3)

progress = status_response.ingest.progress
error_documents = (
    progress.errors.documents
    if progress is not None and progress.errors is not None
    else None
)
if error_documents:
    failure = error_documents[0]
    detail = failure.status_message or failure.status or "unknown error"
    raise RuntimeError(f"GroundX document ingest failed: {detail}")

cancelled_documents = (
    progress.cancelled.documents
    if progress is not None and progress.cancelled is not None
    else None
)
if cancelled_documents:
    cancellation = cancelled_documents[0]
    detail = cancellation.status_message or cancellation.status or "unknown reason"
    raise RuntimeError(f"GroundX document ingest was cancelled: {detail}")

if status != "complete":
    detail = status_response.ingest.status_message or status
    raise RuntimeError(f"GroundX ingest failed: {detail}")
```

## 5. Read The Extracted JSON

Read the completed document from the final status response and call
`get_extract`:

**`Python`**

```python title="Python"
progress = status_response.ingest.progress
documents = (
    progress.complete.documents
    if progress is not None and progress.complete is not None
    else None
)
if not documents:
    raise RuntimeError("GroundX did not return the completed document")
document = documents[0]

output = client.documents.get_extract(document.document_id)
print(output)
```

`documents.get_extract()` returns the extracted JSON. [X-Ray document data](/documentation/guides/document-understanding) is separate diagnostic
data; do not build application output from it.

## 6. Improve Accuracy

Compare the returned JSON with reviewed expected values, then change the
smallest prompt that explains each miss. Add the document's exact labels to
`identifiers`, state formatting rules in `instructions`, and tell repeating
groups what to include or exclude. Change prompts without changing group names,
roles, or relationships.

## Advanced: Repeating And Related Records

Use `repeats: true` for object arrays. `unique_attrs` names fields used to
identify the same record.

By default, all `unique_attrs` fields must match for two repeated records to be
treated as the same record. Optional `identity_match` settings refine that
comparison. They are separate from `match_attrs`, which links a child record to
a parent record.

GroundX Python 4.0.0 and later use one string comparison for record identity
and parent relationships. It ignores capitalization and every whitespace
character. It treats only `0` and `o`, `1`, `i`, and `l`, and `8` and `b` as
equivalent. Punctuation, other characters, and remaining length still matter.
Matching never changes extracted, conflict, diagnostic, or returned values.

| Setting                     | Purpose                                                                           |
| --------------------------- | --------------------------------------------------------------------------------- |
| `threshold_attrs`           | Lists fields that are counted instead of individually required.                   |
| `activate_threshold_at`     | Sets how many threshold fields must be present before the threshold rule applies. |
| `minimum_threshold_matches` | Sets how many threshold fields must match once the rule applies.                  |
| `group_attrs`               | Lists identity fields used to group repeated records.                             |
| `sort_attrs`                | Lists identity fields used to order repeated records.                             |
| `equal_value_shortcuts`     | Lists values that satisfy a threshold comparison immediately.                     |

Every field referenced by `identity_match` must exist in the same group.
`group_attrs` and `sort_attrs` must also appear in `unique_attrs`. Each
`equal_value_shortcuts` key must appear in `threshold_attrs`.

Persisted workflows may contain `identity_match.exact_attrs`. GroundX Python
4.0.0 and later still read it, but it no longer changes string matching. Do not
add it to new workflows.

For example, place both settings inside a repeating group whose fields include
`meter_number`, `service_address`, and `meter_type`:

```yaml
unique_attrs: [meter_number, service_address, meter_type]
identity_match:
  threshold_attrs: [service_address, meter_type]
  activate_threshold_at: 2
  minimum_threshold_matches: 1
```

Here, `meter_number` uses universal string matching. Once both threshold fields
have values, at least one of them must also match.

On a child group, `match_attrs` lists every value used to find its parent.
`passthrough.from` names the parent group. Every match field must use the same
output name in both groups. GroundX:

* compares the extracted values, not their surrounding metadata;
* uses the same capitalization, whitespace, and OCR-confusable string
  comparison as repeated-record identity;
* compares integers and floats numerically;
* ignores blank or missing fields;
* requires both records to provide the same match fields;
* uses the first matching parent when more than one matches.

Matched children appear inside their parent. Unmatched children remain in the
top-level child group. `unique_attrs` does not add relationship fields.

**`v1-parent-child-relationship.yaml`**

```yaml title="v1-parent-child-relationship.yaml"
extraction_policy_version: v1

workflow:
  custom_steps:
    - name: parent_fields
      level: chunk
      kind: summary
    - name: child_fields
      level: chunk
      kind: keys
  agent_chain:
    - parallel:
        - group: parent_records
          chain: [reconcile_meters, qa_meters, save_meters]
        - group: child_records
          chain: [reconcile_charges, save_charges]

parent_records:
  role: meters
  repeats: true
  unique_attrs:
    - relationship_id
  workflow_step: parent_fields
  fields:
    relationship_id:
      workflow_output_key: relationship_id
      prompt:
        description: The identifier shared with related child records.
        identifiers: [Relationship ID]
        instructions: Return the identifier exactly as printed.
        type: str
    parent_name:
      workflow_output_key: parent_name
      prompt:
        description: The name of the parent record.
        identifiers: [Parent Name]
        instructions: Return the parent name exactly as printed.
        type: str

child_records:
  role: charges
  repeats: true
  unique_attrs:
    - child_id
  match_attrs:
    - relationship_id
  passthrough:
    from: parent_records
  workflow_step: child_fields
  fields:
    child_id:
      workflow_output_key: child_id
      prompt:
        description: The identifier for this child record.
        identifiers: [Child ID]
        instructions: Return the child identifier exactly as printed.
        type: str
    relationship_id:
      workflow_output_key: relationship_id
      prompt:
        description: The identifier shared with the related parent record.
        identifiers: [Relationship ID]
        instructions: Return the identifier exactly as printed.
        type: str
    child_value:
      workflow_output_key: child_value
      prompt:
        description: The value for this child record.
        identifiers: [Child Value]
        instructions: Return the value exactly as printed.
        type: str
```

For example, one matched child and one unmatched child produce:

```json
{
  "parent_records": [
    {
      "relationship_id": "A-100",
      "parent_name": "Primary record",
      "child_records": [
        {
          "child_id": "C-1",
          "relationship_id": "A-100",
          "child_value": "Matched value"
        }
      ]
    }
  ],
  "child_records": [
    {
      "child_id": "C-2",
      "relationship_id": "B-200",
      "child_value": "Unmatched value"
    }
  ]
}
```