尧图网站建设 
Cog Model Source 详解模型作者如何用 cog.yaml、Runner 与类型注解定义机器学习模型【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog本篇指南围绕 Cog 的模型源Model Source展开模型作者只需提供一个cog.yaml环境配置、一个包含setup()/run()方法的 Runner 类以及可选的模型权重Cog 就能把它构建成生产可用的 OCI 容器镜像并通过 架构总览 中描述的 CLI、Python SDK 与 coglet 运行时对外提供 HTTP 预测服务。读完本文你将掌握cog.yaml每个配置字段的真实语义、Runner 类的生命周期约定、基于类型注解的输入输出 schema 契约以及文件、密钥、流式输出与异步并发等进阶用法并能在仓库源码层面印证每一项行为。模型作者提供什么一个最小模型的三要素Cog 的设计哲学是用标准 Python 定义模型。一个完整的 Cog 模型在磁盘上只有三个组成部分my-model/ ├── cog.yaml # 运行时环境配置 ├── run.py # Runner 类模型加载与推理逻辑 └── weights/ # 模型权重可选也可以运行时下载其中cog.yaml声明运行环境run.py定义推理逻辑weights/目录存放权重文件。构建系统会读取这份配置产出包含正确 Python 版本、CUDA 库和全部依赖的镜像镜像内的 coglet 运行时则负责加载 Runner、暴露 HTTP 预测接口。三者缺一不可但各自职责清晰这正是模型源一词的含义——它是整个 Cog 管线的输入侧。cog.yaml声明运行时环境cog.yaml是模型源的核心配置文件。一个典型配置如下build: python_version: 3.11 gpu: true python_packages: - torch2.1.0 - transformers4.35.0 system_packages: - ffmpeg run: - curl -o /src/model.bin https://example.com/model.bin run: run.py:Runner concurrency: max: 1字段语义与默认值字段作用说明build.python_versionPython 解释器版本支持 3.103.13仓库源码 pkg/config/config.go 中DefaultPythonVersion为3.13MinimumMinorPythonVersion为 10即低于 3.10 无法构建build.gpu是否启用 CUDA 支持true时镜像会包含 CUDA 运行库并选择 GPU 基础镜像build.python_packagespip 安装的 Python 包字符串数组支持、等版本约束build.python_requirementsrequirements 文件路径当依赖较多时推荐使用例如python_requirements: requirements.txtbuild.system_packagesapt 安装的系统包例如ffmpeg、zsh等系统级依赖build.run构建期间执行的任意 shell 命令可用于下载权重、编译扩展等详见下文build.cuda/build.cudnn指定 CUDA / cuDNN 版本需要自定义 CUDA 版本时使用如cuda: 12.1build.sdk_version容器内安装的 cog Python SDK 版本PEP 440 版本字符串如0.18.0留空安装最新版可被COG_SDK_WHEEL环境变量覆盖runRunner 类路径格式module:ClassName如run.py:Runnerconcurrency.max最大并发预测数大于 1 时要求异步 Runner见下文异步预测器一节以上字段在源码中均有对应结构体pkg/config/config.go的Build结构体定义了GPU、PythonVersion、PythonRequirements、PythonPackages、Run、SystemPackages、CUDA、CuDNN、SDKVersion等字段Config结构体则持有Run即run: run.py:Runner、Concurrency、Weights等顶层配置。python_packages在注释中被标记为 deprecated推荐改用python_requirements但为向后兼容仍然支持pre_install同理是历史遗留字段。build.run 的两种写法build.run的每一项既可以是一条普通字符串命令build: run: - curl -o /src/model.bin https://example.com/model.bin也可以写成带挂载信息的映射形式用于挂载缓存目录等场景build: run: - command: pip install --cache-dir /mnt/cache . mounts: - type: cache id: pip-cache target: /mnt/cachepkg/config/config.go中的RunItem结构体专门处理这两种形态UnmarshalYAML对字符串直接取为Command对映射则解析command与mounts字段。最小可用配置实际项目中的配置往往比文档示例更精简。仓库的 hello-world 示例 只用了两个字段build: python_version: 3.12 run: run.py:RunnerRunner 类模型推理的载体Runner 是一个继承自cog.BaseRunner的 Python 类核心约定是setup()与run()两个方法from cog import BaseRunner, Input, Path class Runner(BaseRunner): def setup(self): Load model into memory. Called once at container start. self.model load_model(./weights) def run(self, prompt: str, steps: int 50) - Path: Run inference. Called for each prediction request. output self.model.generate(prompt, stepssteps) output.save(/tmp/output.png) return Path(/tmp/output.png)setup()一次性初始化调用时机容器启动时只调用一次典型用途加载模型权重、初始化 GPU 上下文、预热缓存执行顺序在 HTTP 服务器开始接收请求之前完成可省略不实现时 Cog 直接进入服务状态生命周期细节实例生命周期、并发模型、崩溃恢复与关闭流程参见 Container Runtime: Predictor Lifecycle。从源码看python/cog/predictor.py中的BaseRunner.setup()签名还接受一个可选的weights参数Optional[Union[Path, str]]可以接收本地路径或 URLhas_setup_weights()通过inspect.signature检测用户是否实现了带weights参数的 setupextract_setup_weights()则从COG_WEIGHTS环境变量读取权重来源——这是 Cog 在镜像外注入权重如 model/weightsource 实现的权重拉取能力的接口。run()每次预测的入口调用时机每个预测请求调用一次输入 schema方法参数的类型注解定义了模型的输入结构输出 schema返回类型注解定义了输出结构同步/异步可以是普通def也可以是async def完整请求链路从 HTTP 请求到响应返回的完整路径参见 Container Runtime: Life of a Prediction。module:ClassName 引用机制cog.yaml中的run: run.py:Runner由python/cog/predictor.py的load_predictor_from_ref()解析按冒号拆分为模块路径与类名若省略类名则默认查找模块中的Runner类并向后兼容Predictor类同时给出 deprecation 警告。加载时会通过_validate_runner_class()检查类必须且只能定义run()或predict()之一BasePredictor本身也已标记为BaseRunner的废弃别名新代码请统一使用BaseRunner。仓库中的 hello-world 示例 展示了最简 Runnerfrom cog import BaseRunner, Input class Runner(BaseRunner): def setup(self) - None: self.prefix hello def run(self, text: str Input(descriptionText to prefix with hello )) - str: return self.prefix text输入类型由类型注解生成输入 schemarun()参数的类型注解会被 Cog 的 schema 生成器见 pkg/schema 与 Schema 架构文档转换为 OpenAPI 输入定义。基本类型def run( self, text: str, # 字符串输入 count: int, # 整数 temperature: float, # 浮点数 verbose: bool, # 布尔 ) - str:文件输入cog.Pathcog.Path类型最强大之处在于客户端可以传 URLCog 自动下载为本地文件后交给模型。from cog import Path def run(self, image: Path) - Path: # 客户端发送: {input: {image: https://example.com/photo.jpg}} # Cog 下载 URLimage 成为本地路径如 /tmp/inputabc123.jpg img PIL.Image.open(image) ...cog.Path继承自pathlib.PosixPath其运行时行为由 python/cog/types.py 实现HTTP/HTTPS URL 会先包装成URLPath一个延迟下载的代理对象convert()方法在真正需要文件时才把内容复制到临时文件data URL 则直接解码最终模型接收到的都是本地文件系统路径。文件名处理上还有若干防御逻辑超过 200 字节的文件名会被截断并保留扩展名非法字符会被替换为下划线。密钥输入cog.Secret对不应出现在日志中的敏感值API Key、Token 等使用cog.Secretfrom cog import Secret def run(self, api_key: Secret) - str: # 日志与 webhook 中该值会被掩码 client SomeAPI(api_key.get_secret_value()) ...python/cog/types.py中的Secret是一个冻结 dataclass__str__在值存在时恒返回**********从而保证任何字符串化操作都不会泄露真实值取真实值必须显式调用get_secret_value()。输入约束Input()用Input()为参数附加元数据与校验规则from cog import Input def run( self, prompt: str Input(descriptionThe text prompt), steps: int Input(default50, ge1, le100, descriptionInference steps), style: str Input(choices[photo, art, sketch]), ) - str:参数作用description展示在 UI 与 schema 中的说明文字default未提供时的默认值必须是不可变字面量ge/le数值下界/上界大于等于/小于等于min_length/max_length字符串长度上下界regex字符串正则校验模式choices允许的取值枚举已废弃推荐用Literaldeprecated标记输入已废弃acceptPath/File 输入允许的 MIME 类型或扩展名如image/*、.safetensors,.bin语法同 HTML accept 属性这些元数据最终由python/cog/input.py中的FieldInfodataclass 承载源码确认regex、accept、deprecated也是受支持字段。值得注意的是Input()明确拒绝default_factory——可变默认值如列表应改用不可变替代如逗号分隔字符串或在run()内部构造。枚举Literalfrom typing import Literal def run( self, size: Literal[small, medium, large] medium, ) - str:列表输入from typing import List from cog import Path def run( self, images: List[Path], # 多个文件输入 tags: List[str], # 多个字符串 ) - str:可选输入from typing import Optional def run( self, seed: Optional[int] None, # 可省略或传 null ) - str:输出类型由返回注解定义产出返回类型注解定义了模型产出什么Cog 据此生成输出 schema详见 Schema 架构文档。基本类型def run(self, prompt: str) - str: return Generated text...文件输出返回指向生成文件的cog.Path运行时 Cog 会上传该文件并把 URL 返回给客户端from cog import Path def run(self, prompt: str) - Path: # 生成文件 output_path /tmp/output.png self.model.generate(prompt).save(output_path) return Path(output_path)仓库的 hello-image 示例 更简洁——直接返回仓库内已有的图片文件from cog import BaseRunner, Path class Runner(BaseRunner): def run(self) - Path: return Path(hello.webp)多文件输出from typing import List from cog import Path def run(self, prompt: str) - List[Path]: paths [] for i in range(4): path f/tmp/output_{i}.png self.model.generate(prompt, seedi).save(path) paths.append(Path(path)) return paths流式输出Iteratorrun()可以返回Iterator逐步产出结果schema 会标记为x-cog-array-type: iterator客户端通过 webhook 或流式响应按产出顺序接收from typing import Iterator def run(self, prompt: str) - Iterator[str]: for token in self.model.generate_stream(prompt): yield tokenstreaming-text 示例 给出了真实 LLM 场景的完整实现用 HuggingFaceTextIteratorStreamer配合后台线程产出 token并用streaming装饰器声明支持流式响应该装饰器在 python/cog/init.py 中定义from cog import BaseRunner, Input, streaming class Runner(BaseRunner): streaming def run( self, prompt: str Input(descriptionPrompt to complete), max_new_tokens: int Input(default128, ge1, le512), ) - Iterator[str]: ... for chunk in streamer: if chunk: yield chunk拼接式流式输出ConcatenateIterator对 LLM 风格的 token 流希望客户端把各分片拼接成完整文本而非当作列表展示时使用ConcatenateIteratorfrom cog import ConcatenateIterator def run(self, prompt: str) - ConcatenateIterator[str]: for token in self.model.generate(prompt): yield token # Hello, , world, ! # 客户端渐进看到: Hello - Hello - Hello world - Hello world!schema 会包含x-cog-array-display: concatenate标记提示输出应拼接而非列表化。ConcatenateIterator在 python/cog/types.py 中定义另有异步版本AsyncConcatenateIterator供 async run 使用。权重加载打包进镜像还是运行时下载模型权重有两条主流加载路径打包进镜像把权重放在源目录中构建时会被复制进镜像my-model/ ├── cog.yaml ├── run.py └── weights/ └── model.safetensorsdef setup(self): self.model load(./weights/model.safetensors)运行时下载在setup()中按需拉取而非打包。常见方式使用 pgetCog 镜像内置的并行下载工具import subprocess def setup(self): subprocess.run([pget, https://example.com/model.tar, ./weights]) self.model load(./weights/model.safetensors)setup 内直接下载def setup(self): # 使用 requests、huggingface_hub 或任何其他方式 snapshot_download(repo_idmeta-llama/Llama-2-7b, local_dir./weights) self.model load(./weights)两种方式的选择取决于部署需求打包权重使镜像更大但启动更快运行时下载保持镜像小巧但要求启动时有网络。仓库还提供更工程化的权重管理方案pkg/weights目录实现了基于锁文件的权重拉取与漂移检测weights/lockfile、weights/manager.gopkg/model/weightsource则实现了从 Hugging Face 等来源导入权重的解析与指纹校验适合对权重可复现性要求较高的场景。异步预测器与并发要实现并发预测run()与setup()都可改为 asyncclass Runner(BaseRunner): async def setup(self): self.model await load_model_async() async def run(self, prompt: str) - str: return await self.model.generate(prompt)前置条件Python 3.11源码 pkg/config/config.go 中MinimumMinorPythonVersionForConcurrency常量即 11cog.yaml中设置concurrency.max 1并发细节参见 Container Runtime。更细粒度的控制可用concurrent(maxN)装饰器python/cog/init.py 中定义。源码明确校验max必须是正整数且max 1时被装饰的函数必须是协程函数或异步生成器否则直接抛出TypeError。仓库的 hello-concurrency 示例 展示了组合用法——异步setupconcurrent(max4)的异步生成器 run并通过current_scope().record_metric()上报指标from cog import BaseRunner, Input, concurrent, current_scope class Runner(BaseRunner): async def setup(self) - None: ... concurrent(max4) async def run( self, total: int Input(default5), interval: int Input(default3), ) - AsyncConcatenateIterator[str]: for index, fruit in enumerate(fruits): ... yield f{fruit}\n await asyncio.sleep(interval) current_scope().record_metric(output_tokens, total)源码参考索引文件作用python/cog/init.py公共 API 导出BaseRunner、Input、Path、Secret、ConcatenateIterator、streaming、concurrent、current_scopepython/cog/predictor.pyBaseRunner类、module:ClassName引用加载、setup weights 检测python/cog/types.pyPath、Secret、URLPath、ConcatenateIterator、AsyncConcatenateIteratorpython/cog/input.pyInput()函数与FieldInfo元数据pkg/config/config.gocog.yaml 解析与Build/Config结构体examples/hello-world最简 Runner 示例examples/hello-imagePath文件输出示例examples/streaming-textIterator流式输出与streaming示例examples/hello-concurrency异步 concurrent(max4)并发示例更进一步架构总览 提供了模型源在整个 Cog 体系中的定位Schema 解释了类型注解如何变成 OpenAPI 契约Prediction API 定义了 HTTP 请求/响应信封格式Container Runtime 说明了 Runner 在容器内的完整生命周期CLI 则介绍如何用命令行构建、测试与部署这些模型。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Cog Model Source 详解：模型作者如何用 cog.yaml、Runner 与类型注解定义机器学习模型
Cog Model Source 详解模型作者如何用 cog.yaml、Runner 与类型注解定义机器学习模型【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog本篇指南围绕 Cog 的模型源Model Source展开模型作者只需提供一个cog.yaml环境配置、一个包含setup()/run()方法的 Runner 类以及可选的模型权重Cog 就能把它构建成生产可用的 OCI 容器镜像并通过 架构总览 中描述的 CLI、Python SDK 与 coglet 运行时对外提供 HTTP 预测服务。读完本文你将掌握cog.yaml每个配置字段的真实语义、Runner 类的生命周期约定、基于类型注解的输入输出 schema 契约以及文件、密钥、流式输出与异步并发等进阶用法并能在仓库源码层面印证每一项行为。模型作者提供什么一个最小模型的三要素Cog 的设计哲学是用标准 Python 定义模型。一个完整的 Cog 模型在磁盘上只有三个组成部分my-model/ ├── cog.yaml # 运行时环境配置 ├── run.py # Runner 类模型加载与推理逻辑 └── weights/ # 模型权重可选也可以运行时下载其中cog.yaml声明运行环境run.py定义推理逻辑weights/目录存放权重文件。构建系统会读取这份配置产出包含正确 Python 版本、CUDA 库和全部依赖的镜像镜像内的 coglet 运行时则负责加载 Runner、暴露 HTTP 预测接口。三者缺一不可但各自职责清晰这正是模型源一词的含义——它是整个 Cog 管线的输入侧。cog.yaml声明运行时环境cog.yaml是模型源的核心配置文件。一个典型配置如下build: python_version: 3.11 gpu: true python_packages: - torch2.1.0 - transformers4.35.0 system_packages: - ffmpeg run: - curl -o /src/model.bin https://example.com/model.bin run: run.py:Runner concurrency: max: 1字段语义与默认值字段作用说明build.python_versionPython 解释器版本支持 3.103.13仓库源码 pkg/config/config.go 中DefaultPythonVersion为3.13MinimumMinorPythonVersion为 10即低于 3.10 无法构建build.gpu是否启用 CUDA 支持true时镜像会包含 CUDA 运行库并选择 GPU 基础镜像build.python_packagespip 安装的 Python 包字符串数组支持、等版本约束build.python_requirementsrequirements 文件路径当依赖较多时推荐使用例如python_requirements: requirements.txtbuild.system_packagesapt 安装的系统包例如ffmpeg、zsh等系统级依赖build.run构建期间执行的任意 shell 命令可用于下载权重、编译扩展等详见下文build.cuda/build.cudnn指定 CUDA / cuDNN 版本需要自定义 CUDA 版本时使用如cuda: 12.1build.sdk_version容器内安装的 cog Python SDK 版本PEP 440 版本字符串如0.18.0留空安装最新版可被COG_SDK_WHEEL环境变量覆盖runRunner 类路径格式module:ClassName如run.py:Runnerconcurrency.max最大并发预测数大于 1 时要求异步 Runner见下文异步预测器一节以上字段在源码中均有对应结构体pkg/config/config.go的Build结构体定义了GPU、PythonVersion、PythonRequirements、PythonPackages、Run、SystemPackages、CUDA、CuDNN、SDKVersion等字段Config结构体则持有Run即run: run.py:Runner、Concurrency、Weights等顶层配置。python_packages在注释中被标记为 deprecated推荐改用python_requirements但为向后兼容仍然支持pre_install同理是历史遗留字段。build.run 的两种写法build.run的每一项既可以是一条普通字符串命令build: run: - curl -o /src/model.bin https://example.com/model.bin也可以写成带挂载信息的映射形式用于挂载缓存目录等场景build: run: - command: pip install --cache-dir /mnt/cache . mounts: - type: cache id: pip-cache target: /mnt/cachepkg/config/config.go中的RunItem结构体专门处理这两种形态UnmarshalYAML对字符串直接取为Command对映射则解析command与mounts字段。最小可用配置实际项目中的配置往往比文档示例更精简。仓库的 hello-world 示例 只用了两个字段build: python_version: 3.12 run: run.py:RunnerRunner 类模型推理的载体Runner 是一个继承自cog.BaseRunner的 Python 类核心约定是setup()与run()两个方法from cog import BaseRunner, Input, Path class Runner(BaseRunner): def setup(self): Load model into memory. Called once at container start. self.model load_model(./weights) def run(self, prompt: str, steps: int 50) - Path: Run inference. Called for each prediction request. output self.model.generate(prompt, stepssteps) output.save(/tmp/output.png) return Path(/tmp/output.png)setup()一次性初始化调用时机容器启动时只调用一次典型用途加载模型权重、初始化 GPU 上下文、预热缓存执行顺序在 HTTP 服务器开始接收请求之前完成可省略不实现时 Cog 直接进入服务状态生命周期细节实例生命周期、并发模型、崩溃恢复与关闭流程参见 Container Runtime: Predictor Lifecycle。从源码看python/cog/predictor.py中的BaseRunner.setup()签名还接受一个可选的weights参数Optional[Union[Path, str]]可以接收本地路径或 URLhas_setup_weights()通过inspect.signature检测用户是否实现了带weights参数的 setupextract_setup_weights()则从COG_WEIGHTS环境变量读取权重来源——这是 Cog 在镜像外注入权重如 model/weightsource 实现的权重拉取能力的接口。run()每次预测的入口调用时机每个预测请求调用一次输入 schema方法参数的类型注解定义了模型的输入结构输出 schema返回类型注解定义了输出结构同步/异步可以是普通def也可以是async def完整请求链路从 HTTP 请求到响应返回的完整路径参见 Container Runtime: Life of a Prediction。module:ClassName 引用机制cog.yaml中的run: run.py:Runner由python/cog/predictor.py的load_predictor_from_ref()解析按冒号拆分为模块路径与类名若省略类名则默认查找模块中的Runner类并向后兼容Predictor类同时给出 deprecation 警告。加载时会通过_validate_runner_class()检查类必须且只能定义run()或predict()之一BasePredictor本身也已标记为BaseRunner的废弃别名新代码请统一使用BaseRunner。仓库中的 hello-world 示例 展示了最简 Runnerfrom cog import BaseRunner, Input class Runner(BaseRunner): def setup(self) - None: self.prefix hello def run(self, text: str Input(descriptionText to prefix with hello )) - str: return self.prefix text输入类型由类型注解生成输入 schemarun()参数的类型注解会被 Cog 的 schema 生成器见 pkg/schema 与 Schema 架构文档转换为 OpenAPI 输入定义。基本类型def run( self, text: str, # 字符串输入 count: int, # 整数 temperature: float, # 浮点数 verbose: bool, # 布尔 ) - str:文件输入cog.Pathcog.Path类型最强大之处在于客户端可以传 URLCog 自动下载为本地文件后交给模型。from cog import Path def run(self, image: Path) - Path: # 客户端发送: {input: {image: https://example.com/photo.jpg}} # Cog 下载 URLimage 成为本地路径如 /tmp/inputabc123.jpg img PIL.Image.open(image) ...cog.Path继承自pathlib.PosixPath其运行时行为由 python/cog/types.py 实现HTTP/HTTPS URL 会先包装成URLPath一个延迟下载的代理对象convert()方法在真正需要文件时才把内容复制到临时文件data URL 则直接解码最终模型接收到的都是本地文件系统路径。文件名处理上还有若干防御逻辑超过 200 字节的文件名会被截断并保留扩展名非法字符会被替换为下划线。密钥输入cog.Secret对不应出现在日志中的敏感值API Key、Token 等使用cog.Secretfrom cog import Secret def run(self, api_key: Secret) - str: # 日志与 webhook 中该值会被掩码 client SomeAPI(api_key.get_secret_value()) ...python/cog/types.py中的Secret是一个冻结 dataclass__str__在值存在时恒返回**********从而保证任何字符串化操作都不会泄露真实值取真实值必须显式调用get_secret_value()。输入约束Input()用Input()为参数附加元数据与校验规则from cog import Input def run( self, prompt: str Input(descriptionThe text prompt), steps: int Input(default50, ge1, le100, descriptionInference steps), style: str Input(choices[photo, art, sketch]), ) - str:参数作用description展示在 UI 与 schema 中的说明文字default未提供时的默认值必须是不可变字面量ge/le数值下界/上界大于等于/小于等于min_length/max_length字符串长度上下界regex字符串正则校验模式choices允许的取值枚举已废弃推荐用Literaldeprecated标记输入已废弃acceptPath/File 输入允许的 MIME 类型或扩展名如image/*、.safetensors,.bin语法同 HTML accept 属性这些元数据最终由python/cog/input.py中的FieldInfodataclass 承载源码确认regex、accept、deprecated也是受支持字段。值得注意的是Input()明确拒绝default_factory——可变默认值如列表应改用不可变替代如逗号分隔字符串或在run()内部构造。枚举Literalfrom typing import Literal def run( self, size: Literal[small, medium, large] medium, ) - str:列表输入from typing import List from cog import Path def run( self, images: List[Path], # 多个文件输入 tags: List[str], # 多个字符串 ) - str:可选输入from typing import Optional def run( self, seed: Optional[int] None, # 可省略或传 null ) - str:输出类型由返回注解定义产出返回类型注解定义了模型产出什么Cog 据此生成输出 schema详见 Schema 架构文档。基本类型def run(self, prompt: str) - str: return Generated text...文件输出返回指向生成文件的cog.Path运行时 Cog 会上传该文件并把 URL 返回给客户端from cog import Path def run(self, prompt: str) - Path: # 生成文件 output_path /tmp/output.png self.model.generate(prompt).save(output_path) return Path(output_path)仓库的 hello-image 示例 更简洁——直接返回仓库内已有的图片文件from cog import BaseRunner, Path class Runner(BaseRunner): def run(self) - Path: return Path(hello.webp)多文件输出from typing import List from cog import Path def run(self, prompt: str) - List[Path]: paths [] for i in range(4): path f/tmp/output_{i}.png self.model.generate(prompt, seedi).save(path) paths.append(Path(path)) return paths流式输出Iteratorrun()可以返回Iterator逐步产出结果schema 会标记为x-cog-array-type: iterator客户端通过 webhook 或流式响应按产出顺序接收from typing import Iterator def run(self, prompt: str) - Iterator[str]: for token in self.model.generate_stream(prompt): yield tokenstreaming-text 示例 给出了真实 LLM 场景的完整实现用 HuggingFaceTextIteratorStreamer配合后台线程产出 token并用streaming装饰器声明支持流式响应该装饰器在 python/cog/init.py 中定义from cog import BaseRunner, Input, streaming class Runner(BaseRunner): streaming def run( self, prompt: str Input(descriptionPrompt to complete), max_new_tokens: int Input(default128, ge1, le512), ) - Iterator[str]: ... for chunk in streamer: if chunk: yield chunk拼接式流式输出ConcatenateIterator对 LLM 风格的 token 流希望客户端把各分片拼接成完整文本而非当作列表展示时使用ConcatenateIteratorfrom cog import ConcatenateIterator def run(self, prompt: str) - ConcatenateIterator[str]: for token in self.model.generate(prompt): yield token # Hello, , world, ! # 客户端渐进看到: Hello - Hello - Hello world - Hello world!schema 会包含x-cog-array-display: concatenate标记提示输出应拼接而非列表化。ConcatenateIterator在 python/cog/types.py 中定义另有异步版本AsyncConcatenateIterator供 async run 使用。权重加载打包进镜像还是运行时下载模型权重有两条主流加载路径打包进镜像把权重放在源目录中构建时会被复制进镜像my-model/ ├── cog.yaml ├── run.py └── weights/ └── model.safetensorsdef setup(self): self.model load(./weights/model.safetensors)运行时下载在setup()中按需拉取而非打包。常见方式使用 pgetCog 镜像内置的并行下载工具import subprocess def setup(self): subprocess.run([pget, https://example.com/model.tar, ./weights]) self.model load(./weights/model.safetensors)setup 内直接下载def setup(self): # 使用 requests、huggingface_hub 或任何其他方式 snapshot_download(repo_idmeta-llama/Llama-2-7b, local_dir./weights) self.model load(./weights)两种方式的选择取决于部署需求打包权重使镜像更大但启动更快运行时下载保持镜像小巧但要求启动时有网络。仓库还提供更工程化的权重管理方案pkg/weights目录实现了基于锁文件的权重拉取与漂移检测weights/lockfile、weights/manager.gopkg/model/weightsource则实现了从 Hugging Face 等来源导入权重的解析与指纹校验适合对权重可复现性要求较高的场景。异步预测器与并发要实现并发预测run()与setup()都可改为 asyncclass Runner(BaseRunner): async def setup(self): self.model await load_model_async() async def run(self, prompt: str) - str: return await self.model.generate(prompt)前置条件Python 3.11源码 pkg/config/config.go 中MinimumMinorPythonVersionForConcurrency常量即 11cog.yaml中设置concurrency.max 1并发细节参见 Container Runtime。更细粒度的控制可用concurrent(maxN)装饰器python/cog/init.py 中定义。源码明确校验max必须是正整数且max 1时被装饰的函数必须是协程函数或异步生成器否则直接抛出TypeError。仓库的 hello-concurrency 示例 展示了组合用法——异步setupconcurrent(max4)的异步生成器 run并通过current_scope().record_metric()上报指标from cog import BaseRunner, Input, concurrent, current_scope class Runner(BaseRunner): async def setup(self) - None: ... concurrent(max4) async def run( self, total: int Input(default5), interval: int Input(default3), ) - AsyncConcatenateIterator[str]: for index, fruit in enumerate(fruits): ... yield f{fruit}\n await asyncio.sleep(interval) current_scope().record_metric(output_tokens, total)源码参考索引文件作用python/cog/init.py公共 API 导出BaseRunner、Input、Path、Secret、ConcatenateIterator、streaming、concurrent、current_scopepython/cog/predictor.pyBaseRunner类、module:ClassName引用加载、setup weights 检测python/cog/types.pyPath、Secret、URLPath、ConcatenateIterator、AsyncConcatenateIteratorpython/cog/input.pyInput()函数与FieldInfo元数据pkg/config/config.gocog.yaml 解析与Build/Config结构体examples/hello-world最简 Runner 示例examples/hello-imagePath文件输出示例examples/streaming-textIterator流式输出与streaming示例examples/hello-concurrency异步 concurrent(max4)并发示例更进一步架构总览 提供了模型源在整个 Cog 体系中的定位Schema 解释了类型注解如何变成 OpenAPI 契约Prediction API 定义了 HTTP 请求/响应信封格式Container Runtime 说明了 Runner 在容器内的完整生命周期CLI 则介绍如何用命令行构建、测试与部署这些模型。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考