Document text extractor node (Public preview)

​

Pre-requisites

orchestrate server start -e <.env file path> -d

​

Configuring document processing in flows

from pydantic import BaseModel, Field
from ibm_watsonx_orchestrate.flow_builder.flows import (
    Flow, flow, START, END
)

from ibm_watsonx_orchestrate.flow_builder.types import DocProcInput, PlainTextReadingOrder


@flow(
    name ="text_extraction_flow_example",
    display_name="text_extraction_flow_example",
    description="This flow consists of one node: a docproc node, which extracts text from the input document",
    input_schema=DocProcInput
)
def build_docproc_flow(aflow: Flow = None) -> Flow:
    doc_proc_node = aflow.docproc(
        name="text_extraction",
        display_name="text_extraction",
        description="Extract text out of a document's contents.",
        task="text_extraction",
        enable_hw=True
    )

    aflow.sequence(START, doc_proc_node, END)
    return aflow

​

Semantic Key-Value Pair (KVP) Extraction

• Node specification: Set kvp_schemas in the node configuration. The node supports semantic KVP extraction in the following cases:
  • If you define kvp_schemas in the input, the node uses those schemas. If you pass an empty array, it falls back to default schemas.
  • If you don’t define kvp_schemas in the input but include them in the node specification, the node uses the specification-defined schemas. Again, if the array is empty, it defaults to the built-in schemas.
  The kvp_schemas is configured as a JSON object. The following example shows how to define kvp_schemas in a node configuration.
  Python

  Copy

  Ask AI

  from pydantic import BaseModel, Field
  from ibm_watsonx_orchestrate.flow_builder.flows import (
      Flow, flow, START, END
  )
  
  from ibm_watsonx_orchestrate.flow_builder.types import DocProcInput, PlainTextReadingOrder
  
  
  @flow(
      name ="kvp_extraction_flow_example_for_api",
      display_name="kvp_extraction_flow_example_for_api",
      description="This flow consists of one node: a docproc node, which extracts kvps from the input document",
      input_schema=DocProcInput
  )
  def build_docproc_flow(aflow: Flow = None) -> Flow:
      doc_proc_node = aflow.docproc(
          name="kvp_extraction",
          display_name="kvp_extraction",
          description="Extract kvp out of a document's contents.",
          task="text_extraction",
          enable_hw=True,
          plain_text_reading_order=PlainTextReadingOrder.block_structure,
          kvp_schemas=[{
          "document_type": "Invoice",
          "document_description": "An invoice is a standard document issued by a seller to a buyer, outlining products or services provided, quantities, prices, and payment terms.",
          "additional_prompt_instructions": "Extract values exactly as they appear in the document, especially numbers and codes.",
          "fields": {
          "invoice_number": {
              "description": "A unique identifier assigned by the vendor for this invoice.",
              "example": "2023-AUS-987654",
              "default": ""
          },
          "document_date": {
              "description": "Date of the document.",
              "example": "2025-07-05",
              "default": ""
          },
          "vendor_name": {
              "description": "Legal or trade name of the company issuing the invoice. Usually located in the header or footer, near the logo, or billing details.",
              "example": "ABC Supply Company Pty Ltd",
              "default": ""
          },
          "vendor_number": {
              "description": "Internal identifier used by the buyer's system to refer to the vendor.",
              "example": "VEND-1023",
              "default": ""
          }
          }
      }]
      )
  
      aflow.sequence(START, doc_proc_node, END)
      return aflow
  

  See all 53 lines
• Runtime input: Set kvp_schemas in the input payload. The following example shows how to define a kvp_schemas in the input payload.
  Python

  Copy

  Ask AI

  import asyncio
  import logging
  import sys
  import json
  from pathlib import Path
  
  from examples.flow_builder.text_extraction.tools.text_extraction_flow import build_docproc_flow
  
  logger = logging.getLogger(__name__)
  
  
  def on_flow_end(result):
      """
      Callback function to be called when the flow is completed.
      """
      print(f"Custom Handler: flow `{flow_run.name}` completed with result: {result}")
  
  
  def on_flow_error(error):
      """
      Callback function to be called when the flow fails.
      """
      print(f"Custom Handler: flow `{flow_run.name}` failed: {error}")
  
  
  async def main(doc_ref: str, kvp_schema_path):
      '''A function demonstrating how to build a flow and save it to a file.'''
      my_flow_definition = await build_docproc_flow().compile_deploy()
      global flow_name
      flow_name = my_flow_definition.flow.spec.display_name
      generated_folder = f"{Path(__file__).resolve().parent}/generated"
      my_flow_definition.dump_spec(f"{generated_folder}/docproc_flow_spec.json")
  
      with open(kvp_schema_path, 'r') as file:
          # Load the JSON data from the file into a Python dictionary
          schema_json = json.load(file)
  
      global flow_run
      flow_run = await my_flow_definition.invoke(
          {
          "document_ref": doc_ref, 
          "language": "en",
          "kvp_schemas": [ schema_json ]
          }, 
          on_flow_end_handler=on_flow_end, on_flow_error_handler=on_flow_error, debug=True)
  
  
  if __name__ == "__main__":
      if len(sys.argv) != 3:
          logger.error(f"Usage: {sys.argv[0]} file_store_path kvp_schema_path")
      else:
          asyncio.run(main(sys.argv[1], sys.argv[2]))
  

  See all 52 lines