# Organizing LLM Output Control and Extraction Techniques in 4 Layers — From Constrained Decoding to Prompt Design

# LLM Output Control: 4-Layer Classification ## Overview ``` ┌─────────────────────────────────────────┐ │ Prompt Layer │ ← Instruction-based control ├─────────────────────────────────────────┤ │ API Layer │ ← Parameter-based control ├─────────────────────────────────────────┤ │ Decoding Layer │ ← Generation algorithm control ├─────────────────────────────────────────┤ │ Post-processing Layer │ ← Output correction/validation └─────────────────────────────────────────┘ Reliability: Decoding > API > Post-processing > Prompt Flexibility: Prompt > Post-processing > API > Decoding ``` --- ## Layer 1: Decoding Layer ### Mechanism Controls the token selection process itself during generation ``` Logits → [Temperature] → [Top-p/Top-k] → [Logit Bias] → Softmax → Token Selection ``` ### Key Parameters ```python # Direct control via transformers library from transformers import AutoModelForCausalLM, AutoTokenizer import torch model_name = "gpt2" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained(model_name) # ── 1. Temperature Control ────────────────────────────── def generate_with_temperature(prompt: str, temperature: float): """ temperature=0.0 → Deterministic (always highest probability) temperature=1.0 → Default distribution temperature=2.0 → High randomness """ inputs = tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=50, temperature=temperature, do_sample=temperature > 0, # False when temp=0 → greedy ) return tokenizer.decode(outputs[0], skip_special_tokens=True) # ── 2. Top-p (Nucleus Sampling) ───────────────────────── def generate_with_top_p(prompt: str, top_p: float = 0.9): """ Selects from tokens whose cumulative probability reaches top_p top_p=0.9 → Uses tokens covering 90% of probability mass """ inputs = tokenizer(prompt, return_tensors="pt") outputs = model.generate( **inputs, max_new_tokens=50, do_sample=True, top_p=top_p, temperature=1.0, ) return tokenizer.decode(outputs[0], skip_special_tokens=True) # ── 3. Logit Bias (Forbidden/Forced Tokens) ───────────── def generate_with_logit_bias(prompt: str): """ Directly manipulates logits to prohibit or force specific tokens """ inputs = tokenizer(prompt, return_tensors="pt") vocab_size = tokenizer.vocab_size # Create logit processor from transformers import LogitsProcessor, LogitsProcessorList class CustomLogitProcessor(LogitsProcessor): def __init__(self, forbidden_token_ids: list, bias: float = -100.0): self.forbidden_ids = forbidden_token_ids self.bias = bias def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor): for token_id in self.forbidden_ids: scores[:, token_id] = self.bias # Nearly impossible return scores # Example: Prohibit specific words # "violence" related tokens forbidden_tokens = tokenizer.encode(" violence", add_special_tokens=False) processor = CustomLogitProcessor(forbidden_token_ids=forbidden_tokens) outputs = model.generate( **inputs, max_new_tokens=50, logits_processor=LogitsProcessorList([processor]), ) return tokenizer.decode(outputs[0], skip_special_tokens=True) # ── 4. Constrained Beam Search ────────────────────────── def generate_constrained(prompt: str, must_include: list[str]): """ Force output to include specific phrases """ inputs = tokenizer(prompt, return_tensors="pt") # Encode constraint phrases constraints_ids = [ tokenizer.encode(phrase, add_special_tokens=False) for phrase in must_include ] from transformers import PhrasalConstraint constraints = [PhrasalConstraint(ids) for ids in constraints_ids] outputs = model.generate( **inputs, max_new_tokens=100, constraints=constraints, num_beams=5, # Beam search required ) return tokenizer.decode(outputs[0], skip_special_tokens=True) ``` ### Structured Output with Outlines ```python # Grammar-based constrained generation import outlines import outlines.models as models from pydantic import BaseModel from enum import Enum # ── Pydantic schema constrained generation ─────────────── class Sentiment(str, Enum): positive = "positive" negative = "negative" neutral = "neutral" class ReviewAnalysis(BaseModel): sentiment: Sentiment confidence: float # 0.0-1.0 key_points: list[str] summary: str # Model initialization model = models.transformers("mistralai/Mistral-7B-v0.1") # Generate output strictly conforming to schema generator = outlines.generate.json(model, ReviewAnalysis) result = generator( "Analyze this review: 'Great product, highly recommend!'" ) # result is guaranteed to be a ReviewAnalysis instance print(type(result)) # <class 'ReviewAnalysis'> print(result.sentiment) # Sentiment.positive (guaranteed) # ── Regex constrained generation ──────────────────────── # Force phone number format phone_generator = outlines.generate.regex( model, r"\d{3}-\d{4}-\d{4}" # Japanese phone number format ) phone = phone_generator("My phone number is:") print(phone) # Guaranteed format like "090-1234-5678" # ── Choice constrained generation ─────────────────────── choice_generator = outlines.generate.choice( model, ["Yes", "No", "Uncertain"] ) answer = choice_generator("Is this statement correct?") print(answer) # Guaranteed to be one of three values ``` --- ## Layer 2: API Layer ### OpenAI API Control ```python from openai import OpenAI from pydantic import BaseModel, Field, validator import json client = OpenAI() # ── 1. JSON Mode ───────────────────────────────────────── def extract_with_json_mode(text: str) -> dict: """ Force JSON output (OpenAI gpt-4-turbo and later) Note: Schema is not guaranteed, only JSON format is """ response = client.chat.completions.create( model="gpt-4-turbo-preview", response_format={"type": "json_object"}, # JSON mode ON messages=[ { "role": "system", "content": "Extract information as JSON. " "Keys: name, age, email" }, {"role": "user", "content": text} ], temperature=0, # Deterministic seed=42, # Reproducibility (best effort) ) raw = response.choices[0].message.content return json.loads(raw) # Parsing basically never fails # ── 2. Structured Outputs (gpt-4o and later) ──────────── class PersonInfo(BaseModel): name: str = Field(description="Full name") age: int = Field(ge=0, le=150, description="Age") email: str = Field(description="Email address") tags: list[str] = Field(default_factory=list) def extract_structured(text: str) -> PersonInfo: """ Schema-level guaranteed output Pydantic model is directly respected """ response = client.beta.chat.completions.parse( model="gpt-4o-2024-08-06", messages=[ {"role": "system", "content": "Extract person information."}, {"role": "user", "content": text} ], response_format=PersonInfo, # Pydantic model specified directly ) # Parsing complete, type guaranteed person = response.choices[0].message.parsed assert isinstance(person, PersonInfo) # Always True return person # ── 3. Function Calling ────────────────────────────────── def use_function_calling(user_query: str): """ Route to specific functions based on intent classification """ tools = [ { "type": "function", "function": { "name": "search_database", "description": "Search product database", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search keywords" }, "category": { "type": "string", "enum": ["electronics", "clothing", "food"], "description": "Product category" }, "max_price": { "type": "number", "description": "Upper price limit" } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "get_order_status", "description": "Get order status", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "pattern": "^ORD-[0-9]{6}$" } }, "required": ["order_id"] } } } ] response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": user_query}], tools=tools, tool_choice="auto", # "auto" | "required" | specific function ) message = response.choices[0].message if message.tool_calls: for tool_call in message.tool_calls: func_name = tool_call.function.name args = json.loads(tool_call.function.arguments) print(f"Called: {func_name}({args})") # Execute actual function... return message # ── 4. Max Tokens & Stop Sequences ────────────────────── def controlled_generation(prompt: str): """ Control output length and stopping conditions """ response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], max_tokens=500, # Maximum length limit stop=["###", "END", "\n\n\n"], # Stop sequences presence_penalty=0.6, # Suppress topic repetition frequency_penalty=0.3, # Suppress word repetition logit_bias={ "1734": -100, # Token ID 1734 (example) completely forbidden "198": -50, # Newline suppression } ) return response.choices[0].message.content ``` --- ## Layer 3: Post-processing Layer ### Validation & Correction Pipeline ```python from dataclasses import dataclass from typing import Any, Callable import re import json from enum import Enum class ValidationStatus(Enum): OK = "ok" FIXED = "fixed" FAILED = "failed" @dataclass class ValidationResult: status: ValidationStatus data: Any errors: list[str] fixes_applied: list[str] # ── 1. JSON Repair ─────────────────────────────────────── def repair_json(raw: str) -> ValidationResult: """ Repair common LLM JSON output errors """ errors = [] fixes = [] # Step1: Extract code blocks code_block = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', raw) if code_block: raw = code_block.group(1) fixes.append("Extracted from code block") # Step2: Try direct parsing try: return ValidationResult( status=ValidationStatus.OK, data=json.loads(raw), errors=[], fixes_applied=fixes ) except json.JSONDecodeError as e: errors.append(f"JSON parse error: {e}") # Step3: Common pattern repair text = raw.strip() # Trailing comma removal text = re.sub(r',\s*}', '}', text) text = re.sub(r',\s*]', ']', text) fixes.append("Removed trailing commas") # Single quote → double quote text = re.sub(r"'([^']*)':", r'"\1":', text) fixes.append("Replaced single quotes with double quotes") # Unquoted keys text = re.sub(r'(\w+):', r'"\1":', text) fixes.append("Added quotes to unquoted keys") # Try parsing again try: return ValidationResult( status=ValidationStatus.FIXED, data=json.loads(text), errors=errors, fixes_applied=fixes ) except json.JSONDecodeError: pass # Step4: Use json_repair library try: from json_repair import repair_json as lib_repair repaired = lib_repair(raw) fixes.append("Used json_repair library") return ValidationResult( status=ValidationStatus.FIXED, data=json.loads(repaired), errors=errors, fixes_applied=fixes ) except Exception as e: errors.append(f"json_repair also failed: {e}") return ValidationResult( status=ValidationStatus.FAILED, data=None, errors=errors, fixes_applied=fixes ) # ── 2. Schema Validation ───────────────────────────────── from pydantic import BaseModel, ValidationError class ProductReview(BaseModel): rating: int = Field(ge=1, le=5) title: str = Field(min_length=1, max_length=100) body: str = Field(min_length=10) pros: list[str] = Field(default_factory=list) cons: list[str] = Field(default_factory=list) def validate_and_coerce(raw_dict: dict) -> ValidationResult: """ Validate schema and auto-coerce where possible """ fixes = [] # Pre-processing: type coercion if "rating" in raw_dict: raw_rating = raw_dict["rating"] if isinstance(raw_rating, str): # "4/5" → 4, "4 stars" → 4 match = re.search(r'\d+', raw_rating) if match: raw_dict["rating"] = int(match.group()) fixes.append(f"Converted rating: '{raw_rating}' → {raw_dict['rating']}") # Validation try: result = ProductReview(**raw_dict) return ValidationResult( status=ValidationStatus.FIXED if fixes else ValidationStatus.OK, data=result, errors=[], fixes_applied=fixes ) except ValidationError as e: return ValidationResult( status=ValidationStatus.FAILED, data=None, errors=[str(err) for err in e.errors()], fixes_applied=fixes ) # ── 3. Content Filtering ───────────────────────────────── import re from dataclasses import dataclass @dataclass class FilterResult: passed: bool filtered_text: str violations: list[dict] class ContentFilter: def __init__(self): self.rules = [ { "name": "phone_number", "pattern": r'\d{2,4}-\d{2,4}-\d{4}', "action": "redact", "replacement": "[PHONE REDACTED]" }, { "name": "email", "pattern": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', "action": "redact", "replacement": "[EMAIL REDACTED]" }, { "name": "excessive_caps", "pattern": r'[A-Z]{10,}', "action": "warn", "replacement": None }, ] self.block_patterns = [ r'(?i)(how to make|instructions for).{0,20}(bomb|weapon|poison)', r'(?i)ignore (all |previous |your )?(instructions|rules|guidelines)', ] def filter(self, text: str) -> FilterResult: violations = [] filtered = text # Block check for pattern in self.block_patterns: if re.search(pattern, text): violations.append({ "type": "blocked", "pattern": pattern, "severity": "high" }) return FilterResult( passed=False, filtered_text="", violations=violations ) # Apply redaction rules for rule in self.rules: matches = re.findall(rule["pattern"], filtered) if matches: violations.append({ "type": rule["name"], "count": len(matches), "severity": "medium" if rule["action"] == "redact" else "low" }) if rule["action"] == "redact": filtered = re.sub( rule["pattern"], rule["replacement"], filtered ) return FilterResult( passed=True, filtered_text=filtered, violations=violations ) # ── 4. Retry with Validation ───────────────────────────── import asyncio from typing import TypeVar, Type T = TypeVar('T', bound=BaseModel) async def generate_with_retry( prompt: str, schema: Type[T], max_retries: int = 3, model: str = "gpt-4o" ) -> T: """ Regenerate with error feedback on validation failure """ last_error = None for attempt in range(max_retries): # Add error info to prompt on retry if last_error and attempt > 0: retry_prompt = f"""{prompt} Previous attempt failed with error: {last_error} Please fix this and try again.""" else: retry_prompt = prompt try: response = client.beta.chat.completions.parse( model=model, messages=[ { "role": "system", "content": f"Output must conform to this schema: {schema.model_json_schema()}" }, {"role": "user", "content": retry_prompt} ], response_format=schema, temperature=0.1 * attempt, # Slightly increase randomness on retry ) result = response.choices[0].message.parsed if result is not None: return result except Exception as e: last_error = str(e) print(f"Attempt {attempt + 1} failed: {e}") await asyncio.sleep(2 ** attempt) # Exponential backoff raise ValueError(f"Failed after {max_retries} retries. Last error: {last_error}") ``` --- ## Layer 4: Prompt Layer ### Systematic Prompt Engineering ```python from string import Template from typing import Optional import yaml # ── 1. Structured Prompt Template ─────────────────────── EXTRACTION_PROMPT = """ ## Role You are an expert information extraction AI. ## Output Format Respond ONLY in the following JSON format: ```json {{ "entities": [ {{ "text": "entity text", "type": "PERSON|ORG|LOCATION|DATE|MONEY", "confidence": 0.0-1.0 }} ], "summary": "one sentence summary", "sentiment": "positive|negative|neutral" }} ``` ## Constraints - Output ONLY the JSON above, no other text - Do not include code blocks or markdown - "confidence" must be a number between 0.0 and 1.0 - "sentiment" must be exactly one of the three values above ## Input Text {input_text} ## Output (JSON only): """ # ── 2. Few-shot Examples ───────────────────────────────── FEW_SHOT_PROMPT = """ Convert natural language to SQL. ## Examples Input: Show me all customers from Tokyo Output: SELECT * FROM customers WHERE city = 'Tokyo'; Input: Average order amount in the past 30 days Output: SELECT AVG(amount) FROM orders WHERE created_at >= DATE_SUB(NOW(), INTERVAL 30 DAY); Input: Product ranking by sales volume (top 5) Output: SELECT p.name, SUM(oi.quantity) as total_sold FROM products p JOIN order_items oi ON p.id = oi.product_id GROUP BY p.id ORDER BY total_sold DESC LIMIT 5; ## New Input Input: {user_query} Output:""" # ── 3. Chain of Thought ────────────────────────────────── COT_PROMPT = """ Analyze the following situation and provide a recommendation. ## Instructions Before reaching your conclusion, explicitly go through these steps: Step 1: Identify key facts Step 2: List applicable constraints Step 3: Enumerate options Step 4: Evaluate pros and cons of each option Step 5: Draw conclusion ## Output Format ``` [Step 1: Key Facts] - ... [Step 2: Constraints] - ... [Step 3: Options] 1. ... 2. ... [Step 4: Evaluation] Option 1: Pros... / Cons... Option 2: Pros... / Cons... [Conclusion] Recommended: ... Rationale: ... ``` ## Situation {situation} """ # ── 4. Prompt Builder Class ────────────────────────────── class PromptBuilder: def __init__(self, config_path: Optional[str] = None): self.templates = {} self.default_constraints = [ "Output in Japanese", "No markdown formatting", "Be concise" ] if config_path: with open(config_path) as f: self.templates = yaml.safe_load(f) def build( self, task_type: str, user_input: str, output_schema: Optional[dict] = None, few_shot_examples: Optional[list] = None, constraints: Optional[list] = None, ) -> str: parts = [] # System role parts.append(f"## Task\n{self._get_task_description(task_type)}\n") # Constraints all_constraints = self.default_constraints.copy() if constraints: all_constraints.extend(constraints) parts.append("## Constraints\n" + "\n".join(f"- {c}" for c in all_constraints)) # Output schema if output_schema: import json parts.append(f"\n## Required Output Format\n```json\n{json.dumps(output_schema, indent=2, ensure_ascii=False)}\n```") # Few-shot examples if few_shot_examples: examples_text = "\n## Examples\n" for ex in few_shot_examples: examples_text += f"\nInput: {ex['input']}\nOutput: {ex['output']}\n" parts.append(examples_text) # User input parts.append(f"\n## Input\n{user_input}") # Output prompt if output_schema: parts.append("\n## Output (JSON only):") return "\n".join(parts) def _get_task_description(self, task_type: str) -> str: descriptions = { "extraction": "Extract structured information from text accurately.", "classification": "Classify the given input into appropriate categories.", "summarization": "Summarize the given text concisely and accurately.", "translation": "Translate accurately while preserving nuance.", "qa": "Answer questions accurately based on given context.", } return descriptions.get(task_type, task_type) # Usage example builder = PromptBuilder() prompt = builder.build( task_type="extraction", user_input="Tanaka Taro (age 32) can be reached at taro@example.com.", output_schema={ "name": "string", "age": "integer", "email": "string" }, constraints=["Extract only explicitly stated information", "Use null for unknown values"] ) print(prompt) ``` --- ## Multi-layer Defense Pattern for Production ```python import asyncio import logging from dataclasses import dataclass from typing import TypeVar, Type, Optional, Any from enum import Enum import time logger = logging.getLogger(__name__) T = TypeVar('T', bound=BaseModel) class ControlMethod(Enum): STRUCTURED_OUTPUT = "structured_output" JSON_MODE = "json_mode" PROMPT_ONLY = "prompt_only" @dataclass class GenerationConfig: model: str = "gpt-4o" temperature: float = 0.1 max_tokens: int = 2000 max_retries: int = 3 timeout: float = 30.0 fallback_model: str = "gpt-4-turbo" class MultiLayerController: """ 4-layer defense output control system """ def __init__(self, config: GenerationConfig): self.config = config self.content_filter = ContentFilter() self.metrics = {"attempts": 0, "successes": 0, "failures": 0} async def generate( self, prompt: str, schema: Type[T], context: Optional[str] = None, ) -> T: """ Layer 1: Prompt → Layer 2: API → Layer 3: Post-process → Retry """ # ── Layer 4: Prompt Optimization ──────────────── enhanced_prompt = self._enhance_prompt(prompt, schema) for attempt in range(self.config.max_retries): self.metrics["attempts"] += 1 try: # ── Layer 2: API Control ───────────────── result = await self._api_call_with_timeout( enhanced_prompt, schema, attempt ) # ── Layer 3: Post-processing ───────────── validated = self._post_process(result, schema) # ── Content Filter ─────────────────────── self._check_content_safety(validated) self.metrics["successes"] += 1 logger.info(f"Success on attempt {attempt + 1}") return validated except ContentViolationError as e: # Content violation: do not retry self.metrics["failures"] += 1 logger.error(f"Content violation: {e}") raise except Exception as e: logger.warning(f"Attempt {attempt + 1} failed: {e}") if attempt < self.config.max_retries - 1: # Add error feedback to prompt enhanced_prompt = self._add_error_feedback( enhanced_prompt, str(e), attempt ) await asyncio.sleep(2 ** attempt) else: self.metrics["failures"] += 1 raise def _enhance_prompt(self, prompt: str, schema: Type[T]) -> str: """Layer 4: Prompt enhancement""" schema_str = schema.model_json_schema() return f"""You must respond with valid JSON conforming to exactly this schema: Schema: ```json {json.dumps(schema_str, indent=2)} ``` Rules: - Output ONLY the JSON object, no other text - All required fields must be included - Values must match specified types - Do not use null for required fields Task: {prompt} JSON Response:""" async def _api_call_with_timeout( self, prompt: str, schema: Type[T], attempt: int ) -> T: """Layer 2: API call with timeout""" # Use fallback model on final retry model = ( self.config.fallback_model if attempt == self.config.max_retries - 1 else self.config.model ) try: # Structured Outputs (most reliable) response = await asyncio.wait_for( asyncio.to_thread( client.beta.chat.completions.parse, model=model, messages=[{"role": "user", "content": prompt}], response_format=schema, temperature=self.config.temperature, max_tokens=self.config.max_tokens, ), timeout=self.config.timeout ) result = response.choices[0].message.parsed if result is None: raise ValueError("Parsing failed (refusal or error)") return result except asyncio.TimeoutError: raise TimeoutError(f"API call timed out after {self.config.timeout}s") def _post_process(self, result: T, schema: Type[T]) -> T: """Layer 3: Post-processing validation""" # Re-validate with Pydantic try: # Convert to dict and re-validate (strict mode) data = result.model_dump() validated = schema.model_validate(data, strict=False) return validated except Exception as e: raise ValueError(f"Post-process validation failed: {e}") def _check_content_safety(self, result: Any) -> None: """Content safety check""" # Recursively check all string fields def check_value(v): if isinstance(v, str): filter_result = self.content_filter.filter(v) if not filter_result.passed: raise ContentViolationError( f"Content violation detected: {filter_result.violations}" ) elif isinstance(v, dict): for val in v.values(): check_value(val) elif isinstance(v, list): for item in v: check_value(item) if hasattr(result, 'model_dump'): check_value(result.model_dump()) def _add_error_feedback( self, prompt: str, error: str, attempt: int ) -> str: """Add error information for retry""" return f"""{prompt} --- Previous attempt #{attempt + 1} failed: Error: {error} Please fix this issue and try again. Focus: Ensure strict compliance with the schema.""" def get_metrics(self) -> dict: total = self.metrics["attempts"] if total > 0: return { **self.metrics, "success_rate": self.metrics["successes"] / total, "failure_rate": self.metrics["failures"] / total, } return self.metrics class ContentViolationError(Exception): pass # ── Usage ───────────────────────────────────────────────── async def main(): class ArticleSummary(BaseModel): title: str = Field(min_length=1, max_length=100) summary: str = Field(min_length=50) key_points: list[str] = Field(min_length=1, max_length=5) category: str = Field(pattern=r'^(tech|business|science|culture)$') sentiment_score: float = Field(ge=-1.0, le=1.0) config = GenerationConfig( model="gpt-4o", temperature=0.1, max_retries=3, timeout=30.0, fallback_model="gpt-4-turbo" ) controller = MultiLayerController(config) article_text = """ OpenAI announced GPT-5, claiming 50% improvement over GPT-4 on benchmarks. The new model features enhanced reasoning and multimodal capabilities. """ result = await controller.generate( prompt=f"Summarize this article: {article_text}", schema=ArticleSummary, ) print(f"Title: {result.title}") print(f"Category: {result.category}") print(f"Sentiment: {result.sentiment_score}") print(f"Key Points: {result.key_points}") print(f"\nMetrics: {controller.get_metrics()}") asyncio.run(main()) ``` --- ## Layer Selection Guide ``` Goal Recommended Layer ───────────────────────────────────────────────────────── JSON format guarantee API (Structured Outputs) Schema strict validation API + Post-processing Specific token prohibition Decoding (Logit Bias) Grammar-constrained output Decoding (Outlines) Accuracy improvement Prompt (Few-shot/CoT) Cost reduction API (max_tokens) PII removal Post-processing (Filter) Idempotency/Reproducibility API (seed) + Prompt Fail-safe Post-processing (Retry) ``` ``` Reliability Ranking: Outlines/Grammar ████████████████████ 100% (mathematically guaranteed) Structured Output ██████████████████░░ 90% (API-level guarantee) JSON Mode █████████████░░░░░░░ 70% (format only, not schema) Post-processing ████████████░░░░░░░░ 65% (validation + repair) Prompt ████████░░░░░░░░░░░░ 40% (model-dependent) ``` For production environments, combining at least **API + Post-processing + Prompt** (3 layers) is recommended.