Extract Data from Documents

1
statement:
2
  prompt:
3
    instructions: Extract billing statement values from the document.
4
  fields:
5
    account_number:
6
      prompt:
7
        description: The utility account number printed on the statement.
8
        identifiers:
9
          - Account Number
10
          - Account #
11
        instructions: Return the account number exactly as printed.
12
        type: str
13
    due_date:
14
      prompt:
15
        description: The payment due date.
16
        identifiers:
17
          - Due Date
18
          - Payment Due
19
        instructions: Return the due date as YYYY-MM-DD.
20
        type: str
21
    total_amount_due:
22
      prompt:
23
        description: The final amount due for the statement.
24
        identifiers:
25
          - Total Amount Due
26
          - Amount Due
27
        instructions: Return only the numeric amount, without a currency symbol.
28
        type: float
29
30
service:
31
  prompt:
32
    instructions: Extract service location values from the statement.
33
  fields:
34
    service_address:
35
      prompt:
36
        description: The service address for the billed location.
37
        identifiers:
38
          - Service Address
39
          - Service Location
40
        instructions: Return the address exactly as printed.
41
        type: str

1
from pathlib import Path
2
3
from groundx.extract import prepare_extraction_yaml
4
5
yaml_text = Path("statement.yaml").read_text()
6
prepared = prepare_extraction_yaml(yaml_text)
7
8
extract_settings = prepared.workflow_groups

1
import os
2
from pathlib import Path
3
4
from groundx import GroundX
5
from groundx.extract import prepare_extraction_yaml
6
7
client = GroundX(api_key=os.environ["GROUNDX_API_KEY"])
8
bucket_id = 1234
9
10
prepared = prepare_extraction_yaml(Path("statement.yaml").read_text())
11
12
workflow_response = client.workflows.create(
13
    name="statement extraction",
14
    extract=prepared.workflow_groups,
15
)
16
17
workflow_id = workflow_response.workflow.workflow_id
18
if workflow_id is None:
19
    raise RuntimeError("GroundX did not return a workflow ID")
20
21
client.workflows.add_to_id(id=bucket_id, workflow_id=workflow_id)

1
import os
2
import time
3
4
from groundx import Document, GroundX
5
6
client = GroundX(api_key=os.environ["GROUNDX_API_KEY"])
7
bucket_id = 1234
8
9
ingest_response = client.ingest(
10
    documents=[
11
        Document(
12
            bucket_id=bucket_id,
13
            file_name="statement.pdf",
14
            file_type="pdf",
15
            file_path="https://example.com/statement.pdf",
16
            process_level="full",
17
        )
18
    ],
19
)
20
21
process_id = ingest_response.ingest.process_id
22
23
while True:
24
    ingest_response = client.documents.get_processing_status_by_id(
25
        process_id=process_id,
26
    )
27
    status = ingest_response.ingest.status
28
29
    if status in {"complete", "error", "cancelled"}:
30
        break
31
32
    time.sleep(3)
33
34
if status != "complete":
35
    raise RuntimeError(f"GroundX ingest ended with status: {status}")

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
process_id = "PROCESS_ID_FROM_INGEST"
8
9
lookup = client.documents.lookup(
10
    id=bucket_id,
11
    status="complete",
12
    n=100,
13
)
14
15
document = next(
16
    (
17
        item
18
        for item in lookup.documents or []
19
        if item.process_id == process_id
20
    ),
21
    None,
22
)
23
24
if document is None:
25
    raise RuntimeError("Could not find the completed GroundX document")
26
27
extraction = client.documents.get_extract(document.document_id)
28
print(extraction)