尧图网站建设 
Cog 并发预测实战用concurrent装饰器与 OpenTelemetry 构建高吞吐推理服务【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/coghello-concurrency是 Cog 仓库中的官方示例项目它以最小可运行的形式演示了 Cog 的两个关键能力通过concurrent(maxN)装饰器让异步run()方法支持并发预测以及通过 OpenTelemetryOTLP协议采集模型遥测数据。读完本文你将掌握 Cog 并发模型的核心规则、409 Conflict 限流机制背后的实现原理以及如何在自有模型中接入 OpenTelemetry 遥测并配置 Honeycomb 数据源。示例项目一览examples/hello-concurrency是一个结构极其精简的 Cog 项目完整文件只有三个cog.yaml构建与运行配置run.py定义 Runner 与并发/遥测逻辑requirements.txtOpenTelemetry 依赖声明其中 cog.yaml 的内容如下build: gpu: false python_version: 3.12 python_requirements: requirements.txt run: run.py:Runner三个关键配置点分别是gpu: false表示该模型不需要 GPUpython_version: 3.12指定 Python 运行时版本run: run.py:Runner声明由run.py中的Runner类提供预测接口。由于本项目无需加载权重run.py 中的setup()只负责初始化遥测并记录耗时。核心机制concurrent装饰器基本用法示例文档指出整个项目的“关键部件”key piece是加在异步run()方法上的concurrent(max4)装饰器from cog import BaseRunner, concurrent class Runner(BaseRunner): concurrent(max4) async def run(self) - str: return hello这段代码声明该模型的预测方法最多允许 4 个预测同时执行。当并发请求超过该阈值时Cog 会拒绝后续预测请求并返回409 Conflict响应。源码级的装饰器实现concurrent装饰器并非魔法其实现在 python/cog/init.py 中清晰可见核心逻辑如下def concurrent( fn: _F | None None, *, max: int 1, # noqa: A002 ) - _F | Callable[[_F], _F]: Configure the maximum concurrency for an async predict handler. if isinstance(max, bool) or not isinstance(max, int): raise TypeError(concurrent max must be an integer) if max 1: raise ValueError(concurrent max must be at least 1) def decorate(inner: _F) - _F: if max 1 and not ( _inspect.iscoroutinefunction(inner) or _inspect.isasyncgenfunction(inner) ): raise TypeError(concurrent max greater than 1 requires an async function) inner.__cog_concurrent_max__ max # type: ignore[attr-defined] return inner if fn is None: return decorate if not callable(fn): raise TypeError( concurrent must be used as concurrent or concurrent(max...) ) return decorate(fn)从这段实现中可以提炼出三条硬性规则max必须是正整数max为bool或非整数会抛出TypeError小于 1 会抛出ValueError。max 1时run()必须是异步函数装饰器通过inspect.iscoroutinefunction或inspect.isasyncgenfunction检查同步函数配max 1会直接报错——因为并发预测依赖事件循环在 await 点上切换同步阻塞代码无法实现真正的并发。并发上限会被标记在函数对象上装饰器把max写入inner.__cog_concurrent_max__属性后续由运行时读取该属性完成并发槽位slot分配。运行时覆盖与动态配置官方文档 docs/python.md 对并发做了更完整的说明异步 Runner 在 cog 0.14.0 中加入cog.concurrent装饰器在 cog 0.21.0 中加入。需要注意两点限制max必须是整数字面量这样 Cog 才能在构建期build time就确定模型的并发配置如果需要运行时动态调整并发应使用COG_MAX_CONCURRENCY环境变量它可以覆盖装饰器中的取值。COG_MAX_CONCURRENCY的解析逻辑位于 crates/coglet-python/src/lib.rs读取环境变量后用parse_max_concurrency解析若值非法则记录告警并回退为默认值 1。这一设计让运维层面可以在不重建镜像的前提下调整并发上限例如按机器规格CPU 核数、显存弹性配置。异步 Runner 与并发调度原理BaseRunner 的 async 化examples/hello-concurrency/run.py 同时把setup()与run()都声明为异步方法。其基类定义在 python/cog/predictor.pyBaseRunner的setup()负责一次性加载模型权重等准备工作run()执行单次预测该文件同时保留了BasePredictor作为弃用的兼容别名。在示例中setup()的执行内容为开启一个名为setup的 trace span、time.sleep(1)模拟耗时初始化、把耗时以model.setup_time_seconds属性写入 span。异步化之后事件循环不会因为某个预测的 await 阻塞而停下整个进程多个并发预测可以共享同一个事件循环交错推进。并发槽位与 permit 池服务端如何落实“最多 N 个并发”答案在 Rust 实现的 coglet 运行时中。crates/coglet/src/permit/目录实现了一个类型化typestate的 permit 池crates/coglet/src/permit/mod.rs 的模块注释明确描述了状态机PermitInUse → PermitIdle预测完成permit 归还池中drop 时自动归还PermitInUse → PermitPoisoned出现异常permit 被孤立PermitPoisoned → PermitIdle不允许——被毒化的槽位会从池中永久移除也就是说每个并发槽位对应一个 permit预测在槽位内运行结束后自动释放异常槽位则被标记淘汰避免坏状态被复用污染后续预测。达到上限时返回 409 Conflict当所有槽位都被占满时新的预测请求会被拒绝。HTTP 层的行为定义在 crates/coglet/src/transport/http/routes.rsErr(CreatePredictionError::AtCapacity) { return ( StatusCode::CONFLICT, Json(serde_json::json!({ error: At capacity - all prediction slots busy, status: failed })), ) .into_response(); }即返回 HTTP409 Conflict响应体为{error: At capacity - all prediction slots busy, status: failed}。同文件中的prediction_at_capacity单元测试routes.rs通过只创建 1 个槽位、用永不完成的 mock 预测占满槽位验证第二个请求确实收到 409 且错误信息包含 “capacity”。深入示例流式输出与指标记录examples/hello-concurrency/run.py 的Runner完整展示了并发 流式输出 指标记录的组合用法concurrent(max4) async def run( # pyright: ignore self, total: int Input(default5), interval: int Input(default3), ) - AsyncConcatenateIterator[str]: # pyright: ignore links [] if setup_context : getattr(self, _setup_context, None): links.append(trace.Link(setup_context)) with tracer.start_as_current_span(predict, linkslinks) as span: span.set_attribute(inputs.total, total) span.set_attribute(inputs.interval, interval) start_time time.time() logging.info( fstarting prediction: cog_version{__version__} total{total} interval{interval} ) fruits [ Apple, Banana, Orange, Grape, Strawberry, Mango, Pineapple, Blueberry, Watermelon, Peach, ][:total] for index, fruit in enumerate(fruits): if index 1 total: yield f{fruit} else: yield f{fruit}\n logging.info(foutput fruit: {fruit}) await asyncio.sleep(interval) logging.info(femit_metric: output_tokens{total}) current_scope().record_metric(output_tokens, total) span.set_attribute(metrics.output_tokens, total) duration time.time() - start_time logging.info(fcompleted prediction in {duration} seconds) span.set_attribute(model.predict_time_seconds, duration)几个值得注意的细节输入定义total输出水果数量默认 5与interval每个输出的间隔秒数默认 3都通过Input(default...)声明Cog 会自动为其生成 OpenAPI Schema。AsyncConcatenateIterator[str]流式输出run()是异步生成器用yield逐行吐数据await asyncio.sleep(interval)制造可控延迟。返回类型为AsyncConcatenateIterator[str]该类型用于声明“逐块拼接的流式字符串输出”。并发 流式意味着 4 个预测可以同时各自流式产出互不阻塞。span 关联setupspan 的上下文被保存为_setup_context预测 span 通过trace.Link(setup_context)与之关联从而把初始化阶段与预测阶段串联成完整 trace。指标记录current_scope().record_metric(output_tokens, total)把自定义指标写入当前预测的作用域。current_scope的实现位于 crates/coglet-python/src/metric_scope.rs每个预测持有独立的Scope并通过 ContextVar 路由确保并发场景下指标不会串到别的预测上。关于并发场景下的指标隔离integration-tests/concurrent/concurrent_test.go 中的TestConcurrentAsyncMetrics专门验证了这一点同时发出 5 个并发预测断言每个预测响应中的prediction_index指标都等于自己的索引防止了“指标被静默丢弃”或“指标张冠李戴”两种并发缺陷。集成测试如何验证并发行为integration-tests/concurrent/concurrent_test.go是理解并发语义的最佳实证材料它包含两个核心用例TestConcurrentPredictionsconcurrent_test.go使用concurrent(max5)的异步预测器5 个 goroutine 同时发起 5 个预测请求每个内部sleep 1s随后立即向/shutdown发送关闭请求。断言所有预测在 3 秒内完成——如果 Cog 串行执行 5 个 1 秒任务需要 5 秒3 秒内完成即证明预测确实是并发执行的同时验证服务端优雅关闭会等待在途预测跑完。TestConcurrentAboveLimitconcurrent_test.gocog.yaml中配置concurrency: {max: 2}先发 2 个长期预测占满槽位再轮询发送溢出请求断言其返回409 Conflict且响应体status为failed、error包含 “capacity” 字样。顺带一提concurrency.max是 YAML 层面的并发配置方式与装饰器等效两个入口最终都会收敛到同一套 permit 池机制。这也解释了示例文档中“如果 Cog 达到最大并发阈值将返回 409 Conflict”的完整链路装饰器标记__cog_concurrent_max__→ coglet 据此初始化并发槽位 → 槽位耗尽时 HTTP 路由返回 409。Telemetry用 OpenTelemetry 采集模型遥测示例的第二大主题是遥测Telemetry。它使用opentelemetry相关依赖见 requirements.txt包含opentelemetry-api、opentelemetry-sdk、opentelemetry-exporter-otlp-proto-http演示如何为模型采集可观测数据。遥测初始化流程run.py 的初始化逻辑分三步第一步读取 Honeycomb token 文件。要求镜像构建目录中存在名为honeycomb_token.key的文件内容为 API tokenhoneycomb_token try: with open(./honeycomb_token.key, r) as f: honeycomb_token f.read().strip() except FileNotFoundError: logging.info(honeycomb_token.key not found; OTEL will be disabled) if not honeycomb_token: os.environ[OTEL_SDK_DISABLED] true文件缺失时不会报错而是通过OTEL_SDK_DISABLEDtrue优雅地禁用 SDK。第二步通过环境变量配置 OTLP 导出目标。三个关键环境变量分别是环境变量示例值作用OTEL_EXPORTER_OTLP_ENDPOINThttps://api.honeycomb.io/指定 OTLP 导出端点使用自定义采集器时改为自己的地址OTEL_EXPORTER_OTLP_HEADERSx-honeycomb-teamtoken携带 Honeycomb 的团队认证头OTEL_SERVICE_NAMEcog-model服务名即数据在 Honeycomb 中的归属数据源示例文档特别说明事件会被发送到名为cog-model的数据源你可以通过修改OTEL_SERVICE_NAME来更换数据源名称若使用自定义端点则通过OTEL_EXPORTER_OTLP_ENDPOINT配置。第三步创建 TracerProvider 并注册导出器。resource Resource( attributes{model.name: replicate/hello-concurrency, cog_version: __version__} ) provider TracerProvider(resourceresource) provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(provider) tracer trace.get_tracer(predict)Resource携带model.name与cog_version两个资源属性所有 span 都会继承它们BatchSpanProcessor负责把 span 批量异步发送到 OTLP 端点。之后通过tracer.start_as_current_span(...)在setup与run中创建 span并写入model.setup_time_seconds、inputs.total、metrics.output_tokens、model.predict_time_seconds等属性。本地调试模式示例文档还提到run.py末尾有一段可取消注释的本地调试代码# Local OTEL debugging # from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor # os.environ[OTEL_EXPORTER_OTLP_ENDPOINT] http://otel-collector.local-otel.orb.local:4318 # os.environ[OTEL_SDK_DISABLED] # provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))这段代码演示了两种本地调试手段一是把导出端点改为本地 OpenTelemetry Collector 地址二是用SimpleSpanProcessor ConsoleSpanExporter把 span 直接打印到控制台——后者无需任何外部服务是最快的验证方式。注意这里演示的是把OTEL_SDK_DISABLED重新置空以恢复 SDK 功能。构建与运行在仓库根目录执行以下命令即可体验完整流程# 构建镜像需在包含 cog.yaml 的项目目录中 cog build # 以 HTTP 服务方式运行默认监听 5000 端口 cog serve # 并发测试同时发起多个预测请求 curl -X POST http://localhost:5000/predictions \ -H Content-Type: application/json \ -d {input: {total: 5, interval: 1}}由于concurrent(max4)最多 4 个预测并行执行第 5 个并发请求会收到409 Conflict。若想运行时调整上限而不重建镜像可设置COG_MAX_CONCURRENCY环境变量覆盖装饰器取值该行为同样记录于 docs/python.md。需要生产级遥测时在项目目录放入honeycomb_token.key后重新构建镜像即可把 span 数据推送到 Honeycomb 的cog-model数据源。小结hello-concurrency用不到 130 行代码串起了 Cog 并发生态的完整链路concurrent(maxN)装饰器负责声明并发上限python/cog/init.pycoglet 的 permit 池负责并发槽位的分配与毒化回收crates/coglet/src/permit/mod.rsHTTP 路由在容量耗尽时返回 409 Conflictcrates/coglet/src/transport/http/routes.rs而current_scope().record_metric与 OpenTelemetry 配合让每个并发预测的指标和 trace 都能被准确归因与采集。无论你是要构建高吞吐的推理 API还是想为模型接入完整的可观测体系这个示例都是可以直接照抄的起点模板。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Cog 并发预测实战：用 `@concurrent` 装饰器与 OpenTelemetry 构建高吞吐推理服务
Cog 并发预测实战用concurrent装饰器与 OpenTelemetry 构建高吞吐推理服务【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/coghello-concurrency是 Cog 仓库中的官方示例项目它以最小可运行的形式演示了 Cog 的两个关键能力通过concurrent(maxN)装饰器让异步run()方法支持并发预测以及通过 OpenTelemetryOTLP协议采集模型遥测数据。读完本文你将掌握 Cog 并发模型的核心规则、409 Conflict 限流机制背后的实现原理以及如何在自有模型中接入 OpenTelemetry 遥测并配置 Honeycomb 数据源。示例项目一览examples/hello-concurrency是一个结构极其精简的 Cog 项目完整文件只有三个cog.yaml构建与运行配置run.py定义 Runner 与并发/遥测逻辑requirements.txtOpenTelemetry 依赖声明其中 cog.yaml 的内容如下build: gpu: false python_version: 3.12 python_requirements: requirements.txt run: run.py:Runner三个关键配置点分别是gpu: false表示该模型不需要 GPUpython_version: 3.12指定 Python 运行时版本run: run.py:Runner声明由run.py中的Runner类提供预测接口。由于本项目无需加载权重run.py 中的setup()只负责初始化遥测并记录耗时。核心机制concurrent装饰器基本用法示例文档指出整个项目的“关键部件”key piece是加在异步run()方法上的concurrent(max4)装饰器from cog import BaseRunner, concurrent class Runner(BaseRunner): concurrent(max4) async def run(self) - str: return hello这段代码声明该模型的预测方法最多允许 4 个预测同时执行。当并发请求超过该阈值时Cog 会拒绝后续预测请求并返回409 Conflict响应。源码级的装饰器实现concurrent装饰器并非魔法其实现在 python/cog/init.py 中清晰可见核心逻辑如下def concurrent( fn: _F | None None, *, max: int 1, # noqa: A002 ) - _F | Callable[[_F], _F]: Configure the maximum concurrency for an async predict handler. if isinstance(max, bool) or not isinstance(max, int): raise TypeError(concurrent max must be an integer) if max 1: raise ValueError(concurrent max must be at least 1) def decorate(inner: _F) - _F: if max 1 and not ( _inspect.iscoroutinefunction(inner) or _inspect.isasyncgenfunction(inner) ): raise TypeError(concurrent max greater than 1 requires an async function) inner.__cog_concurrent_max__ max # type: ignore[attr-defined] return inner if fn is None: return decorate if not callable(fn): raise TypeError( concurrent must be used as concurrent or concurrent(max...) ) return decorate(fn)从这段实现中可以提炼出三条硬性规则max必须是正整数max为bool或非整数会抛出TypeError小于 1 会抛出ValueError。max 1时run()必须是异步函数装饰器通过inspect.iscoroutinefunction或inspect.isasyncgenfunction检查同步函数配max 1会直接报错——因为并发预测依赖事件循环在 await 点上切换同步阻塞代码无法实现真正的并发。并发上限会被标记在函数对象上装饰器把max写入inner.__cog_concurrent_max__属性后续由运行时读取该属性完成并发槽位slot分配。运行时覆盖与动态配置官方文档 docs/python.md 对并发做了更完整的说明异步 Runner 在 cog 0.14.0 中加入cog.concurrent装饰器在 cog 0.21.0 中加入。需要注意两点限制max必须是整数字面量这样 Cog 才能在构建期build time就确定模型的并发配置如果需要运行时动态调整并发应使用COG_MAX_CONCURRENCY环境变量它可以覆盖装饰器中的取值。COG_MAX_CONCURRENCY的解析逻辑位于 crates/coglet-python/src/lib.rs读取环境变量后用parse_max_concurrency解析若值非法则记录告警并回退为默认值 1。这一设计让运维层面可以在不重建镜像的前提下调整并发上限例如按机器规格CPU 核数、显存弹性配置。异步 Runner 与并发调度原理BaseRunner 的 async 化examples/hello-concurrency/run.py 同时把setup()与run()都声明为异步方法。其基类定义在 python/cog/predictor.pyBaseRunner的setup()负责一次性加载模型权重等准备工作run()执行单次预测该文件同时保留了BasePredictor作为弃用的兼容别名。在示例中setup()的执行内容为开启一个名为setup的 trace span、time.sleep(1)模拟耗时初始化、把耗时以model.setup_time_seconds属性写入 span。异步化之后事件循环不会因为某个预测的 await 阻塞而停下整个进程多个并发预测可以共享同一个事件循环交错推进。并发槽位与 permit 池服务端如何落实“最多 N 个并发”答案在 Rust 实现的 coglet 运行时中。crates/coglet/src/permit/目录实现了一个类型化typestate的 permit 池crates/coglet/src/permit/mod.rs 的模块注释明确描述了状态机PermitInUse → PermitIdle预测完成permit 归还池中drop 时自动归还PermitInUse → PermitPoisoned出现异常permit 被孤立PermitPoisoned → PermitIdle不允许——被毒化的槽位会从池中永久移除也就是说每个并发槽位对应一个 permit预测在槽位内运行结束后自动释放异常槽位则被标记淘汰避免坏状态被复用污染后续预测。达到上限时返回 409 Conflict当所有槽位都被占满时新的预测请求会被拒绝。HTTP 层的行为定义在 crates/coglet/src/transport/http/routes.rsErr(CreatePredictionError::AtCapacity) { return ( StatusCode::CONFLICT, Json(serde_json::json!({ error: At capacity - all prediction slots busy, status: failed })), ) .into_response(); }即返回 HTTP409 Conflict响应体为{error: At capacity - all prediction slots busy, status: failed}。同文件中的prediction_at_capacity单元测试routes.rs通过只创建 1 个槽位、用永不完成的 mock 预测占满槽位验证第二个请求确实收到 409 且错误信息包含 “capacity”。深入示例流式输出与指标记录examples/hello-concurrency/run.py 的Runner完整展示了并发 流式输出 指标记录的组合用法concurrent(max4) async def run( # pyright: ignore self, total: int Input(default5), interval: int Input(default3), ) - AsyncConcatenateIterator[str]: # pyright: ignore links [] if setup_context : getattr(self, _setup_context, None): links.append(trace.Link(setup_context)) with tracer.start_as_current_span(predict, linkslinks) as span: span.set_attribute(inputs.total, total) span.set_attribute(inputs.interval, interval) start_time time.time() logging.info( fstarting prediction: cog_version{__version__} total{total} interval{interval} ) fruits [ Apple, Banana, Orange, Grape, Strawberry, Mango, Pineapple, Blueberry, Watermelon, Peach, ][:total] for index, fruit in enumerate(fruits): if index 1 total: yield f{fruit} else: yield f{fruit}\n logging.info(foutput fruit: {fruit}) await asyncio.sleep(interval) logging.info(femit_metric: output_tokens{total}) current_scope().record_metric(output_tokens, total) span.set_attribute(metrics.output_tokens, total) duration time.time() - start_time logging.info(fcompleted prediction in {duration} seconds) span.set_attribute(model.predict_time_seconds, duration)几个值得注意的细节输入定义total输出水果数量默认 5与interval每个输出的间隔秒数默认 3都通过Input(default...)声明Cog 会自动为其生成 OpenAPI Schema。AsyncConcatenateIterator[str]流式输出run()是异步生成器用yield逐行吐数据await asyncio.sleep(interval)制造可控延迟。返回类型为AsyncConcatenateIterator[str]该类型用于声明“逐块拼接的流式字符串输出”。并发 流式意味着 4 个预测可以同时各自流式产出互不阻塞。span 关联setupspan 的上下文被保存为_setup_context预测 span 通过trace.Link(setup_context)与之关联从而把初始化阶段与预测阶段串联成完整 trace。指标记录current_scope().record_metric(output_tokens, total)把自定义指标写入当前预测的作用域。current_scope的实现位于 crates/coglet-python/src/metric_scope.rs每个预测持有独立的Scope并通过 ContextVar 路由确保并发场景下指标不会串到别的预测上。关于并发场景下的指标隔离integration-tests/concurrent/concurrent_test.go 中的TestConcurrentAsyncMetrics专门验证了这一点同时发出 5 个并发预测断言每个预测响应中的prediction_index指标都等于自己的索引防止了“指标被静默丢弃”或“指标张冠李戴”两种并发缺陷。集成测试如何验证并发行为integration-tests/concurrent/concurrent_test.go是理解并发语义的最佳实证材料它包含两个核心用例TestConcurrentPredictionsconcurrent_test.go使用concurrent(max5)的异步预测器5 个 goroutine 同时发起 5 个预测请求每个内部sleep 1s随后立即向/shutdown发送关闭请求。断言所有预测在 3 秒内完成——如果 Cog 串行执行 5 个 1 秒任务需要 5 秒3 秒内完成即证明预测确实是并发执行的同时验证服务端优雅关闭会等待在途预测跑完。TestConcurrentAboveLimitconcurrent_test.gocog.yaml中配置concurrency: {max: 2}先发 2 个长期预测占满槽位再轮询发送溢出请求断言其返回409 Conflict且响应体status为failed、error包含 “capacity” 字样。顺带一提concurrency.max是 YAML 层面的并发配置方式与装饰器等效两个入口最终都会收敛到同一套 permit 池机制。这也解释了示例文档中“如果 Cog 达到最大并发阈值将返回 409 Conflict”的完整链路装饰器标记__cog_concurrent_max__→ coglet 据此初始化并发槽位 → 槽位耗尽时 HTTP 路由返回 409。Telemetry用 OpenTelemetry 采集模型遥测示例的第二大主题是遥测Telemetry。它使用opentelemetry相关依赖见 requirements.txt包含opentelemetry-api、opentelemetry-sdk、opentelemetry-exporter-otlp-proto-http演示如何为模型采集可观测数据。遥测初始化流程run.py 的初始化逻辑分三步第一步读取 Honeycomb token 文件。要求镜像构建目录中存在名为honeycomb_token.key的文件内容为 API tokenhoneycomb_token try: with open(./honeycomb_token.key, r) as f: honeycomb_token f.read().strip() except FileNotFoundError: logging.info(honeycomb_token.key not found; OTEL will be disabled) if not honeycomb_token: os.environ[OTEL_SDK_DISABLED] true文件缺失时不会报错而是通过OTEL_SDK_DISABLEDtrue优雅地禁用 SDK。第二步通过环境变量配置 OTLP 导出目标。三个关键环境变量分别是环境变量示例值作用OTEL_EXPORTER_OTLP_ENDPOINThttps://api.honeycomb.io/指定 OTLP 导出端点使用自定义采集器时改为自己的地址OTEL_EXPORTER_OTLP_HEADERSx-honeycomb-teamtoken携带 Honeycomb 的团队认证头OTEL_SERVICE_NAMEcog-model服务名即数据在 Honeycomb 中的归属数据源示例文档特别说明事件会被发送到名为cog-model的数据源你可以通过修改OTEL_SERVICE_NAME来更换数据源名称若使用自定义端点则通过OTEL_EXPORTER_OTLP_ENDPOINT配置。第三步创建 TracerProvider 并注册导出器。resource Resource( attributes{model.name: replicate/hello-concurrency, cog_version: __version__} ) provider TracerProvider(resourceresource) provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(provider) tracer trace.get_tracer(predict)Resource携带model.name与cog_version两个资源属性所有 span 都会继承它们BatchSpanProcessor负责把 span 批量异步发送到 OTLP 端点。之后通过tracer.start_as_current_span(...)在setup与run中创建 span并写入model.setup_time_seconds、inputs.total、metrics.output_tokens、model.predict_time_seconds等属性。本地调试模式示例文档还提到run.py末尾有一段可取消注释的本地调试代码# Local OTEL debugging # from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor # os.environ[OTEL_EXPORTER_OTLP_ENDPOINT] http://otel-collector.local-otel.orb.local:4318 # os.environ[OTEL_SDK_DISABLED] # provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))这段代码演示了两种本地调试手段一是把导出端点改为本地 OpenTelemetry Collector 地址二是用SimpleSpanProcessor ConsoleSpanExporter把 span 直接打印到控制台——后者无需任何外部服务是最快的验证方式。注意这里演示的是把OTEL_SDK_DISABLED重新置空以恢复 SDK 功能。构建与运行在仓库根目录执行以下命令即可体验完整流程# 构建镜像需在包含 cog.yaml 的项目目录中 cog build # 以 HTTP 服务方式运行默认监听 5000 端口 cog serve # 并发测试同时发起多个预测请求 curl -X POST http://localhost:5000/predictions \ -H Content-Type: application/json \ -d {input: {total: 5, interval: 1}}由于concurrent(max4)最多 4 个预测并行执行第 5 个并发请求会收到409 Conflict。若想运行时调整上限而不重建镜像可设置COG_MAX_CONCURRENCY环境变量覆盖装饰器取值该行为同样记录于 docs/python.md。需要生产级遥测时在项目目录放入honeycomb_token.key后重新构建镜像即可把 span 数据推送到 Honeycomb 的cog-model数据源。小结hello-concurrency用不到 130 行代码串起了 Cog 并发生态的完整链路concurrent(maxN)装饰器负责声明并发上限python/cog/init.pycoglet 的 permit 池负责并发槽位的分配与毒化回收crates/coglet/src/permit/mod.rsHTTP 路由在容量耗尽时返回 409 Conflictcrates/coglet/src/transport/http/routes.rs而current_scope().record_metric与 OpenTelemetry 配合让每个并发预测的指标和 trace 都能被准确归因与采集。无论你是要构建高吞吐的推理 API还是想为模型接入完整的可观测体系这个示例都是可以直接照抄的起点模板。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考