from ai_json_cleanroom import validate_ai_json # Anything your AI returns (messy, wrapped, incomplete) ai_response = "Here's your data:\n```json\n{'name': 'Alice', age: 30} // Python-style syntax\n```\n" # One line to clean and validate result = validate_ai_json(ai_response) if result.json_valid: print(result.data) # Clean: {'name': 'Alice', 'age': 30} else: print(result.errors) # Detailed error information

from ai_json_cleanroom import validate_ai_json # From markdown code fence (AI models often wrap JSON in ```json blocks) markdown_output = 'Here is the data:\n```json\n{"status": "success"}\n```\n' result = validate_ai_json(markdown_output) # Extracted: {"status": "success"} # From mixed text mixed_output = 'The result is {"status": "success"} as requested.' result = validate_ai_json(mixed_output) # Extracted: {"status": "success"} # From generic code fence generic_fence = '```\n{"status": "success"}\n```' result = validate_ai_json(generic_fence) # Extracted: {"status": "success"}

from ai_json_cleanroom import validate_ai_json, ValidateOptions # Single quotes → double quotes result = validate_ai_json("{'name': 'Alice'}") # Repaired: {"name": "Alice"} # Python constants → JSON result = validate_ai_json('{"active": True, "value": None}') # Repaired: {"active": true, "value": null} # Unquoted keys → quoted keys result = validate_ai_json('{name: "Alice", age: 30}') # Repaired: {"name": "Alice", "age": 30} # Comments removal result = validate_ai_json(''' {  "name": "Alice", // user name  /* age field */ "age": 30 } ''') # Repaired: {"name": "Alice", "age": 30} # Trailing commas result = validate_ai_json('{"items": [1, 2, 3,]}') # Repaired: {"items": [1, 2, 3]} # Inner unescaped quotes result = validate_ai_json('{"text": "She said "hello" to me"}') # Repaired: {"text": "She said \"hello\" to me"}

from ai_json_cleanroom import validate_ai_json truncated = '{"users": [{"name": "Alice", "age": 30}, {"name": "Bob", "age":' result = validate_ai_json(truncated) print(result.likely_truncated) # True print(result.errors[0].message) # "No JSON payload found in input." print(result.errors[0].detail) # {'truncation_reasons': ['unclosed_braces_or_brackets', 'suspicious_trailing_character']}

from ai_json_cleanroom import validate_ai_json schema = { "type": "object", "required": ["name", "email"], "properties": { "name": { "type": "string", "minLength": 1, "maxLength": 100 }, "email": { "type": "string", "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$" }, "age": { "type": "integer", "minimum": 0, "maximum": 150 }, "tags": { "type": "array", "minItems": 1, "items": {"type": "string"} } }, "additionalProperties": False } result = validate_ai_json(ai_output, schema=schema) if not result.json_valid: for error in result.errors: print(f"{error.code}: {error.message} at {error.path}")

from ai_json_cleanroom import validate_ai_json expectations = [ { "path": "users[*].email", "required": True, "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$" }, { "path": "users[*].status", "required": True, "in": ["active", "pending", "inactive"] }, { "path": "metadata.version", "required": True, "type": "string", "pattern": r"^\d+\.\d+\.\d+$" }, { "path": "items[*].price", "minimum": 0, "type": "number" } ] result = validate_ai_json(ai_output, expectations=expectations)

from ai_json_cleanroom import validate_ai_json result = validate_ai_json(any_input) # Always safe to access print(f"Valid: {result.json_valid}") print(f"Truncated: {result.likely_truncated}") print(f"Errors: {len(result.errors)}") print(f"Warnings: {len(result.warnings)}") print(f"Data: {result.data}") # None if invalid print(f"Info: {result.info}") # Extraction/parsing metadata # Structured error handling for error in result.errors: print(f"Code: {error.code}") print(f"Path: {error.path}") print(f"Message: {error.message}") print(f"Detail: {error.detail}")

# GPT often returns this mix: input = "{'name': 'Alice', \"age\": 30}" # Mixed quotes # With fix_single_quotes=True: # ✅ Becomes: {"name": "Alice", "age": 30} # With fix_single_quotes=False: # ❌ Parse fails on single quotes

# Claude sometimes returns: input = "{name: 'Alice', age: 30, active: true}" # With quote_unquoted_keys=True: # ✅ Becomes: {"name": "Alice", "age": 30, "active": true}

# Models often mix languages: input = '{"active": True, "deleted": False, "parent": None}' # With replace_constants=True: # ✅ Becomes: {"active": true, "deleted": false, "parent": null}

input = ''' {  "name": "Alice", // user name  /* age field */ "age": 30 } ''' # ✅ Comments are safely removed

# From copy-paste or models trained on web text: input = '{"text": "She said "hello" to me"}' # Smart quotes # With normalize_curly_quotes="always": # ✅ Becomes: {"text": "She said \"hello\" to me"}

from ai_json_cleanroom import validate_ai_json, ValidateOptions # For very messy outputs: options = ValidateOptions( max_total_repairs=500, # Allow more fixes max_repairs_percent=0.05 # Allow 5% of content to be modified ) # For strict validation: options = ValidateOptions( max_total_repairs=10, # Minimal fixes only max_repairs_percent=0.001 # Less than 0.1% modifications )

from openai import OpenAI from ai_json_cleanroom import validate_ai_json client = OpenAI() response = client.chat.completions.create( model="gpt-5.1-2025-11-13", messages=[ {"role": "system", "content": "You are a helpful assistant that outputs JSON."}, {"role": "user", "content": "Generate user profile for Alice Johnson, age 30"} ], response_format={"type": "json_object"} ) # Clean and validate the response result = validate_ai_json( response.choices[0].message.content, schema={ "type": "object", "required": ["name", "age"], "properties": { "name": {"type": "string"}, "age": {"type": "integer", "minimum": 0} } } ) if result.json_valid: user_data = result.data print(f"User: {user_data['name']}, Age: {user_data['age']}") else: # Use structured feedback for retry error_msg = "\n".join([e.message for e in result.errors]) print(f"Validation failed:\n{error_msg}") # Optionally retry with feedback

import anthropic from ai_json_cleanroom import validate_ai_json client = anthropic.Anthropic() message = client.messages.create( model="claude-haiku-4-5", max_tokens=1024, messages=[ { "role": "user", "content": "Generate a JSON object with user info for Alice, age 30" } ] ) # Claude might return: # "Here's the user data:\n```json\n{\"name\": \"Alice\", \"age\": 30}\n```\nLet me know if you need anything else!" result = validate_ai_json(message.content[0].text) if result.json_valid: print(f"Extracted data: {result.data}") print(f"Extraction source: {result.info['source']}") # 'code_fence' else: if result.likely_truncated: print("Response was truncated, increasing max_tokens...") else: print(f"Validation errors: {result.errors}")

from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.output_parsers import JsonOutputParser from ai_json_cleanroom import validate_ai_json, ValidateOptions import json # Initialize LangChain LLM llm = ChatOpenAI(model="gpt-5.1", temperature=0) # Create prompt prompt = ChatPromptTemplate.from_template( "Generate a JSON object with information about {topic}. Return only valid JSON." ) # Get LLM response chain = prompt | llm response = chain.invoke({"topic": "Python programming"}) # Step 1: Clean with ai-json-cleanroom cleaned = validate_ai_json( response.content, options=ValidateOptions( enable_safe_repairs=True, extract_json=True ) ) if cleaned.json_valid: # Step 2: Pass to LangChain parser if needed parser = JsonOutputParser() # Convert back to string for LangChain parser structured = parser.parse(json.dumps(cleaned.data)) print(structured) # Or use cleaned.data directly print(cleaned.data) else: print(f"Cleaning failed: {cleaned.errors}") if cleaned.likely_truncated: print("Retry with higher max_tokens")

from pydantic import BaseModel, Field import instructor from openai import OpenAI from ai_json_cleanroom import validate_ai_json # Define Pydantic model class User(BaseModel): name: str = Field(description="User's full name") age: int = Field(ge=0, le=150, description="User's age") email: str = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$") tags: list[str] = Field(default_factory=list) # Get raw AI output (without Instructor's patching) client = OpenAI() response = client.chat.completions.create( model="gpt-5.1", messages=[ {"role": "user", "content": "Generate user info for Alice, 30 years old"} ] ) raw_output = response.choices[0].message.content # Step 1: Clean with ai-json-cleanroom result = validate_ai_json(raw_output) if result.json_valid: # Step 2: Map to Pydantic with Instructor try: user = User(**result.data) print(f"User: {user.name}, Age: {user.age}, Email: {user.email}") except Exception as e: print(f"Pydantic validation failed: {e}") else: print(f"JSON cleaning failed: {result.errors}")

import instructor from openai import OpenAI from ai_json_cleanroom import validate_ai_json # Patch OpenAI client with Instructor client = instructor.from_openai(OpenAI()) # If Instructor fails, fallback to Cleanroom try: user = client.chat.completions.create( model="gpt-5.1", response_model=User, messages=[{"role": "user", "content": "Generate user info"}] ) except Exception as e: # Fallback: get raw response and clean manually raw_response = client.chat.completions.create( model="gpt-5.1", messages=[{"role": "user", "content": "Generate user info as JSON"}] ) result = validate_ai_json(raw_response.choices[0].message.content) if result.json_valid: user = User(**result.data)

from openai import OpenAI from ai_json_cleanroom import validate_ai_json client = OpenAI() # Collect streaming chunks chunks = [] stream = client.chat.completions.create( model="gpt-5.1", messages=[{"role": "user", "content": "Generate user JSON"}], stream=True ) for chunk in stream: if chunk.choices[0].delta.content: chunks.append(chunk.choices[0].delta.content) # Validate complete output full_output = ''.join(chunks) result = validate_ai_json(full_output) if result.likely_truncated: print("Stream was truncated, consider retrying with higher limits") print(f"Truncation reasons: {result.errors[0].detail.get('truncation_reasons')}") elif result.json_valid: print(f"Valid JSON received: {result.data}") else: print(f"Validation errors: {result.errors}")

from ai_json_cleanroom import validate_ai_json import openai def generate_with_retry(prompt, schema, max_retries=3): """Generate JSON with automatic retry on validation failure.""" client = openai.OpenAI() for attempt in range(max_retries): response = client.chat.completions.create( model="gpt-5.1", messages=[ {"role": "system", "content": "You are a helpful assistant that returns valid JSON."}, {"role": "user", "content": prompt} ] ) result = validate_ai_json( response.choices[0].message.content, schema=schema ) if result.json_valid: return result.data # Build feedback for retry if result.likely_truncated: prompt = f"{prompt}\n\nIMPORTANT: Your previous response was truncated. Please ensure the complete JSON is returned." else: error_messages = [f"- {e.path}: {e.message}" for e in result.errors] feedback = "\n".join(error_messages) prompt = f"{prompt}\n\nYour previous JSON had these issues:\n{feedback}\n\nPlease fix these and return valid JSON." raise ValueError(f"Failed to generate valid JSON after {max_retries} attempts") # Usage schema = { "type": "object", "required": ["name", "email", "age"], "properties": { "name": {"type": "string"}, "email": {"type": "string", "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$"}, "age": {"type": "integer", "minimum": 0} } } user_data = generate_with_retry( "Generate a user profile for Alice Johnson", schema=schema ) print(user_data)

# Cleanroom automatically extracts the JSON part result = validate_ai_json(chatty_response) print(result.data) # Just the JSON: {"status": "success"} print(result.info['source']) # Tells you where it found it: 'balanced_block'

result = validate_ai_json(truncated_response) if result.likely_truncated: # You know exactly what happened print("Response truncated - reasons:", result.errors[0].detail['truncation_reasons']) # Output: ['unclosed_braces_or_brackets', 'unterminated_string'] # Smart retry with higher token limit retry_with_higher_limit()

output = "{'users': [\"Alice\", \"Bob\"], 'count': 2}"

result = validate_ai_json(output) # Automatically fixes to: {"users": ["Alice", "Bob"], "count": 2}

expectations = [ {"path": "users[*].email", "required": True}, {"path": "metadata.version", "pattern": r"^\d+\.\d+\.\d+$"} ] result = validate_ai_json(ai_output, expectations=expectations) # Validates that all users have emails and version is semver

# First, extract with comments preserved to see them raw_response = ai_output # Clean for parsing result = validate_ai_json(raw_response) # The comments are removed for valid JSON print(result.data) # {"temperature": 0.7, "max_tokens": 100} # If you need the comments, parse them separately from raw_response

from ai_json_cleanroom import validate_ai_json # Same code for ALL models def clean_any_ai_output(output): result = validate_ai_json(output) # Default options handle everything if result.json_valid: return result.data elif result.likely_truncated: raise ValueError("Output truncated - increase token limit") else: raise ValueError(f"Could not parse: {result.errors}") # Works with GPT, Claude, Gemini, Llama, etc.

result = validate_ai_json(your_input, options=ValidateOptions( normalize_curly_quotes="always" # Fixes smart quotes )) print(result.errors) # See specific character positions

result = validate_ai_json(claude_output) print(f"Found JSON in: {result.info['source']}") # 'code_fence' = markdown block # 'balanced_block' = found in text # 'raw' = was already clean

result = validate_ai_json(messy_json) # See all repairs applied for warning in result.warnings: if warning.code == "repaired": print(f"Repairs applied: {warning.detail['applied']}") print(f"Number of changes: {warning.detail['counts']}") # See extraction details print(f"Extraction method: {result.info['source']}") print(f"Parser used: {result.info['parse_backend']}")

# Start without schema to see actual structure result = validate_ai_json(output) print(json.dumps(result.data, indent=2)) # Then add schema gradually schema = {"type": "object"} # Start simple # Add requirements one by one

def validate_ai_json( input_data: Union[str, bytes, Dict, List], schema: Optional[Dict[str, Any]] = None, expectations: Optional[List[Dict[str, Any]]] = None, options: Optional[ValidateOptions] = None ) -> ValidationResult: """  Validate AI output against JSON parseability, schema, and expectations.   Args:  input_data: String, bytes, or already-parsed dict/list  schema: JSON Schema subset for validation  expectations: List of path-based validation rules  options: Configuration for parsing, extraction, and repair   Returns:  ValidationResult with json_valid, errors, warnings, data, and info  """

@dataclass class ValidationResult: json_valid: bool # True if parsing and validation succeeded likely_truncated: bool # True if input appears truncated errors: List[ValidationIssue] # Validation errors warnings: List[ValidationIssue] # Non-blocking warnings data: Optional[Union[Dict, List]] # Parsed JSON if valid, else None info: Dict[str, Any] # Extraction/parsing metadata def to_dict(self) -> Dict[str, Any]: """Convert result to dictionary."""

@dataclass class ValidationIssue: code: ErrorCode # Error type (enum) path: str # JSONPath where error occurred message: str # Human-readable description detail: Optional[Dict[str, Any]] # Additional context def to_dict(self) -> Dict[str, Any]: """Convert issue to dictionary."""

@dataclass class ValidateOptions: # Extraction options strict: bool = False extract_json: bool = True allow_json_in_code_fences: bool = True allow_bare_top_level_scalars: bool = False tolerate_trailing_commas: bool = True stop_on_first_error: bool = False # Repair options enable_safe_repairs: bool = True allow_json5_like: bool = True # Master toggle for JSON5-like repairs replace_constants: bool = True # True/False/None → true/false/null replace_nans_infinities: bool = True # NaN/Infinity → null max_total_repairs: int = 200 max_repairs_percent: float = 0.02 # 2% of input size # Granular repair control (new in v1.1) normalize_curly_quotes: str = "always" # "always"|"auto"|"never" fix_single_quotes: bool = True quote_unquoted_keys: bool = True strip_js_comments: bool = True # Custom repair hooks (new in v1.1) custom_repair_hooks: Optional[List[Callable]] = None

class ErrorCode(str, Enum): PARSE_ERROR = "parse_error" TRUNCATED = "truncated" MISSING_REQUIRED = "missing_required" TYPE_MISMATCH = "type_mismatch" ENUM_MISMATCH = "enum_mismatch" CONST_MISMATCH = "const_mismatch" NOT_ALLOWED_EMPTY = "not_allowed_empty" ADDITIONAL_PROPERTY = "additional_property" PATTERN_MISMATCH = "pattern_mismatch" MIN_LENGTH = "min_length" MAX_LENGTH = "max_length" MIN_ITEMS = "min_items" MAX_ITEMS = "max_items" MINIMUM = "minimum" MAXIMUM = "maximum" REPAIRED = "repaired" # Warning: repair was applied # ... and more

def extract_json_payload( text: str, options: Optional[ValidateOptions] = None ) -> Tuple[Optional[str], Dict[str, Any]]: """  Extract JSON string from raw text.   Returns:  (payload, info) - Payload is raw JSON string or None  """ def detect_truncation(s: str) -> Tuple[bool, List[str]]: """  Heuristic truncation detector.   Returns:  (likely_truncated, reasons)  """

from ai_json_cleanroom import validate_ai_json, ValidateOptions options = ValidateOptions( # Master toggle (backward compatibility) allow_json5_like=True, # Individual repair passes (new in v1.1) fix_single_quotes=True, # 'foo' → "foo" quote_unquoted_keys=True, # {foo: 1} → {"foo": 1} strip_js_comments=True, # Remove // and /* */ comments # Python/JS constants replace_constants=True, # True/False/None → true/false/null replace_nans_infinities=True, # NaN/Infinity → null # Curly quotes handling normalize_curly_quotes="auto", # "always"|"auto"|"never" # Safety limits max_total_repairs=200, max_repairs_percent=0.02 ) result = validate_ai_json(ai_output, options=options)

from ai_json_cleanroom import validate_ai_json, ValidateOptions def my_custom_repair(text: str, options: ValidateOptions): """  Custom repair function.   Args:  text: Current text being repaired  options: ValidateOptions instance   Returns:  (modified_text, changes_count, metadata_dict)  """ modified = text changes = 0 metadata = {} # Example: Replace specific domain patterns if "UNDEFINED" in modified: modified = modified.replace("UNDEFINED", "null") changes += modified.count("null") - text.count("null") metadata["undefined_replacements"] = changes return modified, changes, metadata # Use custom hook options = ValidateOptions( custom_repair_hooks=[my_custom_repair] ) result = validate_ai_json(ai_output, options=options) # Check if custom hook was applied if "custom_hook:my_custom_repair" in result.info.get("repair", {}).get("applied", []): print("Custom repair was applied")

from ai_json_cleanroom import validate_ai_json # Validate API response structure expectations = [ # All users must have email and it must match pattern { "path": "data.users[*].email", "required": True, "pattern": r"^[\w\.-]+@[\w\.-]+\.\w+$" }, # All users must have valid status { "path": "data.users[*].status", "required": True, "in": ["active", "pending", "inactive", "banned"] }, # Nested array validation { "path": "data.users[*].orders[*].total", "type": "number", "minimum": 0 }, # Metadata version must be semver { "path": "metadata.api_version", "required": True, "pattern": r"^\d+\.\d+\.\d+$" }, # Optional field with constraints when present { "path": "data.users[*].phone", "required": False, "pattern": r"^\+?[\d\s\-\(\)]+$" } ] result = validate_ai_json(ai_output, expectations=expectations) # Get specific path errors for error in result.errors: if error.code == "path_not_found": print(f"Missing required path: {error.path}") elif error.code == "pattern_mismatch": print(f"Pattern mismatch at {error.path}: {error.message}")

from ai_json_cleanroom import validate_ai_json result = validate_ai_json(messy_ai_output) if result.warnings: for warning in result.warnings: if warning.code == "repaired": applied_repairs = warning.detail.get("applied", []) counts = warning.detail.get("counts", {}) print("Applied repairs:") for repair in applied_repairs: print(f" - {repair}") print("\nRepair counts:") for repair_type, count in counts.items(): print(f" - {repair_type}: {count}") # Check repair info if "repair" in result.info: repair_info = result.info["repair"] print(f"Total repairs applied: {len(repair_info.get('applied', []))}") if "skipped" in repair_info: print(f"Repairs skipped: {repair_info['skipped']}")

# 1. Clean with Cleanroom result = validate_ai_json(ai_output) # 2. Map to Pydantic model if result.json_valid: user = UserModel(**result.data)

# Use as pre-processor before LangChain parsers cleaned = validate_ai_json(response.content) if cleaned.json_valid: chain_result = parser.parse(json.dumps(cleaned.data))

# Get clean data, then apply your business rules result = validate_ai_json(ai_response) if result.json_valid: your_custom_processor(result.data)

# This is a common scenario... try: data = json.loads(ai_output) except: # Try to extract JSON with regex match = re.search(r'\{.*\}', ai_output, re.DOTALL) if match: try: # Fix quotes maybe? fixed = match.group().replace("'", '"') data = json.loads(fixed) except: # Give up raise ValueError("Can't parse AI output")

result = validate_ai_json(ai_output) # Done.