yumoqing/pipeline-llm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(pipeline-llm): 模型治理模块v1.0.0 — 供应商/端点目录/账号钱包/模型注册/组织容错策略/限额限流/双维度记账

Browse Source 
- 8张llm_前缀表(手写幂等DDL): vendor/account/model/api_profile/org_policy/org_quota/user_quota/usage
- gateway.py 门禁链: 限流→限额→策略选模型(主备容错)→账号×端点轮转(偏好过滤+余额加权+冷却避让)→预授权-结算
- api_key AES单层加密(非RC4,盐不对称不可解); 限流Redis分钟窗口(db4) fail-open
- 双维度记账: 一次调用同行记cost(账号侧)+charge(组织池侧); 充值=recharge行
- CRUD+index.ui 8卡片导航+总览; appcodes字典5组; 设计规范+30测试用例
- 策略即开关: 机构无策略→__LEGACY__走旧llm表(向后兼容零改动)
This commit is contained in:
yumoqing 2026-09-01 17:27:17 +08:00
commit 660ad21365
41 changed files with 3454 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
.gitignore vendored Normal file
View File

@ -0,0 +1,20 @@
# build artifacts
build/
*.egg-info/
__pycache__/
*.pyc
dist/
# CRUD generated directories (auto-generated by xls2ui at build time, NOT in repo)
wwwroot/llm_vendor/
wwwroot/llm_account/
wwwroot/llm_model/
wwwroot/llm_api_profile/
wwwroot/llm_org_policy/
wwwroot/llm_org_quota/
wwwroot/llm_user_quota/
wwwroot/llm_usage/
# venv
py3/
venv/

74
README.md Normal file
View File

@ -0,0 +1,74 @@
# pipeline-llm — 产线平台模型治理模块
产线平台的模型层：供应商/账号/模型治理 + 组织容错策略 + 限额限流 + 双维度记账。
所有表 `llm_` 前缀，建在宿主应用库（pipeline）。
## 模块定位
解决五类问题：
1. **供给侧治理**：供应商协议+全球端点目录、账号（钱包）、模型注册、适配模板
2. **容量治理**：供应商限流 → 多账号轮转；多供应商 → 主备模型容错
3. **需求侧治理**：组织级模型容错策略、组织限额限流、个人限额限流
4. **计费**：双维度记账（成本侧记账号、消费侧记组织/个人），模型挂定价 `pricing_program.ppid`
5. **端点选择**：端点目录配在供应商下（全球配置一次），账号选用端点，客户按区域偏好（prefer/must）
## 数据表（8 张）
| 表 | 说明 |
|---|---|
| `llm_vendor` | 供应商（协议 + 端点目录 JSON） |
| `llm_account` | 供应商账号（钱包：余额/充值/选端点，api_key AES 加密） |
| `llm_model` | 模型注册（能力分类/挂定价 ppid/成本售价，name 唯一） |
| `llm_api_profile` | 适配模板（协议×能力形态去重） |
| `llm_org_policy` | 组织容错策略（主备模型 + 端点偏好） |
| `llm_org_quota` | 组织限额限流（池余额 + QPM） |
| `llm_user_quota` | 个人限额限流（额度上限 + QPM） |
| `llm_usage` | 用量流水（双维度记账事实表） |
## 调用门禁链
```
① 个人限流 → ② 组织限流 → ③ 个人额度 → ④ 组织池(预授权)
→ ⑤ 组织策略选模型(主→备) → ⑥ 账号×端点候选池(偏好过滤+余额加权+冷却避让)
→ ⑦ 调用 → ⑧ 结算(实际用量，多退少补，双维度记账)
```
任何一级不过：快速失败 + 明确中文原因（不挂起、不静默）。
## 集成方式（复用不改建）
模块只提供治理引擎与配置管理；实际 LLM 转发仍由宿主已有的
`pipeline_service.llm_bridge` / `llm_proxy` 承担，在两处挂钩子：
- `llm_bridge._get_model_config`：治理启用时前置解析（主备容错+端点选择）
- `llm_proxy.proxy_chat_completion`：调用前后接门禁与结算（运行环境 token 路径）
机构未配置治理（无策略、无新模型）→ 返回 `__LEGACY__`，调用方回退旧 `llm` 表，向后兼容。
## 宿主接入清单
1. `app/pipeline_app.py` init() 加 `load_llm()`（try/except ImportError 兜底）
2. `build.sh`：clone 枚举 + pip install 枚举 + CRUD 生成枚举 + wwwroot 符号链接枚举
3. `scripts/create_tables.py`：IDE 幂等建表枚举加 `pipeline-llm`（或执行本模块 `mysql.ddl.sql`）
4. `scripts/import_init.py`：INIT_MODULES 加本模块 `init/data.json`（appcodes 字典）
5. `conf/rp.json`：`/pipeline-llm/**` 注册到 `logined`；运行 `scripts/load_path.py`
6. `wwwroot/index.ui` 主菜单加「模型治理」项（URL `/pipeline-llm`）
7. `restart-pipeline.sh` 重启（.py 变更必须重启；`api_key` 有缓存，改 key 也须重启）
## 关键约定
- **api_key 用 AES 单层加密**（`appPublic.aes.aes_encode_b64`，password_key 做密钥）。
⚠️ 不用 RC4：`password()/unpassword()` 盐不对称，加密后永久不可解。
- **限流用 Redis 分钟窗口原子计数**（db4，与会话 db3 隔离）；Redis 故障时限流放行（fail-open），
余额仍由 DB 条件 UPDATE 守住。
- **429 → (账号,端点)对冷却 60s**；账号级冷却用 `*` 通配键。
- **双维度记账**：一次调用同行记 `cost`（成本侧，账号余额扣）+ `charge`（消费侧，组织池扣）；
充值是 `status=recharge` 行（对账用）。
- **预授权-结算**：调用前按预估冻结，调用后按实际 token 多退少补。
## 文档
- `docs/design-spec.md` — 完整设计规范（架构/表/机制/功能点/分期）
- `docs/test-cases.md` — 测试用例（30 例，部署后全部执行）
## 开发日志
见 `docs/work-log-2026-09-01.md`。

259
docs/design-spec.md Normal file
View File

@ -0,0 +1,259 @@
# pipeline-llm 模块设计规范
> 版本：1.0 | 日期：2026-09-01 | 状态：待实施
> 宿主：产线平台（pipeline-app，共享 `pipeline` 库） | 表前缀：`llm_`
## 1. 背景与目标
产线平台现有模型层是 `llm` 单表（name/provider/model_id/api_base/api_key/capabilities），
内嵌接入信息、单一 OpenAI-chat 调用形态、能力是标签不是契约。随着模型增多（非 t2t 能力、
多供应商、多账号），需要独立的模型治理模块。本模块解决五类问题：
1. **供给侧治理**：供应商协议/端点目录、账号（钱包）、模型注册、适配模板
2. **容量治理**：供应商限流 → 多账号轮转；多供应商 → 主备模型容错
3. **需求侧治理**：组织级模型容错策略、组织限额限流、个人限额限流
4. **计费**：双维度记账（成本侧记账号、消费侧记组织/个人），挂定价平台 `pricing_program.ppid`
5. **端点选择**：供应商全球端点配置一次，账号选用端点，客户按区域偏好（prefer/must）
## 2. 架构总览
```
供给侧（容量） 需求侧（治理）
───────────── ─────────────
llm_vendor (协议+端点目录) llm_org_policy (组织容错策略)
└─ llm_account (钱包+选端点) llm_org_quota (组织限额限流)
└─ llm_model (模型注册) llm_user_quota (个人限额限流)
└─ llm_api_profile (适配)
↓
llm_usage (用量流水，双维度)
```
**调用门禁链**（所有 LLM 调用收敛点 `llm_call`）：
```
① 个人限流窗口 → ② 组织限流窗口
→ ③ 个人额度余额 → ④ 组织池余额
→ ⑤ 组织容错策略选模型（主→备）
→ ⑥ 账号×端点候选池：区域偏好过滤 → 剩余额度加权轮转 + 冷却避让
→ ⑦ 预授权（冻结预估用量）→ 调用 → ⑧ 结算（实际用量，双维度记账）
任何一级不过：快速失败 + 明确原因（不挂起、不静默）
```
## 3. 数据表（8 张，全部 `llm_` 前缀，建在 `pipeline` 库）
### 3.1 llm_vendor — 供应商
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| name | str(100) | 供应商名（如 阿里云百炼） |
| protocol | str(20) | openai_compat / dashscope_async / custom |
| endpoints | text(JSON) | 端点目录 `[{region, base_url, proxy, timeout, note}]` |
| description | text | |
| status | str(20) | active / disabled，默认 active |
| org_id | str(32) | 0=系统级 |
| created_at/updated_at | str(30) | |
端点目录挂在 vendor 下（全球端点配置一次），账号选用。`endpoints` JSON 元素：
`region`（domestic/international/europe/...）、`base_url`、`proxy`（可选，国外端点走 SOCKS）、
`timeout`（秒，默认 60）、`note`。
### 3.2 llm_account — 供应商账号（钱包）
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| vendor_id | str(32) | → llm_vendor.id |
| name | str(100) | 账号名 |
| api_key | str(500) | **AES 单层加密**（appPublic.aes.aes_encode_b64，password_key 做密钥，对称可解密）。⚠️ 不能用 RC4：rc4.password()/unpassword() 盐不对称，加密后永久不可解（2026-08 实测坑）。改 key 须重启进程（有缓存） |
| endpoint_ids | text(JSON) | 选用的端点索引列表（对应 vendor.endpoints 下标），有序；空=不启用 |
| balance | decimal(14,4) | 余额（充值-消耗），单位与定价币种一致 |
| total_recharge | decimal(14,4) | 累计充值 |
| status | str(20) | active / suspended / exhausted，默认 active |
| org_id | str(32) | |
| created_at/updated_at | str(30) | |
独立充值、独立记账。轮转限流时余额不足的账号直接跳过（不消耗）。
### 3.3 llm_model — 模型注册（旧 `llm` 表的替代）
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| vendor_id | str(32) | → llm_vendor.id |
| account_id | str(32) | 默认账号（可空，运行时轮转选） |
| name | str(100) UNIQUE | 模型注册名（平台内唯一，调用方用此名） |
| vendor_model_id | str(100) | 供应商侧模型名（各家命名不一） |
| capability | str(20) | 能力类型，默认 t2t（见 §6 能力标准） |
| sync_mode | str(10) | sync / async_task，默认 sync |
| profile_id | str(32) | → llm_api_profile.id（适配模板） |
| ppid | str(32) | → 定价平台 `pricing_program.id`（挂定价） |
| default_params | text(JSON) | 默认参数（temperature/max_tokens 等） |
| status | str(20) | active / disabled |
| description | text | |
| org_id | str(32) | |
| created_at/updated_at | str(30) | |
### 3.4 llm_api_profile — 适配模板（协议×能力形态去重）
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| name | str(100) | 如 openai-chat、dashscope-image-async |
| protocol | str(20) | 同 vendor.protocol |
| capability | str(20) | 能力类型 |
| request_template | text | Jinja2：参数 → 供应商请求体 |
| response_template | text | Jinja2：供应商响应 → 统一结果（含 usage） |
| param_schema | text(JSON) | bricks input_fields 格式（参数表单定义） |
| status | str(20) | active / disabled |
| created_at/updated_at | str(30) | |
模板按「协议×能力形态」开，不按模型开——同一端点 N 个模型共用一条，杜绝重复记录。
一期只内置 `openai-chat`（sync），非 chat 形态模型进来时再加对应 profile。
### 3.5 llm_org_policy — 组织模型容错策略
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| org_id | str(32) UNIQUE | 一组织一策略 |
| primary_model_id | str(32) | → llm_model.id 主模型 |
| backup_model_ids | text(JSON) | 备模型 id 列表（有序） |
| endpoint_pref | str(20) | prefer_domestic / prefer_intl / must_domestic / must_intl / any，默认 any |
| status | str(20) | active / disabled |
| created_at/updated_at | str(30) | |
生效优先级：会话/项目选择 > 组织策略 > 平台默认（与现有
`sd_projects.default_model > pipeline_agent_settings.default_llm_id` 链兼容，组织策略插中间）。
must 模式：候选端点耗尽宁可报错冒泡，绝不跨端点。
### 3.6 llm_org_quota — 组织限额限流
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| org_id | str(32) UNIQUE | |
| balance | decimal(14,4) | 组织池余额（充值-消耗） |
| total_recharge | decimal(14,4) | 累计充值 |
| rate_limit | int | 组织 QPM 上限（0=不限） |
| status | str(20) | active / disabled |
| created_at/updated_at | str(30) | |
### 3.7 llm_user_quota — 个人限额限流
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| user_id | str(32) | |
| org_id | str(32) | 所属组织（额度挂组织池） |
| quota_limit | decimal(14,4) | 个人额度上限（0=不限） |
| quota_used | decimal(14,4) | 个人已用 |
| rate_limit | int | 个人 QPM 上限（0=不限） |
| status | str(20) | active / disabled |
| created_at/updated_at | str(30) | |
关系默认「共享池 + 个人上限」：组织充值一个池，个人上限防单人打爆池子。
个人额度耗尽不影响他人；组织池耗尽全员不可用（冒泡提醒充值）。
### 3.8 llm_usage — 用量流水（双维度记账事实表）
| 字段 | 类型 | 说明 |
|---|---|---|
| id | str(32) PK | |
| org_id / user_id | str(32) | 消费侧维度 |
| model_id / account_id | str(32) | 供给侧维度（model_id 空+source=recharge 时为充值记录） |
| endpoint_region | str(20) | 实际走的端点区域 |
| req_tokens / resp_tokens | int | 实际用量 |
| cost | decimal(12,6) | 成本侧金额（账号扣减，按定价） |
| charge | decimal(12,6) | 消费侧金额（组织池扣减，按产品定价） |
| ppid | str(32) | 定价项目（冗余自模型，便于对账） |
| task_ref | str(100) | 调用来源（任务/会话标识） |
| status | str(20) | ok / failed / recharge（充值记录） |
| note | str(200) | 失败原因/充值备注 |
| created_at | str(30) | |
一次调用两条线：**成本侧** cost 记账号（供应商对账），**消费侧** charge 记组织池+个人
（客户计费）。充值是 balance 变动事件，记 `source=recharge` 行（model_id/account_id 空）。
## 4. 核心机制
### 4.1 选择链与轮转（gateway.py `llm_call`）
```python
async def llm_call(prompt, model=None, org_id=None, user_id=None, task_ref='', **kw)
```
1. 解析模型：调用方指定 model → 直接用；否则查组织策略主备链
2. 候选对构建：模型的账号池（account.status=active）× 账号选用端点
3. 区域偏好过滤：must_* 只留对应区域端点对；prefer_* 排序（匹配区域优先）
4. 加权轮转：按账号余额 + 端点窗口剩余配额加权；限流冷却中的对跳过
5. 调用 → 429 标记该(账号,端点)对冷却 → 换下一对；账号池耗尽 → 降备模型；
备模型也耗尽 → must 模式冒泡报错 / prefer 模式跨端点重试
6. 401/403 标记账号失效（不轮转）；5xx 短暂重试（最多 2 次）
### 4.2 预授权-结算（防并发击穿余额）
- 调用前：按 max_tokens 预估（默认 2000 token × 单价）Redis 原子冻结组织池+个人额度
- 调用后：按实际 token 多退少补，写双维度流水
- GeneratorExit/超时：结算兜底（按预估释放，写 failed 流水）
### 4.3 限流（Redis 滑动窗口）
- key：`llm_rl:{user|org}:{id}:{minute}`，原子 INCR + EXPIRE
- 超窗拒绝：返回明确错误（个人限流/组织限流），下窗口自动恢复，不涉及钱
- 供应商侧 429：(账号,端点)对冷却 60s，窗口内不再选它
### 4.4 定价对接（ppid）
- 模型挂 `pricing_program.id`（定价平台的定价项目）
- 消费侧单价从定价平台读取（跨库：定价模块库），一期支持手动设置 `price_per_1k`
兜底（组织策略页配置），二期接定价平台实时价
- 成本侧单价：账号维度配置（供应商结算价），一期也走手动配置
## 5. 功能点清单（共 18 项）
| # | 功能点 | 表/文件 | 优先级 |
|---|---|---|---|
| F01 | 供应商管理（含端点目录配置） | llm_vendor + CRUD | P0 |
| F02 | 账号管理（充值、余额、选端点） | llm_account + CRUD | P0 |
| F03 | 模型注册（挂能力、挂定价、挂适配） | llm_model + CRUD | P0 |
| F04 | 适配模板管理 | llm_api_profile + CRUD | P1（一期内置 openai-chat） |
| F05 | 组织容错策略（主备模型+端点偏好） | llm_org_policy + CRUD | P0 |
| F06 | 组织限额限流（充值、QPM） | llm_org_quota + CRUD | P0 |
| F07 | 个人限额限流 | llm_user_quota + CRUD | P0 |
| F08 | 调用门禁（①-④ 限流限额检查） | gateway.py | P0 |
| F09 | 主备容错（策略选模型→备链） | gateway.py | P0 |
| F10 | 账号×端点轮转（加权+冷却） | gateway.py | P0 |
| F11 | 区域偏好过滤（prefer/must） | gateway.py | P0 |
| F12 | 预授权-结算 | gateway.py | P0 |
| F13 | 双维度记账流水 | llm_usage | P0 |
| F14 | 充值操作（账号/组织池） | recharge dspy | P0 |
| F15 | 用量查询（按组织/模型/账号/时间） | usage 列表页 | P1 |
| F16 | 模型管理入口（主菜单） | index.ui 菜单项 | P0 |
| F17 | 与现有调用链兼容（llm_bridge 切换） | llm_bridge.py 适配 | P0 |
| F18 | 机构未配置冒泡提示 | gateway 错误路径 | P0 |
## 6. 能力类型标准（capabilities 取值，先行收敛）
| 值 | 含义 | 同步性 |
|---|---|---|
| t2t | 文本对话/生成 | sync |
| t2i | 文生图 | async_task |
| i2t | 图生文（多模态理解） | sync |
| t2v | 文生视频 | async_task |
| embedding | 文本向量化 | sync |
| mm-embedding | 多模态向量化 | sync |
| rerank | 重排序 | sync |
| mm-rerank | 多模态重排序 | sync |
| tts | 语音合成 | sync |
| asr | 语音识别 | sync |
新增能力类型走评审（改本文档 + init/data.json appcodes），禁止野生标签。
## 7. 集成点
1. **宿主加载**：`pipeline_app.py` init() 加 `load_llm()`（try/except ImportError 兜底）
2. **菜单**：`wwwroot/index.ui` 主菜单加「模型管理」项 → `/llm`（TabPanel）
3. **RBAC**：`/llm/**` 注册到 superuser；静态资源 any；load_path.py 注册各路径
4. **i18n**：zh/en 双语文案（模块内文案走 appcodes + i18n msg.txt）
5. **建表**：mysql.ddl.sql（json2ddl 生成 + 手写补齐），存量库 ALTER 幂等
6. **兼容**：旧 `llm` 表保留，`llm_bridge.py` 优先查新表（按 name），无则回退旧表——
现有调用零改动，迁移平滑
## 8. 分期
- **一期（本次交付）**：全部 10 张表 + F01-F03/F05-F14/F16-F18 + F04 内置 openai-chat + F15
- **二期（触发：第一个非 t2t 模型）**：异步任务形态适配模板、任务轮询端点
- **三期（触发：定价平台实时价接入）**：成本/消费侧单价自动读取

66
docs/test-cases.md Normal file
View File

@ -0,0 +1,66 @@
# pipeline-llm 测试用例（部署到测试环境后全部执行）
> 测试环境：pipeline@pipeline.opencomputing.cn /d/pipeline/pipeline-app
> 验收标准：浏览器真实执行（curl 不算通过）+ 数据库事实核验
## A. 部署与集成（5 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| A1 | 服务启动 | 重启 pipeline_app | 9090 返回 200，日志有 `[pipeline_llm] loaded` |
| A2 | 建表完整 | SHOW TABLES LIKE 'llm\_%' | 8 张表齐全 |
| A3 | 菜单入口 | 浏览器登录→主菜单 | 「模型管理」项存在，点击打开 /llm 页面（非空白） |
| A4 | RBAC | 未登录访问 /llm/index.ui | 401/跳转登录；登录后 200 |
| A5 | 兼容回退 | 现有驾驶舱对话（走旧 llm 表模型） | 正常回复，不因新模块中断 |
## B. 供给侧配置（6 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| B1 | 供应商+端点目录 | 建供应商，endpoints 配国内+国外两端点 | 保存成功，列表可见两端点 |
| B2 | 账号+充值 | 建账号绑定 B1 供应商，选国内端点，充值 100 | balance=100，total_recharge=100，流水有 recharge 行 |
| B3 | api_key 加密 | 查库 B2 账号 api_key | 密文（非明文），调用时能解密使用 |
| B4 | 模型注册+挂定价 | 建模型挂 vendor+ppid+capability=t2t | 保存成功，ppid 关联可见 |
| B5 | 适配模板 | 内置 openai-chat profile 存在 | request/response 模板非空，参数 schema 合法 |
| B6 | 端点探测 | 账号选用不存在的端点下标 | 保存时校验拒绝（防死配置） |
## C. 需求侧治理（5 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| C1 | 组织策略 | 建组织策略：主模型=B4，备模型空，端点 any | 保存成功 |
| C2 | 组织充值 | 组织池充值 50 | balance=50，流水有记录 |
| C3 | 个人限额 | 建个人配额：额度 10、QPM 5 | 保存成功 |
| C4 | 组织限流 | 同组织 1 分钟内发起 >QPM 次调用 | 超限请求返回「组织限流」明确错误 |
| C5 | 个人限额耗尽 | 个人连续调用至额度用完 | 后续请求返回「个人额度不足」，不影响同组织他人 |
## D. 调用门禁链（7 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| D1 | 正常调用 | 配置齐全后发起一次对话调用 | 成功返回，usage 流水 1 条（双维度金额>0） |
| D2 | 预授权结算 | D1 调用前后查组织池余额 | 冻结→释放，最终扣减=实际 charge |
| D3 | 账号余额不足跳过 | 账号 A 余额 0、账号 B 有余额 | 调用自动走 B，A 不消耗 |
| D4 | 429 轮转 | mock 账号 A 端点返回 429 | 自动切同账号另一端点或账号 B，调用成功 |
| D5 | 主备容错 | 主模型全部账号不可用 | 自动降备模型，调用成功，日志记录降级 |
| D6 | must 端点 | 策略 must_domestic，仅国际端点可用 | 报错冒泡（绝不跨端点），错误信息明确 |
| D7 | 机构未配置 | 无策略无余额的机构调用 | 冒泡提示「未配置模型/余额」，不静默失败 |
## E. 记账与用量（4 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| E1 | 双维度流水 | D1 后查 llm_usage | 同行含成本侧 cost + 消费侧 charge，ppid 冗余正确 |
| E2 | 账号对账 | 账号余额 = 充值 - Σcost | 数值吻合 |
| E3 | 组织对账 | 组织池余额 = 充值 - Σcharge | 数值吻合 |
| E4 | 用量查询页 | 用量列表按组织/模型/时间筛选 | 筛选结果正确，金额汇总与流水一致 |
## F. 回归（3 例）
| # | 用例 | 步骤 | 预期 |
|---|---|---|---|
| F1 | 待办动态表单 | 触发冒泡→待办内提交 | 既有功能不受影响 |
| F2 | 工作空间 | 开发产线打开工作空间 | 仓库树正常（本模块改动不波及） |
| F3 | 驾驶舱对话 | 完整一轮开发产线对话 | 正常，模型解析走新链路 |
**共计 30 例**。执行报告逐条记录：实际命令/截图路径 + 数据库核验结果 + 结论（通过/失败/跳过及原因）。

39
init/data.json Normal file
View File

@ -0,0 +1,39 @@
{
"appcodes": [
{"parentid": "llm_protocol", "parentname": "供应商协议", "items": [
{"k": "openai_compat", "v": "OpenAI兼容"},
{"k": "dashscope_async", "v": "DashScope异步任务"},
{"k": "custom", "v": "自定义"}
]},
{"parentid": "llm_status", "parentname": "状态", "items": [
{"k": "active", "v": "启用"},
{"k": "disabled", "v": "停用"},
{"k": "suspended", "v": "暂停"},
{"k": "exhausted", "v": "余额耗尽"}
]},
{"parentid": "llm_capability", "parentname": "模型能力类型", "items": [
{"k": "t2t", "v": "文本对话/生成"},
{"k": "t2i", "v": "文生图"},
{"k": "i2t", "v": "图生文"},
{"k": "t2v", "v": "文生视频"},
{"k": "embedding", "v": "文本向量化"},
{"k": "mm-embedding", "v": "多模态向量化"},
{"k": "rerank", "v": "重排序"},
{"k": "mm-rerank", "v": "多模态重排序"},
{"k": "tts", "v": "语音合成"},
{"k": "asr", "v": "语音识别"}
]},
{"parentid": "llm_endpoint_pref", "parentname": "端点偏好", "items": [
{"k": "any", "v": "不限"},
{"k": "prefer_domestic", "v": "优先国内"},
{"k": "prefer_intl", "v": "优先国际"},
{"k": "must_domestic", "v": "仅国内(合规)"},
{"k": "must_intl", "v": "仅国际(合规)"}
]},
{"parentid": "llm_usage_status", "parentname": "用量状态", "items": [
{"k": "ok", "v": "成功"},
{"k": "failed", "v": "失败"},
{"k": "recharge", "v": "充值"}
]}
]
}

43
json/llm_account_list.json Normal file
View File

@ -0,0 +1,43 @@
{
"tblname": "llm_account",
"title": "供应商账号",
"params": {
"sortby": "name",
"browserfields": {
"exclouded": [
"id",
"api_key"
],
"alters": {
"vendor_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_vendor_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"api_key",
"balance",
"total_recharge",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_account.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_account.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_account.dspy')}}"
},
"confidential_fields": [
"api_key"
]
}
}

45
json/llm_api_profile_list.json Normal file
View File

@ -0,0 +1,45 @@
{
"tblname": "llm_api_profile",
"title": "适配模板",
"params": {
"sortby": "name",
"browserfields": {
"exclouded": [
"id",
"request_template",
"response_template",
"param_schema"
],
"alters": {
"protocol": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_protocol_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"capability": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_capability_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_api_profile.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_api_profile.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_api_profile.dspy')}}"
}
}
}

54
json/llm_model_list.json Normal file
View File

@ -0,0 +1,54 @@
{
"tblname": "llm_model",
"title": "模型注册",
"params": {
"sortby": "name",
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"vendor_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_vendor_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"account_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_account_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"capability": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_capability_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"ppid": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_ppid_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_model.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_model.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_model.dspy')}}"
}
}
}

42
json/llm_org_policy_list.json Normal file
View File

@ -0,0 +1,42 @@
{
"tblname": "llm_org_policy",
"title": "组织容错策略",
"params": {
"sortby": "created_at",
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"primary_model_id": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_model_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"endpoint_pref": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_endpoint_pref_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_org_policy.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_org_policy.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_org_policy.dspy')}}"
}
}
}

32
json/llm_org_quota_list.json Normal file
View File

@ -0,0 +1,32 @@
{
"tblname": "llm_org_quota",
"title": "组织限额限流",
"params": {
"sortby": "created_at",
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"balance",
"total_recharge",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_org_quota.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_org_quota.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_org_quota.dspy')}}"
}
}
}

43
json/llm_usage_list.json Normal file
View File

@ -0,0 +1,43 @@
{
"tblname": "llm_usage",
"title": "用量流水",
"params": {
"sortby": [
"created_at"
],
"browserfields": {
"exclouded": [
"id"
]
},
"editexclouded": [
"id",
"org_id",
"user_id",
"model_id",
"account_id",
"endpoint_region",
"req_tokens",
"resp_tokens",
"cost",
"charge",
"ppid",
"task_ref",
"status",
"note",
"created_at"
],
"record_toolbar": [
{
"label": "查询统计",
"actiontype": "dspy",
"url": "{{entire_url('../api/llm_usage_query.dspy')}}",
"options": {
"icon": "search",
"cwidth": 20,
"cheight": 12
}
}
]
}
}

31
json/llm_user_quota_list.json Normal file
View File

@ -0,0 +1,31 @@
{
"tblname": "llm_user_quota",
"title": "个人限额限流",
"params": {
"sortby": "created_at",
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"quota_used",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_user_quota.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_user_quota.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_user_quota.dspy')}}"
}
}
}

36
json/llm_vendor_list.json Normal file
View File

@ -0,0 +1,36 @@
{
"tblname": "llm_vendor",
"title": "模型供应商",
"params": {
"sortby": "name",
"browserfields": {
"exclouded": [
"id"
],
"alters": {
"protocol": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_protocol_options.dspy')}}",
"valueField": "value",
"textField": "text"
},
"status": {
"uitype": "code",
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
"valueField": "value",
"textField": "text"
}
}
},
"editexclouded": [
"id",
"created_at",
"updated_at"
],
"editable": {
"new_data_url": "{{entire_url('./add_llm_vendor.dspy')}}",
"update_data_url": "{{entire_url('./update_llm_vendor.dspy')}}",
"delete_data_url": "{{entire_url('./delete_llm_vendor.dspy')}}"
}
}
}

113
models/llm_account.json Normal file
View File

@ -0,0 +1,113 @@
{
"summary": [
{
"name": "llm_account",
"title": "供应商账号(钱包)",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "vendor_id",
"title": "供应商ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "name",
"title": "账号名称",
"type": "str",
"length": 100,
"nullable": "no"
},
{
"name": "api_key",
"title": "API密钥(加密)",
"type": "str",
"length": 500
},
{
"name": "endpoint_ids",
"title": "选用端点索引(JSON)",
"type": "text"
},
{
"name": "balance",
"title": "余额",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "total_recharge",
"title": "累计充值",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "org_id",
"title": "所属机构",
"type": "str",
"length": 32,
"nullable": "no",
"default": "'0'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "idx_llm_account_vendor",
"idxtype": "index",
"idxfields": [
"vendor_id"
]
}
],
"codes": [
{
"field": "vendor_id",
"table": "llm_vendor",
"valuefield": "id",
"textfield": "name"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

100
models/llm_api_profile.json Normal file
View File

@ -0,0 +1,100 @@
{
"summary": [
{
"name": "llm_api_profile",
"title": "适配模板",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "name",
"title": "模板名称",
"type": "str",
"length": 100,
"nullable": "no"
},
{
"name": "protocol",
"title": "协议类型",
"type": "str",
"length": 20,
"nullable": "no"
},
{
"name": "capability",
"title": "能力类型",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'t2t'"
},
{
"name": "request_template",
"title": "请求模板(Jinja2)",
"type": "text"
},
{
"name": "response_template",
"title": "响应模板(Jinja2)",
"type": "text"
},
{
"name": "param_schema",
"title": "参数Schema(bricks)",
"type": "text"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"codes": [
{
"field": "protocol",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_protocol'"
},
{
"field": "capability",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_capability'"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

200
models/llm_model.json Normal file
View File

@ -0,0 +1,200 @@
{
"summary": [
{
"name": "llm_model",
"title": "模型注册",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "vendor_id",
"title": "供应商ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "account_id",
"title": "默认账号ID",
"type": "str",
"length": 32
},
{
"name": "name",
"title": "模型注册名",
"type": "str",
"length": 100,
"nullable": "no"
},
{
"name": "vendor_model_id",
"title": "供应商侧模型名",
"type": "str",
"length": 100
},
{
"name": "capability",
"title": "能力类型",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'t2t'"
},
{
"name": "sync_mode",
"title": "同步性",
"type": "str",
"length": 10,
"nullable": "no",
"default": "'sync'"
},
{
"name": "profile_id",
"title": "适配模板ID",
"type": "str",
"length": 32
},
{
"name": "ppid",
"title": "定价项目ID",
"type": "str",
"length": 32
},
{
"name": "price_input",
"title": "输入单价(元/千token)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "price_output",
"title": "输出单价(元/千token)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "cost_input",
"title": "输入成本(元/千token)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "cost_output",
"title": "输出成本(元/千token)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "default_params",
"title": "默认参数(JSON)",
"type": "text"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "description",
"title": "描述",
"type": "text"
},
{
"name": "org_id",
"title": "所属机构",
"type": "str",
"length": 32,
"nullable": "no",
"default": "'0'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "uk_llm_model_name",
"idxtype": "unique",
"idxfields": [
"name"
]
},
{
"name": "idx_llm_model_vendor",
"idxtype": "index",
"idxfields": [
"vendor_id"
]
}
],
"codes": [
{
"field": "vendor_id",
"table": "llm_vendor",
"valuefield": "id",
"textfield": "name"
},
{
"field": "account_id",
"table": "llm_account",
"valuefield": "id",
"textfield": "name"
},
{
"field": "profile_id",
"table": "llm_api_profile",
"valuefield": "id",
"textfield": "name"
},
{
"field": "capability",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_capability'"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
},
{
"field": "ppid",
"table": "pricing_program",
"valuefield": "id",
"textfield": "name"
}
]
}

97
models/llm_org_policy.json Normal file
View File

@ -0,0 +1,97 @@
{
"summary": [
{
"name": "llm_org_policy",
"title": "组织模型容错策略",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "机构ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "primary_model_id",
"title": "主模型ID",
"type": "str",
"length": 32
},
{
"name": "backup_model_ids",
"title": "备模型ID列表(JSON)",
"type": "text"
},
{
"name": "endpoint_pref",
"title": "端点偏好",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'any'"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "uk_llm_org_policy_org",
"idxtype": "unique",
"idxfields": [
"org_id"
]
}
],
"codes": [
{
"field": "primary_model_id",
"table": "llm_model",
"valuefield": "id",
"textfield": "name"
},
{
"field": "endpoint_pref",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_endpoint_pref'"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

87
models/llm_org_quota.json Normal file
View File

@ -0,0 +1,87 @@
{
"summary": [
{
"name": "llm_org_quota",
"title": "组织限额限流",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "机构ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "balance",
"title": "组织池余额",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "total_recharge",
"title": "累计充值",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "rate_limit",
"title": "组织QPM上限",
"type": "int",
"default": "0"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "uk_llm_org_quota_org",
"idxtype": "unique",
"idxfields": [
"org_id"
]
}
],
"codes": [
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

149
models/llm_usage.json Normal file
View File

@ -0,0 +1,149 @@
{
"summary": [
{
"name": "llm_usage",
"title": "用量流水(双维度记账)",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "机构ID",
"type": "str",
"length": 32
},
{
"name": "user_id",
"title": "用户ID",
"type": "str",
"length": 32
},
{
"name": "model_id",
"title": "模型ID",
"type": "str",
"length": 32
},
{
"name": "account_id",
"title": "账号ID",
"type": "str",
"length": 32
},
{
"name": "endpoint_region",
"title": "端点区域",
"type": "str",
"length": 20
},
{
"name": "req_tokens",
"title": "请求token",
"type": "int",
"default": "0"
},
{
"name": "resp_tokens",
"title": "响应token",
"type": "int",
"default": "0"
},
{
"name": "cost",
"title": "成本金额(账号)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "charge",
"title": "消费金额(组织池)",
"type": "float",
"length": 12,
"dec": 6,
"default": "0"
},
{
"name": "ppid",
"title": "定价项目ID",
"type": "str",
"length": 32
},
{
"name": "task_ref",
"title": "调用来源",
"type": "str",
"length": 100
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'ok'"
},
{
"name": "note",
"title": "备注/原因",
"type": "str",
"length": 200
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
}
],
"indexes": [
{
"name": "idx_llm_usage_org",
"idxtype": "index",
"idxfields": [
"org_id",
"created_at"
]
},
{
"name": "idx_llm_usage_model",
"idxtype": "index",
"idxfields": [
"model_id"
]
}
],
"codes": [
{
"field": "model_id",
"table": "llm_model",
"valuefield": "id",
"textfield": "name"
},
{
"field": "account_id",
"table": "llm_account",
"valuefield": "id",
"textfield": "name"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_usage_status'"
}
]
}

95
models/llm_user_quota.json Normal file
View File

@ -0,0 +1,95 @@
{
"summary": [
{
"name": "llm_user_quota",
"title": "个人限额限流",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "user_id",
"title": "用户ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "org_id",
"title": "所属机构",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "quota_limit",
"title": "个人额度上限",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "quota_used",
"title": "个人已用",
"type": "float",
"length": 14,
"dec": 4,
"default": "0"
},
{
"name": "rate_limit",
"title": "个人QPM上限",
"type": "int",
"default": "0"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "idx_llm_user_quota_user",
"idxtype": "index",
"idxfields": [
"user_id",
"org_id"
]
}
],
"codes": [
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

98
models/llm_vendor.json Normal file
View File

@ -0,0 +1,98 @@
{
"summary": [
{
"name": "llm_vendor",
"title": "模型供应商",
"primary": [
"id"
],
"catelog": "entity"
}
],
"fields": [
{
"name": "id",
"title": "主键ID",
"type": "str",
"length": 32,
"nullable": "no"
},
{
"name": "name",
"title": "供应商名称",
"type": "str",
"length": 100,
"nullable": "no"
},
{
"name": "protocol",
"title": "协议类型",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'openai_compat'"
},
{
"name": "endpoints",
"title": "端点目录(JSON)",
"type": "text"
},
{
"name": "description",
"title": "描述",
"type": "text"
},
{
"name": "status",
"title": "状态",
"type": "str",
"length": 20,
"nullable": "no",
"default": "'active'"
},
{
"name": "org_id",
"title": "所属机构",
"type": "str",
"length": 32,
"nullable": "no",
"default": "'0'"
},
{
"name": "created_at",
"title": "创建时间",
"type": "timestamp",
"nullable": "no"
},
{
"name": "updated_at",
"title": "更新时间",
"type": "timestamp"
}
],
"indexes": [
{
"name": "idx_llm_vendor_status",
"idxtype": "index",
"idxfields": [
"status"
]
}
],
"codes": [
{
"field": "protocol",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_protocol'"
},
{
"field": "status",
"table": "appcodes_kv",
"valuefield": "k",
"textfield": "v",
"cond": "parentid='llm_status'"
}
]
}

133
mysql.ddl.sql Normal file
View File

@ -0,0 +1,133 @@
-- pipeline-llm 模块数据表（8 张，全部 llm_ 前缀，pipeline 库）
-- 手写幂等 DDL（IF NOT EXISTS + 内联索引），由宿主 scripts/create_tables.py 执行
-- 金额用 DECIMAL（对账精度），时间用 datetime
CREATE TABLE IF NOT EXISTS llm_vendor (
`id` varchar(32) NOT NULL comment '主键ID',
`name` varchar(100) NOT NULL comment '供应商名称',
`protocol` varchar(20) NOT NULL DEFAULT 'openai_compat' comment '协议类型',
`endpoints` text comment '端点目录(JSON): [{region,base_url,proxy,timeout,note}]',
`description` text comment '描述',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`org_id` varchar(32) NOT NULL DEFAULT '0' comment '所属机构',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
KEY `idx_llm_vendor_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='模型供应商';
CREATE TABLE IF NOT EXISTS llm_account (
`id` varchar(32) NOT NULL comment '主键ID',
`vendor_id` varchar(32) NOT NULL comment '供应商ID',
`name` varchar(100) NOT NULL comment '账号名称',
`api_key` varchar(500) comment 'API密钥(AES加密)',
`endpoint_ids` text comment '选用端点索引列表(JSON)',
`balance` decimal(14,4) NOT NULL DEFAULT 0 comment '余额',
`total_recharge` decimal(14,4) NOT NULL DEFAULT 0 comment '累计充值',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`org_id` varchar(32) NOT NULL DEFAULT '0' comment '所属机构',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
KEY `idx_llm_account_vendor` (`vendor_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='供应商账号(钱包)';
CREATE TABLE IF NOT EXISTS llm_model (
`id` varchar(32) NOT NULL comment '主键ID',
`vendor_id` varchar(32) NOT NULL comment '供应商ID',
`account_id` varchar(32) comment '默认账号ID',
`name` varchar(100) NOT NULL comment '模型注册名(平台唯一)',
`vendor_model_id` varchar(100) comment '供应商侧模型名',
`capability` varchar(20) NOT NULL DEFAULT 't2t' comment '能力类型',
`sync_mode` varchar(10) NOT NULL DEFAULT 'sync' comment '同步性',
`profile_id` varchar(32) comment '适配模板ID',
`ppid` varchar(32) comment '定价项目ID(pricing_program)',
`price_input` decimal(12,6) NOT NULL DEFAULT 0 comment '输入单价(元/千token)',
`price_output` decimal(12,6) NOT NULL DEFAULT 0 comment '输出单价(元/千token)',
`cost_input` decimal(12,6) NOT NULL DEFAULT 0 comment '输入成本(元/千token)',
`cost_output` decimal(12,6) NOT NULL DEFAULT 0 comment '输出成本(元/千token)',
`default_params` text comment '默认参数(JSON)',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`description` text comment '描述',
`org_id` varchar(32) NOT NULL DEFAULT '0' comment '所属机构',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_llm_model_name` (`name`),
KEY `idx_llm_model_vendor` (`vendor_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='模型注册';
CREATE TABLE IF NOT EXISTS llm_api_profile (
`id` varchar(32) NOT NULL comment '主键ID',
`name` varchar(100) NOT NULL comment '模板名称',
`protocol` varchar(20) NOT NULL comment '协议类型',
`capability` varchar(20) NOT NULL DEFAULT 't2t' comment '能力类型',
`request_template` text comment '请求模板(Jinja2)',
`response_template` text comment '响应模板(Jinja2)',
`param_schema` text comment '参数Schema(bricks input_fields)',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='适配模板(协议x能力形态去重)';
CREATE TABLE IF NOT EXISTS llm_org_policy (
`id` varchar(32) NOT NULL comment '主键ID',
`org_id` varchar(32) NOT NULL comment '机构ID',
`primary_model_id` varchar(32) comment '主模型ID',
`backup_model_ids` text comment '备模型ID列表(JSON)',
`endpoint_pref` varchar(20) NOT NULL DEFAULT 'any' comment '端点偏好',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_llm_org_policy_org` (`org_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='组织模型容错策略';
CREATE TABLE IF NOT EXISTS llm_org_quota (
`id` varchar(32) NOT NULL comment '主键ID',
`org_id` varchar(32) NOT NULL comment '机构ID',
`balance` decimal(14,4) NOT NULL DEFAULT 0 comment '组织池余额',
`total_recharge` decimal(14,4) NOT NULL DEFAULT 0 comment '累计充值',
`rate_limit` int NOT NULL DEFAULT 0 comment '组织QPM上限(0=不限)',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_llm_org_quota_org` (`org_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='组织限额限流';
CREATE TABLE IF NOT EXISTS llm_user_quota (
`id` varchar(32) NOT NULL comment '主键ID',
`user_id` varchar(32) NOT NULL comment '用户ID',
`org_id` varchar(32) NOT NULL comment '所属机构',
`quota_limit` decimal(14,4) NOT NULL DEFAULT 0 comment '个人额度上限(0=不限)',
`quota_used` decimal(14,4) NOT NULL DEFAULT 0 comment '个人已用',
`rate_limit` int NOT NULL DEFAULT 0 comment '个人QPM上限(0=不限)',
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
`updated_at` datetime DEFAULT current_timestamp() ON UPDATE current_timestamp() comment '更新时间',
PRIMARY KEY (`id`),
KEY `idx_llm_user_quota_user` (`user_id`,`org_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='个人限额限流';
CREATE TABLE IF NOT EXISTS llm_usage (
`id` varchar(32) NOT NULL comment '主键ID',
`org_id` varchar(32) comment '机构ID',
`user_id` varchar(32) comment '用户ID',
`model_id` varchar(32) comment '模型ID',
`account_id` varchar(32) comment '账号ID',
`endpoint_region` varchar(20) comment '端点区域',
`req_tokens` int NOT NULL DEFAULT 0 comment '请求token',
`resp_tokens` int NOT NULL DEFAULT 0 comment '响应token',
`cost` decimal(12,6) NOT NULL DEFAULT 0 comment '成本金额(账号侧)',
`charge` decimal(12,6) NOT NULL DEFAULT 0 comment '消费金额(组织池侧)',
`ppid` varchar(32) comment '定价项目ID',
`task_ref` varchar(100) comment '调用来源',
`status` varchar(20) NOT NULL DEFAULT 'ok' comment '状态(ok/failed/recharge)',
`note` varchar(200) comment '备注/原因',
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
PRIMARY KEY (`id`),
KEY `idx_llm_usage_org` (`org_id`,`created_at`),
KEY `idx_llm_usage_model` (`model_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='用量流水(双维度记账)';

1
pipeline_llm/__init__.py Normal file
View File

@ -0,0 +1 @@
from .init import load_llm

653
pipeline_llm/gateway.py Normal file
View File

@ -0,0 +1,653 @@
"""pipeline_llm.gateway — 模型治理引擎：门禁链 + 主备容错 + 账号端点轮转 + 双维度记账。
调用链（与设计规范 §2 一致）：
① 个人限流窗口 → ② 组织限流窗口
→ ③ 个人额度余额 → ④ 组织池余额（预授权冻结）
→ ⑤ 组织容错策略选模型（主→备）
→ ⑥ 账号×端点候选池：区域偏好过滤 → 余额加权轮转 + 限流冷却避让
→ ⑦ 调用（调用方执行）→ ⑧ govern_settle 结算（实际用量，双维度记账）
集成方式（复用不改建）：
- llm_bridge._get_model_config 前置调 govern_resolve：机构配了策略/新模型表才走治理，
否则返回 None 由旧 llm 表逻辑兜底（向后兼容）。
- proxy_chat_completion（运行环境 token 路径）拿到真实 usage 后调 govern_settle 精算。
- llm_bridge 内部路径按 prompt/response 长度估算 token 调 govern_settle（note='est'）。
限流：Redis 分钟窗口原子计数（db4，与会话 db3 隔离）。Redis 不可用时限流放行（fail-open，
记 warning）——限流失效不应阻断业务，余额检查仍在 DB 侧守住。
余额：读-校验-条件 UPDATE（WHERE balance>=amount），并发竞态由 Redis 预授权计数托底。
错误必须真实可行动（用户铁律）：每一级失败返回明确的中文原因，前端原样展示。
"""
import json
import logging
import time
logger = logging.getLogger("pipeline_llm.gateway")
DBNAME = "pipeline"
GOVERN_REDIS_DB = 4 # 治理计数器命名空间（会话用 db3，互不干扰）
_COOLDOWN_SEC = 60 # 429 冷却窗口
_EST_TOKENS_PER_CALL = 800 # 无法估算时的单次调用预估（预授权用）
# 治理开关缓存：机构是否启用治理（有策略记录或有 llm_model 记录）
_govern_cache: dict = {}
_cfg_cache: dict = {} # (account_id) -> {api_key 解密, base_url...}
# ────────────────────────── 基础设施 ──────────────────────────
def _get_db():
from sqlor.dbpools import DBPools
db = DBPools()
if not db.databases:
from appPublic.jsonConfig import getConfig
config = getConfig()
if config.databases:
db.databases = config.databases
return db, DBNAME
def _password_key():
from appPublic.jsonConfig import getConfig
try:
return getConfig().password_key or 'QRIVSRHrthhwyjy176556332'
except Exception:
return 'QRIVSRHrthhwyjy176556332'
def encrypt_api_key(plain: str) -> str:
"""AES 单层加密（对称可解密）。⚠️ 不用 RC4：password/unpassword 盐不对称，加密后不可解。"""
if not plain:
return ''
from appPublic.aes import aes_encode_b64
return aes_encode_b64(_password_key(), plain)
def decrypt_api_key(enc: str) -> str:
"""解密 api_key。非密文（历史明文/环境差异）原样返回。"""
if not enc:
return ''
try:
from appPublic.aes import aes_decode_b64
return aes_decode_b64(_password_key(), enc)
except Exception:
return enc
def _redis():
"""治理用 Redis 连接（db4）。从 session_redis 配置推导地址，不可用返回 None。"""
try:
import redis as _r
from appPublic.jsonConfig import getConfig
url = ''
try:
url = (getConfig().website or {}).get('session_redis', {}).get('url', '')
except Exception:
url = ''
if not url:
url = 'redis://127.0.0.1:6379/3'
# 替换库号：会话用配置的库，治理固定 GOVERN_REDIS_DB
base = url.rsplit('/', 1)[0]
return _r.Redis.from_url(base + '/' + str(GOVERN_REDIS_DB), socket_timeout=2)
except Exception as e:
logger.warning("pipeline_llm: redis 不可用（限流降级放行）: %s", e)
return None
def _row_to_dict(r):
try:
return {k: v for k, v in vars(r).items() if not callable(v)}
except TypeError:
return dict(r)
def _parse_json(s, default):
if not s:
return default
try:
v = json.loads(s)
return v if v is not None else default
except Exception:
return default
def _fnum(v):
try:
return float(v or 0)
except (TypeError, ValueError):
return 0.0
# ────────────────────────── 门禁 ①②：限流 ──────────────────────────
async def _rate_check(scope: str, scope_id: str, limit: int):
"""分钟窗口限流。返回 (ok, msg)。limit<=0 不限；Redis 故障放行。"""
if not limit or limit <= 0 or not scope_id:
return True, ''
r = _redis()
if r is None:
return True, ''
minute = int(time.time() // 60)
key = 'llm_rl:%s:%s:%d' % (scope, scope_id, minute)
try:
cnt = r.incr(key)
r.expire(key, 70)
except Exception as e:
logger.warning("pipeline_llm: 限流计数失败（放行）: %s", e)
return True, ''
if cnt > limit:
label = '个人' if scope == 'user' else '组织'
return False, '%s限流：本分钟已调用 %d 次（上限 %d），请下一分钟再试' % (label, cnt, limit)
return True, ''
# ────────────────────────── 门禁 ③④：额度检查 + 预授权 ──────────────────────────
async def _reserve_quota(sor, org_id: str, user_id: str, amount: float):
"""预授权：组织池 + 个人额度扣减。返回 (ok, msg, 实际冻结金额)。
条件 UPDATE 防超扣；个人不足先回滚组织侧。0 元调用（未定价模型）不冻结。
"""
if amount <= 0:
return True, '', 0.0
# 组织池
recs = await sor.sqlExe(
"SELECT id, balance FROM llm_org_quota WHERE org_id=${o}$ AND status='active' LIMIT 1",
{"o": org_id})
await sor.sqlExe("COMMIT", {})
org_quota_id = ''
if recs:
if _fnum(getattr(recs[0], 'balance', 0)) < amount:
return False, '组织余额不足：当前余额 %.4f，本次预扣 %.4f。请充值组织池（模型治理→组织限额限流）' % (
_fnum(getattr(recs[0], 'balance', 0)), amount), 0.0
org_quota_id = getattr(recs[0], 'id', '')
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance-${a}$ WHERE id=${i}$ AND balance>=${a}$",
{"a": amount, "i": org_quota_id})
await sor.sqlExe("COMMIT", {})
# 个人额度（有记录才校验；共享池+个人上限模式）
if user_id:
recs = await sor.sqlExe(
"SELECT id, quota_limit, quota_used FROM llm_user_quota "
"WHERE user_id=${u}$ AND org_id=${o}$ AND status='active' LIMIT 1",
{"u": user_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
if recs:
lim = _fnum(getattr(recs[0], 'quota_limit', 0))
used = _fnum(getattr(recs[0], 'quota_used', 0))
if lim > 0 and used + amount > lim:
if org_quota_id: # 回滚组织侧
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance+${a}$ WHERE id=${i}$",
{"a": amount, "i": org_quota_id})
await sor.sqlExe("COMMIT", {})
return False, '个人额度不足：已用 %.4f / 上限 %.4f，本次需 %.4f。请联系管理员调高个人额度' % (
used, lim, amount), 0.0
await sor.sqlExe(
"UPDATE llm_user_quota SET quota_used=quota_used+${a}$ WHERE id=${i}$",
{"a": amount, "i": getattr(recs[0], 'id', '')})
await sor.sqlExe("COMMIT", {})
return True, '', amount
async def _release_quota(sor, org_id: str, user_id: str, amount: float):
"""释放预授权（调用失败时）。"""
if amount <= 0:
return
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance+${a}$ WHERE org_id=${o}$ AND status='active'",
{"a": amount, "o": org_id})
if user_id:
await sor.sqlExe(
"UPDATE llm_user_quota SET quota_used=GREATEST(quota_used-${a}$,0) "
"WHERE user_id=${u}$ AND org_id=${o}$",
{"a": amount, "u": user_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
# ────────────────────────── 门禁 ⑤⑥：策略选模型 + 候选轮转 ──────────────────────────
async def _org_policy(sor, org_id: str):
"""组织容错策略。无记录返回 (主模型空, 备链空, 端点偏好 any)。"""
if not org_id:
return '', [], 'any'
recs = await sor.sqlExe(
"SELECT primary_model_id, backup_model_ids, endpoint_pref FROM llm_org_policy "
"WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": org_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return '', [], 'any'
r = recs[0]
return (getattr(r, 'primary_model_id', '') or '',
_parse_json(getattr(r, 'backup_model_ids', ''), []),
getattr(r, 'endpoint_pref', '') or 'any')
async def _model_chain(sor, org_id: str, model_name: str):
"""解析模型链 [主, 备...]，返回 llm_model 记录 dict 列表。
- 调用方指定 model_name：在 llm_model 找到 → [它] + 策略备链补充；找不到 → 空（走旧表兜底）
- 未指定：策略主模型 + 备链；无策略 → 空
"""
chain_ids = []
primary, backups, pref = await _org_policy(sor, org_id)
if model_name:
recs = await sor.sqlExe(
"SELECT id FROM llm_model WHERE status='active' AND (name=${n}$ OR vendor_model_id=${n}$) LIMIT 1",
{"n": model_name})
await sor.sqlExe("COMMIT", {})
if not recs:
return [], pref
chain_ids.append(getattr(recs[0], 'id', ''))
chain_ids.extend([b for b in backups if b not in chain_ids])
else:
if primary:
chain_ids.append(primary)
chain_ids.extend([b for b in backups if b not in chain_ids])
if not chain_ids:
return [], pref
models = []
for mid in chain_ids:
recs = await sor.sqlExe(
"SELECT id, name, vendor_id, vendor_model_id, capability, default_params, "
"price_input, price_output, cost_input, cost_output, ppid, org_id "
"FROM llm_model WHERE id=${i}$ AND status='active' LIMIT 1", {"i": mid})
await sor.sqlExe("COMMIT", {})
if recs:
models.append(_row_to_dict(recs[0]))
return models, pref
async def _vendor_endpoints(sor, vendor_id: str):
"""供应商端点目录。返回 (protocol, [endpoint dict])。"""
recs = await sor.sqlExe(
"SELECT protocol, endpoints FROM llm_vendor WHERE id=${i}$ AND status='active' LIMIT 1",
{"i": vendor_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return '', []
r = recs[0]
return getattr(r, 'protocol', '') or '', _parse_json(getattr(r, 'endpoints', ''), [])
async def _account_candidates(sor, model: dict):
"""模型的候选账号池：同供应商、启用中。"""
recs = await sor.sqlExe(
"SELECT id, name, api_key, endpoint_ids, balance, status FROM llm_account "
"WHERE vendor_id=${v}$ AND status='active'", {"v": model.get('vendor_id', '')})
await sor.sqlExe("COMMIT", {})
return [_row_to_dict(r) for r in (recs or [])]
def _filter_by_pref(candidates_with_ep, pref: str):
"""按端点偏好过滤/排序候选 (account, endpoint, idx) 列表。
must_*：只保留匹配区域，空则返回空（宁可冒泡不跨端点）。
prefer_*：匹配区域排前。any：原序。
"""
if pref in ('must_domestic', 'prefer_domestic'):
want = 'domestic'
elif pref in ('must_intl', 'prefer_intl'):
want = 'international'
else:
return candidates_with_ep
matched = [c for c in candidates_with_ep if (c[1].get('region') or '') == want]
if pref.startswith('must'):
return matched
rest = [c for c in candidates_with_ep if (c[1].get('region') or '') != want]
return matched + rest
async def _pick_candidate(sor, model: dict, pref: str):
"""构建候选并选一个：区域偏好过滤 → 冷却避让 → 余额加权。
返回 (ok, result)：ok 时 result=(account_dict, endpoint_dict)；否则 result=错误信息。
"""
vendor_id = model.get('vendor_id', '')
if not vendor_id:
return False, '模型「%s」未绑定供应商，无法选择账号' % model.get('name', '')
protocol, endpoints = await _vendor_endpoints(sor, vendor_id)
if not endpoints:
return False, '供应商端点目录为空（模型治理→供应商→端点目录），请配置端点'
accounts = await _account_candidates(sor, model)
if not accounts:
return False, '模型「%s」的供应商下无启用账号（或账号余额状态异常）' % model.get('name', '')
# 展开 (账号, 端点) 对：账号选用的端点下标
pairs = []
for acc in accounts:
idxs = _parse_json(acc.get('endpoint_ids'), [])
if not idxs:
continue
for i in idxs:
try:
i = int(i)
except (TypeError, ValueError):
continue
if 0 <= i < len(endpoints):
pairs.append((acc, endpoints[i], i))
if not pairs:
return False, '所有账号都未选用端点（模型治理→供应商账号→选用端点），请配置'
pairs = _filter_by_pref(pairs, pref)
if not pairs:
return False, ('端点偏好为「仅 %s」，但没有可用的该区域端点。按合规要求不跨端点降级——'
'请配置该区域端点或调整组织策略的端点偏好' %
('国内' if 'domestic' in pref else '国际'))
# 冷却避让 + 余额加权选择
r = _redis()
best, best_score = None, -1.0
for acc, ep, i in pairs:
cool_key = 'llm_cool:%s:%d' % (acc.get('id', ''), i)
acct_key = 'llm_cool:%s:*' % (acc.get('id', ''),)
try:
if r is not None and (r.get(cool_key) or r.get(acct_key)):
continue # 限流冷却中（端点级或账号级）
except Exception:
pass
score = _fnum(acc.get('balance'))
if score > best_score:
best, best_score = (acc, ep, i), score
if best is None:
return False, '模型「%s」的全部候选（账号×端点）都在限流冷却中，请稍后重试' % model.get('name', '')
acc, ep, i = best
if _fnum(acc.get('balance')) <= 0:
return False, '模型「%s」的候选账号余额均不足，请为账号充值' % model.get('name', '')
return True, (acc, ep)
def _mark_cooldown(account_id: str, ep_index):
"""429 后把该账号（或账号×端点对）放进冷却。ep_index='*' 表示账号级。"""
r = _redis()
if r is None:
return
try:
r.set('llm_cool:%s:%s' % (account_id, ep_index), '1', ex=_COOLDOWN_SEC)
except Exception:
pass
def mark_cooldown_by_pair(account_id: str, base_url: str):
"""供代理层 429 回调：按 base_url 找到端点下标进冷却（找不到则全部下标冷却）。"""
try:
# 代理层只拿得到 base_url；冷却键按下标，这里无法反查下标——
# 退化为账号级冷却（该账号所有端点），宁可保守
_mark_cooldown(account_id, '*')
except Exception:
pass
# ────────────────────────── 对外主入口 ──────────────────────────
class GovernError(ValueError):
"""治理失败：消息真实可行动，前端原样展示。"""
async def governance_enabled(sor, org_id: str) -> bool:
"""机构是否启用治理：**策略即开关**——该机构有 active 的容错策略才启用。
⚠️ 不能用"新模型表有记录"判断（2026-09-01 修正）：系统级（org_id=0）模型
注册后会让所有未配策略的机构被误判为启用，调用直接报错，断掉旧 llm 表链路。
机构显式建策略才是接入治理的动作；无策略 → 一切照旧（__LEGACY__）。
带缓存（改配置需重启，与 key 缓存一致）。
"""
if not org_id:
return False
if org_id in _govern_cache:
return _govern_cache[org_id]
enabled = False
try:
recs = await sor.sqlExe(
"SELECT COUNT(*) AS c FROM llm_org_policy WHERE org_id=${o}$ AND status='active'",
{"o": org_id})
await sor.sqlExe("COMMIT", {})
enabled = (int(getattr(recs[0], 'c', 0)) if recs else 0) > 0
except Exception as e:
logger.warning("pipeline_llm: governance_enabled 检查失败（按未启用）: %s", e)
enabled = False
_govern_cache[org_id] = enabled
return enabled
async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '',
est_tokens: int = 0, task_ref: str = ''):
"""门禁链 ①-⑥ + 预授权。
返回 (ok, result)：
ok=True → result = {api_base, api_key(明文,仅进程内), model_id, name,
account_id, endpoint_region, reserved(冻结金额),
model_row(定价信息), user_id, org_id, task_ref}
ok=False → result = 错误信息（真实可行动）
机构未启用治理 → (False, '__LEGACY__') 特殊标记：调用方走旧 llm 表逻辑
"""
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
if not await governance_enabled(sor, org_id or ''):
return False, '__LEGACY__'
# ① 个人限流
if user_id:
recs = await sor.sqlExe(
"SELECT rate_limit FROM llm_user_quota WHERE user_id=${u}$ AND org_id=${o}$ "
"AND status='active' LIMIT 1", {"u": user_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
ulimit = int(_fnum(getattr(recs[0], 'rate_limit', 0))) if recs else 0
ok, msg = await _rate_check('user', user_id, ulimit)
if not ok:
return False, msg
# ② 组织限流
recs = await sor.sqlExe(
"SELECT rate_limit FROM llm_org_quota WHERE org_id=${o}$ AND status='active' LIMIT 1",
{"o": org_id})
await sor.sqlExe("COMMIT", {})
olimit = int(_fnum(getattr(recs[0], 'rate_limit', 0))) if recs else 0
ok, msg = await _rate_check('org', org_id, olimit)
if not ok:
return False, msg
# ⑤ 模型链（含 ⑥ 偏好）
models, pref = await _model_chain(sor, org_id or '', model_name or '')
if not models:
if model_name:
return False, '__LEGACY__' # 指定模型不在新表 → 旧表兜底
return False, ('机构未配置模型容错策略且未指定模型：请在模型治理→组织容错策略配置主/备模型，'
'或在调用时指定模型名')
# ⑥ 逐模型尝试（主 → 备容错）
last_err = ''
chosen = None
for m in models:
ok, cand = await _pick_candidate(sor, m, pref)
if ok and isinstance(cand, tuple):
chosen = (m, cand)
break
last_err = cand if isinstance(cand, str) else last_err
logger.info("pipeline_llm: 模型 %s 候选不可用（%s），尝试备模型", m.get('name'), cand)
if chosen is None:
return False, '主/备模型全部不可用。最后原因：%s' % last_err
model, chosen_pair = chosen
acc, ep = chosen_pair
# ③④ 预授权（按预估用量冻结组织池+个人额度）
price_in = _fnum(model.get('price_input'))
price_out = _fnum(model.get('price_output'))
est = int(est_tokens or 0) or _EST_TOKENS_PER_CALL
est_charge = est / 1000.0 * (price_in + price_out) / 2.0
ok, msg, reserved = await _reserve_quota(sor, org_id or '', user_id or '', est_charge)
if not ok:
return False, msg
# 组装结果（api_key 解密，只在本进程内存）
return True, {
'api_base': (ep.get('base_url') or '').rstrip('/'),
'api_key': decrypt_api_key(acc.get('api_key') or ''),
'model_id': model.get('vendor_model_id') or model.get('name', ''),
'name': model.get('name', ''),
'account_id': acc.get('id', ''),
'endpoint_region': ep.get('region') or '',
'endpoint_proxy': ep.get('proxy') or '',
'timeout': int(_fnum(ep.get('timeout')) or 60),
'reserved': reserved,
'model_row': model,
'user_id': user_id or '',
'org_id': org_id or '',
'task_ref': task_ref or '',
}
async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_tokens: int = 0,
note: str = ''):
"""结算 ⑧：按实际用量修正预授权 + 双维度记账。
ctx govern_resolve 成功时返回的 dict
ok_call 上游调用是否成功（失败也记账：释放预授权，写 failed 流水）
"""
if not ctx or not isinstance(ctx, dict):
return
db, dbname = _get_db()
try:
model = ctx.get('model_row') or {}
price_in = _fnum(model.get('price_input'))
price_out = _fnum(model.get('price_output'))
cost_in = _fnum(model.get('cost_input'))
cost_out = _fnum(model.get('cost_output'))
charge = req_tokens / 1000.0 * price_in + resp_tokens / 1000.0 * price_out
cost = req_tokens / 1000.0 * cost_in + resp_tokens / 1000.0 * cost_out
reserved = _fnum(ctx.get('reserved'))
async with db.sqlorContext(dbname) as sor:
if not ok_call:
# 上游限流(429)失败 → 账号级冷却，轮转避让下一轮不再选它
if '429' in str(note or ''):
_mark_cooldown(ctx.get('account_id', ''), '*')
await _release_quota(sor, ctx.get('org_id', ''), ctx.get('user_id', ''), reserved)
else:
# 多退少补：实际消费 vs 预授权
diff = charge - reserved
if diff > 0:
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance-${d}$ "
"WHERE org_id=${o}$ AND status='active'",
{"d": diff, "o": ctx.get('org_id', '')})
elif diff < 0:
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance+${d}$ "
"WHERE org_id=${o}$ AND status='active'",
{"d": -diff, "o": ctx.get('org_id', '')})
# 个人额度按实际修正（预授权按估算扣了，退还差额）
if ctx.get('user_id') and diff < 0:
await sor.sqlExe(
"UPDATE llm_user_quota SET quota_used=GREATEST(quota_used-${d}$,0) "
"WHERE user_id=${u}$ AND org_id=${o}$",
{"d": -diff, "u": ctx.get('user_id', ''), "o": ctx.get('org_id', '')})
await sor.sqlExe("COMMIT", {})
# 成本侧：账号余额扣减
if ctx.get('account_id') and cost > 0:
await sor.sqlExe(
"UPDATE llm_account SET balance=balance-${c}$ WHERE id=${i}$",
{"c": cost, "i": ctx.get('account_id', '')})
await sor.sqlExe("COMMIT", {})
# 双维度流水
from appPublic.uniqueID import getID
await sor.C('llm_usage', {
'id': getID(),
'org_id': ctx.get('org_id', ''),
'user_id': ctx.get('user_id', ''),
'model_id': model.get('id', ''),
'account_id': ctx.get('account_id', ''),
'endpoint_region': ctx.get('endpoint_region', ''),
'req_tokens': int(req_tokens or 0),
'resp_tokens': int(resp_tokens or 0),
'cost': round(cost, 6),
'charge': round(charge, 6),
'ppid': model.get('ppid', '') or '',
'task_ref': ctx.get('task_ref', ''),
'status': 'ok' if ok_call else 'failed',
'note': (note or '')[:200],
})
await sor.sqlExe("COMMIT", {})
except Exception as e:
logger.warning("pipeline_llm: 结算失败（不阻断调用方）: %s", e)
# ────────────────────────── 充值 ──────────────────────────
async def recharge_account(account_id: str, amount, note: str = '', operator: str = ''):
"""供应商账号充值（成本侧钱包）。返回 (ok, msg)。"""
try:
amount = float(amount)
except (TypeError, ValueError):
return False, '充值金额必须是数字'
if amount <= 0:
return False, '充值金额必须大于 0'
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe("SELECT id FROM llm_account WHERE id=${i}$", {"i": account_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return False, '账号不存在'
await sor.sqlExe(
"UPDATE llm_account SET balance=balance+${a}$, total_recharge=total_recharge+${a}$ "
"WHERE id=${i}$", {"a": amount, "i": account_id})
await sor.sqlExe("COMMIT", {})
await _write_recharge_usage(sor, '', '', account_id, amount, note, operator)
_govern_cache.clear()
return True, '账号充值 %.4f 成功' % amount
async def recharge_org(org_id: str, amount, note: str = '', operator: str = ''):
"""组织池充值（消费侧钱包）。无记录自动创建。返回 (ok, msg)。"""
try:
amount = float(amount)
except (TypeError, ValueError):
return False, '充值金额必须是数字'
if amount <= 0:
return False, '充值金额必须大于 0'
if not org_id:
return False, '缺少机构 ID'
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe("SELECT id FROM llm_org_quota WHERE org_id=${o}$", {"o": org_id})
await sor.sqlExe("COMMIT", {})
if recs:
await sor.sqlExe(
"UPDATE llm_org_quota SET balance=balance+${a}$, total_recharge=total_recharge+${a}$ "
"WHERE org_id=${o}$", {"a": amount, "o": org_id})
else:
from appPublic.uniqueID import getID
await sor.C('llm_org_quota', {
'id': getID(), 'org_id': org_id,
'balance': amount, 'total_recharge': amount,
'rate_limit': 0, 'status': 'active'})
await sor.sqlExe("COMMIT", {})
await _write_recharge_usage(sor, org_id, '', '', amount, note, operator)
_govern_cache.clear()
return True, '组织池充值 %.4f 成功' % amount
async def _write_recharge_usage(sor, org_id, user_id, account_id, amount, note, operator):
"""充值流水（llm_usage 里 status=recharge 的行，对账用）。"""
from appPublic.uniqueID import getID
await sor.C('llm_usage', {
'id': getID(),
'org_id': org_id or '',
'user_id': user_id or '',
'model_id': '',
'account_id': account_id or '',
'endpoint_region': '',
'req_tokens': 0,
'resp_tokens': 0,
'cost': round(float(amount), 6) if account_id else 0,
'charge': round(float(amount), 6) if org_id else 0,
'ppid': '',
'task_ref': 'recharge',
'status': 'recharge',
'note': ((note or '') + (' | 操作人:%s' % operator if operator else ''))[:200],
})
await sor.sqlExe("COMMIT", {})
def load_gateway():
"""注册治理函数到 ServerEnv（供 dspy/其他模块调用）。"""
from ahserver.serverenv import ServerEnv
env = ServerEnv()
env.llm_govern_resolve = govern_resolve
env.llm_govern_settle = govern_settle
env.llm_recharge_account = recharge_account
env.llm_recharge_org = recharge_org
env.llm_encrypt_api_key = encrypt_api_key
logger.info("[pipeline_llm] gateway loaded")

455
pipeline_llm/init.py Normal file
View File

@ -0,0 +1,455 @@
"""pipeline_llm 模块入口 — 模型治理（供应商/账号/模型/组织策略/限额限流/记账）。
宿主集成：应用入口调用 load_llm()（try/except ImportError 兜底）。
所有表共享宿主应用的库（get_module_dbname('pipeline_llm')），表名统一 llm_ 前缀。
模块只提供治理引擎与配置管理；实际 LLM 转发仍由宿主已有的
pipeline_service.llm_bridge / llm_proxy 承担——本模块在两个调用链上挂钩子：
- llm_bridge._get_model_config：治理启用时前置解析（主备容错+端点选择）
- llm_proxy.proxy_chat_completion：调用前后接门禁与结算（运行环境路径）
"""
import json
import logging
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
from appPublic.uniqueID import getID
from .gateway import (
govern_resolve, govern_settle, governance_enabled,
recharge_account, recharge_org,
encrypt_api_key, decrypt_api_key, load_gateway, GovernError,
)
logger = logging.getLogger("pipeline_llm")
MODULE_NAME = 'pipeline_llm'
def _dbname():
"""动态取宿主库名（禁硬编码）。未注册 get_module_dbname 时回落 pipeline。"""
env = ServerEnv()
fn = getattr(env, 'get_module_dbname', None)
if callable(fn):
try:
return fn(MODULE_NAME) or 'pipeline'
except Exception:
pass
return 'pipeline'
def _get_sor():
return DBPools(), _dbname()
# ────────────────────── CRUD（供生成的 CRUD 页面调用） ──────────────────────
def _clean(params_kw):
data = dict(params_kw or {})
for k in ('page', 'rows', 'data_filter', 'sortby'):
data.pop(k, None)
return data
async def create_llm_vendor(params_kw):
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
endpoints = data.get('endpoints', '') or ''
if isinstance(endpoints, (list, dict)):
endpoints = json.dumps(endpoints, ensure_ascii=False)
if endpoints.strip():
eps = json.loads(endpoints)
if not isinstance(eps, list):
raise ValueError('endpoints 必须是 JSON 数组')
for i, ep in enumerate(eps):
if not isinstance(ep, dict) or not (ep.get('base_url') or '').strip():
raise ValueError('端点 #%d 缺少 base_url' % (i + 1))
data['endpoints'] = endpoints
data['id'] = getID()
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.C('llm_vendor', data)
result['success'] = True
result['message'] = '创建成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def update_llm_vendor(params_kw):
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
if 'endpoints' in data:
endpoints = data.get('endpoints', '') or ''
if isinstance(endpoints, (list, dict)):
endpoints = json.dumps(endpoints, ensure_ascii=False)
if endpoints.strip():
eps = json.loads(endpoints)
if not isinstance(eps, list):
raise ValueError('endpoints 必须是 JSON 数组')
for i, ep in enumerate(eps):
if not isinstance(ep, dict) or not (ep.get('base_url') or '').strip():
raise ValueError('端点 #%d 缺少 base_url' % (i + 1))
data['endpoints'] = endpoints
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.U('llm_vendor', data)
result['success'] = True
result['message'] = '更新成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def delete_llm_vendor(params_kw):
"""供应商删除前置检查：有账号/模型引用时禁止（防孤儿）。"""
result = {'success': False, 'message': ''}
try:
vid = (params_kw or {}).get('id', '')
if not vid:
raise ValueError('缺少 id')
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT (SELECT COUNT(*) FROM llm_account WHERE vendor_id=${v}$) + "
"(SELECT COUNT(*) FROM llm_model WHERE vendor_id=${v}$) AS c", {"v": vid})
await sor.sqlExe("COMMIT", {})
cnt = int(getattr(recs[0], 'c', 0)) if recs else 0
if cnt > 0:
raise ValueError('该供应商下还有 %d 个账号/模型，请先移除后再删除' % cnt)
await sor.sqlExe("DELETE FROM llm_vendor WHERE id=${v}$", {"v": vid})
await sor.sqlExe("COMMIT", {})
result['success'] = True
result['message'] = '删除成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def create_llm_account(params_kw):
"""账号创建：api_key 自动 AES 加密；校验端点下标合法。"""
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
vid = data.get('vendor_id', '')
if not vid:
raise ValueError('必须选择供应商')
# api_key 加密存储
if data.get('api_key'):
data['api_key'] = encrypt_api_key(data['api_key'])
# 端点下标校验
idxs_raw = data.get('endpoint_ids', '') or ''
if isinstance(idxs_raw, (list, dict)):
idxs_raw = json.dumps(idxs_raw, ensure_ascii=False)
idxs = []
if idxs_raw.strip():
idxs = json.loads(idxs_raw)
if not isinstance(idxs, list):
raise ValueError('endpoint_ids 必须是 JSON 数组')
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe("SELECT endpoints FROM llm_vendor WHERE id=${v}$", {"v": vid})
await sor.sqlExe("COMMIT", {})
if not recs:
raise ValueError('供应商不存在')
eps = []
try:
eps = json.loads(getattr(recs[0], 'endpoints', '') or '') or []
except Exception:
eps = []
for i in idxs:
if not isinstance(i, int) and not (isinstance(i, str) and i.isdigit()):
raise ValueError('endpoint_ids 元素必须是端点下标（数字）')
if int(i) < 0 or int(i) >= len(eps):
raise ValueError('端点下标 %s 超出供应商端点目录（共 %d 个端点）' % (i, len(eps)))
data['endpoint_ids'] = json.dumps([int(i) for i in idxs], ensure_ascii=False) if idxs else ''
data['id'] = getID()
async with db.sqlorContext(dbname) as sor:
await sor.C('llm_account', data)
result['success'] = True
result['message'] = '创建成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def update_llm_account(params_kw):
"""账号更新：api_key 非空才重新加密（空=不改）。"""
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
if data.get('api_key'):
data['api_key'] = encrypt_api_key(data['api_key'])
else:
data.pop('api_key', None)
if 'endpoint_ids' in data:
idxs_raw = data.get('endpoint_ids', '') or ''
if isinstance(idxs_raw, (list, dict)):
idxs_raw = json.dumps(idxs_raw, ensure_ascii=False)
idxs = []
if idxs_raw.strip():
idxs = json.loads(idxs_raw)
if not isinstance(idxs, list):
raise ValueError('endpoint_ids 必须是 JSON 数组')
data['endpoint_ids'] = json.dumps([int(i) for i in idxs], ensure_ascii=False) if idxs else ''
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.U('llm_account', data)
result['success'] = True
result['message'] = '更新成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def delete_llm_account(params_kw):
result = {'success': False, 'message': ''}
try:
aid = (params_kw or {}).get('id', '')
if not aid:
raise ValueError('缺少 id')
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT COUNT(*) AS c FROM llm_model WHERE account_id=${a}$", {"a": aid})
await sor.sqlExe("COMMIT", {})
cnt = int(getattr(recs[0], 'c', 0)) if recs else 0
if cnt > 0:
raise ValueError('该账号被 %d 个模型设为默认账号，请先解除后再删除' % cnt)
await sor.sqlExe("DELETE FROM llm_account WHERE id=${a}$", {"a": aid})
await sor.sqlExe("COMMIT", {})
result['success'] = True
result['message'] = '删除成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def create_llm_model(params_kw):
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
name = (data.get('name', '') or '').strip()
if not name:
raise ValueError('模型注册名不能为空')
if not data.get('vendor_id'):
raise ValueError('必须选择供应商')
dp = data.get('default_params', '') or ''
if isinstance(dp, (list, dict)):
dp = json.dumps(dp, ensure_ascii=False)
if dp.strip():
json.loads(dp) # 校验合法
data['default_params'] = dp
data['id'] = getID()
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe("SELECT id FROM llm_model WHERE name=${n}$", {"n": name})
await sor.sqlExe("COMMIT", {})
if recs:
raise ValueError('模型注册名「%s」已存在（平台内唯一）' % name)
await sor.C('llm_model', data)
result['success'] = True
result['message'] = '创建成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def update_llm_model(params_kw):
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
if 'default_params' in data:
dp = data.get('default_params', '') or ''
if isinstance(dp, (list, dict)):
dp = json.dumps(dp, ensure_ascii=False)
if dp.strip():
json.loads(dp)
data['default_params'] = dp
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
await sor.U('llm_model', data)
result['success'] = True
result['message'] = '更新成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def delete_llm_model(params_kw):
"""模型删除前置检查：被策略引用时禁止。"""
result = {'success': False, 'message': ''}
try:
mid = (params_kw or {}).get('id', '')
if not mid:
raise ValueError('缺少 id')
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT COUNT(*) AS c FROM llm_org_policy WHERE primary_model_id=${m}$", {"m": mid})
await sor.sqlExe("COMMIT", {})
cnt = int(getattr(recs[0], 'c', 0)) if recs else 0
if cnt > 0:
raise ValueError('该模型被 %d 个组织策略设为主模型，请先调整策略' % cnt)
await sor.sqlExe("DELETE FROM llm_model WHERE id=${m}$", {"m": mid})
await sor.sqlExe("COMMIT", {})
result['success'] = True
result['message'] = '删除成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def update_llm_org_policy(params_kw):
"""组织策略保存：校验模型存在、备链去重。"""
result = {'success': False, 'message': ''}
try:
data = _clean(params_kw)
backups = data.get('backup_model_ids', '') or ''
if isinstance(backups, (list, dict)):
backups = json.dumps(backups, ensure_ascii=False)
bl = []
if backups.strip():
bl = json.loads(backups)
if not isinstance(bl, list):
raise ValueError('backup_model_ids 必须是 JSON 数组')
seen = []
for b in bl:
if b and b not in seen:
seen.append(b)
bl = seen
data['backup_model_ids'] = json.dumps(bl, ensure_ascii=False) if bl else ''
primary = data.get('primary_model_id', '') or ''
if primary and primary in bl:
raise ValueError('主模型不能同时出现在备模型链中')
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
if primary:
recs = await sor.sqlExe("SELECT id FROM llm_model WHERE id=${m}$", {"m": primary})
await sor.sqlExe("COMMIT", {})
if not recs:
raise ValueError('主模型不存在或已停用')
org_id = data.get('org_id', '')
if org_id:
recs = await sor.sqlExe("SELECT id FROM llm_org_policy WHERE org_id=${o}$", {"o": org_id})
await sor.sqlExe("COMMIT", {})
if recs:
data['id'] = getattr(recs[0], 'id', '')
await sor.U('llm_org_policy', data)
else:
data['id'] = getID()
await sor.C('llm_org_policy', data)
else:
await sor.U('llm_org_policy', data)
result['success'] = True
result['message'] = '保存成功'
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def llm_usage_query(params_kw):
"""用量查询（列表页工具栏）：按机构/模型/账号/状态/时间范围过滤，返回汇总+明细。"""
result = {'success': False, 'message': ''}
try:
pk = params_kw or {}
conds = ["1=1"]
params = {}
if pk.get('org_id'):
conds.append("org_id=${o}$"); params['o'] = pk['org_id']
if pk.get('model_id'):
conds.append("model_id=${m}$"); params['m'] = pk['model_id']
if pk.get('account_id'):
conds.append("account_id=${a}$"); params['a'] = pk['account_id']
if pk.get('status'):
conds.append("status=${s}$"); params['s'] = pk['status']
if pk.get('date_from'):
conds.append("created_at>=${df}$"); params['df'] = pk['date_from']
if pk.get('date_to'):
conds.append("created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59'
where = " AND ".join(conds)
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT u.id, u.org_id, u.user_id, u.model_id, m.name AS model_name, "
"u.account_id, a.name AS account_name, u.endpoint_region, "
"u.req_tokens, u.resp_tokens, u.cost, u.charge, u.ppid, u.task_ref, "
"u.status, u.note, u.created_at "
"FROM llm_usage u "
"LEFT JOIN llm_model m ON m.id=u.model_id "
"LEFT JOIN llm_account a ON a.id=u.account_id "
"WHERE %s ORDER BY u.created_at DESC LIMIT 200" % where, params)
sums = await sor.sqlExe(
"SELECT COUNT(*) AS cnt, COALESCE(SUM(req_tokens),0) AS req_t, "
"COALESCE(SUM(resp_tokens),0) AS resp_t, COALESCE(SUM(cost),0) AS cost_s, "
"COALESCE(SUM(charge),0) AS charge_s FROM llm_usage WHERE %s" % where, params)
await sor.sqlExe("COMMIT", {})
rows = []
for r in (recs or []):
d = {}
for k, v in vars(r).items():
if not callable(v):
d[k] = v
rows.append(d)
s = sums[0] if sums else None
result['success'] = True
result['rows'] = rows
result['summary'] = {
'count': int(getattr(s, 'cnt', 0)) if s else 0,
'req_tokens': int(getattr(s, 'req_t', 0)) if s else 0,
'resp_tokens': int(getattr(s, 'resp_t', 0)) if s else 0,
'cost': round(float(getattr(s, 'cost_s', 0)), 6) if s else 0,
'charge': round(float(getattr(s, 'charge_s', 0)), 6) if s else 0,
}
except Exception as e:
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
async def llm_dashboard(params_kw):
"""治理总览（首页卡片数据源）：各机构策略/余额/近期用量。"""
result = {'success': True}
try:
db, dbname = _get_sor()
async with db.sqlorContext(dbname) as sor:
vendors = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_vendor WHERE status='active'", {})
accounts = await sor.sqlExe(
"SELECT COUNT(*) AS c, COALESCE(SUM(balance),0) AS b FROM llm_account WHERE status='active'", {})
models = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_model WHERE status='active'", {})
orgs = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_org_policy WHERE status='active'", {})
usage = await sor.sqlExe(
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='ok'", {})
await sor.sqlExe("COMMIT", {})
result['vendors'] = int(getattr(vendors[0], 'c', 0)) if vendors else 0
result['accounts'] = int(getattr(accounts[0], 'c', 0)) if accounts else 0
result['account_balance'] = round(float(getattr(accounts[0], 'b', 0)), 4) if accounts else 0
result['models'] = int(getattr(models[0], 'c', 0)) if models else 0
result['org_policies'] = int(getattr(orgs[0], 'c', 0)) if orgs else 0
result['usage_calls'] = int(getattr(usage[0], 'c', 0)) if usage else 0
result['usage_charge'] = round(float(getattr(usage[0], 'ch', 0)), 4) if usage else 0
except Exception as e:
result['success'] = False
result['message'] = str(e)
return json.dumps(result, ensure_ascii=False, default=str)
def load_llm():
"""模块加载：注册全部函数到 ServerEnv。"""
load_gateway()
env = ServerEnv()
env.create_llm_vendor = create_llm_vendor
env.update_llm_vendor = update_llm_vendor
env.delete_llm_vendor = delete_llm_vendor
env.create_llm_account = create_llm_account
env.update_llm_account = update_llm_account
env.delete_llm_account = delete_llm_account
env.create_llm_model = create_llm_model
env.update_llm_model = update_llm_model
env.delete_llm_model = delete_llm_model
env.update_llm_org_policy = update_llm_org_policy
env.llm_usage_query = llm_usage_query
env.llm_dashboard = llm_dashboard
logger.info("[pipeline_llm] v1.0.0 loaded — 模型治理模块就绪")

17
pyproject.toml Normal file
View File

@ -0,0 +1,17 @@
[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "pipeline_llm"
version = "1.0.0"
description = "Pipeline platform model governance module — vendors/accounts/endpoints, org policies, quotas, failover rotation, dual-dimension accounting"
requires-python = ">=3.8"
dependencies = [
"aiohttp",
"redis",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["pipeline_llm*"]

36
scripts/load_path.py Normal file
View File

@ -0,0 +1,36 @@
#!/usr/bin/env python3
"""pipeline_llm module RBAC permission registration.
Run from app root: py3/bin/python pkgs/pipeline-llm/scripts/load_path.py
Requires app-root wrapper set_role_perm.py.
"""
import subprocess
MOD = "pipeline-llm"
PATHS_ANY = [] # 本模块无免登录资源（全部需登录）
PATHS_LOGINED = [
f"/{MOD}/",
f"/{MOD}/index.ui",
]
# api/ 与生成的 CRUD 目录用通配符一次覆盖
PATHS_WILDCARD_LOGINED = [
f"/{MOD}/**",
]
def register_paths():
for path in PATHS_ANY:
subprocess.run(["py3/bin/python", "set_role_perm.py", "any", path])
print(f" any: {path}")
for path in PATHS_LOGINED + PATHS_WILDCARD_LOGINED:
subprocess.run(["py3/bin/python", "set_role_perm.py", "logined", path])
print(f" logined: {path}")
if __name__ == "__main__":
print(f"=== {MOD} (pipeline_llm) RBAC registration ===")
register_paths()
print("Done.")

10
wwwroot/api/get_llm_account_options.dspy Normal file
View File

@ -0,0 +1,10 @@
# get_llm_account_options.dspy — 供应商账号下拉（active）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT id, name FROM llm_account WHERE status='active' ORDER BY name", {})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "id", ""), "text": getattr(r, "name", "")} for r in (recs or [])]

11
wwwroot/api/get_llm_capability_options.dspy Normal file
View File

@ -0,0 +1,11 @@
# get_llm_capability_options.dspy — appcodes_kv 下拉选项（parentid=llm_capability）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT k, v FROM appcodes_kv WHERE parentid=${p}$ ORDER BY k",
{"p": "llm_capability"})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "k", ""), "text": getattr(r, "v", "") or getattr(r, "k", "")} for r in (recs or [])]

11
wwwroot/api/get_llm_endpoint_pref_options.dspy Normal file
View File

@ -0,0 +1,11 @@
# get_llm_endpoint_pref_options.dspy — appcodes_kv 下拉选项（parentid=llm_endpoint_pref）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT k, v FROM appcodes_kv WHERE parentid=${p}$ ORDER BY k",
{"p": "llm_endpoint_pref"})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "k", ""), "text": getattr(r, "v", "") or getattr(r, "k", "")} for r in (recs or [])]

10
wwwroot/api/get_llm_model_options.dspy Normal file
View File

@ -0,0 +1,10 @@
# get_llm_model_options.dspy — 模型下拉（active，策略主备模型用）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT id, name FROM llm_model WHERE status='active' ORDER BY name", {})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "id", ""), "text": getattr(r, "name", "")} for r in (recs or [])]

10
wwwroot/api/get_llm_ppid_options.dspy Normal file
View File

@ -0,0 +1,10 @@
# get_llm_ppid_options.dspy — 定价项目下拉（pricing_program，挂定价用）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT id, name FROM pricing_program ORDER BY name", {})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "id", ""), "text": getattr(r, "name", "")} for r in (recs or [])]

11
wwwroot/api/get_llm_protocol_options.dspy Normal file
View File

@ -0,0 +1,11 @@
# get_llm_protocol_options.dspy — appcodes_kv 下拉选项（parentid=llm_protocol）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT k, v FROM appcodes_kv WHERE parentid=${p}$ ORDER BY k",
{"p": "llm_protocol"})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "k", ""), "text": getattr(r, "v", "") or getattr(r, "k", "")} for r in (recs or [])]

11
wwwroot/api/get_llm_status_options.dspy Normal file
View File

@ -0,0 +1,11 @@
# get_llm_status_options.dspy — appcodes_kv 下拉选项（parentid=llm_status）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT k, v FROM appcodes_kv WHERE parentid=${p}$ ORDER BY k",
{"p": "llm_status"})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "k", ""), "text": getattr(r, "v", "") or getattr(r, "k", "")} for r in (recs or [])]

10
wwwroot/api/get_llm_vendor_options.dspy Normal file
View File

@ -0,0 +1,10 @@
# get_llm_vendor_options.dspy — 供应商下拉（active）
dbname = get_module_dbname('pipeline_llm')
try:
async with get_sor_context(request._run_ns, dbname) as sor:
recs = await sor.sqlExe(
"SELECT id, name FROM llm_vendor WHERE status='active' ORDER BY name", {})
await sor.sqlExe("COMMIT", {})
except Exception as e:
return {"success": False, "message": str(e), "data": []}
return [{"value": getattr(r, "id", ""), "text": getattr(r, "name", "")} for r in (recs or [])]

11
wwwroot/api/llm_account_recharge.dspy Normal file
View File

@ -0,0 +1,11 @@
# llm_account_recharge.dspy — 供应商账号充值（成本侧钱包）
uid = await get_user()
if not uid:
return {"success": False, "message": "未登录"}
account_id = (params_kw or {}).get('account_id', '') or (params_kw or {}).get('id', '')
amount = (params_kw or {}).get('amount', 0)
note = (params_kw or {}).get('note', '') or ''
if not account_id:
return {"success": False, "message": "缺少账号"}
ok, msg = await llm_recharge_account(account_id, amount, note, uid)
return {"success": ok, "message": msg}

5
wwwroot/api/llm_dashboard.dspy Normal file
View File

@ -0,0 +1,5 @@
# llm_dashboard.dspy — 模型治理总览数据
uid = await get_user()
if not uid:
return {"success": False, "message": "未登录"}
return json.loads(await llm_dashboard({}))

48
wwwroot/api/llm_dashboard_widget.dspy Normal file
View File

@ -0,0 +1,48 @@
# llm_dashboard_widget.dspy — 模型治理总览（顶部统计条，返回 Bricks widget）
dbname = get_module_dbname('pipeline_llm')
data = {"vendors": 0, "accounts": 0, "account_balance": 0, "models": 0,
"org_policies": 0, "usage_calls": 0, "usage_charge": 0}
try:
async with get_sor_context(request._run_ns, dbname) as sor:
vendors = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_vendor WHERE status='active'", {})
accounts = await sor.sqlExe(
"SELECT COUNT(*) AS c, COALESCE(SUM(balance),0) AS b FROM llm_account WHERE status='active'", {})
models = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_model WHERE status='active'", {})
orgs = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_org_policy WHERE status='active'", {})
usage = await sor.sqlExe(
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='ok'", {})
await sor.sqlExe("COMMIT", {})
data['vendors'] = int(getattr(vendors[0], 'c', 0)) if vendors else 0
data['accounts'] = int(getattr(accounts[0], 'c', 0)) if accounts else 0
data['account_balance'] = round(float(getattr(accounts[0], 'b', 0)), 4) if accounts else 0
data['models'] = int(getattr(models[0], 'c', 0)) if models else 0
data['org_policies'] = int(getattr(orgs[0], 'c', 0)) if orgs else 0
data['usage_calls'] = int(getattr(usage[0], 'c', 0)) if usage else 0
data['usage_charge'] = round(float(getattr(usage[0], 'ch', 0)), 4) if usage else 0
except Exception:
pass
stats = [
('供应商', str(data['vendors'])),
('账号', str(data['accounts'])),
('账号总余额', str(data['account_balance'])),
('模型', str(data['models'])),
('组织策略', str(data['org_policies'])),
('调用次数', str(data['usage_calls'])),
('累计消费', str(data['usage_charge'])),
]
boxes = []
for label, val in stats:
boxes.append({
"widgettype": "VBox",
"options": {"css": "card", "padding": "12px 20px", "minWidth": "110px"},
"subwidgets": [
{"widgettype": "Text", "options": {"text": val, "cfontsize": 1.6, "color": "#1e293b", "halign": "middle"}},
{"widgettype": "Text", "options": {"text": label, "cfontsize": 0.9, "color": "#64748b", "halign": "middle"}},
]
})
return {
"widgettype": "HBox",
"options": {"gap": "12px", "padding": "0 0 16px 0", "wrap": "wrap"},
"subwidgets": boxes
}

11
wwwroot/api/llm_org_recharge.dspy Normal file
View File

@ -0,0 +1,11 @@
# llm_org_recharge.dspy — 组织池充值（消费侧钱包）
uid = await get_user()
if not uid:
return {"success": False, "message": "未登录"}
org_id = (params_kw or {}).get('org_id', '') or ''
amount = (params_kw or {}).get('amount', 0)
note = (params_kw or {}).get('note', '') or ''
if not org_id:
return {"success": False, "message": "缺少机构"}
ok, msg = await llm_recharge_org(org_id, amount, note, uid)
return {"success": ok, "message": msg}

6
wwwroot/api/llm_usage_query.dspy Normal file
View File

@ -0,0 +1,6 @@
# llm_usage_query.dspy — 用量查询统计（工具栏入口，透传过滤条件）
uid = await get_user()
if not uid:
return {"success": False, "message": "未登录"}
pk = dict(params_kw or {})
return json.loads(await llm_usage_query(pk))

271
wwwroot/index.ui Normal file
View File

@ -0,0 +1,271 @@
{
"widgettype": "VBox",
"options": {
"width": "100%",
"height": "100%",
"padding": "0"
},
"subwidgets": [
{
"widgettype": "HBox",
"options": {
"width": "100%",
"alignItems": "center",
"marginBottom": "24px"
},
"subwidgets": [
{
"widgettype": "Title2",
"options": {
"text": "模型治理"
}
},
{
"widgettype": "Filler"
},
{
"widgettype": "Text",
"options": {
"text": "供应商/账号/模型 · 容错策略 · 限额限流 · 双维度记账",
"cfontsize": 1.2
}
}
]
},
{
"widgettype": "urlwidget",
"options": {
"url": "{{entire_url('/pipeline-llm/api/llm_dashboard_widget.dspy')}}"
}
},
{
"widgettype": "VScrollPanel",
"options": {
"css": "filler"
},
"subwidgets": [
{
"widgettype": "VBox",
"options": {
"spacing": 24
},
"subwidgets": [
{
"widgettype": "ResponsableBox",
"options": {
"gap": "16px",
"minWidth": "300px"
},
"subwidgets": [
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_vendor')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "供应商与端点", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "供应商协议 + 全球端点目录（国内/国际，配置一次全账号共用）", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_account')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "供应商账号（钱包）", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "多账号独立充值记账、选用端点；轮转突破供应商限流", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_model')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "模型注册", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "能力分类（t2t/…）、挂定价项目（ppid）、成本/售价", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_api_profile')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "适配模板", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "协议×能力形态的请求/响应模板（按形态去重，不按模型开）", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_org_policy')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "组织容错策略", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "组织级主模型 + 备模型链 + 端点偏好（prefer/must）", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_org_quota')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "组织限额限流", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "组织池充值/余额 + 组织 QPM 上限", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_user_quota')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "个人限额限流", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "个人额度上限 + 个人 QPM（共享组织池模式）", "cfontsize": 1.2}}
]
},
{
"widgettype": "VBox",
"options": {
"css": "card",
"cwidth": 25,
"padding": "16px",
"cursor": "pointer"
},
"binds": [
{
"wid": "self",
"event": "click",
"actiontype": "urlwidget",
"target": "app.llm_content",
"options": {
"url": "{{entire_url('/pipeline-llm/llm_usage')}}"
},
"mode": "replace"
}
],
"subwidgets": [
{"widgettype": "Title4", "options": {"text": "用量流水", "marginBottom": "8px"}},
{"widgettype": "Text", "options": {"text": "双维度记账：成本侧（账号）+ 消费侧（组织/个人）", "cfontsize": 1.2}}
]
}
]
}
]
},
{
"widgettype": "VBox",
"id": "llm_content"
}
]
}
]
}