模型推理服务人工智能后端大模型MLOpsLLMOps【免费下载链接】BentoMLThe easiest way to serve AI apps and models - Build Model Inference APIs, Job queues, LLM apps, Multi-model pipelines, and more!项目地址https://gitcode.com/gh_mirrors/be/BentoML点击查看免费下载模型组合Model Composition是 BentoML 中把多个模型编排在一起、构建复杂 AI 应用如 RAG、AI Agent、多模型流水线的核心能力既可以在同一个 Service 内运行多个模型并暴露独立或组合的 API也可以把模型拆分到多个 Service 中按顺序Sequential或并行Concurrent协作甚至构建同时包含串行与并行路径的推理图Inference Graph。读完本文你将掌握bentoml.servicebentoml.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(taskzero-shot-classification, modelself.model_a_path, hypothesis_templateThis text is about {}) self.pipeline_b pipeline(tasksentiment-analysis, modelself.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.pyHuggingFaceModel继承自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、2memory默认以 Gi 为单位字符串可写512Mi、2Gitraffic还支持max_concurrency超出即拒绝的并发上限、concurrency服务并发处理能力等字段。__init__中初始化流水线类属性拿到模型路径后在__init__中构建 transformers pipelineService 实例化逻辑见 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编排模型接力流水线顺序编排让模型依次工作——前一个模型的输出成为后一个模型的输入非常适合「先预处理、再推理」的多阶段流水线。下面的示例中PreprocessingServiceCPU、2Gi 内存先处理输入InferenceServiceGPU、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(tasktext-classification, modelself.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(tasktext-classification, modelself.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远程服务地址、deploymentBentoCloud 上的部署名可配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()会构造RemoteProxysrc/_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(taskzero-shot-classification, modelself.model_a_path, hypothesis_templateThis 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(tasksentiment-analysis, modelself.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.pytest_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( tasktext-generation, modelself.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( tasktext-generation, modelself.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( tasktext-classification, modelself.model_path, tokenizerself.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_lengthMAX_LENGTH, num_return_sequencesNUM_RETURN_SEQUENCE, ), self.distilgpt2_generator.to_async.generate( # type: ignore original_sentence, max_lengthMAX_LENGTH, num_return_sequencesNUM_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该工作流的执行顺序是接收一段文本提示prompt作为输入GPT2 与 DistilGPT2并行基于该提示各生成一段新文本BERT串行地对每段生成文本打分返回包含generated生成文本与score质量分数的列表。配合开头的推理图可以直观理解其拓扑ParallelService的多个文本生成副本共享同一输入并行执行、按流量独立自动扩缩容输出汇入SequentialService的文本分类副本完成打分最终输出 JSON。从实现层面看Service.__attrs_post_init__src/_bentoml_sdk/service/factory.py会自动扫描类中所有Dependency类型的类属性并收集进Service.dependenciesall_services()会递归展开全部依赖形成完整的服务拓扑若两个依赖定义了同名冲突的 Service会抛出BentoMLConfigException提示依赖冲突。这也解释了为何bentoml.depends声明的服务会在一次bentoml serve中被统一编排、统一构建。五、底层机制速览一次组合调用经历了什么结合上文各段源码可以把一次「组合调用」的完整链路归纳为声明期HuggingFaceModel(...)作为类属性被Service.__attrs_post_init__收集进Service.models仅记录模型引用model_id、revision、endpoint 等元数据不会立即下载src/_bentoml_sdk/models/huggingface.py 的to_info()/to_create_schema()还会生成模型清单与远端注册信息依赖解析bentoml.depends(...)生成的Dependency描述符在首次类属性访问时调用get()src/_bentoml_sdk/service/dependency.py同进程依赖直接实例化self.on()远程依赖则构建RemoteProxyHTTP 客户端媒体类型视情况为application/json或 pickle并把自身登记到全局_dependencies列表以便服务退出时统一close()清理连接调用期在bentoml.api方法内部通过依赖.to_async.方法(...)或asyncio.gather(依赖A.to_async.方法(...), 依赖B.to_async.方法(...))发起跨服务调用to_async由_AsyncWrapper提供对同步方法用线程池包装、对异步方法直接透传src/_bentoml_sdk/service/factory.py序列化HTTP 层经由 src/_bentoml_impl/client/proxy.py 的AsyncHTTPClient/SyncHTTPClient完成请求的编码与响应的解码超时默认取目标服务traffic.timeout的 1.01 倍。六、注意事项与限制resources字段仅在部署平台生效文档明确指出resources配置GPU、内存、CPU 等只在 BentoCloud 等部署环境中决定实例规格本地bentoml serve运行时它不会限制本机资源占用。HuggingFaceModel返回本地路径其值是字符串下载后的模型目录路径必须传入有效的 Hugging Face 模型 IDrevision、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 与多模型流水线的核心编排能力。按是否需要独立扩缩容/异构硬件可按下表快速选型场景方案关键 API模型共享同一硬件、API 独立单 Service 多模型bentoml.service 多个bentoml.api模型需独立扩缩容、流水线接力多 Service 顺序编排bentoml.dependsto_async模型需并行推理后聚合多 Service 并发编排bentoml.dependsasyncio.gather串行与并行混合的复杂工作流推理图Inference Graph依赖组合 asyncio.gather 循环串行需要深入了解的部分可直接阅读仓库中的对应实现依赖机制源码、Service 工厂与异步包装、Service 配置 Schema、HuggingFace 模型引用、远程代理客户端以及组合服务的端到端测试 test_asgi.py。更完整的 Service API 说明可参考官方文档 services 与 distributed-services 相关内容。赞分享模型推理服务人工智能后端大模型MLOpsLLMOps【免费下载链接】BentoMLThe easiest way to serve AI apps and models - Build Model Inference APIs, Job queues, LLM apps, Multi-model pipelines, and more!项目地址https://gitcode.com/gh_mirrors/be/BentoML点击查看免费下载相关推荐Ray Serve 模型组合Model Composition实战用 DeploymentHandle 编排多阶段 AI 服务Ray Serve 模型组合Model Composition实战用 DeploymentHandle 编排多阶段 AI 服务 导读 在 Ray Serv人工智能分布式训练强化学习任务调度模型推理服务后端3个实用技巧彻底解决Mac外接显示器控制难题3个实用技巧彻底解决Mac外接显示器控制难题 还在为Mac外接显示器无法调节亮度而烦恼吗当你在深夜工作想要降低显示器亮度保护眼睛时却发现苹果系统无法识别外模型推理服务人工智能后端大模型MLOpsLLMOpsTriton Inference Server 业务逻辑脚本BLS完全指南在 Python 模型中编排多模型推理Triton Inference Server 业务逻辑脚本BLS完全指南在 Python 模型中编排多模型推理 本篇指南聚焦 Triton Infere模型推理服务AI 应用后端上一篇RPCS3补丁系统为PS3游戏注入全新生命力的智能模块下一篇Neural-SLAM 项目使用教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
