在 Sage 系统中的作用：为 llmage（大模型管理）提供计费定价服务。每个 LLM 模型通过 llm.ppid 字段关联到 pricing 模块的定价项目（pricing_program），调用结束后由 llmage 的 llm_charging() 通过 buffered_charging(ppid, usages) 计算费用。

核心职责：

定价项目管理：创建定价项目，定义定价规格（pricing_spec）
定价项目时序：支持按时间段定义不同定价策略（enabled_date / expired_date）
定价数据管理：通过 Excel 批量导入/导出定价表，自动生成下拉菜单
定价计算引擎：根据输入数据匹配定价规则，通过公式计算费用
定价测试：提供测试界面验证定价规则是否正确

目录结构

pricing/
├── pyproject.toml              # 构建配置
├── setup.cfg                   # 包元数据：name=pricing
├── README.md
├── pricing/                    # Python 源码包
│   ├── __init__.py             # 空
│   ├── init.py                 # 模块初始化，注册函数到 ServerEnv
│   ├── pricing.py              # 定价引擎核心：PricingProgram 类 + 匹配逻辑
│   └── write_pattern.py        # Excel 定价模版生成与解析
├── json/                       # CRUD 定义（bricks-framework）
│   ├── pricing_program.json    # 定价项目管理
│   └── pricing_program_timing.json  # 定价项目时序
├── models/                     # 表定义（xlsx 格式）
│   ├── pricing_program.xlsx
│   └── pricing_program_timing.xlsx
├── wwwroot/                    # Web 前端资源
│   ├── menu.ui                 # 菜单
│   ├── test_pricing_program.ui # 定价测试窗口
│   ├── test_pricing_program.dspy # 定价测试逻辑
│   ├── pricing_test.ui         # 定价验证窗口
│   ├── get_all_pricing_programs.dspy # 获取所有定价项目
│   ├── pi_get_all_specs.dspy   # 获取所有规格
│   ├── upload_pricing_data.dspy   # 上传定价数据
│   ├── download_pricing_data.dspy   # 下载定价数据
│   ├── download_pricing_pattern.dspy # 下载定价模版
│   ├── load_pricing_data.ui    # 加载定价数据界面
│   ├── get_platform_providers.dspy # 获取平台供应商
│   ├── imgs/                   # 图标
│   └── pricing_item/           # 定价项管理
│       ├── index.ui
│       ├── add_pricing_item.dspy
│       ├── update_pricing_item.dspy
│       ├── delete_pricing_item.dspy
│       ├── get_pricing_item.dspy
│       └── get_spec_fields_by_psid.dspy
└── script/
    └── perms.json              # RBAC 权限配置

数据库表结构

表关系

pricing_program (定价项目) ──1:N──> pricing_program_timing (定价项目时序)
pricing_program_timing ──1:N──> pricing_item (定价项)

核心表说明

表名	说明	关键字段
pricing_program	定价项目定义	id, name, ownerid, providerid, pricing_spec(YAML格式), pricing_belong, discount
pricing_program_timing	定价项目时序（分时段定价）	id, ppid, name, enabled_date, expired_date, pricing_data(YAML格式)
pricing_item	定价项	id, pptid, name, formula, spec_values

pricing_spec（定价规格）定义格式

pricing_program.pricing_spec 字段是 YAML 格式，定义定价的维度和字段：

model:
  type: str
  label: "模型"
  options:
    - "viduq3-pro"
    - "viduq3-turbo"

resolution:
  type: str
  label: "分辨率"
  options:
    - "1024p"
    - "720p"
    - "540p"

duration:
  type: int
  label: "时长"

off_peak:
  type: int
  label: "错峰执行"
  options:
    - 0  # 正常时段
    - 1  # 错峰

price:
  type: float
  label: 单价

字段属性说明：

type: 字段类型（str、int、float、bool、factor）
label: 显示名称
options: 可选值列表（用于前端下拉菜单和定价匹配）
value_mode: 匹配模式（可选，默认 =）
- = 或省略：精确匹配
- between: 区间匹配（格式：小值 ~ 大值，可加 = 表示包含端点）
- in: 枚举匹配（格式：值1 值2 ...）
- >、>=、<、<=: 比较运算

必须有 price 字段，其 type 必须为 float。

pricing_data（定价数据）格式

pricing_program_timing.pricing_data 是 YAML 格式，包含 fields 和 pricings 两部分：

fields:
  model:
    type: str
    label: "模型"
  resolution:
    type: str
    label: "分辨率"
  formula:
    type: str
    label: "计算公式"
  price:
    type: float
    label: "单价"

pricings:
  - model: viduq3-pro
    resolution: 1024p
    price: 0.5
    formula: price * 1

  - model: viduq3-pro
    resolution: 720p
    price: 0.3
    formula: price * 1

  - model: viduq3-turbo
    resolution: 1024p
    price: 0.2
    formula: price * 1

定价匹配规则

调用 get_pricing_from_ymalstr(config_data, yamlstr) 时：

遍历 pricings 中的每条定价规则
对每条规则的每个字段，用 check_value() 判断输入数据是否匹配：
- value_mode == '=': 精确匹配 data_value == spec_value
- value_mode == 'between': 区间匹配（如 0 ~ 10000 表示 0 <= value < 10000）
- value_mode == 'in': 枚举匹配（如 gpt-4 gpt-3.5 表示其中之一）
- value_mode == '>'/'<'>='/<=': 比较运算
所有字段都匹配的定价规则才会被选中
执行 formula 字段中定义的计算公式，得到 amount
返回所有匹配的定价项列表

数据映射（mappings）：

model_mappings:  # 模型名称映射
  "doubao-seed-2-0-pro-260215": "doubao-seed-2-0-pro"
  "doubao-seed-2-0-lite-260215": "doubao-seed-2-0-lite"

输入数据中的值会先通过 mappings 转换，再与定价规则匹配。

Python API

模块初始化

def load_pricing():
    env = ServerEnv()
    env.get_pricing_program = get_pricing_program
    env.write_pricing_patten = PricingProgram.write_pricing_patten
    env.pricing_program_charging = PricingProgram.charging
    env.buffered_charging = PricingProgram.buffered_charging
    env.load_pricing_data = PricingProgram.load_pricing_data
    env.get_pricing_program = PricingProgram.get_pricing_program
    env.test_pricing = test_pricing

其他模块的 .dspy 文件可通过 globals() 直接使用这些函数。

PricingProgram 类（pricing.py）

定价计算

# 带缓存的计费（推荐，llmage 使用此方法）
pp = PricingProgram()
prices = await pp.buffered_charging(ppid, data)
# data: {'prompt_tokens': 52, 'completion_tokens': 1416, 'model': 'xxx', ...}
# 返回: [DictObject(amount=0.001, cost=0.001*discount, data={...}, formula=...)]

# 不带缓存的计费（传入 sor）
prices = await PricingProgram.charging(sor, ppid, data)

定价数据管理

# 获取定价项目（带缓存，按日期）
d = await PricingProgram.get_ppid_pricing(ppid)
# 返回: DictObject(name, ownerid, providerid, pricing_belong, discount, pricing_data)

# 获取定价项目定义
pp = await PricingProgram.get_pricing_program(ppid)

# 从 Excel 文件加载定价数据
await PricingProgram.load_pricing_data(pptid, webpath_xlsx)

# 生成定价 Excel 模版
fpath = await PricingProgram.write_pricing_patten(request, ppid)

YAML/DB 转换

# DB 到应用（YAML 字符串 → dict）
PricingProgram.pp_db2app(pp)         # pricing_program
PricingProgram.ppt_db2app(ppt)       # pricing_program_timing

# 应用到 DB（dict → YAML 字符串）
PricingProgram.pp_app2db(pp)
PricingProgram.ppt_app2db(ppt)

独立函数

# 测试定价（给定 pptid 和输入数据，返回总金额）
amount = await test_pricing(pptid, data)

# 获取定价项目定义
pp = await get_pricing_program(ppid)

核心匹配引擎

# 静态方法，直接在内存 YAML 上匹配
prices = PricingProgram.get_pricing_from_ymalstr(config_data, yamlstr)
# config_data: 输入数据字典
# yamlstr: pricing_data 的 YAML 字符串
# 返回: 匹配的定价项列表，每项包含 amount（通过 formula 计算）

前端页面

pricing_program.json — 定价项目管理

按 ownerid（登录用户所在组织）过滤
pricing_spec 字段不在浏览器中显示（YAML 格式复杂）
子表：pricing_program_timing（定价项目时序）
工具栏：测试 按钮，弹出测试窗口

pricing_program_timing.json — 定价项目时序

按 ownerid 过滤
子表：pricing_item（定价细项）
工具栏：
- 定价模版：下载 Excel 模版
- 上传定价数据：上传 Excel 定价文件
- 下载定价数据：下载当前定价数据
- 验证定价：测试定价规则

Excel 模版与数据导入

生成模版

通过 write_pricing_patten(request, ppid) 生成 Excel 模版：

根据 pricing_spec 中的字段生成列头
options 字段自动生成 Excel 下拉菜单（DataValidation）
bool 类型生成 TRUE/FALSE 下拉
factor 类型字段不参与定价（仅作为因子）

导入定价数据

通过 load_pricing_data(pptid, webpath_xlsx) 从 Excel 导入：

解析 Excel 第一行作为字段名
将 label 映射回字段名（Excel 列头是 label，需映射回 spec 中的 key）
转换为 YAML 格式存入 pricing_program_timing.pricing_data

在 Sage 系统中的协同关系

llmage (模型管理) ──→ llm.ppid ──→ pricing_program.id
                                        │
                                        ├── pricing_spec (定价维度定义)
                                        │
                                        └── pricing_program_timing ──→ pricing_data (定价表)
                                                                        │
                                                                        └── pricings[] ──→ formula 计算金额

调用流程：

llmage 调用外部 LLM API，获取 token 用量（prompt_tokens, completion_tokens）
记录到 llmusage 表（accounting_status='created'）
后台记账任务 backend_accounting() 每 10 秒轮询
调用 llm_charging(ppid, llmusage) → buffered_charging(ppid, usages)
pricing 模块匹配定价规则，计算 amount 和 cost
更新 llmusage.amount 和 llmusage.cost
调用 llm_accounting() 记账到 accounting 模块

HTTP API 接口

获取定价展示数据

GET /pricing/api/get_pricing_display.dspy?ppid={ppid}
Authorization: Bearer {token}

说明： 返回指定定价项目的可读定价数据，供前端展示。自动解析 pricing_data 字段，转换为人类可读的价格表。

返回示例（Token 定价）：

{
  "status": "ok",
  "data": {
    "ppid": "5i1JIpqERgCWqKQ4DCegD",
    "name": "通义千问 qwen3.7-max",
    "pricing_type": "per_use",
    "items": [
      {
        "filters": {"model": "qwen3.7-max"},
        "filter_labels": {"模型": "qwen3.7-max"},
        "price_factors": [
          {
            "factor": "uncache_tokens",
            "label": "非缓存tokens",
            "unit_price": 6.0,
            "unit": "百万",
            "unit_label": "元/百万"
          }
        ]
      },
      {
        "filters": {"model": "qwen3.7-max"},
        "filter_labels": {"模型": "qwen3.7-max"},
        "price_factors": [
          {
            "factor": "cached_tokens",
            "label": "缓存tokens",
            "unit_price": 1.2,
            "unit": "百万",
            "unit_label": "元/百万"
          }
        ]
      }
    ],
    "display_text": "【通义千问 qwen3.7-max】定价:\n  - 非缓存tokens: 6.0 元/百万\n  - 缓存tokens: 1.2 元/百万"
  }
}

返回示例（视频生成定价）：

{
  "status": "ok",
  "data": {
    "ppid": "vidu_video_pricing",
    "name": "viduq3视频定价",
    "pricing_type": "per_use",
    "items": [
      {
        "filters": {"model": "viduq3-turbo", "resolution": "1080p", "off_peak": "0"},
        "filter_labels": {"模型": "viduq3-turbo", "分辨率": "1080p", "错峰": "0"},
        "price_factors": [
          {
            "factor": "duration",
            "label": "时长",
            "unit_price": 0.56,
            "unit": "秒",
            "unit_label": "元/秒"
          }
        ]
      }
    ],
    "display_text": "【viduq3视频定价】定价:\n  - 时长: 0.56 元/秒 [model=viduq3-turbo, resolution=1080p, off_peak=0]"
  }
}

返回字段说明：

ppid: 定价项目 ID
name: 定价项目名称
pricing_type: 定价类型（per_use 按次计费）
items: 定价项列表
- filters: 过滤条件（如适用模型、分辨率等）
- filter_labels: 过滤条件的中文标签
- price_factors: 计价因子列表
  - factor: 因子字段名
  - label: 因子中文名称
  - unit_price: 单位价格（已转换为展示价，无需再乘）
  - unit: 单位（百万、秒、次等）
  - unit_label: 单位标签（元/百万、元/秒等）
  - tiered: 阶梯定价（仅当价格不同时展示）
display_text: 人类可读的价格表文本

pricing_data 使用指南

两种格式

pricing_data 支持两种格式：

1. 新格式（推荐）：price_factors + unit_prices + unit

适用于简单定价场景，直接指定计价因子和单位价格。

示例 1：Token 定价（文本生成）

unit_values:
  百万: 1000000
fields:
  model:
    type: str
    role: filter
    label: 模型
  uncache_tokens:
    type: float
    role: factor
    label: 非缓存tokens
  cached_tokens:
    type: float
    role: factor
    label: 缓存tokens
  completion_tokens:
    type: float
    role: factor
    label: 输出tokens
pricings:
  # 非缓存输入定价
  - price_factors: uncache_tokens
    unit_prices: 6.0
    unit: 百万
    filters:
      - model: qwen3.7-max
  
  # 缓存输入定价
  - price_factors: cached_tokens
    unit_prices: 1.2
    unit: 百万
    filters:
      - model: qwen3.7-max
  
  # 输出定价
  - price_factors: completion_tokens
    unit_prices: 18.0
    unit: 百万
    filters:
      - model: qwen3.7-max

示例 2：视频生成定价（多维度过滤）

unit_values:
  次: 1
fields:
  model:
    type: str
    role: filter
    label: 模型
  resolution:
    type: str
    role: filter
    label: 分辨率
  duration:
    type: int
    role: filter
    label: 时长(秒)
    value_mode: between
  off_peak:
    type: str
    role: filter
    label: 错峰执行
  flat:
    type: float
    role: factor
    label: 固定费用
pricings:
  - price_factors: flat
    unit_prices: 85.0
    unit: 次
    filters:
      - model: viduq2-pro
      - resolution: 1080p
      - duration: '1'
      - off_peak: false
  
  - price_factors: flat
    unit_prices: 43.0
    unit: 次
    filters:
      - model: viduq2-pro
      - resolution: 1080p
      - duration: '1'
      - off_peak: true
  
  - price_factors: duration
    unit_prices: 0.56
    unit: 秒
    filters:
      - model: viduq3-turbo
      - resolution: 1080p
      - off_peak: 0

2. 旧格式：formula（公式计算）

适用于复杂定价场景，使用 Python 表达式计算金额。

示例：按 Token 数量计算

fields:
  model:
    type: str
    role: filter
    label: 模型
  prompt_tokens:
    type: float
    role: factor
    label: 输入tokens
  completion_tokens:
    type: float
    role: factor
    label: 输出tokens
pricings:
  - model: gpt-4
    formula: (3.2 * prompt_tokens + 16 * completion_tokens) / 1000000.0
  
  - model: gpt-3.5
    formula: (0.5 * prompt_tokens + 1.5 * completion_tokens) / 1000000.0

关键规则

unit_values 必须定义：在 pricing_data 顶部定义 unit_values，如 百万: 1000000，用于单位换算
unit_prices 存储展示价：新格式中 unit_prices 直接存储展示价格（如 6.0 表示 6.0 元/百万），不需要再乘以单位值
filters 支持两种格式：
- 列表格式：filters: [{model: xxx}, {resolution: yyy}]
- 字典格式：直接作为字段写在 pricing 条目中
value_mode 默认精确匹配：不指定时做 = 精确匹配，可指定 between、in、>、< 等
formula 可引用任意字段：旧格式中 formula 可引用 config_data 中的任意字段名
多条件匹配：如果多条定价规则都匹配，全部返回（不是只返回第一条）

关键设计要点

YAML 驱动定价规则：所有定价逻辑用 YAML 描述，支持字段定义、选项、匹配模式、计算公式
多维度条件匹配：支持 between、in、=、>、<、>=、<= 等多种匹配模式
公式计算：每条定价规则可定义独立的计算公式（如 (3.2 * prompt_tokens + 16 * completion_tokens) / 1000000）
数据映射（mappings）：输入值可预先映射（如 doubao-seed-2-0-pro-260215 → doubao-seed-2-0-pro）
分时段定价：pricing_program_timing 支持按 enabled_date/expired_date 定义不同时期的定价
折扣支持：pricing_program.discount 字段控制折扣，最终 cost = amount * discount
Excel 批量操作：通过 Excel 导入导出定价数据，自动生成下拉菜单，降低维护门槛
内存缓存：PricingProgram.pricing_data 按 {ppid}.{date} 键缓存，避免重复查询
定价项（pricing_item）：可单独管理定价细项，用于更复杂的定价结构

依赖关系

pricing
├── sqlor          # 数据库 ORM
├── apppublic      # 工具库（日志、唯一ID、时间工具等）
├── ahserver       # Web 服务器框架
├── openpyxl       # Excel 文件读写（依赖安装）
└── pyyaml         # YAML 解析（依赖安装）

开发注意事项

pricing_spec 必须包含 price 字段：且 type 为 float，否则定价匹配会失败
formula 字段：每条定价规则的 formula 是 Python 表达式，通过 eval() 执行，可引用 config_data 中的任意字段
value_mode 默认值为 =：不指定时做精确匹配
between 区间格式：0 ~ 100 表示 0 <= value < 100，0 =~ 100 表示 0 <= value <= 100
factor 类型字段：在 Excel 模版中被跳过（不生成列），仅作为计算因子使用
pricing_data 结构：必须是 {fields: {...}, pricings: [...]} 格式，缺少任一部分都会报错
缓存过期策略：缓存按日期键管理，每个 ppid 只保留最近 2 天的缓存
discount 应用：discount 在 buffered_charging() 和 charging() 中乘以 amount 得到 cost
Excel 列头是 label：导入时通过 label 映射回 spec 中的 key 字段名，label 必须完全匹配
多匹配规则：如果多条定价规则都匹配输入数据，全部返回（不是只返回第一条），调用方需自行处理

README.md Unescape Escape

pricing 模块开发文档

模块概述