结构化输出技术：从JSON Mode到Function Calling

import json
from openai import OpenAI

client = OpenAI()

# OpenAI JSON Mode
response = client.chat.completions.create(
    model="gpt-4o-2026",
    response_format={"type": "json_object"},
    messages=[
        {
            "role": "system",
            "content": """你是一个信息提取助手。
请将用户输入提取为JSON格式，包含以下字段：
- name: 姓名
- age: 年龄（整数）
- skills: 技能列表（字符串数组）
- experience: 工作经验（整数，单位年）
"""
        },
        {
            "role": "user",
            "content": "张三，28岁，精通Python和JavaScript，有5年开发经验"
        }
    ]
)

result = json.loads(response.choices[0].message.content)
print(result)
# 输出: {"name": "张三", "age": 28, "skills": ["Python", "JavaScript"], "experience": 5}

# 2026年最新：JSON Schema强化约束
response = client.chat.completions.create(
    model="gpt-4o-2026",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "employee_info",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string", "minLength": 1, "maxLength": 50},
                    "age": {"type": "integer", "minimum": 18, "maximum": 65},
                    "skills": {
                        "type": "array",
                        "items": {"type": "string"},
                        "minItems": 1,
                        "maxItems": 20
                    },
                    "experience": {"type": "integer", "minimum": 0, "maximum": 50},
                    "level": {
                        "type": "string",
                        "enum": ["junior", "mid", "senior", "expert"]
                    }
                },
                "required": ["name", "age", "skills", "experience", "level"],
                "additionalProperties": False
            }
        }
    },
    messages=[...]
)

from dataclasses import dataclass
from typing import Callable
import inspect

@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: dict  # JSON Schema
    
class FunctionCallingSystem:
    """
    2026年Function Calling最佳实践
    """
    
    def __init__(self, model_client):
        self.model = model_client
        self.tools: dict[str, ToolDefinition] = {}
        self.handlers: dict[str, Callable] = {}
    
    def register_function(self, func: Callable, description: str):
        """注册可调用函数"""
        # 自动从函数签名生成Schema
        sig = inspect.signature(func)
        params = {}
        required = []
        
        for name, param in sig.parameters.items():
            param_type = param.annotation
            json_type = self._python_type_to_json(param_type)
            params[name] = {
                "type": json_type,
                "description": self._extract_param_doc(func, name)
            }
            if param.default == inspect.Parameter.empty:
                required.append(name)
        
        tool = ToolDefinition(
            name=func.__name__,
            description=description or func.__doc__,
            parameters={
                "type": "object",
                "properties": params,
                "required": required
            }
        )
        
        self.tools[func.__name__] = tool
        self.handlers[func.__name__] = func
    
    async def execute_with_functions(self, user_message: str) -> str:
        """带函数调用的对话"""
        messages = [{"role": "user", "content": user_message}]
        tools = [t.__dict__ for t in self.tools.values()]
        
        while True:
            response = await self.model.chat(
                messages=messages,
                tools=tools,
                tool_choice="auto"
            )
            
            message = response.choices[0].message
            messages.append(message)
            
            if not message.tool_calls:
                # 模型没有调用工具，返回最终回复
                return message.content
            
            # 执行函数调用
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
                
                # 参数验证
                validation = self._validate_arguments(func_name, func_args)
                if not validation["valid"]:
                    result = f"参数错误: {validation['errors']}"
                else:
                    # 执行函数
                    try:
                        handler = self.handlers[func_name]
                        result = await handler(**func_args)
                    except Exception as e:
                        result = f"执行错误: {str(e)}"
                
                # 将结果返回给模型
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result)
                })

# 定义工具函数
@dataclass
class SearchResult:
    title: str
    url: str
    snippet: str

async def search_web(query: str, max_results: int = 5) -> list[dict]:
    """搜索网络内容
    
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数（默认5）
    """
    # 实际搜索逻辑
    results = await search_engine.search(query, limit=max_results)
    return [{"title": r.title, "url": r.url, "snippet": r.snippet} 
            for r in results]

async def get_weather(city: str, unit: str = "celsius") -> dict:
    """获取指定城市的天气信息
    
    Args:
        city: 城市名称
        unit: 温度单位（celsius或fahrenheit）
    """
    weather = await weather_api.get(city, unit)
    return {
        "city": city,
        "temperature": weather.temp,
        "condition": weather.condition,
        "humidity": weather.humidity
    }

async def send_email(to: str, subject: str, body: str) -> dict:
    """发送邮件
    
    Args:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件正文
    """
    await email_service.send(to, subject, body)
    return {"status": "sent", "to": to}

# 注册并使用
system = FunctionCallingSystem(model_client)
system.register_function(search_web, "搜索网络获取最新信息")
system.register_function(get_weather, "查询天气信息")
system.register_function(send_email, "发送邮件")

# 执行
response = await system.execute_with_functions(
    "帮我查一下北京今天的天气，然后把结果发邮件给 zhangsan@example.com"
)

class ConstrainedDecoder:
    """
    约束解码器
    在每个生成步骤中，只允许符合Schema的token
    """
    
    def __init__(self, model, grammar):
        self.model = model
        self.grammar = grammar  # 定义输出格式的语法
        
    def generate(self, prompt: str, max_tokens: int = 500) -> str:
        """约束生成"""
        generated = ""
        
        for _ in range(max_tokens):
            # 获取下一个token的概率分布
            logits = self.model.get_next_token_logits(prompt + generated)
            
            # 获取当前状态下的合法token
            valid_tokens = self.grammar.get_valid_tokens(generated)
            
            # 将非法token的概率设为负无穷
            for token_id in range(len(logits)):
                if token_id not in valid_tokens:
                    logits[token_id] = float('-inf')
            
            # 选择概率最高的合法token
            next_token = self.model.sample(logits, temperature=0.0)
            
            if next_token == self.model.eos_token_id:
                break
            
            generated += self.model.decode(next_token)
        
        return generated

from outlines import models, generate
from pydantic import BaseModel, Field
from typing import Literal

# 定义输出Schema
class MovieReview(BaseModel):
    title: str = Field(description="电影标题")
    rating: int = Field(ge=1, le=10, description="评分1-10")
    sentiment: Literal["positive", "neutral", "negative"]
    summary: str = Field(min_length=50, max_length=200, 
                         description="50-200字摘要")
    key_points: list[str] = Field(min_items=2, max_items=5)
    recommendation: bool

# 约束生成
model = models.transformers("meta-llama/Llama-4-70B")
generator = generate.json(model, MovieReview)

review = generator(
    "请对电影《流浪地球3》进行评价。"
    "考虑剧情、特效、演技等方面。"
)
# 输出保证符合MovieReview schema
print(review.title)        # "流浪地球3"
print(review.rating)       # 8
print(review.sentiment)    # "positive"
print(review.key_points)   # ["特效震撼", "剧情紧凑", ...]

# 使用正则表达式约束
import outlines

model = models.transformers("Qwen/Qwen3-72B")

# 约束为电话号码格式
phone_generator = generate.regex(
    model,
    r"\d{3}-\d{3}-\d{4}"
)
phone = phone_generator("生成一个随机电话号码")
# 保证格式: "123-456-7890"

# 约束为选择题格式
choice_generator = generate.choice(
    model,
    ["A", "B", "C", "D"]
)
answer = choice_generator("以下哪个是Python的数据类型？\nA. int\nB. string\nC. float\nD. 以上都是")
# 保证输出: "D"

# 约束为CSV格式
csv_generator = generate.format(
    model,
    "csv",
    columns=["name", "age", "city"]
)

class StructuredExtractionPipeline:
    """
    多步骤结构化信息提取
    适用于复杂文档解析场景
    """
    
    def __init__(self, model):
        self.model = model
    
    async def extract_from_document(self, document: str) -> dict:
        """从文档中提取结构化信息"""
        
        # 步骤1：文档分类
        doc_type = await self._classify_document(document)
        
        # 步骤2：根据类型选择提取Schema
        schema = self._get_schema_for_type(doc_type)
        
        # 步骤3：分段提取（长文档）
        if len(document) > 5000:
            segments = self._split_document(document)
            partial_results = []
            for segment in segments:
                result = await self._extract_segment(segment, schema)
                partial_results.append(result)
            # 合并结果
            final_result = await self._merge_results(partial_results, schema)
        else:
            final_result = await self._extract_segment(document, schema)
        
        # 步骤4：验证和修正
        final_result = await self._validate_and_fix(final_result, schema)
        
        return final_result
    
    async def _extract_segment(self, segment: str, schema: dict) -> dict:
        """使用Function Calling提取"""
        response = await self.model.chat(
            messages=[
                {
                    "role": "system",
                    "content": f"""你是信息提取专家。
请从给定文本中提取信息，严格按照以下Schema输出：
{json.dumps(schema, ensure_ascii=False, indent=2)}

注意：
1. 只输出Schema中定义的字段
2. 如果信息不存在，使用null
3. 日期格式：YYYY-MM-DD
4. 金额格式：数字，单位为元
"""
                },
                {"role": "user", "content": segment}
            ],
            response_format={"type": "json_schema", "json_schema": schema}
        )
        
        return json.loads(response.choices[0].message.content)
    
    async def _validate_and_fix(self, result: dict, schema: dict) -> dict:
        """验证并修正提取结果"""
        errors = self._validate_schema(result, schema)
        
        if errors:
            # 让模型修正错误
            fix_prompt = f"""以下提取结果有错误，请修正：

原始结果：{json.dumps(result, ensure_ascii=False)}
错误信息：{json.dumps(errors, ensure_ascii=False)}
Schema要求：{json.dumps(schema, ensure_ascii=False)}

请输出修正后的JSON：
"""
            response = await self.model.chat(
                messages=[{"role": "user", "content": fix_prompt}],
                response_format={"type": "json_object"}
            )
            return json.loads(response.choices[0].message.content)
        
        return result

class RobustStructuredOutput:
    """带错误处理的结构化输出"""
    
    MAX_RETRIES = 3
    
    async def generate_with_retry(self, prompt: str, schema: dict) -> dict:
        """带重试的结构化生成"""
        
        for attempt in range(self.MAX_RETRIES):
            try:
                # 尝试生成
                response = await self._generate(prompt, schema)
                
                # 验证
                self._validate(response, schema)
                
                return response
                
            except JSONDecodeError as e:
                # JSON解析失败
                print(f"尝试 {attempt+1}: JSON解析失败 - {e}")
                prompt = self._add_error_hint(prompt, str(e), schema)
                
            except SchemaValidationError as e:
                # Schema验证失败
                print(f"尝试 {attempt+1}: Schema验证失败 - {e}")
                prompt = self._add_correction_hint(prompt, str(e))
                
            except Exception as e:
                print(f"尝试 {attempt+1}: 未知错误 - {e}")
        
        # 所有重试失败，使用降级方案
        return await self._fallback_extraction(prompt, schema)
    
    async def _fallback_extraction(self, prompt: str, schema: dict) -> dict:
        """降级方案：使用XML标签提取"""
        xml_prompt = f"""{prompt}

请用以下XML格式输出：
<result>
"""
        for field, spec in schema["properties"].items():
            xml_prompt += f"  <{field}></{field}>\n"
        xml_prompt += "</result>"
        
        response = await self.model.generate(xml_prompt)
        
        # 解析XML
        return self._parse_xml_response(response, schema)

# 1. 缓存Schema解析
from functools import lru_cache

@lru_cache(maxsize=100)
def get_compiled_schema(schema_json: str):
    """缓存编译后的Schema"""
    return jsonschema.Draft7Validator(
        json.loads(schema_json)
    )

# 2. 批量处理
async def batch_structured_generation(
    prompts: list[str], 
    schema: dict,
    concurrency: int = 10
):
    """并发批量结构化生成"""
    semaphore = asyncio.Semaphore(concurrency)
    
    async def process_one(prompt):
        async with semaphore:
            return await generate_structured(prompt, schema)
    
    return await asyncio.gather(*[process_one(p) for p in prompts])

# 3. 流式解析
async def stream_structured_output(prompt: str, schema: dict):
    """流式输出+实时解析"""
    buffer = ""
    async for chunk in model.stream(prompt):
        buffer += chunk
        # 尝试增量解析
        try:
            partial = json.loads(buffer)
            yield partial
        except:
            continue  # 等待更多数据