Output Control & Formatting：精确控制 AI 输出的全面指南

def markdown_format_control():
    prompt = """请严格按照以下 Markdown 格式输出：

# 标题
## 摘要
<200 字以内的摘要>

## 核心要点
- 要点 1：<具体内容>
- 要点 2：<具体内容>
- 要点 3：<具体内容>

## 数据表格
| 指标 | 值 | 说明 |
|------|----|------|
| <名称> | <数值> | <说明> |

## 结论
<150 字以内>

---
注意：每个部分都必须输出，缺失则不合格。"""
    return prompt

# 格式验证函数
import re

def validate_markdown_structure(text: str) -> dict:
    """验证 Markdown 结构完整性"""
    checks = {
        "has_title": bool(re.search(r"^# ", text, re.MULTILINE)),
        "has_summary": bool(re.search(r"^## 摘要", text, re.MULTILINE)),
        "has_keypoints": bool(re.search(r"^## 核心要点", text, re.MULTILINE)),
        "has_table": bool(re.search(r"\|.*\|.*\|", text)),
        "has_conclusion": bool(re.search(r"^## 结论", text, re.MULTILINE)),
        "summary_length_ok": None,
    }
    
    # 检查摘要长度
    summary_match = re.search(
        r"## 摘要\s*\n(.+?)(?=\n##)", text, re.DOTALL
    )
    if summary_match:
        checks["summary_length_ok"] = len(summary_match.group(1)) < 200
    
    return checks

输出必须是一个合法的 JSON 对象，格式如下：
{
  "status": "success | error",
  "data": {
    // 按需填充
  },
  "metadata": {
    "timestamp": "<当前时间戳>",
    "confidence": 0.0-1.0
  }
}

要求：
1. 所有 key 必须使用双引号
2. 字符串值必须使用双引号
3. 不能包含注释或 Markdown 代码块标记
4. 不能有其他文字描述
5. 一定输出纯 JSON

import json
import re

def extract_and_fix_json(text: str) -> dict:
    """从模型输出中提取并修复 JSON"""
    
    # Step 1: 尝试提取 JSON 块
    # 移除代码块标记
    text = re.sub(r'```json\s*|\s*```', '', text)
    
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    
    # Step 2: 常见修复
    fixes = [
        # 修复单引号
        lambda s: s.replace("'", '"'),
        # 修复末尾逗号
        lambda s: re.sub(r',(\s*[}\]])', r'\1', s),
        # 修复注释
        lambda s: re.sub(r'//.*?\n', '\n', s),
        # 修复布尔值
        lambda s: s.replace('True', 'true').replace('False', 'false'),
    ]
    
    for fix in fixes:
        try:
            return json.loads(fix(text))
        except json.JSONDecodeError:
            continue
    
    # Step 3: 最后手段 - 重新生成
    raise ValueError("无法修复 JSON 输出")

<response>
  <metadata>
    <model>gpt-4</model>
    <timestamp>2026-06-25T12:00:00Z</timestamp>
    <type>analysis</type>
  </metadata>
  <content>
    <section id="overview">
      <title>概述</title>
      <paragraph>{overview_text}</paragraph>
    </section>
    <section id="details">
      <title>详细信息</title>
      <item key="point_1">{detail_1}</item>
      <item key="point_2">{detail_2}</item>
    </section>
  </content>
  <validation>
    <passed>true</passed>
    <score>{quality_score}</score>
  </validation>
</response>

from pydantic import BaseModel, Field, validator
from typing import List, Optional
from openai import OpenAI

class ProductReview(BaseModel):
    """结构化的产品评价输出"""
    product_name: str = Field(description="产品名称")
    rating: int = Field(ge=1, le=5, description="评分 1-5")
    pros: List[str] = Field(min_length=1, max_length=5, description="优点列表")
    cons: List[str] = Field(min_length=1, max_length=5, description="缺点列表")
    summary: str = Field(max_length=200, description="总结摘要")
    recommend: bool = Field(description="是否推荐")
    
    @validator('rating')
    def rating_matches_sentiment(cls, v, values):
        if 'cons' in values and len(values['cons']) > 3 and v > 4:
            raise ValueError('评分与缺点数量不符')
        return v

def structured_review_analysis(review_text: str) -> ProductReview:
    client = OpenAI()
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"分析以下产品评价并提取结构化信息：\n{review_text}"
        }],
        functions=[{
            "name": "extract_review",
            "description": "提取产品评价的结构化信息",
            "parameters": ProductReview.schema()
        }],
        function_call={"name": "extract_review"}
    )
    
    return ProductReview.parse_raw(
        response.choices[0].message.function_call.arguments
    )

from enum import Enum

class Sentiment(Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"
    MIXED = "mixed"

class Category(Enum):
    COMPLAINT = "complaint"
    INQUIRY = "inquiry"
    FEEDBACK = "feedback"
    SUGGESTION = "suggestion"
    OTHER = "other"

def constrained_classification(text: str) -> tuple:
    """强制输出限定在枚举范围内"""
    prompt = f"""从以下选项中选择情感倾向（仅输出一个词）：
{', '.join([s.value for s in Sentiment])}

同时从以下选项选择类别（仅输出一个词）：
{', '.join([c.value for c in Category])}

文本：{text}

情感：
类别："""
    
    # ... 调用 LLM 并严格验证输出
    # 如果输出不在枚举范围内，重试或降级处理
    pass

def progressive_expansion(topic: str, max_length: int) -> str:
    """从核心内容开始，逐步扩展文字到目标长度"""
    
    # Phase 1: 生成核心骨架
    skeleton_prompt = f"为 '{topic}' 写一个三行的核心概要。"
    # ... 获取 skeleton
    
    # Phase 2: 扩展到段落
    expand_prompt = f"""基于以下概要，将内容扩展到约 {max_length//2} 字：
{skeleton}

要求：添加具体例子和数据佐证。"""
    # ... 获取 expanded
    
    # Phase 3: 精细化调整
    final_prompt = f"""将以下内容调整到恰好 {max_length} 字：
{expanded}

如果超出则精简，不足则补充细节。"""
    
    # ... 返回最终结果
    pass

class OutputValidator:
    """多层次输出验证"""
    
    def __init__(self, rules: dict):
        self.rules = rules
    
    def validate(self, output: str) -> dict:
        results = {
            "passed": True,
            "checks": [],
            "errors": []
        }
        
        # Layer 1: 结构完整性
        if self.rules.get("require_structure"):
            for section in self.rules["required_sections"]:
                if section not in output:
                    results["passed"] = False
                    results["errors"].append(f"缺少必要段落: {section}")
        
        # Layer 2: 格式正确性
        if self.rules.get("require_json"):
            try:
                json.loads(output)
            except json.JSONDecodeError as e:
                results["passed"] = False
                results["errors"].append(f"JSON 格式错误: {e}")
        
        # Layer 3: 内容合理性
        if self.rules.get("min_length"):
            if len(output) < self.rules["min_length"]:
                results["passed"] = False
                results["errors"].append("内容长度不足")
        
        # Layer 4: 安全过滤
        for pattern in self.rules.get("block_patterns", []):
            if re.search(pattern, output, re.IGNORECASE):
                results["passed"] = False
                results["errors"].append(f"检测到违禁内容: {pattern}")
        
        results["checks"] = len(results["errors"])
        return results

# 使用
validator = OutputValidator({
    "require_structure": True,
    "required_sections": ["摘要", "正文", "结论"],
    "require_json": False,
    "min_length": 100,
    "block_patterns": [r"(密码|passwd|secret)"]
})

class OutputPostProcessor:
    """输出后处理管道"""
    
    def __init__(self):
        self.pipeline = []
    
    def add_step(self, name: str, fn):
        self.pipeline.append((name, fn))
    
    def process(self, output: str) -> str:
        for name, fn in self.pipeline:
            try:
                output = fn(output)
            except Exception as e:
                print(f"后处理步骤 '{name}' 失败: {e}")
        return output

# 示例管道
processor = OutputPostProcessor()
processor.add_step("trim_whitespace", lambda x: x.strip())
processor.add_step("fix_encoding", lambda x: x.encode('utf-8', errors='replace').decode('utf-8'))
processor.add_step("remove_markdown_code_block", 
    lambda x: re.sub(r'```\w*\n?|```', '', x))
processor.add_step("ensure_trailing_newline", 
    lambda x: x if x.endswith('\n') else x + '\n')