193 lines
8.3 KiB
Python
193 lines
8.3 KiB
Python
"""pipeline_llm.product_interface — 产线模型的产品层接口（照 llmage 样板）。
|
||
|
||
产品模块（product_management）通过 env.product_interface 调度本模块完成
|
||
展示/预检/执行/计费。资源层只算「真实定价（原价）」，完全不碰折扣——
|
||
客户折扣→售价、供应商折扣→成本，由产品层 calculate_product_cost 接
|
||
（见 resource-productization 技能的折扣分工铁律）。
|
||
|
||
resource_ref_id = llm_model.id。
|
||
"""
|
||
|
||
import logging
|
||
|
||
from ahserver.serverenv import ServerEnv
|
||
from sqlor.dbpools import get_sor_context
|
||
|
||
from .gateway import _row_to_dict
|
||
|
||
logger = logging.getLogger("pipeline_llm.product_interface")
|
||
|
||
_MODULE_NAME = 'pipeline_llm'
|
||
|
||
|
||
async def _resolve_model(resource_ref_id):
|
||
"""resource_ref_id → llm_model 行。返回 (model_dict, err)。"""
|
||
env = ServerEnv()
|
||
async with get_sor_context(env, _MODULE_NAME) as sor:
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, vendor_model_id, capability, status, ppid, org_id, "
|
||
"description "
|
||
"FROM llm_model WHERE id=${i}$ LIMIT 1", {"i": resource_ref_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return None, '模型不存在（llm_model.id=%s）' % resource_ref_id
|
||
return _row_to_dict(recs[0]), None
|
||
|
||
|
||
async def get_product_display(resource_ref_id):
|
||
"""展示定价信息。2026-09-04 起模型表不存单价，定价统一走 ppid 定价引擎。"""
|
||
m, err = await _resolve_model(resource_ref_id)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
ppid = m.get('ppid') or ''
|
||
pricing_text = '按定价方案计费（ppid=%s）' % ppid if ppid else '未挂定价方案'
|
||
return {
|
||
'success': True,
|
||
'pricing_text': pricing_text,
|
||
'pricing_detail': {
|
||
'ppid': ppid,
|
||
'capability': m.get('capability') or 't2t',
|
||
'model_id': m.get('vendor_model_id') or m.get('name', ''),
|
||
},
|
||
'extra_info': m.get('description') or '',
|
||
}
|
||
|
||
|
||
async def check_product_availability(resource_ref_id, user_org_id=None):
|
||
"""模型状态 + 定价方案是否就绪。"""
|
||
m, err = await _resolve_model(resource_ref_id)
|
||
if err:
|
||
return {'available': False, 'reason': err}
|
||
if (m.get('status') or '') != 'active':
|
||
return {'available': False, 'reason': '模型已停用（%s）' % m.get('name')}
|
||
if not (m.get('ppid') or ''):
|
||
return {'available': False, 'reason': '模型未挂定价方案（ppid 为空），请先配置定价'}
|
||
return {'available': True, 'reason': ''}
|
||
|
||
|
||
async def check_product_consumable(resource_ref_id, user_id, user_org_id):
|
||
"""消费前综合预检：可用性 + 客户资金账户余额。"""
|
||
avail = await check_product_availability(resource_ref_id, user_org_id)
|
||
if not avail.get('available'):
|
||
return {'consumable': False, 'reason': avail.get('reason', ''),
|
||
'pricing_available': False}
|
||
env = ServerEnv()
|
||
balance = None
|
||
if user_org_id and hasattr(env, 'getCustomerBalance'):
|
||
try:
|
||
acc_dbname = env.get_module_dbname('accounting')
|
||
from sqlor.dbpools import DBPools
|
||
async with DBPools().sqlorContext(acc_dbname) as sor:
|
||
balance = await env.getCustomerBalance(sor, user_org_id)
|
||
except Exception:
|
||
balance = None # accounting 未接入 → 不拦
|
||
if balance is None:
|
||
return {'consumable': True, 'reason': '', 'pricing_available': True}
|
||
if float(balance) <= 0:
|
||
return {'consumable': False, 'reason': '客户资金账户余额不足（%.2f），请先充值' % balance,
|
||
'pricing_available': True}
|
||
return {'consumable': True, 'reason': '', 'pricing_available': True}
|
||
|
||
|
||
async def execute_product_service(resource_ref_id, user_id, user_org_id, request_data):
|
||
"""产品消费引擎入口不承接模型调用（防双重计费）。
|
||
|
||
模型的真实消费路径是产线会话/推理入口（llm_bridge → inference），
|
||
该路径已完成成本侧结算并写 pending 用量流水，由异步三方账出客户账单。
|
||
"""
|
||
return {'success': False, 'status': 'unsupported',
|
||
'message': '产线模型消费走推理入口（会话/llm_v1），不经产品消费引擎'}
|
||
|
||
|
||
async def execute_product_service_stream(resource_ref_id, user_id, user_org_id, request_data):
|
||
"""同 execute_product_service：不承接，防双重计费。"""
|
||
yield {'success': False, 'status': 'unsupported',
|
||
'message': '产线模型消费走推理入口（会话/llm_v1），不经产品消费引擎',
|
||
'done': True}
|
||
|
||
|
||
async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None):
|
||
"""资源层真实定价（原价）。不碰折扣（折扣由产品层按产品精确计算）。
|
||
|
||
usage_data: {'prompt_tokens': N, 'completion_tokens': N}（或按量因子 dict；
|
||
键名唯一契约 = 上游 usage 原文结构，2026-09-10 起 req_tokens/resp_tokens 别名
|
||
随冗余列一并废除）
|
||
"""
|
||
m, err = await _resolve_model(resource_ref_id)
|
||
if err:
|
||
return {'success': False, 'message': err}
|
||
ppid = m.get('ppid') or ''
|
||
if not ppid:
|
||
return {'success': False, 'message': '模型未挂定价方案（ppid 为空）'}
|
||
env = ServerEnv()
|
||
if not hasattr(env, 'buffered_charging'):
|
||
return {'success': False, 'message': '定价引擎（pricing 模块）未加载'}
|
||
usage = dict(usage_data or {})
|
||
try:
|
||
prices = await env.buffered_charging(ppid, usage)
|
||
except Exception as e:
|
||
logger.warning("pipeline_llm: 定价计算失败 ppid=%s: %s", ppid, e)
|
||
return {'success': False, 'message': '定价计算失败: %s' % e}
|
||
if prices is None:
|
||
return {'success': False, 'message': '定价计算返回空（ppid=%s）' % ppid}
|
||
amount = sum(float(getattr(p, 'amount', 0) or 0) for p in prices)
|
||
cost = sum(float(getattr(p, 'cost', 0) or 0) for p in prices)
|
||
return {
|
||
'success': True,
|
||
'original_amount': round(amount, 6),
|
||
'amount': round(amount, 6), # 资源层不碰折扣，产品层再乘客户折扣
|
||
'cost': round(cost, 6),
|
||
'discount': 1.0,
|
||
'pricing_program_id': ppid,
|
||
}
|
||
|
||
|
||
async def load_product_category_product(parent_category_id):
|
||
"""产品导入：把模型注册表包装成产品（标准化格式）。
|
||
|
||
类别 = 能力类型（t2t/t2i/...，取 appcodes llm_capability 的中文名）；
|
||
产品 = 所有 active 的模型。只读源数据返回，写入由产品模块负责。
|
||
"""
|
||
env = ServerEnv()
|
||
async with get_sor_context(env, _MODULE_NAME) as sor:
|
||
models = await sor.sqlExe(
|
||
"SELECT id, name, vendor_model_id, capability, status, org_id, description, "
|
||
"ppid FROM llm_model "
|
||
"WHERE status='active' ORDER BY capability, name", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
caps = await sor.sqlExe(
|
||
"SELECT k, v FROM appcodes_kv WHERE parentid='llm_capability'", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not models:
|
||
return {'success': False, 'error': '模型注册表没有可导入的启用模型'}
|
||
models = [_row_to_dict(m) for m in models]
|
||
cap_names = {getattr(c, 'k', ''): getattr(c, 'v', '') for c in (caps or [])}
|
||
seen_caps = []
|
||
for m in models:
|
||
cap = m.get('capability') or 't2t'
|
||
if cap not in seen_caps:
|
||
seen_caps.append(cap)
|
||
categories = [{
|
||
'source_id': cap,
|
||
'name': cap_names.get(cap, cap),
|
||
'description': '产线平台模型（%s）' % cap_names.get(cap, cap),
|
||
'product_type': 'pipeline_llm_model',
|
||
'product_type_title': '产线模型按量',
|
||
'sort_order': 0,
|
||
} for cap in seen_caps]
|
||
products = []
|
||
for m in models:
|
||
cap = m.get('capability') or 't2t'
|
||
name = m.get('name') or ''
|
||
products.append({
|
||
'source_category_id': cap,
|
||
'resource_ref_id': m.get('id', ''),
|
||
'product_code': m.get('vendor_model_id', '') or name,
|
||
'product_name': name,
|
||
'product_type': 'pipeline_llm_model',
|
||
'brief_intro': m.get('description', '') or '',
|
||
'sort_order': 0,
|
||
'providerid': '', # 模型供应关系暂无对应供应商机构，成本侧用定价引擎
|
||
})
|
||
return {'success': True, 'categories': categories, 'products': products}
|