287 lines
10 KiB
Python
287 lines
10 KiB
Python
"""
|
|
数据集市 ETL — 增量同步 llmusage → 集市表
|
|
"""
|
|
import json
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from appPublic.uniqueID import getID
|
|
from appPublic.dictObject import DictObject
|
|
from appPublic.jsonConfig import getConfig
|
|
from appPublic.log import debug, info, exception
|
|
from sqlor.dbpools import DBPools
|
|
from ahserver.serverenv import ServerEnv
|
|
|
|
MODULE_NAME = "sage_datamark"
|
|
|
|
|
|
async def ensure_tables(sor):
|
|
"""创建集市表（如不存在）"""
|
|
tables = [
|
|
"""CREATE TABLE IF NOT EXISTS dm_model_call_fact (
|
|
id VARCHAR(32) PRIMARY KEY,
|
|
luid VARCHAR(32) NOT NULL,
|
|
llmid VARCHAR(32) NOT NULL,
|
|
model VARCHAR(100),
|
|
catelogid VARCHAR(10),
|
|
userid VARCHAR(32) NOT NULL,
|
|
userorgid VARCHAR(32) NOT NULL,
|
|
ownerid VARCHAR(32),
|
|
providerid VARCHAR(32),
|
|
call_date DATE NOT NULL,
|
|
call_hour TINYINT NOT NULL,
|
|
call_time DATETIME NOT NULL,
|
|
ttft_ms INT,
|
|
ttot_ms INT,
|
|
prompt_tokens INT DEFAULT 0,
|
|
completion_tokens INT DEFAULT 0,
|
|
status VARCHAR(20),
|
|
fail_reason VARCHAR(200),
|
|
amount DECIMAL(10,4) DEFAULT 0,
|
|
currency VARCHAR(5) DEFAULT 'CNY',
|
|
distributor_orgid VARCHAR(32),
|
|
sale_userid VARCHAR(32),
|
|
created_at DATETIME DEFAULT NOW(),
|
|
INDEX idx_call_date (call_date),
|
|
INDEX idx_llmid (llmid),
|
|
INDEX idx_userorgid (userorgid),
|
|
INDEX idx_ownerid (ownerid),
|
|
INDEX idx_status (status),
|
|
INDEX idx_model (model)
|
|
)""",
|
|
|
|
"""CREATE TABLE IF NOT EXISTS dm_model_perf_daily (
|
|
id VARCHAR(32) PRIMARY KEY,
|
|
stat_date DATE NOT NULL,
|
|
llmid VARCHAR(32) NOT NULL,
|
|
model VARCHAR(100),
|
|
catelogid VARCHAR(10),
|
|
providerid VARCHAR(32),
|
|
ownerid VARCHAR(32),
|
|
total_calls INT DEFAULT 0,
|
|
success_calls INT DEFAULT 0,
|
|
fail_calls INT DEFAULT 0,
|
|
success_rate DECIMAL(5,2),
|
|
avg_ttft_ms DECIMAL(10,1),
|
|
p50_ttft_ms DECIMAL(10,1),
|
|
p95_ttft_ms DECIMAL(10,1),
|
|
p99_ttft_ms DECIMAL(10,1),
|
|
avg_ttot_ms DECIMAL(10,1),
|
|
p50_ttot_ms DECIMAL(10,1),
|
|
p95_ttot_ms DECIMAL(10,1),
|
|
max_concurrent INT,
|
|
avg_concurrent DECIMAL(5,1),
|
|
total_prompt_tokens BIGINT DEFAULT 0,
|
|
total_completion_tokens BIGINT DEFAULT 0,
|
|
total_amount DECIMAL(12,4) DEFAULT 0,
|
|
userorgid VARCHAR(32),
|
|
distributor_orgid VARCHAR(32),
|
|
UNIQUE KEY uk_date_model_org (stat_date, llmid, userorgid),
|
|
INDEX idx_date (stat_date),
|
|
INDEX idx_model (model)
|
|
)""",
|
|
|
|
"""CREATE TABLE IF NOT EXISTS dm_provider_cost_daily (
|
|
id VARCHAR(32) PRIMARY KEY,
|
|
stat_date DATE NOT NULL,
|
|
model VARCHAR(100) NOT NULL,
|
|
catelogid VARCHAR(10),
|
|
providerid VARCHAR(32) NOT NULL,
|
|
provider_name VARCHAR(100),
|
|
total_calls INT DEFAULT 0,
|
|
total_tokens BIGINT DEFAULT 0,
|
|
total_amount DECIMAL(12,4) DEFAULT 0,
|
|
unit_price DECIMAL(10,6),
|
|
avg_ttft_ms DECIMAL(10,1),
|
|
UNIQUE KEY uk_date_model_provider (stat_date, model, providerid),
|
|
INDEX idx_model (model),
|
|
INDEX idx_date (stat_date)
|
|
)""",
|
|
]
|
|
for sql in tables:
|
|
await sor.sqlExe(sql, {})
|
|
|
|
|
|
def parse_usages(usages_str):
|
|
"""从 llmusage.usages JSON 中提取 token 数"""
|
|
try:
|
|
u = json.loads(usages_str) if isinstance(usages_str, str) else usages_str
|
|
if isinstance(u, dict):
|
|
return (
|
|
u.get('prompt_tokens', 0) or 0,
|
|
u.get('completion_tokens', 0) or 0,
|
|
u.get('total_tokens', 0) or 0
|
|
)
|
|
except:
|
|
pass
|
|
return 0, 0, 0
|
|
|
|
|
|
async def sync_call_fact(sor, last_sync=None):
|
|
"""增量同步 llmusage → dm_model_call_fact"""
|
|
if last_sync is None:
|
|
last_sync = (datetime.now() - timedelta(minutes=10)).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
sql = """
|
|
SELECT lu.*, l.model as llm_model, l.catelogid, l.providerid, l.ownerid
|
|
FROM llmusage lu
|
|
JOIN llm l ON l.id = lu.llmid
|
|
WHERE lu.use_time >= ${last_sync}$
|
|
ORDER BY lu.use_time
|
|
"""
|
|
rows = await sor.sqlExe(sql, {'last_sync': last_sync})
|
|
info('sage_datamark ETL: syncing ' + str(len(rows)) + ' rows from ' + last_sync)
|
|
|
|
count = 0
|
|
for r in rows:
|
|
r = DictObject(r)
|
|
# Skip existing
|
|
existing = await sor.sqlExe(
|
|
'SELECT id FROM dm_model_call_fact WHERE luid=${luid}$',
|
|
{'luid': r.id})
|
|
if existing:
|
|
continue
|
|
|
|
prompt_tokens, completion_tokens, _ = parse_usages(r.usages)
|
|
call_time = datetime.strptime(r.use_time[:19], '%Y-%m-%d %H:%M:%S')
|
|
|
|
ns = {
|
|
'id': getID(),
|
|
'luid': r.id,
|
|
'llmid': r.llmid,
|
|
'model': r.llm_model or r.model,
|
|
'catelogid': r.catelogid,
|
|
'userid': r.userid,
|
|
'userorgid': r.userorgid,
|
|
'ownerid': r.ownerid,
|
|
'providerid': r.providerid,
|
|
'call_date': r.use_date or call_time.strftime('%Y-%m-%d'),
|
|
'call_hour': call_time.hour,
|
|
'call_time': r.use_time,
|
|
'ttft_ms': int(r.responsed_seconds * 1000) if r.responsed_seconds else None,
|
|
'ttot_ms': int(r.finish_seconds * 1000) if r.finish_seconds else None,
|
|
'prompt_tokens': prompt_tokens,
|
|
'completion_tokens': completion_tokens,
|
|
'status': r.status or 'SUCCEEDED',
|
|
'amount': r.amount or 0,
|
|
'currency': getattr(r, 'currency', 'CNY') or 'CNY',
|
|
}
|
|
await sor.C('dm_model_call_fact', ns)
|
|
count += 1
|
|
|
|
info('sage_datamark ETL: inserted ' + str(count) + ' new rows')
|
|
return count
|
|
|
|
|
|
async def aggregate_daily_perf(sor, stat_date=None):
|
|
"""聚合当天 dm_model_call_fact → dm_model_perf_daily"""
|
|
if stat_date is None:
|
|
stat_date = datetime.now().strftime('%Y-%m-%d')
|
|
|
|
# 按 llmid + userorgid 聚合
|
|
sql = """
|
|
SELECT
|
|
call_date, llmid, model, catelogid, providerid, ownerid,
|
|
COALESCE(userorgid, '') as userorgid,
|
|
COUNT(*) as total_calls,
|
|
SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success_calls,
|
|
SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail_calls,
|
|
AVG(ttft_ms) as avg_ttft,
|
|
AVG(ttot_ms) as avg_ttot,
|
|
SUM(prompt_tokens) as total_prompt,
|
|
SUM(completion_tokens) as total_completion,
|
|
SUM(amount) as total_amount
|
|
FROM dm_model_call_fact
|
|
WHERE call_date = ${stat_date}$
|
|
GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, userorgid
|
|
"""
|
|
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
|
|
info('sage_datamark AGG: aggregating ' + str(len(rows)) + ' groups for ' + stat_date)
|
|
|
|
for r in rows:
|
|
r = DictObject(r)
|
|
# Delete old row, insert new
|
|
await sor.sqlExe(
|
|
'DELETE FROM dm_model_perf_daily WHERE stat_date=${d}$ AND llmid=${l}$ AND COALESCE(userorgid,\"\")=${u}$',
|
|
{'d': stat_date, 'l': r.llmid, 'u': r.userorgid or ''})
|
|
|
|
ns = {
|
|
'id': getID(),
|
|
'stat_date': stat_date,
|
|
'llmid': r.llmid,
|
|
'model': r.model,
|
|
'catelogid': r.catelogid,
|
|
'providerid': r.providerid,
|
|
'ownerid': r.ownerid,
|
|
'total_calls': r.total_calls,
|
|
'success_calls': r.success_calls,
|
|
'fail_calls': r.fail_calls,
|
|
'success_rate': round(r.success_calls * 100.0 / r.total_calls, 2) if r.total_calls else 0,
|
|
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
|
'avg_ttot_ms': round(r.avg_ttot, 1) if r.avg_ttot else None,
|
|
'total_prompt_tokens': r.total_prompt or 0,
|
|
'total_completion_tokens': r.total_completion or 0,
|
|
'total_amount': r.total_amount or 0,
|
|
'userorgid': r.userorgid or None,
|
|
}
|
|
await sor.C('dm_model_perf_daily', ns)
|
|
|
|
return len(rows)
|
|
|
|
|
|
async def aggregate_provider_cost(sor, stat_date=None):
|
|
"""聚合同模型供应商性价比 — dm_provider_cost_daily"""
|
|
if stat_date is None:
|
|
stat_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
|
|
sql = """
|
|
SELECT
|
|
model, providerid,
|
|
COUNT(*) as total_calls,
|
|
SUM(prompt_tokens + completion_tokens) as total_tokens,
|
|
SUM(amount) as total_amount,
|
|
AVG(ttft_ms) as avg_ttft
|
|
FROM dm_model_call_fact
|
|
WHERE call_date = ${stat_date}$
|
|
GROUP BY model, providerid
|
|
"""
|
|
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
|
|
|
|
for r in rows:
|
|
r = DictObject(r)
|
|
await sor.sqlExe(
|
|
'DELETE FROM dm_provider_cost_daily WHERE stat_date=${d}$ AND model=${m}$ AND providerid=${p}$',
|
|
{'d': stat_date, 'm': r.model, 'p': r.providerid})
|
|
|
|
ns = {
|
|
'id': getID(),
|
|
'stat_date': stat_date,
|
|
'model': r.model,
|
|
'providerid': r.providerid,
|
|
'total_calls': r.total_calls,
|
|
'total_tokens': r.total_tokens or 0,
|
|
'total_amount': r.total_amount or 0,
|
|
'unit_price': round(float(r.total_amount) / r.total_tokens, 8) if r.total_tokens else None,
|
|
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
|
}
|
|
await sor.C('dm_provider_cost_daily', ns)
|
|
|
|
return len(rows)
|
|
|
|
|
|
async def run_etl_sync(sor):
|
|
"""每5分钟运行: 确保表存在 + 增量同步"""
|
|
await ensure_tables(sor)
|
|
await sync_call_fact(sor)
|
|
|
|
|
|
async def run_etl_aggregate(sor):
|
|
"""每小时运行: 聚合天级性能 + 并发统计"""
|
|
await ensure_tables(sor)
|
|
await aggregate_daily_perf(sor)
|
|
|
|
|
|
async def run_etl_provider_cost(sor):
|
|
"""每天运行: 供应商性价比"""
|
|
await ensure_tables(sor)
|
|
await aggregate_provider_cost(sor)
|