config and example fixes backend

Author: kazmer97

lib/idp_common_pkg/idp_common/classification/service.py

@@ -44,6 +44,7 @@
 )
 from idp_common.models import Document, Section, Status
 from idp_common.utils import extract_json_from_text, extract_structured_data_from_text
+from idp_common.utils.few_shot_example_builder import build_few_shot_examples_content
 
 logger = logging.getLogger(__name__)
 
@@ -865,7 +866,7 @@ def _build_content(
         content.extend(before_examples_content)
 
         # Add few-shot examples from config
-        examples_content = self._build_few_shot_examples_content()
+        examples_content = build_few_shot_examples_content(self.config)
         content.extend(examples_content)
 
         # Add the part after examples
@@ -875,139 +876,6 @@ def _build_content(
 
         return content
 
-    def _build_few_shot_examples_content(self) -> List[Dict[str, Any]]:
-        """
-        Build content items for few-shot examples from the configuration.
-
-        Returns:
-            List of content items containing text and image content for examples
-        """
-        content = []
-        classes = self.config.classes or []
-
-        for schema in classes:
-            classification_meta = schema.get(X_AWS_IDP_CLASSIFICATION, {})
-            examples = classification_meta.get("examples", [])
-            for example in examples:
-                class_prompt = example.get("classPrompt")
-
-                # Only process this example if it has a non-empty class_prompt
-                if not class_prompt or not class_prompt.strip():
-                    logger.info(
-                        f"Skipping example with empty classPrompt: {example.get('name')}"
-                    )
-                    continue
-
-                content.append({"text": class_prompt})
-
-                image_path = example.get("imagePath")
-                if image_path:
-                    try:
-                        # Load image content from the path
-
-                        from idp_common import image, s3
-
-                        # Get list of image files from the path (supports directories/prefixes)
-                        image_files = self._get_image_files_from_path(image_path)
-
-                        # Process each image file
-                        for image_file_path in image_files:
-                            try:
-                                # Load image content
-                                if image_file_path.startswith("s3://"):
-                                    # Direct S3 URI
-                                    image_content = s3.get_binary_content(
-                                        image_file_path
-                                    )
-                                else:
-                                    # Local file
-                                    with open(image_file_path, "rb") as f:
-                                        image_content = f.read()
-
-                                # Prepare image content for Bedrock
-                                image_attachment = (
-                                    image.prepare_bedrock_image_attachment(
-                                        image_content
-                                    )
-                                )
-                                content.append(image_attachment)
-
-                            except Exception as e:
-                                logger.warning(
-                                    f"Failed to load image {image_file_path}: {e}"
-                                )
-                                continue
-
-                    except Exception as e:
-                        raise ValueError(
-                            f"Failed to load example images from {image_path}: {e}"
-                        )
-
-        return content
-
-    def _get_image_files_from_path(self, image_path: str) -> List[str]:
-        """
-        Get list of image files from a path that could be a single file, directory, or S3 prefix.
-
-        Args:
-            image_path: Path to image file, directory, or S3 prefix
-
-        Returns:
-            List of image file paths/URIs sorted by filename
-        """
-        import os
-
-        from idp_common import s3
-
-        # Handle S3 URIs
-        if image_path.startswith("s3://"):
-            # Check if it's a direct file or a prefix
-            if image_path.endswith(
-                (".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff", ".tif", ".webp")
-            ):
-                # Direct S3 file
-                return [image_path]
-            else:
-                # S3 prefix - list all images
-                return s3.list_images_from_path(image_path)
-        else:
-            # Handle local paths
-            config_bucket = os.environ.get("CONFIGURATION_BUCKET")
-            root_dir = os.environ.get("ROOT_DIR")
-
-            if config_bucket:
-                # Use environment bucket with imagePath as key
-                s3_uri = f"s3://{config_bucket}/{image_path}"
-
-                # Check if it's a direct file or a prefix
-                if image_path.endswith(
-                    (".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff", ".tif", ".webp")
-                ):
-                    # Direct S3 file
-                    return [s3_uri]
-                else:
-                    # S3 prefix - list all images
-                    return s3.list_images_from_path(s3_uri)
-            elif root_dir:
-                # Use relative path from ROOT_DIR
-                full_path = os.path.join(root_dir, image_path)
-                full_path = os.path.normpath(full_path)
-
-                if os.path.isfile(full_path):
-                    # Single local file
-                    return [full_path]
-                elif os.path.isdir(full_path):
-                    # Local directory - list all images
-                    return s3.list_images_from_path(full_path)
-                else:
-                    # Path doesn't exist
-                    logger.warning(f"Image path does not exist: {full_path}")
-                    return []
-            else:
-                raise ValueError(
-                    "No CONFIGURATION_BUCKET or ROOT_DIR set. Cannot read example images from local filesystem."
-                )
-
     def classify_page_bedrock(
         self,
         page_id: str,

lib/idp_common_pkg/idp_common/config/migration.py

@@ -10,12 +10,12 @@
     SCHEMA_ITEMS,
     SCHEMA_REQUIRED,
     SCHEMA_DESCRIPTION,
-    SCHEMA_EXAMPLES,
     TYPE_OBJECT,
     TYPE_ARRAY,
     TYPE_STRING,
     # AWS IDP extensions
     X_AWS_IDP_DOCUMENT_TYPE,
+    X_AWS_IDP_EXAMPLES,
     X_AWS_IDP_LIST_ITEM_DESCRIPTION,
     X_AWS_IDP_ORIGINAL_NAME,
     X_AWS_IDP_EVALUATION_METHOD,
@@ -151,9 +151,7 @@ def migrate_legacy_to_schema(
 
         # Migrate examples if present
         if LEGACY_EXAMPLES in class_config:
-            migrated_class[LEGACY_EXAMPLES] = _migrate_examples(
-                class_config[LEGACY_EXAMPLES]
-            )
+            migrated_class[X_AWS_IDP_EXAMPLES] = class_config[LEGACY_EXAMPLES]
 
         legacy_attributes = class_config.get(LEGACY_ATTRIBUTES, [])
 
@@ -280,98 +278,6 @@ def _migrate_list_attribute(attr: Dict[str, Any]) -> Dict[str, Any]:
     return schema_attr
 
 
-def _migrate_examples(legacy_examples: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """
-    Migrate legacy examples to JSON Schema examples format.
-
-    Legacy format has:
-    - classPrompt: Classification prompt
-    - attributesPrompt: Extraction prompt
-    - imagePath: Path to example image
-    - name: Example name
-
-    JSON Schema format uses standard 'examples' array with AWS extensions.
-
-    Args:
-        legacy_examples: List of legacy example dictionaries
-
-    Returns:
-        List of migrated examples in JSON Schema format
-    """
-    if not legacy_examples:
-        return []
-
-    migrated_examples = []
-
-    for example in legacy_examples:
-        # Create a JSON Schema example with AWS IDP extensions
-        migrated_example = {}
-
-        # Preserve the name if present
-        if LEGACY_NAME in example:
-            migrated_example[LEGACY_NAME] = example[LEGACY_NAME]
-
-        # Add AWS IDP extensions for few-shot learning
-        if LEGACY_CLASS_PROMPT in example:
-            migrated_example[X_AWS_IDP_CLASS_PROMPT] = example[LEGACY_CLASS_PROMPT]
-
-        if LEGACY_ATTRIBUTES_PROMPT in example:
-            migrated_example[X_AWS_IDP_ATTRIBUTES_PROMPT] = example[
-                LEGACY_ATTRIBUTES_PROMPT
-            ]
-
-        if LEGACY_IMAGE_PATH in example:
-            migrated_example[X_AWS_IDP_IMAGE_PATH] = example[LEGACY_IMAGE_PATH]
-
-        # Parse the attributes prompt to create actual example values if possible
-        # This helps with standard JSON Schema validation
-        if LEGACY_ATTRIBUTES_PROMPT in example:
-            try:
-                # Try to extract JSON values from the attributes prompt
-                import re
-                import json
-
-                prompt = example[LEGACY_ATTRIBUTES_PROMPT]
-                # Look for JSON-like structure in the prompt
-                json_match = re.search(r"\{.*\}", prompt, re.DOTALL)
-                parsed = False
-                if json_match:
-                    json_str = json_match.group()
-                    try:
-                        example_values = json.loads(json_str)
-                        if isinstance(example_values, dict):
-                            for key, value in example_values.items():
-                                migrated_example[key] = value
-                            parsed = True
-                    except (json.JSONDecodeError, ValueError):
-                        parsed = False
-
-                if not parsed:
-                    # Try to find key-value pairs in format: "key": "value"
-                    pairs = re.findall(
-                        r'"([^"]+)":\s*("(?:[^"\\]|\\.)*"|\bnull\b)', prompt
-                    )
-                    if pairs:
-                        # Build a JSON object from the pairs
-                        json_str = (
-                            "{" + ", ".join([f'"{k}": {v}' for k, v in pairs]) + "}"
-                        )
-                        try:
-                            example_values = json.loads(json_str)
-                            # Add the parsed values to the example
-                            for key, value in example_values.items():
-                                migrated_example[key] = value
-                        except (json.JSONDecodeError, ValueError):
-                            pass  # If parsing fails, just keep the prompt as-is
-            except Exception:
-                pass  # Safely ignore parsing errors
-
-        if migrated_example:  # Only add if we have content
-            migrated_examples.append(migrated_example)
-
-    return migrated_examples
-
-
 def _add_aws_extensions(legacy_attr: Dict[str, Any], schema: Dict[str, Any]) -> None:
     """Add AWS extension fields back to legacy format (for reverse migration if needed)."""
     if X_AWS_IDP_EVALUATION_METHOD in schema:
@@ -542,9 +448,13 @@ def _convert_classes_to_json_schema(
         if required:
             schema[SCHEMA_REQUIRED] = required
 
-        # Add examples if present
+        # Add examples if present (check both legacy and new key)
         if LEGACY_EXAMPLES in doc_type_class and doc_type_class[LEGACY_EXAMPLES]:
-            schema[SCHEMA_EXAMPLES] = doc_type_class[LEGACY_EXAMPLES]
+            schema[X_AWS_IDP_EXAMPLES] = doc_type_class[LEGACY_EXAMPLES]
+        elif (
+            X_AWS_IDP_EXAMPLES in doc_type_class and doc_type_class[X_AWS_IDP_EXAMPLES]
+        ):
+            schema[X_AWS_IDP_EXAMPLES] = doc_type_class[X_AWS_IDP_EXAMPLES]
 
         if defs:
             schema[DEFS_FIELD] = defs

lib/idp_common_pkg/idp_common/config/models.py

@@ -542,6 +542,32 @@ class IDPConfig(BaseModel):
         default_factory=EvaluationConfig, description="Evaluation configuration"
     )
 
+    # Criteria validation specific fields (used in pattern-2/criteria-validation)
+    summary: Optional[Dict[str, Any]] = Field(
+        default=None, description="Summary configuration for criteria validation"
+    )
+    criteria_types: Optional[List[str]] = Field(
+        default=None, description="List of criteria types for validation"
+    )
+    request_bucket: Optional[str] = Field(
+        default=None, description="S3 bucket for user history/request data"
+    )
+    request_history_prefix: Optional[str] = Field(
+        default=None, description="S3 prefix for request history"
+    )
+    criteria_bucket: Optional[str] = Field(
+        default=None, description="S3 bucket for criteria documents"
+    )
+    output_bucket: Optional[str] = Field(
+        default=None, description="S3 bucket for processing output"
+    )
+    textract_page_tracker: Optional[str] = Field(
+        default=None, description="S3 bucket for Textract page tracking"
+    )
+    cost_report_bucket: Optional[str] = Field(
+        default=None, description="S3 bucket for cost reports"
+    )
+
     model_config = ConfigDict(
         # Do not allow extra fields - all config should be explicit
         extra="forbid",

lib/idp_common_pkg/idp_common/config/schema_constants.py

@@ -51,6 +51,10 @@
 # Evaluation method for attribute comparison
 X_AWS_IDP_EVALUATION_METHOD = "x-aws-idp-evaluation-method"
 
+
+
+X_AWS_IDP_EXAMPLES= "x-aws-idp-examples"
+
 # Valid evaluation methods
 EVALUATION_METHOD_EXACT = "EXACT"
 EVALUATION_METHOD_NUMERIC_EXACT = "NUMERIC_EXACT"
@@ -118,7 +122,6 @@
 SCHEMA_ITEMS = "items"
 SCHEMA_REQUIRED = "required"
 SCHEMA_DESCRIPTION = "description"
-SCHEMA_EXAMPLES = "examples"  # Standard JSON Schema examples field
 
 # JSON Schema type values
 TYPE_OBJECT = "object"