Skip to content

Models

stickler.structured_object_evaluator.models

Models for structured object evaluation.

stickler.structured_object_evaluator.models.structured_model

Structured model comparison using Pydantic models.

This module provides the StructuredModel class for defining structured data models with comparison configuration and evaluation capabilities.

stickler.structured_object_evaluator.models.structured_model.StructuredModel

Bases: BaseModel

Base class for models with structured comparison capabilities.

This class extends Pydantic's BaseModel with the ability to compare instances using configurable comparison metrics for each field.

Architecture - Delegation Pattern:

StructuredModel uses a delegation pattern where comparison logic is distributed across specialized helper classes. This refactoring reduced the class from 2584 lines to ~500 lines while maintaining all functionality.

The delegation pattern works as follows: 1. StructuredModel maintains the public API (compare, compare_with, compare_field) 2. All implementation details are delegated to specialized helper classes 3. Each helper class has a single, well-defined responsibility 4. Helpers receive the StructuredModel instance as a parameter (composition) 5. This avoids circular dependencies and keeps the architecture clean

Helper Classes and Their Responsibilities:

Model Creation: - ModelFactory: Creates dynamic StructuredModel subclasses from JSON configuration - Validates configuration structure - Converts field definitions to Pydantic fields - Creates model classes using Pydantic's create_model()

Comparison Orchestration: - ComparisonEngine: Main orchestrator for the comparison process - Coordinates between dispatcher, collectors, and calculators - Implements single-traversal optimization - Manages compare_recursive and compare_with methods

Field Comparison Routing: - ComparisonDispatcher: Routes field comparisons to appropriate handlers - Uses match-statement based dispatch for clarity - Handles null cases and type mismatches - Delegates to specialized comparators based on field type

Field-Level Comparison: - FieldComparator: Compares primitive and structured fields - Handles string, int, float comparisons - Handles nested StructuredModel comparisons - Applies threshold-based binary classification

  • PrimitiveListComparator: Compares lists of primitive values
  • Uses Hungarian matching for optimal pairing
  • Returns hierarchical structure for API consistency
  • Handles empty list cases

  • StructuredListComparator: Compares lists of StructuredModels

  • Uses Hungarian matching with object-level similarity
  • Performs threshold-gated recursive analysis
  • Calculates nested field metrics

Metrics Calculation: - ConfusionMatrixCalculator: Calculates confusion matrix metrics - Computes TP, FP, TN, FN, FD, FA counts - Handles list-level and field-level metrics - Calculates nested field metrics for structured lists

  • AggregateMetricsCalculator: Rolls up child metrics to parent nodes
  • Performs recursive traversal of result tree
  • Sums child aggregate metrics to parent
  • Provides universal field-level granularity

  • DerivedMetricsCalculator: Calculates derived metrics

  • Computes precision, recall, F1, accuracy
  • Supports both traditional and FD-inclusive recall
  • Delegates to MetricsHelper for calculations

  • ConfusionMatrixBuilder: Orchestrates all metrics calculation

  • Coordinates between the three calculator classes
  • Ensures correct calculation order
  • Builds complete confusion matrices

Non-Match Documentation: - NonMatchCollector: Documents non-matching fields - Collects object-level non-matches for lists - Collects field-level non-matches (legacy format) - Handles nested StructuredModel recursion

Existing Helpers (Pre-Refactoring): - HungarianHelper: Hungarian algorithm for list matching - MetricsHelper: Derived metrics calculation formulas - ConfigurationHelper: Field configuration management - ComparisonHelper: Comparison utility methods - EvaluatorFormatHelper: Output formatting for evaluators - NonMatchesHelper: Non-match collection utilities - FieldHelper: Field type and null checking utilities

Benefits of Delegation Pattern:
  1. Maintainability: Each class has a single responsibility
  2. Testability: Components can be tested in isolation
  3. Extensibility: Easy to add new field types or metrics
  4. Readability: Clear separation of concerns
  5. Performance: No overhead - delegation is just function calls
Migration Notes:
  • All public APIs remain unchanged (complete backward compatibility)
  • All tests pass without modification (80+ test files)
  • Performance characteristics maintained (single-traversal optimization)
  • No breaking changes for existing users
Features:
  • Field-level comparison configuration via ComparableField
  • Nested model comparison with recursive evaluation
  • Integration with ANLS* comparators
  • JSON schema generation with comparison metadata
  • Unordered list comparison using Hungarian matching
  • Confusion matrix metrics (TP, FP, FN, TN, FA, FD)
  • Aggregate metrics rollup from nested fields
  • Retention of extra fields not defined in the model
  • Dynamic model creation from JSON configuration
  • Threshold-gated recursive analysis for performance
Example Usage:

from stickler.structured_object_evaluator.models import StructuredModel from stickler.structured_object_evaluator.models import ComparableField from stickler.comparators import LevenshteinComparator

class Product(StructuredModel): ... name: str = ComparableField( ... comparator=LevenshteinComparator(), ... threshold=0.8, ... weight=2.0 ... ) ... price: float = ComparableField( ... comparator=NumericComparator(), ... threshold=0.9 ... )

gt = Product(name="Widget", price=29.99) pred = Product(name="Widgit", price=29.99) # Typo in name

Simple comparison (returns overall similarity score)

score = gt.compare(pred) print(f"Similarity: {score:.2f}")

Detailed comparison with confusion matrix

result = gt.compare_with(pred, include_confusion_matrix=True) print(f"TP: {result['overall']['tp']}, FD: {result['overall']['fd']}") print(f"F1: {result['aggregate']['derived']['cm_f1']:.2f}")

Source code in stickler/structured_object_evaluator/models/structured_model.py
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78
  79
  80
  81
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94
  95
  96
  97
  98
  99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
class StructuredModel(BaseModel):
    """Base class for models with structured comparison capabilities.

    This class extends Pydantic's BaseModel with the ability to compare
    instances using configurable comparison metrics for each field.

    Architecture - Delegation Pattern:
    ----------------------------------
    StructuredModel uses a delegation pattern where comparison logic is
    distributed across specialized helper classes. This refactoring reduced
    the class from 2584 lines to ~500 lines while maintaining all functionality.

    The delegation pattern works as follows:
    1. StructuredModel maintains the public API (compare, compare_with, compare_field)
    2. All implementation details are delegated to specialized helper classes
    3. Each helper class has a single, well-defined responsibility
    4. Helpers receive the StructuredModel instance as a parameter (composition)
    5. This avoids circular dependencies and keeps the architecture clean

    Helper Classes and Their Responsibilities:
    ------------------------------------------

    **Model Creation:**
    - ModelFactory: Creates dynamic StructuredModel subclasses from JSON configuration
      - Validates configuration structure
      - Converts field definitions to Pydantic fields
      - Creates model classes using Pydantic's create_model()

    **Comparison Orchestration:**
    - ComparisonEngine: Main orchestrator for the comparison process
      - Coordinates between dispatcher, collectors, and calculators
      - Implements single-traversal optimization
      - Manages compare_recursive and compare_with methods

    **Field Comparison Routing:**
    - ComparisonDispatcher: Routes field comparisons to appropriate handlers
      - Uses match-statement based dispatch for clarity
      - Handles null cases and type mismatches
      - Delegates to specialized comparators based on field type

    **Field-Level Comparison:**
    - FieldComparator: Compares primitive and structured fields
      - Handles string, int, float comparisons
      - Handles nested StructuredModel comparisons
      - Applies threshold-based binary classification

    - PrimitiveListComparator: Compares lists of primitive values
      - Uses Hungarian matching for optimal pairing
      - Returns hierarchical structure for API consistency
      - Handles empty list cases

    - StructuredListComparator: Compares lists of StructuredModels
      - Uses Hungarian matching with object-level similarity
      - Performs threshold-gated recursive analysis
      - Calculates nested field metrics

    **Metrics Calculation:**
    - ConfusionMatrixCalculator: Calculates confusion matrix metrics
      - Computes TP, FP, TN, FN, FD, FA counts
      - Handles list-level and field-level metrics
      - Calculates nested field metrics for structured lists

    - AggregateMetricsCalculator: Rolls up child metrics to parent nodes
      - Performs recursive traversal of result tree
      - Sums child aggregate metrics to parent
      - Provides universal field-level granularity

    - DerivedMetricsCalculator: Calculates derived metrics
      - Computes precision, recall, F1, accuracy
      - Supports both traditional and FD-inclusive recall
      - Delegates to MetricsHelper for calculations

    - ConfusionMatrixBuilder: Orchestrates all metrics calculation
      - Coordinates between the three calculator classes
      - Ensures correct calculation order
      - Builds complete confusion matrices

    **Non-Match Documentation:**
    - NonMatchCollector: Documents non-matching fields
      - Collects object-level non-matches for lists
      - Collects field-level non-matches (legacy format)
      - Handles nested StructuredModel recursion

    **Existing Helpers (Pre-Refactoring):**
    - HungarianHelper: Hungarian algorithm for list matching
    - MetricsHelper: Derived metrics calculation formulas
    - ConfigurationHelper: Field configuration management
    - ComparisonHelper: Comparison utility methods
    - EvaluatorFormatHelper: Output formatting for evaluators
    - NonMatchesHelper: Non-match collection utilities
    - FieldHelper: Field type and null checking utilities

    Benefits of Delegation Pattern:
    --------------------------------
    1. **Maintainability**: Each class has a single responsibility
    2. **Testability**: Components can be tested in isolation
    3. **Extensibility**: Easy to add new field types or metrics
    4. **Readability**: Clear separation of concerns
    5. **Performance**: No overhead - delegation is just function calls

    Migration Notes:
    ----------------
    - All public APIs remain unchanged (complete backward compatibility)
    - All tests pass without modification (80+ test files)
    - Performance characteristics maintained (single-traversal optimization)
    - No breaking changes for existing users

    Features:
    ---------
    - Field-level comparison configuration via ComparableField
    - Nested model comparison with recursive evaluation
    - Integration with ANLS* comparators
    - JSON schema generation with comparison metadata
    - Unordered list comparison using Hungarian matching
    - Confusion matrix metrics (TP, FP, FN, TN, FA, FD)
    - Aggregate metrics rollup from nested fields
    - Retention of extra fields not defined in the model
    - Dynamic model creation from JSON configuration
    - Threshold-gated recursive analysis for performance

    Example Usage:
    --------------
    >>> from stickler.structured_object_evaluator.models import StructuredModel
    >>> from stickler.structured_object_evaluator.models import ComparableField
    >>> from stickler.comparators import LevenshteinComparator
    >>> 
    >>> class Product(StructuredModel):
    ...     name: str = ComparableField(
    ...         comparator=LevenshteinComparator(),
    ...         threshold=0.8,
    ...         weight=2.0
    ...     )
    ...     price: float = ComparableField(
    ...         comparator=NumericComparator(),
    ...         threshold=0.9
    ...     )
    >>> 
    >>> gt = Product(name="Widget", price=29.99)
    >>> pred = Product(name="Widgit", price=29.99)  # Typo in name
    >>> 
    >>> # Simple comparison (returns overall similarity score)
    >>> score = gt.compare(pred)
    >>> print(f"Similarity: {score:.2f}")
    >>> 
    >>> # Detailed comparison with confusion matrix
    >>> result = gt.compare_with(pred, include_confusion_matrix=True)
    >>> print(f"TP: {result['overall']['tp']}, FD: {result['overall']['fd']}")
    >>> print(f"F1: {result['aggregate']['derived']['cm_f1']:.2f}")
    """

    # Default match threshold - can be overridden in subclasses
    match_threshold: ClassVar[float] = 0.7

    extra_fields: Dict[str, Any] = Field(default_factory=dict, exclude=True)

    model_config = {
        "arbitrary_types_allowed": True,
        "extra": "allow",  # Allow extra fields to be stored in extra_fields
    }

    def __init_subclass__(cls, **kwargs):
        """Validate field configurations when a StructuredModel subclass is defined."""
        super().__init_subclass__(**kwargs)

        # Validate field configurations using class annotations since model_fields isn't populated yet
        if hasattr(cls, "__annotations__"):
            for field_name, field_type in cls.__annotations__.items():
                if field_name == "extra_fields":
                    continue

                # Get the field default value if it exists
                field_default = getattr(cls, field_name, None)

                # Since ComparableField is now always a function that returns a Field,
                # we need to check if field_default has comparison metadata
                if hasattr(field_default, "json_schema_extra") and callable(
                    field_default.json_schema_extra
                ):
                    # Check for comparison metadata
                    temp_schema = {}
                    field_default.json_schema_extra(temp_schema)
                    if "x-comparison" in temp_schema:
                        # This field was created with ComparableField function - validate constraints
                        if cls._is_list_of_structured_model_type(field_type):
                            comparison_config = temp_schema["x-comparison"]

                            # Threshold validation - only flag if explicitly set to non-default value
                            threshold = comparison_config.get("threshold", 0.5)
                            if threshold != 0.5:  # Default threshold value
                                raise ValueError(
                                    f"Field '{field_name}' is a List[StructuredModel] and cannot have a "
                                    f"'threshold' parameter in ComparableField. Hungarian matching uses each "
                                    f"StructuredModel's 'match_threshold' class attribute instead. "
                                    f"Set 'match_threshold = {threshold}' on the list element class."
                                )

                            # Comparator validation - only flag if explicitly set to non-default type
                            comparator_type = comparison_config.get(
                                "comparator_type", "LevenshteinComparator"
                            )
                            if (
                                comparator_type != "LevenshteinComparator"
                            ):  # Default comparator type
                                raise ValueError(
                                    f"Field '{field_name}' is a List[StructuredModel] and cannot have a "
                                    f"'comparator' parameter in ComparableField. Object comparison uses each "
                                    f"StructuredModel's individual field comparators instead."
                                )

    @classmethod
    def _is_list_of_structured_model_type(cls, field_type) -> bool:
        """Check if a field type annotation represents List[StructuredModel].

        Args:
            field_type: The field type annotation

        Returns:
            True if the field is a List[StructuredModel] type
        """
        # Handle direct imports and typing constructs
        origin = get_origin(field_type)
        if origin is list or origin is List:
            args = get_args(field_type)
            if args:
                element_type = args[0]
                # Check if element type is a StructuredModel subclass
                try:
                    return inspect.isclass(element_type) and issubclass(
                        element_type, StructuredModel
                    )
                except (TypeError, AttributeError):
                    return False

        # Handle Union types (like Optional[List[StructuredModel]])
        elif origin is Union:
            args = get_args(field_type)
            for arg in args:
                if cls._is_list_of_structured_model_type(arg):
                    return True

        return False

    @classmethod
    def from_json(cls, json_data: Dict[str, Any]) -> "StructuredModel":
        """Create a StructuredModel instance from JSON data.

        This method handles missing fields gracefully and stores extra fields
        in the extra_fields attribute.

        Args:
            json_data: Dictionary containing the JSON data

        Returns:
            StructuredModel instance created from the JSON data
        """
        return ConfigurationHelper.from_json(cls, json_data)

    @classmethod
    def model_from_json(cls, config: Dict[str, Any]) -> Type["StructuredModel"]:
        """Create a StructuredModel subclass from JSON configuration using Pydantic's create_model().

        This method leverages Pydantic's native dynamic model creation capabilities to ensure
        full compatibility with all Pydantic features while adding structured comparison
        functionality through inherited StructuredModel methods.

        The generated model inherits all StructuredModel capabilities:
        - compare_with() method for detailed comparisons
        - Field-level comparison configuration
        - Hungarian algorithm for list matching
        - Confusion matrix generation
        - JSON schema with comparison metadata

        Args:
            config: JSON configuration with fields, comparators, and model settings.
                   Required keys:
                   - fields: Dict mapping field names to field configurations
                   Optional keys:
                   - model_name: Name for the generated class (default: "DynamicModel")
                   - match_threshold: Overall matching threshold (default: 0.7)

                   Field configuration format:
                   {
                       "type": "str|int|float|bool|List[str]|etc.",  # Required
                       "comparator": "LevenshteinComparator|ExactComparator|etc.",  # Optional
                       "threshold": 0.8,  # Optional, default 0.5
                       "weight": 2.0,     # Optional, default 1.0
                       "required": true,  # Optional, default false
                       "default": "value", # Optional
                       "description": "Field description",  # Optional
                       "alias": "field_alias",  # Optional
                       "examples": ["example1", "example2"]  # Optional
                   }

        Returns:
            A fully functional StructuredModel subclass created with create_model()

        Raises:
            ValueError: If configuration is invalid or contains unsupported types/comparators
            KeyError: If required configuration keys are missing

        Examples:
            >>> config = {
            ...     "model_name": "Product",
            ...     "match_threshold": 0.8,
            ...     "fields": {
            ...         "name": {
            ...             "type": "str",
            ...             "comparator": "LevenshteinComparator",
            ...             "threshold": 0.8,
            ...             "weight": 2.0,
            ...             "required": True
            ...         },
            ...         "price": {
            ...             "type": "float",
            ...             "comparator": "NumericComparator",
            ...             "default": 0.0
            ...         }
            ...     }
            ... }
            >>> ProductClass = StructuredModel.model_from_json(config)
            >>> isinstance(ProductClass.model_fields, dict)  # Full Pydantic compatibility
            True
            >>> product = ProductClass(name="Widget", price=29.99)
            >>> product.name
            'Widget'
            >>> result = product.compare_with(ProductClass(name="Widget", price=29.99))
            >>> result["overall_score"]
            1.0
        """
        # Delegate to ModelFactory for dynamic model creation
        from .model_factory import ModelFactory

        return ModelFactory.create_model_from_json(config, base_class=cls)

    @classmethod
    def from_json_schema(cls, schema: Dict[str, Any]) -> Type["StructuredModel"]:
        """Create a StructuredModel subclass from a JSON Schema document.

        This method accepts standard JSON Schema documents and creates fully functional
        StructuredModel classes with comparison capabilities. Supports JSON Schema draft-07+.

        Comparison behavior can be customized using x-aws-stickler-* extension fields:

        Field-Level Extensions:
        -----------------------
        - x-aws-stickler-comparator: Comparator algorithm name (built-in or registered custom)
        - x-aws-stickler-threshold: Similarity threshold for match/no-match (0.0-1.0, default: 0.5)
        - x-aws-stickler-weight: Field importance in overall scoring (>0.0, default: 1.0)
        - x-aws-stickler-clip-under-threshold: Clip scores below threshold to 0.0 (bool, default: false)
        - x-aws-stickler-aggregate: Include field metrics in parent aggregation (bool, default: false)

        Model-Level Extensions:
        -----------------------
        - x-aws-stickler-model-name: Generated class name (default: "DynamicModel")
        - x-aws-stickler-match-threshold: Overall match threshold (default: 0.7)

        Supported Features:
        -------------------
        - Primitive types: string, number, integer, boolean
        - Nested objects and arrays (primitive/object items)
        - Required fields, defaults, descriptions
        - Schema references ($ref with #/definitions/ and #/$defs/)

        Default Type Mappings:
        ----------------------
        - string → LevenshteinComparator (threshold: 0.5)
        - number/integer → NumericComparator (threshold: 0.5)
        - boolean → ExactComparator (threshold: 1.0)
        - arrays → Hungarian matching with element-appropriate comparators
        - objects → Recursive field-by-field comparison

        Args:
            schema: JSON Schema document as a dictionary

        Returns:
            StructuredModel subclass created from the schema

        Raises:
            ValueError: If schema is invalid or contains unsupported features
            jsonschema.exceptions.SchemaError: If schema doesn't conform to JSON Schema spec

        Examples:
            Basic usage with standard JSON Schema:
            >>> schema = {
            ...     "type": "object",
            ...     "properties": {
            ...         "name": {"type": "string"},
            ...         "age": {"type": "integer"},
            ...         "email": {"type": "string"}
            ...     },
            ...     "required": ["name", "email"]
            ... }
            >>> PersonModel = StructuredModel.from_json_schema(schema)
            >>> person1 = PersonModel(name="Alice", age=30, email="alice@example.com")
            >>> person2 = PersonModel(name="Alicia", age=30, email="alice@example.com")
            >>> result = person1.compare_with(person2)
            >>> # name field uses LevenshteinComparator, age uses NumericComparator

            Advanced usage with x-aws-stickler-* extensions:
            >>> schema = {
            ...     "type": "object",
            ...     "x-aws-stickler-model-name": "Product",
            ...     "x-aws-stickler-match-threshold": 0.8,
            ...     "properties": {
            ...         "name": {
            ...             "type": "string",
            ...             "x-aws-stickler-comparator": "LevenshteinComparator",
            ...             "x-aws-stickler-threshold": 0.9,
            ...             "x-aws-stickler-weight": 2.0,
            ...             "x-aws-stickler-aggregate": true
            ...         },
            ...         "price": {
            ...             "type": "number",
            ...             "x-aws-stickler-comparator": "NumericComparator",
            ...             "x-aws-stickler-threshold": 0.95,
            ...             "x-aws-stickler-clip-under-threshold": true
            ...         }
            ...     },
            ...     "required": ["name"]
            ... }
            >>> ProductModel = StructuredModel.from_json_schema(schema)
            >>> result = product1.compare_with(product2)
            >>> # name field has weight=2.0, price field clips scores below 0.95
            """

        return cls._from_json_schema_internal(schema, field_path="")

    @classmethod
    def _from_json_schema_internal(
        cls, schema: Dict[str, Any], field_path: str
    ) -> Type["StructuredModel"]:
        """Internal method for creating StructuredModel from JSON Schema with field path tracking.

        This is used internally for recursive calls to track field paths for error messages.
        External callers should use from_json_schema() instead.

        Args:
            schema: JSON Schema document as a dictionary
            field_path: Current field path for error messages (e.g., "address.street")

        Returns:
            StructuredModel subclass created from the schema
        """
        # Import dependencies
        from ..utils.json_schema_validator import validate_json_schema
        from .json_schema_field_converter import JsonSchemaFieldConverter
        from .model_factory import ModelFactory

        # Subtask 4.2: Validate JSON Schema
        try:
            validate_json_schema(schema)
        except Exception as e:
            raise ValueError(
                f"Invalid JSON Schema: {e}. "
                f"Please ensure the schema conforms to JSON Schema draft-07 specification."
            )

        # Subtask 4.3: Extract model-level configuration
        model_name = schema.get("x-aws-stickler-model-name", "DynamicModel")
        match_threshold = schema.get("x-aws-stickler-match-threshold", 0.7)

        # Validate model name
        if not isinstance(model_name, str) or not model_name.isidentifier():
            raise ValueError(
                f"x-aws-stickler-model-name must be a valid Python identifier, "
                f"got: {model_name}"
            )

        # Validate match threshold
        if not isinstance(match_threshold, (int, float)):
            raise ValueError(
                f"x-aws-stickler-match-threshold must be a number, "
                f"got: {type(match_threshold).__name__}"
            )

        if not (0.0 <= match_threshold <= 1.0):
            raise ValueError(
                f"x-aws-stickler-match-threshold must be between 0.0 and 1.0, "
                f"got: {match_threshold}"
            )

        # Subtask 4.4: Convert fields and create model
        # Ensure schema has properties
        if "properties" not in schema:
            raise ValueError(
                "JSON Schema must contain 'properties' key for object type"
            )

        properties = schema.get("properties", {})
        required = schema.get("required", [])

        # Create converter and convert properties to field definitions
        converter = JsonSchemaFieldConverter(schema, field_path=field_path)
        field_definitions = converter.convert_properties_to_fields(properties, required)

        # Create the model using ModelFactory
        return ModelFactory.create_model_from_fields(
            model_name=model_name,
            field_definitions=field_definitions,
            match_threshold=match_threshold,
            base_class=cls
        )

    @classmethod
    def _is_structured_field_type(cls, field_info) -> bool:
        """Check if a field represents a structured type that needs special handling.

        Args:
            field_info: Pydantic field info object

        Returns:
            True if the field is a List[StructuredModel] or StructuredModel type
        """
        return ConfigurationHelper.is_structured_field_type(field_info)

    @classmethod
    def _get_comparison_info(cls, field_name: str) -> ComparableField:
        """Extract comparison info from a field.

        Args:
            field_name: Name of the field to get comparison info for

        Returns:
            ComparableField object with comparison configuration
        """
        return ConfigurationHelper.get_comparison_info(cls, field_name)



    @classmethod
    def _is_aggregate_field(cls, field_name: str) -> bool:
        """Check if field is marked for confusion matrix aggregation.

        Args:
            field_name: Name of the field to check

        Returns:
            True if the field is marked for aggregation, False otherwise
        """
        return ConfigurationHelper.is_aggregate_field(cls, field_name)

    def _is_truly_null(self, val: Any) -> bool:
        """Check if a value is truly null (None).

        DEPRECATED: Delegates to NullHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .null_helper import NullHelper
        return NullHelper.is_truly_null(val)

    def _should_use_hierarchical_structure(self, val: Any, field_name: str) -> bool:
        """Check if a list value should maintain hierarchical structure.

        For lists, we need to check if they should maintain hierarchical structure
        based on their field type configuration.

        Args:
            val: Value to check (typically a list)
            field_name: Name of the field being evaluated

        Returns:
            True if the value should use hierarchical structure, False otherwise
        """
        if isinstance(val, list):
            # Check if this field is configured as List[StructuredModel]
            field_info = self.__class__.model_fields.get(field_name)
            if field_info and self._is_structured_field_type(field_info):
                return True
        return False

    def _is_effectively_null_for_lists(self, val: Any) -> bool:
        """Check if a list value is effectively null (None or empty list).

        DEPRECATED: Delegates to NullHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .null_helper import NullHelper
        return NullHelper.is_effectively_null_for_lists(val)

    def _is_effectively_null_for_primitives(self, val: Any) -> bool:
        """Check if a primitive value is effectively null.

        DEPRECATED: Delegates to NullHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .null_helper import NullHelper
        return NullHelper.is_effectively_null_for_primitives(val)

    def _is_list_field(self, field_name: str) -> bool:
        """Check if a field is ANY list type.

        Args:
            field_name: Name of the field to check

        Returns:
            True if the field is a list type (List[str], List[StructuredModel], etc.)
        """
        field_info = self.__class__.model_fields.get(field_name)
        if not field_info:
            return False

        field_type = field_info.annotation
        # Handle Optional types and direct List types
        if hasattr(field_type, "__origin__"):
            origin = field_type.__origin__
            if origin is list or origin is List:
                return True
            elif origin is Union:  # Optional[List[...]] case
                args = field_type.__args__
                for arg in args:
                    if hasattr(arg, "__origin__") and (
                        arg.__origin__ is list or arg.__origin__ is List
                    ):
                        return True
        return False

    def _handle_list_field_dispatch(
        self, gt_val: Any, pred_val: Any, weight: float
    ) -> dict:
        """Handle list field comparison using match statements.

        DEPRECATED: This method now delegates to ComparisonDispatcher.
        Kept for backward compatibility with any external callers.

        Args:
            gt_val: Ground truth list value
            pred_val: Predicted list value
            weight: Field weight for scoring

        Returns:
            Comparison result dictionary
        """
        from .comparison_dispatcher import ComparisonDispatcher
        dispatcher = ComparisonDispatcher(self)
        return dispatcher.handle_list_field_dispatch(gt_val, pred_val, weight)

    def _create_true_negative_result(self, weight: float) -> dict:
        """Create a true negative result.

        DEPRECATED: Delegates to ResultHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .result_helper import ResultHelper
        return ResultHelper.create_true_negative_result(weight)

    def _create_false_alarm_result(self, weight: float) -> dict:
        """Create a false alarm result.

        DEPRECATED: Delegates to ResultHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .result_helper import ResultHelper
        return ResultHelper.create_false_alarm_result(weight)

    def _create_false_negative_result(self, weight: float) -> dict:
        """Create a false negative result.

        DEPRECATED: Delegates to ResultHelper for consistency.
        Kept for backward compatibility with any external callers.
        """
        from .result_helper import ResultHelper
        return ResultHelper.create_false_negative_result(weight)

    def _handle_struct_list_empty_cases(
        self,
        gt_list: List["StructuredModel"],
        pred_list: List["StructuredModel"],
        weight: float,
    ) -> dict:
        """Handle empty list cases with beautiful match statements.

        DEPRECATED: Delegates to ResultHelper for consistency.
        Kept for backward compatibility with any external callers.

        Args:
            gt_list: Ground truth list (may be None)
            pred_list: Predicted list (may be None)
            weight: Field weight for scoring

        Returns:
            Result dictionary if early exit needed, None if should continue processing
        """
        from .result_helper import ResultHelper

        # Normalize None to empty lists for consistent handling
        gt_len = len(gt_list or [])
        pred_len = len(pred_list or [])

        return ResultHelper.create_empty_list_result(gt_len, pred_len, weight)

    def _calculate_object_level_metrics(
        self,
        gt_list: List["StructuredModel"],
        pred_list: List["StructuredModel"],
        match_threshold: float,
    ) -> tuple:
        """Calculate object-level metrics using Hungarian matching.

        Args:
            gt_list: Ground truth list
            pred_list: Predicted list
            match_threshold: Threshold for considering objects as matches

        Returns:
            Tuple of (object_metrics_dict, matched_pairs, matched_gt_indices, matched_pred_indices)
        """
        # Use Hungarian matching for OBJECT-LEVEL counts - OPTIMIZED: Single call gets all info
        hungarian_helper = HungarianHelper()
        hungarian_info = hungarian_helper.get_complete_matching_info(gt_list, pred_list)
        matched_pairs = hungarian_info["matched_pairs"]

        # Count OBJECTS, not individual fields
        tp_objects = 0  # Objects with similarity >= match_threshold
        fd_objects = 0  # Objects with similarity < match_threshold
        for gt_idx, pred_idx, similarity in matched_pairs:
            if similarity >= match_threshold:
                tp_objects += 1
            else:
                fd_objects += 1

        # Count unmatched objects
        matched_gt_indices = {idx for idx, _, _ in matched_pairs}
        matched_pred_indices = {idx for _, idx, _ in matched_pairs}
        fn_objects = len(gt_list) - len(matched_gt_indices)  # Unmatched GT objects
        fa_objects = len(pred_list) - len(
            matched_pred_indices
        )  # Unmatched pred objects

        # Build list-level metrics counting OBJECTS (not fields)
        object_level_metrics = {
            "tp": tp_objects,
            "fa": fa_objects,
            "fd": fd_objects,
            "fp": fa_objects + fd_objects,  # Total false positives
            "tn": 0,  # No true negatives at object level for non-empty lists
            "fn": fn_objects,
        }

        return (
            object_level_metrics,
            matched_pairs,
            matched_gt_indices,
            matched_pred_indices,
        )

    def _calculate_struct_list_similarity(
        self,
        gt_list: List["StructuredModel"],
        pred_list: List["StructuredModel"],
        info: "ComparableField",
    ) -> float:
        """Calculate raw similarity score for structured list.

        Args:
            gt_list: Ground truth list
            pred_list: Predicted list
            info: Field comparison info

        Returns:
            Raw similarity score between 0.0 and 1.0
        """
        if len(pred_list) > 0:
            match_result = self._compare_unordered_lists(
                gt_list, pred_list, info.comparator, info.threshold
            )
            return match_result.get("overall_score", 0.0)
        else:
            return 0.0

    def _compare_unordered_lists(
        self,
        list1: List[Any],
        list2: List[Any],
        comparator: BaseComparator,
        threshold: float,
    ) -> Dict[str, Any]:
        """Compare two lists as unordered collections using Hungarian matching.

        Args:
            list1: First list
            list2: Second list
            comparator: Comparator to use for item comparison
            threshold: Minimum score to consider a match

        Returns:
            Dictionary with confusion matrix metrics including:
            - tp: True positives (matches >= threshold)
            - fd: False discoveries (matches < threshold)
            - fa: False alarms (unmatched prediction items)
            - fn: False negatives (unmatched ground truth items)
            - fp: Total false positives (fd + fa)
            - overall_score: Similarity score for backward compatibility
        """
        return ComparisonHelper.compare_unordered_lists(
            list1, list2, comparator, threshold
        )

    def compare_field(self, field_name: str, other_value: Any) -> float:
        """Compare a single field with a value using the configured comparator.

        Args:
            field_name: Name of the field to compare
            other_value: Value to compare with

        Returns:
            Similarity score between 0.0 and 1.0
        """
        # Get our field value
        my_value = getattr(self, field_name)

        # If both values are StructuredModel instances, use recursive compare_with
        if isinstance(my_value, StructuredModel) and isinstance(
            other_value, StructuredModel
        ):
            # Use compare_with for rich comparison
            comparison_result = my_value.compare_with(
                other_value,
                include_confusion_matrix=False,
                document_non_matches=False,
                evaluator_format=False,
                recall_with_fd=False,
            )
            # Apply field-level threshold if configured
            info = self._get_comparison_info(field_name)
            raw_score = comparison_result["overall_score"]
            return (
                raw_score
                if raw_score >= info.threshold or not info.clip_under_threshold
                else 0.0
            )

        # CRITICAL FIX: For lists, don't clip under threshold for partial matches
        if isinstance(my_value, list) and isinstance(other_value, list):
            # Get field info
            info = self._get_comparison_info(field_name)

            # Use the raw comparison result without threshold clipping for lists
            result = ComparisonHelper.compare_unordered_lists(
                my_value, other_value, info.comparator, info.threshold
            )

            # Return the overall score directly (don't clip based on threshold for lists)
            return result["overall_score"]

        # For other fields, use existing logic
        return ComparisonHelper.compare_field_with_threshold(
            self, field_name, other_value
        )

    def compare_field_raw(self, field_name: str, other_value: Any) -> float:
        """Compare a single field with a value WITHOUT applying thresholds.

        This version is used by the compare method to get raw similarity scores.

        Args:
            field_name: Name of the field to compare
            other_value: Value to compare with

        Returns:
            Raw similarity score between 0.0 and 1.0 without threshold filtering
        """
        # Get our field value
        my_value = getattr(self, field_name)

        # If both values are StructuredModel instances, use recursive compare_with
        if isinstance(my_value, StructuredModel) and isinstance(
            other_value, StructuredModel
        ):
            # Use compare_with for rich comparison, but extract the raw score
            comparison_result = my_value.compare_with(
                other_value,
                include_confusion_matrix=False,
                document_non_matches=False,
                evaluator_format=False,
                recall_with_fd=False,
            )
            return comparison_result["overall_score"]

        # For non-StructuredModel fields, use existing logic
        return ComparisonHelper.compare_field_raw(self, field_name, other_value)

    def compare_recursive(self, other: "StructuredModel") -> dict:
        """The ONE clean recursive function that handles everything.

        Enhanced to capture BOTH confusion matrix metrics AND similarity scores
        in a single traversal to eliminate double traversal inefficiency.

        PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

        Args:
            other: Another instance of the same model to compare with

        Returns:
            Dictionary with clean hierarchical structure:
            - overall: TP, FP, TN, FN, FD, FA counts + similarity_score + all_fields_matched
            - fields: Recursive structure for each field with scores
            - non_matches: List of non-matching items
        """
        from .comparison_engine import ComparisonEngine
        engine = ComparisonEngine(self)
        return engine.compare_recursive(other)

    def _dispatch_field_comparison(
        self, field_name: str, gt_val: Any, pred_val: Any
    ) -> dict:
        """Enhanced case-based dispatch using match statements for clean logic flow.

        DEPRECATED: This method now delegates to ComparisonDispatcher.
        Kept for backward compatibility with any external callers.
        """
        from .comparison_dispatcher import ComparisonDispatcher
        dispatcher = ComparisonDispatcher(self)
        return dispatcher.dispatch_field_comparison(field_name, gt_val, pred_val)







    def _calculate_aggregate_metrics(self, result: dict) -> dict:
        """Calculate aggregate metrics for all nodes in the result tree.

        This method delegates to AggregateMetricsCalculator for the actual implementation.

        CRITICAL FIX: Enhanced deep nesting traversal to handle arbitrary nesting depth.
        The aggregate field contains the sum of all primitive field confusion matrices
        below that node in the tree. This provides universal field-level granularity.

        Args:
            result: Result from compare_recursive with hierarchical structure

        Returns:
            Modified result with 'aggregate' fields added at each level
        """
        from .aggregate_metrics_calculator import AggregateMetricsCalculator
        calculator = AggregateMetricsCalculator()
        return calculator.calculate_aggregate_metrics(result)

    def _add_derived_metrics_to_result(self, result: dict, recall_with_fd: bool = False) -> dict:
        """Walk through result and add 'derived' fields with F1, precision, recall, accuracy.

        This method delegates to DerivedMetricsCalculator for the actual implementation.

        Args:
            result: Result from compare_recursive with basic TP, FP, FN, etc. metrics
            recall_with_fd: If True, include FD in recall denominator (TP/(TP+FN+FD))
                           If False, use traditional recall (TP/(TP+FN))

        Returns:
            Modified result with 'derived' fields added at each level
        """
        from .derived_metrics_calculator import DerivedMetricsCalculator
        calculator = DerivedMetricsCalculator()
        return calculator.add_derived_metrics_to_result(result, recall_with_fd)

    def _has_basic_metrics(self, metrics_dict: dict) -> bool:
        """Check if a dictionary has basic confusion matrix metrics.

        Args:
            metrics_dict: Dictionary to check

        Returns:
            True if it has the basic metrics (tp, fp, fn, etc.)
        """
        basic_metrics = ["tp", "fp", "fn", "tn", "fa", "fd"]
        return all(metric in metrics_dict for metric in basic_metrics)

    def _classify_field_for_confusion_matrix(
        self, field_name: str, other_value: Any, threshold: float = None
    ) -> Dict[str, Any]:
        """Classify a field comparison according to the confusion matrix rules.

        This method delegates to ConfusionMatrixCalculator for the actual implementation.

        Args:
            field_name: Name of the field being compared
            other_value: Value to compare with
            threshold: Threshold for matching (uses field's threshold if None)

        Returns:
            Dictionary with TP, FP, TN, FN, FD counts and derived metrics
        """
        from .confusion_matrix_calculator import ConfusionMatrixCalculator
        calculator = ConfusionMatrixCalculator(self)
        return calculator.classify_field_for_confusion_matrix(field_name, other_value, threshold)

    def _calculate_list_confusion_matrix(
        self, field_name: str, other_list: List[Any]
    ) -> Dict[str, Any]:
        """Calculate confusion matrix for a list field, including nested field metrics.

        This method delegates to ConfusionMatrixCalculator for the actual implementation.

        Args:
            field_name: Name of the list field being compared
            other_list: Predicted list to compare with

        Returns:
            Dictionary with:
            - Top-level TP, FP, TN, FN, FD, FA counts and derived metrics for the list field
            - nested_fields: Dict with metrics for individual fields within list items (e.g., "transactions.date")
            - non_matches: List of individual object-level non-matches for detailed analysis
        """
        from .confusion_matrix_calculator import ConfusionMatrixCalculator
        calculator = ConfusionMatrixCalculator(self)
        return calculator.calculate_list_confusion_matrix(field_name, other_list)

    def _calculate_nested_field_metrics(
        self,
        list_field_name: str,
        gt_list: List["StructuredModel"],
        pred_list: List["StructuredModel"],
        threshold: float,
    ) -> Dict[str, Dict[str, Any]]:
        """Calculate confusion matrix metrics for individual fields within list items.

        This method delegates to ConfusionMatrixCalculator for the actual implementation.

        THRESHOLD-GATED RECURSION: Only perform recursive field analysis for object pairs
        with similarity >= StructuredModel.match_threshold. Poor matches and unmatched
        items are treated as atomic units.

        Args:
            list_field_name: Name of the parent list field (e.g., "transactions")
            gt_list: Ground truth list of StructuredModel objects
            pred_list: Predicted list of StructuredModel objects
            threshold: Matching threshold (not used for threshold-gating)

        Returns:
            Dictionary mapping nested field paths to their confusion matrix metrics
            E.g., {"transactions.date": {...}, "transactions.description": {...}}
        """
        from .confusion_matrix_calculator import ConfusionMatrixCalculator
        calculator = ConfusionMatrixCalculator(self)
        return calculator.calculate_nested_field_metrics(list_field_name, gt_list, pred_list, threshold)

    def _calculate_single_nested_field_metrics(
        self,
        parent_field_name: str,
        gt_nested: "StructuredModel",
        pred_nested: "StructuredModel",
        parent_is_aggregate: bool = False,
    ) -> Dict[str, Dict[str, Any]]:
        """Calculate confusion matrix metrics for fields within a single nested StructuredModel.

        This method delegates to ConfusionMatrixCalculator for the actual implementation.

        Args:
            parent_field_name: Name of the parent field (e.g., "address")
            gt_nested: Ground truth nested StructuredModel
            pred_nested: Predicted nested StructuredModel
            parent_is_aggregate: Whether the parent field should aggregate child metrics

        Returns:
            Dictionary mapping nested field paths to their confusion matrix metrics
            E.g., {"address.street": {...}, "address.city": {...}}
        """
        from .confusion_matrix_calculator import ConfusionMatrixCalculator
        calculator = ConfusionMatrixCalculator(self)
        return calculator.calculate_single_nested_field_metrics(
            parent_field_name, gt_nested, pred_nested, parent_is_aggregate
        )

    def _collect_enhanced_non_matches(
        self, recursive_result: dict, other: "StructuredModel"
    ) -> List[Dict[str, Any]]:
        """Collect enhanced non-matches with object-level granularity.

        This method delegates to NonMatchCollector for the actual implementation.

        Args:
            recursive_result: Result from compare_recursive containing field comparison details
            other: The predicted StructuredModel instance

        Returns:
            List of non-match dictionaries with enhanced object-level information
        """
        from .non_match_collector import NonMatchCollector
        collector = NonMatchCollector(self)
        return collector.collect_enhanced_non_matches(recursive_result, other)

    def _collect_non_matches(
        self, other: "StructuredModel", base_path: str = ""
    ) -> List[NonMatchField]:
        """Collect non-matches for detailed analysis.

        This method delegates to NonMatchCollector for the actual implementation.

        Args:
            other: Other model to compare with
            base_path: Base path for field naming (e.g., "address")

        Returns:
            List of NonMatchField objects documenting non-matches
        """
        from .non_match_collector import NonMatchCollector
        collector = NonMatchCollector(self)
        return collector.collect_non_matches(other, base_path)

    def compare(self, other: "StructuredModel") -> float:
        """Compare this model with another and return a scalar similarity score.

        Returns the overall weighted average score regardless of sufficient/necessary field matching.
        This provides a more nuanced score for use in comparators.

        Args:
            other: Another instance of the same model to compare with

        Returns:
            Similarity score between 0.0 and 1.0
        """
        # We'll calculate the overall weighted score directly instead of using compare_with
        # This ensures that sufficient/necessary field rules don't cause a zero score
        # when at least some fields match

        total_score = 0.0
        total_weight = 0.0

        for field_name in self.__class__.model_fields:
            # Skip the extra_fields attribute in comparison
            if field_name == "extra_fields":
                continue
            if hasattr(other, field_name):
                # Get field configuration
                info = self.__class__._get_comparison_info(field_name)
                # Use weight from ComparableField object
                weight = info.weight

                # Compare field values WITHOUT applying thresholds
                field_score = self.compare_field_raw(
                    field_name, getattr(other, field_name)
                )

                # Update total score
                total_score += field_score * weight
                total_weight += weight

        # Calculate overall score
        if total_weight > 0:
            return total_score / total_weight
        else:
            return 0.0

    def compare_with(
        self,
        other: "StructuredModel",
        include_confusion_matrix: bool = False,
        document_non_matches: bool = False,
        evaluator_format: bool = False,
        recall_with_fd: bool = False,
        add_derived_metrics: bool = True,
    ) -> Dict[str, Any]:
        """Compare this model with another instance using SINGLE TRAVERSAL optimization.

        PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

        Args:
            other: Another instance of the same model to compare with
            include_confusion_matrix: Whether to include confusion matrix calculations
            document_non_matches: Whether to document non-matches for analysis
            evaluator_format: Whether to format results for the evaluator
            recall_with_fd: If True, include FD in recall denominator (TP/(TP+FN+FD))
                            If False, use traditional recall (TP/(TP+FN))
            add_derived_metrics: Whether to add derived metrics to confusion matrix

        Returns:
            Dictionary with comparison results including:
            - field_scores: Scores for each field
            - overall_score: Weighted average score
            - all_fields_matched: Whether all fields matched
            - confusion_matrix: (optional) Confusion matrix data if requested
            - non_matches: (optional) Non-match documentation if requested
        """
        from .comparison_engine import ComparisonEngine
        engine = ComparisonEngine(self)
        return engine.compare_with(
            other,
            include_confusion_matrix=include_confusion_matrix,
            document_non_matches=document_non_matches,
            evaluator_format=evaluator_format,
            recall_with_fd=recall_with_fd,
            add_derived_metrics=add_derived_metrics,
        )

    def _convert_score_to_binary_metrics(
        self, score: float, threshold: float = 0.5
    ) -> Dict[str, float]:
        """Convert similarity score to binary classification metrics using MetricsHelper.

        Args:
            score: Similarity score [0-1]
            threshold: Threshold for considering a match

        Returns:
            Dictionary with TP, FP, FN, TN counts converted to metrics
        """
        metrics_helper = MetricsHelper()
        return metrics_helper.convert_score_to_binary_metrics(score, threshold)

    def _format_for_evaluator(
        self,
        result: Dict[str, Any],
        other: "StructuredModel",
        recall_with_fd: bool = False,
    ) -> Dict[str, Any]:
        """Format comparison results for evaluator compatibility.

        Args:
            result: Standard comparison result from compare_with
            other: The other model being compared
            recall_with_fd: Whether to include FD in recall denominator

        Returns:
            Dictionary in evaluator format with overall, fields, confusion_matrix
        """
        return EvaluatorFormatHelper.format_for_evaluator(
            self, result, other, recall_with_fd
        )

    def _calculate_list_item_metrics(
        self,
        field_name: str,
        gt_list: List[Any],
        pred_list: List[Any],
        recall_with_fd: bool = False,
    ) -> List[Dict[str, Any]]:
        """Calculate metrics for individual items in a list field.

        Args:
            field_name: Name of the list field
            gt_list: Ground truth list
            pred_list: Prediction list
            recall_with_fd: Whether to include FD in recall denominator

        Returns:
            List of metrics dictionaries for each matched item pair
        """
        return EvaluatorFormatHelper.calculate_list_item_metrics(
            field_name, gt_list, pred_list, recall_with_fd
        )

    @classmethod
    def model_json_schema(cls, **kwargs):
        """Override to add model-level comparison metadata.

        Extends the standard Pydantic JSON schema with comparison metadata
        at the field level.

        Args:
            **kwargs: Arguments to pass to the parent method

        Returns:
            JSON schema with added comparison metadata
        """
        schema = super().model_json_schema(**kwargs)

        # Add comparison metadata to each field in the schema
        for field_name, field_info in cls.model_fields.items():
            if field_name == "extra_fields":
                continue

            # Get the schema property for this field
            if field_name not in schema.get("properties", {}):
                continue

            field_props = schema["properties"][field_name]

            # Since ComparableField is now always a function, check for json_schema_extra
            if hasattr(field_info, "json_schema_extra") and callable(
                field_info.json_schema_extra
            ):
                # Fallback: Check for json_schema_extra function
                temp_schema = {}
                field_info.json_schema_extra(temp_schema)

                if "x-comparison" in temp_schema:
                    # Copy the comparison metadata from the temp schema to the real schema
                    field_props["x-comparison"] = temp_schema["x-comparison"]

        return schema

__init_subclass__(**kwargs)

Validate field configurations when a StructuredModel subclass is defined.

Source code in stickler/structured_object_evaluator/models/structured_model.py
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def __init_subclass__(cls, **kwargs):
    """Validate field configurations when a StructuredModel subclass is defined."""
    super().__init_subclass__(**kwargs)

    # Validate field configurations using class annotations since model_fields isn't populated yet
    if hasattr(cls, "__annotations__"):
        for field_name, field_type in cls.__annotations__.items():
            if field_name == "extra_fields":
                continue

            # Get the field default value if it exists
            field_default = getattr(cls, field_name, None)

            # Since ComparableField is now always a function that returns a Field,
            # we need to check if field_default has comparison metadata
            if hasattr(field_default, "json_schema_extra") and callable(
                field_default.json_schema_extra
            ):
                # Check for comparison metadata
                temp_schema = {}
                field_default.json_schema_extra(temp_schema)
                if "x-comparison" in temp_schema:
                    # This field was created with ComparableField function - validate constraints
                    if cls._is_list_of_structured_model_type(field_type):
                        comparison_config = temp_schema["x-comparison"]

                        # Threshold validation - only flag if explicitly set to non-default value
                        threshold = comparison_config.get("threshold", 0.5)
                        if threshold != 0.5:  # Default threshold value
                            raise ValueError(
                                f"Field '{field_name}' is a List[StructuredModel] and cannot have a "
                                f"'threshold' parameter in ComparableField. Hungarian matching uses each "
                                f"StructuredModel's 'match_threshold' class attribute instead. "
                                f"Set 'match_threshold = {threshold}' on the list element class."
                            )

                        # Comparator validation - only flag if explicitly set to non-default type
                        comparator_type = comparison_config.get(
                            "comparator_type", "LevenshteinComparator"
                        )
                        if (
                            comparator_type != "LevenshteinComparator"
                        ):  # Default comparator type
                            raise ValueError(
                                f"Field '{field_name}' is a List[StructuredModel] and cannot have a "
                                f"'comparator' parameter in ComparableField. Object comparison uses each "
                                f"StructuredModel's individual field comparators instead."
                            )

compare(other)

Compare this model with another and return a scalar similarity score.

Returns the overall weighted average score regardless of sufficient/necessary field matching. This provides a more nuanced score for use in comparators.

Parameters:

Name Type Description Default
other StructuredModel

Another instance of the same model to compare with

required

Returns:

Type Description
float

Similarity score between 0.0 and 1.0

Source code in stickler/structured_object_evaluator/models/structured_model.py
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
def compare(self, other: "StructuredModel") -> float:
    """Compare this model with another and return a scalar similarity score.

    Returns the overall weighted average score regardless of sufficient/necessary field matching.
    This provides a more nuanced score for use in comparators.

    Args:
        other: Another instance of the same model to compare with

    Returns:
        Similarity score between 0.0 and 1.0
    """
    # We'll calculate the overall weighted score directly instead of using compare_with
    # This ensures that sufficient/necessary field rules don't cause a zero score
    # when at least some fields match

    total_score = 0.0
    total_weight = 0.0

    for field_name in self.__class__.model_fields:
        # Skip the extra_fields attribute in comparison
        if field_name == "extra_fields":
            continue
        if hasattr(other, field_name):
            # Get field configuration
            info = self.__class__._get_comparison_info(field_name)
            # Use weight from ComparableField object
            weight = info.weight

            # Compare field values WITHOUT applying thresholds
            field_score = self.compare_field_raw(
                field_name, getattr(other, field_name)
            )

            # Update total score
            total_score += field_score * weight
            total_weight += weight

    # Calculate overall score
    if total_weight > 0:
        return total_score / total_weight
    else:
        return 0.0

compare_field(field_name, other_value)

Compare a single field with a value using the configured comparator.

Parameters:

Name Type Description Default
field_name str

Name of the field to compare

required
other_value Any

Value to compare with

required

Returns:

Type Description
float

Similarity score between 0.0 and 1.0

Source code in stickler/structured_object_evaluator/models/structured_model.py
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
def compare_field(self, field_name: str, other_value: Any) -> float:
    """Compare a single field with a value using the configured comparator.

    Args:
        field_name: Name of the field to compare
        other_value: Value to compare with

    Returns:
        Similarity score between 0.0 and 1.0
    """
    # Get our field value
    my_value = getattr(self, field_name)

    # If both values are StructuredModel instances, use recursive compare_with
    if isinstance(my_value, StructuredModel) and isinstance(
        other_value, StructuredModel
    ):
        # Use compare_with for rich comparison
        comparison_result = my_value.compare_with(
            other_value,
            include_confusion_matrix=False,
            document_non_matches=False,
            evaluator_format=False,
            recall_with_fd=False,
        )
        # Apply field-level threshold if configured
        info = self._get_comparison_info(field_name)
        raw_score = comparison_result["overall_score"]
        return (
            raw_score
            if raw_score >= info.threshold or not info.clip_under_threshold
            else 0.0
        )

    # CRITICAL FIX: For lists, don't clip under threshold for partial matches
    if isinstance(my_value, list) and isinstance(other_value, list):
        # Get field info
        info = self._get_comparison_info(field_name)

        # Use the raw comparison result without threshold clipping for lists
        result = ComparisonHelper.compare_unordered_lists(
            my_value, other_value, info.comparator, info.threshold
        )

        # Return the overall score directly (don't clip based on threshold for lists)
        return result["overall_score"]

    # For other fields, use existing logic
    return ComparisonHelper.compare_field_with_threshold(
        self, field_name, other_value
    )

compare_field_raw(field_name, other_value)

Compare a single field with a value WITHOUT applying thresholds.

This version is used by the compare method to get raw similarity scores.

Parameters:

Name Type Description Default
field_name str

Name of the field to compare

required
other_value Any

Value to compare with

required

Returns:

Type Description
float

Raw similarity score between 0.0 and 1.0 without threshold filtering

Source code in stickler/structured_object_evaluator/models/structured_model.py
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
def compare_field_raw(self, field_name: str, other_value: Any) -> float:
    """Compare a single field with a value WITHOUT applying thresholds.

    This version is used by the compare method to get raw similarity scores.

    Args:
        field_name: Name of the field to compare
        other_value: Value to compare with

    Returns:
        Raw similarity score between 0.0 and 1.0 without threshold filtering
    """
    # Get our field value
    my_value = getattr(self, field_name)

    # If both values are StructuredModel instances, use recursive compare_with
    if isinstance(my_value, StructuredModel) and isinstance(
        other_value, StructuredModel
    ):
        # Use compare_with for rich comparison, but extract the raw score
        comparison_result = my_value.compare_with(
            other_value,
            include_confusion_matrix=False,
            document_non_matches=False,
            evaluator_format=False,
            recall_with_fd=False,
        )
        return comparison_result["overall_score"]

    # For non-StructuredModel fields, use existing logic
    return ComparisonHelper.compare_field_raw(self, field_name, other_value)

compare_recursive(other)

The ONE clean recursive function that handles everything.

Enhanced to capture BOTH confusion matrix metrics AND similarity scores in a single traversal to eliminate double traversal inefficiency.

PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

Parameters:

Name Type Description Default
other StructuredModel

Another instance of the same model to compare with

required

Returns:

Type Description
dict

Dictionary with clean hierarchical structure:

dict
  • overall: TP, FP, TN, FN, FD, FA counts + similarity_score + all_fields_matched
dict
  • fields: Recursive structure for each field with scores
dict
  • non_matches: List of non-matching items
Source code in stickler/structured_object_evaluator/models/structured_model.py
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
def compare_recursive(self, other: "StructuredModel") -> dict:
    """The ONE clean recursive function that handles everything.

    Enhanced to capture BOTH confusion matrix metrics AND similarity scores
    in a single traversal to eliminate double traversal inefficiency.

    PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

    Args:
        other: Another instance of the same model to compare with

    Returns:
        Dictionary with clean hierarchical structure:
        - overall: TP, FP, TN, FN, FD, FA counts + similarity_score + all_fields_matched
        - fields: Recursive structure for each field with scores
        - non_matches: List of non-matching items
    """
    from .comparison_engine import ComparisonEngine
    engine = ComparisonEngine(self)
    return engine.compare_recursive(other)

compare_with(other, include_confusion_matrix=False, document_non_matches=False, evaluator_format=False, recall_with_fd=False, add_derived_metrics=True)

Compare this model with another instance using SINGLE TRAVERSAL optimization.

PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

Parameters:

Name Type Description Default
other StructuredModel

Another instance of the same model to compare with

required
include_confusion_matrix bool

Whether to include confusion matrix calculations

False
document_non_matches bool

Whether to document non-matches for analysis

False
evaluator_format bool

Whether to format results for the evaluator

False
recall_with_fd bool

If True, include FD in recall denominator (TP/(TP+FN+FD)) If False, use traditional recall (TP/(TP+FN))

False
add_derived_metrics bool

Whether to add derived metrics to confusion matrix

True

Returns:

Type Description
Dict[str, Any]

Dictionary with comparison results including:

Dict[str, Any]
  • field_scores: Scores for each field
Dict[str, Any]
  • overall_score: Weighted average score
Dict[str, Any]
  • all_fields_matched: Whether all fields matched
Dict[str, Any]
  • confusion_matrix: (optional) Confusion matrix data if requested
Dict[str, Any]
  • non_matches: (optional) Non-match documentation if requested
Source code in stickler/structured_object_evaluator/models/structured_model.py
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
def compare_with(
    self,
    other: "StructuredModel",
    include_confusion_matrix: bool = False,
    document_non_matches: bool = False,
    evaluator_format: bool = False,
    recall_with_fd: bool = False,
    add_derived_metrics: bool = True,
) -> Dict[str, Any]:
    """Compare this model with another instance using SINGLE TRAVERSAL optimization.

    PHASE 2: Delegates to ComparisonEngine while maintaining identical behavior.

    Args:
        other: Another instance of the same model to compare with
        include_confusion_matrix: Whether to include confusion matrix calculations
        document_non_matches: Whether to document non-matches for analysis
        evaluator_format: Whether to format results for the evaluator
        recall_with_fd: If True, include FD in recall denominator (TP/(TP+FN+FD))
                        If False, use traditional recall (TP/(TP+FN))
        add_derived_metrics: Whether to add derived metrics to confusion matrix

    Returns:
        Dictionary with comparison results including:
        - field_scores: Scores for each field
        - overall_score: Weighted average score
        - all_fields_matched: Whether all fields matched
        - confusion_matrix: (optional) Confusion matrix data if requested
        - non_matches: (optional) Non-match documentation if requested
    """
    from .comparison_engine import ComparisonEngine
    engine = ComparisonEngine(self)
    return engine.compare_with(
        other,
        include_confusion_matrix=include_confusion_matrix,
        document_non_matches=document_non_matches,
        evaluator_format=evaluator_format,
        recall_with_fd=recall_with_fd,
        add_derived_metrics=add_derived_metrics,
    )

from_json(json_data) classmethod

Create a StructuredModel instance from JSON data.

This method handles missing fields gracefully and stores extra fields in the extra_fields attribute.

Parameters:

Name Type Description Default
json_data Dict[str, Any]

Dictionary containing the JSON data

required

Returns:

Type Description
StructuredModel

StructuredModel instance created from the JSON data

Source code in stickler/structured_object_evaluator/models/structured_model.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@classmethod
def from_json(cls, json_data: Dict[str, Any]) -> "StructuredModel":
    """Create a StructuredModel instance from JSON data.

    This method handles missing fields gracefully and stores extra fields
    in the extra_fields attribute.

    Args:
        json_data: Dictionary containing the JSON data

    Returns:
        StructuredModel instance created from the JSON data
    """
    return ConfigurationHelper.from_json(cls, json_data)

from_json_schema(schema) classmethod

Create a StructuredModel subclass from a JSON Schema document.

This method accepts standard JSON Schema documents and creates fully functional StructuredModel classes with comparison capabilities. Supports JSON Schema draft-07+.

Comparison behavior can be customized using x-aws-stickler-* extension fields:

Field-Level Extensions:
  • x-aws-stickler-comparator: Comparator algorithm name (built-in or registered custom)
  • x-aws-stickler-threshold: Similarity threshold for match/no-match (0.0-1.0, default: 0.5)
  • x-aws-stickler-weight: Field importance in overall scoring (>0.0, default: 1.0)
  • x-aws-stickler-clip-under-threshold: Clip scores below threshold to 0.0 (bool, default: false)
  • x-aws-stickler-aggregate: Include field metrics in parent aggregation (bool, default: false)
Model-Level Extensions:
  • x-aws-stickler-model-name: Generated class name (default: "DynamicModel")
  • x-aws-stickler-match-threshold: Overall match threshold (default: 0.7)
Supported Features:
  • Primitive types: string, number, integer, boolean
  • Nested objects and arrays (primitive/object items)
  • Required fields, defaults, descriptions
  • Schema references ($ref with #/definitions/ and #/$defs/)
Default Type Mappings:
  • string → LevenshteinComparator (threshold: 0.5)
  • number/integer → NumericComparator (threshold: 0.5)
  • boolean → ExactComparator (threshold: 1.0)
  • arrays → Hungarian matching with element-appropriate comparators
  • objects → Recursive field-by-field comparison

Parameters:

Name Type Description Default
schema Dict[str, Any]

JSON Schema document as a dictionary

required

Returns:

Type Description
Type[StructuredModel]

StructuredModel subclass created from the schema

Raises:

Type Description
ValueError

If schema is invalid or contains unsupported features

SchemaError

If schema doesn't conform to JSON Schema spec

Examples:

Basic usage with standard JSON Schema:

>>> schema = {
...     "type": "object",
...     "properties": {
...         "name": {"type": "string"},
...         "age": {"type": "integer"},
...         "email": {"type": "string"}
...     },
...     "required": ["name", "email"]
... }
>>> PersonModel = StructuredModel.from_json_schema(schema)
>>> person1 = PersonModel(name="Alice", age=30, email="alice@example.com")
>>> person2 = PersonModel(name="Alicia", age=30, email="alice@example.com")
>>> result = person1.compare_with(person2)
>>> # name field uses LevenshteinComparator, age uses NumericComparator

Advanced usage with x-aws-stickler-* extensions:

>>> schema = {
...     "type": "object",
...     "x-aws-stickler-model-name": "Product",
...     "x-aws-stickler-match-threshold": 0.8,
...     "properties": {
...         "name": {
...             "type": "string",
...             "x-aws-stickler-comparator": "LevenshteinComparator",
...             "x-aws-stickler-threshold": 0.9,
...             "x-aws-stickler-weight": 2.0,
...             "x-aws-stickler-aggregate": true
...         },
...         "price": {
...             "type": "number",
...             "x-aws-stickler-comparator": "NumericComparator",
...             "x-aws-stickler-threshold": 0.95,
...             "x-aws-stickler-clip-under-threshold": true
...         }
...     },
...     "required": ["name"]
... }
>>> ProductModel = StructuredModel.from_json_schema(schema)
>>> result = product1.compare_with(product2)
>>> # name field has weight=2.0, price field clips scores below 0.95
Source code in stickler/structured_object_evaluator/models/structured_model.py
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
@classmethod
def from_json_schema(cls, schema: Dict[str, Any]) -> Type["StructuredModel"]:
    """Create a StructuredModel subclass from a JSON Schema document.

    This method accepts standard JSON Schema documents and creates fully functional
    StructuredModel classes with comparison capabilities. Supports JSON Schema draft-07+.

    Comparison behavior can be customized using x-aws-stickler-* extension fields:

    Field-Level Extensions:
    -----------------------
    - x-aws-stickler-comparator: Comparator algorithm name (built-in or registered custom)
    - x-aws-stickler-threshold: Similarity threshold for match/no-match (0.0-1.0, default: 0.5)
    - x-aws-stickler-weight: Field importance in overall scoring (>0.0, default: 1.0)
    - x-aws-stickler-clip-under-threshold: Clip scores below threshold to 0.0 (bool, default: false)
    - x-aws-stickler-aggregate: Include field metrics in parent aggregation (bool, default: false)

    Model-Level Extensions:
    -----------------------
    - x-aws-stickler-model-name: Generated class name (default: "DynamicModel")
    - x-aws-stickler-match-threshold: Overall match threshold (default: 0.7)

    Supported Features:
    -------------------
    - Primitive types: string, number, integer, boolean
    - Nested objects and arrays (primitive/object items)
    - Required fields, defaults, descriptions
    - Schema references ($ref with #/definitions/ and #/$defs/)

    Default Type Mappings:
    ----------------------
    - string → LevenshteinComparator (threshold: 0.5)
    - number/integer → NumericComparator (threshold: 0.5)
    - boolean → ExactComparator (threshold: 1.0)
    - arrays → Hungarian matching with element-appropriate comparators
    - objects → Recursive field-by-field comparison

    Args:
        schema: JSON Schema document as a dictionary

    Returns:
        StructuredModel subclass created from the schema

    Raises:
        ValueError: If schema is invalid or contains unsupported features
        jsonschema.exceptions.SchemaError: If schema doesn't conform to JSON Schema spec

    Examples:
        Basic usage with standard JSON Schema:
        >>> schema = {
        ...     "type": "object",
        ...     "properties": {
        ...         "name": {"type": "string"},
        ...         "age": {"type": "integer"},
        ...         "email": {"type": "string"}
        ...     },
        ...     "required": ["name", "email"]
        ... }
        >>> PersonModel = StructuredModel.from_json_schema(schema)
        >>> person1 = PersonModel(name="Alice", age=30, email="alice@example.com")
        >>> person2 = PersonModel(name="Alicia", age=30, email="alice@example.com")
        >>> result = person1.compare_with(person2)
        >>> # name field uses LevenshteinComparator, age uses NumericComparator

        Advanced usage with x-aws-stickler-* extensions:
        >>> schema = {
        ...     "type": "object",
        ...     "x-aws-stickler-model-name": "Product",
        ...     "x-aws-stickler-match-threshold": 0.8,
        ...     "properties": {
        ...         "name": {
        ...             "type": "string",
        ...             "x-aws-stickler-comparator": "LevenshteinComparator",
        ...             "x-aws-stickler-threshold": 0.9,
        ...             "x-aws-stickler-weight": 2.0,
        ...             "x-aws-stickler-aggregate": true
        ...         },
        ...         "price": {
        ...             "type": "number",
        ...             "x-aws-stickler-comparator": "NumericComparator",
        ...             "x-aws-stickler-threshold": 0.95,
        ...             "x-aws-stickler-clip-under-threshold": true
        ...         }
        ...     },
        ...     "required": ["name"]
        ... }
        >>> ProductModel = StructuredModel.from_json_schema(schema)
        >>> result = product1.compare_with(product2)
        >>> # name field has weight=2.0, price field clips scores below 0.95
        """

    return cls._from_json_schema_internal(schema, field_path="")

model_from_json(config) classmethod

Create a StructuredModel subclass from JSON configuration using Pydantic's create_model().

This method leverages Pydantic's native dynamic model creation capabilities to ensure full compatibility with all Pydantic features while adding structured comparison functionality through inherited StructuredModel methods.

The generated model inherits all StructuredModel capabilities: - compare_with() method for detailed comparisons - Field-level comparison configuration - Hungarian algorithm for list matching - Confusion matrix generation - JSON schema with comparison metadata

Parameters:

Name Type Description Default
config Dict[str, Any]

JSON configuration with fields, comparators, and model settings. Required keys: - fields: Dict mapping field names to field configurations Optional keys: - model_name: Name for the generated class (default: "DynamicModel") - match_threshold: Overall matching threshold (default: 0.7)

Field configuration format: { "type": "str|int|float|bool|List[str]|etc.", # Required "comparator": "LevenshteinComparator|ExactComparator|etc.", # Optional "threshold": 0.8, # Optional, default 0.5 "weight": 2.0, # Optional, default 1.0 "required": true, # Optional, default false "default": "value", # Optional "description": "Field description", # Optional "alias": "field_alias", # Optional "examples": ["example1", "example2"] # Optional }

required

Returns:

Type Description
Type[StructuredModel]

A fully functional StructuredModel subclass created with create_model()

Raises:

Type Description
ValueError

If configuration is invalid or contains unsupported types/comparators

KeyError

If required configuration keys are missing

Examples:

>>> config = {
...     "model_name": "Product",
...     "match_threshold": 0.8,
...     "fields": {
...         "name": {
...             "type": "str",
...             "comparator": "LevenshteinComparator",
...             "threshold": 0.8,
...             "weight": 2.0,
...             "required": True
...         },
...         "price": {
...             "type": "float",
...             "comparator": "NumericComparator",
...             "default": 0.0
...         }
...     }
... }
>>> ProductClass = StructuredModel.model_from_json(config)
>>> isinstance(ProductClass.model_fields, dict)  # Full Pydantic compatibility
True
>>> product = ProductClass(name="Widget", price=29.99)
>>> product.name
'Widget'
>>> result = product.compare_with(ProductClass(name="Widget", price=29.99))
>>> result["overall_score"]
1.0
Source code in stickler/structured_object_evaluator/models/structured_model.py
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
@classmethod
def model_from_json(cls, config: Dict[str, Any]) -> Type["StructuredModel"]:
    """Create a StructuredModel subclass from JSON configuration using Pydantic's create_model().

    This method leverages Pydantic's native dynamic model creation capabilities to ensure
    full compatibility with all Pydantic features while adding structured comparison
    functionality through inherited StructuredModel methods.

    The generated model inherits all StructuredModel capabilities:
    - compare_with() method for detailed comparisons
    - Field-level comparison configuration
    - Hungarian algorithm for list matching
    - Confusion matrix generation
    - JSON schema with comparison metadata

    Args:
        config: JSON configuration with fields, comparators, and model settings.
               Required keys:
               - fields: Dict mapping field names to field configurations
               Optional keys:
               - model_name: Name for the generated class (default: "DynamicModel")
               - match_threshold: Overall matching threshold (default: 0.7)

               Field configuration format:
               {
                   "type": "str|int|float|bool|List[str]|etc.",  # Required
                   "comparator": "LevenshteinComparator|ExactComparator|etc.",  # Optional
                   "threshold": 0.8,  # Optional, default 0.5
                   "weight": 2.0,     # Optional, default 1.0
                   "required": true,  # Optional, default false
                   "default": "value", # Optional
                   "description": "Field description",  # Optional
                   "alias": "field_alias",  # Optional
                   "examples": ["example1", "example2"]  # Optional
               }

    Returns:
        A fully functional StructuredModel subclass created with create_model()

    Raises:
        ValueError: If configuration is invalid or contains unsupported types/comparators
        KeyError: If required configuration keys are missing

    Examples:
        >>> config = {
        ...     "model_name": "Product",
        ...     "match_threshold": 0.8,
        ...     "fields": {
        ...         "name": {
        ...             "type": "str",
        ...             "comparator": "LevenshteinComparator",
        ...             "threshold": 0.8,
        ...             "weight": 2.0,
        ...             "required": True
        ...         },
        ...         "price": {
        ...             "type": "float",
        ...             "comparator": "NumericComparator",
        ...             "default": 0.0
        ...         }
        ...     }
        ... }
        >>> ProductClass = StructuredModel.model_from_json(config)
        >>> isinstance(ProductClass.model_fields, dict)  # Full Pydantic compatibility
        True
        >>> product = ProductClass(name="Widget", price=29.99)
        >>> product.name
        'Widget'
        >>> result = product.compare_with(ProductClass(name="Widget", price=29.99))
        >>> result["overall_score"]
        1.0
    """
    # Delegate to ModelFactory for dynamic model creation
    from .model_factory import ModelFactory

    return ModelFactory.create_model_from_json(config, base_class=cls)

model_json_schema(**kwargs) classmethod

Override to add model-level comparison metadata.

Extends the standard Pydantic JSON schema with comparison metadata at the field level.

Parameters:

Name Type Description Default
**kwargs

Arguments to pass to the parent method

{}

Returns:

Type Description

JSON schema with added comparison metadata

Source code in stickler/structured_object_evaluator/models/structured_model.py
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
@classmethod
def model_json_schema(cls, **kwargs):
    """Override to add model-level comparison metadata.

    Extends the standard Pydantic JSON schema with comparison metadata
    at the field level.

    Args:
        **kwargs: Arguments to pass to the parent method

    Returns:
        JSON schema with added comparison metadata
    """
    schema = super().model_json_schema(**kwargs)

    # Add comparison metadata to each field in the schema
    for field_name, field_info in cls.model_fields.items():
        if field_name == "extra_fields":
            continue

        # Get the schema property for this field
        if field_name not in schema.get("properties", {}):
            continue

        field_props = schema["properties"][field_name]

        # Since ComparableField is now always a function, check for json_schema_extra
        if hasattr(field_info, "json_schema_extra") and callable(
            field_info.json_schema_extra
        ):
            # Fallback: Check for json_schema_extra function
            temp_schema = {}
            field_info.json_schema_extra(temp_schema)

            if "x-comparison" in temp_schema:
                # Copy the comparison metadata from the temp schema to the real schema
                field_props["x-comparison"] = temp_schema["x-comparison"]

    return schema

stickler.structured_object_evaluator.models.comparable_field

Field module for structured model evaluation.

This module provides the ComparableField function for creating fields in structured models with comparison configuration parameters.

stickler.structured_object_evaluator.models.comparable_field.ComparableField(comparator=None, threshold=0.5, weight=1.0, default=None, aggregate=False, clip_under_threshold=True, alias=None, description=None, examples=None, **field_kwargs)

Create a Pydantic Field with comparison metadata.

This function creates a proper Pydantic Field with embedded comparison configuration, enabling both comparison functionality and native Pydantic features like aliases.

Parameters:

Name Type Description Default
comparator Optional[BaseComparator]

Comparator to use for field comparison (default: LevenshteinComparator)

None
threshold float

Minimum similarity score to consider a match (default: 0.5)

0.5
weight float

Weight of this field in overall score calculation (default: 1.0)

1.0
default Any

Default value for the field (default: None)

None
aggregate bool

DEPRECATED - This parameter is deprecated and will be removed in a future version. Use the new universal 'aggregate' field in compare_with() output instead.

False
clip_under_threshold bool

Whether to zero out scores below threshold (default: True)

True
alias Optional[str]

Pydantic field alias for serialization (default: None)

None
description Optional[str]

Field description for documentation (default: None)

None
examples Optional[list]

Example values for the field (default: None)

None
**field_kwargs

Additional Pydantic Field arguments

{}

Returns:

Type Description

Pydantic Field with embedded comparison metadata

Example

class MyModel(StructuredModel): # Basic usage (no alias): name: str = ComparableField(threshold=0.8)

# With alias (new feature):
email: str = ComparableField(
    threshold=0.9,
    alias="email_address",
    description="User's email",
    examples=["user@example.com"]
)
Source code in stickler/structured_object_evaluator/models/comparable_field.py
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def ComparableField(
    comparator: Optional[BaseComparator] = None,
    threshold: float = 0.5,
    weight: float = 1.0,
    default: Any = None,
    aggregate: bool = False,
    clip_under_threshold: bool = True,
    # Pydantic Field parameters (all optional, just like Field)
    alias: Optional[str] = None,
    description: Optional[str] = None,
    examples: Optional[list] = None,
    **field_kwargs,
):
    """Create a Pydantic Field with comparison metadata.

    This function creates a proper Pydantic Field with embedded comparison configuration,
    enabling both comparison functionality and native Pydantic features like aliases.

    Args:
        comparator: Comparator to use for field comparison (default: LevenshteinComparator)
        threshold: Minimum similarity score to consider a match (default: 0.5)
        weight: Weight of this field in overall score calculation (default: 1.0)
        default: Default value for the field (default: None)
        aggregate: DEPRECATED - This parameter is deprecated and will be removed in a future version.
                  Use the new universal 'aggregate' field in compare_with() output instead.
        clip_under_threshold: Whether to zero out scores below threshold (default: True)
        alias: Pydantic field alias for serialization (default: None)
        description: Field description for documentation (default: None)
        examples: Example values for the field (default: None)
        **field_kwargs: Additional Pydantic Field arguments

    Returns:
        Pydantic Field with embedded comparison metadata

    Example:
        class MyModel(StructuredModel):
            # Basic usage (no alias):
            name: str = ComparableField(threshold=0.8)

            # With alias (new feature):
            email: str = ComparableField(
                threshold=0.9,
                alias="email_address",
                description="User's email",
                examples=["user@example.com"]
            )
    """
    # Issue deprecation warning if aggregate=True is used
    if aggregate:
        warnings.warn(
            "The 'aggregate' parameter in ComparableField is deprecated and will be removed "
            "in a future version. All nodes now automatically include an 'aggregate' field "
            "in the compare_with() output that sums primitive field metrics below that node.",
            DeprecationWarning,
            stacklevel=2,
        )

    # Create the actual comparator instance
    actual_comparator = comparator or LevenshteinComparator()

    # Create serializable metadata for JSON schema compatibility
    serializable_metadata = {
        "comparator_type": actual_comparator.__class__.__name__,
        "comparator_name": getattr(actual_comparator, "name", "unknown"),
        "comparator_config": getattr(actual_comparator, "config", {}),
        "threshold": threshold,
        "weight": weight,
        "clip_under_threshold": clip_under_threshold,
        "aggregate": aggregate,
    }

    # Create json_schema_extra function that stores runtime data
    def json_schema_extra_func(schema: Dict[str, Any]) -> None:
        schema["x-comparison"] = serializable_metadata

    # HYBRID APPROACH: Store runtime instances as function attributes
    # This works around FieldInfo's __slots__ restriction
    json_schema_extra_func._comparator_instance = actual_comparator
    json_schema_extra_func._threshold = threshold
    json_schema_extra_func._weight = weight
    json_schema_extra_func._clip_under_threshold = clip_under_threshold
    json_schema_extra_func._aggregate = aggregate
    json_schema_extra_func._comparison_metadata = serializable_metadata

    # Merge with existing json_schema_extra if provided
    existing_json_schema_extra = field_kwargs.get("json_schema_extra", {})
    if callable(existing_json_schema_extra):

        def enhanced_json_schema_extra(schema: Dict[str, Any]) -> None:
            existing_json_schema_extra(schema)
            json_schema_extra_func(schema)

        # Copy our runtime data to the enhanced function
        enhanced_json_schema_extra._comparator_instance = actual_comparator
        enhanced_json_schema_extra._threshold = threshold
        enhanced_json_schema_extra._weight = weight
        enhanced_json_schema_extra._clip_under_threshold = clip_under_threshold
        enhanced_json_schema_extra._aggregate = aggregate
        enhanced_json_schema_extra._comparison_metadata = serializable_metadata
        final_json_schema_extra = enhanced_json_schema_extra
    elif isinstance(existing_json_schema_extra, dict):

        def enhanced_json_schema_extra(schema: Dict[str, Any]) -> None:
            schema.update(existing_json_schema_extra)
            json_schema_extra_func(schema)

        # Copy our runtime data to the enhanced function
        enhanced_json_schema_extra._comparator_instance = actual_comparator
        enhanced_json_schema_extra._threshold = threshold
        enhanced_json_schema_extra._weight = weight
        enhanced_json_schema_extra._clip_under_threshold = clip_under_threshold
        enhanced_json_schema_extra._aggregate = aggregate
        enhanced_json_schema_extra._comparison_metadata = serializable_metadata
        final_json_schema_extra = enhanced_json_schema_extra
    else:
        final_json_schema_extra = json_schema_extra_func

    # Remove json_schema_extra from field_kwargs to avoid duplication
    clean_field_kwargs = {
        k: v for k, v in field_kwargs.items() if k != "json_schema_extra"
    }

    # Create the Field
    field = Field(
        default=default,
        alias=alias,
        description=description,
        examples=examples,
        json_schema_extra=final_json_schema_extra,
        **clean_field_kwargs,
    )

    return field

stickler.structured_object_evaluator.models.non_match_field

Models for documenting non-matches in structured object evaluation.

This module provides data models for documenting and tracking non-matches (false positives, false negatives, etc.) during structured object evaluation. It also includes utilities for filtering, exporting, and analyzing non-matches.

stickler.structured_object_evaluator.models.non_match_field.NonMatchField

Bases: BaseModel

Model for documenting non-matches in structured object evaluation.

This class stores detailed information about each non-match detected during the evaluation process, enabling more thorough analysis and debugging of evaluation results.

Source code in stickler/structured_object_evaluator/models/non_match_field.py
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class NonMatchField(BaseModel):
    """Model for documenting non-matches in structured object evaluation.

    This class stores detailed information about each non-match detected
    during the evaluation process, enabling more thorough analysis and
    debugging of evaluation results.
    """

    field_path: str = Field(
        description="Dot-notation path to the field (e.g., 'address.city')"
    )
    non_match_type: NonMatchType = Field(description="Type of non-match")
    ground_truth_value: Any = Field(description="Original ground truth value")
    prediction_value: Any = Field(description="Predicted value")
    similarity_score: Optional[float] = Field(
        default=None, description="Similarity score if available"
    )
    details: Dict[str, Any] = Field(
        default_factory=dict, description="Additional context or details"
    )
    document_id: Optional[str] = Field(
        default=None, description="ID of the document this non-match belongs to"
    )

    def __str__(self) -> str:
        """Return a string representation of the non-match document."""
        similarity_str = (
            f", similarity: {self.similarity_score:.4f}"
            if self.similarity_score is not None
            else ""
        )
        doc_id_str = f" (doc: {self.document_id})" if self.document_id else ""
        return (
            f"{self.non_match_type.value.upper()} at '{self.field_path}'{similarity_str}{doc_id_str}\n"
            f"  GT: {self.ground_truth_value}\n"
            f"  Pred: {self.prediction_value}"
        )

    @staticmethod
    def filter_by_type(
        documents: List["NonMatchField"], match_type: NonMatchType
    ) -> List["NonMatchField"]:
        """
        Filter non-match documents by their type.

        Args:
            documents: List of NonMatchField instances to filter
            match_type: Type of non-match to filter for

        Returns:
            Filtered list of NonMatchField instances
        """
        return [doc for doc in documents if doc.non_match_type == match_type]

    @staticmethod
    def export_to_dict(
        documents: List["NonMatchField"],
    ) -> Dict[str, List[Dict[str, Any]]]:
        """
        Export a list of non-match documents to a dictionary for serialization.

        Args:
            documents: List of NonMatchField instances

        Returns:
            Dictionary with categorized non-matches
        """
        result = {"false_alarms": [], "false_discoveries": [], "false_negatives": []}

        for doc in documents:
            # Create a simplified entry
            entry = {
                "field_path": doc.field_path,
                "ground_truth": str(doc.ground_truth_value),
                "prediction": str(doc.prediction_value),
            }

            if doc.similarity_score is not None:
                entry["similarity_score"] = doc.similarity_score

            if doc.details:
                entry["details"] = doc.details

            if doc.non_match_type == NonMatchType.FALSE_ALARM:
                result["false_alarms"].append(entry)
            elif doc.non_match_type == NonMatchType.FALSE_DISCOVERY:
                result["false_discoveries"].append(entry)
            elif doc.non_match_type == NonMatchType.FALSE_NEGATIVE:
                result["false_negatives"].append(entry)

        return result

    @staticmethod
    def export_to_json(documents: List["NonMatchField"], output_path: str):
        """
        Export a list of non-match documents to a JSON file.

        Args:
            documents: List of NonMatchField instances
            output_path: Path to save the JSON file
        """
        # Create parent directories if needed
        Path(output_path).parent.mkdir(parents=True, exist_ok=True)

        # Export as dictionary
        data = NonMatchField.export_to_dict(documents)

        # Write to file
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2)

    @staticmethod
    def print_summary(documents: List["NonMatchField"], detailed: bool = False):
        """
        Print a summary of non-match documents.

        Args:
            documents: List of NonMatchField instances
            detailed: Whether to print detailed information for each document
        """
        # Count by type
        false_alarms = NonMatchField.filter_by_type(documents, NonMatchType.FALSE_ALARM)
        false_discoveries = NonMatchField.filter_by_type(
            documents, NonMatchType.FALSE_DISCOVERY
        )
        false_negatives = NonMatchField.filter_by_type(
            documents, NonMatchType.FALSE_NEGATIVE
        )

        # Print summary counts
        print(f"Non-matches summary:")
        print(f"- False Alarms: {len(false_alarms)}")
        print(f"- False Discoveries: {len(false_discoveries)}")
        print(f"- False Negatives: {len(false_negatives)}")

        # Print details if requested
        if detailed and documents:
            print("\nDetailed non-matches:")
            for i, doc in enumerate(documents):
                print(f"\nNon-match #{i + 1}:")
                print(f"- Type: {doc.non_match_type}")
                print(f"- Field: {doc.field_path}")
                print(f"- Ground truth: {doc.ground_truth_value}")
                print(f"- Prediction: {doc.prediction_value}")
                if doc.similarity_score is not None:
                    print(f"- Similarity: {doc.similarity_score:.4f}")

__str__()

Return a string representation of the non-match document.

Source code in stickler/structured_object_evaluator/models/non_match_field.py
47
48
49
50
51
52
53
54
55
56
57
58
59
def __str__(self) -> str:
    """Return a string representation of the non-match document."""
    similarity_str = (
        f", similarity: {self.similarity_score:.4f}"
        if self.similarity_score is not None
        else ""
    )
    doc_id_str = f" (doc: {self.document_id})" if self.document_id else ""
    return (
        f"{self.non_match_type.value.upper()} at '{self.field_path}'{similarity_str}{doc_id_str}\n"
        f"  GT: {self.ground_truth_value}\n"
        f"  Pred: {self.prediction_value}"
    )

export_to_dict(documents) staticmethod

Export a list of non-match documents to a dictionary for serialization.

Parameters:

Name Type Description Default
documents List[NonMatchField]

List of NonMatchField instances

required

Returns:

Type Description
Dict[str, List[Dict[str, Any]]]

Dictionary with categorized non-matches

Source code in stickler/structured_object_evaluator/models/non_match_field.py
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@staticmethod
def export_to_dict(
    documents: List["NonMatchField"],
) -> Dict[str, List[Dict[str, Any]]]:
    """
    Export a list of non-match documents to a dictionary for serialization.

    Args:
        documents: List of NonMatchField instances

    Returns:
        Dictionary with categorized non-matches
    """
    result = {"false_alarms": [], "false_discoveries": [], "false_negatives": []}

    for doc in documents:
        # Create a simplified entry
        entry = {
            "field_path": doc.field_path,
            "ground_truth": str(doc.ground_truth_value),
            "prediction": str(doc.prediction_value),
        }

        if doc.similarity_score is not None:
            entry["similarity_score"] = doc.similarity_score

        if doc.details:
            entry["details"] = doc.details

        if doc.non_match_type == NonMatchType.FALSE_ALARM:
            result["false_alarms"].append(entry)
        elif doc.non_match_type == NonMatchType.FALSE_DISCOVERY:
            result["false_discoveries"].append(entry)
        elif doc.non_match_type == NonMatchType.FALSE_NEGATIVE:
            result["false_negatives"].append(entry)

    return result

export_to_json(documents, output_path) staticmethod

Export a list of non-match documents to a JSON file.

Parameters:

Name Type Description Default
documents List[NonMatchField]

List of NonMatchField instances

required
output_path str

Path to save the JSON file

required
Source code in stickler/structured_object_evaluator/models/non_match_field.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@staticmethod
def export_to_json(documents: List["NonMatchField"], output_path: str):
    """
    Export a list of non-match documents to a JSON file.

    Args:
        documents: List of NonMatchField instances
        output_path: Path to save the JSON file
    """
    # Create parent directories if needed
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)

    # Export as dictionary
    data = NonMatchField.export_to_dict(documents)

    # Write to file
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)

filter_by_type(documents, match_type) staticmethod

Filter non-match documents by their type.

Parameters:

Name Type Description Default
documents List[NonMatchField]

List of NonMatchField instances to filter

required
match_type NonMatchType

Type of non-match to filter for

required

Returns:

Type Description
List[NonMatchField]

Filtered list of NonMatchField instances

Source code in stickler/structured_object_evaluator/models/non_match_field.py
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def filter_by_type(
    documents: List["NonMatchField"], match_type: NonMatchType
) -> List["NonMatchField"]:
    """
    Filter non-match documents by their type.

    Args:
        documents: List of NonMatchField instances to filter
        match_type: Type of non-match to filter for

    Returns:
        Filtered list of NonMatchField instances
    """
    return [doc for doc in documents if doc.non_match_type == match_type]

print_summary(documents, detailed=False) staticmethod

Print a summary of non-match documents.

Parameters:

Name Type Description Default
documents List[NonMatchField]

List of NonMatchField instances

required
detailed bool

Whether to print detailed information for each document

False
Source code in stickler/structured_object_evaluator/models/non_match_field.py
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
@staticmethod
def print_summary(documents: List["NonMatchField"], detailed: bool = False):
    """
    Print a summary of non-match documents.

    Args:
        documents: List of NonMatchField instances
        detailed: Whether to print detailed information for each document
    """
    # Count by type
    false_alarms = NonMatchField.filter_by_type(documents, NonMatchType.FALSE_ALARM)
    false_discoveries = NonMatchField.filter_by_type(
        documents, NonMatchType.FALSE_DISCOVERY
    )
    false_negatives = NonMatchField.filter_by_type(
        documents, NonMatchType.FALSE_NEGATIVE
    )

    # Print summary counts
    print(f"Non-matches summary:")
    print(f"- False Alarms: {len(false_alarms)}")
    print(f"- False Discoveries: {len(false_discoveries)}")
    print(f"- False Negatives: {len(false_negatives)}")

    # Print details if requested
    if detailed and documents:
        print("\nDetailed non-matches:")
        for i, doc in enumerate(documents):
            print(f"\nNon-match #{i + 1}:")
            print(f"- Type: {doc.non_match_type}")
            print(f"- Field: {doc.field_path}")
            print(f"- Ground truth: {doc.ground_truth_value}")
            print(f"- Prediction: {doc.prediction_value}")
            if doc.similarity_score is not None:
                print(f"- Similarity: {doc.similarity_score:.4f}")

stickler.structured_object_evaluator.models.non_match_field.NonMatchType

Bases: str, Enum

Enum defining the types of non-matches.

Source code in stickler/structured_object_evaluator/models/non_match_field.py
15
16
17
18
19
20
class NonMatchType(str, Enum):
    """Enum defining the types of non-matches."""

    FALSE_ALARM = "false_alarm"  # GT null, prediction non-null
    FALSE_DISCOVERY = "false_discovery"  # Both non-null but don't match
    FALSE_NEGATIVE = "false_negative"  # GT non-null, prediction null

stickler.structured_object_evaluator.models.field

Field module for structured model evaluation.

This module contains the ComparableField class used to define fields in structured models with comparison configuration parameters.

stickler.structured_object_evaluator.models.field.CustomField

Bases: FieldInfo

Field with comparable properties for structured model evaluation.

This extends pydantic's Field with additional attributes that control how the field is compared during evaluation.

Attributes:

Name Type Description
comparator

The comparator to use for this field

threshold

The threshold for determining if values match

weight

The weight of this field in the overall score

description

Human-readable description of the field

Source code in stickler/structured_object_evaluator/models/field.py
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class CustomField(FieldInfo):
    """
    Field with comparable properties for structured model evaluation.

    This extends pydantic's Field with additional attributes that control how the field
    is compared during evaluation.

    Attributes:
        comparator: The comparator to use for this field
        threshold: The threshold for determining if values match
        weight: The weight of this field in the overall score
        description: Human-readable description of the field
    """

    def __init__(
        self,
        default: Any = ...,
        *,
        comparator: Optional[BaseComparator] = None,
        threshold: float = DEFAULT_THRESHOLD,
        weight: float = DEFAULT_WEIGHT,
        description: Optional[str] = None,
        **kwargs,
    ):
        """
        Initialize comparable field.

        Args:
            default: Default value for the field
            comparator: The comparator to use for this field
            threshold: The threshold for determining if values match
            weight: The weight of this field in the overall score
            description: Human-readable description of the field
            **kwargs: Additional field parameters
        """
        # Fix: Pass all kwargs together with default as keyword args,
        # since pydantic expects a specific format
        kwargs["default"] = default
        super().__init__(**kwargs)

        # Store comparison configuration
        self.comparator = comparator or LevenshteinComparator()
        self.threshold = threshold
        self.weight = weight
        self.description = description

    def get_config(self) -> Dict[str, Any]:
        """
        Get field configuration.

        Returns:
            Dictionary with field configuration
        """
        return {
            "comparator": self.comparator,
            "threshold": self.threshold,
            "weight": self.weight,
            "description": self.description,
        }

    def __repr__(self) -> str:
        """
        String representation.

        Returns:
            String representation
        """
        return (
            f"ComparableField("
            f"comparator={self.comparator}, "
            f"threshold={self.threshold}, "
            f"weight={self.weight})"
        )

__init__(default=..., *, comparator=None, threshold=DEFAULT_THRESHOLD, weight=DEFAULT_WEIGHT, description=None, **kwargs)

Initialize comparable field.

Parameters:

Name Type Description Default
default Any

Default value for the field

...
comparator Optional[BaseComparator]

The comparator to use for this field

None
threshold float

The threshold for determining if values match

DEFAULT_THRESHOLD
weight float

The weight of this field in the overall score

DEFAULT_WEIGHT
description Optional[str]

Human-readable description of the field

None
**kwargs

Additional field parameters

{}
Source code in stickler/structured_object_evaluator/models/field.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    default: Any = ...,
    *,
    comparator: Optional[BaseComparator] = None,
    threshold: float = DEFAULT_THRESHOLD,
    weight: float = DEFAULT_WEIGHT,
    description: Optional[str] = None,
    **kwargs,
):
    """
    Initialize comparable field.

    Args:
        default: Default value for the field
        comparator: The comparator to use for this field
        threshold: The threshold for determining if values match
        weight: The weight of this field in the overall score
        description: Human-readable description of the field
        **kwargs: Additional field parameters
    """
    # Fix: Pass all kwargs together with default as keyword args,
    # since pydantic expects a specific format
    kwargs["default"] = default
    super().__init__(**kwargs)

    # Store comparison configuration
    self.comparator = comparator or LevenshteinComparator()
    self.threshold = threshold
    self.weight = weight
    self.description = description

__repr__()

String representation.

Returns:

Type Description
str

String representation

Source code in stickler/structured_object_evaluator/models/field.py
78
79
80
81
82
83
84
85
86
87
88
89
90
def __repr__(self) -> str:
    """
    String representation.

    Returns:
        String representation
    """
    return (
        f"ComparableField("
        f"comparator={self.comparator}, "
        f"threshold={self.threshold}, "
        f"weight={self.weight})"
    )

get_config()

Get field configuration.

Returns:

Type Description
Dict[str, Any]

Dictionary with field configuration

Source code in stickler/structured_object_evaluator/models/field.py
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_config(self) -> Dict[str, Any]:
    """
    Get field configuration.

    Returns:
        Dictionary with field configuration
    """
    return {
        "comparator": self.comparator,
        "threshold": self.threshold,
        "weight": self.weight,
        "description": self.description,
    }

stickler.structured_object_evaluator.models.comparison_info

Comparison configuration for structured model fields.

stickler.structured_object_evaluator.models.comparison_info.ComparisonInfo

Container for comparison configuration.

This class holds the configuration for how a field should be compared, including which comparator to use, the threshold for considering a match, and the weight in scoring.

Attributes:

Name Type Description
comparator

The comparator to use for string similarity

threshold

Minimum score to consider a match (like ANLS threshold)

weight

Weight of this field in the overall score calculation

Source code in stickler/structured_object_evaluator/models/comparison_info.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class ComparisonInfo:
    """Container for comparison configuration.

    This class holds the configuration for how a field should be compared,
    including which comparator to use, the threshold for considering a match,
    and the weight in scoring.

    Attributes:
        comparator: The comparator to use for string similarity
        threshold: Minimum score to consider a match (like ANLS threshold)
        weight: Weight of this field in the overall score calculation
    """

    def __init__(
        self,
        comparator: Optional[BaseComparator] = None,
        threshold: float = 0.5,
        weight: float = 1.0,
    ):
        """Initialize comparison configuration.

        Args:
            comparator: Comparator to use (default: LevenshteinComparator)
            threshold: Minimum similarity score to consider a match (default: 0.5)
            weight: Weight of this field in the overall score (default: 1.0)
        """
        self.comparator = comparator or LevenshteinComparator()
        self.threshold = threshold
        self.weight = weight

    def compare(self, value1: Any, value2: Any) -> float:
        """Compare two values and return a similarity score between 0 and 1.

        Args:
            value1: First value to compare
            value2: Second value to compare

        Returns:
            Similarity score between 0.0 and 1.0, with 0.0 if below threshold
        """
        # Handle None values
        if value1 is None or value2 is None:
            return 1.0 if value1 == value2 else 0.0

        # Use the comparator to calculate similarity
        similarity = self.comparator.compare(value1, value2)

        # Apply threshold (if below threshold, return 0)
        return 0.0 if similarity < self.threshold else similarity

    def __repr__(self) -> str:
        """Return string representation."""
        return f"ComparisonInfo(comparator={self.comparator}, threshold={self.threshold}, weight={self.weight})"

    def to_dict(self) -> Dict[str, Any]:
        """Convert to a serializable dictionary for JSON schema."""
        return {
            "comparator_type": self.comparator.__class__.__name__,
            "comparator_name": getattr(self.comparator, "name", "unknown"),
            "comparator_config": getattr(self.comparator, "config", {}),
            "threshold": self.threshold,
            "weight": self.weight,
        }

__init__(comparator=None, threshold=0.5, weight=1.0)

Initialize comparison configuration.

Parameters:

Name Type Description Default
comparator Optional[BaseComparator]

Comparator to use (default: LevenshteinComparator)

None
threshold float

Minimum similarity score to consider a match (default: 0.5)

0.5
weight float

Weight of this field in the overall score (default: 1.0)

1.0
Source code in stickler/structured_object_evaluator/models/comparison_info.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    comparator: Optional[BaseComparator] = None,
    threshold: float = 0.5,
    weight: float = 1.0,
):
    """Initialize comparison configuration.

    Args:
        comparator: Comparator to use (default: LevenshteinComparator)
        threshold: Minimum similarity score to consider a match (default: 0.5)
        weight: Weight of this field in the overall score (default: 1.0)
    """
    self.comparator = comparator or LevenshteinComparator()
    self.threshold = threshold
    self.weight = weight

__repr__()

Return string representation.

Source code in stickler/structured_object_evaluator/models/comparison_info.py
61
62
63
def __repr__(self) -> str:
    """Return string representation."""
    return f"ComparisonInfo(comparator={self.comparator}, threshold={self.threshold}, weight={self.weight})"

compare(value1, value2)

Compare two values and return a similarity score between 0 and 1.

Parameters:

Name Type Description Default
value1 Any

First value to compare

required
value2 Any

Second value to compare

required

Returns:

Type Description
float

Similarity score between 0.0 and 1.0, with 0.0 if below threshold

Source code in stickler/structured_object_evaluator/models/comparison_info.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def compare(self, value1: Any, value2: Any) -> float:
    """Compare two values and return a similarity score between 0 and 1.

    Args:
        value1: First value to compare
        value2: Second value to compare

    Returns:
        Similarity score between 0.0 and 1.0, with 0.0 if below threshold
    """
    # Handle None values
    if value1 is None or value2 is None:
        return 1.0 if value1 == value2 else 0.0

    # Use the comparator to calculate similarity
    similarity = self.comparator.compare(value1, value2)

    # Apply threshold (if below threshold, return 0)
    return 0.0 if similarity < self.threshold else similarity

to_dict()

Convert to a serializable dictionary for JSON schema.

Source code in stickler/structured_object_evaluator/models/comparison_info.py
65
66
67
68
69
70
71
72
73
def to_dict(self) -> Dict[str, Any]:
    """Convert to a serializable dictionary for JSON schema."""
    return {
        "comparator_type": self.comparator.__class__.__name__,
        "comparator_name": getattr(self.comparator, "name", "unknown"),
        "comparator_config": getattr(self.comparator, "config", {}),
        "threshold": self.threshold,
        "weight": self.weight,
    }