Finalize orchestration for invoice data extraction

Author: jamesmcroft

src/AIDocumentPipeline/function_app.py

@@ -1,6 +1,6 @@
 import azure.functions as func
 import azure.durable_functions as df
-from invoices import process_invoice_batch_workflow
+from invoices import process_invoice_batch_workflow, extract_invoice_data_workflow
 from invoices.activities import extract_invoice_data, get_invoice_folders
 from shared.storage import write_bytes_to_blob
 
@@ -9,3 +9,4 @@
 app.register_functions(extract_invoice_data.bp)
 app.register_functions(get_invoice_folders.bp)
 app.register_functions(process_invoice_batch_workflow.bp)
+app.register_functions(extract_invoice_data_workflow.bp)

src/AIDocumentPipeline/invoices/activities/extract_invoice_data.py

@@ -32,7 +32,7 @@ def run(input: Request) -> InvoiceData | None:
             system_prompt="You are an AI assistant that extracts data from documents and returns them as structured JSON objects. Do not return as a code block.",
             extraction_prompt=f"Extract the data from this invoice. If a value is not present, provide null. Use the following structure: {InvoiceData.empty().to_dict()}",
             endpoint=app_config.openai_endpoint,
-            deployment_name=app_config.openai_vision_completion_deployment,
+            deployment_name=app_config.openai_completion_deployment,
             max_tokens=4096,
             temperature=0.1,
             top_p=0.1

src/AIDocumentPipeline/invoices/extract_invoice_data_workflow.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+from shared.storage import write_bytes_to_blob
+from invoices.activities import extract_invoice_data
+from shared.workflow_result import WorkflowResult
+import azure.durable_functions as df
+import azure.functions as func
+import logging
+from shared import config as app_config
+
+name = "ExtractInvoiceDataWorkflow"
+bp = df.Blueprint()
+
+
+@bp.function_name(name)
+@bp.orchestration_trigger(context_name="context", orchestration=name)
+def run(context: df.DurableOrchestrationContext) -> WorkflowResult:
+    # Step 1: Extract the input from the context
+    input = context.get_input()
+    result = WorkflowResult(input.name)
+
+    # Step 2: Validate the input
+    validation_result = input.validate()
+    if not validation_result.is_valid:
+        result.merge(validation_result)
+        return result
+
+    result.add_message("InvoiceFolder.validate", "input is valid")
+
+    # Step 3: Get the invoice folders from the blob container
+    for invoice in input.invoice_file_names:
+        invoice_data = yield context.call_activity(extract_invoice_data.name, extract_invoice_data.Request(input.container_name, invoice))
+
+        if not invoice_data:
+            result.add_error(extract_invoice_data.name,
+                             f"Failed to extract data for {invoice}.")
+            continue
+
+        invoice_data_stored = yield context.call_activity(write_bytes_to_blob.name, write_bytes_to_blob.Request(app_config.invoices_storage_account_name, input.container_name, f"{invoice}.Data.json", invoice_data))
+
+        if not invoice_data_stored:
+            result.add_error(write_bytes_to_blob.name,
+                             f"Failed to store extracted data for {invoice}.")
+            continue
+
+    return result

src/AIDocumentPipeline/invoices/process_invoice_batch_workflow.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
+from invoices import extract_invoice_data_workflow
 from shared.workflow_result import WorkflowResult
 from invoices.invoice_batch_request import InvoiceBatchRequest
-from shared import config as app_config
 import azure.durable_functions as df
 import azure.functions as func
 import logging
-from invoices.activities import extract_invoice_data, get_invoice_folders
+from invoices.activities import get_invoice_folders
 
 name = "ProcessInvoiceBatchWorkflow"
 http_trigger_name = "ProcessInvoiceBatchHttp"
@@ -49,5 +49,15 @@ def run(context: df.DurableOrchestrationContext) -> WorkflowResult:
                        f"Retrieved {len(invoice_folders)} invoice folders.")
 
     # Step 4: Process the invoices in each folder.
+    extract_invoice_data_tasks = []
+    for folder in invoice_folders:
+        extract_invoice_data_task = context.call_sub_orchestrator(
+            extract_invoice_data_workflow.name, folder)
+        extract_invoice_data_tasks.append(extract_invoice_data_task)
+
+    yield context.task_all(extract_invoice_data_tasks)
+
+    for task in extract_invoice_data_tasks:
+        logging.info(f"Task {task} completed.")
 
     return result

src/AIDocumentPipeline/shared/config.py

@@ -2,14 +2,8 @@
 
 otlp_exporter_endpoint = os.environ.get("OTLP_EXPORTER_ENDPOINT", None)
 openai_endpoint = os.environ.get("OPENAI_ENDPOINT", None)
-openai_embedding_deployment = os.environ.get(
-    "OPENAI_EMBEDDING_DEPLOYMENT", None)
 openai_completion_deployment = os.environ.get(
     "OPENAI_COMPLETION_DEPLOYMENT", None)
-openai_vision_completion_deployment = os.environ.get(
-    "OPENAI_VISION_COMPLETION_DEPLOYMENT", None)
-document_intelligence_endpoint = os.environ.get(
-    "DOCUMENT_INTELLIGENCE_ENDPOINT", None)
 managed_identity_client_id = os.environ.get("MANAGED_IDENTITY_CLIENT_ID", None)
 invoices_storage_account_name = os.environ.get(
     "INVOICES_STORAGE_ACCOUNT_NAME", None)

src/AIDocumentPipeline/shared/documents/document_data_extractor.py

@@ -33,7 +33,7 @@ def from_bytes(self, document_bytes: bytes, options: DocumentDataExtractorOption
 
         for image_uri in image_uris:
             user_content.append({
-                "type": "image_uri",
+                "type": "image_url",
                 "image_url": {
                     "url": image_uri
                 }