Pipeline Technical Guide

Internal architecture, data flow, service responsibilities, entity model internals, NiFi integration, merge/materialization mechanics, and configuration reference.
This guide is for pipeline developers and operators — people building or modifying the processing services, configuring NiFi flows, debugging entity state, or extending the annotation system.

---

Table of Contents

• Architecture
• Service Inventory
• NiFi Entity Model (Internals)
• NiFi Integration Layer
• Processing Lifecycle
• Service Details
  • nifi-dashboard
  • data-cutter
  • document-extractor
  • audio-transcription
  • image-augmentation
  • object-detection
• Fragment Internals
• Merge and Materialization Internals
• Annotation Merge Mechanics
• RAG Pipeline Internals
• Configuration Reference

---

Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                           Apache NiFi                                   │
│                     (Flow orchestration layer)                          │
│                                                                         │
│  FlowFiles carry serialized NifiEntityModel JSON                        │
│  Routing by MIME type (os_item_content_type) and concept membership      │
│  InvokeHTTP calls each service's /api/nifi/* endpoints                  │
└────┬──────────┬──────────┬──────────┬──────────┬──────────┬────────────┘
     │          │          │          │          │          │
     ▼          ▼          ▼          ▼          ▼          ▼
┌─────────┐┌─────────┐┌─────────┐┌─────────┐┌─────────┐┌─────────────┐
│  nifi-  ││  data-  ││document-││  audio- ││ image-  ││   object-   │
│dashboard││ cutter  ││extractor││ transc. ││ augment.││  detection  │
│         ││         ││         ││         ││         ││             │
│FastAPI  ││FastAPI  ││FastAPI  ││FastAPI  ││FastAPI  ││FastAPI      │
│+Celery  ││+Celery  ││+Celery  ││+Celery  ││+Celery  ││+Celery      │
└─────────┘└─────────┘└─────────┘└─────────┘└─────────┘└─────────────┘
     │          │          │          │          │          │
     └──────────┴──────────┴──────────┴──────────┴──────────┘
                              │
           ┌──────────────────┼──────────────────┐
           │                  │                  │
    ┌──────▼──────┐   ┌──────▼──────┐   ┌───────▼──────┐
    │  OpenSearch  │   │    Timbr    │   │   Octostar   │
    │ (search idx) │   │ (ontology)  │   │  (platform)  │
    └─────────────┘   └─────────────┘   └──────────────┘

All services share:

• streamlit-octostar-utils — NiFi entity model, route framework, context manager, fragmenter, contents abstraction, NLP utilities
• octostar client — Platform API (entity sync, upsert, attachment write)
• FastAPI + Celery pattern — HTTP endpoints for NiFi InvokeHTTP, Celery workers for async processing
• config.yaml — Per-service configuration (queues, task limits, model settings)
• manifest.yaml — Platform registration (alias, resource limits, UI integration)

---

Service Inventory

Service	Alias	NiFi Endpoints	Celery Queues	Key Dependencies
nifi-dashboard	nifi-dashboard	/pre-process, /post-process, /search-send, /rag-embed, /firehose-hydrate	default, embedding, fast+embedding	OpenSearch, Timbr, Redis
data-cutter	data-cutter	/split-document, /split-audio, /extract-keyframes-from-video, /extract-audio-from-video, /split-primitive, /merge-fragments, /merge-primitive, /create-rag-chunks, /materialize-fragments	default, libreoffice	FFmpeg, LibreOffice, octostar_ai
document-extractor	document-extractor	/extract-text, /extract-metadata, /extract-ocr, /detect-language, /extract-ner-entities, /extract-spreadsheet-entities, /summarize, /summarize-deep	default, exiftool, tika, ocr, ner, llm, llm_deep	Apache Tika, ExifTool, EasyOCR/Tesseract, PyMuPDF, SpaCy/Flair
audio-transcription	audio-transcription	/transcribe-audio, /translate-text	default	Whisper (HuggingFace) or AssemblyAI, Argos Translate
image-augmentation	image-augmentation	/caption-and-tag-images	default, llm	OpenAI-compatible vision LLM, SentenceTransformers
object-detection	object-detection	/detect-faces, /detect-general-embeddings	default	MTCNN, FaceNet (InceptionResnetV1), CLIP ViT-B/32, Wand/ImageMagick

---

NiFi Entity Model (Internals)

Defined in streamlit_octostar_utils.api_crafter.nifi.

NifiEntityModel (Wire Format / Pydantic)

class NifiEntityModel(BaseModel):
    request: RequestModel
    record: RecordModel          # extra="allow"
    annotations: Dict[str, Any] = {}
    children: List[Union[NifiOTMRelationshipProxyModel, NifiProxyEntityModel]] = []
    contents: Optional[Dict[str, Any]] = None

RequestModel

class RequestModel(BaseModel):
    jwt: Optional[str]
    ontology_name: str
    ontology_info: OntologyInfoModel  # parents, relationships, label_keys
    entity_timestamp: Optional[str]
    sync_params: dict = {}
    nifi_attributes: dict = {}
    config: dict = {}                 # Per-processor config; fragment state here
    metrics: dict = {}
    is_temporary: bool = False
    exception: dict = {}
    last_processor_name: Optional[str]

RecordModel

class RecordModel(BaseModel):
    model_config = ConfigDict(extra="allow")
    entity_id: str
    os_entity_uid: str          # Must equal entity_id
    entity_type: str
    os_concept: str             # Must equal entity_type
    os_workspace: Optional[str]
    entity_label: Optional[str]

Invariants enforced at construction:

• os_entity_uid == entity_id
• os_concept == entity_type

NifiEntity (Runtime Object)

Wraps NifiEntityModel with mutable state and helper methods:

• record — dict-like access to record fields
• annotations — mutable dict
• contents — Contents instance or None (see Contents section)
• children — list of NifiEntityProxy / NifiOTMRelationshipProxy
• add_annotations(json, merge_method, recurse) — merges into annotations
• add_child_fragment(child, ...) — creates fragment with OTM relationships
• add_child_entity(child, ...) — creates non-fragment child (clone)
• add_child_file(child, ...) — creates file child (attachable)
• is_fragmented() / is_root_fragment(fragmenter_keylist) — fragment state queries
• is_child_concept(concept_name) — ontology membership check
• update_last_timestamp() — sets os_last_updated_at
• prepare_contents() — materializes contents to memory or temp blob
• output_as_independent / output_as_child / drop_on_output — output routing flags

Reserved Record Fields

Excluded from upsert operations: os_entity_uid, entity_id, entity_type, os_concept, entity_label, os_created_at, os_created_by, os_last_updated_at, os_last_updated_by

Named Relationships (Constants)

FRAGMENT_ENTITY_NAME    = "os_fragment"
FILE_ENTITY_NAME        = "os_file"
TAG_ENTITY_NAME         = "os_tag"
RELATIONSHIP_ENTITY_NAME = "os_relationship"

FRAGMENT_RELATIONSHIP          = RelationshipName("is_fragment_of", "otm")
PREVIOUS_FRAGMENT_RELATIONSHIP = RelationshipName(...)
NEXT_FRAGMENT_RELATIONSHIP     = RelationshipName(...)
SOURCE_FRAGMENT_ENTITY_RELATIONSHIP = RelationshipName(...)
FILE_RELATIONSHIP              = RelationshipName("generator_of", "mtm")
TAG_RELATIONSHIP               = RelationshipName(...)

Contents System

Defined in streamlit_octostar_utils.api_crafter.contents:

Class	location	Locator Shape	Behavior
MemoryContents	memory	{location, data (base64), entity_type, filetype}	In-memory; ≤ 5 MB (MAX_IN_MEMORY_SIZE_BYTES)
WorkspaceAttachmentContents	workspace_attachment	{location, pointer: "workspace_id/entity_id"}	Read/write via Octostar v2 attachment API
TemporaryAttachmentContents	temporary_attachment	{location, filename}	Presigned temp blob
ReferenceContents	reference	{location, read_source, write_target}	Proxy — reads from one Contents, writes to another

Contents.from_locator(locator, client) dispatches on locator["location"].

---

NiFi Integration Layer

Request/Response Flow

NiFi InvokeHTTP
      │
      │  POST /api/nifi/{operation}?processor_suffix=0
      │  Body: JSON array of serialized NifiEntityModel objects
      │
      ▼
┌──────────────────────────────────────────────────────┐
│  NifiRoute.submit_task()                              │
│                                                       │
│  processor_name = "processor.{alias}.{op}.{suffix}"   │
│  Celery task dispatched with body bytes               │
│                                                       │
│  Returns: {"task_id": "..."}                          │
└────────────────────┬─────────────────────────────────┘
                     │
                     ▼
┌──────────────────────────────────────────────────────┐
│  Celery Worker — @NifiRoute.nifi_task wrapper         │
│                                                       │
│  1. json.loads(body_bytes)                            │
│  2. NifiContextManager(body) as nifi_context          │
│  3. nifi_context.receive_input(body, processor_name)  │
│     → NifiEntityBatch[] (grouped by config hash)      │
│  4. User function(task, nifi_context, batches, ...)   │
│  5. nifi_context.send_output(batches, processor_name) │
│     → Sync entities → Serialize → Return JSON         │
└──────────────────────────────────────────────────────┘

NifiContextManager

with NifiContextManager(json_data) as ctx:
    batches = ctx.receive_input(json_data, processor_name)
    # ... process entities ...
    ctx.send_output(batches, processor_name)

receive_input****:

1. Deserializes NifiEntity instances (with Contents objects)
2. Sorts by entity_timestamp
3. Deduplicates by entity_id
4. Groups into NifiEntityBatch objects by MD5 hash of the per-processor config slice
5. Each batch has: entities, config (processor config dict), config_key

send_output****:

1. Sets last_processor_name on all entities
2. Flattens entity trees (respecting output_as_independent / output_as_child / drop_on_output)
3. Deduplicates and sorts
4. Calls sync_entities() for non-lazy sync targets
5. Serializes and returns JSON

Sync Mechanism (request_entity_sync)

nifi_context.request_entity_sync(entity, {
    SyncFlag.UPSERT_ENTITY_ALL: True,           # Upsert all non-reserved record fields
    SyncFlag.UPSERT_ENTITY_SPECIFIC_FIELDS: {"fields": [...]},  # Or specific fields
    SyncFlag.WRITE_CONTENTS: True,              # Upload attachment bytes
})

SyncFlag enum:

Flag	Value	Payload	Effect
UPSERT_ENTITY_ALL	0	bool	Upserts all non-reserved record fields
UPSERT_ENTITY_SPECIFIC_FIELDS	1	{"fields": [...]}	Upserts named fields only
WRITE_CONTENTS	10	bool	Triggers attachment upload to workspace
FETCH_RELATIONSHIPS	20	list[str]	(Not implemented — raises NotImplementedError)

Sync sequence:

1. Reindex lock — calls update_processing_status with do_not_reindex_before/do_not_update_before timestamps to prevent stale overwrites
2. Write attachables — WorkspaceAttachmentContents.truncate/write/flush for entities with WRITE_CONTENTS
3. Upsert entities — upsert_entities.sync with field sets per entity
4. Clear sync_params on processed entities

---

Processing Lifecycle

Full Pipeline Sequence

1. EVENT / UPLOAD
   │  Entity created in Octostar → event stream or manual trigger
   │
2. FIREHOSE HYDRATE  (nifi-dashboard)
   │  Redis/event → fetch entity rows from Timbr → serialize as NifiEntityModel
   │  Dedup: checks processing_status.do_not_reindex_before in OpenSearch
   │  Entity type filtering via config
   │
3. PRE-PROCESS  (nifi-dashboard)
   │  Clears: annotations, relationships, processing_status in OpenSearch
   │  Deletes: existing os_fragment children by source_entity_uid
   │  Filters out: entities that are themselves fragments
   │  Sets: processing_status = RUNNING
   │
4. CONTENT SPLITTING  (data-cutter)
   │  Route depends on MIME:
   │  • Documents (PDF, DOCX, DOC, ODT) → split_document → document_pages + document_images
   │  • Presentations (PPTX, PPT, ODP) → split_document → document_pages (slides) + document_images
   │  • Spreadsheets (XLSX, CSV, ODS, …) → split_document → table_rows (+ table_row_groups)
   │  • Video → extract_keyframes + extract_audio → video_keyframes + video_audio
   │  • Audio → split_audio → audio_split segments
   │  • Archives / executables / other → no splitting (metadata-only)
   │
5. ENRICHMENT  (per media type)
   │  • Text extraction (document-extractor: extract-text, extract-ocr)
   │  • Metadata extraction (document-extractor: extract-metadata)
   │  • Language detection (document-extractor: detect-language)
   │  • NER (document-extractor: extract-ner-entities)
   │  • Summarization (document-extractor: summarize / summarize-deep)
   │  • Transcription (audio-transcription: transcribe-audio)
   │  • Translation (audio-transcription: translate-text)
   │  • Image captioning + tagging (image-augmentation: caption-and-tag-images)
   │  • Face detection + embeddings (object-detection: detect-faces)
   │  • Image embeddings (object-detection: detect-general-embeddings)
   │
6. MERGE FRAGMENTS  (data-cutter)
   │  Post-order tree walk: combine child annotations into parents
   │  Configurable merge operations per field
   │
7. MATERIALIZE FRAGMENTS  (data-cutter)
   │  Accumulate fragment annotations → write to parent in OpenSearch
   │  Uses Elasticsearch Painless bulk_update scripts
   │
8. RAG CHUNKS  (data-cutter)
   │  OSEntitySplitter: record fields + annotations → text chunks
   │  Writes rag:chunks and rag:fulltext
   │
9. RAG EMBED  (nifi-dashboard)
   │  Reads rag:chunks → encodes metadata.text → writes embedding vectors
   │
10. POST-PROCESS  (nifi-dashboard)
    │  Optional: fetch/flush relationships
    │
11. SEARCH SEND  (nifi-dashboard)
    │  Writes annotations + relationships to OpenSearch
    │  Sets processing_status = COMPLETED

Entity Routing Rules

NiFi routes entities to services based on record fields:

Condition	Routed To
os_has_attachment AND os_item_content_type starts with image/ AND is_child_concept("os_attachable")	image-augmentation, object-detection
os_has_attachment AND os_item_content_type starts with audio/ AND is_child_concept("os_attachable")	audio-transcription
os_has_attachment AND os_item_content_type matches document/spreadsheet MIMEs AND is_child_concept("os_attachable")	data-cutter (split), document-extractor
os_has_attachment AND os_item_content_type starts with video/ AND is_child_concept("os_attachable")	data-cutter (keyframes + audio extraction)
os_has_attachment AND no specific splitter/enrichment match (archives, executables, other)	document-extractor (metadata only); Tika text extraction attempted but may yield nothing
Has extract:txt annotation	document-extractor (NER, summarize), data-cutter (RAG)
Has rag:chunks annotation	nifi-dashboard (rag-embed)

---

Service Details

nifi-dashboard

Responsibilities: Pipeline bookends (pre/post), search indexing, RAG embedding, event hydration, monitoring. To be migrated to octostar-api.

Routes:

Route	Celery Queue	Entity Filter	Writes
pre-process	default	All (drops fragments)	Clears OpenSearch fields; processing_status=RUNNING
post-process	default	All	Optional relationship fetch/flush
search-send	default	All (per do_index config)	annotations + relationships → OpenSearch
rag-embed	embedding	Entities with rag:chunks	embedding on each chunk
firehose-hydrate	default	Event stream entities	Serialized NifiEntityModel output

Key implementation details:

• firehose-hydrate queries Timbr SELECT * FROM timbr.{entity_type} WHERE os_entity_uid IN (...) in batches of 5,000
• search-send uses opensearch_utils.convert_clientside to map annotations to the live index mapping
• pre-process runs clear_fragments which queries timbr.os_fragment by source_entity_uid and batch-deletes
• Prometheus metrics endpoint accepts NiFiMetricsPayload with counters like nifi_octostar_*
• Processing status tracked via ProcessingStatusCode enum with update_processing_status.sync

data-cutter

Responsibilities: Media splitting, fragment creation/merge/materialization, RAG chunk creation.

Routes:

Route	Queue	Input	Output
split-document	libreoffice	Document attachment	Fragment tree (document_pages + document_images for page/slide types; table_rows for spreadsheets)
split-audio	default	Audio attachment	Audio segment fragments
extract-keyframes-from-video	default	Video attachment	Keyframe image fragments
extract-audio-from-video	default	Video attachment	Single audio fragment
split-primitive	default	Entity field value	Clone entities with chunked field
merge-fragments	default	Fragment tree	Merged annotations on parents
merge-primitive	default	Multiple field paths	Single output field
create-rag-chunks	default	Entity record + annotations	rag:chunks, rag:fulltext
materialize-fragments	default	Fragment entities	Parent OpenSearch doc updated

Key implementation details:

• Document splitting uses format-specific plugins under plugins/splitters/document/ registered by MIME; each plugin defines its own get_collection_type() — page/slide-based splitters produce document_pages (+ document_page_groups), while row-based splitters (XLSX, CSV, ODS) produce table_rows (+ table_row_groups)
• DocumentSplitter.split() produces a tree; _create_fragment_entities walks it and calls add_child_fragment
• Image extraction: extract_images() normalizes via validate_and_normalize_image; images attached to the leaf fragment whose range_value contains the source_index
• FFmpeg used for: audio extraction (get_audio), keyframe extraction (fps + saliency/hash filter), audio splitting (segment extraction)
• NifiFragmenter.as_nifi_fragments creates stable UUIDs via UUID5(FRAGMENTS_NAMESPACE, MD5(payload))
• merge_fragments uses NifiFragmenter.build_fragment_tree_from_children_entities → post-order reduce
• materialize_fragments accumulates via materializer plugins (plugins/materializers/) → Painless bulk scripts
• Merge plugins: FIRST, LAST, strconcat, lstconcat under plugins/mergers/

document-extractor

Responsibilities: Text extraction, file metadata, OCR, NER, language detection, summarization.

Routes:

Route	Queue	Writes Annotation
extract-text	tika	extract:txt
extract-metadata	exiftool	extract:metadata
extract-ocr	ocr	extract:txt (image entities only)
detect-language	default	extract:txtLang
extract-ner-entities	ner	ner:entities
extract-spreadsheet-entities	llm	ner:entities
summarize	llm	summary:txt
summarize-deep	llm_deep	summary:txt

Key implementation details:

• extract-text uses Apache Tika JVM per worker; scanned PDF detection (low text + high image coverage) triggers PyMuPDF + RapidOCR per-page fallback
• Context stitching: if metadata contains os_ContextBefore/os_ContextAfter, wraps extracted body text
• extract-ocr only processes image/* MIME entities; backend: EasyOCR or Tesseract (configurable via ocr_method)
• NER cascade: SpaCy/Flair/regex → optional LLM postprocessing (llm_ner.py)
• Summarization is section-aware: uses section:type annotation to select prompts; hierarchical for fragment trees
• Summarization MIME allowlist: by default only application/pdf, DOCX, DOC, and ODT are eligible (configurable via allowed_mime_types); presentations, spreadsheets, CSV, and plain text are skipped
• Summarization builds fragment tree via _build_tree_from_fragments (Timbr os_fragment rows)
• detect-language only runs on entities with extract:txt but missing extract:txtLang

audio-transcription

Responsibilities: Speech-to-text, translation.

Routes:

Route	Writes Annotations
transcribe-audio	extract:txt, extract:txtLang
translate-text	translate:txt:{to_language}, translate:src

Key implementation details:

• Whisper: HuggingFace automatic-speech-recognition pipeline with chunk_length_s=30, stride_length_s=(4,2), return_timestamps=True
• AssemblyAI: speech_model=best, language_detection=True; segments from get_sentences()
• Timestamps are [start_seconds, end_seconds]; formatted as HH:MM:SS.mmm
• Caveat: AssemblyAI raw timestamp values are milliseconds (not seconds) — units differ between backends unless normalized downstream
• Language detection via streamlit_octostar_utils.nlp.language.detect_language on joined segment texts
• Translation reads extract:txt + batch.config["to_language"] (default: "english")
• NiFi task filters: audio/* MIME + os_has_attachment + is_child_concept("os_attachable")

image-augmentation

Responsibilities: Vision-LLM captioning, tagging, semantic classification.

Routes:

Route	Writes Annotations
caption-and-tag-images	extract:txt, extract:txtLang, image:annotations, image:annotations-key, image:annotations-sensitive, image:annotations-setting

Key implementation details:

• Processing pipeline: decode (Pillow/Wand) → resize (512px max, LANCZOS) → JPEG encode → base64 data URI → OpenAI vision API → parse JSON → normalize → semantic classification → build ImageAnalysis
• ImageAnalysis Pydantic model (see Developer Guide for full enum values)
• Labels extracted by extract_labels(): walks full to_dict() tree, True booleans → label key, string/enum values → label, skips keys containing "description"
• Semantic classification: sliding-window text chunks vs precomputed embeddings for object/sensitive concept lists
• image:annotations merge: deduplication by label key (and optional bbox)
• Full ImageAnalysis dict is not stored as a single annotation — only flattened label lists

object-detection

Responsibilities: Face detection with embeddings, whole-image embeddings.

Routes:

Route	Writes Annotations	Creates Fragments
detect-faces	image:annotations	face_crops (conditional)
detect-general-embeddings	image:annotations	—

Key implementation details:

• Face detection: MTCNN (keep_all=True, min_face_size=40, thresholds=[0.7, 0.8, 0.9]) → InceptionResnetV1 (vggface2) L2-normalized embeddings
• Image embedding: CLIP ViT-B/32 L2-normalized features
• Face crop branching logic:
  • No faces → no annotations added
  • 1 face covering ≥ 70% image area → annotations on parent, no fragments
  • Otherwise → child PNG fragments per face, each with single detection annotation
• Embedding format: {"#type": "VECTOR", "data": [...], "metadata": {"model_name": ..., "model_version": ..., "dim": ...}}
• Image loading: Pillow primary, Wand/ImageMagick fallback for HEIC etc.
• Device selection: CELERY_WORKER_NAME suffix or PID % num_devices for GPU round-robin
• NiFi task filters: image/* MIME + os_has_attachment + is_child_concept("os_attachable") + truthy os_entity_uid

---

Fragment Internals

Fragment Creation

NifiEntity.add_child_fragment(child, ...) sets on the child record:

• All parent record keys starting with fragment are copied
• os_parent_uid → parent's os_entity_uid
• source_entity_uid → root entity's UID
• previous_entity_uid / next_entity_uid → sibling linkage
• OTM relationships: is_fragment_of, previous/next fragment, source fragment entity

Fragment UUID Generation

NifiFragmenter.create_fragment_uuid(os_entity_uid, stable_fragment_identifier, processor_name, os_workspace) — deterministic UUID5 via _FRAGMENTS_NAMESPACE + MD5 of concatenated parameters. This ensures idempotent re-processing.

Fragment Tracking State

Lives at entity.request.config.fragment:

{
    "fragments_stack": ["processor.data-cutter.split-document.0"],
    "processor.data-cutter.split-document.0": {
        "identifier": "shared-uuid-for-siblings",
        "count": 12,
        "index": 3,
        "root_uid": "parent-os-entity-uid",
        "root_type": "os_document",
        "merge_params": {}    # Populated by push_defragment_strategy
    }
}

fragments_stack: LIFO — last fragmenter key is the "current" fragmentation level. as_nifi_fragments prepends; merge_fragments pops.

Required fields per fragment: index, count, identifier.

Fragment Grouping

NifiFragmenter.identify_fragment_groups(nifi_batches) → dict[fragmenter_keylist, list[NifiEntity]] using top of each entity's fragments_stack.

Fragment Tree Operations

• build_fragment_tree_from_children_entities(root, keylist) — recursive tree of {entity, index, merge_params, children}
• iterate_fragments_tree(tree, order="post"|"pre") — generator
• reduce_fragments_tree(tree, leaf_fn, parent_fn) — bottom-up reduce
• extract_tree_entities(tree) — flat list

Output Routing

Flag	Default	Effect
output_as_independent	False	Entity appears as a top-level FlowFile in NiFi output
output_as_child	True	Entity appears nested under its parent
drop_on_output	False	Entity is excluded from output entirely

Video keyframes and face crops use output_as_independent=True, output_as_child=False so they enter the NiFi flow as standalone entities that get routed by their own MIME type.

---

Merge and Materialization Internals

Merge Fragments

Route: data-cutter /merge-fragments

Algorithm:

1. Identify fragment groups by fragments_stack top
2. Build tree from build_fragment_tree_from_children_entities
3. Post-order traversal: at each parent node, merge children's field values using configured operation
4. Pop fragmenter key from fragments_stack

Merge tree node shape:

class NestedMetadata:
    data: dict              # Entity field data at field_path
    index: int              # Fragment index
    children: List[NestedMetadata]
    merge_params: dict      # Per-node operation overrides

Batch config:

Key	Type	Default	Description
field_path	str	"annotations"	Entity base to merge (record, annotations, relationships, content)
merge_depth	int	-1	Max tree depth to merge (-1 = unlimited)
keep_root_only	bool	False	Discard non-root entities after merge
merge_params.op	str	—	Merge operation name
merge_params.recurse	bool	—	Recurse into nested dicts
merge_params.fields	list[str]	—	Restrict merge to these keys

Available merge operation plugins (loaded from plugins/mergers/):

Name	Behavior
FIRST	Keep first child's value (by fragment_index)
LAST	Keep last child's value
STRCONCAT	Concatenate strings with configurable separator
LSTCONCAT	Concatenate lists

Materialize Fragments

Route: data-cutter /materialize-fragments

Algorithm:

1. Group fragment entities by fragments_stack
2. For each group, load materializer plugins (or all if not configured)
3. Walk fragments: extract_value per materializer, accumulate results
4. Build Painless script via get_painless_script
5. bulk_update.sync on parent entity in OpenSearch

Materializer plugin interface (utils/base_materializer.py):

class BaseMaterializer:
    name: str                    # Plugin identifier
    source_path: str             # Annotation key to read from fragments
    def extract_value(entity)    # Pull value from a fragment
    def accumulate(values)       # Combine extracted values
    def get_painless_script()    # Elasticsearch Painless update script

Result location in OpenSearch: annotations.materialized.<source_path>

---

Annotation Merge Mechanics

add_annotations Method

entity.add_annotations(
    json: dict,                              # Annotations to add
    merge_method: Callable[[Any, Any], Any], # (old_value, new_value) -> merged
    recurse: Union[bool, int] = False        # Recurse into nested dicts
)

Delegates to recursive_update_dict(self.annotations, json, merge_method, recurse).

recursive_update_dict Algorithm

For each key k in the incoming dict d2:

1. If d2[k] is a dict AND d1[k] is a dict AND recurse is truthy → recurse (if recurse is int, decrement per level)
2. If d2[k] is a dict but not recursing → d1[k] = merge_method(d1[k], d2[k])
3. Scalar / non-dict: if key exists → d1[k] = merge_method(d1[k], d2[k]); else assign d2[k]

Common Merge Methods in the Codebase

These are the add_annotations merge methods used within services during processing. This is distinct from the query-time merge strategies in get_entity_annotations (documented in the Developer Guide) which operates on fragment annotations at read time.

Pattern	Lambda	Used For
Replace	lambda _, v2: v2	extract:txtLang, summary:txt, translate:*
Append	lambda v1, v2: v1 + "\n" + v2 if v1 else v2	extract:txt
Dedup list	Keeps unique by label + optional bbox key	image:annotations*
Recursive dict	recurse=True or recurse=1	ner:entities, extract:metadata

---

RAG Pipeline Internals

Chunk Creation (data-cutter: create-rag-chunks)

Uses octostar_ai.OSEntitySplitter:

1. Builds text from entity record fields (record_fields_to_parse, default: description, contents) and annotations
2. Splits into chunks (splitter implementation in octostar_ai)
3. Each chunk: {"metadata": {**chunk.metadata, "text": chunk.page_content}}
4. Writes rag:chunks (max rag_max_chunks, default 100) and rag:fulltext (concatenation)

Embedding (nifi-dashboard: rag-embed)

1. Reads entity.annotations["rag:chunks"]
2. Extracts metadata.text from each chunk
3. Batch-encodes with configured embedding model (SentenceTransformer)
4. Writes embedding dict (#type: VECTOR) into each chunk's metadata

Embedding Model Config

From nifi-dashboard/config.yaml:

embedder_model:
  type: sentence_transformer
  model_name: "sentence-transformers/all-MiniLM-L6-v2"
  batch_size: 32
  device: "cpu"   # or "cuda", or list for round-robin

---

Configuration Reference

Per-Service config.yaml Structure

All services share a common pattern:

logging:
  level: INFO
  format: "%(asctime)s ..."
  log_requests: true

celery:
  preload_time_limit: 300
  memory_monitor:
    sample_interval: 5
    gc_collect_on_end: true
    malloc_trim_on_end: true

queues:
  default:
    n_workers: 2          # 0 = queue disabled, routes not registered
    max_tasks_per_child: 100
    warn_memory_per_child: 1000  # MB
  # ... additional queues per service

tasks:
  task_name:
    time_limit: 300
    soft_time_limit: 290
    rate_limit: null
    batch_size: 4
    parallelism_max: 3
    parallelism_throttle: null

data-cutter Specific Config

tasks:
  split_document:
    page_splitting:
      enabled: true
      max_pages: 20       # Pages per fragment group
      context_chars: 200  # Overlap chars between groups
    table_splitting:
      enabled: false
      max_rows: 100
    extract_images: true
    allowed_mime_types: [...]  # MIME whitelist

  create_rag_chunks:
    rag_max_chunks: 100
    record_fields_to_parse: ["description", "contents"]

audio_extraction:
  quality: ...
  sample_rate: ...
  channels: ...
  output_format: ...

keyframe_extraction:
  fps: ...
  saliency_percentage: ...
  max_keyframes: ...
  similarity_threshold: ...
  output_format: png|jpg

audio_splitter:
  segment_duration: 30
  max_size_mb: ...
  overlap_seconds: ...

NiFi Batch Config Overrides

Batch-level configuration is passed via batch.config and overrides config.yaml defaults:

Service	Config Key	Type	Effect
data-cutter	write_fragment_contents	bool	Whether to persist fragment binary data
data-cutter	fragment_parent_source	str	Resolve parent entity for fragment tree
data-cutter	page_splitting.max_pages	int	Override pages per group
data-cutter	extract_images	bool	Enable/disable image extraction from documents
document-extractor	ocr_method	str	"easyocr" / "tesseract" / "none"
document-extractor	with_scores	bool	Include NER confidence scores
document-extractor	with_context	bool	Include NER context snippets
document-extractor	record_fields_to_parse	list	Fields for NER input
document-extractor	max_llm_calls	int	Summarization LLM call budget
document-extractor	max_input_chars	int	Summarization input cap
document-extractor	allowed_mime_types	list	Override summarization MIME filter
audio-transcription	to_language	str	Translation target (default: "english")
object-detection	write_fragment_contents	bool	Whether to persist face crop images
object-detection	fragment_parent_source	str	Parent entity resolution