875 lines
40 KiB
Python
875 lines
40 KiB
Python
"""pipeline_llm 模块入口 — 模型治理（供应商/账号/模型/组织策略/限额限流/记账）。
|
||
|
||
宿主集成：应用入口调用 load_pipeline_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 html
|
||
import json
|
||
import logging
|
||
import os
|
||
|
||
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,
|
||
)
|
||
from .inference import chat_inference, list_models, load_inference
|
||
from .selection import load_selection
|
||
from .product_interface import load_product_category_product
|
||
|
||
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()
|
||
|
||
|
||
# ────────────────────── 端点目录输入归一化 ──────────────────────
|
||
|
||
def normalize_endpoints(text):
|
||
"""端点目录输入归一化：支持「一行一个 URL」或 JSON 数组两种写法。
|
||
|
||
行格式：URL [region=domestic|international] [timeout=秒]，例如：
|
||
https://dashscope.aliyuncs.com
|
||
https://api.example.com region=international timeout=30
|
||
缺省 region=domestic、timeout=60。存储格式恒为 JSON 数组
|
||
[{base_url, region, timeout}]（治理链按此读取）。
|
||
非法输入抛 ValueError（中文提示，直接展示给用户）。
|
||
|
||
2026-09-12 端点可互换约定：base_url 一律归一为主机根（scheme+域名）——
|
||
输入带路径段（旧数据/旧习惯的 /api/v1、/compatible-mode/v1 等）时确定性
|
||
剥掉；接口路径归 profile.path 承载（完整相对路径）。同 (主机根, region)
|
||
去重合并，timeout 取最大值（单端点须覆盖该主机全部接口形态的超时需求）。
|
||
"""
|
||
from urllib.parse import urlparse
|
||
s = (text or '').strip()
|
||
if not s:
|
||
return ''
|
||
if s.startswith('['):
|
||
eps = json.loads(s)
|
||
if not isinstance(eps, list):
|
||
raise ValueError('endpoints 必须是 JSON 数组')
|
||
else:
|
||
eps = []
|
||
for ln_no, ln in enumerate([l.strip() for l in s.splitlines()], 1):
|
||
if not ln:
|
||
continue
|
||
parts = ln.split()
|
||
url = parts[0].rstrip('/')
|
||
if not (url.startswith('http://') or url.startswith('https://')):
|
||
raise ValueError('第 %d 行不是合法 URL（须以 http(s):// 开头）：%s' % (ln_no, parts[0]))
|
||
ep = {'base_url': url, 'region': 'domestic', 'timeout': 60}
|
||
for kv in parts[1:]:
|
||
if '=' not in kv:
|
||
raise ValueError('第 %d 行选项 %s 应为 key=value 形式（region= / timeout=）' % (ln_no, kv))
|
||
k, v = kv.split('=', 1)
|
||
if k == 'region':
|
||
if v not in ('domestic', 'international'):
|
||
raise ValueError('第 %d 行 region 只能是 domestic 或 international' % ln_no)
|
||
ep['region'] = v
|
||
elif k == 'timeout':
|
||
ep['timeout'] = int(v)
|
||
else:
|
||
raise ValueError('第 %d 行未知选项 %s（支持 region= / timeout=）' % (ln_no, k))
|
||
eps.append(ep)
|
||
merged = []
|
||
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))
|
||
pu = urlparse((ep.get('base_url') or '').rstrip('/'))
|
||
if not pu.netloc:
|
||
raise ValueError('端点 #%d base_url 非法（无主机名）：%s' % (i + 1, ep.get('base_url')))
|
||
root = pu.scheme + '://' + pu.netloc
|
||
region = ep.get('region') or 'domestic'
|
||
try:
|
||
to = int(ep.get('timeout') or 60)
|
||
except (TypeError, ValueError):
|
||
to = 60
|
||
hit = None
|
||
for m in merged:
|
||
if m['base_url'] == root and m['region'] == region:
|
||
hit = m
|
||
break
|
||
if hit is None:
|
||
m = {'base_url': root, 'region': region, 'timeout': to}
|
||
if ep.get('protocol'):
|
||
m['protocol'] = ep['protocol']
|
||
merged.append(m)
|
||
else:
|
||
hit['timeout'] = max(hit['timeout'], to)
|
||
return json.dumps(merged, ensure_ascii=False)
|
||
|
||
|
||
def _norm_endpoint_ids(raw):
|
||
"""endpoint_ids 归一化：JSON 数组或逗号分隔数字均可，返回 JSON 数组串。"""
|
||
if isinstance(raw, (list, dict)):
|
||
raw = json.dumps(raw, ensure_ascii=False)
|
||
raw = (raw or '').strip()
|
||
if not raw:
|
||
return ''
|
||
if raw.startswith('['):
|
||
idxs = json.loads(raw)
|
||
else:
|
||
idxs = [x.strip() for x in raw.split(',') if x.strip()]
|
||
if not isinstance(idxs, list):
|
||
raise ValueError('endpoint_ids 必须是 JSON 数组')
|
||
for i in idxs:
|
||
if not (isinstance(i, int) or (isinstance(i, str) and i.lstrip('-').isdigit())):
|
||
raise ValueError('endpoint_ids 元素必须是端点下标（数字）')
|
||
return json.dumps([int(i) for i in idxs], ensure_ascii=False)
|
||
|
||
|
||
MASK = '******' # 列表脱敏占位；编辑表单原样提交视为「不修改」
|
||
|
||
|
||
# ────────────────────── 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)
|
||
data['endpoints'] = normalize_endpoints(data.get('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:
|
||
data['endpoints'] = normalize_endpoints(data.get('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') and data['api_key'] != MASK:
|
||
data['api_key'] = encrypt_api_key(data['api_key'])
|
||
else:
|
||
data.pop('api_key', None)
|
||
# 端点下标校验（JSON 或逗号分隔均可）
|
||
data['endpoint_ids'] = _norm_endpoint_ids(data.get('endpoint_ids', ''))
|
||
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 json.loads(data['endpoint_ids'] or '[]'):
|
||
if int(i) < 0 or int(i) >= len(eps):
|
||
raise ValueError('端点下标 %s 超出供应商端点目录（共 %d 个端点）' % (i, len(eps)))
|
||
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 非空才重新加密（空=不改）。
|
||
|
||
vendor_id 归属不可改：账号在哪个供应商下创建就固定在哪个供应商下，
|
||
防止子表（按供应商过滤）入口下篡改表单把账号挪到别的供应商。
|
||
"""
|
||
result = {'success': False, 'message': ''}
|
||
try:
|
||
data = _clean(params_kw)
|
||
aid = data.get('id', '')
|
||
if not aid:
|
||
raise ValueError('缺少 id')
|
||
data.pop('vendor_id', None)
|
||
if data.get('api_key') and data['api_key'] != MASK:
|
||
data['api_key'] = encrypt_api_key(data['api_key'])
|
||
else:
|
||
data.pop('api_key', None)
|
||
if 'endpoint_ids' in data:
|
||
data['endpoint_ids'] = _norm_endpoint_ids(data.get('endpoint_ids', ''))
|
||
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):
|
||
"""账号删除：被模型引用时禁止；带 vendor_id 时校验归属（子表范围隔离）。"""
|
||
result = {'success': False, 'message': ''}
|
||
try:
|
||
pk = params_kw or {}
|
||
aid = pk.get('id', '')
|
||
if not aid:
|
||
raise ValueError('缺少 id')
|
||
db, dbname = _get_sor()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT vendor_id FROM llm_account WHERE id=${a}$", {"a": aid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
raise ValueError('账号不存在')
|
||
# 子表入口会带 vendor_id：校验账号确实属于该供应商，防跨供应商误删/越权删
|
||
expect_vid = pk.get('vendor_id', '')
|
||
if expect_vid and getattr(recs[0], 'vendor_id', '') != expect_vid:
|
||
raise ValueError('账号不属于该供应商，禁止删除')
|
||
cnt_recs = await sor.sqlExe(
|
||
"SELECT COUNT(*) AS c FROM llm_model WHERE account_id=${a}$", {"a": aid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
cnt = int(getattr(cnt_recs[0], 'c', 0)) if cnt_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
|
||
# 后续步骤模板链：JSON 数组校验
|
||
qp = data.get('query_profile_ids', '') or ''
|
||
if isinstance(qp, (list, dict)):
|
||
qp = json.dumps(qp, ensure_ascii=False)
|
||
if qp.strip():
|
||
ql = json.loads(qp)
|
||
if not isinstance(ql, list):
|
||
raise ValueError('query_profile_ids 必须是 JSON 数组')
|
||
data['query_profile_ids'] = qp
|
||
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
|
||
if 'query_profile_ids' in data:
|
||
qp = data.get('query_profile_ids', '') or ''
|
||
if isinstance(qp, (list, dict)):
|
||
qp = json.dumps(qp, ensure_ascii=False)
|
||
if qp.strip():
|
||
ql = json.loads(qp)
|
||
if not isinstance(ql, list):
|
||
raise ValueError('query_profile_ids 必须是 JSON 数组')
|
||
data['query_profile_ids'] = qp
|
||
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 {}
|
||
# 条件列统一加 u. 前缀（2026-09-10）：JOIN llm_model/llm_account/sd_projects
|
||
# 后 org_id/status/created_at 多表同名，无前缀会触发 MariaDB 1052 歧义
|
||
conds = ["1=1"]
|
||
params = {}
|
||
if pk.get('org_id'):
|
||
conds.append("u.org_id=${o}$"); params['o'] = pk['org_id']
|
||
if pk.get('model_id'):
|
||
conds.append("u.model_id=${m}$"); params['m'] = pk['model_id']
|
||
if pk.get('account_id'):
|
||
conds.append("u.account_id=${a}$"); params['a'] = pk['account_id']
|
||
if pk.get('status'):
|
||
conds.append("u.status=${s}$"); params['s'] = pk['status']
|
||
# 项目过滤（2026-09-10 用户定夺：按项目统计费用）；
|
||
# 不传/null = 不区分项目（全部），'0' = 精确查非项目调用
|
||
_pid = pk.get('project_id')
|
||
if _pid:
|
||
conds.append("u.project_id=${pid}$"); params['pid'] = _pid
|
||
if pk.get('date_from'):
|
||
conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from']
|
||
if pk.get('date_to'):
|
||
conds.append("u.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(
|
||
# token 数从 usages JSON 提取（req/resp_tokens 冗余列已删 m0021，
|
||
# usages 是唯一事实源）；JSON_VALID 守卫防脏行报错；出参键名保持
|
||
# req_tokens/resp_tokens 兼容既有前端
|
||
"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, "
|
||
"CASE WHEN JSON_VALID(u.usages) THEN "
|
||
"COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') "
|
||
"ELSE '0' END AS req_tokens, "
|
||
"CASE WHEN JSON_VALID(u.usages) THEN "
|
||
"COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') "
|
||
"ELSE '0' END AS resp_tokens, "
|
||
"u.cost, u.charge, u.ppid, u.project_id, "
|
||
"CASE WHEN u.project_id='0' THEN '（非项目调用）' "
|
||
"ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, "
|
||
"u.task_ref, u.call_id, "
|
||
"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 "
|
||
"LEFT JOIN sd_projects sp ON sp.id=u.project_id "
|
||
"WHERE %s ORDER BY u.created_at DESC LIMIT 200" % where, params)
|
||
# token 汇总走 JSON_EXTRACT（列已删 m0021）：CAST DECIMAL 防字符串数字；
|
||
# 非法/缺失 JSON 行按 0 计入，不中断汇总
|
||
sums = await sor.sqlExe(
|
||
"SELECT COUNT(*) AS cnt, "
|
||
"COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN "
|
||
"CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.prompt_tokens')),'0') "
|
||
"AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, "
|
||
"COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN "
|
||
"CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.completion_tokens')),'0') "
|
||
"AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t, "
|
||
"COALESCE(SUM(cost),0) AS cost_s, "
|
||
"COALESCE(SUM(charge),0) AS charge_s FROM llm_usage u WHERE %s" % where, params)
|
||
await sor.sqlExe("COMMIT", {})
|
||
rows = []
|
||
for r in (recs or []):
|
||
# sqlor 行是 DictObject（dict 子类），必须 dict() 拷贝——
|
||
# vars() 取到的是空实例 __dict__（同 gateway._row_to_dict 根因）
|
||
try:
|
||
d = dict(r)
|
||
except (TypeError, ValueError):
|
||
d = {k: v for k, v in vars(r).items() if not callable(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_project_cost_query(params_kw):
|
||
"""按项目统计费用（2026-09-10 用户定夺）：llm_usage.project_id GROUP BY。
|
||
|
||
过滤：org_id / date_from / date_to / status（缺省统计全部状态）。
|
||
分组：project_id（'0' = 非项目调用哨兵值，一组显示'（非项目调用）'；
|
||
其余 JOIN sd_projects 取项目名，项目已删显示'(项目已删除)'）。
|
||
project_id 过滤：不传/null = 不区分项目（全部分组都出）。
|
||
汇总口径：calls 调用次数 / cost 成本合计 / charge 客户应付合计 /
|
||
req_t、resp_t token 合计（usages JSON 唯一事实源，JSON_VALID 守卫）。
|
||
"""
|
||
result = {'success': False, 'message': ''}
|
||
try:
|
||
pk = params_kw or {}
|
||
conds = ["1=1"]
|
||
params = {}
|
||
if pk.get('org_id'):
|
||
conds.append("u.org_id=${o}$"); params['o'] = pk['org_id']
|
||
if pk.get('status'):
|
||
conds.append("u.status=${s}$"); params['s'] = pk['status']
|
||
if pk.get('date_from'):
|
||
conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from']
|
||
if pk.get('date_to'):
|
||
conds.append("u.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.project_id, "
|
||
"CASE WHEN u.project_id='0' THEN '（非项目调用）' "
|
||
"ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, "
|
||
"COUNT(*) AS calls, "
|
||
"COALESCE(SUM(u.cost),0) AS cost_s, "
|
||
"COALESCE(SUM(u.charge),0) AS charge_s, "
|
||
"COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN "
|
||
"CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') "
|
||
"AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, "
|
||
"COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN "
|
||
"CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') "
|
||
"AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t "
|
||
"FROM llm_usage u "
|
||
"LEFT JOIN sd_projects sp ON sp.id=u.project_id "
|
||
"WHERE %s GROUP BY u.project_id, sp.name "
|
||
"ORDER BY cost_s DESC, calls DESC LIMIT 500" % where, params)
|
||
await sor.sqlExe("COMMIT", {})
|
||
rows = []
|
||
for r0 in (recs or []):
|
||
try:
|
||
d = dict(r0)
|
||
except (TypeError, ValueError):
|
||
d = {k: v for k, v in vars(r0).items() if not callable(v)}
|
||
rows.append({
|
||
'project_id': d.get('project_id'),
|
||
'project_name': d.get('project_name') or '',
|
||
'calls': int(d.get('calls', 0) or 0),
|
||
'cost': round(float(d.get('cost_s', 0) or 0), 6),
|
||
'charge': round(float(d.get('charge_s', 0) or 0), 6),
|
||
'req_tokens': int(d.get('req_t', 0) or 0),
|
||
'resp_tokens': int(d.get('resp_t', 0) or 0),
|
||
})
|
||
result['success'] = True
|
||
result['rows'] = rows
|
||
result['summary'] = {
|
||
'project_count': len([x for x in rows
|
||
if x['project_id'] and x['project_id'] != '0']),
|
||
'calls': sum(x['calls'] for x in rows),
|
||
'cost': round(sum(x['cost'] for x in rows), 6),
|
||
'charge': round(sum(x['charge'] for x in rows), 6),
|
||
}
|
||
except Exception as e:
|
||
result['message'] = str(e)
|
||
return json.dumps(result, ensure_ascii=False, default=str)
|
||
|
||
|
||
# ────────────────────── 调用原文追踪：「查看IO」弹窗 ──────────────────────
|
||
|
||
_IO_BODY_MAX = 150000 # 弹窗单侧最大展示字符（完整原文仍在盘上，超长提示文件路径）
|
||
|
||
_KIND_TEXT = {'chat': '对话', 'gen': '同步生成', 'submit': '异步提交', 'query': '任务轮询'}
|
||
|
||
|
||
def _read_trace_file(rel_path):
|
||
"""读 trace 原文文件。rel_path 为库中 filesroot 相对路径（llm_trace/<日>/<call>_<seq>.req.json）。
|
||
|
||
安全铁律：路径来自数据库，realpath 后必须落在 llm_trace 目录内，逃逸=拒绝访问
|
||
（防路径穿越；realpath 同时防符号链接指向目录外）。返回 (text, err)。
|
||
"""
|
||
from .trace import _filesroot, trace_root
|
||
if not rel_path:
|
||
return '', '无原文文件（该侧落盘失败或已过保留期被清理）'
|
||
root = os.path.realpath(trace_root())
|
||
real = os.path.realpath(os.path.join(_filesroot(), str(rel_path)))
|
||
if not real.startswith(root + os.sep):
|
||
return '', '[非法路径，拒绝访问]'
|
||
try:
|
||
with open(real, 'r', encoding='utf-8') as f:
|
||
return f.read(), ''
|
||
except Exception as e:
|
||
return '', '[文件读取失败: %s]' % str(e)[:120]
|
||
|
||
|
||
def _pretty_or_raw(text):
|
||
"""落盘 JSON 美化（indent=2）便于阅读；非合法 JSON 原样返回。"""
|
||
try:
|
||
return json.dumps(json.loads(text), ensure_ascii=False, indent=2)
|
||
except Exception:
|
||
return text
|
||
|
||
|
||
def _msg_widget(message):
|
||
return {"widgettype": "Message",
|
||
"options": {"title": "提示", "message": message,
|
||
"cwidth": 24, "cheight": 8, "timeout": 3}}
|
||
|
||
|
||
async def llm_call_trace_io(params_kw, user_org_id='0'):
|
||
"""「查看IO」弹窗内容组装：TabPanel（上下文摘要/请求原文/响应原文）。
|
||
|
||
dspy 薄代理调本函数（读表+读文件+组装 widget 全在模块侧）。
|
||
机构隔离：平台业主（org_id='0'）可看全部；普通机构只看本机构或 org_id 为空的行。
|
||
返回 widget 描述 dict（外层 PopupWindow 壳由列表页 binds popup_options 生成，
|
||
此处再包 PopupWindow 会套娃）。
|
||
"""
|
||
pk = params_kw or {}
|
||
trace_id = str(pk.get('id') or '').strip()
|
||
if not trace_id or len(trace_id) > 32:
|
||
return _msg_widget('缺少或非法的 trace id 参数')
|
||
|
||
db, dbname = _get_sor()
|
||
|
||
def _row2dict(r):
|
||
# sqlExe 行可能是 DictObject(dict 子类) 或普通对象——与 llm_usage_query 同款双兜底
|
||
try:
|
||
return dict(r)
|
||
except (TypeError, ValueError):
|
||
return {k: v for k, v in vars(r).items() if not callable(v)}
|
||
|
||
async with db.sqlorContext(dbname) as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT * FROM llm_call_trace WHERE id=${id}$", {'id': trace_id})
|
||
row = _row2dict(recs[0]) if recs else None
|
||
batch_rows, usage_rows = [], []
|
||
if row and row.get('call_id'):
|
||
call_id = row['call_id']
|
||
recs2 = await sor.sqlExe(
|
||
"SELECT seq, kind, method, status_code, elapsed_ms, url, note "
|
||
"FROM llm_call_trace WHERE call_id=${c}$ ORDER BY seq, created_at",
|
||
{'c': call_id})
|
||
batch_rows = [_row2dict(r) for r in (recs2 or [])]
|
||
recs3 = await sor.sqlExe(
|
||
"SELECT u.status, u.accounting_status, u.cost, u.charge, u.task_ref, "
|
||
"u.project_id, m.name AS model_name "
|
||
"FROM llm_usage u LEFT JOIN llm_model m ON m.id=u.model_id "
|
||
"WHERE u.call_id=${c}$", {'c': call_id})
|
||
usage_rows = [_row2dict(r) for r in (recs3 or [])]
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
if not row:
|
||
return _msg_widget('未找到调用追踪记录（可能已过保留期被清理）')
|
||
row_org = str(row.get('org_id') or '')
|
||
if str(user_org_id or '0') != '0' and row_org not in ('', str(user_org_id)):
|
||
return _msg_widget('无权查看其他机构的调用原文')
|
||
|
||
# 读文件在 DB 块外（行/批次/usage 已全部拷贝为本地 dict，不再依赖 sor）
|
||
req_text, req_err = _read_trace_file(row.get('req_path') or '')
|
||
resp_text, resp_err = _read_trace_file(row.get('resp_path') or '')
|
||
if req_err:
|
||
req_text = req_err
|
||
else:
|
||
req_text = _pretty_or_raw(req_text)
|
||
if len(req_text) > _IO_BODY_MAX:
|
||
req_text = req_text[:_IO_BODY_MAX] + \
|
||
'\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('req_path') or '')
|
||
if resp_err:
|
||
resp_text = resp_err
|
||
else:
|
||
resp_text = _pretty_or_raw(resp_text)
|
||
if len(resp_text) > _IO_BODY_MAX:
|
||
resp_text = resp_text[:_IO_BODY_MAX] + \
|
||
'\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('resp_path') or '')
|
||
|
||
# ── 摘要 tab（本行 + 计费流水上下文 + 同批次往返）──
|
||
lines = [
|
||
'批次 call_id: %s' % (row.get('call_id') or ''),
|
||
'往返 seq/类型: %s / %s' % (
|
||
row.get('seq'), _KIND_TEXT.get(str(row.get('kind') or ''), row.get('kind') or '')),
|
||
'HTTP: %s 状态码: %s 耗时: %s ms' % (
|
||
row.get('method') or '', row.get('status_code'), row.get('elapsed_ms')),
|
||
'URL: %s' % (row.get('url') or ''),
|
||
'机构: %s' % (row_org or '(空)'),
|
||
'时间: %s' % (row.get('created_at') or ''),
|
||
]
|
||
if row.get('note'):
|
||
lines.append('错误摘要: %s' % row['note'])
|
||
if usage_rows:
|
||
lines.append('')
|
||
lines.append('—— 计费流水（llm_usage，同 call_id，%d 条）——' % len(usage_rows))
|
||
for u in usage_rows:
|
||
lines.append('模型: %s 状态: %s 记账: %s 成本: %s 应付: %s 来源: %s 项目: %s' % (
|
||
u.get('model_name') or '(无)', u.get('status') or '',
|
||
u.get('accounting_status') or '-', u.get('cost'), u.get('charge'),
|
||
u.get('task_ref') or '-', u.get('project_id') or '-'))
|
||
if batch_rows:
|
||
lines.append('')
|
||
lines.append('—— 同批次往返（%d 次）——' % len(batch_rows))
|
||
for t in batch_rows:
|
||
lines.append('seq=%s %s %s 状态:%s %sms %s%s' % (
|
||
t.get('seq'),
|
||
_KIND_TEXT.get(str(t.get('kind') or ''), t.get('kind') or ''),
|
||
t.get('method') or '', t.get('status_code'), t.get('elapsed_ms'),
|
||
(t.get('url') or '')[:100],
|
||
(' err: ' + str(t.get('note'))[:80]) if t.get('note') else ''))
|
||
summary_text = '\n'.join(lines)
|
||
|
||
# 摘要 tab 用 Html widget：Text 的 innerHTML 会吞换行且不转义（正文含上游
|
||
# 响应片段时 < > 会破坏结构/注入），Html + <pre> + escape 是唯一安全渲染。
|
||
summary_html = ('<pre style="white-space:pre-wrap;word-break:break-all;'
|
||
'font-size:13px;margin:0;font-family:inherit;">'
|
||
+ html.escape(summary_text) + '</pre>')
|
||
|
||
# mode:null → CodeMirror 纯文本+行号（仓库 3parties 只有 codemirror.js
|
||
# 内建 null 模式和 codemirror-python.js；不指定 mode 会去拉不存在的
|
||
# codemirror-python.js 之外的 mode 文件报 warn，null 是最干净的退化）。
|
||
def _code_tab(name, label, content):
|
||
return {"name": name, "label": label,
|
||
"content": {"widgettype": "CodeEditor",
|
||
"options": {"value": content, "mode": "null",
|
||
"readonly": True, "theme": False,
|
||
"width": "100%", "height": "100%"}}}
|
||
|
||
return {
|
||
"widgettype": "TabPanel",
|
||
"options": {
|
||
"width": "100%", "height": "100%", "tab_pos": "top",
|
||
"items": [
|
||
{"name": "summary", "label": "上下文摘要",
|
||
"content": {"widgettype": "VBox",
|
||
"options": {"width": "100%", "height": "100%",
|
||
"overflow": "auto", "padding": "12px"},
|
||
"subwidgets": [
|
||
{"widgettype": "Html",
|
||
"options": {"width": "100%", "height": "auto",
|
||
"html": summary_html}}]}},
|
||
_code_tab('req', '请求原文（上行）', req_text),
|
||
_code_tab('resp', '响应原文（下行）', resp_text),
|
||
]
|
||
}
|
||
}
|
||
|
||
|
||
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='SUCCEEDED'", {})
|
||
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_pipeline_llm():
|
||
"""模块加载：注册全部函数到 ServerEnv。（规范名，产品层按 load_{resource_module} 约定）"""
|
||
load_gateway()
|
||
load_inference()
|
||
load_selection()
|
||
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_project_cost_query = llm_project_cost_query
|
||
env.llm_call_trace_io = llm_call_trace_io
|
||
env.llm_dashboard = llm_dashboard
|
||
# 产品层标准接口（product_management.product_interface 调度）
|
||
from .product_interface import (
|
||
get_product_display, check_product_availability, check_product_consumable,
|
||
execute_product_service, execute_product_service_stream,
|
||
calculate_product_cost, load_product_category_product,
|
||
)
|
||
env.product_interface = {
|
||
'module_name': 'pipeline_llm',
|
||
'get_product_display': get_product_display,
|
||
'check_product_availability': check_product_availability,
|
||
'check_product_consumable': check_product_consumable,
|
||
'execute_product_service': execute_product_service,
|
||
'execute_product_service_stream': execute_product_service_stream,
|
||
'calculate_product_cost': calculate_product_cost,
|
||
'load_product_category_product': load_product_category_product,
|
||
}
|
||
env.load_product_category_product_llm = load_product_category_product
|
||
# 平台待办 provider（软注册，2026-09-11：账号钱包余额不足 → 属主机构
|
||
# operator 角色催充值待办；pipeline-service 没装/没钩子时只告警不崩，
|
||
# 与 ticket 模块同款模式）
|
||
try:
|
||
from pipeline_service.human_task_capability import register_todo_provider
|
||
from .todos import list_wallet_todos
|
||
register_todo_provider(list_wallet_todos)
|
||
logger.info("[pipeline_llm] wallet todo provider registered")
|
||
except ImportError:
|
||
logger.warning("[pipeline_llm] pipeline_service.register_todo_provider 不可用，"
|
||
"钱包余额待办不会出现在平台待办（模块可独立运行）")
|
||
logger.info("[pipeline_llm] v1.0.0 loaded — 模型治理模块就绪")
|
||
|
||
|
||
# 兼容旧名（宿主曾用 load_llm；规范名为 load_pipeline_llm）
|
||
load_llm = load_pipeline_llm
|