> ## Documentation Index > Fetch the complete documentation index at: https://developer.kodexa.ai/llms.txt > Use this file to discover all available pages before exploring further. # 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 ```python theme={null} 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 | ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} # 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: ```python theme={null} # 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: ```python theme={null} 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. ```python theme={null} 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. ```python theme={null} 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: ```python theme={null} 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. ```python theme={null} 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: ```python theme={null} context.update_status("Extracting page 5 of 20", progress=5, progress_max=20) ``` ### Cancellation Check for user-initiated cancellation: ```python theme={null} 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: ```python theme={null} 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 ```python theme={null} 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)}") ```