anycv-sdk 0.1.0

适用于AnyCV视觉识别算法的统一SDK和FastAPI服务

AnyCV SDK

新增配置优先接入方式：算法项目可以只提供 algorithm.py 和 anycv.yaml，通过 adapter: python + entrypoint 自动生成标准 FastAPI 接口。详见 配置优先的 SDK 接入方式。

AnyCV SDK 是一个面向视觉算法项目的统一接入层。它把不同算法的模型加载、单帧推理、RTMP 流式任务、健康检查、Swagger 文档和参数解析收敛到同一套 Python SDK 与 FastAPI 服务里。

当前项目只保留两条主线：

• anycv 主环境：本仓库作为统一算法运行环境，通过 configs/algorithms.yaml 管理多个算法，启动一个 FastAPI 服务统一暴露算法接口。
• SDK 分发包：把 anycv_sdk 打成 SDK 包给算法同事使用。算法同事只需要实现算法类并引入 SDK，即可自动暴露健康检查、单帧推理、OPTIONS 探测和可选流式接口。

项目结构

anycv/
├─ pyproject.toml                     # Python 包元信息、依赖、命令行入口
├─ configs/
│  └─ algorithms.yaml                 # 统一多算法服务配置
├─ src/
│  └─ anycv_sdk/
│     ├─ __init__.py                  # SDK 对外导出入口
│     ├─ cli.py                       # anycv-service 命令行入口
│     ├─ service.py                   # 统一多算法 FastAPI 服务
│     ├─ standalone.py                # 单算法模板 FastAPI 服务
│     ├─ core/
│     │  ├─ adapter.py                # 多算法适配器基类
│     │  ├─ config.py                 # YAML 配置加载和路径解析
│     │  ├─ engine.py                 # 多算法运行引擎
│     │  ├─ images.py                 # 图片解码、base64 编码
│     │  ├─ models.py                 # 配置、结果、流任务数据模型
│     │  ├─ params.py                 # HTTP 表单参数解析与合并
│     │  └─ streams.py                # 本地推流子进程管理
│     └─ algorithms/
│        ├─ __init__.py               # adapter 名称到具体类的注册入口
│        ├─ base_yolo.py              # YOLO 算法通用加载逻辑
│        ├─ crowd.py                  # 人群检测/人数统计适配器
│        ├─ floating_plastic.py       # 漂浮塑料垃圾检测适配器
│        ├─ illegal_parking_obb.py    # 车辆违停 OBB 检测适配器
│        └─ yolo_results.py           # YOLO 结果转换工具
├─ algorithms/                        # 原始算法脚本和算法侧配置
├─ templates/
│  └─ algorithm_project/              # 单算法项目模板
└─ tests/                             # SDK 单元测试和接口契约测试

src/anycv_sdk 会被打进 SDK 包；根目录下的 algorithms/ 主要保存原始算法脚本、示例配置和本项目运行所需资源。模型权重建议放在 weights/ 或服务器统一模型目录中，不建议提交到 Git，也不会自动打进 SDK 包。

当前算法

算法 ID	路由	类型	流式任务
crowd-person-track-001	/detect/crowd	人群检测/跟踪统计	支持
floating-plastic-detect-001	/detect/floating-plastic	水面漂浮塑料垃圾检测	支持
illegal-parking-obb-001	/detect/illegal-parking-obb	车辆 OBB 检测	支持

算法清单以 configs/algorithms.yaml 为准。服务启动时只注册 status: 1 的算法。

快速运行

Windows PowerShell：

uv python install 3.12
uv sync --extra algorithms --extra dev
uv run anycv-service --config configs\algorithms.yaml --host 0.0.0.0 --port 8000

Ubuntu / Linux：

uv python install 3.12
uv sync --extra algorithms --extra dev
uv run anycv-service --config configs/algorithms.yaml --host 0.0.0.0 --port 8000

启动后访问：

• GET http://127.0.0.1:8000/
• GET http://127.0.0.1:8000/api/alg/v1/health
• GET http://127.0.0.1:8000/docs

如果使用 GPU，建议按服务器 CUDA 版本单独安装对应 PyTorch，例如：

uv pip install --force-reinstall torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu128

模型权重需要按 configs/algorithms.yaml 中的 modelPath 放到对应目录，或把 modelPath 改成服务器上的绝对路径。

HTTP 接口约定

健康检查

GET /api/alg/v1/health

返回服务状态、已注册算法、模型路径、加载状态和当前活跃推流任务。

单帧推理

POST /detect/{algorithm-route}
Content-Type: multipart/form-data

必须包含：

• file：图片文件字段。

可选字段：

• conf：置信度阈值。
• iou：NMS IoU 阈值。
• imgsz：推理输入尺寸。
• device：推理设备，例如 0、cuda:0、cpu、auto。
• classes：类别过滤，例如 9,10 或列表参数。

响应必须包含：

• result_image：JPEG base64 字符串，不包含 data:image/... 前缀。
• detections：结构化检测结果列表，具体字段由算法决定。
• cost_ms：推理耗时，单位毫秒。

启动流式任务

POST /detect/{algorithm-route}/stream
Content-Type: application/json

请求体：

{
  "input_url": "rtmp://source/live/a",
  "output_server": "rtmp://target/live/result",
  "stream_key": "stream-a",
  "width": 1920,
  "height": 1080,
  "bitrate": 2500
}

width、height、bitrate 传 0 时使用算法默认配置。

停止流式任务

DELETE /detect/{algorithm-route}/stream/{stream_key}

SDK 会按 algorithm_id + stream_key 找到本地子进程并终止。

新增算法编写流程规范

1. 选择接入方式

当前项目只维护两种新增算法路径：

1. 接入 anycv 主环境：算法需要纳入本仓库统一服务时，在 src/anycv_sdk/algorithms/ 编写适配器，并在 configs/algorithms.yaml 注册算法。
2. 使用 SDK 独立暴露接口：算法同事独立维护算法项目时，安装 anycv-sdk，继承 AlgorithmTemplate，调用 create_algorithm_app() 自动生成标准接口。

如果算法需要视频流能力，仍然通过 stream.command 托管本地算法脚本。

2. 编写适配器

在 src/anycv_sdk/algorithms/ 下新增文件，例如 my_algorithm.py：

from __future__ import annotations

from typing import Any

from anycv_sdk.algorithms.base_yolo import YoloAdapter, resolve_device
from anycv_sdk.core.images import encode_jpeg_base64
from anycv_sdk.core.models import AlgorithmResult


class MyAlgorithmAdapter(YoloAdapter):
    def infer(self, image: Any, params: dict[str, Any] | None = None) -> AlgorithmResult:
        model = self.ensure_loaded()
        merged = self.merge_params(params)
        result = model.predict(
            source=image,
            conf=float(merged.get("conf", 0.25)),
            imgsz=int(merged.get("imgsz", 1280)),
            device=resolve_device(merged.get("device", "auto")),
            verbose=False,
        )[0]
        return AlgorithmResult(
            {
                "detections": [],
                "result_image": encode_jpeg_base64(result.plot()),
                "algorithmId": self.config.algorithm_id,
                "algorithmName": self.config.algorithm_name,
            }
        )

适配器要求：

• load() 必须幂等，重复调用不能重复占用显存；继承 YoloAdapter 时已内置该逻辑。
• infer() 接收 SDK 解码后的 OpenCV BGR 图片。
• 所有请求参数必须通过 self.merge_params(params) 与配置默认值合并。
• 返回结果必须包含 result_image。
• detections 建议始终返回列表，没有目标时返回 []，不要返回 None。
• 算法异常应抛出 AlgorithmError 或普通异常，由服务层统一转换为 HTTP 错误。

3. 注册 adapter

在 src/anycv_sdk/algorithms/__init__.py 的 create_adapter() 中增加分支：

if config.adapter == "my_algorithm":
    from anycv_sdk.algorithms.my_algorithm import MyAlgorithmAdapter

    return MyAlgorithmAdapter(config)

config.adapter 的值必须与 YAML 配置中的 adapter 字段一致。

4. 增加配置

在 configs/algorithms.yaml 的 algorithms 下追加：

- adapter: my_algorithm
  algorithmId: my-algorithm-001
  algorithmName: 示例算法
  category: detection
  route: /detect/my-algorithm
  supportStream: 1
  timeoutMs: 30000
  status: 1
  modelPath: /data/models/anycv/my_algorithm/model.pt
  runtime:
    mode: main
  defaults:
    conf: 0.25
    imgsz: 1280
    device: auto
  stream:
    cwd: ../algorithms/my_algorithm
    script: detect_rtmp_push.py
    log_file: logs/{stream_key}.log
    command:
      - "{python}"
      - "-u"
      - "{script}"
      - "--url"
      - "{input_url}"
      - "--push_url"
      - "{output_server}"
      - "--model"
      - "{model_path}"
      - "--conf"
      - "{conf}"
      - "--imgsz"
      - "{imgsz}"
      - "--device"
      - "{device}"

配置规范：

• algorithmId 一旦给平台使用，不要随意修改。
• route 必须唯一，并以 / 开头。
• modelPath 生产环境建议使用绝对路径。
• status: 0 表示保留配置但不注册接口。
• defaults 只放允许请求覆盖的安全参数。
• stream.command 每个元素是一个命令行参数，不要写成一整条 shell 字符串。
• stream.command 支持 {input_url}、{output_server}、{stream_key}、{model_path}、{python}、{script} 和 defaults 中的字段。

5. 增加测试

至少覆盖：

• 配置能正常加载，路由出现在 load_app_config() 结果中。
• 单帧推理返回包含 result_image 的结构。
• 流式算法能正确渲染 stream.command。
• 如果新增了结果转换逻辑，需要用假对象测试字段结构，避免强依赖真实模型。

运行：

uv run pytest

6. 更新文档和版本

新增算法进入主服务后，应同步更新：

• configs/algorithms.yaml 中的注释。
• 本 README 的当前算法表格。
• 部署手册中的模型文件放置说明。
• pyproject.toml 的 version，如果需要对外重新发 SDK 包。

SDK 打包和分发

当前项目使用 pyproject.toml 管理依赖、使用 uv 管理 .venv 环境，同时保留 setuptools 和 src 布局用于打包。包名为 anycv-sdk，命令行入口为 anycv-service。

打包前建议先跑测试：

uv sync --extra algorithms --extra dev
uv run pytest

构建 wheel 和源码包：

uv pip install build
uv run python -m build

构建完成后会生成：

dist/
├─ anycv_sdk-0.1.0-py3-none-any.whl
└─ anycv_sdk-0.1.0.tar.gz

算法同事的单算法项目安装本地 wheel：

uv pip install "dist/anycv_sdk-0.1.0-py3-none-any.whl"

如果算法项目需要运行 YOLO/OpenCV，还需要安装算法扩展依赖：

uv pip install numpy opencv-python ultralytics

注意：

• wheel 默认只包含 src/anycv_sdk 包代码。
• 模型权重不会被打包，需要由部署流程单独下发。
• 根目录 algorithms/ 下的原始脚本不会自动进入 SDK 包；算法同事应在自己的算法项目中维护算法脚本和模型权重。
• 对算法同事分发 SDK 时，应固定版本号，并在算法项目中锁定依赖版本。

常用命令

# 本地开发安装
uv sync --extra algorithms --extra dev

# 启动统一多算法服务
uv run anycv-service --config configs/algorithms.yaml --host 0.0.0.0 --port 8000

# 强制启动时加载全部启用模型
uv run anycv-service --config configs/algorithms.yaml --load-on-startup

# 禁止启动时预加载模型，改为首次请求懒加载
uv run anycv-service --config configs/algorithms.yaml --no-load-on-startup

# 运行测试
uv run pytest

# 构建 SDK 包
uv run python -m build

相关文档

• 部署手册
• SDK 使用手册