用 Instructor 实现 Demonstration Ensembling（DENSE）：最大化利用少样本示例的集成提示技术

用 Instructor 实现 Demonstration Ensembling（DENSE）：最大化利用少样本示例的集成提示技术

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor

DENSE（Demonstration Ensembling，演示集成）是一种通过将训练集中的示例划分为多个子集、分别构造多份 few-shot 提示并多次调用模型，再对多个输出进行聚合的提示工程技术。本文基于 docs/prompting/ensembling/dense.md 展开，结合本项目仓库源码讲解其原理、完整实现与调优要点，读完你可以直接在 Instructor 项目中用少量代码落地这套"示例利用率最大化"的分类与推理方案。

核心思想：把有限的示例"用足"

少样本（few-shot）提示的性能通常受限于一次提示中能塞入的示例数量。DENSE 的思路非常直接：与其把所有示例挤进一次提示，不如把示例分成多个互斥子集，分别构造多份提示并行调用模型，最后对多个回答做聚合（如多数投票）得出最终答案。

这一技术源自论文《Exploring Demonstration Ensembling for In Context Learning》（arXiv:2308.08780）。其核心动机是：单个子集只覆盖示例空间的一部分，而多个子集叠加后能更全面地"教"模型理解任务模式，从而提升分类与推理的准确率。

用 Instructor 实现时，整个过程可以抽象为三个阶段：

1. 划分：把示例列表划分成num_responses个大小相等的子集；
2. 并行推理：用每个子集分别构造系统提示，异步并发调用模型得到多个结构化响应；
3. 聚合：对全部响应的答案字段做多数投票（majority vote），输出票数最高的结果。

原文档特别指出：为了简单，示例采用"遍历并等分"的方式得到大小相同的簇。但在真实业务中，你完全可以根据场景改用embedding 聚类（embedding clustering）等相似度采样方式来划分示例，让每个子集内部语义更内聚。

与 Self-Consistency 的区别：多样性来自"提示"而非"采样"

DENSE 与另一篇同目录文档 Self-Consistency（Self-Consistency Improves Chain Of Thought Reasoning In Language Models）容易混淆，两者都用到了"多次调用 + 多数投票"，但多样性来源截然不同：

从实现上看，Self-Consistency 的多次调用彼此独立（coros = [generate_self_consistent_response(prompt) for _ in range(num_responses)]），而 DENSE 的每次调用携带的是不同的示例子集，这正是它名字中 "Demonstration"（演示/示例）的由来。你可以把 DENSE 理解为 Self-Consistency 在 few-shot 场景下的增强变体。

用 Instructor 实现 DENSE：完整代码与逐段解析

原文给出了一个可运行的完整示例，其核心是异步客户端 + 结构化输出模型 +asyncio.gather并发。下面先给出完整代码（与原文档一致），再分段解析关键点。

import instructor from pydantic import BaseModel import asyncio from collections import Counter from typing import Literal from textwrap import dedent class DemonstrationResponse(BaseModel): correct_answer: Literal["Positive", "Negative", "Neutral"] client = instructor.from_provider("openai/gpt-5-nano", async_client=True) async def generate_self_consistent_response(prompt: str, examples: list[str]): concetenated_examples = "\n".join(examples) return await client.create( model="gpt-4o", messages=[ { "role": "system", "content": dedent( f""" You are an intelligent AI System that excels at classifying user queries into three possible labels: - Positive - Negative - Neutral You are about to be given a user query and asked to classify it into one of the three categories. Make sure to refer closely to the examples provided to you, examining each individual example before coming up with the final answer. Here are the examples: {concetenated_examples} """ ), }, {"role": "user", "content": prompt}, ], response_model=DemonstrationResponse, temperature=0, ) async def generate_self_consistent_responses( prompt: str, num_responses: int, examples: list[str] ): assert ( len(examples) % num_responses == 0 ), "The number of examples must be evenly divisible by num_responses" # Batch the examples into num_responses batches batch_size = len(examples) // num_responses coros = [ generate_self_consistent_response(prompt, examples[i : i + batch_size]) for i in range(0, len(examples), batch_size) ] responses = await asyncio.gather(*coros) return responses if __name__ == "__main__": user_query = "What is the weather like today?" examples = [ "I love this product! [Positive]", "This is the worst service ever. [Negative]", "The movie was okay, not great but not terrible. [Neutral]", "I'm so happy with my new phone! [Positive]", "The food was terrible and the service was slow. [Negative]", "It's an average day, nothing special. [Neutral]", "Fantastic experience, will come again! [Positive]", "I wouldn't recommend this to anyone. [Negative]", "The book was neither good nor bad. [Neutral]", "Absolutely thrilled with the results! [Positive]", ] responses = asyncio.run(generate_self_consistent_responses(user_query, 5, examples)) answer_counts = Counter([response.correct_answer for response in responses]) most_common_answer, _ = answer_counts.most_common(1)[0] print(most_common_answer) #> Neutral

1. 用结构化输出模型约束答案空间

class DemonstrationResponse(BaseModel): correct_answer: Literal["Positive", "Negative", "Neutral"]

这里的关键是Literal联合类型 + PydanticBaseModel。Instructor 会把该模型转换成工具调用（tool call）的 JSON Schema，强制模型从三个预定义标签中选一个，从而杜绝模型自由发挥输出Positive 🙂之类不可聚合的变体。聚合阶段之所以能放心使用Counter做多数投票，正是因为有这一层结构化约束兜底。

2. 通过from_provider创建异步客户端

client = instructor.from_provider("openai/gpt-5-nano", async_client=True)

from_provider是本仓库统一的多提供商入口，其定义位于 instructor/v2/auto_client.py。从函数签名可以看出，它接受model（字符串或已知模型名）、async_client（布尔值）、cache、mode等参数：

• 当async_client=True时返回AsyncInstructor（异步客户端），从而可以在async def函数里使用await client.create(...)；
• 当async_client=False（默认）时返回同步的Instructor。

从源码看（instructor/init.py），from_provider通过模块级懒加载机制从.auto_client导入，因此即使只安装了部分提供商 SDK 也不会影响包导入性能。

client.create(...)是 Instructor 的核心调用入口，参数包括：

提示：from_provider("openai/gpt-5-nano", ...)中openai/gpt-5-nano用于路由到 OpenAI 提供商并确定默认模型，而create()里的model="gpt-4o"是本次请求实际使用的模型，两者可以独立指定。这是本仓库统一提供商接口的常见用法（可参考 docs/integrations/model-selection.md）。

3. 动态拼接示例子集

concetenated_examples = "\n".join(examples)

每个子集内的示例通过换行拼接后，经 f-string 注入系统提示中的 "Here are the examples:" 段落。textwrap.dedent用于去除多行字符串的公共缩进，保证送入模型的提示格式干净一致。系统提示明确要求模型"逐个仔细检查每个示例后再给出最终答案"，这正是 DENSE 想让模型充分吸收子集示例的关键措辞。

4. 等分示例 + 并发聚合

async def generate_self_consistent_responses(prompt, num_responses, examples): assert len(examples) % num_responses == 0, "The number of examples must be evenly divisible by num_responses" batch_size = len(examples) // num_responses coros = [ generate_self_consistent_response(prompt, examples[i : i + batch_size]) for i in range(0, len(examples), batch_size) ] responses = await asyncio.gather(*coros) return responses

这是整个方案的"装配线"：

• 断言约束：要求len(examples)能被num_responses整除，保证每个子集大小一致。本示例 10 个示例、5 个子集，每个子集恰好 2 个示例。
• 等分切片：examples[i : i + batch_size]将示例列表切成num_responses个互不重叠的连续片段。你可以据此推断：示例的排列顺序会直接影响各子集的内容构成，因此对示例做随机打乱或按类别均匀重排，往往能改善子集的代表性。
• 并发调度：asyncio.gather(*coros)一次性并发发出num_responses个异步请求。相比串行 for 循环，网络等待时间被完全摊平，5 个子集调用几乎同时完成。
• 返回结构：返回值是一个DemonstrationResponse列表，顺序与coros对应。

5. 多数投票得到最终答案

answer_counts = Counter([response.correct_answer for response in responses]) most_common_answer, _ = answer_counts.most_common(1)[0] print(most_common_answer) #> Neutral

对 5 次子集推理的结果做Counter统计，取most_common(1)[0]得到票数最高的标签。在原文档的演示中，最终输出为Neutral——即对于 "What is the weather like today?" 这样的中性问句，多数子集判定为中性，符合直觉。

进阶调优与边界条件

在把 DENSE 迁移到自己的任务前，有几个值得注意的工程细节：

1. 整除性约束：assert len(examples) % num_responses == 0保证了等分，但也会限制参数选择。如果你的示例数无法整除，可以：
   • 调整num_responses为示例数的因数；
   • 或丢弃尾部少量示例，让子集数量与示例数匹配；
   • 或改造切片逻辑（如examples[i::num_responses]的跨步采样），允许子集大小差 1。
2. 子集划分策略：原文档明确提示，除等分外可考虑embedding 聚类——先用嵌入模型（如text-embedding-3-small，用法可参考 docs/prompting/ensembling/cosp.md 中的句子级 embedding 计算）对示例编码，再按语义相似度聚类分组，使每个子集内部语义更聚焦。对于类别分布不均的数据集，按类别分层抽样也能避免某个子集全是单一标签。
3. 投票权重：多数投票假设每个子集贡献等权。如果你的场景中某些子集明显更可靠（如聚类后与目标查询更相似的簇），可以改为加权投票，或参考 docs/prompting/ensembling/diverse.md 中 DiVeRSe 的做法——用另一个模型对每个响应打分（Poor/Average/Good/Excellent 映射为 0.25/0.5/0.75/1.0），按分数聚合而不是按票数聚合。
4. 与温度的关系：示例中temperature=0是为了消除同一子集内的采样随机性。若你的任务允许一定随机性，可调高temperature并结合 Self-Consistency（同目录 docs/prompting/ensembling/self_consistency.md）再做一层多轮采样，进一步放大候选多样性。
5. 异步环境适配：示例使用asyncio.run(...)在脚本入口运行协程。若你运行在已有的事件循环内（如 FastAPI），应使用await而不是再次asyncio.run，可参考 docs/concepts/async.md 与 examples/learn-async/run.py 中的异步客户端模式。

与其他 Ensembling 技术的关系

DENSE 是 docs/prompting/ensembling 目录下十种集成提示技术之一，该目录由 docs/prompting/index.md 统一索引。与它同族的方案各有侧重：

• COSP：在无标注数据上用归一化熵 + 重复度自动挑选高质量 few-shot 示例，再叠加 Self-Consistency；
• DiVeRSe：生成多样提示 + 验证器打分替代多数投票；
• MoRE：用多个专用专家提示（事实/多跳/数学/常识）分头推理，再按得分选出最佳；
• USP / Universal Self-Consistency：对自由形式答案做聚类/语义一致性的通用聚合。

DENSE 的独特定位在于：它把"示例"当作可切分的资源来调度，是"如何最有效地把手头标注数据喂给模型"这个问题的最朴素也最可靠的答案——实现成本低、可解释性强，且天然适配结构化输出与并发调用。

小结

通过本仓库的 Instructor，DENSE 的全部落地代码不过百行：一个Literal约束的响应模型保证答案可聚合，from_provider(..., async_client=True)提供异步客户端，等分切片 +asyncio.gather实现并发多提示推理，Counter.most_common完成多数投票。它不依赖任何额外训练或微调，只通过"多次使用不同示例子集"这一个动作，就显著提升了对有限标注数据的利用率，是 few-shot 场景下一个极易上手、便于扩展的基线方案。

参考资料

• 原始文档：docs/prompting/ensembling/dense.md
• 论文：Exploring Demonstration Ensembling for In Context Learning（arXiv:2308.08780）
• 统一提供商入口源码：instructor/v2/auto_client.py
• 相关技术：Self-Consistency、COSP、DiVeRSe、MoRE

【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor