48. Structured Output 实现：vLLM如何确保生成内容的结构化与可靠性

class JSONSchemaParser:
    def __init__(self):
        self.schema = None
        self.validators = {}
    
    def parse(self, schema: dict) -> "JSONSchemaParser":
        """解析JSON Schema"""
        self.schema = schema
        self._parse_schema(schema)
        return self
    
    def _parse_schema(self, schema: dict):
        """递归解析Schema"""
        if "type" not in schema:
            raise ValueError("Schema must have a 'type' field")
        
        schema_type = schema["type"]
        
        if schema_type == "object":
            self._parse_object_schema(schema)
        elif schema_type == "array":
            self._parse_array_schema(schema)
        elif schema_type in ["string", "number", "integer", "boolean", "null"]:
            self._parse_primitive_schema(schema)
        elif schema_type == "anyOf" or schema_type == "oneOf":
            self._parse_anyof_schema(schema)
        elif schema_type == "allOf":
            self._parse_allof_schema(schema)
    
    def _parse_object_schema(self, schema: dict):
        """解析对象类型Schema"""
        # 处理必填字段
        required = schema.get("required", [])
        
        # 处理属性
        properties = schema.get("properties", {})
        for prop_name, prop_schema in properties.items():
            self.validators[prop_name] = self._create_validator(prop_schema)
    
    def _parse_array_schema(self, schema: dict):
        """解析数组类型Schema"""
        # 处理数组项
        items = schema.get("items", {})
        self.validators["items"] = self._create_validator(items)
        
        # 处理数组长度限制
        self.validators["minItems"] = schema.get("minItems", 0)
        self.validators["maxItems"] = schema.get("maxItems", None)
    
    def _parse_primitive_schema(self, schema: dict):
        """解析基本类型Schema"""
        # 处理类型特定的验证规则
        schema_type = schema["type"]
        if schema_type == "string":
            self.validators["pattern"] = schema.get("pattern", None)
            self.validators["minLength"] = schema.get("minLength", 0)
            self.validators["maxLength"] = schema.get("maxLength", None)
            self.validators["format"] = schema.get("format", None)
        elif schema_type in ["number", "integer"]:
            self.validators["minimum"] = schema.get("minimum", None)
            self.validators["maximum"] = schema.get("maximum", None)
            self.validators["exclusiveMinimum"] = schema.get("exclusiveMinimum", None)
            self.validators["exclusiveMaximum"] = schema.get("exclusiveMaximum", None)
            self.validators["multipleOf"] = schema.get("multipleOf", None)
    
    def _create_validator(self, schema: dict):
        """创建验证器"""
        # 递归创建验证器
        parser = JSONSchemaParser()
        return parser.parse(schema)

class FSMCompiler:
    def __init__(self):
        self.states = set()
        self.initial_state = None
        self.final_states = set()
        self.transitions = {}
    
    def compile(self, spec: dict) -> "FSM":
        """编译格式规范为有限状态机"""
        if "type" not in spec:
            raise ValueError("Spec must have a 'type' field")
        
        spec_type = spec["type"]
        
        if spec_type == "json":
            return self._compile_json_spec(spec)
        elif spec_type == "regex":
            return self._compile_regex_spec(spec)
        elif spec_type == "xml":
            return self._compile_xml_spec(spec)
        else:
            raise ValueError(f"Unsupported spec type: {spec_type}")
    
    def _compile_json_spec(self, spec: dict) -> "FSM":
        """编译JSON格式规范为有限状态机"""
        # 解析JSON Schema
        schema_parser = JSONSchemaParser()
        schema = schema_parser.parse(spec["schema"])
        
        # 构建有限状态机
        fsm = FSM()
        
        # 添加初始状态
        initial_state = fsm.add_state("initial")
        fsm.set_initial_state(initial_state)
        
        # 添加最终状态
        final_state = fsm.add_state("final")
        fsm.add_final_state(final_state)
        
        # 构建状态转移
        self._build_json_fsm_states(fsm, initial_state, final_state, schema)
        
        return fsm
    
    def _build_json_fsm_states(self, fsm: "FSM", current_state: "State", final_state: "State", schema: "JSONSchemaParser"):
        """构建JSON FSM状态转移"""
        # 根据Schema类型构建不同的状态转移
        # 这里简化实现，仅展示核心逻辑
        pass
    
    def _compile_regex_spec(self, spec: dict) -> "FSM":
        """编译正则表达式规范为有限状态机"""
        # 将正则表达式转换为NFA，再转换为DFA
        regex = spec["pattern"]
        nfa = self._regex_to_nfa(regex)
        dfa = self._nfa_to_dfa(nfa)
        min_dfa = self._minimize_dfa(dfa)
        return min_dfa
    
    def _regex_to_nfa(self, regex: str) -> "FSM":
        """将正则表达式转换为NFA"""
        # 实现正则表达式到NFA的转换
        pass
    
    def _nfa_to_dfa(self, nfa: "FSM") -> "FSM":
        """将NFA转换为DFA"""
        # 实现子集构造算法，将NFA转换为DFA
        pass
    
    def _minimize_dfa(self, dfa: "FSM") -> "FSM":
        """最小化DFA"""
        # 实现DFA最小化算法
        pass

class ConstrainedDecoder:
    def __init__(self, fsm: "FSM"):
        self.fsm = fsm
        self.current_states = {fsm.initial_state}
    
    def get_allowed_tokens(self, logits: torch.Tensor) -> torch.Tensor:
        """根据当前状态获取允许的token"""
        # 获取当前所有可能的下一个token
        allowed_chars = set()
        for state in self.current_states:
            transitions = self.fsm.get_transitions(state)
            for char, next_state in transitions.items():
                allowed_chars.add(char)
        
        # 将字符转换为token
        allowed_tokens = []
        for token_id in range(logits.shape[-1]):
            token = tokenizer.decode([token_id])
            if token in allowed_chars:
                allowed_tokens.append(token_id)
        
        # 创建掩码，只保留允许的token
        mask = torch.zeros_like(logits)
        mask[:, allowed_tokens] = 1
        
        return mask
    
    def update_state(self, token: str):
        """根据生成的token更新FSM状态"""
        new_states = set()
        for state in self.current_states:
            transitions = self.fsm.get_transitions(state)
            if token in transitions:
                new_states.add(transitions[token])
        
        # 处理ε转换
        self._handle_epsilon_transitions(new_states)
        
        self.current_states = new_states
    
    def _handle_epsilon_transitions(self, states: set):
        """处理ε转换"""
        # 实现ε转换的处理
        pass
    
    def is_complete(self) -> bool:
        """检查当前状态是否为最终状态"""
        return any(state in self.fsm.final_states for state in self.current_states)
    
    def reset(self):
        """重置解码器状态"""
        self.current_states = {self.fsm.initial_state}

class StructuredOutputValidator:
    def __init__(self, spec: dict):
        self.spec = spec
        self.spec_type = spec["type"]
        
        # 根据规范类型创建验证器
        if self.spec_type == "json":
            self.validator = self._create_json_validator(spec)
        elif self.spec_type == "regex":
            self.validator = self._create_regex_validator(spec)
        elif self.spec_type == "xml":
            self.validator = self._create_xml_validator(spec)
        else:
            raise ValueError(f"Unsupported spec type: {self.spec_type}")
    
    def _create_json_validator(self, spec: dict) -> callable:
        """创建JSON验证器"""
        import jsonschema
        schema = spec["schema"]
        
        def validate_json(content: str) -> bool:
            try:
                # 解析JSON
                import json
                data = json.loads(content)
                
                # 验证JSON Schema
                jsonschema.validate(instance=data, schema=schema)
                return True
            except (json.JSONDecodeError, jsonschema.ValidationError):
                return False
        
        return validate_json
    
    def _create_regex_validator(self, spec: dict) -> callable:
        """创建正则表达式验证器"""
        import re
        pattern = spec["pattern"]
        regex = re.compile(pattern)
        
        def validate_regex(content: str) -> bool:
            return bool(regex.fullmatch(content))
        
        return validate_regex
    
    def _create_xml_validator(self, spec: dict) -> callable:
        """创建XML验证器"""
        from lxml import etree
        xsd_schema = spec["schema"]
        schema = etree.XMLSchema(etree.fromstring(xsd_schema.encode()))
        
        def validate_xml(content: str) -> bool:
            try:
                xml_doc = etree.fromstring(content.encode())
                return schema.validate(xml_doc)
            except etree.XMLSyntaxError:
                return False
        
        return validate_xml
    
    def validate(self, content: str) -> bool:
        """验证内容是否符合格式规范"""
        return self.validator(content)
    
    def get_validation_error(self, content: str) -> str:
        """获取验证错误信息"""
        try:
            if self.spec_type == "json":
                import json
                import jsonschema
                data = json.loads(content)
                jsonschema.validate(instance=data, schema=self.spec["schema"])
                return ""
            elif self.spec_type == "regex":
                import re
                pattern = self.spec["pattern"]
                regex = re.compile(pattern)
                if regex.fullmatch(content):
                    return ""
                else:
                    return f"Content does not match pattern: {pattern}"
            elif self.spec_type == "xml":
                from lxml import etree
                xml_doc = etree.fromstring(content.encode())
                schema = etree.XMLSchema(etree.fromstring(self.spec["schema"].encode()))
                if schema.validate(xml_doc):
                    return ""
                else:
                    return schema.error_log.last_error
            else:
                return f"Unsupported spec type: {self.spec_type}"
        except Exception as e:
            return str(e)

from vllm import LLM, SamplingParams

# 定义JSON Schema
json_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0, "maximum": 120},
        "email": {"type": "string", "format": "email"},
        "address": {
            "type": "object",
            "properties": {
                "street": {"type": "string"},
                "city": {"type": "string"},
                "country": {"type": "string"}
            },
            "required": ["street", "city", "country"]
        },
        "hobbies": {
            "type": "array",
            "items": {"type": "string"},
            "minItems": 1,
            "maxItems": 5
        }
    },
    "required": ["name", "age", "email", "address"]
}

# 创建LLM实例
llm = LLM(model="meta-llama/Llama-2-7b-chat-hf")

# 创建采样参数，启用结构化输出
sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=200,
    structured_output={
        "type": "json",
        "schema": json_schema
    }
)

# 生成结构化输出
prompt = "Generate a user profile with name, age, email, address, and hobbies"
outputs = llm.generate([prompt], sampling_params)

# 输出结果
for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"Prompt: {prompt}")
    print(f"Generated: {generated_text}")

from vllm import LLM, SamplingParams

# 创建LLM实例
llm = LLM(model="meta-llama/Llama-2-7b-chat-hf")

# 创建采样参数，启用结构化输出（正则表达式）
sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=100,
    structured_output={
        "type": "regex",
        "pattern": r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$"
    }
)

# 生成结构化输出
prompt = "Generate a timestamp in the format YYYY-MM-DD HH:MM:SS"
outputs = llm.generate([prompt], sampling_params)

# 输出结果
for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"Prompt: {prompt}")
    print(f"Generated: {generated_text}")

# 1. 定义新的格式规范类型
class CustomFormatSpec:
    def __init__(self, spec: dict):
        self.spec = spec
    
    def validate(self, content: str) -> bool:
        # 实现自定义验证逻辑
        pass

# 2. 注册新的格式类型
from vllm.structured_output import FORMAT_REGISTRY

@FORMAT_REGISTRY.register("custom")
def create_custom_validator(spec: dict):
    return CustomFormatSpec(spec)

# 3. 使用新的格式类型
sampling_params = SamplingParams(
    structured_output={
        "type": "custom",
        "spec": {"key": "value"}
    }
)

# 1. 实现自定义验证器
class CustomValidator:
    def __init__(self, config: dict):
        self.config = config
    
    def validate(self, content: str) -> bool:
        # 实现自定义验证逻辑
        return True
    
    def get_allowed_tokens(self, logits: torch.Tensor) -> torch.Tensor:
        # 实现自定义token过滤逻辑
        return logits

# 2. 注册自定义验证器
from vllm.structured_output import VALIDATOR_REGISTRY

@VALIDATOR_REGISTRY.register("custom")
def create_custom_validator(spec: dict):
    return CustomValidator(spec)

# 安装vLLM
pip install vllm

# 安装其他依赖
pip install jsonschema lxml

# 启动vLLM服务，启用结构化输出功能
python -m vllm.entrypoints.api_server \
    --model meta-llama/Llama-2-7b-chat-hf \
    --port 8000 \
    --num-gpus 1

# 使用curl测试结构化输出API
curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "meta-llama/Llama-2-7b-chat-hf", "messages": [{"role": "user", "content": "Generate a user profile with name, age, email, and address"}], "temperature": 0.7, "max_tokens": 200, "structured_output": {"type": "json", "schema": {"type": "object", "properties": {"name": {"type": "string"}, "age": {"type": "integer"}, "email": {"type": "string"}, "address": {"type": "string"}}, "required": ["name", "age", "email", "address"]}}}'

from vllm import LLM, SamplingParams

# 创建LLM实例
llm = LLM(model="meta-llama/Llama-2-7b-chat-hf")

# 创建采样参数，启用结构化输出
sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=200,
    structured_output={
        "type": "json",
        "schema": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
                "email": {"type": "string"},
                "address": {"type": "string"}
            },
            "required": ["name", "age", "email", "address"]
        }
    }
)

# 生成结构化输出
prompt = "Generate a user profile with name, age, email, and address"
outputs = llm.generate([prompt], sampling_params)

# 输出结果
for output in outputs:
    print(f"Generated: {output.outputs[0].text}")