LLM 输出限制：Structured Outputs、受限编码和提示词工程

在使用大型语言模型（LLM）时，我们常常面临一个挑战：如何从模型输出中准确提取自己所需的信息。例如，当我们希望模型输出 JSON 格式的数据时，由于模型生成的内容并不总是稳定，可能需要额外编写大量的正则表达式来匹配并提取其中的有效信息。然而，由于 LLM 的能力，导致其输出结构并不永远可靠。

现阶段，让LLM按要求生成特定格式文本的主要方法有几种种：

微调：使模型的输出遵循特定格式
OpenAI Json-mode/Structured Outputs/function-calling: 这些功能允许模型生成更严格、结构化的输出，但受限于openAI平台。
格式约束：在decoding阶段进行约束，限制模型的输出，
Prompt Engineering：最简单的办法，但不稳定。
多阶段prompting：通过多个步骤的提示逐步引导模型生成所需的格式。

本文将聚焦在Structured Outputs，受限编码，和prompt-engineering的角度，探讨它们在生成特定格式文本中的应用和效果。

Json Mode

仅特定模型和平台支持

以openAI 为例，在openai.chat.completions.create 参数中增加response_format={"type":"json_object"} 即可（具体参见：response_format ）。

需要在prompt中要求输出json格式
不能保证完全按要求的格式结构输出
但非100%成功率，存在一些需要额外检测和适当处理的edge case。

Handling edge cases

根据OpenAI官方文档提供的处理方案

https://platform.openai.com/docs/guides/structured-outputs/json-mode

      we_did_not_specify_stop_tokens = True
      
      try:
          response = client.chat.completions.create(
              model="gpt-3.5-turbo-0125",
              messages=[
                  {"role": "system", "content": "You are a helpful assistant designed to output JSON."},
                  {"role": "user", "content": "Who won the world series in 2020? Please respond in the format {winner: ...}"}
              ],
              response_format={"type": "json_object"}
          )
      
          # Check if the conversation was too long for the context window, resulting in incomplete JSON 
          if response.choices[0].message.finish_reason == "length":
              # your code should handle this error case
              pass
      
          # Check if the OpenAI safety system refused the request and generated a refusal instead
          if response.choices[0].message[0].get("refusal"):
              # your code should handle this error case
              # In this case, the .content field will contain the explanation (if any) that the model generated for why it is refusing
              print(response.choices[0].message[0]["refusal"])
      
          # Check if the model's output included restricted content, so the generation of JSON was halted and may be partial
          if response.choices[0].message.finish_reason == "content_filter":
              # your code should handle this error case
              pass
      
          if response.choices[0].message.finish_reason == "stop":
              # In this case the model has either successfully finished generating the JSON object according to your schema, or the model generated one of the tokens you provided as a "stop token"
      
              if we_did_not_specify_stop_tokens:
                  # If you didn't specify any stop tokens, then the generation is complete and the content key will contain the serialized JSON object
                  # This is guaranteed to parse successfully and should now contain  "{"winner": "Los Angeles Dodgers"}"
                  print(response.choices[0].message.content)
              else:
                  # Check if the response.choices[0].message.content ends with one of your stop tokens and handle appropriately
                  pass
      except Exception as e:
          # Your code should handle errors here, for example a network error calling the API
          print(e)

使用pydantic的方案

      from pydantic import BaseModel, EmailStr, ValidationError
      
      # 定义你期望的 JSON 数据模型
      class UserModel(BaseModel):
          name: str
          age: int
          email: EmailStr
      
      # 检查 JSON 是否符合模型的函数
      def validate_json(json_str):
          try:
              # 将输入的 JSON 字符串转换为 UserModel 实例
              user = UserModel.parse_raw(json_str)
              
              # 如果验证通过，返回字典
              return user.dict()
          except ValidationError as ve:
              print(f"JSON validation error: {ve.json()}")
              return None
      
      # 示例用法
      json_str = '{"name": "John Doe", "email": "john.doe@example.com"}'
      validated_json = validate_json(json_str)
      
      if validated_json is not None:
          print("JSON is valid and conforms to the schema:")
          print(validated_json)
      else:
          print("JSON is invalid.")

Json-Mode 更多是对于输出json的格式进行检查(即Json格式的有效性)

Structured Outputs

在模型升级到2024-08-06版本后， OpenAI增加了Structured Outputs功能。

其旨在保证模型始终生成符合所提供的JSON Schema的响应，而无需处理edge case，以及json内容错误。

其优势在于：

无需验证或重试格式不正确的响应
可通过编程检测模型是否安全生成（在API返回结果中增加了"refusal”字段）
无需复杂的prompt用于限制输出结构

相较于Json-Mode，虽然两者都确保生成有效的JSON，但只有Structured Outputs能确保输出json-schema的一致性。因此理论上无需额外的json-schema检查的代码步骤。

如何使用

以官方文档所提供的`Structured data extraction`为例：

from pydantic import BaseModel
from openai import OpenAI

client = OpenAI()

class ResearchPaperExtraction(BaseModel):
    title: str
    authors: list[str]
    abstract: str
    keywords: list[str]

completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "You are an expert at structured data extraction. You will be given unstructured text from a research paper and should convert it into the given structure."},
        {"role": "user", "content": "..."}
    ],
    response_format=ResearchPaperExtraction,
)

research_paper = completion.choices[0].message.parsed

output

   {
     "title": "Application of Quantum Algorithms in Interstellar Navigation: A New Frontier",
     "authors": [
       "Dr. Stella Voyager",
       "Dr. Nova Star",
       "Dr. Lyra Hunter"
     ],
     "abstract": "This paper investigates the utilization of quantum algorithms to improve interstellar navigation systems. By leveraging quantum superposition and entanglement, our proposed navigation system can calculate optimal travel paths through space-time anomalies more efficiently than classical methods. Experimental simulations suggest a significant reduction in travel time and fuel consumption for interstellar missions.",
     "keywords": [
       "Quantum algorithms",
       "interstellar navigation",
       "space-time anomalies",
       "quantum superposition",
       "quantum entanglement",
       "space travel"
     ]
   }

和在LLM输出的结果中通过`pydantic`进行格式检查类似，只是直接将格式要求传给了模型，减少了额外工作。

等价于

...
"response_format": {
      "type": "json_schema",
      "json_schema": {
        "name": "research_paper_extraction",
        "schema": {
          "type": "object",
          "properties": {
            "title": { "type": "string" },
            "authors": {
              "type": "array",
              "items": { "type": "string" }
            },
            "abstract": { "type": "string" },
            "keywords": {
              "type": "array",
              "items": { "type": "string" }
            }
          },
          "required": ["title", "authors", "abstract", "keywords"],
          "additionalProperties": false
        },
        "strict": true
      }
    }

此外，文档中还提供了COT的Structured Outputs例子

Chain of thought

   from pydantic import BaseModel
   from openai import OpenAI
   
   client = OpenAI()
   
   class Step(BaseModel):
       explanation: str
       output: str
   
   class MathReasoning(BaseModel):
       steps: list[Step]
       final_answer: str
   
   completion = client.beta.chat.completions.parse(
       model="gpt-4o-2024-08-06",
       messages=[
           {"role": "system", "content": "You are a helpful math tutor. Guide the user through the solution step by step."},
           {"role": "user", "content": "how can I solve 8x + 7 = -23"}
       ],
       response_format=MathReasoning,
   )
   
   math_reasoning = completion.choices[0].message.parsed

output:

{
"steps": [
  {
    "explanation": "Start with the equation 8x + 7 = -23.",
    "output": "8x + 7 = -23"
  },
  {
    "explanation": "Subtract 7 from both sides to isolate the term with the variable.",
    "output": "8x = -23 - 7"
  },
  {
    "explanation": "Simplify the right side of the equation.",
    "output": "8x = -30"
  },
  {
    "explanation": "Divide both sides by 8 to solve for x.",
    "output": "x = -30 / 8"
  },
  {
    "explanation": "Simplify the fraction.",
    "output": "x = -15 / 4"
  }
],
"final_answer": "x = -15 / 4"
}

相比于之前普通版本的COT：

在测试阶段，程序化检查中间Chain的中间过程变得更加简单
更优美的提取最终结果
限制分析过程中的每一步格式，以提高正确率？

OpenAI API中提供了两种形式的Structured Outputs

Function Calling
使用 json_schema 响应格式

在function calling （tool-using）时，在定义tools时需增加"strict": true 字段，如：

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_delivery_date",
            "description": "Get the delivery date for a customer's order. Call this whenever you need to know the delivery date, for example when a customer asks 'Where is my package'",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "The customer's order ID.",
                    },
                },
                "required": ["order_id"],
                "additionalProperties": False,
            },
        },
        "strict": True,
    }
]

注意："strict": true,会产生一定的延迟（10s以下）

Structured Outputs功能引入了一个新的字段refusal，用于在模型拒绝执行请求时提供反馈。当用户生成的输入时，模型可能会因为安全原因拒绝履行请求，此时API响应会包含一个名为“refusal”的新字段来指示模型拒绝了请求。当API响应中出现refusal字段时，开发者可以程序化地检测到模型生成了拒绝而不是符合模式的输出。

Constrained Decoding

What Is Constrained Decoding

constrained decoding is a technique that manipulates a generative model’s token generation process to constrain its next-token predictions to only tokens that do not violate the required output structure.

仅对严格需要生成的输出部分，从模型的下一个Token概率分布中的兼容Token集进行采样。

简单来说，对于LLM的输出过程，模型会根据之前的词语序列，计算出下一个词可能出现的概率，并根据这个概率生成下一个词。这一过程会不断重复，直到生成完整的句子或段落（或遇到[EOS]标记）。在普通的生成过程中，模型的概率空间是所有可能的Token。

相对而言，在引导生成（Guided Generation）过程中，模型的输出会受到额外的引导或约束，以确保生成的内容更加符合某些特定的要求或目标。这种引导可以通过调整概率分布、加入上下文信息、或施加额外规则来实现。

具体来说，对于给定的输入序列 $\tilde{S}_t$ 和模型参数 $\theta$，

模型首先计算出一个未加约束的概率分布 $\alpha$，表示下一个Token $\tilde{s}_{t+1}$ 的可能性，

$$ \alpha = \text{LM}(\tilde{S}_t, \theta) $$

为了引入约束或限制，构造一个布尔掩码 $m$，这个掩码将限制下一个Token的选择范围。掩码函数 $m(\tilde{S}_t)$将原始分布中的部分不符合约束条件的Token概率置为0，形成一个新的受限分布 $\tilde{\alpha}$：

$$ \tilde{\alpha} = m(\tilde{S}_t) \odot \alpha $$

即在原始分布$\alpha$ 上应用掩码 $m$，只保留那些满足约束条件的Token的概率，其他Token的概率被置为0。

接下来，模型根据受限分布 $\tilde{\alpha}$ 进行采样，选择下一个Token $\tilde{s}_{t+1}$

$$ \tilde{s}_{t+1} \sim \text{Categorical}(\tilde{\alpha}) $$

这个步骤意味着模型只会从符合约束条件的Token集合中选择下一个Token，确保生成的输出遵循预定的规则或格式。

以上内容出自 outlines的paper：Efficient Guided Generation for Large Language Model 具体内容可自行查看。

以常见的Json格式约束为例（Fast JSON Decoding for Local LLMs with Compressed Finite State Machine），

有多种框架可以利用本地模型实现约束解码，如： outlines，SGLang等。

vllm+outlines 实现结构化输出

依赖环境：

! pip install vllm
! pip install openai
! pip install instructor

注意：需要使用到GPU，建议使用colab

启动vllm模型服务，选择适合GPU显存大小的模型：

#! vllm serve Qwen/Qwen1.5-4B-Chat --max_model_len=10000
! vllm serve Qwen/Qwen1.5-1.8B --max_model_len=10000

引导生成json结构：

from openai import OpenAI
# Set OpenAI's API key and API base to use vLLM's API server.
openai_api_key = "EMPTY"
openai_api_base = "http://localhost:8000/v1"

api_client = OpenAI(api_key='YOUR_API_KEY', base_url='http://localhost:8000/v1')
model_name = api_client.models.list().data[0].id
json_schema = {
    "type": "object",
    "properties": {
        "thought": {"type": "string"},
        "answer": {"type": "string"}
    },
    "required": ["thought", "answer"]
}

query = "What is the capital of China?"

# Feed the random question into the existing query
system_prompt = "Respond only with a json object containing the following fields and nothing else: thought, answer."
completion = api_client.chat.completions.create(
    model=model_name,
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": query}
    ],
    extra_body={
        "stop_token_ids": [128009], 
        "response_format": {"type": "json_object"},
        "guided_json": json_schema
    }
)

输出：

print(completion.choices[0].message.content.strip())
# {"thought": "What is the capital of China?", "answer": "Beijing"}

另一个例子，引导生成符合结构的《Harry Potter》角色属性：

定义角色属性结构

from openai import OpenAI
from pydantic import BaseModel, Field
import base64
import instructor
import time
from enum import Enum
from typing import List
from glob import glob

class Property(BaseModel):
    key: str = Field(description="Must be snake case")
    value: str

class House(Enum):
    Griffindor = "gryffindor"
    Hufflepuff = "hufflepuff"
    Ravenclaw = "ravenclaw"
    Slytherin = "slytherin"
    Targaryen = "targaryen"

# This schema is what guides generation
class Character(BaseModel):
    name: str
    age: int
    house: House
    properties: List[Property]

api_client = OpenAI(api_key='YOUR_API_KEY', base_url='http://localhost:8000/v1')
client = instructor.from_openai(api_client, mode=instructor.Mode.MD_JSON)
model_name = api_client.models.list().data[0].id

system_prompt = """\
You are a highly creative assistant tasked with generating fictional characters for a fantasy novel. 
Each character belongs to a specific house, and their properties should reflect their house's traits. 
However, ensure that the characters are entirely original and do not directly reference or resemble any characters, 
events, or settings from the Harry Potter series. Avoid using names, events, or any elements that would make the 
character too similar to existing Harry Potter content. The goal is to create unique and novel characters that 
can stand on their own in a fantasy world.
"""

def run(name):
    try:
        response = client.chat.completions.create(
            model=model_name,
            response_model=Character,
            messages=[
                {
                    'role': 'system',
                    'content': system_prompt
                },
                {
                    'role': 'user',
                    'content': name
                }
            ],
            temperature=0.1,  # Adjust temperature for creativity
        )
        return response
    except Exception as e:
        print(f"An error occurred: {e}")
        return None

 # Example usage
character_name = "Voldemort"
character = run(character_name)
if character:
    print(character)
else:
    print("Failed to generate character.")

# name='Voldemort' age=50 house=<House.Slytherin: 'slytherin'> properties=[Property(key='power', value='dark magic'), Property(key='reputation', value='evil'), Property(key='history', value='murdered his parents and became the Dark Lord')]

Prompt-based Methods

根据奥卡姆剃刀法制，“如无必要，勿增实体”，或许使用prompt也可以实现，避免引入不必要的复杂性。

使用Prompt Engineering 要求LLM输出符合格式要求的json（或者其他）时，可能可以用的一些小Tips：

明确的指令和提示，并提供具体的例子: 往往LLM的few-shot learning 的能力还行，模仿个简单的json格式基本不在话下。

使用占位符： 为具体的json元素标注数据类型和格式，便于LLM理解。这一步骤类似于给json定义一个schema。如

{
  "user": "示例用户",
  "id": 12345,
  "items": [
    {"item_name": "示例物品1", "quantity": 2},
    {"item_name": "示例物品2", "quantity": 1}
  ]
}

而不是一句“将结果输出成json格式”

调整温度： 采用较低的温度设置
分步骤生成： 对于复杂的JSON结构，可以要求模型逐步生成每个部分，然后将它们组合起来；或先生成JSON的高层结构，然后再深入生成每个子部分。
让LLM自己检查：类似“三段式翻译” 之类的prompt， LLM首先生成第一版json，而后LLM对LLM进行检查，并提出修改建议，最后基于建议和初版输出生成最终版结果。
格式有效性检查：使用如Pydantic等工具进行验证和解析，如果模型生成了不正确的JSON，可以通过循环过程直到输出有效的JSON为止；或使用一些第三方库修复json, 如json_repair。

这里推荐一个开源项目， TypeChat，一个由微软开发，用于让LLM输出类型定义的结果。TypeChat 自动处理生成提示、验证 LLM 输出是否符合定义的类型，并在必要时修复不符合的输出。

其核心是通过prompt的方式，要求LLM按特定要求输出。

以官方提供的Calendar例子为例：

其实现功能即，将自然语言转换成json格式，用于后续的代办事项的工具使用。

如输入：

📅> I need to get my tires changed from 12:00 to 2:00 pm on Friday March 15, 2024

则会输出：

{
  "actions": [
    {
      "actionType": "add event",
      "event": {
        "day": "Friday March 15, 2024",
        "timeRange": {
          "startTime": "12:00 pm",
          "endTime": "2:00 pm"
        },
        "description": "get my tires changed"
      }
    }
  ]
}

首先需要事先定义具体且复杂的schema，如 schema.py

from typing_extensions import Literal, NotRequired, TypedDict, Annotated, Doc

class UnknownAction(TypedDict):
    """
    if the user types text that can not easily be understood as a calendar action, this action is used
    """

    actionType: Literal["Unknown"]
    text: Annotated[str, Doc("text typed by the user that the system did not understand")]


class EventTimeRange(TypedDict, total=False):
    startTime: str
    endTime: str
    duration: str


class Event(TypedDict):
    day: Annotated[str, Doc("date (example: March 22, 2024) or relative date (example: after EventReference)")]
    timeRange: EventTimeRange
    description: str
    location: NotRequired[str]
    participants: NotRequired[Annotated[list[str], Doc("a list of people or named groups like 'team'")]]


class EventReference(TypedDict, total=False):
    """
    properties used by the requester in referring to an event
    these properties are only specified if given directly by the requester
    """

    day: Annotated[str, Doc("date (example: March 22, 2024) or relative date (example: after EventReference)")]
    dayRange: Annotated[str, Doc("(examples: this month, this week, in the next two days)")]
    timeRange: EventTimeRange
    description: str
    location: str
    participants: list[str]


class FindEventsAction(TypedDict):
    actionType: Literal["find events"]
    eventReference: Annotated[EventReference, Doc("one or more event properties to use to search for matching events")]


class ChangeDescriptionAction(TypedDict):
    actionType: Literal["change description"]
    eventReference: NotRequired[Annotated[EventReference, Doc("event to be changed")]]
    description: Annotated[str, Doc("new description for the event")]


class ChangeTimeRangeAction(TypedDict):
    actionType: Literal["change time range"]
    eventReference: NotRequired[Annotated[EventReference, Doc("event to be changed")]]
    timeRange: Annotated[EventTimeRange, Doc("new time range for the event")]


class AddParticipantsAction(TypedDict):
    actionType: Literal["add participants"]
    eventReference: NotRequired[
        Annotated[EventReference, Doc("event to be augmented; if not specified assume last event discussed")]
    ]
    participants: NotRequired[Annotated[list[str], "new participants (one or more)"]]


class RemoveEventAction(TypedDict):
    actionType: Literal["remove event"]
    eventReference: EventReference


class AddEventAction(TypedDict):
    actionType: Literal["add event"]
    event: Event


Actions = (
    AddEventAction
    | RemoveEventAction
    | AddParticipantsAction
    | ChangeTimeRangeAction
    | ChangeDescriptionAction
    | FindEventsAction
    | UnknownAction
)


class CalendarActions(TypedDict):
    actions: list[Actions]

这些类表示各种日历操作，每个操作类都有一个actionType字段，确定操作类型：

• FindEventsAction：查找事件，包含一个eventReference字段，用于指定要查找的事件属性。

• ChangeDescriptionAction：更改事件描述，包含eventReference（可选）和新的description字段。

• ChangeTimeRangeAction：更改事件的时间范围，包含eventReference（可选）和新的timeRange字段。

• AddParticipantsAction：为事件添加参与者，包含eventReference（可选）和参与者列表participants（可选）。

• RemoveEventAction：删除事件，包含一个eventReference字段。

• AddEventAction：添加新事件，包含一个event字段。

在TypeChat的核心 prompt如下：

prompt = f"""
You are a service that translates user requests into JSON objects of type "{self._type_name}" according to the following TypeScript definitions:
```
{self._schema_str}
```
The following is a user request:
'''
{intent}
'''
The following is the user request translated into a JSON object with 2 spaces of indentation and no properties with the value undefined:
"""

以上面的例子，最终生成的prompt：

"""
You are a service that translates user requests into JSON objects of type "CalendarActions" according to the following TypeScript definitions:
```
interface CalendarActions {
    actions: Array<
        | AddEventAction
        | RemoveEventAction
        | AddParticipantsAction
        | ChangeTimeRangeAction
        | ChangeDescriptionAction
        | FindEventsAction
        | UnknownAction
    >;
}

// if the user types text that can not easily be understood as a calendar action, this action is used
interface UnknownAction {
    actionType: "Unknown";
    // text typed by the user that the system did not understand
    text: string;
}

interface FindEventsAction {
    actionType: "find events";
    // one or more event properties to use to search for matching events
    eventReference: EventReference;
}

// properties used by the requester in referring to an event
// these properties are only specified if given directly by the requester
interface EventReference {
    // date (example: March 22, 2024) or relative date (example: after EventReference)
    day?: string;
    // (examples: this month, this week, in the next two days)
    dayRange?: string;
    timeRange?: EventTimeRange;
    description?: string;
    location?: string;
    participants?: string[];
}

interface EventTimeRange {
    startTime?: string;
    endTime?: string;
    duration?: string;
}

interface ChangeDescriptionAction {
    actionType: "change description";
    // event to be changed
    eventReference?: EventReference;
    // new description for the event
    description: string;
}

interface ChangeTimeRangeAction {
    actionType: "change time range";
    // event to be changed
    eventReference?: EventReference;
    // new time range for the event
    timeRange: EventTimeRange;
}

interface AddParticipantsAction {
    actionType: "add participants";
    // event to be augmented; if not specified assume last event discussed
    eventReference?: EventReference;
    // new participants (one or more)
    participants?: string[];
}

interface RemoveEventAction {
    actionType: "remove event";
    eventReference: EventReference;
}

interface AddEventAction {
    actionType: "add event";
    event: Event;
}

interface Event {
    // date (example: March 22, 2024) or relative date (example: after EventReference)
    day: string;
    timeRange: EventTimeRange;
    description: string;
    location?: string;
    // a list of people or named groups like 'team'
    participants?: string[];
}
```

The following is a user request:

I need to get my tires changed from 12:00 to 2:00 pm on Friday March 15, 2024

The following is the user request translated into a JSON object with 2 spaces of indentation and no properties with the value undefined:

"""