From db1a757f51d068d6bac54039e41c69550c45599f Mon Sep 17 00:00:00 2001 From: torst Date: Mon, 20 Jul 2026 10:50:58 -0400 Subject: [PATCH] Add shared prompt editor workflows Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- batch_processing/batch_method.py | 6 + file_handling/data_import.py | 92 ++- live_processing/keyword_extraction_live.py | 73 +- live_processing/multi_label_live.py | 87 +- live_processing/single_label_live.py | 86 +- prompt_editor/__init__.py | 2 + prompt_editor/config.py | 102 +++ prompt_editor/engine.py | 638 ++++++++++++++ prompt_editor/presets.py | 168 ++++ prompt_editor/storage.py | 114 +++ prompt_editor/ui.py | 913 +++++++++++++++++++++ tests/test_prompt_editor.py | 143 ++++ ui/main_window.py | 78 +- 13 files changed, 2203 insertions(+), 299 deletions(-) create mode 100644 prompt_editor/__init__.py create mode 100644 prompt_editor/config.py create mode 100644 prompt_editor/engine.py create mode 100644 prompt_editor/presets.py create mode 100644 prompt_editor/storage.py create mode 100644 prompt_editor/ui.py create mode 100644 tests/test_prompt_editor.py diff --git a/batch_processing/batch_method.py b/batch_processing/batch_method.py index 2380c09..6753af0 100644 --- a/batch_processing/batch_method.py +++ b/batch_processing/batch_method.py @@ -10,6 +10,7 @@ from batch_processing.batch_error_handling import handle_batch_fail from file_handling.data_conversion import to_long_df, save_as_csv, join_datasets from file_handling.data_import import import_data +from prompt_editor.engine import load_prompt_editor_batch_results from settings import config, secrets_store from settings.user_config import get_setting from openai import OpenAI @@ -179,6 +180,11 @@ def get_batch_results(batch_id: str) -> None: Results are automatically saved as a DataFrame with columns for quote, label, and confidence from the classification response. """ + prompt_editor_df = load_prompt_editor_batch_results(batch_id) + if prompt_editor_df is not None: + save_as_csv(prompt_editor_df) + return + client = get_client() status = get_batch_status(batch_id) diff --git a/file_handling/data_import.py b/file_handling/data_import.py index 3651303..42c1a79 100644 --- a/file_handling/data_import.py +++ b/file_handling/data_import.py @@ -16,6 +16,7 @@ from __future__ import annotations import csv +from dataclasses import dataclass from pathlib import Path from typing import Iterable, Optional, Sequence, Tuple, List @@ -200,7 +201,7 @@ def rebuild(self, column_labels: list[str]): # ----------------------------- # Public API: Import Wizard # ----------------------------- -def import_data( +def import_tabular_data( parent: Optional[tk.Misc] = None, title: str = "Import Data", filetypes: Sequence[tuple[str, str]] = ( @@ -210,7 +211,7 @@ def import_data( ("TSV files", "*.tsv"), ("Text files", "*.txt"), ), -) -> Optional[tuple[list[str], str]]: +) -> Optional[ImportedDataset]: """ Open an "Import Wizard" dialog: select a file, preview it, choose 'has headers', pick exactly one column via radio buttons, choose a dataset name, and return that @@ -447,11 +448,10 @@ def _sel_hdr(): _initial_blank_preview() # start 5x5 # -------------- Import handler -------------- - result_values: Optional[list[str]] = None - result_dataset_name: Optional[str] = None + result_dataset: Optional[ImportedDataset] = None def _do_import(): - nonlocal result_values, result_dataset_name + nonlocal result_dataset path = file_var.get().strip() if not path: messagebox.showerror("Error", "No file selected.", parent=dlg) @@ -464,16 +464,15 @@ def _do_import(): messagebox.showerror("Load error", str(e), parent=dlg) return - body = rows[1:] if (has_headers.get() and rows) else rows + header, body = normalize_tabular_rows(rows, has_headers.get()) if not body: messagebox.showerror("Error", "The file appears to have no data rows.", parent=dlg) return col_idx = selected_col.get() - out: list[str] = [] - for r in body: - val = r[col_idx] if col_idx < len(r) else "" - out.append("" if val is None else str(val)) + if col_idx >= len(header): + messagebox.showerror("Error", "Selected column is out of range for the imported file.", parent=dlg) + return # Resolve dataset_name mode = dataset_name_mode.get() @@ -493,8 +492,15 @@ def _do_import(): messagebox.showerror("Dataset Name error", "Please type a dataset_name in the 'Other' field.", parent=dlg) return - result_values = out - result_dataset_name = dataset_name + result_dataset = ImportedDataset( + file_path=path, + dataset_name=dataset_name, + has_headers=has_headers.get(), + selected_column_index=col_idx, + selected_column_name=header[col_idx], + columns=header, + rows=rows_to_records(header, body), + ) dlg.destroy() process_btn.configure(command=_do_import) @@ -513,6 +519,64 @@ def _do_import(): owner.wait_window(dlg) _safe_destroy(created_root) - if result_values is None or result_dataset_name is None: + return result_dataset + + +def import_data( + parent: Optional[tk.Misc] = None, + title: str = "Import Data", + filetypes: Sequence[tuple[str, str]] = ( + ("All files", "*.*"), + ("Excel files", "*.xlsx *.xls"), + ("CSV files", "*.csv"), + ("TSV files", "*.tsv"), + ("Text files", "*.txt"), + ), +) -> Optional[tuple[list[str], str]]: + imported = import_tabular_data(parent=parent, title=title, filetypes=filetypes) + if imported is None: return None - return result_values, result_dataset_name + return imported.selected_values, imported.dataset_name +@dataclass +class ImportedDataset: + file_path: str + dataset_name: str + has_headers: bool + selected_column_index: int + selected_column_name: str + columns: list[str] + rows: list[dict[str, str]] + + @property + def selected_values(self) -> list[str]: + return [row.get(self.selected_column_name, "") for row in self.rows] + + +def normalize_tabular_rows(rows: list[list[str]], has_headers: bool) -> tuple[list[str], list[list[str]]]: + if rows: + if has_headers: + header = rows[0] + body = rows[1:] + else: + max_cols = max((len(r) for r in rows), default=1) + header = [f"Column {i+1}" for i in range(max_cols)] + body = rows + else: + header, body = [f"Column {i+1}" for i in range(5)], [] + + normalized_header = [ + str(h) if (h is not None and str(h).strip() != "") else f"Column {i+1}" + for i, h in enumerate(header) + ] + return normalized_header, body + + +def rows_to_records(columns: Sequence[str], rows: Sequence[Sequence[str]]) -> list[dict[str, str]]: + records: list[dict[str, str]] = [] + for row in rows: + record: dict[str, str] = {} + for idx, column in enumerate(columns): + value = row[idx] if idx < len(row) else "" + record[column] = "" if value is None else str(value) + records.append(record) + return records diff --git a/live_processing/keyword_extraction_live.py b/live_processing/keyword_extraction_live.py index 109681f..7592519 100644 --- a/live_processing/keyword_extraction_live.py +++ b/live_processing/keyword_extraction_live.py @@ -1,76 +1,15 @@ -""" -Live text classification processing using OpenAI API. +"""Legacy keyword-extraction entry point backed by the prompt editor preset.""" -This module provides real-time text classification functionality that processes -text snippets immediately using OpenAI's API, as opposed to batch processing. -This is useful for smaller datasets or when immediate results are needed. -""" +from __future__ import annotations from typing import Optional import tkinter as tk -from tkinter import messagebox -from pydantic import BaseModel, ValidationError, Field, ConfigDict -from file_handling.data_import import import_data -from file_handling.data_conversion import save_as_csv, to_long_df -from settings import config -from batch_processing.batch_method import get_client -# Progress UI lives in a separate module -from ui.progress_ui import ProgressController +from prompt_editor.presets import KEYWORD_EXTRACTION_PRESET_ID +from prompt_editor.ui import open_prompt_editor def keyword_extraction_pipeline(parent: Optional[tk.Misc] = None): - """ - Prompt for quotes CSVs, extract keywords from each quote, - show progress, then save results to CSV. - """ - try: - client = get_client() - except Exception as e: - messagebox.showerror("API Key Required", str(e)) + if parent is None: return - - # Get quotes data - from_import = import_data(parent, "Select the quotes data") - if from_import is None: - return # user hit Cancel - quotes, quotes_nickname = from_import - - class KeywordExtraction(BaseModel): - id: int | None = None - quote: str - keywords: list[str] = Field(..., min_length=1) - model_config = ConfigDict() - - total = len(quotes) - - progress = ProgressController.open(parent=parent, total_count=total, title="Processing quotes…") - - results: list[KeywordExtraction] = [] - try: - for idx, q in enumerate(quotes, start=1): - try: - resp = client.responses.parse( - model=config.model, - input=[{"role": "system", "content": "You are an expert at structured data extraction."}, - {"role": "user", "content": f"Extract the keywords from this quote: {q}"}], - text_format=KeywordExtraction, - ) - decision = resp.output_parsed - row = KeywordExtraction( - id=idx, - quote=q, - **decision.model_dump(exclude={'id', 'quote'}) # <- prevents duplicate kwargs - ) - results.append(row) - except ValidationError as ve: - print(f"[VALIDATION ERROR] {str(q)[:60]}... -> {ve}") - except Exception as e: - print(f"[API ERROR] {str(q)[:60]}... -> {e}") - finally: - progress.update(idx, message=f"Processed {idx} of {total} quotes") - finally: - progress.close() - - df = to_long_df(results) - save_as_csv(df) \ No newline at end of file + open_prompt_editor(parent, preset_id=KEYWORD_EXTRACTION_PRESET_ID, execution_mode="live") diff --git a/live_processing/multi_label_live.py b/live_processing/multi_label_live.py index e40b8f0..65f1106 100644 --- a/live_processing/multi_label_live.py +++ b/live_processing/multi_label_live.py @@ -1,88 +1,15 @@ -""" -Live text classification processing using OpenAI API. +"""Legacy multi-label entry point backed by the prompt editor preset.""" -This module provides real-time text classification functionality that processes -text snippets immediately using OpenAI's API, as opposed to batch processing. -This is useful for smaller datasets or when immediate results are needed. -""" +from __future__ import annotations -from typing import Optional, List +from typing import Optional import tkinter as tk -from tkinter import messagebox -from pydantic import BaseModel, ValidationError, Field, ConfigDict -from file_handling.data_import import import_data -from file_handling.data_conversion import make_str_enum, save_as_csv, to_long_df -from settings import config -from batch_processing.batch_method import get_client -# Progress UI lives in a separate module -from ui.progress_ui import ProgressController +from prompt_editor.presets import MULTI_LABEL_PRESET_ID +from prompt_editor.ui import open_prompt_editor def multi_label_pipeline(parent: Optional[tk.Misc] = None): - """ - Prompt for labels/quotes CSVs, classify each quote with 1+ labels, - show progress, then save results to CSV. - """ - try: - client = get_client() - except Exception as e: - messagebox.showerror("API Key Required", str(e)) + if parent is None: return - - # Get labels data - from_import = import_data(parent, "Select the labels data") - if from_import is None: - return # user hit Cancel - label_values, labels_nickname = from_import - labels = make_str_enum("Label", label_values) - - # Get quotes data - from_import = import_data(parent, "Select the quotes data") - if from_import is None: - return # user hit Cancel - quotes, quotes_nickname = from_import - - class LabeledQuoteMulti(BaseModel): - id: int | None = None - quote: str - label: List[labels] = Field(..., min_length=1) - model_config = ConfigDict(use_enum_values=True, extra='forbid') - - total = len(quotes) - - progress = ProgressController.open(parent=parent, total_count=total, title="Processing quotes…") - - results: list[LabeledQuoteMulti] = [] - try: - for idx, q in enumerate(quotes, start=1): - try: - resp = client.responses.parse( - model=config.model, - input=[{ - "role": "user", - "content": ( - "Label this quote with labels from the allowed set only. " - f"Allowed: {', '.join(labels)}\nQuote: {q}" - ), - }], - text_format=LabeledQuoteMulti, - ) - decision = resp.output_parsed - row = LabeledQuoteMulti( - id=idx, - quote=q, - **decision.model_dump(exclude={'id', 'quote'}) # <- prevents duplicate kwargs - ) - results.append(row) - except ValidationError as ve: - print(f"[VALIDATION ERROR] {str(q)[:60]}... -> {ve}") - except Exception as e: - print(f"[API ERROR] {str(q)[:60]}... -> {e}") - finally: - progress.update(idx, message=f"Processed {idx} of {total} quotes") - finally: - progress.close() - - df = to_long_df(results) - save_as_csv(df) + open_prompt_editor(parent, preset_id=MULTI_LABEL_PRESET_ID, execution_mode="live") diff --git a/live_processing/single_label_live.py b/live_processing/single_label_live.py index 896232c..a5da844 100644 --- a/live_processing/single_label_live.py +++ b/live_processing/single_label_live.py @@ -1,89 +1,15 @@ -""" -Live text classification processing using OpenAI API. +"""Legacy single-label entry point backed by the prompt editor preset.""" -This module provides real-time text classification functionality that processes -text snippets immediately using OpenAI's API, as opposed to batch processing. -This is useful for smaller datasets or when immediate results are needed. -""" +from __future__ import annotations from typing import Optional import tkinter as tk -from tkinter import messagebox -from pydantic import BaseModel, ValidationError, Field, ConfigDict -from file_handling.data_import import import_data -from file_handling.data_conversion import make_str_enum, save_as_csv, to_long_df -from settings import config -from batch_processing.batch_method import get_client -# Progress UI lives in a separate module -from ui.progress_ui import ProgressController +from prompt_editor.presets import SINGLE_LABEL_PRESET_ID +from prompt_editor.ui import open_prompt_editor def single_label_pipeline(parent: Optional[tk.Misc] = None): - """ - Prompt for labels/quotes CSVs, classify each quote with exactly one label, - show progress, then save results to CSV. - """ - try: - client = get_client() - except Exception as e: - messagebox.showerror("API Key Required", str(e)) + if parent is None: return - - # Get labels data - from_import = import_data(parent, "Select the labels data") - if from_import is None: - return # user hit Cancel - label_values, labels_nickname = from_import - labels = make_str_enum("Label", label_values) - - # Get quotes data - from_import = import_data(parent, "Select the quotes data") - if from_import is None: - return # user hit Cancel - quotes, quotes_nickname = from_import - - class LabeledQuote(BaseModel): - id: int | None = None - quote: str - label: labels # STRICT: must be one of labels - model_config = ConfigDict(use_enum_values=True, extra='forbid') - - - total = len(quotes) - - progress = ProgressController.open(parent=parent, total_count=total, title="Processing quotes…") - - results: list[LabeledQuote] = [] - try: - for idx, q in enumerate(quotes, start=1): - try: - resp = client.responses.parse( - model=config.model, - input=[{ - "role": "user", - "content": ( - f"Label this quote with exactly one label from the allowed set.\n" - f"Quote: {q}" - ), - }], - text_format=LabeledQuote, - ) - decision = resp.output_parsed - row = LabeledQuote( - id=idx, - quote=q, - **decision.model_dump(exclude={'id', 'quote'}) # <- prevents duplicate kwargs - ) - results.append(row) - except ValidationError as ve: - print(f"[VALIDATION ERROR] {str(q)[:60]}... -> {ve}") - except Exception as e: - print(f"[API ERROR] {str(q)[:60]}... -> {e}") - finally: - progress.update(idx, message=f"Processed {idx} of {total} quotes") - finally: - progress.close() - - df = to_long_df(results) - save_as_csv(df) \ No newline at end of file + open_prompt_editor(parent, preset_id=SINGLE_LABEL_PRESET_ID, execution_mode="live") diff --git a/prompt_editor/__init__.py b/prompt_editor/__init__.py new file mode 100644 index 0000000..b0aa074 --- /dev/null +++ b/prompt_editor/__init__.py @@ -0,0 +1,2 @@ +"""Shared prompt editor modules for live and batch LLM workflows.""" + diff --git a/prompt_editor/config.py b/prompt_editor/config.py new file mode 100644 index 0000000..8cddfeb --- /dev/null +++ b/prompt_editor/config.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + +PromptExecutionMode = Literal["live", "batch"] +PromptTemplateKind = Literal["preset", "custom"] +PlaceholderRenderMode = Literal["row_value", "full_column", "unique_values", "joined_text", "constant"] +OutputSourceType = Literal["model_output", "passthrough", "system"] +OutputFieldType = Literal["enum", "text", "integer", "boolean", "list[str]"] + +PRIMARY_SOURCE_ID = "primary" +LIVE_ROW_WARNING_THRESHOLD = 50 + + +class PromptEditorBaseModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class AdvancedSettings(PromptEditorBaseModel): + temperature: float | None = None + max_output_tokens: int | None = None + reasoning_effort: str | None = None + detail_level: str | None = None + + +class PrimarySourceDefinition(PromptEditorBaseModel): + source_id: str = PRIMARY_SOURCE_ID + dataset_name: str = "" + file_path: str | None = None + has_headers: bool = True + selected_column: str = "" + available_columns: list[str] = Field(default_factory=list) + passthrough_columns: list[str] = Field(default_factory=list) + rows: list[dict[str, str]] = Field(default_factory=list) + + +class AdditionalSourceDefinition(PromptEditorBaseModel): + source_id: str + dataset_name: str + file_path: str | None = None + has_headers: bool = True + selected_column: str = "" + available_columns: list[str] = Field(default_factory=list) + render_mode: PlaceholderRenderMode = "unique_values" + rows: list[dict[str, str]] = Field(default_factory=list) + + +class PlaceholderBinding(PromptEditorBaseModel): + token: str + source_id: str + source_kind: Literal["primary", "additional", "system"] + column_name: str | None = None + render_mode: PlaceholderRenderMode = "row_value" + constant_value: str | None = None + + +class OutputDefinition(PromptEditorBaseModel): + name: str + source_type: OutputSourceType + field_type: OutputFieldType + required: bool = True + instructions: str = "" + enum_source_id: str | None = None + enum_source_column: str | None = None + passthrough_column: str | None = None + system_value: str | None = None + + +class PromptTemplateConfig(PromptEditorBaseModel): + version: int = 1 + template_name: str + template_kind: PromptTemplateKind = "custom" + preset_id: str | None = None + execution_mode: PromptExecutionMode = "live" + model: str = "gpt-4o-mini" + advanced: AdvancedSettings = Field(default_factory=AdvancedSettings) + system_prompt: str = "" + user_prompt_template: str = "" + primary_source: PrimarySourceDefinition = Field(default_factory=PrimarySourceDefinition) + additional_sources: list[AdditionalSourceDefinition] = Field(default_factory=list) + placeholder_bindings: list[PlaceholderBinding] = Field(default_factory=list) + output_definitions: list[OutputDefinition] = Field(default_factory=list) + live_row_warning_threshold: int = LIVE_ROW_WARNING_THRESHOLD + + +class BatchRowContext(PromptEditorBaseModel): + custom_id: str + row_number: int + row_data: dict[str, str] = Field(default_factory=dict) + + +class BatchJobRecord(PromptEditorBaseModel): + batch_id: str + project_key: str + template_name: str + preset_id: str | None = None + model: str + output_definitions: list[OutputDefinition] + row_contexts: list[BatchRowContext] = Field(default_factory=list) + enum_values_by_output: dict[str, list[str]] = Field(default_factory=dict) diff --git a/prompt_editor/engine.py b/prompt_editor/engine.py new file mode 100644 index 0000000..96c0d5e --- /dev/null +++ b/prompt_editor/engine.py @@ -0,0 +1,638 @@ +from __future__ import annotations + +import io +import json +import re +from collections.abc import Sequence +from copy import deepcopy +from typing import Any + +import pandas as pd +from pydantic import BaseModel, ConfigDict, Field, ValidationError, create_model + +from batch_processing.batch_creation import forbid_additional_props +from batch_processing.batch_error_handling import handle_batch_fail +from file_handling.data_conversion import make_str_enum +from file_handling.data_import import ( + ImportedDataset, + _load_tabular, + normalize_tabular_rows, + rows_to_records, +) +from prompt_editor.config import ( + AdditionalSourceDefinition, + BatchJobRecord, + BatchRowContext, + OutputDefinition, + PlaceholderBinding, + PrimarySourceDefinition, + PromptTemplateConfig, +) +from prompt_editor.storage import get_project_key, load_batch_job, save_batch_job +from settings import secrets_store +from ui.progress_ui import ProgressController +from openai import OpenAI + +TOKEN_PATTERN = re.compile(r"<([a-zA-Z0-9_]+)>") + + +class _StructuredResponseModel(BaseModel): + model_config = ConfigDict(use_enum_values=True, extra="forbid") + + +def get_client() -> OpenAI: + api_key = secrets_store.load_api_key() + if not api_key: + raise Exception("OpenAI API key not configured. Please set it in Settings.") + return OpenAI(api_key=api_key) + + +def get_batch_status(batch_id: str) -> Any: + client = get_client() + return client.batches.retrieve(batch_id) + + +def make_source_id(name: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", name.strip().lower()).strip("-") + return slug or "source" + + +def build_primary_source(imported: ImportedDataset, passthrough_columns: list[str] | None = None) -> PrimarySourceDefinition: + selected_column = imported.selected_column_name + passthrough = passthrough_columns or ([selected_column] if selected_column else []) + return PrimarySourceDefinition( + dataset_name=imported.dataset_name, + file_path=imported.file_path, + has_headers=imported.has_headers, + selected_column=selected_column, + available_columns=imported.columns, + passthrough_columns=passthrough, + rows=deepcopy(imported.rows), + ) + + +def build_additional_source( + imported: ImportedDataset, + source_id: str | None = None, + render_mode: str = "unique_values", +) -> AdditionalSourceDefinition: + return AdditionalSourceDefinition( + source_id=source_id or make_source_id(imported.dataset_name), + dataset_name=imported.dataset_name, + file_path=imported.file_path, + has_headers=imported.has_headers, + selected_column=imported.selected_column_name, + available_columns=imported.columns, + render_mode=render_mode, + rows=deepcopy(imported.rows), + ) + + +def load_source_rows(file_path: str, has_headers: bool) -> tuple[list[str], list[dict[str, str]]]: + rows = _load_tabular(file_path) + columns, body = normalize_tabular_rows(rows, has_headers) + return columns, rows_to_records(columns, body) + + +def hydrate_config(config: PromptTemplateConfig) -> PromptTemplateConfig: + hydrated = config.model_copy(deep=True) + + if hydrated.primary_source.file_path and not hydrated.primary_source.rows: + columns, rows = load_source_rows(hydrated.primary_source.file_path, hydrated.primary_source.has_headers) + hydrated.primary_source.available_columns = columns + hydrated.primary_source.rows = rows + + for source in hydrated.additional_sources: + if source.file_path and not source.rows: + columns, rows = load_source_rows(source.file_path, source.has_headers) + source.available_columns = columns + source.rows = rows + + return hydrated + + +def _find_source(config: PromptTemplateConfig, source_id: str) -> PrimarySourceDefinition | AdditionalSourceDefinition: + if source_id == config.primary_source.source_id: + return config.primary_source + for source in config.additional_sources: + if source.source_id == source_id: + return source + raise ValueError(f"Unknown source binding: {source_id}") + + +def list_available_tokens(config: PromptTemplateConfig) -> list[str]: + return [binding.token for binding in config.placeholder_bindings] + + +def _non_empty_values(rows: list[dict[str, str]], column_name: str) -> list[str]: + values: list[str] = [] + for row in rows: + value = str(row.get(column_name, "")).strip() + if value: + values.append(value) + return values + + +def enum_values_for_output( + config: PromptTemplateConfig, + output_def: OutputDefinition, + enum_values_override: dict[str, list[str]] | None = None, +) -> list[str]: + if enum_values_override and output_def.name in enum_values_override: + values = enum_values_override[output_def.name] + if values: + return values + + if not output_def.enum_source_id: + raise ValueError(f"Output '{output_def.name}' is missing an enum source.") + + source = _find_source(config, output_def.enum_source_id) + column_name = output_def.enum_source_column or source.selected_column + if not column_name: + raise ValueError(f"Output '{output_def.name}' has no enum source column.") + if column_name not in source.available_columns and source.available_columns: + raise ValueError(f"Column '{column_name}' is not available in source '{source.dataset_name}'.") + + seen: set[str] = set() + ordered: list[str] = [] + for value in _non_empty_values(source.rows, column_name): + if value not in seen: + seen.add(value) + ordered.append(value) + if not ordered: + raise ValueError(f"Output '{output_def.name}' resolved to an empty allowed-label set.") + return ordered + + +def build_response_model( + config: PromptTemplateConfig, + enum_values_override: dict[str, list[str]] | None = None, +) -> tuple[type[BaseModel], dict, dict[str, list[str]]]: + model_fields: dict[str, tuple[Any, Any]] = {} + enum_cache: dict[str, list[str]] = {} + + model_outputs = [output for output in config.output_definitions if output.source_type == "model_output"] + if not model_outputs: + raise ValueError("Add at least one model output field before running the prompt.") + + for index, output_def in enumerate(model_outputs, start=1): + description = output_def.instructions.strip() or None + + if output_def.field_type == "enum": + enum_values = enum_values_for_output(config, output_def, enum_values_override) + enum_cache[output_def.name] = enum_values + enum_type = make_str_enum(f"EnumField{index}", enum_values) + annotation = enum_type if output_def.required else enum_type | None + field_value = Field(... if output_def.required else None, description=description) + elif output_def.field_type == "text": + annotation = str if output_def.required else str | None + field_value = Field(... if output_def.required else None, description=description) + elif output_def.field_type == "integer": + annotation = int if output_def.required else int | None + field_value = Field(... if output_def.required else None, description=description) + elif output_def.field_type == "boolean": + annotation = bool if output_def.required else bool | None + field_value = Field(... if output_def.required else None, description=description) + elif output_def.field_type == "list[str]": + if output_def.enum_source_id: + enum_values = enum_values_for_output(config, output_def, enum_values_override) + enum_cache[output_def.name] = enum_values + item_type = make_str_enum(f"EnumListField{index}", enum_values) + annotation = list[item_type] if output_def.required else list[item_type] | None + else: + annotation = list[str] if output_def.required else list[str] | None + default = ... if output_def.required else None + field_value = Field(default, min_length=1 if output_def.required else None, description=description) + else: + raise ValueError(f"Unsupported output field type: {output_def.field_type}") + + model_fields[output_def.name] = (annotation, field_value) + + response_model = create_model("PromptEditorResponse", __base__=_StructuredResponseModel, **model_fields) + strict_schema = forbid_additional_props(response_model.model_json_schema()) + return response_model, strict_schema, enum_cache + + +def _render_source_value( + rows: list[dict[str, str]], + column_name: str, + render_mode: str, + row_data: dict[str, str] | None, + constant_value: str | None = None, +) -> str: + if render_mode == "constant": + return constant_value or "" + if render_mode == "row_value": + if row_data is None: + return "" + return str(row_data.get(column_name, "")) + + values = _non_empty_values(rows, column_name) + if render_mode == "full_column": + return "\n".join(values) + if render_mode == "unique_values": + seen: set[str] = set() + unique_values: list[str] = [] + for value in values: + if value not in seen: + seen.add(value) + unique_values.append(value) + return "\n".join(unique_values) + if render_mode == "joined_text": + return " ".join(values) + raise ValueError(f"Unsupported placeholder render mode: {render_mode}") + + +def render_prompt_text(template_text: str, values: dict[str, str]) -> str: + def _replace(match: re.Match[str]) -> str: + token = match.group(1) + if token not in values: + raise ValueError(f"Prompt token <{token}> is not bound.") + return values[token] + + return TOKEN_PATTERN.sub(_replace, template_text) + + +def resolve_placeholder_values(config: PromptTemplateConfig, row_data: dict[str, str] | None) -> dict[str, str]: + values: dict[str, str] = {} + for binding in config.placeholder_bindings: + if binding.source_kind == "system": + values[binding.token] = binding.constant_value or "" + continue + + source = _find_source(config, binding.source_id) + column_name = binding.column_name or source.selected_column + if not column_name: + raise ValueError(f"Placeholder <{binding.token}> is missing a source column.") + if source.available_columns and column_name not in source.available_columns: + raise ValueError( + f"Placeholder <{binding.token}> references unknown column '{column_name}' in '{source.dataset_name}'." + ) + + values[binding.token] = _render_source_value( + rows=source.rows, + column_name=column_name, + render_mode=binding.render_mode if binding.source_kind != "additional" else binding.render_mode or source.render_mode, + row_data=row_data, + constant_value=binding.constant_value, + ) + + return values + + +def render_prompts_for_row(config: PromptTemplateConfig, row_index: int = 0) -> tuple[str, str]: + hydrated = hydrate_config(config) + validate_config(hydrated) + row_data = hydrated.primary_source.rows[row_index] + placeholder_values = resolve_placeholder_values(hydrated, row_data) + system_prompt = render_prompt_text(hydrated.system_prompt, placeholder_values) if hydrated.system_prompt else "" + user_prompt = render_prompt_text(hydrated.user_prompt_template, placeholder_values) + + guidance_lines = [ + f"- {output.name}: {output.instructions.strip()}" + for output in hydrated.output_definitions + if output.source_type == "model_output" and output.instructions.strip() + ] + if guidance_lines: + user_prompt = f"{user_prompt}\n\nOutput field instructions:\n" + "\n".join(guidance_lines) + return system_prompt, user_prompt + + +def build_request_body(config: PromptTemplateConfig, row_index: int, strict_schema: dict) -> dict[str, Any]: + system_prompt, user_prompt = render_prompts_for_row(config, row_index) + body: dict[str, Any] = { + "model": config.model, + "input": [], + "text": { + "format": { + "type": "json_schema", + "name": "PromptEditorResponse", + "schema": strict_schema, + "strict": True, + } + }, + "metadata": { + "template_name": config.template_name, + "row_number": str(row_index + 1), + }, + } + if system_prompt: + body["input"].append({"role": "system", "content": system_prompt}) + body["input"].append({"role": "user", "content": user_prompt}) + + if config.advanced.temperature is not None: + body["temperature"] = config.advanced.temperature + if config.advanced.max_output_tokens is not None: + body["max_output_tokens"] = config.advanced.max_output_tokens + if config.advanced.reasoning_effort: + body["reasoning"] = {"effort": config.advanced.reasoning_effort} + if config.advanced.detail_level: + body["text"]["verbosity"] = config.advanced.detail_level + return body + + +def extract_response_text(payload: Any) -> str: + if hasattr(payload, "model_dump"): + payload = payload.model_dump(mode="json") + if isinstance(payload, dict): + output = payload.get("output", []) + else: + output = getattr(payload, "output", []) + if not output: + raise ValueError("Model response did not include any output content.") + + first_output = output[0] + content = first_output.get("content", []) if isinstance(first_output, dict) else getattr(first_output, "content", []) + if not content: + raise ValueError("Model response did not include any content blocks.") + + first_content = content[0] + text = first_content.get("text") if isinstance(first_content, dict) else getattr(first_content, "text", None) + if text is None: + raise ValueError("Model response did not include text output.") + return str(text) + + +def parse_structured_response(raw_text: str, response_model: type[BaseModel]) -> dict[str, Any]: + try: + payload = json.loads(raw_text) + except json.JSONDecodeError as exc: + raise ValueError(f"Model returned invalid JSON: {exc.msg}") from exc + + try: + parsed = response_model.model_validate(payload) + except ValidationError as exc: + raise ValueError(f"Model output did not match the required schema: {exc}") from exc + return parsed.model_dump(mode="json") + + +def _coerce_output_value(value: Any, field_type: str) -> Any: + if value is None: + return None + if field_type == "integer": + return int(value) + if field_type == "boolean": + if isinstance(value, bool): + return value + lowered = str(value).strip().lower() + if lowered in {"true", "1", "yes"}: + return True + if lowered in {"false", "0", "no"}: + return False + if field_type == "list[str]" and not isinstance(value, list): + return [str(value)] + return value + + +def build_export_row( + output_definitions: list[OutputDefinition], + row_number: int, + row_data: dict[str, str], + parsed_output: dict[str, Any] | None, + raw_response: str, + error_message: str | None = None, +) -> dict[str, Any]: + export_row: dict[str, Any] = { + "row_number": row_number, + "status": "error" if error_message else "success", + "error_message": error_message or "", + "raw_response": raw_response, + } + + for output in output_definitions: + if output.source_type == "model_output": + value = None if parsed_output is None else parsed_output.get(output.name) + elif output.source_type == "passthrough": + column_name = output.passthrough_column or output.name + value = row_data.get(column_name, "") + elif output.source_type == "system": + value = output.system_value or "" + else: + raise ValueError(f"Unsupported output source type: {output.source_type}") + export_row[output.name] = _coerce_output_value(value, output.field_type) + + return export_row + + +def export_rows_to_dataframe(rows: list[dict[str, Any]]) -> pd.DataFrame: + df = pd.json_normalize(rows, sep=".") + list_columns = [ + column + for column in df.columns + if df[column].apply(lambda value: isinstance(value, Sequence) and not isinstance(value, (str, bytes))).any() + ] + + for column in list_columns: + df[column] = df[column].apply( + lambda value: [None] + if value is None + else value + if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) + else [value] + ) + + if list_columns: + df = df.explode(list_columns, ignore_index=True) + return df + + +def validate_config(config: PromptTemplateConfig) -> None: + if not config.primary_source.rows: + raise ValueError("Import a primary source dataset before running the prompt.") + if not config.primary_source.selected_column: + raise ValueError("Select a primary source column before running the prompt.") + if config.primary_source.available_columns and config.primary_source.selected_column not in config.primary_source.available_columns: + raise ValueError("The selected primary source column is no longer available.") + if not config.user_prompt_template.strip(): + raise ValueError("Enter a user prompt template before running the prompt.") + + tokens_in_prompts = set(TOKEN_PATTERN.findall(config.system_prompt + "\n" + config.user_prompt_template)) + bindings = {binding.token for binding in config.placeholder_bindings} + missing = sorted(token for token in tokens_in_prompts if token not in bindings) + if missing: + raise ValueError(f"Missing placeholder bindings for: {', '.join(f'<{token}>' for token in missing)}") + + for output in config.output_definitions: + if output.source_type == "passthrough": + column_name = output.passthrough_column or output.name + if column_name not in config.primary_source.passthrough_columns: + raise ValueError( + f"Passthrough output '{output.name}' must reference a column selected in passthrough columns." + ) + + +def run_live_prompt(config: PromptTemplateConfig, parent=None) -> pd.DataFrame: + client = get_client() + hydrated = hydrate_config(config) + validate_config(hydrated) + response_model, strict_schema, _ = build_response_model(hydrated) + + rows: list[dict[str, Any]] = [] + total = len(hydrated.primary_source.rows) + progress = ProgressController.open(parent=parent, total_count=total, title="Processing rows…") + try: + for index, row_data in enumerate(hydrated.primary_source.rows, start=1): + raw_response = "" + parsed_output: dict[str, Any] | None = None + error_message: str | None = None + try: + request_body = build_request_body(hydrated, index - 1, strict_schema) + response = client.responses.create(**request_body) + raw_response = extract_response_text(response) + parsed_output = parse_structured_response(raw_response, response_model) + except Exception as exc: + error_message = str(exc) + rows.append( + build_export_row( + output_definitions=hydrated.output_definitions, + row_number=index, + row_data=row_data, + parsed_output=parsed_output, + raw_response=raw_response, + error_message=error_message, + ) + ) + progress.update(index, message=f"Processed {index} of {total} rows") + finally: + progress.close() + + return export_rows_to_dataframe(rows) + + +def run_single_test_row(config: PromptTemplateConfig) -> dict[str, Any]: + client = get_client() + hydrated = hydrate_config(config) + validate_config(hydrated) + response_model, strict_schema, _ = build_response_model(hydrated) + response = client.responses.create(**build_request_body(hydrated, 0, strict_schema)) + raw_response = extract_response_text(response) + parsed_output = parse_structured_response(raw_response, response_model) + return build_export_row( + output_definitions=hydrated.output_definitions, + row_number=1, + row_data=hydrated.primary_source.rows[0], + parsed_output=parsed_output, + raw_response=raw_response, + ) + + +def submit_batch_prompt(config: PromptTemplateConfig) -> Any: + client = get_client() + hydrated = hydrate_config(config) + validate_config(hydrated) + _, strict_schema, enum_cache = build_response_model(hydrated) + + batch_buffer = io.BytesIO() + batch_buffer.name = "batchinput.jsonl" + row_contexts: list[BatchRowContext] = [] + + for index, row_data in enumerate(hydrated.primary_source.rows, start=1): + custom_id = f"row-{index:05d}" + request_body = build_request_body(hydrated, index - 1, strict_schema) + line = { + "custom_id": custom_id, + "method": "POST", + "url": "/v1/responses", + "body": request_body, + } + batch_buffer.write((json.dumps(line, ensure_ascii=False) + "\n").encode("utf-8")) + row_contexts.append(BatchRowContext(custom_id=custom_id, row_number=index, row_data=deepcopy(row_data))) + + batch_buffer.seek(0) + uploaded = client.files.create(file=batch_buffer, purpose="batch") + batch = client.batches.create( + input_file_id=uploaded.id, + endpoint="/v1/responses", + completion_window="24h", + metadata={ + "model": hydrated.model, + "type": hydrated.template_name, + "dataset(s)": hydrated.primary_source.dataset_name, + }, + ) + + save_batch_job( + BatchJobRecord( + batch_id=batch.id, + project_key=get_project_key(), + template_name=hydrated.template_name, + preset_id=hydrated.preset_id, + model=hydrated.model, + output_definitions=hydrated.output_definitions, + row_contexts=row_contexts, + enum_values_by_output=enum_cache, + ) + ) + return batch + + +def load_prompt_editor_batch_results(batch_id: str) -> pd.DataFrame | None: + record = load_batch_job(batch_id) + if record is None: + return None + + client = get_client() + status = get_batch_status(batch_id) + if status.output_file_id is None: + handle_batch_fail(client, status) + return None + + file_response = client.files.content(status.output_file_id).content + results = [ + json.loads(line) + for line in file_response.decode("utf-8").splitlines() + if line.strip() + ] + + config = PromptTemplateConfig( + template_name=record.template_name, + model=record.model, + output_definitions=record.output_definitions, + primary_source=PrimarySourceDefinition(rows=[]), + ) + response_model, _, _ = build_response_model(config, record.enum_values_by_output) + + row_lookup = {context.custom_id: context for context in record.row_contexts} + seen_ids: set[str] = set() + exported_rows: list[dict[str, Any]] = [] + + for result in results: + custom_id = result.get("custom_id", "") + seen_ids.add(custom_id) + context = row_lookup.get(custom_id, BatchRowContext(custom_id=custom_id, row_number=0, row_data={})) + raw_response = "" + parsed_output: dict[str, Any] | None = None + error_message: str | None = None + try: + body = result["response"]["body"] + raw_response = extract_response_text(body) + parsed_output = parse_structured_response(raw_response, response_model) + except Exception as exc: + error_message = str(exc) + exported_rows.append( + build_export_row( + output_definitions=record.output_definitions, + row_number=context.row_number, + row_data=context.row_data, + parsed_output=parsed_output, + raw_response=raw_response, + error_message=error_message, + ) + ) + + for custom_id, context in row_lookup.items(): + if custom_id in seen_ids: + continue + exported_rows.append( + build_export_row( + output_definitions=record.output_definitions, + row_number=context.row_number, + row_data=context.row_data, + parsed_output=None, + raw_response="", + error_message="No result was returned for this row.", + ) + ) + + exported_rows.sort(key=lambda row: row["row_number"]) + return export_rows_to_dataframe(exported_rows) diff --git a/prompt_editor/presets.py b/prompt_editor/presets.py new file mode 100644 index 0000000..5d64389 --- /dev/null +++ b/prompt_editor/presets.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from prompt_editor.config import ( + AdditionalSourceDefinition, + OutputDefinition, + PlaceholderBinding, + PrimarySourceDefinition, + PromptTemplateConfig, +) +from settings.user_config import get_setting + +SINGLE_LABEL_PRESET_ID = "single-label-classification" +MULTI_LABEL_PRESET_ID = "multi-label-classification" +KEYWORD_EXTRACTION_PRESET_ID = "keyword-extraction" + + +def _base_template(name: str, preset_id: str) -> PromptTemplateConfig: + return PromptTemplateConfig( + template_name=name, + template_kind="preset", + preset_id=preset_id, + model=get_setting("model", "gpt-4o-mini"), + primary_source=PrimarySourceDefinition( + dataset_name="quotes", + selected_column="quote", + ), + ) + + +def single_label_preset() -> PromptTemplateConfig: + return _base_template("Single Label Classification", SINGLE_LABEL_PRESET_ID).model_copy( + update={ + "system_prompt": "", + "user_prompt_template": ( + "Label this quote with exactly one label from the allowed set.\n" + "Allowed labels:\n\n\n" + "Quote:\n" + ), + "additional_sources": [ + AdditionalSourceDefinition( + source_id="labels", + dataset_name="labels", + selected_column="label", + render_mode="unique_values", + ) + ], + "placeholder_bindings": [ + PlaceholderBinding( + token="quote", + source_id="primary", + source_kind="primary", + render_mode="row_value", + ), + PlaceholderBinding( + token="labels", + source_id="labels", + source_kind="additional", + render_mode="unique_values", + ), + ], + "output_definitions": [ + OutputDefinition( + name="label", + source_type="model_output", + field_type="enum", + required=True, + instructions="Choose exactly one label from the allowed set.", + enum_source_id="labels", + enum_source_column="label", + ) + ], + } + ) + + +def multi_label_preset() -> PromptTemplateConfig: + return _base_template("Multi-Label Classification", MULTI_LABEL_PRESET_ID).model_copy( + update={ + "system_prompt": "", + "user_prompt_template": ( + "Label this quote using only labels from the allowed set.\n" + "Return one or more labels when they apply.\n" + "Allowed labels:\n\n\n" + "Quote:\n" + ), + "additional_sources": [ + AdditionalSourceDefinition( + source_id="labels", + dataset_name="labels", + selected_column="label", + render_mode="unique_values", + ) + ], + "placeholder_bindings": [ + PlaceholderBinding( + token="quote", + source_id="primary", + source_kind="primary", + render_mode="row_value", + ), + PlaceholderBinding( + token="labels", + source_id="labels", + source_kind="additional", + render_mode="unique_values", + ), + ], + "output_definitions": [ + OutputDefinition( + name="label", + source_type="model_output", + field_type="list[str]", + required=True, + instructions="Return one or more labels from the allowed set only.", + enum_source_id="labels", + enum_source_column="label", + ) + ], + } + ) + + +def keyword_extraction_preset() -> PromptTemplateConfig: + return _base_template("Keyword Extraction", KEYWORD_EXTRACTION_PRESET_ID).model_copy( + update={ + "system_prompt": "You are an expert at structured data extraction.", + "user_prompt_template": "Extract the keywords from this quote:\n", + "placeholder_bindings": [ + PlaceholderBinding( + token="quote", + source_id="primary", + source_kind="primary", + render_mode="row_value", + ) + ], + "output_definitions": [ + OutputDefinition( + name="keywords", + source_type="model_output", + field_type="list[str]", + required=True, + instructions="Return a concise list of keywords drawn from the quote.", + ) + ], + } + ) + + +PRESET_FACTORIES = { + SINGLE_LABEL_PRESET_ID: single_label_preset, + MULTI_LABEL_PRESET_ID: multi_label_preset, + KEYWORD_EXTRACTION_PRESET_ID: keyword_extraction_preset, +} + + +def get_preset(preset_id: str) -> PromptTemplateConfig: + try: + return PRESET_FACTORIES[preset_id]() + except KeyError as exc: + raise ValueError(f"Unknown prompt editor preset: {preset_id}") from exc + + +def list_presets() -> list[tuple[str, str]]: + return [ + (SINGLE_LABEL_PRESET_ID, "Single Label Classification"), + (MULTI_LABEL_PRESET_ID, "Multi-Label Classification"), + (KEYWORD_EXTRACTION_PRESET_ID, "Keyword Extraction"), + ] diff --git a/prompt_editor/storage.py b/prompt_editor/storage.py new file mode 100644 index 0000000..899fa0a --- /dev/null +++ b/prompt_editor/storage.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from prompt_editor.config import BatchJobRecord, PromptTemplateConfig +from settings.user_config import ensure_user_config_dir + + +def _project_root(project_root: str | Path | None = None) -> Path: + return Path(project_root or Path.cwd()).resolve() + + +def get_project_key(project_root: str | Path | None = None) -> str: + root = _project_root(project_root) + digest = hashlib.sha1(str(root).encode("utf-8")).hexdigest()[:12] + return f"{root.name}-{digest}" + + +def get_prompt_editor_dir(project_root: str | Path | None = None) -> Path: + directory = ensure_user_config_dir() / "prompt_editor" / get_project_key(project_root) + directory.mkdir(parents=True, exist_ok=True) + return directory + + +def get_templates_file(project_root: str | Path | None = None) -> Path: + return get_prompt_editor_dir(project_root) / "templates.json" + + +def get_batch_jobs_file(project_root: str | Path | None = None) -> Path: + return get_prompt_editor_dir(project_root) / "batch_jobs.json" + + +def _load_json(path: Path, default): + try: + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + pass + return default + + +def _save_json(path: Path, payload) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_suffix(path.suffix + ".tmp") + temp_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + temp_path.replace(path) + + +def _persistable_template(config: PromptTemplateConfig) -> dict: + data = config.model_dump(mode="json") + data["primary_source"]["rows"] = [] + for source in data["additional_sources"]: + source["rows"] = [] + return data + + +def list_templates(project_root: str | Path | None = None) -> list[PromptTemplateConfig]: + raw_templates = _load_json(get_templates_file(project_root), []) + templates: list[PromptTemplateConfig] = [] + for raw in raw_templates: + try: + templates.append(PromptTemplateConfig.model_validate(raw)) + except Exception: + continue + return templates + + +def load_template(template_name: str, project_root: str | Path | None = None) -> PromptTemplateConfig | None: + for template in list_templates(project_root): + if template.template_name == template_name: + return template + return None + + +def save_template(config: PromptTemplateConfig, project_root: str | Path | None = None) -> None: + templates = list_templates(project_root) + persisted = _persistable_template(config) + saved = False + output: list[dict] = [] + for template in templates: + if template.template_name == config.template_name: + output.append(persisted) + saved = True + else: + output.append(_persistable_template(template)) + if not saved: + output.append(persisted) + output.sort(key=lambda item: item["template_name"].lower()) + _save_json(get_templates_file(project_root), output) + + +def delete_template(template_name: str, project_root: str | Path | None = None) -> None: + templates = [t for t in list_templates(project_root) if t.template_name != template_name] + _save_json(get_templates_file(project_root), [_persistable_template(t) for t in templates]) + + +def save_batch_job(record: BatchJobRecord, project_root: str | Path | None = None) -> None: + path = get_batch_jobs_file(project_root) + jobs = _load_json(path, {}) + jobs[record.batch_id] = record.model_dump(mode="json") + _save_json(path, jobs) + + +def load_batch_job(batch_id: str, project_root: str | Path | None = None) -> BatchJobRecord | None: + jobs = _load_json(get_batch_jobs_file(project_root), {}) + raw = jobs.get(batch_id) + if raw is None: + return None + try: + return BatchJobRecord.model_validate(raw) + except Exception: + return None diff --git a/prompt_editor/ui.py b/prompt_editor/ui.py new file mode 100644 index 0000000..f245739 --- /dev/null +++ b/prompt_editor/ui.py @@ -0,0 +1,913 @@ +from __future__ import annotations + +import json +import tkinter as tk +from tkinter import messagebox, simpledialog, ttk + +from file_handling.data_conversion import save_as_csv +from file_handling.data_import import import_tabular_data +from prompt_editor.config import ( + AdditionalSourceDefinition, + OutputDefinition, + PlaceholderBinding, + PrimarySourceDefinition, + PromptTemplateConfig, +) +from prompt_editor.engine import ( + build_additional_source, + build_primary_source, + list_available_tokens, + make_source_id, + render_prompts_for_row, + run_live_prompt, + run_single_test_row, + submit_batch_prompt, +) +from prompt_editor.presets import get_preset, list_presets +from prompt_editor.storage import delete_template, list_templates, load_template, save_template +from settings.models_registry import get_models, refresh_models +from settings.user_config import get_setting +from ui.batch_operations import refresh_batches_async + +REASONING_OPTIONS = ["", "low", "medium", "high"] +DETAIL_OPTIONS = ["", "low", "medium", "high"] + + +class PlaceholderDialog(simpledialog.Dialog): + def __init__(self, parent, config: PromptTemplateConfig, binding: PlaceholderBinding | None = None): + self.config_model = config + self.binding = binding + self.result: PlaceholderBinding | None = None + super().__init__(parent, title="Placeholder Binding") + + def body(self, master): + ttk.Label(master, text="Token").grid(row=0, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Source kind").grid(row=1, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Source").grid(row=2, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Column").grid(row=3, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Render mode").grid(row=4, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Constant value").grid(row=5, column=0, sticky="w", padx=8, pady=6) + + self.var_token = tk.StringVar(value="" if self.binding is None else self.binding.token) + self.var_source_kind = tk.StringVar(value="primary" if self.binding is None else self.binding.source_kind) + self.var_source_id = tk.StringVar(value="primary" if self.binding is None else self.binding.source_id) + self.var_column = tk.StringVar(value="" if self.binding is None else (self.binding.column_name or "")) + self.var_render_mode = tk.StringVar(value="row_value" if self.binding is None else self.binding.render_mode) + self.var_constant = tk.StringVar(value="" if self.binding is None else (self.binding.constant_value or "")) + + ttk.Entry(master, textvariable=self.var_token, width=28).grid(row=0, column=1, sticky="ew", padx=8, pady=6) + self.cmb_kind = ttk.Combobox( + master, + textvariable=self.var_source_kind, + values=["primary", "additional", "system"], + state="readonly", + width=25, + ) + self.cmb_kind.grid(row=1, column=1, sticky="ew", padx=8, pady=6) + + self.cmb_source = ttk.Combobox(master, textvariable=self.var_source_id, state="readonly", width=25) + self.cmb_source.grid(row=2, column=1, sticky="ew", padx=8, pady=6) + self.cmb_column = ttk.Combobox(master, textvariable=self.var_column, state="readonly", width=25) + self.cmb_column.grid(row=3, column=1, sticky="ew", padx=8, pady=6) + self.cmb_mode = ttk.Combobox( + master, + textvariable=self.var_render_mode, + values=["row_value", "full_column", "unique_values", "joined_text", "constant"], + state="readonly", + width=25, + ) + self.cmb_mode.grid(row=4, column=1, sticky="ew", padx=8, pady=6) + self.ent_constant = ttk.Entry(master, textvariable=self.var_constant, width=28) + self.ent_constant.grid(row=5, column=1, sticky="ew", padx=8, pady=6) + + master.columnconfigure(1, weight=1) + self.cmb_kind.bind("<>", lambda _event: self._refresh_state()) + self.cmb_source.bind("<>", lambda _event: self._refresh_columns()) + self.cmb_mode.bind("<>", lambda _event: self._refresh_state()) + self._refresh_state() + return self.cmb_kind + + def _sources(self) -> list[tuple[str, str, list[str]]]: + sources = [("primary", self.config_model.primary_source.dataset_name or "primary", self.config_model.primary_source.available_columns)] + for source in self.config_model.additional_sources: + sources.append((source.source_id, source.dataset_name, source.available_columns)) + return sources + + def _refresh_columns(self): + selected = self.var_source_id.get() + columns = [] + for source_id, _name, available_columns in self._sources(): + if source_id == selected: + columns = available_columns + break + self.cmb_column["values"] = columns + if columns and self.var_column.get() not in columns: + self.var_column.set(columns[0]) + + def _refresh_state(self): + source_kind = self.var_source_kind.get() + if source_kind == "system": + self.cmb_source["values"] = [] + self.cmb_source.set("") + self.cmb_column["values"] = [] + self.cmb_column.set("") + self.cmb_mode.set("constant") + self.cmb_mode.configure(state="disabled") + self.cmb_source.configure(state="disabled") + self.cmb_column.configure(state="disabled") + self.ent_constant.configure(state="normal") + return + + self.cmb_mode.configure(state="readonly") + self.cmb_source.configure(state="readonly") + self.ent_constant.configure(state="disabled") + source_values = [] + for source_id, dataset_name, _columns in self._sources(): + if source_kind == "primary" and source_id == "primary": + source_values.append(source_id) + elif source_kind == "additional" and source_id != "primary": + source_values.append(source_id) + self.cmb_source["values"] = source_values + if self.var_source_id.get() not in source_values: + self.var_source_id.set(source_values[0] if source_values else "") + self.cmb_column.configure(state="readonly") + self._refresh_columns() + + def validate(self): + token = self.var_token.get().strip() + if not token: + messagebox.showerror("Prompt Editor", "Token is required.", parent=self) + return False + if self.var_source_kind.get() != "system": + if not self.var_source_id.get().strip(): + messagebox.showerror("Prompt Editor", "Source is required.", parent=self) + return False + if not self.var_column.get().strip(): + messagebox.showerror("Prompt Editor", "Column is required for non-system bindings.", parent=self) + return False + return True + + def apply(self): + self.result = PlaceholderBinding( + token=self.var_token.get().strip(), + source_kind=self.var_source_kind.get().strip(), + source_id=self.var_source_id.get().strip() or "primary", + column_name=self.var_column.get().strip() or None, + render_mode=self.var_render_mode.get().strip(), + constant_value=self.var_constant.get(), + ) + + +class OutputDialog(simpledialog.Dialog): + def __init__(self, parent, config: PromptTemplateConfig, output_def: OutputDefinition | None = None): + self.config_model = config + self.output_def = output_def + self.result: OutputDefinition | None = None + super().__init__(parent, title="Output Field") + + def body(self, master): + ttk.Label(master, text="Name").grid(row=0, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Source type").grid(row=1, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Field type").grid(row=2, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Required").grid(row=3, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Instructions").grid(row=4, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="Enum source").grid(row=5, column=0, sticky="w", padx=8, pady=6) + ttk.Label(master, text="System value").grid(row=6, column=0, sticky="w", padx=8, pady=6) + + self.var_name = tk.StringVar(value="" if self.output_def is None else self.output_def.name) + self.var_source_type = tk.StringVar(value="model_output" if self.output_def is None else self.output_def.source_type) + self.var_field_type = tk.StringVar(value="text" if self.output_def is None else self.output_def.field_type) + self.var_required = tk.BooleanVar(value=True if self.output_def is None else self.output_def.required) + self.var_instructions = tk.StringVar(value="" if self.output_def is None else self.output_def.instructions) + self.var_enum_source = tk.StringVar( + value="" + if self.output_def is None or not self.output_def.enum_source_id + else f"{self.output_def.enum_source_id}:{self.output_def.enum_source_column or ''}" + ) + self.var_system_value = tk.StringVar(value="" if self.output_def is None else (self.output_def.system_value or "")) + + ttk.Entry(master, textvariable=self.var_name, width=28).grid(row=0, column=1, sticky="ew", padx=8, pady=6) + self.cmb_source_type = ttk.Combobox( + master, + textvariable=self.var_source_type, + values=["model_output", "system"], + state="readonly", + width=25, + ) + self.cmb_source_type.grid(row=1, column=1, sticky="ew", padx=8, pady=6) + self.cmb_field_type = ttk.Combobox( + master, + textvariable=self.var_field_type, + values=["enum", "text", "integer", "boolean", "list[str]"], + state="readonly", + width=25, + ) + self.cmb_field_type.grid(row=2, column=1, sticky="ew", padx=8, pady=6) + ttk.Checkbutton(master, variable=self.var_required).grid(row=3, column=1, sticky="w", padx=8, pady=6) + ttk.Entry(master, textvariable=self.var_instructions, width=28).grid(row=4, column=1, sticky="ew", padx=8, pady=6) + self.cmb_enum_source = ttk.Combobox(master, textvariable=self.var_enum_source, state="readonly", width=25) + self.cmb_enum_source.grid(row=5, column=1, sticky="ew", padx=8, pady=6) + ttk.Entry(master, textvariable=self.var_system_value, width=28).grid(row=6, column=1, sticky="ew", padx=8, pady=6) + + master.columnconfigure(1, weight=1) + self.cmb_source_type.bind("<>", lambda _event: self._refresh_state()) + self.cmb_field_type.bind("<>", lambda _event: self._refresh_state()) + self._refresh_state() + return self.cmb_source_type + + def _refresh_state(self): + enum_values = [] + sources = [self.config_model.primary_source] + self.config_model.additional_sources + for source in sources: + for column in source.available_columns: + enum_values.append(f"{source.source_id}:{column}") + self.cmb_enum_source["values"] = enum_values + + is_system = self.var_source_type.get() == "system" + allows_enum = self.var_field_type.get() in {"enum", "list[str]"} and not is_system + self.cmb_enum_source.configure(state="readonly" if allows_enum else "disabled") + if not allows_enum: + self.var_enum_source.set("") + + def validate(self): + if not self.var_name.get().strip(): + messagebox.showerror("Prompt Editor", "Output name is required.", parent=self) + return False + if self.var_source_type.get() == "system" and not self.var_system_value.get(): + messagebox.showerror("Prompt Editor", "System outputs need a value.", parent=self) + return False + if self.var_field_type.get() == "enum" and not self.var_enum_source.get(): + messagebox.showerror("Prompt Editor", "Enum outputs need an enum source.", parent=self) + return False + return True + + def apply(self): + enum_source_id = None + enum_source_column = None + if self.var_enum_source.get(): + enum_source_id, enum_source_column = self.var_enum_source.get().split(":", 1) + + self.result = OutputDefinition( + name=self.var_name.get().strip(), + source_type=self.var_source_type.get().strip(), + field_type=self.var_field_type.get().strip(), + required=self.var_required.get(), + instructions=self.var_instructions.get().strip(), + enum_source_id=enum_source_id, + enum_source_column=enum_source_column or None, + system_value=self.var_system_value.get() if self.var_source_type.get() == "system" else None, + ) + + +class PromptEditorWindow(tk.Toplevel): + def __init__(self, parent: tk.Misc, initial_config: PromptTemplateConfig | None = None): + super().__init__(parent) + self.title("Prompt Editor") + self.transient(parent) + self.geometry("1200x860") + + self.current_config = initial_config.model_copy(deep=True) if initial_config else PromptTemplateConfig( + template_name="Untitled Prompt", + model=get_setting("model", "gpt-4o-mini"), + user_prompt_template="", + ) + self._last_prompt_widget = None + + self._build_ui() + self._apply_config_to_ui(self.current_config) + + def _build_ui(self): + root = ttk.Frame(self, padding=12) + root.grid(row=0, column=0, sticky="nsew") + self.columnconfigure(0, weight=1) + self.rowconfigure(0, weight=1) + root.columnconfigure(0, weight=1) + root.rowconfigure(1, weight=1) + + top = ttk.Frame(root) + top.grid(row=0, column=0, sticky="ew", pady=(0, 8)) + for column in range(8): + top.columnconfigure(column, weight=1 if column in {1, 4} else 0) + + ttk.Label(top, text="Template").grid(row=0, column=0, sticky="w", padx=(0, 6)) + self.var_template_name = tk.StringVar() + ttk.Entry(top, textvariable=self.var_template_name, width=28).grid(row=0, column=1, sticky="ew", padx=(0, 8)) + + ttk.Label(top, text="Presets").grid(row=0, column=2, sticky="w", padx=(0, 6)) + self.var_preset = tk.StringVar() + self.cmb_preset = ttk.Combobox( + top, + textvariable=self.var_preset, + values=[label for _preset_id, label in list_presets()], + state="readonly", + width=30, + ) + self.cmb_preset.grid(row=0, column=3, sticky="ew", padx=(0, 6)) + ttk.Button(top, text="Load Preset", command=self._load_selected_preset).grid(row=0, column=4, sticky="w", padx=(0, 12)) + + ttk.Label(top, text="Saved").grid(row=0, column=5, sticky="w", padx=(0, 6)) + self.var_saved_template = tk.StringVar() + self.cmb_saved_templates = ttk.Combobox(top, textvariable=self.var_saved_template, state="readonly", width=28) + self.cmb_saved_templates.grid(row=0, column=6, sticky="ew", padx=(0, 6)) + ttk.Button(top, text="Load", command=self._load_selected_template).grid(row=0, column=7, sticky="w") + + second = ttk.Frame(root) + second.grid(row=2, column=0, sticky="ew", pady=(8, 0)) + ttk.Button(second, text="Save Template", command=self._save_template).pack(side="left") + ttk.Button(second, text="Delete Template", command=self._delete_template).pack(side="left", padx=6) + ttk.Button(second, text="Close", command=self.destroy).pack(side="right") + + self.notebook = ttk.Notebook(root) + self.notebook.grid(row=1, column=0, sticky="nsew") + + self.sources_tab = ttk.Frame(self.notebook, padding=10) + self.prompt_tab = ttk.Frame(self.notebook, padding=10) + self.outputs_tab = ttk.Frame(self.notebook, padding=10) + self.review_tab = ttk.Frame(self.notebook, padding=10) + self.notebook.add(self.sources_tab, text="Sources") + self.notebook.add(self.prompt_tab, text="Prompt") + self.notebook.add(self.outputs_tab, text="Outputs") + self.notebook.add(self.review_tab, text="Review & Run") + + self._build_sources_tab() + self._build_prompt_tab() + self._build_outputs_tab() + self._build_review_tab() + self._refresh_templates_menu() + + def _build_sources_tab(self): + self.sources_tab.columnconfigure(0, weight=1) + self.sources_tab.columnconfigure(1, weight=1) + self.sources_tab.rowconfigure(2, weight=1) + + primary_frame = ttk.LabelFrame(self.sources_tab, text="Primary source") + primary_frame.grid(row=0, column=0, columnspan=2, sticky="ew", pady=(0, 10)) + primary_frame.columnconfigure(1, weight=1) + + ttk.Button(primary_frame, text="Import primary dataset", command=self._import_primary_source).grid( + row=0, column=0, padx=8, pady=8, sticky="w" + ) + ttk.Button(primary_frame, text="Reload file", command=self._reload_primary_source).grid( + row=0, column=1, padx=8, pady=8, sticky="w" + ) + + self.lbl_primary_summary = ttk.Label(primary_frame, text="No primary dataset loaded.") + self.lbl_primary_summary.grid(row=1, column=0, columnspan=2, sticky="w", padx=8, pady=(0, 8)) + + ttk.Label(primary_frame, text="Passthrough columns").grid(row=2, column=0, sticky="nw", padx=8, pady=4) + self.lst_passthrough = tk.Listbox(primary_frame, selectmode="extended", exportselection=False, height=7) + self.lst_passthrough.grid(row=2, column=1, sticky="ew", padx=8, pady=4) + ttk.Button(primary_frame, text="Apply passthrough selection", command=self._apply_passthrough_columns).grid( + row=3, column=1, sticky="w", padx=8, pady=(0, 8) + ) + + addl_frame = ttk.LabelFrame(self.sources_tab, text="Additional sources") + addl_frame.grid(row=1, column=0, sticky="nsew", padx=(0, 8)) + addl_frame.columnconfigure(0, weight=1) + addl_frame.rowconfigure(0, weight=1) + self.tree_sources = ttk.Treeview( + addl_frame, + columns=("source_id", "dataset", "column", "mode", "rows"), + show="headings", + height=10, + ) + for column, heading in { + "source_id": "Source ID", + "dataset": "Dataset", + "column": "Column", + "mode": "Mode", + "rows": "Rows", + }.items(): + self.tree_sources.heading(column, text=heading) + self.tree_sources.column(column, width=120, anchor="w") + self.tree_sources.grid(row=0, column=0, sticky="nsew", padx=8, pady=8) + source_buttons = ttk.Frame(addl_frame) + source_buttons.grid(row=1, column=0, sticky="w", padx=8, pady=(0, 8)) + ttk.Button(source_buttons, text="Add source", command=self._add_additional_source).pack(side="left") + ttk.Button(source_buttons, text="Reload source", command=self._reload_selected_source).pack(side="left", padx=6) + ttk.Button(source_buttons, text="Remove source", command=self._remove_selected_source).pack(side="left") + + placeholder_frame = ttk.LabelFrame(self.sources_tab, text="Placeholder bindings") + placeholder_frame.grid(row=1, column=1, rowspan=2, sticky="nsew") + placeholder_frame.columnconfigure(0, weight=1) + placeholder_frame.rowconfigure(0, weight=1) + self.tree_bindings = ttk.Treeview( + placeholder_frame, + columns=("token", "kind", "source", "column", "mode"), + show="headings", + height=18, + ) + for column, heading in { + "token": "Token", + "kind": "Kind", + "source": "Source", + "column": "Column", + "mode": "Mode", + }.items(): + self.tree_bindings.heading(column, text=heading) + self.tree_bindings.column(column, width=110, anchor="w") + self.tree_bindings.grid(row=0, column=0, sticky="nsew", padx=8, pady=8) + binding_buttons = ttk.Frame(placeholder_frame) + binding_buttons.grid(row=1, column=0, sticky="w", padx=8, pady=(0, 8)) + ttk.Button(binding_buttons, text="Add binding", command=self._add_binding).pack(side="left") + ttk.Button(binding_buttons, text="Edit binding", command=self._edit_binding).pack(side="left", padx=6) + ttk.Button(binding_buttons, text="Remove binding", command=self._remove_binding).pack(side="left") + + def _build_prompt_tab(self): + self.prompt_tab.columnconfigure(0, weight=1) + self.prompt_tab.rowconfigure(1, weight=1) + self.prompt_tab.rowconfigure(3, weight=1) + + ttk.Label(self.prompt_tab, text="System prompt").grid(row=0, column=0, sticky="w") + self.txt_system = tk.Text(self.prompt_tab, height=10, wrap="word") + self.txt_system.grid(row=1, column=0, sticky="nsew", pady=(4, 6)) + ttk.Button(self.prompt_tab, text="Insert Placeholder", command=lambda: self._insert_placeholder(self.txt_system)).grid( + row=1, column=1, sticky="n", padx=(8, 0) + ) + + ttk.Label(self.prompt_tab, text="User prompt template").grid(row=2, column=0, sticky="w", pady=(8, 0)) + self.txt_user = tk.Text(self.prompt_tab, height=14, wrap="word") + self.txt_user.grid(row=3, column=0, sticky="nsew", pady=(4, 6)) + ttk.Button(self.prompt_tab, text="Insert Placeholder", command=lambda: self._insert_placeholder(self.txt_user)).grid( + row=3, column=1, sticky="n", padx=(8, 0) + ) + + def _build_outputs_tab(self): + self.outputs_tab.columnconfigure(0, weight=1) + self.outputs_tab.rowconfigure(0, weight=1) + self.tree_outputs = ttk.Treeview( + self.outputs_tab, + columns=("name", "source", "field_type", "required", "details"), + show="headings", + height=18, + ) + for column, heading in { + "name": "Name", + "source": "Source type", + "field_type": "Field type", + "required": "Required", + "details": "Details", + }.items(): + self.tree_outputs.heading(column, text=heading) + self.tree_outputs.column(column, width=160, anchor="w") + self.tree_outputs.grid(row=0, column=0, sticky="nsew") + + buttons = ttk.Frame(self.outputs_tab) + buttons.grid(row=1, column=0, sticky="w", pady=(8, 0)) + ttk.Button(buttons, text="Add output", command=self._add_output).pack(side="left") + ttk.Button(buttons, text="Edit output", command=self._edit_output).pack(side="left", padx=6) + ttk.Button(buttons, text="Remove output", command=self._remove_output).pack(side="left") + + def _build_review_tab(self): + self.review_tab.columnconfigure(0, weight=1) + self.review_tab.rowconfigure(1, weight=1) + + controls = ttk.Frame(self.review_tab) + controls.grid(row=0, column=0, sticky="ew") + for column in range(8): + controls.columnconfigure(column, weight=1 if column in {1, 3, 5} else 0) + + ttk.Label(controls, text="Model").grid(row=0, column=0, sticky="w", padx=(0, 6)) + self.var_model = tk.StringVar() + self.cmb_model = ttk.Combobox(controls, textvariable=self.var_model, values=get_models(), state="readonly") + self.cmb_model.grid(row=0, column=1, sticky="ew", padx=(0, 8)) + ttk.Button(controls, text="↻", width=3, command=self._refresh_models).grid(row=0, column=2, sticky="w", padx=(0, 8)) + + ttk.Label(controls, text="Default mode").grid(row=0, column=3, sticky="w", padx=(0, 6)) + self.var_execution_mode = tk.StringVar(value="live") + mode_frame = ttk.Frame(controls) + mode_frame.grid(row=0, column=4, sticky="w", padx=(0, 8)) + ttk.Radiobutton(mode_frame, text="Live", value="live", variable=self.var_execution_mode).pack(side="left") + ttk.Radiobutton(mode_frame, text="Batch", value="batch", variable=self.var_execution_mode).pack(side="left", padx=(6, 0)) + + advanced = ttk.LabelFrame(self.review_tab, text="Advanced") + advanced.grid(row=2, column=0, sticky="ew", pady=(8, 8)) + for column in range(8): + advanced.columnconfigure(column, weight=1 if column % 2 == 1 else 0) + + ttk.Label(advanced, text="Temperature").grid(row=0, column=0, sticky="w", padx=8, pady=6) + self.var_temperature = tk.StringVar() + ttk.Entry(advanced, textvariable=self.var_temperature, width=10).grid(row=0, column=1, sticky="w", padx=8, pady=6) + ttk.Label(advanced, text="Max output tokens").grid(row=0, column=2, sticky="w", padx=8, pady=6) + self.var_max_output_tokens = tk.StringVar() + ttk.Entry(advanced, textvariable=self.var_max_output_tokens, width=10).grid(row=0, column=3, sticky="w", padx=8, pady=6) + ttk.Label(advanced, text="Reasoning").grid(row=0, column=4, sticky="w", padx=8, pady=6) + self.var_reasoning = tk.StringVar() + ttk.Combobox(advanced, textvariable=self.var_reasoning, values=REASONING_OPTIONS, state="readonly", width=12).grid( + row=0, column=5, sticky="w", padx=8, pady=6 + ) + ttk.Label(advanced, text="Detail level").grid(row=0, column=6, sticky="w", padx=8, pady=6) + self.var_detail = tk.StringVar() + ttk.Combobox(advanced, textvariable=self.var_detail, values=DETAIL_OPTIONS, state="readonly", width=12).grid( + row=0, column=7, sticky="w", padx=8, pady=6 + ) + + ttk.Label(self.review_tab, text="Preview / test output").grid(row=3, column=0, sticky="w") + self.txt_preview = tk.Text(self.review_tab, height=20, wrap="word") + self.txt_preview.grid(row=1, column=0, sticky="nsew", pady=(8, 0)) + + buttons = ttk.Frame(self.review_tab) + buttons.grid(row=4, column=0, sticky="ew", pady=(8, 0)) + ttk.Button(buttons, text="Render Preview", command=self._render_preview).pack(side="left") + ttk.Button(buttons, text="Test First Row", command=self._run_test_row).pack(side="left", padx=6) + ttk.Button(buttons, text="Run Live", command=self._run_live).pack(side="left", padx=6) + ttk.Button(buttons, text="Run Batch", command=self._run_batch).pack(side="left", padx=6) + + def _refresh_templates_menu(self): + template_names = [template.template_name for template in list_templates()] + self.cmb_saved_templates["values"] = template_names + + def _apply_config_to_ui(self, config: PromptTemplateConfig): + self.current_config = config.model_copy(deep=True) + self.var_template_name.set(self.current_config.template_name) + self.var_model.set(self.current_config.model) + self.var_execution_mode.set(self.current_config.execution_mode) + self.var_temperature.set("" if self.current_config.advanced.temperature is None else str(self.current_config.advanced.temperature)) + self.var_max_output_tokens.set( + "" if self.current_config.advanced.max_output_tokens is None else str(self.current_config.advanced.max_output_tokens) + ) + self.var_reasoning.set(self.current_config.advanced.reasoning_effort or "") + self.var_detail.set(self.current_config.advanced.detail_level or "") + + self.txt_system.delete("1.0", "end") + self.txt_system.insert("1.0", self.current_config.system_prompt) + self.txt_user.delete("1.0", "end") + self.txt_user.insert("1.0", self.current_config.user_prompt_template) + + self._refresh_primary_source_ui() + self._refresh_sources_tree() + self._refresh_bindings_tree() + self._sync_passthrough_outputs() + self._refresh_outputs_tree() + self._refresh_templates_menu() + + def _config_from_ui(self) -> PromptTemplateConfig: + temperature = self.var_temperature.get().strip() + max_output_tokens = self.var_max_output_tokens.get().strip() + config = self.current_config.model_copy(deep=True) + config.template_name = self.var_template_name.get().strip() or "Untitled Prompt" + config.model = self.var_model.get().strip() or get_setting("model", "gpt-4o-mini") + config.execution_mode = self.var_execution_mode.get().strip() or "live" + config.system_prompt = self.txt_system.get("1.0", "end-1c") + config.user_prompt_template = self.txt_user.get("1.0", "end-1c") + config.advanced.temperature = float(temperature) if temperature else None + config.advanced.max_output_tokens = int(max_output_tokens) if max_output_tokens else None + config.advanced.reasoning_effort = self.var_reasoning.get().strip() or None + config.advanced.detail_level = self.var_detail.get().strip() or None + return config + + def _refresh_primary_source_ui(self): + primary = self.current_config.primary_source + if not primary.rows: + self.lbl_primary_summary.configure(text="No primary dataset loaded.") + self.lst_passthrough.delete(0, "end") + return + + self.lbl_primary_summary.configure( + text=( + f"{primary.dataset_name} | rows: {len(primary.rows)} | " + f"prompt column: {primary.selected_column}" + ) + ) + self.lst_passthrough.delete(0, "end") + for column in primary.available_columns: + self.lst_passthrough.insert("end", column) + if column in primary.passthrough_columns: + self.lst_passthrough.selection_set("end") + + def _refresh_sources_tree(self): + self.tree_sources.delete(*self.tree_sources.get_children()) + for index, source in enumerate(self.current_config.additional_sources): + self.tree_sources.insert( + "", + "end", + iid=str(index), + values=(source.source_id, source.dataset_name, source.selected_column, source.render_mode, len(source.rows)), + ) + + def _refresh_bindings_tree(self): + self.tree_bindings.delete(*self.tree_bindings.get_children()) + for index, binding in enumerate(self.current_config.placeholder_bindings): + self.tree_bindings.insert( + "", + "end", + iid=str(index), + values=(binding.token, binding.source_kind, binding.source_id, binding.column_name or "", binding.render_mode), + ) + + def _refresh_outputs_tree(self): + self.tree_outputs.delete(*self.tree_outputs.get_children()) + for index, output_def in enumerate(self.current_config.output_definitions): + detail = output_def.enum_source_id or output_def.passthrough_column or output_def.system_value or "" + self.tree_outputs.insert( + "", + "end", + iid=str(index), + values=(output_def.name, output_def.source_type, output_def.field_type, "yes" if output_def.required else "no", detail), + ) + + def _sync_passthrough_outputs(self): + manual_outputs = [output for output in self.current_config.output_definitions if output.source_type != "passthrough"] + passthrough_outputs = [ + OutputDefinition( + name=column, + source_type="passthrough", + field_type="text", + required=True, + passthrough_column=column, + ) + for column in self.current_config.primary_source.passthrough_columns + ] + self.current_config.output_definitions = manual_outputs + passthrough_outputs + + def _load_selected_preset(self): + label = self.var_preset.get().strip() + if not label: + return + preset_lookup = {display: preset_id for preset_id, display in list_presets()} + preset_id = preset_lookup[label] + self._apply_config_to_ui(get_preset(preset_id)) + + def _load_selected_template(self): + template_name = self.var_saved_template.get().strip() + if not template_name: + return + template = load_template(template_name) + if template is None: + messagebox.showerror("Prompt Editor", "Saved template could not be loaded.", parent=self) + return + self._apply_config_to_ui(template) + + def _save_template(self): + try: + config = self._config_from_ui() + save_template(config) + self.current_config = config + self._refresh_templates_menu() + self.var_saved_template.set(config.template_name) + messagebox.showinfo("Prompt Editor", "Template saved.", parent=self) + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + def _delete_template(self): + template_name = self.var_template_name.get().strip() + if not template_name: + return + delete_template(template_name) + self._refresh_templates_menu() + messagebox.showinfo("Prompt Editor", "Template deleted.", parent=self) + + def _import_primary_source(self): + imported = import_tabular_data(self, "Select the primary row dataset") + if imported is None: + return + self.current_config.primary_source = build_primary_source(imported) + if not any(binding.source_id == "primary" for binding in self.current_config.placeholder_bindings): + self.current_config.placeholder_bindings.append( + PlaceholderBinding( + token=make_source_id(imported.selected_column_name) or "quote", + source_id="primary", + source_kind="primary", + column_name=imported.selected_column_name, + render_mode="row_value", + ) + ) + else: + for binding in self.current_config.placeholder_bindings: + if binding.source_id == "primary" and binding.source_kind == "primary" and not binding.column_name: + binding.column_name = imported.selected_column_name + self._refresh_primary_source_ui() + self._sync_passthrough_outputs() + self._refresh_outputs_tree() + self._refresh_bindings_tree() + + def _reload_primary_source(self): + primary = self.current_config.primary_source + if not primary.file_path: + return + imported = import_tabular_data(self, "Reload primary row dataset") + if imported is None: + return + self.current_config.primary_source = build_primary_source(imported, passthrough_columns=primary.passthrough_columns) + self._refresh_primary_source_ui() + + def _apply_passthrough_columns(self): + selected = [self.lst_passthrough.get(index) for index in self.lst_passthrough.curselection()] + self.current_config.primary_source.passthrough_columns = selected + self._sync_passthrough_outputs() + self._refresh_outputs_tree() + + def _add_additional_source(self): + imported = import_tabular_data(self, "Select an additional source") + if imported is None: + return + source_id = make_source_id(imported.dataset_name) + existing_ids = {source.source_id for source in self.current_config.additional_sources} + suffix = 2 + base_id = source_id + while source_id in existing_ids or source_id == "primary": + source_id = f"{base_id}-{suffix}" + suffix += 1 + source = build_additional_source(imported, source_id=source_id) + self.current_config.additional_sources.append(source) + if not any(binding.source_id == source.source_id for binding in self.current_config.placeholder_bindings): + self.current_config.placeholder_bindings.append( + PlaceholderBinding( + token=source.source_id.replace("-", "_"), + source_id=source.source_id, + source_kind="additional", + column_name=source.selected_column, + render_mode=source.render_mode, + ) + ) + self._refresh_sources_tree() + self._refresh_bindings_tree() + + def _reload_selected_source(self): + selection = self.tree_sources.selection() + if not selection: + return + index = int(selection[0]) + imported = import_tabular_data(self, "Reload additional source") + if imported is None: + return + existing = self.current_config.additional_sources[index] + self.current_config.additional_sources[index] = build_additional_source( + imported, + source_id=existing.source_id, + render_mode=existing.render_mode, + ) + self._refresh_sources_tree() + self._refresh_bindings_tree() + + def _remove_selected_source(self): + selection = self.tree_sources.selection() + if not selection: + return + index = int(selection[0]) + removed = self.current_config.additional_sources.pop(index) + self.current_config.placeholder_bindings = [ + binding for binding in self.current_config.placeholder_bindings if binding.source_id != removed.source_id + ] + self._refresh_sources_tree() + self._refresh_bindings_tree() + + def _add_binding(self): + dialog = PlaceholderDialog(self, self._config_from_ui()) + if dialog.result is None: + return + self.current_config.placeholder_bindings.append(dialog.result) + self._refresh_bindings_tree() + + def _edit_binding(self): + selection = self.tree_bindings.selection() + if not selection: + return + index = int(selection[0]) + dialog = PlaceholderDialog(self, self._config_from_ui(), self.current_config.placeholder_bindings[index]) + if dialog.result is None: + return + self.current_config.placeholder_bindings[index] = dialog.result + self._refresh_bindings_tree() + + def _remove_binding(self): + selection = self.tree_bindings.selection() + if not selection: + return + self.current_config.placeholder_bindings.pop(int(selection[0])) + self._refresh_bindings_tree() + + def _add_output(self): + dialog = OutputDialog(self, self._config_from_ui()) + if dialog.result is None: + return + self.current_config.output_definitions.append(dialog.result) + self._sync_passthrough_outputs() + self._refresh_outputs_tree() + + def _edit_output(self): + selection = self.tree_outputs.selection() + if not selection: + return + index = int(selection[0]) + output_def = self.current_config.output_definitions[index] + if output_def.source_type == "passthrough": + messagebox.showinfo( + "Prompt Editor", + "Passthrough outputs are managed from the passthrough column selection in Sources.", + parent=self, + ) + return + dialog = OutputDialog(self, self._config_from_ui(), output_def) + if dialog.result is None: + return + self.current_config.output_definitions[index] = dialog.result + self._sync_passthrough_outputs() + self._refresh_outputs_tree() + + def _remove_output(self): + selection = self.tree_outputs.selection() + if not selection: + return + index = int(selection[0]) + if self.current_config.output_definitions[index].source_type == "passthrough": + messagebox.showinfo( + "Prompt Editor", + "Remove passthrough outputs by changing the passthrough column selection in Sources.", + parent=self, + ) + return + self.current_config.output_definitions.pop(index) + self._refresh_outputs_tree() + + def _insert_placeholder(self, target_widget: tk.Text): + tokens = list_available_tokens(self.current_config) + if not tokens: + messagebox.showinfo("Prompt Editor", "Add a placeholder binding first.", parent=self) + return + token = simpledialog.askstring( + "Insert Placeholder", + "Available tokens:\n" + "\n".join(f"<{token_name}>" for token_name in tokens) + "\n\nType the token name to insert:", + parent=self, + ) + if not token: + return + token = token.strip().strip("<>") + if token not in tokens: + messagebox.showerror("Prompt Editor", f"<{token}> is not a known placeholder.", parent=self) + return + target_widget.insert("insert", f"<{token}>") + + def _refresh_models(self): + try: + models = refresh_models() + self.cmb_model["values"] = models + if models and self.var_model.get() not in models: + self.var_model.set(models[0]) + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + def _write_preview(self, text: str): + self.txt_preview.delete("1.0", "end") + self.txt_preview.insert("1.0", text) + + def _render_preview(self): + try: + config = self._config_from_ui() + system_prompt, user_prompt = render_prompts_for_row(config) + preview = f"System prompt:\n{system_prompt or '(none)'}\n\nUser prompt:\n{user_prompt}" + self._write_preview(preview) + self.notebook.select(self.review_tab) + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + def _run_test_row(self): + try: + config = self._config_from_ui() + result = run_single_test_row(config) + self._write_preview(json.dumps(result, indent=2, ensure_ascii=False)) + self.notebook.select(self.review_tab) + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + def _run_live(self): + try: + config = self._config_from_ui() + row_count = len(config.primary_source.rows) + if row_count > config.live_row_warning_threshold: + proceed = messagebox.askyesno( + "Prompt Editor", + ( + f"This live run will process {row_count} rows. " + f"Batch mode is recommended above {config.live_row_warning_threshold} rows.\n\n" + "Run live anyway?" + ), + parent=self, + ) + if not proceed: + return + df = run_live_prompt(config, parent=self) + save_as_csv(df) + self._write_preview(f"Live run complete.\nRows exported: {len(df)}") + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + def _run_batch(self): + try: + config = self._config_from_ui() + batch = submit_batch_prompt(config) + if hasattr(self.master, "tree_ongoing"): + refresh_batches_async(self.master) + self._write_preview(f"Batch submitted.\nBatch ID: {batch.id}") + messagebox.showinfo("Prompt Editor", "Batch submitted.", parent=self) + except Exception as exc: + messagebox.showerror("Prompt Editor", str(exc), parent=self) + + +def open_prompt_editor(parent: tk.Misc, preset_id: str | None = None, execution_mode: str | None = None): + config = get_preset(preset_id) if preset_id else None + if config is not None and execution_mode: + config.execution_mode = execution_mode + window = PromptEditorWindow(parent, initial_config=config) + window.grab_set() + return window diff --git a/tests/test_prompt_editor.py b/tests/test_prompt_editor.py new file mode 100644 index 0000000..c28ca96 --- /dev/null +++ b/tests/test_prompt_editor.py @@ -0,0 +1,143 @@ +import json + +import pytest +from pydantic import ValidationError + +from prompt_editor.config import ( + AdditionalSourceDefinition, + OutputDefinition, + PlaceholderBinding, + PrimarySourceDefinition, + PromptTemplateConfig, +) +from prompt_editor.engine import build_response_model, render_prompts_for_row +from prompt_editor.presets import multi_label_preset, single_label_preset +from prompt_editor.storage import load_template, save_template + + +@pytest.fixture +def prompt_editor_config_dir(tmp_path, monkeypatch): + config_root = tmp_path / "config-root" + config_root.mkdir() + monkeypatch.setattr("prompt_editor.storage.ensure_user_config_dir", lambda: config_root) + return config_root + + +def _base_config() -> PromptTemplateConfig: + return PromptTemplateConfig( + template_name="Code quotes", + system_prompt="Use .", + user_prompt_template="Classify using .", + primary_source=PrimarySourceDefinition( + dataset_name="quotes", + selected_column="quote", + available_columns=["quote", "speaker"], + passthrough_columns=["quote"], + rows=[{"quote": "A hopeful excerpt", "speaker": "R1"}], + ), + additional_sources=[ + AdditionalSourceDefinition( + source_id="labels", + dataset_name="labels", + selected_column="label", + available_columns=["label"], + rows=[{"label": "positive"}, {"label": "negative"}, {"label": "positive"}], + ) + ], + placeholder_bindings=[ + PlaceholderBinding(token="quote", source_id="primary", source_kind="primary", column_name="quote"), + PlaceholderBinding( + token="labels", + source_id="labels", + source_kind="additional", + column_name="label", + render_mode="unique_values", + ), + ], + output_definitions=[ + OutputDefinition( + name="label", + source_type="model_output", + field_type="enum", + enum_source_id="labels", + enum_source_column="label", + ), + OutputDefinition( + name="quote", + source_type="passthrough", + field_type="text", + passthrough_column="quote", + ), + ], + ) + + +class TestPromptTemplatePersistence: + def test_template_roundtrip_strips_runtime_rows(self, prompt_editor_config_dir): + config = _base_config() + + save_template(config) + + loaded = load_template("Code quotes") + assert loaded is not None + assert loaded.template_name == "Code quotes" + assert loaded.primary_source.rows == [] + assert loaded.additional_sources[0].rows == [] + + raw = json.loads( + (prompt_editor_config_dir / "prompt_editor").rglob("templates.json").__next__().read_text(encoding="utf-8") + ) + assert raw[0]["primary_source"]["rows"] == [] + + +class TestPromptRendering: + def test_placeholder_rendering_uses_unique_values_for_additional_sources(self): + system_prompt, user_prompt = render_prompts_for_row(_base_config()) + + assert "positive\nnegative" in system_prompt + assert "A hopeful excerpt" in user_prompt + assert user_prompt.count("positive") == 1 + + +class TestSchemaGeneration: + def test_multi_label_schema_is_array_of_allowed_enum_values(self): + config = multi_label_preset() + config.primary_source.available_columns = ["quote"] + config.primary_source.selected_column = "quote" + config.primary_source.rows = [{"quote": "A quote"}] + config.primary_source.passthrough_columns = ["quote"] + config.additional_sources[0].available_columns = ["label"] + config.additional_sources[0].selected_column = "label" + config.additional_sources[0].rows = [{"label": "a"}, {"label": "b"}] + config.output_definitions.append( + OutputDefinition(name="quote", source_type="passthrough", field_type="text", passthrough_column="quote") + ) + + response_model, schema, _ = build_response_model(config) + + response_model.model_validate({"label": ["a", "b"]}) + with pytest.raises(ValidationError): + response_model.model_validate({"label": ["a", "other"]}) + assert schema["additionalProperties"] is False + + +class TestStrictSingleLabelBehavior: + def test_single_label_preset_accepts_only_allowed_values(self): + config = single_label_preset() + config.primary_source.available_columns = ["quote"] + config.primary_source.selected_column = "quote" + config.primary_source.rows = [{"quote": "A quote"}] + config.primary_source.passthrough_columns = ["quote"] + config.additional_sources[0].available_columns = ["label"] + config.additional_sources[0].selected_column = "label" + config.additional_sources[0].rows = [{"label": "positive"}, {"label": "negative"}] + config.output_definitions.append( + OutputDefinition(name="quote", source_type="passthrough", field_type="text", passthrough_column="quote") + ) + + response_model, _schema, _ = build_response_model(config) + + valid = response_model.model_validate({"label": "positive"}) + assert valid.model_dump()["label"] == "positive" + with pytest.raises(ValidationError): + response_model.model_validate({"label": "neutral"}) diff --git a/ui/main_window.py b/ui/main_window.py index 9f898d4..a0a0d1f 100644 --- a/ui/main_window.py +++ b/ui/main_window.py @@ -2,18 +2,12 @@ Main window GUI for CodebookAI text classification application. This module provides the primary user interface for the CodebookAI application, -including batch job management, live processing controls, and settings access. +including batch job management, prompt-editor controls, and settings access. The interface displays ongoing and completed batch jobs in tabbed tables. Updated per request: - Removed add_btn, tools_btn, and settings_btn (refresh button retained). -- Added a top menu bar with the following structure: - File > Settings, Exit - Data Prep > Sample - LLM Tools > Live Methods > Single Label Text Classification, Multi-Label Text Classification, Text Extraction - > Batch Methods > Single Label Text Classification - Data Analysis > Reliability Statistics - Help > Github Repo +- Added a top menu bar with prompt-editor access and built-in presets. - Added a "Batches" title above the table area at the bottom of the page. """ @@ -27,6 +21,12 @@ from live_processing.correlogram import open_correlogram_wizard from live_processing.reliability_calculator import open_reliability_wizard from live_processing.sampler import sample_data +from prompt_editor.presets import ( + KEYWORD_EXTRACTION_PRESET_ID, + MULTI_LABEL_PRESET_ID, + SINGLE_LABEL_PRESET_ID, +) +from prompt_editor.ui import open_prompt_editor # Handle imports based on how the script is run try: @@ -35,7 +35,6 @@ from ui_utils import center_window from ui_helpers import make_tab_with_tree, popup_menu, popup_menu_below_widget from batch_operations import ( - call_batch_async, refresh_batches_async, call_batch_download_async, cancel_batch_async, @@ -46,26 +45,11 @@ from ui.ui_utils import center_window from ui.ui_helpers import make_tab_with_tree, popup_menu, popup_menu_below_widget from ui.batch_operations import ( - call_batch_async, refresh_batches_async, call_batch_download_async, cancel_batch_async, ) -# Ensure live modules can be imported when run directly -try: - import live_processing.multi_label_live - import live_processing.single_label_live - import live_processing.keyword_extraction_live -except ImportError: - import sys - import os - - sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - import live_processing.multi_label_live - import live_processing.single_label_live - import live_processing.keyword_extraction_live - APP_TITLE = "CodebookAI" APP_SUBTITLE = "A qualitative research tool based on OpenAI's Playground API." WINDOW_SIZE = (1000, 620) # width, height @@ -135,46 +119,24 @@ def build_ui(root: tk.Tk) -> None: data_prep_menu.add_command(label="Sample", command=lambda: sample_data(root)) menubar.add_cascade(label="Data Prep", menu=data_prep_menu) - # LLM Tools > Live Methods / Batch Methods + # LLM Tools llm_tools_menu = tk.Menu(menubar, tearoff=False) + llm_tools_menu.add_command(label="Prompt Editor", command=lambda: open_prompt_editor(root)) - # Live Methods submenu - live_methods_menu = tk.Menu(llm_tools_menu, tearoff=False) - - def _single_label_live_call(): - live_processing.single_label_live.single_label_pipeline(root) - - def _multi_label_live_call(): - live_processing.multi_label_live.multi_label_pipeline(root) - - def _keyword_extraction_live_call(): - live_processing.keyword_extraction_live.keyword_extraction_pipeline(root) - - live_methods_menu.add_command( - label="Single Label Text Classification", command=_single_label_live_call - ) - live_methods_menu.add_command( - label="Multi-Label Text Classification", command=_multi_label_live_call + presets_menu = tk.Menu(llm_tools_menu, tearoff=False) + presets_menu.add_command( + label="Single Label Classification", + command=lambda: open_prompt_editor(root, preset_id=SINGLE_LABEL_PRESET_ID), ) - live_methods_menu.add_command(label="Keyword Extraction", command=_keyword_extraction_live_call) - - # Batch Methods submenu - batch_methods_menu = tk.Menu(llm_tools_menu, tearoff=False) - batch_methods_menu.add_command( - label="Single Label Text Classification", - command=lambda: call_batch_async(root, type="single_label"), + presets_menu.add_command( + label="Multi-Label Classification", + command=lambda: open_prompt_editor(root, preset_id=MULTI_LABEL_PRESET_ID), ) - batch_methods_menu.add_command( - label="Multi-Label Text Classification", - command=lambda: call_batch_async(root, type="multi_label"), - ) - batch_methods_menu.add_command( + presets_menu.add_command( label="Keyword Extraction", - command=lambda: call_batch_async(root, type="keyword_extraction"), + command=lambda: open_prompt_editor(root, preset_id=KEYWORD_EXTRACTION_PRESET_ID), ) - - llm_tools_menu.add_cascade(label="Live Methods", menu=live_methods_menu) - llm_tools_menu.add_cascade(label="Batch Methods", menu=batch_methods_menu) + llm_tools_menu.add_cascade(label="Built-in Presets", menu=presets_menu) menubar.add_cascade(label="LLM Tools", menu=llm_tools_menu) # Data Analysis