|
|
最近闭源大模型们都陆续支持结构化输出，这一章我们先结合demo看下开源和闭源对结构化输出的支持，随后会介绍Constrained Decoding和Format Restricting Instructions 两种结构化输出约束方案，最后会给出结构化输出对比自然语言输出的一些观点。
代码示例
闭源 - OpenAI
闭源三巨头都是支持结构化输出的，上面链了OpenAI和Gemini关于结构化输出的相关API文档。这里我们就以OpenAI为例，聊下结构化输出。
这里并非指OpenAI很早就支持的Json Mode，而JSON Mode的升级版Structure Output，只对gpt-4o-mini-2024-07-18和gpt-4o-2024-08-06之后的模型版本支持。简单说原来的JSON Mode只保证模型输出一个合法可以解析的json而已，对json的字段，字段类型，取值不做任何约束，而Strucutre Ouput则会进一步对JSON里面的具体字段和类型进行约束。这里我们举个例子，从基金季报中抽取基金经理对市场不同行业的观点，对观点进行情绪分类，并关联相关的申万一级行业。（哈哈并不是说这是最优的解决方案只是想把抽取，分类，生成任务融在一个case里面）
首先我们先定义抽取任务的结构体，申万一级行业的枚举值和情绪的枚举值，这里结构化输出都是使用pydantic定义的。通过枚举值定义我们可以约束模型输出的取值范围，而通过抽取结构定义我们可以约束模型输出的结构。不过这里对Enum的取值数量有限制一次输出的枚举值总量不能超过500，毕竟是直接作为模型上文，枚举值太多一是慢二是贵，三是不稳定。
from enum import Enumfrom typing import Listfrom pydantic import BaseModel, Fieldclass SWIndustry(Enum): AGRICULTURE = "农林牧渔" MINING = "采掘" CHEMICALS = "化工" STEEL = "钢铁" NONFERROUS_METALS = "有色金属" BUILDING_MATERIALS = "建筑材料" ELECTRICAL_EQUIPMENT = "电气设备" APPLIANCES = "家用电器" FOOD_BEVERAGE = "食品饮料" TEXTILE_APPAREL = "纺织服装" LIGHT_MANUFACTURING = "轻工制造" PHARMACEUTICALS = "医药生物" PUBLIC_UTILITIES = "公用事业" TRANSPORTATION = "交通运输" REAL_ESTATE = "房地产" COMMERCE_TRADE = "商业贸易" COMPUTER = "计算机" MEDIA = "传媒" COMMUNICATION = "通信" BANKING = "银行" NON_BANK_FINANCIAL = "非银金融" AUTOMOBILE = "汽车" MACHINERY = "机械设备" DEFENSE_MILITARY = "国防军工" BUILDING_CONSTRUCTION = "建筑装饰" ELECTRONICS = "电子" COMPREHENSIVE = "综合" LEISURE_SERVICES = "休闲服务" COMPUTER_APPLICATIONS = "计算机应用" CHEMICAL_FIBERS = "化纤" METAL_PRODUCTS = "金属制品"class ViewAspect(Enum): POSITIVE = '正面' NETURAL = '中性' NEGATIVE = '负面' class View(BaseModel): extract_view: str = Field(description="抽取文档中中对某个金融行业、或行业相关的主题或概念表达观点的句子") extract_view_entities: List[str] = Field(description="抽取观点金融主体,该主体必须出现在观点句子中，可以是金融行业，或行业相关概念或主题") related_industry: list[SWIndustry] = Field(description=f"观点金融主体最相关的1个或多个申万一级行业") view_aspect: ViewAspect = Field(description=f'对观点情绪进行精准分类，模糊情绪均为中性')class ViewExtraction(BaseModel): views: list[View] = Field( ..., description="每个观点都应该是一个单独的对象，包含原文中表达观点的句子，观点主体，观点情绪分类和关联的申万一级行业", )然后只需要把以上的结构体作为response_format的参数输入openai即可
from openai import AzureOpenAIclient = AzureOpenAI( api_key = '...', api_version="2024-08-01-preview", azure_endpoint= "...")completion = client.beta.chat.completions.parse( model='gpt-4o', messages=[ { "role": "system", "content": "你是一个完美的金融观点解析系统，可以从文档中抽取观点，和观点对应主体并对观点进行分类。请从以下文档中抽取一系列观点", }, { "role": "user", "content": content, }, ], response_format= ViewExtraction)这样我们就能得到结构化的输出如下
再举一个function calling的例子，假设我们有两个工具一个Bing搜索，一个是基金信息查询工具，模型需要根据用户提问选择一个或多个工具来解决问题。
from typing import Literal,Union,Optionalclass BingSearch(BaseModel): query: str = Field(description="网页搜索query") class FundInfo(BaseModel): """ 可以通过基金代码或基金名称，查询基金基础信息 """ fund_code_or_name: Optional[str] = Field(description="提问提及的基金代码或名称，没有则为空") lookup_field: Literal["fund_manager", "unit_value","contract_date","manage_fee","net_value"]class Task(BaseModel): name: str = Field(description="任务名称") tool: Union[BingSearch, FundInfo] = Field(description="完成任务所需调用的工具") class TaskSequence(BaseModel): reason: str = Field(description="先逐步思考要解决用户的问题需要哪些步骤") task_actions: List[Task] = Field(description="任务列表，按执行顺序依次排列") completion = client.beta.chat.completions.parse( model='gpt-4o', messages=[ { "role": "system", "content": "你是一个金融工具助手，可以完美根据用户提问选择需要调用的工具列表", }, { "role": "user", "content": '天弘中证500当前的管理费是多少，是否随着基金规模的增加而增加', }, ], response_format= TaskSequence)然后我们就能得到下面的结构化输出啦~
开源实现 - Instructor
开源也有一些方案是针对结构化输出的，例如Instructor, Outlines。简单对比的话如果你用API调模型那Instructor更合适，如果你自己部署模型调用那Outlines更合适，vllm这些推理框架最新的版本也已经融入了Outlines。这里我们就选Instructor进行介绍。还是上面的例子，输出格式的定义相同，针对不满足openai版本条件的老模型，我们可以使用instructor来实现结构化输出。
from openai import AzureOpenAIimport instructorclient = AzureOpenAI( api_key = '...', api_version="2024-08-01-preview", azure_endpoint= "...")client = instructor.from_openai(client)resp = client.chat.completions.create( model='thgpt4o', response_model= ViewExtraction, messages=[ { "role": "system", "content": "你是一个完美的金融观点解析系统，可以从文档中抽取观点，和观点对应主体并对观点进行分类。请从以下文档中抽取一系列观点", }, { "role": "user", "content": content, }, ],)
那instructor，openai这些结构化输出能力都是如何实现的呢？下面我们来看几种约束模型给出结构化输出的方案
实现原理
这里提供两种不同的实现方案，一种是基于条件解码的强约束方案，和基于指令的弱约束方案，并且会给出不同方案对模型推理效果的影响。
Constrained Decoding
开源项目Outlines的两位作者Brandon T. Willard和R´emi Louf是比较早提出大模型可控生成方案的大佬。
条件解码方案其实就是在每一步解码时都对输出词表进行MASK（Regular Expression Guided Masking），只允许模型对当前位置符合输出格式的Token进行预测，把原始基于完整词表的softmax，转换成对于局部掩码词表的softmax。
那问题其实就简化成了在每一步推理时，如何选择该进行掩码的Token呢？毕竟GPT预测是自左向右，无法获得完整Token序列。论文把基于输出格式掩码的问题，转换成了基于有限状态机的状态转移问题（FSM）。简单解释下FSM其实就是由一组状态和状态之间的转移过程组成，词表中的字符满足条件的可以匹配到FSM的某个或某几个状态，从而在碰到字符A后，就可以确认几种满足条件的状态转移路径，从而根据后面的路径确认掩码词表。
因为词表中的每个字符究竟满足哪些状态，每个状态后有哪些可能的转移状态这些都是预先计算好的，因此并不需要在推理中动态计算，相反可以预先构建好每个词表到状态，再到后续转移状态的mapping。在解码过程中只需要根据解码字符读取mapping，对下一个字符进行对应的掩码即可。因此算法的时间复杂度是O（1），空间复杂度是O（状态数）。
这里我们还是举论文中的例子。我们的输出要求是满足浮点数“([0-9])?.?[0-9]”。这个输出约束可以被转换成FSM中的4种不同状态，每个状态有不同的转移状态（哈哈下面的例子是DeepSeek给大家举的）
- 状态 0: 初始状态，可以进入状态1和2
- 状态 1: 匹配数字 [0-9]，可以继续在状态1或者去状态2
- 状态 2: 匹配小数点 [.]，只能进入状态3
- 状态 3: 匹配小数点后的数字 [0-9]，可以继续在状态3
假设我们的词表只有5个字符{"A", ".", "42", ".2", "1"}，那整个FSM掩码过程如下（以下词表选择过程是读取预先构建好的index）
- 步骤 1：初始化 FSM。我们从状态 0 开始。
- 步骤 2：查找当前状态下允许的词汇。在状态 0，根据 FSM，我们可以匹配数字 [0-9] 或小数点 [.]。因此，我们允许的词汇是：{".", "42", ".2", "1"}。
- 步骤 3：选择一个词汇并更新 FSM 状态。假设我们选择了 ".2"。选择 ".2" 后，FSM 从状态 0 进入状态 2（因为匹配了小数点 [.]）。然后，FSM 继续匹配 "2"，进入状态 3。
- 步骤 4：继续生成下一个词汇。在状态 3，我们只能匹配数字 [0-9]。因此，允许的词汇是：{"42", "1"}。假设我们选择了 "1"。选择 "1" 后，FSM 保持在状态 3。
- 步骤 5：生成结束。如果我们选择了一个表示结束的特殊词汇（如 EOS），生成过程结束。
基于已经构建好的FSM进行解码的步骤在Outlines里面如下（./generator/generatae.py）（哈哈下面的代码是cursor帮忙直接定位到的）
def sequence_generator(model, sampler, fsms, token_ids, sequence_weights, attention_masks, fsm_states, rng): while True: # 1. 获取模型输出的logits logits, kv_cache = model(token_ids, attention_masks, kv_cache) # 2. 获取FSM允许的下一个token allowed_tokens = get_allowed_tokens(fsms, fsm_states) # 3. 基于allowed_tokens对logits进行mask,不允许的token均为-inf biased_logits = bias_logits(logits, allowed_tokens) # 4. 采样下一个token next_token_ids, ancestors, sequence_weights = sampler(biased_logits, sequence_weights, rng) # 5. 更新FSM状态 fsm_states = get_next_fsm_states(fsms, fsm_states, next_token_ids) # 6. 检查是否生成完成 is_finished = is_generation_finished(fsms, fsm_states)Format Restricting Instructions
FRI是更简单的实现方案，也就是在指令中加入对应输出的约束。这里还是拿Instructor来举例子吧，虽然这并不准确，因为Instructor调用的API接口背后还是做了Constrained Decoding的逻辑，Instructor其实只是从中做了一层Adapter。但是不妨碍我们通过instructor的实现来看下如何把pydantic的定义转换成结构化输出的指令约束。
在上面使用instructor.from_openai(client)时，Instructor会打猴子补丁，在常规openai的接口上，增加response_model的预处理，和对输出的retry机制（patch.py)
@overloaddef from_openai( client: openai.OpenAI, mode: instructor.Mode = instructor.Mode.TOOLS, **kwargs: Any,) -> Instructor: pass @overloaddef patch( client: OpenAI, mode: Mode = Mode.TOOLS,) -> OpenAI: ...def patch( # type: ignore client: OpenAI | AsyncOpenAI | None = None, create: Callable[T_ParamSpec, T_Retval] | None = None, mode: Mode = Mode.TOOLS,) -> OpenAI | AsyncOpenAI: """ Patch the `client.chat.completions.create` method Enables the following features: - `response_model` parameter to parse the response from OpenAI's API - `max_retries` parameter to retry the function if the response is not valid - `validation_context` parameter to validate the response using the pydantic model - `strict` parameter to use strict json parsing - `hooks` parameter to hook into the completion process """ logger.debug(f"Patching `client.chat.completions.create` with {mode=}") if create is not None: func = create elif client is not None: func = client.chat.completions.create else: raise ValueError("Either client or create must be provided") @wraps(func) # type: ignore def new_create_sync( response_model: type[T_Model] | None = None, validation_context: dict[str, Any] | None = None, context: dict[str, Any] | None = None, max_retries: int | Retrying = 1, strict: bool = True, hooks: Hooks | None = None, *args: T_ParamSpec.args, **kwargs: T_ParamSpec.kwargs, ) -> T_Model: context = handle_context(context, validation_context) response_model, new_kwargs = handle_response_model( response_model=response_model, mode=mode, **kwargs ) new_kwargs = handle_templating(new_kwargs, context) response = retry_sync( func=func, # type: ignore response_model=response_model, context=context, max_retries=max_retries, args=args, hooks=hooks, strict=strict, kwargs=new_kwargs, mode=mode, ) return response # type: ignore new_create = new_create_async if func_is_async else new_create_sync if client is not None: client.chat.completions.create = new_create # type: ignore return client else: return new_create # type: ignore其中handle_response_model的部分会针对不同模型的API接口进行不同的指令处理，上面使用OpenAI时使用了工具调用模式来实现结构化输出。
def openai_schema(cls) -> dict[str, Any]: """ Return the schema in the format of OpenAI's schema as jsonschema Note: Its important to add a docstring to describe how to best use this class, it will be included in the description attribute and be part of the prompt. Returns: model_json_schema (dict): A dictionary in the format of OpenAI's schema as jsonschema """ schema = cls.model_json_schema() docstring = parse(cls.__doc__ or "") parameters = { k: v for k, v in schema.items() if k not in ("title", "description") } for param in docstring.params: if (name := param.arg_name) in parameters["properties"] and ( description := param.description ): if "description" not in parameters["properties"][name]: parameters["properties"][name]["description"] = description parameters["required"] = sorted( k for k, v in parameters["properties"].items() if "default" not in v ) if "description" not in schema: if docstring.short_description: schema["description"] = docstring.short_description else: schema["description"] = ( f"Correctly extracted `{cls.__name__}` with all " f"the required parameters with correct types" ) return { "name": schema["title"], "description": schema["description"], "parameters": parameters, }拿前面基金Function Call的例子来说，实际进入GPT模型的指令被转换成了以下函数调用的指令格式
{'name': 'TaskSequence', 'description': 'Correctly extracted `TaskSequence` with all the required parameters with correct types', 'parameters': {'$defs': {'BingSearch': {'properties': {'query': {'description': '网页搜索query', 'title': 'Query', 'type': 'string'}}, 'required': ['query'], 'title': 'BingSearch', 'type': 'object'}, 'FundInfo': {'description': '可以通过基金代码或基金名称，查询基金基础信息', 'properties': {'fund_code_or_name': {'anyOf': [{'type': 'string'}, {'type': 'null'}], 'description': '提问提及的基金代码或名称，没有则为空', 'title': 'Fund Code Or Name'}, 'lookup_field': {'enum': ['fund_manager', 'unit_value', 'contract_date', 'manage_fee', 'net_value'], 'title': 'Lookup Field', 'type': 'string'}}, 'required': ['fund_code_or_name', 'lookup_field'], 'title': 'FundInfo', 'type': 'object'}, 'Task': {'properties': {'name': {'description': '任务名称', 'title': 'Name', 'type': 'string'}, 'tool': {'anyOf': [{'$ref': '#/$defs/BingSearch'}, {'$ref': '#/$defs/FundInfo'}], 'description': '完成任务所需调用的工具', 'title': 'Tool'}}, 'required': ['name', 'tool'], 'title': 'Task', 'type': 'object'}}, 'properties': {'reason': {'description': '先逐步思考要解决用户的问题需要哪些步骤', 'title': 'Reason', 'type': 'string'}, 'task_actions': {'description': '任务列表，按执行顺序依次排列', 'items': {'$ref': '#/$defs/Task'}, 'title': 'Task Actions', 'type': 'array'}}, 'required': ['reason', 'task_actions'], 'type': 'object'}}FRI缺少严格约束，所以只能依赖模型的指令遵从能力，有一定概率输出结果会无法还原成原始的的Pydantic类型。下面我们看另一种强约束的方案。
优劣对比
针对上述的两种结构化解码方案，对比常规的自然语言推理对模型效果的影响几何？我先是读到的第一篇论文（Let Me Speak Freely），核心结论其实是结构化输出会影响模型的推理效果。
但是随后Outlines的作者们就发了一篇博客指出了论文的几个核心问题。双方各自站的立场不同，但逻辑上个博客指出的几个论文的核心问题确实很有说服力，包括
- 论文使用自然语言推理和使用结构化输出推理的指令不同，因此效果不可比
- 论文使用了第二个大模型对结构化输出的结果进行解析（引入了更多错误），实际上正确的使用方式应该是直接使用推理输出来还原pydantic model即可，毕竟大家使用结构化输出的其中一个原因就是更好解析。
- 论文使用的结构化输出prompt质量有待提升
博客给出的最终结论是在GSM8k，Last Letter，Shuffled Object这三个任务上结构化输出相比NL输出都有提升。并且直接给出了基于Outlines的结果复现代码github repo（这里强烈建议大家去瞅瞅上面的博客，对于结构化输出有些很有意思的见解）
但是吸取前面盲目偏信前一篇论文的教训，其实在平时的任务尝试上，个人感觉结构化输出的效果和具体任务，Prompt（fewshot）质量，模型本身的指令能力强相关。因此还是倾向于在应用时充分对比NL和Structure的效果后再做应用。在大模型时代很多结论都有领域和模型局限性，大家需要在自己的场景上审慎判断哈哈~
“新年伊始，愿各位代码如诗行云流水，bug如朝露见光即散；创意如泉涌，论文如宝藏，实验如神助，成功率百分百！科研路上，你我皆是‘码’到成功的幸运儿！🎉”
想看更全的大模型论文·微调预训练数据·开源框架·AIGC应用 >> DecryPrompt |
|
English
简体中文
繁體中文
한국 사람
日本語
Deutsch
русский
بالعربية
TÜRKÇE
português
คนไทย
french
