MAX Pipelines Pipeline Variants 类型体系全解析：文本生成、Embeddings、图像与音频的输入输出类型指南

导读

本文聚焦 Modular MAX 平台中 max.pipelines.modeling.types.pipeline_variants 模块的类型体系，系统讲解文本生成（含多模态 VLM 支持）、Embeddings、像素（图像）生成与音频生成四类 pipeline 变体所依赖的请求、输入与输出类型。读完本文，你将掌握如何构造 TextGenerationRequest 会话消息（含 tool calling 与图片/视频内容）、理解 TextGenerationInputs 的批次语义与 BatchType 调度含义，并能正确使用 EmbeddingsGenerationOutput 等结果类型完成服务端与调度层的对接。

模块定位与文档入口

本文对应的官方 API 文档页为 pipelines.modeling.types.pipeline_variants.rst，该页面通过 Sphinx autosummary 将 max.pipelines.modeling.types.pipeline_variants 模块下的公开类型按四类能力分组罗列：

• Text generation：BatchType、ImageContentPart、MessageContent、TextContentPart、TextGenerationInputs、TextGenerationRequest、TextGenerationRequestFunction、TextGenerationRequestMessage、TextGenerationRequestTool、VideoContentPart
• Audio generation：AudioGenerationInputs
• Embeddings：EmbeddingsContext、EmbeddingsGenerationContextType、EmbeddingsGenerationInputs、EmbeddingsGenerationOutput
• Image generation：PixelGenerationInputs

这些类型在仓库中的实际实现位于 pipeline_variants/init.py 及其四个子模块：text_generation.py、audio_generation.py、embeddings_generation.py、pixel_generation.py。__init__.py 中的 __all__ 还额外导出了 CompletedBatchStats（批次完成统计），它未出现在 RST 索引中，但在调度与指标场景中同样重要。直接导入方式如下：

from max.pipelines.modeling.types.pipeline_variants import (
    TextGenerationRequest,
    TextGenerationInputs,
    EmbeddingsGenerationInputs,
    EmbeddingsGenerationOutput,
    PixelGenerationInputs,
    AudioGenerationInputs,
)

所有 pipeline 变体类型都继承自 pipeline.py 中定义的基类/协议：PipelineInputs（输入标记基类）与 PipelineOutput（输出协议，要求实现 is_done 属性）。PipelineOutputsDict 类型别名则定义了 dict[RequestID, PipelineOutputType]，是 Pipeline.execute() 的统一返回形态。

文本生成：从请求到批次输入的完整类型链

文本生成是 pipeline_variants 中最庞大的一组类型，覆盖了从用户请求（TextGenerationRequest）、会话消息（TextGenerationRequestMessage）、多模态内容（MessageContent 及其 content part）到调度批次输入（TextGenerationInputs）的全链路。实现位于 text_generation.py。

1. TextGenerationRequest：不可变文本生成请求

TextGenerationRequest（text_generation.py#L325-L519）是一个 frozen=True 的 dataclass，用于描述一次文本 token 生成请求。其关键字段如下：

字段	类型	默认值	说明
request_id	RequestID	必填	请求唯一标识
model_name	str	必填	目标模型名，需与服务器可用模型匹配
prompt	str | Sequence[int] | None	None	兼容 legacy completion API，可传字符串或 token ID 序列
messages	list[TextGenerationRequestMessage]	[]	聊天补全用会话消息列表
images	list[bytes]	[]	多模态图片字节数组
videos	list[bytes]	[]	多模态视频字节数组，预处理时解码为帧
decoded_images	list[PILImage]	[]	已在 API 服务端解码的 PIL 图像，与 images 一一对应
tools	list[TextGenerationRequestTool] | None	None	可供模型调用的工具定义
response_format	TextGenerationResponseFormat | None	None	结构化输出格式（json_schema / grammar）
timestamp_ns	int	0	服务端收到请求的纳秒时间戳
request_path	str	"/"	请求端点路径，用于路由与日志
logprobs	int	0	每个 token 返回的 top-logprob 数量，0 表示不返回
echo	bool	False	为 True 时响应包含原始 prompt
chat_template_options	dict | None	None	应用聊天模板时的附加选项
sampling_params	SamplingParams	SamplingParams()	token 采样配置
target_endpoint	str | None	None	分离式服务（disaggregated serving）下指定目标实例路由
dkv_cache_hint	dict | None	None	分布式 KV cache 的 Orchestrator 缓存提示
cache_salt	str | None	None	按请求隔离 prefix-cache 条目的盐值

__post_init__ 中实现了一系列运行时校验（text_generation.py#L453-L493）：

• messages 中的 dict 会被自动转换为 TextGenerationRequestMessage 实例；
• prompt 与 messages 互斥，同时提供会抛出 ValueError；
• 提供 images 或 videos 时不允许使用字符串 prompt（必须走 messages）；
• 多模态一致性校验：images 数量必须等于消息中 ImageContentPart 的数量（number_of_images），videos 数量必须等于 VideoContentPart 的数量，否则报错，从源头杜绝"有图无占位符"或"占位符无图"的错位。

此外还提供了 images_for_processing() 方法：优先返回预先解码的 decoded_images（API 服务端在准入时已完整解码并校验过图像，避免 tokenizer 二次解码），离线/测试调用方则回退到原始 images 字节。

cache_salt 与 dkv_cache_hint 的工程细节：cache_salt 会与 kv_cache_hash_seed 通过 XOR 组合来播种 block hash，在 sha256/sha256_64 算法下提供加密级隔离保证，在 ahash64 下为尽力而为；OpenAI schema 层将其截断为 512 字符。dkv_cache_hint 由服务层原样序列化到 TextContext.dkv_cache_hint，交由 Rust 侧 dKV connector 解析，用于把每个 block 路由到持有它的实例——服务层本身并不读取其内容。

2. 会话消息与多模态内容：TextGenerationRequestMessage 与 MessageContent

TextGenerationRequestMessage（text_generation.py#L157-L322)遵循 OpenAI 对话规范，role 支持 system、user、assistant、tool、function 以及厂商扩展的 root（部分聊天模板将其排序在 system 之上）。核心字段：

• content：str | list[MessageContent]，默认为空字符串（兼容仅携带 tool_calls 的 assistant 消息）；
• tool_calls：OpenAI 形状的 {id, type, function: {name, arguments}} 列表，原样透传给聊天模板，保证多轮工具调用的 prompt 正确渲染；
• tool_call_id：工具消息回应的目标调用标识；
• reasoning_content：assistant 回合伴随产生的思考内容。

content 的 field_validator 提供了宽松的归一化逻辑：None 折叠为空字符串；字符串直接通过；dict 列表按 type 分派——text 取 text/content 字段、image/video 生成对应占位 part、image_url/video_url 明确报错（内部格式要求图片必须以字节形式放在 TextGenerationRequest.images 中，消息内容里只放 type='image' 占位符）。flatten_content() 将消息扁平化为 {role, content} dict 并保留工具调用元数据，供纯文本聊天模板消费。

MessageContent 是联合类型别名 TextContentPart | ImageContentPart | VideoContentPart（text_generation.py#L149），三个 part 都是 frozen=True 的 pydantic 模型：

• TextContentPart：type="text" + text 字段；
• ImageContentPart：type="image"，可选厂商提示 detail（OpenAI 质量档位）与 max_long_side_pixel（图像预处理最大长边像素），None 表示未设置、模型可忽略；
• VideoContentPart：type="video"，可选 fps（采样帧率）、max_frames（最大采样帧数）、detail、max_long_side_pixel。

TextGenerationRequestMessage 还提供了 number_of_images / number_of_videos 缓存属性，供上层做一致性校验。

3. 工具调用定义：TextGenerationRequestTool 与 Function

工具定义采用两层 TypedDict 结构（text_generation.py#L57-L77）：

class TextGenerationRequestFunction(TypedDict):
    name: str
    description: str | None
    parameters: dict[str, Any]  # 通常遵循 JSON Schema

class TextGenerationRequestTool(TypedDict):
    type: str                 # 工具类别
    function: TextGenerationRequestFunction

该结构对应 OpenAI Function Calling 约定：function.parameters 以 JSON Schema 描述函数入参，模型可在生成过程中决定是否调用。

4. BatchType 与 TextGenerationInputs：调度层的批次语义

BatchType（text_generation.py#L522-L528）是双值枚举，直接反映服务端调度器的两阶段执行模式：

• CE（Context Encoding）：上下文编码批次；
• TG（Token Generation）：token 生成批次。

TextGenerationInputs（text_generation.py#L602-L698）是文本生成 pipeline 步骤的批次输入，泛型参数为 TextGenerationContextType。其 batches: list[list[TextGenerationContextType]] 支持数据并行多副本——每个内层 batch 对应一个设备副本。__post_init__ 自动完成三项统计：

• input_tokens：flat_batch 中各上下文 tokens.active_length 之和；
• context_tokens：tokens.processed_length 之和；
• batch_type：只要存在任一 generated_length == 0 的上下文，整个批次即判定为 CE，否则为 TG。

per_replica_input_tokens / per_replica_context_tokens 在构造时冻结（排除 DP padding 哑上下文，即 _is_padding_ctx=True 的条目），因为调度过程中 token 窗口会持续变化，后续读取 active_length 已无法还原构造时刻的批次描述。此外还提供 batch_size、enable_echo、enable_log_probs、batch_top_log_probs、batch_echo 等便捷属性，以及 __bool__（按 flat_batch 是否为空判断）。

同文件中的 CompletedBatchStats（text_generation.py#L531-L599）记录已同步批次的执行统计：batch_type、batch_size、num_input_tokens、num_context_tokens、execution_time_s，并针对投机解码（speculative decoding）补充 num_output_tokens、draft_tokens_generated、draft_tokens_accepted、avg_acceptance_length、max_acceptance_length、acceptance_rate_per_position。其 prompt_throughput 与 generation_throughput 属性分别计算输入侧与生成侧的 tokens/秒吞吐（TG 批次在有输出 token 计数时按实际输出数计算，否则按每请求 1 token 估算）。

采样参数与响应格式：请求的关联类型

TextGenerationRequest.sampling_params 的类型为 SamplingParams，实现在 sampling_params.py，常用字段及默认值：

字段	默认值	说明
top_k	-1	仅从概率最高的 K 个 token 中采样，-1 表示全部，贪心设为 1
top_p	1	累计概率阈值
min_p	0.0	相对最可能 token 概率的最低保留阈值，0 禁用
temperature	1	随机性控制，贪心设为 0
thinking_temperature	None	<think> 块内 token 的温度覆盖，需要配置 reasoning parser
frequency_penalty / presence_penalty	0.0	频率/存在惩罚
repetition_penalty	1.0	重复惩罚，>1 时按除法压低已出现 token
max_new_tokens / min_new_tokens	None / 0	新 token 数量上下限
ignore_eos	False	忽略 EOS 继续生成
stop / stop_token_ids	None	字符串/token id 停止条件
detokenize	True	是否将输出 token 解码为文本
seed	加密安全随机值	随机数种子
logits_processors	None	logits 后处理回调序列

SamplingParams.from_input_and_generation_config() 会按"用户显式参数 > 模型 HuggingFace GenerationConfig > 类默认值"的优先级合并三者，并且当 do_sample=False 时自动切到贪心默认值。

TextGenerationResponseFormat（context.py#L46-L95）用于结构化输出控制：type 可为 json_object 或 grammar；json_schema=None 表示无约束，而显式 {} 表示"强制输出任意合法 JSON 值"（两者语义不同）；grammar 优先于 json_schema（用于 Kimi 工具调用语法等模型特定约束）；grammar_enforced 控制是否从第一个 token 起强制 grammar；requires_structured_output_flag 标记是否需要服务端开启 --enable-structured-output 标志（纯工具调用 grammar 由服务端控制，无需该标志）。

Embeddings 变体：上下文协议、输入与输出

Embeddings 类型的实现在 embeddings_generation.py，其设计原则是去掉文本生成特有的一切状态，只保留单步嵌入所需的最小接口。

EmbeddingsContext（L29-L64）是 @runtime_checkable 的 Protocol，继承 BaseContext，仅要求两个属性：tokens: TokenBuffer（输入 token）与 model_name: str（嵌入模型名）。docstring 明确列出被排除的文本生成特性：eos_token_ids、matcher（结构化输出 grammar）、json_schema、log_probabilities 与 token 生成迭代状态——这为嵌入 pipeline 提供了比文本生成上下文更轻量的类型约束。

EmbeddingsGenerationContextType = TypeVar(..., bound=EmbeddingsContext) 是该协议的 TypeVar 绑定。EmbeddingsGenerationInputs（L72-L81）是 frozen dataclass，batches: list[dict[RequestID, EmbeddingsContext]] 支持多副本，batch 属性将其合并为单一 dict[RequestID, EmbeddingsContext]。

EmbeddingsGenerationOutput（L84-L101）是 msgspec.Struct(tag=True, omit_defaults=True)，携带 embeddings: npt.NDArray[np.floating[Any]]（NumPy 浮点数组），其 is_done 恒为 True——嵌入生成是单步操作，天然满足 PipelineOutput 协议（文件末尾通过 _check_embeddings_output_implements_pipeline_output 做了类型层面的运行时验证）。

图像生成变体：PixelGenerationInputs 与请求参数

像素（图像）生成类型在 pixel_generation.py。模块内的私有 _PixelGenerationRequest（L33-L62）集中定义了图像生成的请求参数及校验：

字段	默认值	说明
prompt / secondary_prompt	必填 / None	主提示与辅助提示
negative_prompt / secondary_negative_prompt	None	负向提示
guidance_scale	3.5	CFG 引导强度
true_cfg_scale	1.0	真实 CFG 强度
height / width	None	输出尺寸，须为正数
num_inference_steps	50	推理步数，须为正数
num_images_per_prompt	1	每提示生成张数，须为正数
seed	None	随机种子
input_image	npt.NDArray[np.uint8] | None	可选输入图像（如图像编辑/图生图）

校验逻辑包括：prompt 不能为空；height/width 若给定必须为正；num_inference_steps 与 num_images_per_prompt 必须为正。公开类型 PixelGenerationInputs（L65-L72）则是 PipelineInputs 的泛型子类，batch: dict[RequestID, PixelGenerationContextType] 将请求 ID 映射到像素生成上下文。

音频生成变体：AudioGenerationInputs

音频生成变体是四类中最精简的，实现在 audio_generation.py。AudioGenerationInputs 为 frozen dataclass，泛型绑定 AudioGenerationContextType，唯一字段 batch: dict[RequestID, AudioGenerationContextType]，结构上与 PixelGenerationInputs 一致。从源码结构看，音频生成 pipeline 的请求参数与上下文细节封装在 AudioGenerationContextType 对应的 context 实现中，pipeline_variants 层仅提供统一的输入容器。

类型体系在服务端与调度层的落地

这组类型并非孤立的数据类，它们贯穿了 MAX 服务端的请求准入、调度与结果回传链路：

• OpenAI 路由层：在 openai_routes.py 中，/v1/chat/completions 等路由将请求体组装为 TextGenerationRequest（例如 L2222-L2243），其中 cache_salt 由 _get_cache_salt() 依据 X-Cache-Salt 请求头或请求体提取，且仅在服务设置 use_client_cache_salt=True 时生效，并受 _CACHE_SALT_MAX_LEN 长度限制。
• Embeddings 调度器：embeddings_scheduler.py 将批处理请求组装为 EmbeddingsGenerationInputs(batches=<a href="https://link.gitcode.com/i/de52196c761baadd56574f5e03624b41" target="_blank">batch_to_execute])（[L116），并消费 dict[RequestID, EmbeddingsGenerationOutput] 形式的批次响应。
• Worker 边界：zmq_interface.py 与 llm.py 负责跨进程序列化——EmbeddingsGenerationOutput 作为 msgspec.Struct 可被 EngineQueue 正确反序列化，llm.py 中会对响应类型做运行时断言（expected EmbeddingsGenerationOutput）。这解释了为何 PipelineOutput 被设计为 Protocol 而非抽象基类：msgspec.Struct 无法多重继承普通 ABC。
• 通用 Pipeline 接口：所有变体最终都对接 pipeline.py 中 Pipeline[PipelineInputsType, PipelineOutputType] 抽象类的 execute(inputs) -> PipelineOutputsDict 与 release(request_id) 契约。

小结

pipeline_variants 模块是 MAX 平台"请求 → 批次输入 → 输出"类型链路的枢纽：文本生成侧以 TextGenerationRequest 承载 OpenAI 风格会话（含工具调用与多模态内容）、以 TextGenerationInputs + BatchType 表达调度批次；Embeddings、像素、音频侧分别以轻量协议与统一输入容器覆盖各自能力。理解这套类型体系，是阅读 MAX 服务端源码、扩展自定义 pipeline 或接入 OpenAI 兼容接口的起点。建议继续阅读 types 包目录 下的 pipeline.py、context 包 中的 SamplingParams 与 TextContext，以及 serve 目录 中的路由与调度实现，以获得完整上下文。