LLM结构化输出指南

STRUCTURED_PROMPT = """
请按以下JSON格式输出，不要包含其他内容：

{
  "intent": "用户意图分类",
  "confidence": 0.0-1.0之间的数值,
  "entities": [
    {"type": "实体类型", "value": "实体值"}
  ],
  "response": "回复内容"
}

用户输入：{input}
"""

FEW_SHOT_PROMPT = """
请按JSON格式输出意图分析结果。

示例1：
输入：我想订一张明天去北京的机票
输出：{"intent": "book_flight", "confidence": 0.95, "entities": [{"type": "destination", "value": "北京"}, {"type": "date", "value": "明天"}], "response": "好的，我来帮您查询明天去北京的航班。"}

示例2：
输入：今天天气怎么样
输出：{"intent": "weather_query", "confidence": 0.9, "entities": [{"type": "date", "value": "今天"}], "response": "让我为您查询今天的天气。"}

现在请分析：
输入：{input}
输出：
"""

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="qwen3-32b",
    messages=[{"role": "user", "content": "分析以下文本的情感，返回JSON"}],
    response_format={"type": "json_object"}  # 强制JSON输出
)

result = json.loads(response.choices[0].message.content)

from pydantic import BaseModel
from typing import List

class SentimentAnalysis(BaseModel):
    sentiment: str
    confidence: float
    key_phrases: List[str]
    summary: str

response = client.beta.chat.completions.parse(
    model="qwen3-32b",
    messages=[{"role": "user", "content": "分析这段文本的情感"}],
    response_format=SentimentAnalysis  # 直接传入Pydantic模型
)

analysis = response.choices[0].message.parsed  # 已解析的Pydantic对象
print(analysis.sentiment)
print(analysis.confidence)

# vLLM支持的Guided Decoding
from vllm import LLM, SamplingParams

llm = LLM(model="qwen3-32b")

# 通过JSON Schema约束输出
json_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer", "minimum": 0, "maximum": 150},
        "skills": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["name", "age"]
}

sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=512,
    guided_decoding={"json": json_schema}  # 约束解码
)

outputs = llm.generate(["提取人物信息：张三，25岁，会Python和Java"], sampling_params)
result = json.loads(outputs[0].outputs[0].text)

class StructuredOutputRunner:
    def __init__(self, llm, schema, max_retries=3):
        self.llm = llm
        self.schema = schema
        self.max_retries = max_retries
    
    async def generate(self, prompt):
        for attempt in range(self.max_retries):
            try:
                response = await self.llm.generate(prompt)
                data = json.loads(response)
                
                # 验证schema
                validated = self.schema(**data)
                return validated
            
            except json.JSONDecodeError:
                # JSON解析失败，添加错误提示重试
                prompt = f"{prompt}\n\n上一次输出不是合法JSON，请只输出JSON。"
            except Exception as e:
                # Schema验证失败
                prompt = f"{prompt}\n\n错误：{e}。请修正输出格式。"
        
        raise ValueError(f"Failed to generate valid output after {self.max_retries} attempts")

def partial_json_parse(text):
    """处理不完整的JSON输出"""
    # 尝试补全不完整的JSON
    text = text.strip()
    
    # 统计括号
    open_braces = text.count('{') - text.count('}')
    open_brackets = text.count('[') - text.count(']')
    
    # 补全
    text += ']' * max(open_brackets, 0)
    text += '}' * max(open_braces, 0)
    
    try:
        return json.loads(text)
    except:
        return None