From ab98fc53b70cc11165d3061b0a0027b66c706202 Mon Sep 17 00:00:00 2001 From: Athul Date: Thu, 11 Jun 2026 13:38:44 +0530 Subject: [PATCH 1/3] UN-2771 Include text extraction time in API deployment metrics The structure tool timed indexing but not the text extraction (LLMWhisperer/X2Text) call, so API responses with include_metrics=True reported indexing time only. Time dynamic_extraction the same way and merge it into the result metrics as extraction.time_taken(s). Bump structure tool to 0.0.102. Co-Authored-By: Claude Fable 5 --- tools/structure/src/config/properties.json | 2 +- tools/structure/src/constants.py | 1 + tools/structure/src/main.py | 14 +++++++++++++- 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/tools/structure/src/config/properties.json b/tools/structure/src/config/properties.json index c8697e7307..a9d1a6029d 100644 --- a/tools/structure/src/config/properties.json +++ b/tools/structure/src/config/properties.json @@ -2,7 +2,7 @@ "schemaVersion": "0.0.1", "displayName": "Structure Tool", "functionName": "structure_tool", - "toolVersion": "0.0.101", + "toolVersion": "0.0.102", "description": "This is a template tool which can answer set of input prompts designed in the Prompt Studio", "input": { "description": "File that needs to be indexed and parsed for answers" diff --git a/tools/structure/src/constants.py b/tools/structure/src/constants.py index 8da6a5701a..294fb3015b 100644 --- a/tools/structure/src/constants.py +++ b/tools/structure/src/constants.py @@ -77,6 +77,7 @@ class SettingsKeys: TOOL = "tool" METRICS = "metrics" INDEXING = "indexing" + EXTRACTION = "extraction" EXECUTION_ID = "execution_id" IS_DIRECTORY_MODE = "is_directory_mode" LLM_PROFILE_ID = "llm_profile_id" diff --git a/tools/structure/src/main.py b/tools/structure/src/main.py index f68143a6c8..fa8ec39530 100644 --- a/tools/structure/src/main.py +++ b/tools/structure/src/main.py @@ -318,6 +318,7 @@ def run( ) extracted_text = "" + extraction_metrics = {} usage_kwargs: dict[Any, Any] = dict() if skip_extraction_and_indexing: self.stream_log( @@ -328,6 +329,7 @@ def run( usage_kwargs[UsageKwargs.RUN_ID] = self.file_execution_id usage_kwargs[UsageKwargs.FILE_NAME] = self.source_file_name usage_kwargs[UsageKwargs.EXECUTION_ID] = self.execution_id + extraction_start_time = datetime.datetime.now() extracted_text = STHelper.dynamic_extraction( file_path=input_file, enable_highlight=is_highlight_enabled, @@ -338,6 +340,13 @@ def run( tool=self, execution_run_data_folder=str(execution_run_data_folder), ) + extraction_metrics = { + SettingsKeys.EXTRACTION: { + "time_taken(s)": STHelper.elapsed_time( + start_time=extraction_start_time + ) + } + } index_metrics = {} if is_summarization_enabled: @@ -458,7 +467,10 @@ def run( "No text is extracted from the document to add to the metadata" ) if merged_metrics := self._merge_metrics( - structured_output.get(SettingsKeys.METRICS, {}), index_metrics + self._merge_metrics( + structured_output.get(SettingsKeys.METRICS, {}), index_metrics + ), + extraction_metrics, ): structured_output[SettingsKeys.METRICS] = merged_metrics # Update GUI From ec778bc95c32e6dbff2ec9581a76600414e7ec42 Mon Sep 17 00:00:00 2001 From: Athul Date: Thu, 11 Jun 2026 16:22:22 +0530 Subject: [PATCH 2/3] UN-2771 Rework: capture extraction time in the worker pipeline instead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review, the structure tool's Docker path is deprecated — the live flow is the celery-based LegacyExecutor structure pipeline. Time the extract step there and merge {'extraction': {'time_taken(s)': ...}} into the result metrics alongside the existing per-output indexing timing. Structure tool changes reverted (no tool version bump needed). Co-Authored-By: Claude Fable 5 --- tools/structure/src/config/properties.json | 2 +- tools/structure/src/constants.py | 1 - tools/structure/src/main.py | 14 +------------- workers/executor/executors/legacy_executor.py | 14 ++++++++++++-- 4 files changed, 14 insertions(+), 17 deletions(-) diff --git a/tools/structure/src/config/properties.json b/tools/structure/src/config/properties.json index a9d1a6029d..c8697e7307 100644 --- a/tools/structure/src/config/properties.json +++ b/tools/structure/src/config/properties.json @@ -2,7 +2,7 @@ "schemaVersion": "0.0.1", "displayName": "Structure Tool", "functionName": "structure_tool", - "toolVersion": "0.0.102", + "toolVersion": "0.0.101", "description": "This is a template tool which can answer set of input prompts designed in the Prompt Studio", "input": { "description": "File that needs to be indexed and parsed for answers" diff --git a/tools/structure/src/constants.py b/tools/structure/src/constants.py index 294fb3015b..8da6a5701a 100644 --- a/tools/structure/src/constants.py +++ b/tools/structure/src/constants.py @@ -77,7 +77,6 @@ class SettingsKeys: TOOL = "tool" METRICS = "metrics" INDEXING = "indexing" - EXTRACTION = "extraction" EXECUTION_ID = "execution_id" IS_DIRECTORY_MODE = "is_directory_mode" LLM_PROFILE_ID = "llm_profile_id" diff --git a/tools/structure/src/main.py b/tools/structure/src/main.py index fa8ec39530..f68143a6c8 100644 --- a/tools/structure/src/main.py +++ b/tools/structure/src/main.py @@ -318,7 +318,6 @@ def run( ) extracted_text = "" - extraction_metrics = {} usage_kwargs: dict[Any, Any] = dict() if skip_extraction_and_indexing: self.stream_log( @@ -329,7 +328,6 @@ def run( usage_kwargs[UsageKwargs.RUN_ID] = self.file_execution_id usage_kwargs[UsageKwargs.FILE_NAME] = self.source_file_name usage_kwargs[UsageKwargs.EXECUTION_ID] = self.execution_id - extraction_start_time = datetime.datetime.now() extracted_text = STHelper.dynamic_extraction( file_path=input_file, enable_highlight=is_highlight_enabled, @@ -340,13 +338,6 @@ def run( tool=self, execution_run_data_folder=str(execution_run_data_folder), ) - extraction_metrics = { - SettingsKeys.EXTRACTION: { - "time_taken(s)": STHelper.elapsed_time( - start_time=extraction_start_time - ) - } - } index_metrics = {} if is_summarization_enabled: @@ -467,10 +458,7 @@ def run( "No text is extracted from the document to add to the metadata" ) if merged_metrics := self._merge_metrics( - self._merge_metrics( - structured_output.get(SettingsKeys.METRICS, {}), index_metrics - ), - extraction_metrics, + structured_output.get(SettingsKeys.METRICS, {}), index_metrics ): structured_output[SettingsKeys.METRICS] = merged_metrics # Update GUI diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index eae4d05b2f..db89d4b98d 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -619,10 +619,12 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: ) step = 1 + extraction_metrics: dict = {} try: # ---- Step 1: Extract ---- if not skip_extraction: step += 1 + extraction_start = time.monotonic() extract_ctx = ExecutionContext( executor_name=context.executor_name, operation=Operation.EXTRACT.value, @@ -640,6 +642,9 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: return _failure(extract_result) _absorb(extract_result) extracted_text = extract_result.data.get(IKeys.EXTRACTED_TEXT, "") + extraction_metrics = { + "extraction": {"time_taken(s)": time.monotonic() - extraction_start} + } # ---- Step 2: Summarize (if enabled) ---- if is_summarization: @@ -700,6 +705,7 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: source_file_name=source_file_name, extracted_text=extracted_text, index_metrics=index_metrics, + extraction_metrics=extraction_metrics, ) output_map = structured_output.get(PSKeys.OUTPUT, {}) or {} @@ -787,6 +793,7 @@ def _finalize_pipeline_result( source_file_name: str, extracted_text: str, index_metrics: dict, + extraction_metrics: dict | None = None, ) -> None: """Populate metadata/metrics in structured_output after pipeline completion.""" if "metadata" not in structured_output: @@ -794,10 +801,13 @@ def _finalize_pipeline_result( structured_output["metadata"]["file_name"] = source_file_name if extracted_text: structured_output["metadata"]["extracted_text"] = extracted_text - if index_metrics: + new_metrics = self._merge_pipeline_metrics( + index_metrics or {}, extraction_metrics or {} + ) + if new_metrics: existing_metrics = structured_output.get("metrics", {}) structured_output["metrics"] = self._merge_pipeline_metrics( - existing_metrics, index_metrics + existing_metrics, new_metrics ) def _run_pipeline_summarize( From 81d01ed182eae88499db5517ee0ee8ab376eb4af Mon Sep 17 00:00:00 2001 From: Athul Date: Tue, 14 Jul 2026 10:19:04 +0530 Subject: [PATCH 3/3] UN-2771 Address review: namespace the metric, share the duration key, align clocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Nest extraction timing under a reserved "_pipeline" namespace (PSKeys.PIPELINE) so it cannot collide with a user-defined output/prompt named "extraction" at the top level of the metrics dict. - Use PSKeys.EXTRACTION instead of the hardcoded literal, and capture the duration in a named local for parity with the indexing path. - Promote the duplicated "time_taken(s)" literal to PSKeys.TIME_TAKEN. - Align the indexing path onto time.monotonic(): it measured a duration with wall-clock datetime.now(), which is wrong across NTP/system-clock adjustments and left the two producers of time_taken(s) on different clocks. Drops the now-unused local datetime import. - Document the extraction_metrics shape on _finalize_pipeline_result and type it dict[str, dict] | None. Tests: 5 new cases in test_phase5d.py — metric recorded, name-collision guard, index+extraction merge, skip-extraction, and extract-failure (no timing recorded). Worker suite: 723 passed, same 6 pre-existing failures as main. Co-Authored-By: Claude Opus 4.8 (1M context) --- workers/executor/executors/constants.py | 5 + workers/executor/executors/legacy_executor.py | 28 ++-- workers/tests/test_phase5d.py | 120 ++++++++++++++++++ 3 files changed, 145 insertions(+), 8 deletions(-) diff --git a/workers/executor/executors/constants.py b/workers/executor/executors/constants.py index 9eddab8423..868a47c773 100644 --- a/workers/executor/executors/constants.py +++ b/workers/executor/executors/constants.py @@ -51,6 +51,11 @@ class PromptServiceConstants: CHALLENGE = "challenge" ENABLE_CHALLENGE = "enable_challenge" EXTRACTION = "extraction" + # Reserved namespace for pipeline-level metrics so they cannot collide with + # user-defined output/prompt names at the top level of the metrics dict. + PIPELINE = "_pipeline" + # Wire-format key every metrics producer reports its duration under. + TIME_TAKEN = "time_taken(s)" SUMMARIZE = "summarize" SINGLE_PASS_EXTRACTION = "single-pass-extraction" SIMPLE_PROMPT_STUDIO = "simple-prompt-studio" diff --git a/workers/executor/executors/legacy_executor.py b/workers/executor/executors/legacy_executor.py index ea5404e55e..07b7d8c4f1 100644 --- a/workers/executor/executors/legacy_executor.py +++ b/workers/executor/executors/legacy_executor.py @@ -642,8 +642,14 @@ def _failure(child_result: ExecutionResult) -> ExecutionResult: return _failure(extract_result) _absorb(extract_result) extracted_text = extract_result.data.get(IKeys.EXTRACTED_TEXT, "") + extraction_time = time.monotonic() - extraction_start + # Nest under a reserved "_pipeline" namespace so the metric never + # collides with a user-defined output named "extraction" during + # the top-level metrics merge (see _merge_pipeline_metrics). extraction_metrics = { - "extraction": {"time_taken(s)": time.monotonic() - extraction_start} + PSKeys.PIPELINE: { + PSKeys.EXTRACTION: {PSKeys.TIME_TAKEN: extraction_time} + } } # ---- Step 2: Summarize (if enabled) ---- @@ -806,9 +812,17 @@ def _finalize_pipeline_result( source_file_name: str, extracted_text: str, index_metrics: dict, - extraction_metrics: dict | None = None, + extraction_metrics: dict[str, dict] | None = None, ) -> None: - """Populate metadata/metrics in structured_output after pipeline completion.""" + """Populate metadata/metrics in structured_output after pipeline completion. + + Args: + extraction_metrics: Pipeline-level extraction timing, shaped as + ``{"_pipeline": {"extraction": {"time_taken(s)": float}}}``. + Nested under the reserved ``_pipeline`` namespace to avoid + colliding with user-defined output names during the merge. + ``None``/empty when extraction is skipped. + """ if "metadata" not in structured_output: structured_output["metadata"] = {} structured_output["metadata"]["file_name"] = source_file_name @@ -965,8 +979,6 @@ def _index_pipeline_output( index_records: list[dict], ) -> None: """Index a single structure-pipeline output entry in-place.""" - import datetime - chunk_size = output.get("chunk-size", 0) if chunk_size == 0: return @@ -987,7 +999,7 @@ def _index_pipeline_output( return seen_params.add(param_key) - indexing_start = datetime.datetime.now() + indexing_start = time.monotonic() logger.info( "Pipeline indexing: chunk_size=%s chunk_overlap=%s vector_db=%s", chunk_size, @@ -1039,9 +1051,9 @@ def _index_pipeline_output( if child_records: index_records.extend(child_records) - elapsed = (datetime.datetime.now() - indexing_start).total_seconds() + elapsed = time.monotonic() - indexing_start output_name = output.get("name", "") - index_metrics[output_name] = {"indexing": {"time_taken(s)": elapsed}} + index_metrics[output_name] = {"indexing": {PSKeys.TIME_TAKEN: elapsed}} @staticmethod def _merge_pipeline_metrics(metrics1: dict, metrics2: dict) -> dict: diff --git a/workers/tests/test_phase5d.py b/workers/tests/test_phase5d.py index 8b3252c9e6..72f1715242 100644 --- a/workers/tests/test_phase5d.py +++ b/workers/tests/test_phase5d.py @@ -304,6 +304,126 @@ def test_index_metrics_merged(self, executor): assert "indexing" in metrics["field_a"] +# --------------------------------------------------------------------------- +# Tests — Extraction timing metric (UN-2771) +# --------------------------------------------------------------------------- + + +class TestExtractionMetrics: + """Extraction duration is reported under the reserved _pipeline namespace. + + The metric is nested as ``metrics["_pipeline"]["extraction"]`` rather than a + bare top-level ``"extraction"`` key, because sibling keys in this dict are + user-defined output/prompt names — a prompt literally named "extraction" + would otherwise collide with it. + """ + + def _executor_with_extraction(self, executor, answer_metrics=None): + executor._handle_extract = MagicMock( + return_value=ExecutionResult( + success=True, data={"extracted_text": "text"} + ) + ) + executor._handle_index = MagicMock( + return_value=ExecutionResult(success=True, data={"doc_id": "d1"}) + ) + executor._handle_answer_prompt = MagicMock( + return_value=ExecutionResult( + success=True, + data={"output": {}, "metrics": answer_metrics or {}}, + ) + ) + return executor + + def _run(self, executor): + ctx = _make_pipeline_context({ + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": _base_pipeline_options(), + }) + return executor._handle_structure_pipeline(ctx) + + def test_extraction_metrics_recorded(self, executor): + """A normal run reports a non-negative float extraction duration.""" + self._executor_with_extraction(executor) + result = self._run(executor) + + assert result.success + time_taken = result.data["metrics"]["_pipeline"]["extraction"][ + "time_taken(s)" + ] + assert isinstance(time_taken, float) + assert time_taken >= 0 + + def test_extraction_metric_does_not_collide_with_output_name(self, executor): + """A prompt named "extraction" coexists with the pipeline metric.""" + self._executor_with_extraction( + executor, + answer_metrics={"extraction": {"llm": {"time_taken(s)": 2.0}}}, + ) + result = self._run(executor) + + metrics = result.data["metrics"] + # The user's prompt metrics are untouched... + assert metrics["extraction"] == {"llm": {"time_taken(s)": 2.0}} + # ...and the pipeline timing lives in its own namespace. + assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + + def test_index_and_extraction_metrics_merged(self, executor): + """Per-output indexing and pipeline extraction metrics coexist.""" + self._executor_with_extraction( + executor, + answer_metrics={"field_a": {"llm": {"time_taken(s)": 2.0}}}, + ) + executor._run_pipeline_index = MagicMock( + return_value=({"field_a": {"indexing": {"time_taken(s)": 0.5}}}, []) + ) + result = self._run(executor) + + metrics = result.data["metrics"] + assert "llm" in metrics["field_a"] + assert "indexing" in metrics["field_a"] + assert "time_taken(s)" in metrics["_pipeline"]["extraction"] + + def test_skip_extraction_records_no_extraction_metric(self, executor): + """Smart-table path skips extract, so no extraction metric is added.""" + executor._handle_extract = MagicMock() + executor._handle_index = MagicMock() + executor._handle_answer_prompt = MagicMock( + return_value=ExecutionResult(success=True, data={"output": {}}) + ) + opts = _base_pipeline_options() + opts["skip_extraction_and_indexing"] = True + + ctx = _make_pipeline_context({ + "extract_params": _base_extract_params(), + "index_template": _base_index_template(), + "answer_params": _base_answer_params(), + "pipeline_options": opts, + }) + result = executor._handle_structure_pipeline(ctx) + + assert result.success + executor._handle_extract.assert_not_called() + assert "_pipeline" not in result.data.get("metrics", {}) + + def test_extract_failure_records_no_extraction_metric(self, executor): + """Timing is taken after the failure early-return, so a failed extract + must not report a duration.""" + executor._handle_extract = MagicMock( + return_value=ExecutionResult.failure(error="x2text error") + ) + executor._handle_index = MagicMock() + executor._handle_answer_prompt = MagicMock() + executor._finalize_pipeline_result = MagicMock() + + result = self._run(executor) + + assert not result.success + executor._finalize_pipeline_result.assert_not_called() + + # --------------------------------------------------------------------------- # Tests — Embedding usage record propagation # ---------------------------------------------------------------------------