Extract Data from Documents | Documentation

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:

1 {
2   "statement": {
3     "account_number": "123456789",
4     "due_date": "2026-06-30",
5     "total_amount_due": 128.55
6   },
7   "service": {
8     "service_address": "100 Main St, Denver, CO 80202"
9   }
10 }

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

1 extraction_policy_version: v1
2 
3 workflow:
4   custom_steps:
5     - name: statement_fields
6       level: chunk
7       kind: instruct
8     - name: service_fields
9       level: chunk
10       kind: instruct
11   agent_chain:
12     - parallel:
13         - group: statement
14           chain: [reconcile_statement, qa_statement, save_statement]
15         - group: service
16           chain: [reconcile_statement, qa_statement, save_statement]
17 
18 statement:
19   role: statement
20   workflow_step: statement_fields
21   fields:
22     account_number:
23       workflow_output_key: account_number
24       prompt:
25         description: The utility account number printed on the statement.
26         identifiers: [Account Number, "Account #"]
27         instructions: Return the account number exactly as printed.
28         type: str
29     due_date:
30       workflow_output_key: due_date
31       prompt:
32         description: The payment due date.
33         identifiers: [Due Date, Payment Due]
34         instructions: Return the date as YYYY-MM-DD.
35         type: str
36     total_amount_due:
37       workflow_output_key: total_amount_due
38       prompt:
39         description: The final amount due for the statement.
40         identifiers: [Total Amount Due, Amount Due]
41         instructions: Return only the numeric amount.
42         type: float
43 
44 service:
45   role: statement
46   workflow_step: service_fields
47   fields:
48     service_address:
49       workflow_output_key: service_address
50       prompt:
51         description: The billed service address.
52         identifiers: [Service Address, Service Location]
53         instructions: Return the address exactly as printed.
54         type: str

3. Create And Assign The Workflow

Install the extraction helpers:

1 pip install "groundx[extract]"

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

Python

1 import os
2 
3 from groundx import GroundX
4 
5 client = GroundX(api_key=os.environ["GROUNDX_API_KEY"])
6 bucket_id = 1234
7 
8 response = client.create_extraction_workflow(
9     path="statement.yaml",
10     name="statement extraction",
11 )
12 workflow_id = response.workflow.workflow_id
13 if workflow_id is None:
14     raise RuntimeError("GroundX did not return a workflow ID")
15 
16 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

1 client.update_extraction_workflow(
2     "workflow-id",
3     path="statement.yaml",
4     name="statement extraction",
5 )

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

Python

1 definition = client.load_extraction_definition(workflow_id="workflow-id")

4. Ingest A Document

Upload the document to the assigned bucket with full processing:

Python

1 import time
2 
3 from groundx import Document
4 
5 ingest_response = client.ingest(
6     documents=[
7         Document(
8             bucket_id=bucket_id,
9             file_name="statement.pdf",
10             file_type="pdf",
11             file_path="https://example.com/statement.pdf",
12             process_level="full",
13         )
14     ],
15 )
16 
17 process_id = ingest_response.ingest.process_id
18 while True:
19     status_response = client.documents.get_processing_status_by_id(
20         process_id=process_id,
21     )
22     status = status_response.ingest.status
23     if status in {"complete", "error", "cancelled"}:
24         break
25     time.sleep(3)
26 
27 progress = status_response.ingest.progress
28 error_documents = (
29     progress.errors.documents
30     if progress is not None and progress.errors is not None
31     else None
32 )
33 if error_documents:
34     failure = error_documents[0]
35     detail = failure.status_message or failure.status or "unknown error"
36     raise RuntimeError(f"GroundX document ingest failed: {detail}")
37 
38 cancelled_documents = (
39     progress.cancelled.documents
40     if progress is not None and progress.cancelled is not None
41     else None
42 )
43 if cancelled_documents:
44     cancellation = cancelled_documents[0]
45     detail = cancellation.status_message or cancellation.status or "unknown reason"
46     raise RuntimeError(f"GroundX document ingest was cancelled: {detail}")
47 
48 if status != "complete":
49     detail = status_response.ingest.status_message or status
50     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

1 progress = status_response.ingest.progress
2 documents = (
3     progress.complete.documents
4     if progress is not None and progress.complete is not None
5     else None
6 )
7 if not documents:
8     raise RuntimeError("GroundX did not return the completed document")
9 document = documents[0]
10 
11 output = client.documents.get_extract(document.document_id)
12 print(output)

documents.get_extract() returns the extracted JSON. X-Ray document data 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.

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:

1 unique_attrs: [meter_number, service_address, meter_type]
2 identity_match:
3   threshold_attrs: [service_address, meter_type]
4   activate_threshold_at: 2
5   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

1 extraction_policy_version: v1
2 
3 workflow:
4   custom_steps:
5     - name: parent_fields
6       level: chunk
7       kind: summary
8     - name: child_fields
9       level: chunk
10       kind: keys
11   agent_chain:
12     - parallel:
13         - group: parent_records
14           chain: [reconcile_meters, qa_meters, save_meters]
15         - group: child_records
16           chain: [reconcile_charges, save_charges]
17 
18 parent_records:
19   role: meters
20   repeats: true
21   unique_attrs:
22     - relationship_id
23   workflow_step: parent_fields
24   fields:
25     relationship_id:
26       workflow_output_key: relationship_id
27       prompt:
28         description: The identifier shared with related child records.
29         identifiers: [Relationship ID]
30         instructions: Return the identifier exactly as printed.
31         type: str
32     parent_name:
33       workflow_output_key: parent_name
34       prompt:
35         description: The name of the parent record.
36         identifiers: [Parent Name]
37         instructions: Return the parent name exactly as printed.
38         type: str
39 
40 child_records:
41   role: charges
42   repeats: true
43   unique_attrs:
44     - child_id
45   match_attrs:
46     - relationship_id
47   passthrough:
48     from: parent_records
49   workflow_step: child_fields
50   fields:
51     child_id:
52       workflow_output_key: child_id
53       prompt:
54         description: The identifier for this child record.
55         identifiers: [Child ID]
56         instructions: Return the child identifier exactly as printed.
57         type: str
58     relationship_id:
59       workflow_output_key: relationship_id
60       prompt:
61         description: The identifier shared with the related parent record.
62         identifiers: [Relationship ID]
63         instructions: Return the identifier exactly as printed.
64         type: str
65     child_value:
66       workflow_output_key: child_value
67       prompt:
68         description: The value for this child record.
69         identifiers: [Child Value]
70         instructions: Return the value exactly as printed.
71         type: str

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

1 {
2   "parent_records": [
3     {
4       "relationship_id": "A-100",
5       "parent_name": "Primary record",
6       "child_records": [
7         {
8           "child_id": "C-1",
9           "relationship_id": "A-100",
10           "child_value": "Matched value"
11         }
12       ]
13     }
14   ],
15   "child_records": [
16     {
17       "child_id": "C-2",
18       "relationship_id": "B-200",
19       "child_value": "Unmatched value"
20     }
21   ]
22 }

1. Start With The JSON You Want

2. Write The Workflow YAML

3. Create And Assign The Workflow

4. Ingest A Document

5. Read The Extracted JSON

6. Improve Accuracy

Advanced: Repeating And Related Records