feat(llm): 用量落库 + 只读汇总端点 + 面板用量小卡(契约 v0.64)
- model 层新增可选 UsageSink(nil ⇒ 逐字不变,既有用例零改动) - 组合根注入 UsageWriter:有界队列 1024 + 单写者 + 批量事务(128/批、2s 兜底), 队列满丢弃 + 限流 WARN,绝不阻塞模型调用;sink 挂 RuntimeLLM(热重载不断) - 迁移 0017:表 llm_usage(13 列,session_id 可空)+ idx_created_at + idx_session_id - 新端点 GET /api/v1/admin/llm/usage(owner|admin 读;缺省近 24h、上限 90 天、 超限 400 不静默裁剪;hitRate = cacheHit/(cacheHit+cacheMiss),分母 0 ⇒ 0; 绝不含凭据与正文;未接线/查库失败 ⇒ 500 usage_unavailable) - LLM_USAGE_RETENTION(默认 720h,≤0 = 关闭清理):启动清一次 + 每小时 - 面板 LLM 页「用量(近 24h)」小卡 + Docs 两份同步 v0.64 - 已知缺口(如实登记):session_id 恒 NULL,需 loop 侧在请求上打标,留列后接
This commit is contained in:
1 parent
18381f65fd
commit
be38651ba4
32 files changed
+2233
-44
No files matched your search
@@ -93,6 +93,9 @@ func run() error {
|
||||
// 文档回收(切片 S7):启动期清扫一次 + 此后按 DOCUMENT_GC_INTERVAL 定时清扫。
|
||||
// 它不等这一次跑完就去拉网关(见 wire_document_gc.go 的取舍 1)。
|
||||
a.startDocumentGC(ctx)
|
||||
// LLM 用量落库与保留期清理(切片 U1):单写者 + 有界队列的异步写入器,
|
||||
// 加上按 LLM_USAGE_RETENTION 的过期样本清理。
|
||||
a.startUsage(ctx)
|
||||
a.log.Info("cyrene subsystems started",
|
||||
"personas", a.personas.CountPersonas(),
|
||||
"tools", len(a.caps.List(capreg.KindTool)),
|
||||
|
||||
@@ -85,6 +85,12 @@ type app struct {
|
||||
// docGC 是文档回收的常驻任务(切片 S7):未被引用且超过保留期的文档连同
|
||||
// sidecar 与元数据行一起回收。GC 被配置关闭时为 nil(见 wire_document_gc.go)。
|
||||
docGC *documentGC
|
||||
// usage 是 LLM 用量的异步落库出口(切片 U1):注入模型栈后,每次成功模型
|
||||
// 调用都会把非敏感计数样本塞进它的有界队列。库未接时为 nil(= 只记日志)。
|
||||
usage *model.UsageWriter
|
||||
// usageGC 是用量样本的保留期清理常驻任务。`LLM_USAGE_RETENTION ≤ 0` 时为
|
||||
// nil(清理关闭,见 wire_usage.go)。
|
||||
usageGC *usageRetentionGC
|
||||
// capacitySink 把 capacity_awaken 状态迁移广播给 Web UI(design/14 §5)。
|
||||
// 由 *gateway.Server 满足(BroadcastCapacityAwaken);以接口形式持有,
|
||||
// 组合根才能用假出口做单元测试。
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||||
)
|
||||
|
||||
@@ -84,6 +85,10 @@ func (a *app) wireFrontend() error {
|
||||
DataDir: resolveDataDir(a.cfg),
|
||||
Database: a.db,
|
||||
Vector: a.vectorState(),
|
||||
// Usage 是 `llm_usage` 表的只读汇总出口(切片 U1):网关复用组合根已打开的
|
||||
// 连接池,自己不开池、不读环境变量。库未接时为 nil(端点回 500
|
||||
// usage_unavailable,不回假汇总)。
|
||||
Usage: model.NewUsageReader(a.db),
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("gateway: %w", err)
|
||||
|
||||
@@ -128,7 +128,15 @@ func (a *app) wireRuntime() error {
|
||||
// 一等来源——把 `.env` 补齐进面板配置里缺失的键(面板已有的键绝不覆盖,组集合
|
||||
// 完全由面板决定),幂等;失败只记日志、不阻断启动。
|
||||
materializeModels(modelsPath, a.cfg.LLM, a.log)
|
||||
runtimeStack := model.NewRuntimeLLM(modelsPath, a.cfg.LLM, a.log)
|
||||
// 用量落库出口(切片 U1):先建 sink,再建模型栈 —— 栈里**每一条**链
|
||||
// (主/备/本地第三级/按角色跨协议)的每一步适配器都会带上它,因此任何一条
|
||||
// 路径上的模型调用都计成本。库未接时 sink 为 nil,行为与之前逐字相同。
|
||||
a.usage = a.newUsageSink()
|
||||
llmOpts := make([]model.RuntimeOption, 0, 1)
|
||||
if a.usage != nil {
|
||||
llmOpts = append(llmOpts, model.WithRuntimeUsageSink(a.usage))
|
||||
}
|
||||
runtimeStack := model.NewRuntimeLLM(modelsPath, a.cfg.LLM, a.log, llmOpts...)
|
||||
a.modelAdapter = runtimeStack
|
||||
a.llm = runtimeStack
|
||||
|
||||
|
||||
@@ -104,6 +104,8 @@ func (a *app) pushOutletConfig() backthink.PushOutletConfig {
|
||||
func (a *app) close(ctx context.Context) {
|
||||
// 文档回收循环要先用连接池(每轮一次查询),因此排在关库之前。
|
||||
a.docGC.stop()
|
||||
// 用量落库与保留期清理同样要用连接池:先把队列里剩下的样本写完,再停清理。
|
||||
a.stopUsage()
|
||||
if a.backthink != nil {
|
||||
_ = a.backthink.Stop(ctx)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
package main
|
||||
|
||||
// 本文件把 **LLM 用量落库与保留期清理**接进组合根(切片 U1,主人裁决「2.4 落」):
|
||||
// 构造异步写入器、把它注入模型栈,并在启动期清一次过期样本、此后按间隔定时清。
|
||||
//
|
||||
// 三条取舍写在这里(写入器实现见 `internal/runtime/model/usage_store.go`):
|
||||
//
|
||||
// 1. **写入绝不阻塞模型调用**:模型栈拿到的是一个有界队列 + 单写者的 sink,
|
||||
// `RecordUsage` 不做任何等待(队列满就丢弃并限流 WARN)。因此库慢、库锁、
|
||||
// 库挂了,主人的每次对话都不受影响。
|
||||
// 2. **失败不退出进程**:整批落库失败只记一条带来源的 WARN;保留期清理失败
|
||||
// 同样是 WARN 后等下一轮 —— 用量是观测数据,不该决定进程生死。
|
||||
// 3. **关闭 = 显式配 ≤0**:`LLM_USAGE_RETENTION=0`(或负值)不构造清理器,
|
||||
// 由下面的一条 INFO 写明「清理已关闭」(静默的「配了却不生效」不存在)。
|
||||
//
|
||||
// 清理间隔刻意不做成 env 键:保留期才是部署者的旋钮,清扫频率没有可调价值。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// usagePurgeInterval 是用量保留期清理的定时间隔(启动期另有一次立即清扫)。
|
||||
const usagePurgeInterval = time.Hour
|
||||
|
||||
// usageRetentionGC 是常驻的用量保留期清理任务:一个连接 + 一个可停的循环。
|
||||
type usageRetentionGC struct {
|
||||
db *sql.DB
|
||||
retention time.Duration
|
||||
log *slog.Logger
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// newUsageSink 按当前配置构造用量写入器;库未接时返回 nil(= 只记日志,
|
||||
// 与这一版之前逐字相同)。
|
||||
func (a *app) newUsageSink() *model.UsageWriter {
|
||||
return model.NewUsageWriter(a.db, a.log)
|
||||
}
|
||||
|
||||
// newUsageRetentionGC 按当前配置构造保留期清理器;库未接或保留期 ≤0(关闭清理)
|
||||
// 时返回 nil。关闭原因由 startUsage 记一条 INFO。
|
||||
func (a *app) newUsageRetentionGC() *usageRetentionGC {
|
||||
if a.db == nil || a.cfg.Server.LLMUsageRetention <= 0 {
|
||||
return nil
|
||||
}
|
||||
return &usageRetentionGC{
|
||||
db: a.db,
|
||||
retention: a.cfg.Server.LLMUsageRetention,
|
||||
log: logger.WithSource(a.log, "cmd/cyrene"),
|
||||
}
|
||||
}
|
||||
|
||||
// startUsage 拉起用量落库的单写者与保留期清理(各自关闭时是空操作)。
|
||||
// 它不等清理跑完就去拉网关(与文档 GC 同一取舍:慢盘上的一次 DELETE 不该把
|
||||
// 「昔涟能不能起来」绑在磁盘上)。
|
||||
func (a *app) startUsage(ctx context.Context) {
|
||||
if a.usage != nil {
|
||||
a.usage.Start(ctx)
|
||||
}
|
||||
gc := a.newUsageRetentionGC()
|
||||
if gc == nil {
|
||||
a.log.Info("llm usage retention disabled",
|
||||
"env", "LLM_USAGE_RETENTION", "value", a.cfg.Server.LLMUsageRetention.String())
|
||||
return
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
gc.cancel = cancel
|
||||
gc.done = make(chan struct{})
|
||||
a.usageGC = gc
|
||||
go gc.loop(runCtx)
|
||||
}
|
||||
|
||||
// stopUsage 停掉用量落库与保留期清理(先让单写者把队列里剩下的样本写完,再停
|
||||
// 清理)。两者都要用连接池,因此必须在关库**之前**调用;未启动时是空操作。
|
||||
func (a *app) stopUsage() {
|
||||
if a.usage != nil {
|
||||
a.usage.Stop()
|
||||
}
|
||||
a.usageGC.stop()
|
||||
}
|
||||
|
||||
// loop 是清理循环:启动期清扫一次 → 定时清扫。
|
||||
func (g *usageRetentionGC) loop(ctx context.Context) {
|
||||
defer close(g.done)
|
||||
g.sweep(ctx)
|
||||
ticker := time.NewTicker(usagePurgeInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
g.sweep(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sweep 跑一轮清理。删除条数**不是**每次都要记(正常运转时每轮都是 0,记满日志
|
||||
// 只是噪声);真的删了才记一条 INFO,失败一律 WARN 并让循环继续。
|
||||
func (g *usageRetentionGC) sweep(ctx context.Context) {
|
||||
before := time.Now().Add(-g.retention)
|
||||
removed, err := model.PurgeUsage(ctx, g.db, before)
|
||||
if err != nil {
|
||||
g.log.Warn("llm usage retention sweep failed", "err", err)
|
||||
return
|
||||
}
|
||||
if removed > 0 {
|
||||
g.log.Info("llm usage retention sweep removed rows",
|
||||
"rows", removed, "retention", g.retention.String())
|
||||
}
|
||||
}
|
||||
|
||||
// stop 停掉清理循环并等它退出;未启动时是空操作。
|
||||
func (g *usageRetentionGC) stop() {
|
||||
if g == nil || g.cancel == nil {
|
||||
return
|
||||
}
|
||||
g.cancel()
|
||||
<-g.done
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// 本文件钉住用量落库与保留期清理**被接上组合根**(切片 U1;与
|
||||
// TestWireDocumentGC_IsWired 同一手法):
|
||||
//
|
||||
// - sink 必须装进模型栈,且模型栈必须由组合根构造 —— 少任一处,落库都会
|
||||
// **静默不工作**(日志照记、库里一行没有),而这不会让任何既有用例变红;
|
||||
// - 写入器与清理器必须在**关库之前**停掉(两者都要用那条连接池);
|
||||
// - 启动期必须真的把两者拉起来(`a.startUsage(ctx)`);
|
||||
// - 保留期 ≤0 = 关闭清理,且关闭是**显式**的(记一条 INFO)。
|
||||
func TestWireUsage_IsWired(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
seams := map[string][]string{
|
||||
"wire_runtime.go": {
|
||||
"a.usage = a.newUsageSink()",
|
||||
"model.WithRuntimeUsageSink(a.usage)",
|
||||
},
|
||||
"wire_usage.go": {
|
||||
"model.NewUsageWriter(a.db, a.log)",
|
||||
"a.cfg.Server.LLMUsageRetention",
|
||||
"model.PurgeUsage(ctx, g.db, before)",
|
||||
"a.log.Info(\"llm usage retention disabled\"",
|
||||
},
|
||||
"wire_services.go": {
|
||||
// 先让单写者把队列里剩下的样本写完,再停清理,最后才关库。
|
||||
"a.stopUsage()",
|
||||
},
|
||||
"main.go": {
|
||||
"a.startUsage(ctx)",
|
||||
},
|
||||
"wire_frontend.go": {
|
||||
// 管理端点的只读汇总出口(网关复用组合根已打开的连接池)。
|
||||
"Usage: model.NewUsageReader(a.db)",
|
||||
},
|
||||
}
|
||||
for file, wants := range seams {
|
||||
src, err := os.ReadFile(file)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", file, err)
|
||||
}
|
||||
for _, want := range wants {
|
||||
if !strings.Contains(string(src), want) {
|
||||
t.Fatalf("%s 里找不到 %q —— 用量落库没有被接上组合根", file, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 新 env 键必须与代码里的 env 标签逐字一致(AGENTS §10 铁律)。
|
||||
raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example"))
|
||||
if err != nil {
|
||||
t.Fatalf("read .env.example: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(raw), "LLM_USAGE_RETENTION=") {
|
||||
t.Fatalf(".env.example 里找不到 %q(新 env 键必须同步模板,AGENTS §10)", "LLM_USAGE_RETENTION=")
|
||||
}
|
||||
if tag := envTagOf(t, "LLMUsageRetention"); tag != "LLM_USAGE_RETENTION" {
|
||||
t.Fatalf("config.ServerConfig.LLMUsageRetention 的 env 标签 = %q, 与模板不一致", tag)
|
||||
}
|
||||
}
|
||||
|
||||
// envTagOf 返回 config.ServerConfig 上某字段的 env 标签。
|
||||
func envTagOf(t *testing.T, field string) string {
|
||||
t.Helper()
|
||||
typ, ok := reflect.TypeOf(config.ServerConfig{}).FieldByName(field)
|
||||
if !ok {
|
||||
t.Fatalf("config.ServerConfig 里没有字段 %q", field)
|
||||
}
|
||||
return typ.Tag.Get("env")
|
||||
}
|
||||
|
||||
// usageGCDB 打开一个带迁移的内存库。
|
||||
func usageGCDB(t *testing.T) *sql.DB {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
conn, err := db.Open(ctx, ":memory:", slog.New(slog.NewTextHandler(os.Stderr, nil)))
|
||||
if err != nil {
|
||||
t.Fatalf("open in-memory db: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
if err := db.MigrateUp(ctx, conn, slog.New(slog.NewTextHandler(os.Stderr, nil))); err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
return conn
|
||||
}
|
||||
|
||||
// usageGCApp 构造一个只带清理器所需依赖的 app。
|
||||
func usageGCApp(t *testing.T, conn *sql.DB, retention time.Duration) *app {
|
||||
t.Helper()
|
||||
return &app{
|
||||
cfg: &config.Config{Server: config.ServerConfig{LLMUsageRetention: retention}},
|
||||
db: conn,
|
||||
log: slog.New(slog.NewTextHandler(os.Stderr, nil)),
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageRetentionGC_SweepRemovesExpiredOnly 钉住清理判据:早于保留期的样本被
|
||||
// 删、保留期内的原样留着。
|
||||
func TestUsageRetentionGC_SweepRemovesExpiredOnly(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
conn := usageGCDB(t)
|
||||
insertUsageAt(t, conn, "2020-01-01T00:00:00.000Z")
|
||||
insertUsageAt(t, conn, "2099-01-01T00:00:00.000Z")
|
||||
|
||||
gc := usageGCApp(t, conn, 720*time.Hour).newUsageRetentionGC()
|
||||
if gc == nil {
|
||||
t.Fatal("保留期 > 0 时必须构造清理器")
|
||||
}
|
||||
gc.sweep(ctx)
|
||||
|
||||
if rows := countUsage(t, conn); rows != 1 {
|
||||
t.Fatalf("清理后剩 %d 行, want 1(只删过期那条)", rows)
|
||||
}
|
||||
var remaining string
|
||||
if err := conn.QueryRowContext(ctx, `SELECT created_at FROM llm_usage`).Scan(&remaining); err != nil {
|
||||
t.Fatalf("read remaining: %v", err)
|
||||
}
|
||||
if remaining != "2099-01-01T00:00:00.000Z" {
|
||||
t.Fatalf("剩下的是 %q, want 未过期那条", remaining)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageRetentionGC_DisabledWhenNonPositive 钉住「≤0 = 关闭清理」是显式关:
|
||||
// 0 与负值都不构造清理器(因而一行都不会被删),正数才构造。
|
||||
func TestUsageRetentionGC_DisabledWhenNonPositive(t *testing.T) {
|
||||
t.Parallel()
|
||||
conn := usageGCDB(t)
|
||||
for _, retention := range []time.Duration{0, -time.Hour} {
|
||||
if gc := usageGCApp(t, conn, retention).newUsageRetentionGC(); gc != nil {
|
||||
t.Fatalf("LLM_USAGE_RETENTION=%s 时不该构造清理器", retention)
|
||||
}
|
||||
}
|
||||
// 关闭状态下过期行原样留着(没有任何代码路径去删它)。
|
||||
insertUsageAt(t, conn, "2020-01-01T00:00:00.000Z")
|
||||
if rows := countUsage(t, conn); rows != 1 {
|
||||
t.Fatalf("关闭清理却动了数据:剩 %d 行", rows)
|
||||
}
|
||||
if gc := usageGCApp(t, conn, time.Hour).newUsageRetentionGC(); gc == nil {
|
||||
t.Fatal("LLM_USAGE_RETENTION=1h 时必须构造清理器")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageWriter_NilDBMeansNoSink 钉住「库未接 ⇒ sink 为 nil ⇒ 只记日志」的
|
||||
// 组合根口径(与 model 层的 nil-sink 行为合成一条链)。
|
||||
func TestUsageWriter_NilDBMeansNoSink(t *testing.T) {
|
||||
t.Parallel()
|
||||
a := &app{log: slog.New(slog.NewTextHandler(os.Stderr, nil))}
|
||||
if sink := a.newUsageSink(); sink != nil {
|
||||
t.Fatalf("库未接时 newUsageSink 必须返回 nil,实得 %#v", sink)
|
||||
}
|
||||
if gc := a.newUsageRetentionGC(); gc != nil {
|
||||
t.Fatal("库未接时不该构造清理器")
|
||||
}
|
||||
// 未注入 sink 的模型栈照常构造(不 panic),且模型栈不因 nil sink 变红。
|
||||
stack := model.NewRuntimeLLM(filepath.Join(t.TempDir(), "models.json"),
|
||||
config.LLMConfig{}, a.log)
|
||||
if stack == nil {
|
||||
t.Fatal("nil sink 不该影响模型栈构造")
|
||||
}
|
||||
}
|
||||
|
||||
// insertUsageAt 写一行带指定 created_at 的用量样本。
|
||||
func insertUsageAt(t *testing.T, conn *sql.DB, createdAt string) {
|
||||
t.Helper()
|
||||
if _, err := conn.ExecContext(context.Background(),
|
||||
`INSERT INTO llm_usage (purpose, adapter, model, prompt_tokens, created_at)
|
||||
VALUES ('chat', 'openai', 'm', 1, ?)`, createdAt); err != nil {
|
||||
t.Fatalf("insert usage row: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// countUsage 返回 llm_usage 行数。
|
||||
func countUsage(t *testing.T, conn *sql.DB) int64 {
|
||||
t.Helper()
|
||||
var rows int64
|
||||
if err := conn.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM llm_usage`).Scan(&rows); err != nil {
|
||||
t.Fatalf("count llm_usage: %v", err)
|
||||
}
|
||||
return rows
|
||||
}
|
||||
Reference in new issue
Block a user