From bc55f4f639ad352fd9314b9f6f477971b6ee4e83 Mon Sep 17 00:00:00 2001 From: mimran-khan Date: Thu, 25 Jun 2026 01:28:02 +0530 Subject: [PATCH] feat: add schema complexity analyzer and automatic mode selection Introduce two complementary utilities that help users get better extraction results on the first attempt: 1. `analyze_schema(response_model)` inspects a Pydantic model's JSON schema and reports complexity metrics (depth, field count, recursion, large enums) along with actionable findings and a 0-100 complexity score. 2. `select_mode(response_model, provider)` picks the optimal extraction mode for a given provider by cross-referencing schema complexity with provider capabilities. Handles recursive schemas, deep nesting, and high-complexity patterns by routing to the mode most likely to succeed. Both are exposed at `instructor.analyze_schema` and `instructor.select_mode` for convenient top-level access. --- instructor/__init__.py | 11 + instructor/v2/core/__init__.py | 8 + instructor/v2/core/auto_mode.py | 171 +++++++++++++ instructor/v2/core/schema_analyzer.py | 346 +++++++++++++++++++++++++ tests/test_auto_mode.py | 317 +++++++++++++++++++++++ tests/test_schema_analyzer.py | 348 ++++++++++++++++++++++++++ 6 files changed, 1201 insertions(+) create mode 100644 instructor/v2/core/auto_mode.py create mode 100644 instructor/v2/core/schema_analyzer.py create mode 100644 tests/test_auto_mode.py create mode 100644 tests/test_schema_analyzer.py diff --git a/instructor/__init__.py b/instructor/__init__.py index 908afa267..f89073968 100644 --- a/instructor/__init__.py +++ b/instructor/__init__.py @@ -61,6 +61,11 @@ from .v2.providers.vertexai.client import from_vertexai as from_vertexai from .v2.providers.writer.client import from_writer as from_writer from .v2.providers.xai.client import from_xai as from_xai + from .v2.core.auto_mode import select_mode as select_mode + from .v2.core.schema_analyzer import ( + SchemaAnalysis as SchemaAnalysis, + analyze_schema as analyze_schema, + ) from .v2.validation import ( llm_validator as llm_validator, openai_moderation as openai_moderation, @@ -106,6 +111,9 @@ "openai_moderation", "hooks", "v2", + "analyze_schema", + "SchemaAnalysis", + "select_mode", ] _LAZY_IMPORTS: dict[str, tuple[str, str | None]] = { @@ -158,6 +166,9 @@ "from_xai": (".v2.providers.xai.client", "from_xai"), "from_perplexity": (".v2.providers.perplexity.client", "from_perplexity"), "from_genai": (".v2.providers.genai.client", "from_genai"), + "analyze_schema": (".v2.core.schema_analyzer", "analyze_schema"), + "SchemaAnalysis": (".v2.core.schema_analyzer", "SchemaAnalysis"), + "select_mode": (".v2.core.auto_mode", "select_mode"), } diff --git a/instructor/v2/core/__init__.py b/instructor/v2/core/__init__.py index 59125e06d..9c2d8155e 100644 --- a/instructor/v2/core/__init__.py +++ b/instructor/v2/core/__init__.py @@ -15,6 +15,10 @@ "ReaskHandler", "ResponseParser", "normalize_mode", + "analyze_schema", + "SchemaAnalysis", + "SchemaFinding", + "select_mode", ] _LAZY_ATTRS: dict[str, tuple[str, str]] = { @@ -27,6 +31,10 @@ "ReaskHandler": ("instructor.v2.core.protocols", "ReaskHandler"), "ResponseParser": ("instructor.v2.core.protocols", "ResponseParser"), "normalize_mode": ("instructor.v2.core.registry", "normalize_mode"), + "analyze_schema": ("instructor.v2.core.schema_analyzer", "analyze_schema"), + "SchemaAnalysis": ("instructor.v2.core.schema_analyzer", "SchemaAnalysis"), + "SchemaFinding": ("instructor.v2.core.schema_analyzer", "SchemaFinding"), + "select_mode": ("instructor.v2.core.auto_mode", "select_mode"), } diff --git a/instructor/v2/core/auto_mode.py b/instructor/v2/core/auto_mode.py new file mode 100644 index 000000000..fa8a59c7f --- /dev/null +++ b/instructor/v2/core/auto_mode.py @@ -0,0 +1,171 @@ +"""Advisory automatic mode selection based on provider capabilities. + +Selects modes using schema complexity and provider support. + +This module is strictly advisory. It provides a recommendation that users can +pass to from_openai(..., mode=recommended) or ignore entirely. It does NOT +automatically override user-specified modes, and it is NOT called internally +by from_provider or any other client factory. + +The intent is to give users a starting point when they are unsure which mode +to use for a given schema, especially when dealing with complex or recursive +models that behave differently across providers. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +from pydantic import BaseModel + +from instructor.v2.core.mode import Mode +from instructor.v2.core.providers import Provider +from instructor.v2.core.provider_specs import PROVIDER_SPECS +from instructor.v2.core.schema_analyzer import ( + SchemaAnalysis, + analyze_schema, +) + +if TYPE_CHECKING: + from instructor.v2.core.provider_specs import ProviderSpec + +logger = logging.getLogger("instructor.v2.auto_mode") + + +def select_mode( + response_model: type[BaseModel], + provider: Provider, + *, + prefer_strict: bool = True, + analysis: SchemaAnalysis | None = None, +) -> Mode: + """Return an advisory mode recommendation for a provider and response model. + + This function is purely advisory. It does NOT automatically override any + user-specified mode, and it is NOT called internally by from_provider or + any client factory. Users opt in explicitly: + + mode = select_mode(MyModel, Provider.OPENAI) + client = instructor.from_openai(openai.OpenAI(), mode=mode) + + The recommendation is based on: + - Provider support (not all providers support all modes) + - Schema complexity (simple schemas work with TOOLS, complex with JSON_SCHEMA) + - Recursion (recursive models need MD_JSON or JSON modes) + - Depth (deeply nested schemas benefit from JSON_SCHEMA) + + Args: + response_model: The Pydantic model to extract. + provider: The target LLM provider. + prefer_strict: If True, prefer stricter modes (JSON_SCHEMA over MD_JSON) + when both are available and suitable. Default True. + analysis: Pre-computed schema analysis. If None, computed internally. + + Returns: + The recommended Mode enum value. Users should pass this to their + client factory or ignore it if they have a reason to prefer a + specific mode. + + Note: + This is advisory only and not integrated into from_provider. + The heuristics are intentionally conservative - when in doubt, + TOOLS mode is preferred since providers optimize for it. + + Examples: + >>> from pydantic import BaseModel + >>> class Simple(BaseModel): + ... name: str + >>> select_mode(Simple, Provider.OPENAI) + + + >>> class Complex(BaseModel): + ... items: list["Complex"] + >>> select_mode(Complex, Provider.OPENAI) + + """ + if analysis is None: + analysis = analyze_schema(response_model) + + spec = PROVIDER_SPECS.get(provider) + supported = _get_supported_modes(spec) + + if not supported: + logger.debug(f"No supported modes found for {provider}, defaulting to TOOLS") + return Mode.TOOLS + + if analysis.has_recursion: + return _pick_recursive_mode(supported, prefer_strict) + + if analysis.complexity_score > 70: + return _pick_high_complexity_mode(supported, prefer_strict) + + if analysis.max_depth > 4: + return _pick_deep_schema_mode(supported, prefer_strict) + + return _pick_simple_mode(supported, prefer_strict) + + +def _get_supported_modes(spec: ProviderSpec | None) -> set[Mode]: + """Extract supported modes from a provider spec.""" + if spec is None: + return {Mode.TOOLS, Mode.JSON_SCHEMA, Mode.MD_JSON} + return set(spec.supported_modes) + + +def _pick_recursive_mode(supported: set[Mode], prefer_strict: bool) -> Mode: # noqa: ARG001 + """Pick mode for recursive schemas. + + Recursive models cannot be represented in tool-call schemas on most providers. + MD_JSON or JSON modes handle them gracefully since the LLM generates free-form + JSON that is then validated. + """ + preference_order = [Mode.MD_JSON, Mode.JSON, Mode.JSON_SCHEMA, Mode.TOOLS] + return _first_supported(preference_order, supported) + + +def _pick_high_complexity_mode(supported: set[Mode], prefer_strict: bool) -> Mode: + """Pick mode for highly complex schemas (score > 70). + + JSON_SCHEMA mode gives the LLM explicit structural guidance, reducing errors + on complex schemas. Falls back to MD_JSON, then TOOLS. + """ + if prefer_strict: + preference_order = [Mode.JSON_SCHEMA, Mode.TOOLS, Mode.MD_JSON, Mode.JSON] + else: + preference_order = [Mode.MD_JSON, Mode.JSON_SCHEMA, Mode.TOOLS, Mode.JSON] + return _first_supported(preference_order, supported) + + +def _pick_deep_schema_mode(supported: set[Mode], prefer_strict: bool) -> Mode: + """Pick mode for deeply nested schemas (depth > 4). + + TOOLS mode can struggle with deep nesting since each level adds function + call overhead. JSON_SCHEMA provides the full structure inline. + """ + if prefer_strict: + preference_order = [Mode.JSON_SCHEMA, Mode.TOOLS, Mode.MD_JSON, Mode.JSON] + else: + preference_order = [Mode.TOOLS, Mode.JSON_SCHEMA, Mode.MD_JSON, Mode.JSON] + return _first_supported(preference_order, supported) + + +def _pick_simple_mode(supported: set[Mode], prefer_strict: bool) -> Mode: + """Pick mode for simple schemas (low complexity, shallow depth). + + TOOLS mode is most reliable for simple schemas since it leverages native + function calling which providers optimize for. + """ + if prefer_strict: + preference_order = [Mode.TOOLS, Mode.JSON_SCHEMA, Mode.MD_JSON, Mode.JSON] + else: + preference_order = [Mode.TOOLS, Mode.MD_JSON, Mode.JSON_SCHEMA, Mode.JSON] + return _first_supported(preference_order, supported) + + +def _first_supported(preference: list[Mode], supported: set[Mode]) -> Mode: + """Return the first mode from preference that is in the supported set.""" + for mode in preference: + if mode in supported: + return mode + return Mode.TOOLS diff --git a/instructor/v2/core/schema_analyzer.py b/instructor/v2/core/schema_analyzer.py new file mode 100644 index 000000000..2dec93c60 --- /dev/null +++ b/instructor/v2/core/schema_analyzer.py @@ -0,0 +1,346 @@ +"""Schema complexity pre-flight analyzer. + +Analyzes Pydantic model JSON schemas before API calls to detect complexity +patterns that commonly cause LLM extraction failures. Provides actionable +warnings and mode recommendations based on schema characteristics. +""" + +from __future__ import annotations + +import warnings +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + +from pydantic import BaseModel + + +class Severity(Enum): + """Severity level for schema analysis findings.""" + + INFO = "info" + WARNING = "warning" + ERROR = "error" + + +@dataclass +class SchemaFinding: + """A single finding from schema analysis.""" + + severity: Severity + code: str + message: str + path: str = "" + suggestion: str = "" + + +@dataclass +class SchemaAnalysis: + """Complete analysis result for a schema.""" + + findings: list[SchemaFinding] = field(default_factory=list) + total_fields: int = 0 + max_depth: int = 0 + has_recursion: bool = False + num_required: int = 0 + num_optional: int = 0 + estimated_token_overhead: int = 0 + + @property + def has_errors(self) -> bool: + return any(f.severity == Severity.ERROR for f in self.findings) + + @property + def has_warnings(self) -> bool: + return any(f.severity == Severity.WARNING for f in self.findings) + + @property + def complexity_score(self) -> int: + """Score from 0-100 representing schema complexity. + + Higher scores indicate more complex schemas that are harder for LLMs. + """ + score = 0 + score += min(self.total_fields * 2, 30) + score += min(self.max_depth * 8, 32) + score += 20 if self.has_recursion else 0 + score += min(self.estimated_token_overhead // 100, 18) + return min(score, 100) + + @property + def recommended_mode(self) -> str | None: + """Suggest the best mode based on complexity characteristics.""" + if self.has_recursion: + return "MD_JSON" + if self.complexity_score > 70: + return "JSON_SCHEMA" + if self.max_depth > 4: + return "TOOLS" + return None + + +# Thresholds for warnings and errors +MAX_DEPTH_WARNING = 4 +MAX_DEPTH_ERROR = 7 +MAX_FIELDS_WARNING = 20 +MAX_FIELDS_ERROR = 50 +MAX_ENUM_VALUES_WARNING = 15 +MAX_ENUM_VALUES_ERROR = 50 +MAX_REQUIRED_FIELDS_WARNING = 12 +MAX_PROPERTIES_PER_OBJECT_WARNING = 10 +ESTIMATED_TOKENS_PER_FIELD = 15 + + +def analyze_schema( + response_model: type[BaseModel], + *, + warn: bool = False, +) -> SchemaAnalysis: + """Analyze a Pydantic model's schema for complexity issues. + + Inspects the JSON schema generated from a Pydantic model and identifies + patterns that commonly cause LLM extraction failures, including: + + - Deeply nested object hierarchies + - Recursive/self-referencing models + - Large enum value sets + - Too many required fields + - Excessively wide objects (many properties) + + Args: + response_model: The Pydantic model class to analyze. + warn: If True, emit Python warnings for findings at WARNING+ severity. + + Returns: + SchemaAnalysis with findings and metrics. + + Examples: + >>> from pydantic import BaseModel + >>> class Simple(BaseModel): + ... name: str + ... age: int + >>> result = analyze_schema(Simple) + >>> result.complexity_score < 20 + True + + >>> class Nested(BaseModel): + ... child: "Nested | None" = None + >>> result = analyze_schema(Nested) + >>> result.has_recursion + True + """ + schema = response_model.model_json_schema() + analysis = SchemaAnalysis() + + defs = schema.get("$defs", {}) + + _walk_schema( + schema, + analysis=analysis, + defs=defs, + path="$", + depth=0, + visited_refs=set(), + ) + + analysis.estimated_token_overhead = ( + analysis.total_fields * ESTIMATED_TOKENS_PER_FIELD + ) + + _check_global_thresholds(analysis) + + if warn: + for finding in analysis.findings: + if finding.severity in (Severity.WARNING, Severity.ERROR): + msg = f"[instructor] {finding.code}: {finding.message}" + if finding.suggestion: + msg += f" Suggestion: {finding.suggestion}" + warnings.warn(msg, stacklevel=2) + + return analysis + + +def _walk_schema( + node: dict[str, Any], + *, + analysis: SchemaAnalysis, + defs: dict[str, Any], + path: str, + depth: int, + visited_refs: set[str], +) -> None: + """Recursively walk a JSON schema node and accumulate metrics.""" + analysis.max_depth = max(analysis.max_depth, depth) + + if "$ref" in node: + ref_name = node["$ref"].split("/")[-1] + if ref_name in visited_refs: + analysis.has_recursion = True + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="RECURSIVE_REF", + message=f"Recursive reference detected: {ref_name}", + path=path, + suggestion=( + "Recursive schemas can confuse some LLMs. " + "Consider using MD_JSON mode or flattening the structure." + ), + ) + ) + return + if ref_name in defs: + _walk_schema( + defs[ref_name], + analysis=analysis, + defs=defs, + path=f"{path}.${ref_name}", + depth=depth, + visited_refs=visited_refs | {ref_name}, + ) + return + + if "anyOf" in node or "oneOf" in node: + variants = node.get("anyOf") or node.get("oneOf", []) + for i, variant in enumerate(variants): + _walk_schema( + variant, + analysis=analysis, + defs=defs, + path=f"{path}[variant_{i}]", + depth=depth, + visited_refs=visited_refs, + ) + return + + if "allOf" in node: + for i, sub in enumerate(node["allOf"]): + _walk_schema( + sub, + analysis=analysis, + defs=defs, + path=f"{path}[allOf_{i}]", + depth=depth, + visited_refs=visited_refs, + ) + return + + if "enum" in node: + enum_values = node["enum"] + count = len(enum_values) + if count > MAX_ENUM_VALUES_ERROR: + analysis.findings.append( + SchemaFinding( + severity=Severity.ERROR, + code="ENUM_TOO_LARGE", + message=f"Enum at {path} has {count} values (max recommended: {MAX_ENUM_VALUES_ERROR})", + path=path, + suggestion="Split into categories or use a free-text field with validation.", + ) + ) + elif count > MAX_ENUM_VALUES_WARNING: + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="ENUM_LARGE", + message=f"Enum at {path} has {count} values (>{MAX_ENUM_VALUES_WARNING})", + path=path, + suggestion="Large enums increase token usage and error rates.", + ) + ) + return + + node_type = node.get("type") + if node_type == "object" or "properties" in node: + properties = node.get("properties", {}) + required = node.get("required", []) + prop_count = len(properties) + + analysis.total_fields += prop_count + analysis.num_required += len(required) + analysis.num_optional += prop_count - len(required) + + if prop_count > MAX_PROPERTIES_PER_OBJECT_WARNING: + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="WIDE_OBJECT", + message=f"Object at {path} has {prop_count} properties", + path=path, + suggestion="Consider grouping related fields into sub-objects.", + ) + ) + + for prop_name, prop_schema in properties.items(): + _walk_schema( + prop_schema, + analysis=analysis, + defs=defs, + path=f"{path}.{prop_name}", + depth=depth + 1, + visited_refs=visited_refs, + ) + return + + if node_type == "array": + items = node.get("items", {}) + _walk_schema( + items, + analysis=analysis, + defs=defs, + path=f"{path}[]", + depth=depth + 1, + visited_refs=visited_refs, + ) + return + + +def _check_global_thresholds(analysis: SchemaAnalysis) -> None: + """Check global metrics against thresholds.""" + if analysis.max_depth > MAX_DEPTH_ERROR: + analysis.findings.append( + SchemaFinding( + severity=Severity.ERROR, + code="DEPTH_EXCESSIVE", + message=f"Schema depth is {analysis.max_depth} (max recommended: {MAX_DEPTH_ERROR})", + suggestion="Flatten nested structures or extract sub-models into separate calls.", + ) + ) + elif analysis.max_depth > MAX_DEPTH_WARNING: + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="DEPTH_HIGH", + message=f"Schema depth is {analysis.max_depth} (>{MAX_DEPTH_WARNING})", + suggestion="Deep nesting increases extraction errors. Consider flattening.", + ) + ) + + if analysis.total_fields > MAX_FIELDS_ERROR: + analysis.findings.append( + SchemaFinding( + severity=Severity.ERROR, + code="TOO_MANY_FIELDS", + message=f"Schema has {analysis.total_fields} total fields (max recommended: {MAX_FIELDS_ERROR})", + suggestion="Split into multiple extraction calls or reduce schema scope.", + ) + ) + elif analysis.total_fields > MAX_FIELDS_WARNING: + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="MANY_FIELDS", + message=f"Schema has {analysis.total_fields} total fields (>{MAX_FIELDS_WARNING})", + suggestion="Consider whether all fields are needed in a single extraction.", + ) + ) + + if analysis.num_required > MAX_REQUIRED_FIELDS_WARNING: + analysis.findings.append( + SchemaFinding( + severity=Severity.WARNING, + code="MANY_REQUIRED", + message=f"Schema has {analysis.num_required} required fields (>{MAX_REQUIRED_FIELDS_WARNING})", + suggestion="Make some fields optional with defaults to reduce extraction pressure.", + ) + ) diff --git a/tests/test_auto_mode.py b/tests/test_auto_mode.py new file mode 100644 index 000000000..23296620e --- /dev/null +++ b/tests/test_auto_mode.py @@ -0,0 +1,317 @@ +"""Tests for the advisory automatic mode selection system. + +Covers simple models, recursive schemas, deep nesting, high complexity, +provider-specific constraints, and edge cases around fallback behavior. +""" + +from __future__ import annotations + +from pydantic import BaseModel + +from instructor.v2.core.auto_mode import select_mode +from instructor.v2.core.mode import Mode +from instructor.v2.core.providers import Provider +from instructor.v2.core.schema_analyzer import analyze_schema + + +# --- Test Models (ordered so dependencies come before usage) --- + + +class SimpleUser(BaseModel): + name: str + age: int + + +class RecursiveTree(BaseModel): + label: str + children: list[RecursiveTree] = [] + + +class OptionalRecursive(BaseModel): + name: str + parent: OptionalRecursive | None = None + + +# Deep nesting chain (leaf to root) +class LevelE(BaseModel): + value: str + + +class LevelD(BaseModel): + e: LevelE + + +class LevelC(BaseModel): + d: LevelD + + +class LevelB(BaseModel): + c: LevelC + + +class LevelA(BaseModel): + b: LevelB + + +class DeeplyNestedModel(BaseModel): + a: LevelA + + +# High complexity chain (leaf to root) +class ComplexBottomA(BaseModel): + z1: str + z2: str + z3: str + z4: str + z5: str + z6: str + z7: str + + +class ComplexDeeperA(BaseModel): + bottom: ComplexBottomA + y1: str + y2: str + y3: str + + +class ComplexDeepA(BaseModel): + deeper: ComplexDeeperA + x1: str + x2: str + x3: str + x4: str + x5: str + + +class ComplexInnerA(BaseModel): + deep: ComplexDeepA + w1: str + w2: str + w3: str + w4: str + w5: str + + +class ComplexSectionA(BaseModel): + inner: ComplexInnerA + v1: str + v2: str + v3: str + v4: str + v5: str + + +class ComplexInnerB(BaseModel): + n1: str + n2: str + n3: str + n4: str + n5: str + n6: str + n7: str + n8: str + n9: str + n10: str + n11: str + n12: str + + +class ComplexSectionB(BaseModel): + inner: ComplexInnerB + m1: str + m2: str + m3: str + m4: str + m5: str + m6: str + m7: str + + +class HighComplexityModel(BaseModel): + """A model designed to exceed the 70-point complexity threshold. + + Combines deep nesting (5+ levels) with many fields to push the score over 70. + """ + + section_a: ComplexSectionA + section_b: ComplexSectionB + f01: str + f02: str + f03: str + f04: str + f05: str + f06: str + f07: str + f08: str + f09: str + f10: str + + +# Medium complexity chain (leaf to root) +class RefItem(BaseModel): + url: str + label: str + + +class MetaBlock(BaseModel): + author: str + version: int + refs: list[RefItem] + + +class MediumComplexity(BaseModel): + title: str + description: str + tags: list[str] + metadata: MetaBlock + + +# --- Tests --- + + +class TestSimpleModelSelection: + def test_openai_simple_picks_tools(self): + mode = select_mode(SimpleUser, Provider.OPENAI) + assert mode == Mode.TOOLS + + def test_anthropic_simple_picks_tools(self): + mode = select_mode(SimpleUser, Provider.ANTHROPIC) + assert mode == Mode.TOOLS + + def test_genai_simple_picks_tools(self): + mode = select_mode(SimpleUser, Provider.GENAI) + assert mode == Mode.TOOLS + + def test_cohere_simple_picks_tools(self): + mode = select_mode(SimpleUser, Provider.COHERE) + assert mode == Mode.TOOLS + + def test_mistral_simple_picks_tools(self): + mode = select_mode(SimpleUser, Provider.MISTRAL) + assert mode == Mode.TOOLS + + +class TestRecursiveModelSelection: + def test_recursive_prefers_md_json_openai(self): + mode = select_mode(RecursiveTree, Provider.OPENAI) + assert mode == Mode.MD_JSON + + def test_recursive_prefers_md_json_anthropic(self): + mode = select_mode(RecursiveTree, Provider.ANTHROPIC) + # Anthropic does not support MD_JSON, should fallback to JSON + assert mode in (Mode.MD_JSON, Mode.JSON) + + def test_recursive_prefers_md_json_genai(self): + mode = select_mode(RecursiveTree, Provider.GENAI) + # GenAI supports JSON but not MD_JSON + assert mode == Mode.JSON + + def test_optional_recursive_also_detected(self): + mode = select_mode(OptionalRecursive, Provider.OPENAI) + assert mode == Mode.MD_JSON + + +class TestDeepNestingSelection: + def test_deep_openai_prefers_json_schema_strict(self): + mode = select_mode(DeeplyNestedModel, Provider.OPENAI, prefer_strict=True) + assert mode == Mode.JSON_SCHEMA + + def test_deep_openai_prefers_tools_relaxed(self): + mode = select_mode(DeeplyNestedModel, Provider.OPENAI, prefer_strict=False) + assert mode == Mode.TOOLS + + def test_deep_bedrock_falls_to_tools(self): + # Bedrock supports TOOLS and MD_JSON only + mode = select_mode(DeeplyNestedModel, Provider.BEDROCK) + assert mode in (Mode.TOOLS, Mode.MD_JSON) + + +class TestHighComplexitySelection: + def test_high_complexity_strict_prefers_json_schema(self): + analysis = analyze_schema(HighComplexityModel) + assert analysis.complexity_score > 70, ( + f"Test model score {analysis.complexity_score} too low, expected >70" + ) + mode = select_mode(HighComplexityModel, Provider.OPENAI, prefer_strict=True) + assert mode == Mode.JSON_SCHEMA + + def test_high_complexity_relaxed_prefers_md_json(self): + mode = select_mode(HighComplexityModel, Provider.OPENAI, prefer_strict=False) + assert mode == Mode.MD_JSON + + def test_high_complexity_mistral(self): + mode = select_mode(HighComplexityModel, Provider.MISTRAL, prefer_strict=True) + assert mode == Mode.JSON_SCHEMA + + +class TestProviderConstraints: + def test_perplexity_only_supports_md_json(self): + mode = select_mode(SimpleUser, Provider.PERPLEXITY) + assert mode == Mode.MD_JSON + + def test_bedrock_no_json_schema(self): + mode = select_mode(HighComplexityModel, Provider.BEDROCK, prefer_strict=True) + assert mode in (Mode.TOOLS, Mode.MD_JSON) + + def test_genai_supports_tools_and_json(self): + mode = select_mode(SimpleUser, Provider.GENAI) + assert mode in (Mode.TOOLS, Mode.JSON) + + +class TestPrecomputedAnalysis: + def test_accepts_precomputed_analysis(self): + analysis = analyze_schema(SimpleUser) + mode = select_mode(SimpleUser, Provider.OPENAI, analysis=analysis) + assert mode == Mode.TOOLS + + def test_precomputed_recursive_analysis(self): + analysis = analyze_schema(RecursiveTree) + mode = select_mode(RecursiveTree, Provider.OPENAI, analysis=analysis) + assert mode == Mode.MD_JSON + + +class TestFallbackBehavior: + def test_unknown_provider_defaults_to_tools(self): + mode = select_mode(SimpleUser, Provider.UNKNOWN) + assert mode == Mode.TOOLS + + def test_ollama_defaults_to_tools(self): + mode = select_mode(SimpleUser, Provider.OLLAMA) + assert mode == Mode.TOOLS + + +class TestPreferStrictFlag: + def test_prefer_strict_true_vs_false_simple(self): + strict = select_mode(SimpleUser, Provider.OPENAI, prefer_strict=True) + relaxed = select_mode(SimpleUser, Provider.OPENAI, prefer_strict=False) + assert strict == relaxed == Mode.TOOLS + + def test_prefer_strict_affects_deep_model(self): + strict = select_mode(DeeplyNestedModel, Provider.OPENAI, prefer_strict=True) + relaxed = select_mode(DeeplyNestedModel, Provider.OPENAI, prefer_strict=False) + assert strict == Mode.JSON_SCHEMA + assert relaxed == Mode.TOOLS + + +class TestMediumComplexity: + def test_medium_model_picks_tools(self): + result = analyze_schema(MediumComplexity) + assert result.complexity_score < 70 + mode = select_mode(MediumComplexity, Provider.OPENAI) + assert mode == Mode.TOOLS + + +class TestIntegrationWithAnalyzer: + def test_analysis_and_mode_selection_consistent(self): + analysis = analyze_schema(RecursiveTree) + assert analysis.has_recursion + mode = select_mode(RecursiveTree, Provider.OPENAI, analysis=analysis) + assert mode == Mode.MD_JSON + + def test_simple_analysis_and_mode_consistent(self): + analysis = analyze_schema(SimpleUser) + assert not analysis.has_recursion + assert analysis.complexity_score < 30 + mode = select_mode(SimpleUser, Provider.OPENAI, analysis=analysis) + assert mode == Mode.TOOLS diff --git a/tests/test_schema_analyzer.py b/tests/test_schema_analyzer.py new file mode 100644 index 000000000..d125fa66b --- /dev/null +++ b/tests/test_schema_analyzer.py @@ -0,0 +1,348 @@ +"""Tests for the schema complexity pre-flight analyzer. + +Covers simple schemas, deeply nested models, recursive structures, large enums, +wide objects, and mixed complexity patterns to ensure accurate detection and +scoring. +""" + +from __future__ import annotations + +import warnings +from enum import Enum + +from pydantic import BaseModel + +from instructor.v2.core.schema_analyzer import ( + SchemaFinding, + Severity, + analyze_schema, +) + + +# --- Test Models (ordered so dependencies come before usage) --- + + +class SimpleModel(BaseModel): + name: str + age: int + + +class Address(BaseModel): + street: str + city: str + zip_code: str + + +class NestedModel(BaseModel): + address: Address + name: str + + +class RecursiveModel(BaseModel): + value: str + children: list[RecursiveModel] = [] + + +class OptionalRecursiveModel(BaseModel): + name: str + parent: OptionalRecursiveModel | None = None + + +class Level5(BaseModel): + value: str + + +class Level4(BaseModel): + level5: Level5 + + +class Level3(BaseModel): + level4: Level4 + + +class Level2(BaseModel): + level3: Level3 + + +class Level1(BaseModel): + level2: Level2 + + +class DeeplyNested(BaseModel): + level1: Level1 + + +class WideModel(BaseModel): + field_01: str + field_02: str + field_03: str + field_04: str + field_05: str + field_06: str + field_07: str + field_08: str + field_09: str + field_10: str + field_11: str + field_12: str + + +BigValues = [f"cat_{i}" for i in range(25)] +BigCategory = Enum("BigCategory", {v: v for v in BigValues}) + + +class ModelWithLargeEnum(BaseModel): + category: BigCategory + + +HugeValues = [f"item_{i}" for i in range(60)] +HugeEnum = Enum("HugeEnum", {v: v for v in HugeValues}) + + +class ModelWithHugeEnum(BaseModel): + item: HugeEnum + + +class ModelWithManyRequired(BaseModel): + f01: str + f02: str + f03: str + f04: str + f05: str + f06: str + f07: str + f08: str + f09: str + f10: str + f11: str + f12: str + f13: str + + +class EmptyModel(BaseModel): + pass + + +class DataA(BaseModel): + kind: str = "a" + value_a: int + + +class DataB(BaseModel): + kind: str = "b" + value_b: str + + +class ModelWithUnion(BaseModel): + data: DataA | DataB + + +class ModelWithOptionals(BaseModel): + required_field: str + optional_1: str | None = None + optional_2: int | None = None + optional_3: float | None = None + + +class ListItem(BaseModel): + name: str + quantity: int + + +class ModelWithList(BaseModel): + items: list[ListItem] + + +# --- Tests --- + + +class TestSimpleSchemas: + def test_simple_model_low_complexity(self): + result = analyze_schema(SimpleModel) + assert result.complexity_score < 20 + assert result.total_fields == 2 + assert result.max_depth == 1 + assert not result.has_recursion + assert not result.has_errors + assert not result.has_warnings + + def test_empty_model(self): + result = analyze_schema(EmptyModel) + assert result.complexity_score == 0 + assert result.total_fields == 0 + assert result.max_depth == 0 + assert not result.has_recursion + + def test_model_with_optionals(self): + result = analyze_schema(ModelWithOptionals) + assert result.num_required == 1 + assert result.num_optional == 3 + assert result.total_fields == 4 + + def test_nested_model_adds_depth(self): + result = analyze_schema(NestedModel) + assert result.max_depth >= 2 + assert result.total_fields >= 5 + + +class TestRecursiveSchemas: + def test_recursive_model_detected(self): + result = analyze_schema(RecursiveModel) + assert result.has_recursion + assert any(f.code == "RECURSIVE_REF" for f in result.findings) + + def test_optional_recursive_model_detected(self): + result = analyze_schema(OptionalRecursiveModel) + assert result.has_recursion + + def test_recursive_model_recommends_md_json(self): + result = analyze_schema(RecursiveModel) + assert result.recommended_mode == "MD_JSON" + + +class TestDeepNesting: + def test_deeply_nested_reports_high_depth(self): + result = analyze_schema(DeeplyNested) + assert result.max_depth >= 5 + + def test_depth_warning_threshold(self): + result = analyze_schema(DeeplyNested) + depth_findings = [ + f for f in result.findings if f.code in ("DEPTH_HIGH", "DEPTH_EXCESSIVE") + ] + assert len(depth_findings) > 0 + + +class TestWideObjects: + def test_wide_model_triggers_warning(self): + result = analyze_schema(WideModel) + wide_findings = [f for f in result.findings if f.code == "WIDE_OBJECT"] + assert len(wide_findings) > 0 + + def test_wide_model_counts_fields_correctly(self): + result = analyze_schema(WideModel) + assert result.total_fields == 12 + + +class TestEnums: + def test_large_enum_triggers_warning(self): + result = analyze_schema(ModelWithLargeEnum) + enum_findings = [f for f in result.findings if f.code == "ENUM_LARGE"] + assert len(enum_findings) > 0 + + def test_huge_enum_triggers_error(self): + result = analyze_schema(ModelWithHugeEnum) + enum_findings = [f for f in result.findings if f.code == "ENUM_TOO_LARGE"] + assert len(enum_findings) > 0 + + +class TestRequiredFields: + def test_many_required_triggers_warning(self): + result = analyze_schema(ModelWithManyRequired) + required_findings = [f for f in result.findings if f.code == "MANY_REQUIRED"] + assert len(required_findings) > 0 + + def test_required_count_accurate(self): + result = analyze_schema(ModelWithManyRequired) + assert result.num_required == 13 + + +class TestUnionTypes: + def test_union_model_handles_variants(self): + result = analyze_schema(ModelWithUnion) + assert result.total_fields >= 2 + assert not result.has_recursion + + def test_union_model_low_complexity(self): + result = analyze_schema(ModelWithUnion) + assert result.complexity_score < 50 + + +class TestListFields: + def test_list_model_adds_depth(self): + result = analyze_schema(ModelWithList) + assert result.max_depth >= 2 + + def test_list_item_fields_counted(self): + result = analyze_schema(ModelWithList) + assert result.total_fields >= 3 + + +class TestComplexityScore: + def test_score_bounded_0_to_100(self): + for model in [SimpleModel, RecursiveModel, DeeplyNested, WideModel]: + result = analyze_schema(model) + assert 0 <= result.complexity_score <= 100 + + def test_simple_lower_than_recursive(self): + simple = analyze_schema(SimpleModel) + recursive = analyze_schema(RecursiveModel) + assert simple.complexity_score < recursive.complexity_score + + def test_simple_lower_than_deeply_nested(self): + simple = analyze_schema(SimpleModel) + deep = analyze_schema(DeeplyNested) + assert simple.complexity_score < deep.complexity_score + + +class TestTokenOverhead: + def test_estimated_tokens_proportional_to_fields(self): + simple = analyze_schema(SimpleModel) + wide = analyze_schema(WideModel) + assert wide.estimated_token_overhead > simple.estimated_token_overhead + + def test_empty_model_zero_overhead(self): + result = analyze_schema(EmptyModel) + assert result.estimated_token_overhead == 0 + + +class TestWarningEmission: + def test_warn_flag_emits_python_warnings(self): + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + analyze_schema(WideModel, warn=True) + assert len(w) > 0 + assert any("[instructor]" in str(warning.message) for warning in w) + + def test_no_warnings_for_simple_model(self): + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + analyze_schema(SimpleModel, warn=True) + assert len(w) == 0 + + +class TestSchemaFinding: + def test_finding_has_all_fields(self): + finding = SchemaFinding( + severity=Severity.WARNING, + code="TEST", + message="test message", + path="$.field", + suggestion="do something", + ) + assert finding.severity == Severity.WARNING + assert finding.code == "TEST" + assert finding.path == "$.field" + + def test_severity_enum_values(self): + assert Severity.INFO.value == "info" + assert Severity.WARNING.value == "warning" + assert Severity.ERROR.value == "error" + + +class TestSchemaAnalysisProperties: + def test_has_errors_false_for_simple(self): + result = analyze_schema(SimpleModel) + assert not result.has_errors + + def test_has_errors_true_for_huge_enum(self): + result = analyze_schema(ModelWithHugeEnum) + assert result.has_errors + + def test_has_warnings_false_for_simple(self): + result = analyze_schema(SimpleModel) + assert not result.has_warnings + + def test_has_warnings_true_for_wide(self): + result = analyze_schema(WideModel) + assert result.has_warnings