企业官网建设流程全解析

• 模型推理服务
• 人工智能
• 后端
• 大模型
• MLOps
• LLMOps

模型组合（Model Composition）是 BentoML 中把多个模型编排在一起、构建复杂 AI 应用（如 RAG、AI Agent、多模型流水线）的核心能力：既可以在同一个 Service 内运行多个模型并暴露独立或组合的 API，也可以把模型拆分到多个 Service 中按顺序（Sequential）或并行（Concurrent）协作，甚至构建同时包含串行与并行路径的推理图（Inference Graph）。读完本文，你将掌握@bentoml.service+@bentoml.api的组合用法、bentoml.depends服务间依赖调用、to_async异步包装与asyncio.gather并行聚合，并理解其背后的源码实现原理。

一、什么是模型组合，何时需要它

BentoML 的模型组合指：把多个模型组织起来共同完成一次推理任务，模型之间既可以依次接力（一个模型的输出是另一个模型的输入），也可以同时运行（各自独立推理后再合并结果）。官方文档（docs/source/get-started/model-composition.rst）归纳了以下典型使用场景：

• 处理不同类型的数据：例如用不同的模型分别处理图像与文本，再把结果组合起来；
• 提升准确率与性能：组合多个模型的结果（如集成学习 / Ensemble）改善最终预测质量；
• 异构硬件编排：让不同模型运行在不同硬件上，例如计算密集型模型跑 GPU、轻量模型跑 CPU，按需分配资源；
• 多阶段流水线：用专门化的模型或服务编排预处理（preprocessing）、推理（inference）与后处理（postprocessing）等串行步骤。

组合的粒度可以是一个 Service 内部，也可以是多个 Service 之间，取决于你对「独立扩缩容」和「异构硬件」的需求。@bentoml.service装饰器中的resources字段用于声明部署所需的资源（如 GPU），但需注意：该字段只在 BentoCloud 等部署平台上生效（详见 bentoml 配置参考）。

二、方案一：在单个 Service 中运行多个模型

如果多个模型共享同一套硬件设备（如同一块 GPU），你可以在同一个 Service 类中挂载多个模型，并为每个模型暴露独立 API，或提供一个组合多个模型结果的 API。这样一次部署即可同时服务多个模型。

import bentoml from bentoml.models import HuggingFaceModel from transformers import pipeline from typing import List # Run two models in the same Service on the same hardware device @bentoml.service( resources={"gpu": 1, "memory": "4GiB"}, traffic={"timeout": 20}, ) class MultiModelService: # Retrieve model references from HF by specifying its HF ID model_a_path = HuggingFaceModel("FacebookAI/roberta-large-mnli") model_b_path = HuggingFaceModel("distilbert/distilbert-base-uncased") def __init__(self) -> None: # Initialize pipelines for each model self.pipeline_a = pipeline(task="zero-shot-classification", model=self.model_a_path, hypothesis_template="This text is about {}") self.pipeline_b = pipeline(task="sentiment-analysis", model=self.model_b_path) # Define an API for data processing with model A @bentoml.api def process_a(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> dict: return self.pipeline_a(input_data, labels) # Define an API for data processing with model B @bentoml.api def process_b(self, input_data: str) -> dict: return self.pipeline_b(input_data)[0] # Define an API endpoint that combines the processing of both models @bentoml.api def combined_process(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> dict: classification = self.pipeline_a(input_data, labels) sentiment = self.pipeline_b(input_data)[0] return { "classification": classification, "sentiment": sentiment }

要点解析：

• 模型引用以类属性声明：HuggingFaceModel("FacebookAI/roberta-large-mnli")只是模型引用，并不会在类定义时立即下载。从源码看（src/_bentoml_sdk/models/huggingface.py），HuggingFaceModel继承自Model[str]抽象基类，其resolve()方法在实例访问该属性时才通过huggingface_hub.snapshot_download实际下载，并返回模型在本地的下载路径字符串。它支持revision（默认"main"）、endpoint（默认https://huggingface.co，可用环境变量HF_ENDPOINT覆盖）、include/exclude文件过滤等参数。
• 资源与流量配置：resources={"gpu": 1, "memory": "4GiB"}声明 1 张 GPU 与 4 GiB 内存；traffic={"timeout": 20}设置请求超时 20 秒。可参考 src/_bentoml_sdk/service/config.py 中ResourceSchema与TrafficSchema的定义：cpu传整数/浮点表示核数，传字符串可带单位（如"100m"、"0.5"、"2"）；memory默认以 Gi 为单位，字符串可写"512Mi"、"2Gi"；traffic还支持max_concurrency（超出即拒绝的并发上限）、concurrency（服务并发处理能力）等字段。
• __init__中初始化流水线：类属性拿到模型路径后，在__init__中构建 transformers pipeline；Service 实例化逻辑见 src/_bentoml_sdk/service/factory.py 的Service.__call__。
• API 暴露方式：@bentoml.api把方法暴露为 HTTP 端点（src/_bentoml_sdk/decorators.py），支持route、name、input_spec、output_spec、batchable、max_batch_size（默认 100）、max_latency_ms（默认 60000）等参数。

注意：HuggingFaceModel函数返回的是下载后的模型路径字符串，必须传入 Hugging Face 上显示的模型 ID，例如HuggingFaceModel("FacebookAI/roberta-large-mnli")。更多模型加载与管理细节可参考文档 model-loading-and-management。

三、方案二：在独立 Service 中运行与扩缩容多个模型

当多个模型需要独立扩缩容或需要不同硬件时，应把它们拆分到不同的 Service。Service 之间通过bentoml.depends建立依赖关系：一个 Service 可以调用另一个 Service 暴露的 API，如同调用本地方法。

3.1 顺序（Sequential）编排：模型接力流水线

顺序编排让模型依次工作——前一个模型的输出成为后一个模型的输入，非常适合「先预处理、再推理」的多阶段流水线。下面的示例中，PreprocessingService（CPU、2Gi 内存）先处理输入，InferenceService（GPU、4Gi 内存）再基于预处理结果完成推理，两个 Service 资源规格不同、可独立扩缩容：

import bentoml from bentoml.models import HuggingFaceModel from transformers import pipeline from typing import Dict, Any @bentoml.service(resources={"cpu": "2", "memory": "2Gi"}) class PreprocessingService: model_a_path = HuggingFaceModel("distilbert/distilbert-base-uncased") def __init__(self) -> None: # Initialize pipeline for model A self.pipeline_a = pipeline(task="text-classification", model=self.model_a_path) @bentoml.api def preprocess(self, input_data: str) -> Dict[str, Any]: # Dummy preprocessing steps return self.pipeline_a(input_data)[0] @bentoml.service(resources={"gpu": 1, "memory": "4Gi"}) class InferenceService: model_b_path = HuggingFaceModel("distilbert/distilroberta-base") preprocessing_service = bentoml.depends(PreprocessingService) def __init__(self) -> None: # Initialize pipeline for model B self.pipeline_b = pipeline(task="text-classification", model=self.model_b_path) @bentoml.api async def predict(self, input_data: str) -> Dict[str, Any]: # Dummy inference on preprocessed data # Implement your custom logic here preprocessed_data = await self.preprocessing_service.to_async.preprocess(input_data) final_result = self.pipeline_b(input_data)[0] return { "preprocessing_result": preprocessed_data, "final_result": final_result }

关键机制：

• bentoml.depends声明依赖：preprocessing_service = bentoml.depends(PreprocessingService)接收被依赖的 Service 类作为参数，之后即可调用其暴露的 API。从源码（src/_bentoml_sdk/service/dependency.py）看，depends()实际上返回一个Dependency描述符，它支持三种依赖来源：on（本 Bento 内的 Service 类）、url（远程服务地址）、deployment（BentoCloud 上的部署名，可配cluster）。Dependency.__get__在首次访问类属性时通过get()解析：若目标 Service 在同进程内则直接返回其实例（self.on()），否则包装为远程代理RemoteProxy。
• to_async包装：Service.to_async把同步 API 方法包装为协程（见 src/_bentoml_sdk/service/factory.py 中_AsyncWrapper的实现——对同步函数通过anyio.to_thread.run_sync在线程池中执行，避免阻塞事件循环）；Service.to_sync则把异步方法包装回同步调用。官方文档特别强调：在异步上下文中直接调用同步阻塞函数会阻塞事件循环，不推荐，因此应使用.to_async。
• 跨服务调用走 HTTP：当被依赖的 Service 是远程进程时，Dependency.get()会构造RemoteProxy（src/_bentoml_impl/client/proxy.py），内部同时持有AsyncHTTPClient与SyncHTTPClient，通过.to_async/.to_sync属性切换调用风格；其超时默认取自目标服务traffic.timeout并加 1% 余量。

3.2 并发（Concurrent）编排：并行推理 + 结果聚合

并发编排让多个相互独立的模型同时运行，再把结果聚合在一起，适合集成模型（Ensemble）等需要综合多模型预测以提升准确率的场景。实现上使用asyncio.gather并行发起对多个依赖 Service 的调用：

import asyncio import bentoml from bentoml.models import HuggingFaceModel from transformers import pipeline from typing import Dict, Any, List @bentoml.service(resources={"gpu": 1, "memory": "4Gi"}) class ModelAService: model_a_path = HuggingFaceModel("FacebookAI/roberta-large-mnli") def __init__(self) -> None: # Initialize pipeline for model A self.pipeline_a = pipeline(task="zero-shot-classification", model=self.model_a_path, hypothesis_template="This text is about {}") @bentoml.api def predict(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> Dict[str, Any]: # Dummy preprocessing steps return self.pipeline_a(input_data, labels) @bentoml.service(resources={"gpu": 1, "memory": "4Gi"}) class ModelBService: model_b_path = HuggingFaceModel("distilbert/distilbert-base-uncased") def __init__(self) -> None: # Initialize pipeline for model B self.pipeline_b = pipeline(task="sentiment-analysis", model=self.model_b_path) @bentoml.api def predict(self, input_data: str) -> Dict[str, Any]: # Dummy preprocessing steps return self.pipeline_b(input_data)[0] @bentoml.service(resources={"cpu": "4", "memory": "8Gi"}) class EnsembleService: service_a = bentoml.depends(ModelAService) service_b = bentoml.depends(ModelBService) @bentoml.api async def ensemble_predict(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> Dict[str, Any]: result_a, result_b = await asyncio.gather( self.service_a.to_async.predict(input_data, labels), self.service_b.to_async.predict(input_data) ) # Dummy aggregation return { "zero_shot_classification": result_a, "sentiment_analysis": result_b }

要点：EnsembleService通过两个bentoml.depends同时依赖ModelAService与ModelBService，在ensemble_predict中用asyncio.gather让两次to_async.predict并行执行，最后按需聚合。asyncio.gather会并发调度多个协程，配合to_async的线程池包装，即便下游 API 是同步实现的也能并行等待。仓库中的端到端测试 tests/e2e/bento_new_sdk/test_asgi.py（test_composed_service）验证了bentoml.depends组合服务后通过Service.to_asgi()挂载并正常完成跨服务调用的行为。

四、方案三：推理图（Inference Graph）——串行与并行混合编排

当工作流需要同时包含并行与串行路径时，可以构建推理图。下方示例是一个经典的「文本生成 → 生成质量打分」流水线：GPT2 与 DistilGPT2并行生成文本，BERT 再串行地对每段生成文本打分：

import asyncio import typing as t import transformers import bentoml MAX_LENGTH = 128 NUM_RETURN_SEQUENCE = 1 @bentoml.service( resources={"gpu": 1, "memory": "4Gi"} ) class GPT2: model_path = bentoml.models.HuggingFaceModel("openai-community/gpt2") def __init__(self): self.generation_pipeline_1 = transformers.pipeline( task="text-generation", model=self.model_path, ) @bentoml.api def generate(self, sentence: str) -> t.List[t.Any]: return self.generation_pipeline_1(sentence) @bentoml.service( resources={"gpu": 1, "memory": "4Gi"} ) class DistilGPT2: model_path = bentoml.models.HuggingFaceModel("distilbert/distilgpt2") def __init__(self): self.generation_pipeline_2 = transformers.pipeline( task="text-generation", model=self.model_path, ) @bentoml.api def generate(self, sentence: str) -> t.List[t.Any]: return self.generation_pipeline_2(sentence) @bentoml.service( resources={"cpu": "2", "memory": "2Gi"} ) class BertBaseUncased: model_path = bentoml.models.HuggingFaceModel("google-bert/bert-base-uncased") def __init__(self): self.classification_pipeline = transformers.pipeline( task="text-classification", model=self.model_path, tokenizer=self.model_path, ) @bentoml.api def classify_generated_texts(self, sentence: str) -> float | str: score = self.classification_pipeline(sentence)[0]["score"] # type: ignore return score @bentoml.service( resources={"cpu": "4", "memory": "8Gi"} ) class InferenceGraph: gpt2_generator = bentoml.depends(GPT2) distilgpt2_generator = bentoml.depends(DistilGPT2) bert_classifier = bentoml.depends(BertBaseUncased) @bentoml.api async def generate_score( self, original_sentence: str = "I have an idea!" ) -> t.List[t.Dict[str, t.Any]]: generated_sentences = [ # type: ignore result[0]["generated_text"] for result in await asyncio.gather( # type: ignore self.gpt2_generator.to_async.generate( # type: ignore original_sentence, max_length=MAX_LENGTH, num_return_sequences=NUM_RETURN_SEQUENCE, ), self.distilgpt2_generator.to_async.generate( # type: ignore original_sentence, max_length=MAX_LENGTH, num_return_sequences=NUM_RETURN_SEQUENCE, ), ) ] results = [] for sentence in generated_sentences: # type: ignore score = await self.bert_classifier.to_async.classify_generated_texts( sentence ) # type: ignore results.append( { "generated": sentence, "score": score, } ) return results

该工作流的执行顺序是：

1. 接收一段文本提示（prompt）作为输入；
2. GPT2 与 DistilGPT2并行基于该提示各生成一段新文本；
3. BERT串行地对每段生成文本打分；
4. 返回包含generated（生成文本）与score（质量分数）的列表。

配合开头的推理图可以直观理解其拓扑：ParallelService的多个文本生成副本共享同一输入并行执行、按流量独立自动扩缩容，输出汇入SequentialService的文本分类副本完成打分，最终输出 JSON。

从实现层面看，Service.__attrs_post_init__（src/_bentoml_sdk/service/factory.py）会自动扫描类中所有Dependency类型的类属性并收集进Service.dependencies，all_services()会递归展开全部依赖形成完整的服务拓扑；若两个依赖定义了同名冲突的 Service，会抛出BentoMLConfigException提示依赖冲突。这也解释了为何bentoml.depends声明的服务会在一次bentoml serve中被统一编排、统一构建。

五、底层机制速览：一次组合调用经历了什么

结合上文各段源码，可以把一次「组合调用」的完整链路归纳为：

1. 声明期：HuggingFaceModel(...)作为类属性被Service.__attrs_post_init__收集进Service.models，仅记录模型引用（model_id、revision、endpoint 等元数据），不会立即下载（src/_bentoml_sdk/models/huggingface.py 的to_info()/to_create_schema()还会生成模型清单与远端注册信息）；
2. 依赖解析：bentoml.depends(...)生成的Dependency描述符在首次类属性访问时调用get()（src/_bentoml_sdk/service/dependency.py）：同进程依赖直接实例化self.on()；远程依赖则构建RemoteProxy（HTTP 客户端，媒体类型视情况为application/json或 pickle），并把自身登记到全局_dependencies列表以便服务退出时统一close()清理连接；
3. 调用期：在@bentoml.api方法内部通过依赖.to_async.方法(...)或asyncio.gather(依赖A.to_async.方法(...), 依赖B.to_async.方法(...))发起跨服务调用；to_async由_AsyncWrapper提供，对同步方法用线程池包装、对异步方法直接透传（src/_bentoml_sdk/service/factory.py）；
4. 序列化：HTTP 层经由 src/_bentoml_impl/client/proxy.py 的AsyncHTTPClient/SyncHTTPClient完成请求的编码与响应的解码，超时默认取目标服务traffic.timeout的 1.01 倍。

六、注意事项与限制

• resources字段仅在部署平台生效：文档明确指出resources配置（GPU、内存、CPU 等）只在 BentoCloud 等部署环境中决定实例规格；本地bentoml serve运行时它不会限制本机资源占用。
• HuggingFaceModel返回本地路径：其值是字符串（下载后的模型目录路径），必须传入有效的 Hugging Face 模型 ID；revision、endpoint、include/exclude可在类属性上配置，构建 Bento 时会连同模型元数据一起固化（Service.on_load_bento会用 Bento 内的模型信息回填 model_id 与 revision）。
• 避免阻塞事件循环：异步 API 中不要直接调用同步阻塞函数，应通过被依赖服务的.to_async属性调用（内部用线程池隔离）。
• LLM 到 LLM 的流式传递暂不支持：官方文档说明，把某个 LLM 的输出流式地直接作为另一个 LLM 的输入（构建复合 LLM 系统）目前尚未在 BentoML 中支持，但已列入 roadmap；社区可在官方论坛或 GitHub issue 中参与讨论。这意味着现阶段编排 LLM 时，需要将上游的完整生成结果作为下游输入。

七、小结与延伸

模型组合是 BentoML 支撑 RAG、AI Agent 与多模型流水线的核心编排能力。按是否需要独立扩缩容/异构硬件，可按下表快速选型：

需要深入了解的部分，可直接阅读仓库中的对应实现：依赖机制源码、Service 工厂与异步包装、Service 配置 Schema、HuggingFace 模型引用、远程代理客户端，以及组合服务的端到端测试 test_asgi.py。更完整的 Service API 说明可参考官方文档 services 与 distributed-services 相关内容。

• 模型推理服务
• 人工智能
• 后端
• 大模型
• MLOps
• LLMOps