Processing

Track Kodexa document processing steps, knowledge items, and execution pipelines from Python using the processing module of the Python SDK.

The processing module provides classes for tracking document processing history, recording knowledge items, and managing execution pipelines.

ProcessingStep

A ProcessingStep represents a unit of work in a document processing pipeline. Steps form a DAG (directed acyclic graph) with parent-child relationships, enabling you to track the full lineage of how a document was processed.

Creating Steps

from kodexa_document import ProcessingStep

# Create a processing step
step = ProcessingStep(name="PDF Extraction")

# With metadata
step = ProcessingStep(
    name="Invoice Classification",
    metadata={"model_version": "2.1", "threshold": 0.85},
    presentation_metadata={"icon": "file-invoice", "color": "blue"}
)

Fields

Field	Type	Description
id	str	UUID (auto-generated)
name	str	Step name (required)
start_timestamp	datetime	When the step started
duration	int	Duration in milliseconds
metadata	dict	Arbitrary key-value metadata
presentation_metadata	dict	UI display hints
children	List[ProcessingStep]	Child steps
parents	List[ProcessingStep]	Parent steps
internal_steps	List[ProcessingStep]	Internal sub-steps
knowledge_items	List[KnowledgeItem]	Associated knowledge items
model_use	List[ModelUse]	LLM calls attributed to this step

Model Usage

model_use records the LLM calls a step made, so a document’s real model spend is visible in the Explain Plan and in model-use analytics. Steps that recorded their usage only as freeform metadata['llm_usage'] — extraction, classification and chunking steps — surface it as a typed ModelUse entry whenever the document is loaded; the freeform metadata is left in place for older readers, and a typed entry always takes precedence over it. The ModelUse class is importable from kodexa_document.processing.types.

Field	Type	Description
model	str	Model id the call used
provider	str	Provider that served the call, from a fixed set (anthropic, anthropic_bedrock, bedrock, cohere_bedrock, openai, openai_azure, google_gemini, xai, perplexity, other)
input_tokens	int	Prompt tokens
output_tokens	int	Completion tokens
total_tokens	int	Total tokens (summed from input and output when not supplied)
duration_ms	int	Call duration in milliseconds
cost_micros	int	Cost priced at the time of the call, in millionths of a US dollar; 0 when no price was available or when the entry came from freeform metadata
finish_reason	str	Why the call stopped
request_id	str	Provider request identifier

step = ProcessingStep(name="Data Extraction")

for use in step.model_use:
    print(f"{use.model} ({use.provider}): {use.total_tokens} tokens in {use.duration_ms}ms")

Recording an LLM Call

record_llm_call() records a call’s usage together with the prompt, thinking and response blocks shown in the Explain Plan:

step.record_llm_call(
    model="my-model-id",
    provider="anthropic_bedrock",
    response="Extracted 12 fields",
    input_tokens=1200,
    output_tokens=180,
    duration_ms=940,
)

A provider outside that fixed set is stored as other, with the name you passed kept on original_provider_name.

A ModelUse entry is added only when at least one usage field is set — a model, provider, token count, cost, duration, finish reason or request id. A wrapper that only narrates the prompt and response therefore leaves the platform’s own recorded entry as the step’s single usage record. The prompt, thinking, response and schema body blocks are recorded either way.

Parent-Child Relationships

Build processing hierarchies:

# Create a pipeline
pipeline = ProcessingStep(name="Document Pipeline")

# Add child steps
extraction = ProcessingStep(name="Text Extraction")
pipeline.add_child(extraction)  # Bidirectional link

classification = ProcessingStep(name="Classification")
pipeline.add_child(classification)

tagging = ProcessingStep(name="Entity Tagging")
extraction.add_child(tagging)

# Children know their parents
print(tagging.parents[0].name)  # "Text Extraction"

Merging Steps

Combine multiple processing branches:

# Two independent processing steps
ocr_step = ProcessingStep(name="OCR Processing")
nlp_step = ProcessingStep(name="NLP Analysis")

# Merge them into a combined step
merged = ProcessingStep.merge_with(ocr_step, nlp_step)
print(merged.name)  # "Merged Step"
print(len(merged.parents))  # 2

Serialization

Steps serialize to JSON with circular reference handling:

step = ProcessingStep(name="My Step")
child = ProcessingStep(name="Child Step")
step.add_child(child)

# To dict/JSON
step_dict = step.to_dict()
json_str = step.to_json()

# From dict/JSON
restored = ProcessingStep.from_dict(step_dict)
restored = ProcessingStep.from_json(json_str)

The to_dict() method uses a seen set to handle circular references from bidirectional parent-child links. The from_dict() method uses a step_cache to reconstruct these references.

KnowledgeItem

A KnowledgeItem represents a piece of knowledge produced or consumed during processing.

from kodexa_document.processing.processing_step import KnowledgeItem, KnowledgeFeature

item = KnowledgeItem(
    title="Invoice #12345",
    description="Extracted invoice data",
    slug="invoice-12345",
    knowledge_item_type_ref="knowledge-item-type://my-org/invoice",
    properties={"total": 1234.56, "vendor": "Acme Corp"}
)

Fields

Field	Type	Description
id	str	UUID (auto-generated)
title	str	Display title
description	str	Description
slug	str	URL-friendly identifier
sequence_order	int	Ordering within a set
knowledge_item_type_ref	str	Reference to the item type
properties	dict	Arbitrary properties
features	List[KnowledgeFeature]	Associated features

KnowledgeFeature

A KnowledgeFeature attaches structured metadata to knowledge items.

feature = KnowledgeFeature(
    feature_type_ref="knowledge-feature-type://my-org/confidence-score",
    slug="extraction-confidence",
    properties={"score": 0.95, "model": "invoice-v2"},
    extended_properties={"per_field_scores": {"total": 0.99, "date": 0.91}}
)

item.features.append(feature)

Fields

Field	Type	Description
id	str	UUID (auto-generated)
feature_type_ref	str	Reference to the feature type
slug	str	URL-friendly identifier
active	bool	Whether this feature is active (default: True)
properties	dict	Core properties
extended_properties	dict	Additional properties

Attaching to Processing Steps

Knowledge items are associated with processing steps:

step = ProcessingStep(name="Invoice Extraction")

item = KnowledgeItem(
    title="Invoice #12345",
    knowledge_item_type_ref="knowledge-item-type://my-org/invoice"
)

feature = KnowledgeFeature(
    feature_type_ref="knowledge-feature-type://my-org/extraction-result",
    properties={"status": "complete", "field_count": 12}
)
item.features.append(feature)

step.knowledge_items.append(item)

PipelineContext

PipelineContext tracks the state of a running execution pipeline. It is primarily used by module developers building custom processing steps.

from kodexa_document.platform.kodexa import PipelineContext

context = PipelineContext(
    execution_id="exec-abc-123",
    content_provider=my_content_provider
)

Fields

Field	Type	Description
execution_id	str	UUID for this execution (auto-generated if not provided)
statistics	PipelineStatistics	Tracks documents processed
output_document	Document	The output document
content_objects	list	Content objects in the pipeline
stop_on_exception	bool	Whether to halt on errors (default: True)
current_document	Document	The document currently being processed
document_family	DocumentFamily	The document family context
document_store	Store	The associated document store

Status Updates

Report progress during execution:

context.update_status("Extracting page 5 of 20", progress=5, progress_max=20)

Cancellation

Check for user-initiated cancellation:

if context.is_cancelled():
    print("Execution cancelled by user")
    return

RemoteStep

RemoteStep wraps a reference to a module on the platform and can process documents remotely:

from kodexa_document.platform.client import RemoteStep

step = RemoteStep(
    ref="my-org/fast-pdf-model:1.0.0",
    step_type="ACTION",
    options={"dpi": 300, "language": "en"}
)

# Process a document through the remote module
result_doc = step.process(document, context)

Complete Example

from datetime import datetime
from kodexa_document import ProcessingStep
from kodexa_document.processing.processing_step import KnowledgeItem, KnowledgeFeature

# Build a processing pipeline history
pipeline = ProcessingStep(
    name="Invoice Processing Pipeline",
    start_timestamp=datetime.now(),
    metadata={"version": "3.0"}
)

# Stage 1: PDF extraction
extraction = ProcessingStep(
    name="PDF Extraction",
    metadata={"model": "fast-pdf-model", "pages": 3}
)
pipeline.add_child(extraction)

# Stage 2: Classification
classification = ProcessingStep(
    name="Document Classification",
    metadata={"result": "invoice", "confidence": 0.98}
)
pipeline.add_child(classification)

# Stage 3: Data extraction (depends on both previous steps)
data_extraction = ProcessingStep(
    name="Data Extraction",
    metadata={"fields_extracted": 12}
)
extraction.add_child(data_extraction)
classification.add_child(data_extraction)

# Attach knowledge item to the data extraction step
item = KnowledgeItem(
    title="Invoice #12345",
    knowledge_item_type_ref="knowledge-item-type://acme/invoice",
    properties={"vendor": "Acme Corp", "total": 1234.56}
)

item.features.append(KnowledgeFeature(
    feature_type_ref="knowledge-feature-type://acme/extraction-score",
    properties={"overall_confidence": 0.95}
))

data_extraction.knowledge_items.append(item)

# Serialize the full pipeline
pipeline_json = pipeline.to_json()

# Deserialize later
restored = ProcessingStep.from_json(pipeline_json)
print(f"Pipeline: {restored.name}")
print(f"Steps: {len(restored.children)}")