feat(model): llm_usage.session_id 真正落库(补完 U1 缺口)
通道:新增 model/session_ctx.go 的 WithSessionID / SessionIDFrom(与 tool.WithTurnPersona
同族的 ctx 范式,避开 model 反向 import loop 的环),loop 在 Run 顶部播种;适配器
record(ctx,...) 接会话;usageSessionArg 空串映射为 NULL。
兼容:sink 未注入或会话未播种时行为逐字不变;既有 model 测试零断言改动即全绿。
文件:model/{session_ctx.go(新), usage.go, usage_store.go, openai.go, anthropic.go,
usage_session_test.go(新)} + loop/{run.go, usage_session_test.go(新)}。
变异三处:去 loop 打标 / 写侧不绑会话 / 适配器不读 ctx ⇒ 各自红行,还原复绿。
契约与日志字段集未动;读侧不按 session 过滤,面板汇总口径不变。
This commit is contained in:
1 parent
cb564fe981
commit
59a1ac4056
9 files changed
+359
-18
No files matched your search
@@ -234,7 +234,7 @@ func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Optio
|
||||
if err != nil {
|
||||
return Response{}, err
|
||||
}
|
||||
a.record(req, start, out)
|
||||
a.record(ctx, req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -291,7 +291,7 @@ func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEv
|
||||
return Response{}, err
|
||||
}
|
||||
out.Model = model
|
||||
a.record(req, start, out)
|
||||
a.record(ctx, req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -411,8 +411,9 @@ func (a *AnthropicChatAdapter) upstreamError(resp *http.Response) error {
|
||||
|
||||
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
// 所有调用点都覆盖 —— 见 usage.go),再喂给可选的 CallRecord 计账出口。
|
||||
func (a *AnthropicChatAdapter) record(req Request, start time.Time, resp Response) {
|
||||
logUsage(a.log, a.usageSink, a.name, req, resp, time.Since(start))
|
||||
// ctx 用来取本次调用所属的会话(loop 播种,见 session_ctx.go)。
|
||||
func (a *AnthropicChatAdapter) record(ctx context.Context, req Request, start time.Time, resp Response) {
|
||||
logUsage(ctx, a.log, a.usageSink, a.name, req, resp, time.Since(start))
|
||||
if a.recorder == nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -152,7 +152,7 @@ func (a *OpenAIChatAdapter) Chat(ctx context.Context, req Request, opts Options)
|
||||
return Response{}, fmt.Errorf("model: %w: decode response: %v", ErrUpstream, err)
|
||||
}
|
||||
out := responseFromWire(&wire)
|
||||
a.record(req, start, out)
|
||||
a.record(ctx, req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -189,7 +189,7 @@ func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onEvent
|
||||
return Response{}, err
|
||||
}
|
||||
out.Model = model
|
||||
a.record(req, start, out)
|
||||
a.record(ctx, req, start, out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -241,8 +241,9 @@ func (a *OpenAIChatAdapter) upstreamError(resp *http.Response) error {
|
||||
|
||||
// record 上报一次成功调用的用量:先记一条结构化用量日志(含缓存命中率,
|
||||
// 所有调用点都覆盖 —— 见 usage.go),再喂给可选的 CallRecord 计账出口。
|
||||
func (a *OpenAIChatAdapter) record(req Request, start time.Time, resp Response) {
|
||||
logUsage(a.log, a.usageSink, a.name, req, resp, time.Since(start))
|
||||
// ctx 用来取本次调用所属的会话(loop 播种,见 session_ctx.go)。
|
||||
func (a *OpenAIChatAdapter) record(ctx context.Context, req Request, start time.Time, resp Response) {
|
||||
logUsage(ctx, a.log, a.usageSink, a.name, req, resp, time.Since(start))
|
||||
if a.recorder == nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
package model
|
||||
|
||||
import "context"
|
||||
|
||||
// 本文件承载「本次调用属于哪个会话」的 context 通道(切片 U1 留的接线缺口)。
|
||||
//
|
||||
// 背景:模型适配器跑在 loop 之下、没有会话概念,于是 `llm_usage.session_id`
|
||||
// 一直是 NULL —— 列与索引在切片 U1 已建好(迁移 0017),缺的是**从循环边界
|
||||
// 把会话标识带到适配器边界**的那一跳(usage.go / usage_store.go 此前都留了
|
||||
// 注释说明这一点)。没有这一跳,「哪次调用贵在哪个会话身上」这个问题永远答不出来。
|
||||
//
|
||||
// 通道形态与 tool/turncontext.go、loop/session_ctx.go 同族:key 由**消费方**
|
||||
// (本包)持有,上游 loop 在 Run 顶部播种一次;此后所有经该 ctx 发起的模型调用
|
||||
// —— 流式的 ChatStream 与非流式的 Chat(agent.RunStep)**两条**路径,以及
|
||||
// fallback 链、persona 栈这些原样透传 ctx 的包装 —— 在适配器内部都能读到它。
|
||||
// 反过来(key 放在 loop 里)做不到:本包在 loop 之下,导入 loop 会成环(铁律:
|
||||
// 单二进制内模块间只走进程内接口 + 依赖注入)。
|
||||
//
|
||||
// 为什么不挂在 model.Request 上:Request 是要编码成线格式发给提供商的那份边界
|
||||
// 对象,而会话标识属于**本地记账元数据** —— 挂上去就等于给「将来某个适配器顺手
|
||||
// 把它序列化出去」留了口子,且 buildRequest 之外的调用点(后台流程、直接调用
|
||||
// 适配器的包装)全都要各自记得填。ctx 值只在进程内流动,覆盖面还更全。
|
||||
//
|
||||
// 纪律:它只是记账标签,**绝不**参与身份/权限判定(红线:身份由 channel 层
|
||||
// actor 元数据 + 策略引擎决定,绝不靠任何透传字段推断)。未播种时读到空串,
|
||||
// 落库据此写 NULL 而不是空串。
|
||||
|
||||
// sessionIDKey 是本次模型调用所属会话 id 的 context key。
|
||||
type sessionIDKey struct{}
|
||||
|
||||
// WithSessionID 把会话 id 挂到 ctx 上(循环在 Run 顶部调用一次)。
|
||||
//
|
||||
// 空 id 是空操作:不挂载即「本次调用没有会话归属」(管理面板试连、未接循环的
|
||||
// 后台流程),读取侧返回空串,用量记 NULL —— 与接线前逐字一致。
|
||||
func WithSessionID(ctx context.Context, sessionID string) context.Context {
|
||||
if ctx == nil || sessionID == "" {
|
||||
return ctx
|
||||
}
|
||||
return context.WithValue(ctx, sessionIDKey{}, sessionID)
|
||||
}
|
||||
|
||||
// SessionIDFrom 返回 ctx 上挂载的会话 id;未挂载(或 ctx 为 nil)时返回空串。
|
||||
func SessionIDFrom(ctx context.Context) string {
|
||||
if ctx == nil {
|
||||
return ""
|
||||
}
|
||||
s, _ := ctx.Value(sessionIDKey{}).(string)
|
||||
return s
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
@@ -36,6 +37,10 @@ const usageLogEvent = "model usage"
|
||||
type UsageSample struct {
|
||||
// Adapter 是服务该次调用的适配器/提供商分组标识。
|
||||
Adapter string
|
||||
// SessionID 是本次调用所属的会话(loop 经 ctx 播种,见 session_ctx.go)。
|
||||
// 空串表示**没有会话归属**(管理面板试连、未接循环的后台流程)——落库时
|
||||
// 写 NULL,绝不写 "",否则「无归属」与「属于某会话」在列里混成同一个值。
|
||||
SessionID string
|
||||
// Model 是产出该响应的模型名。
|
||||
Model string
|
||||
// Purpose 是调用用途标签(如 "chat" / "memory_extract" / "engage_judge")。
|
||||
@@ -64,10 +69,12 @@ type UsageSink interface {
|
||||
}
|
||||
|
||||
// usageSampleOf 把一次调用归一化成 UsageSample。它是日志与落库**唯一**的字段来源,
|
||||
// 两处口径因此不可能分叉。
|
||||
func usageSampleOf(adapter string, req Request, resp Response, took time.Duration) UsageSample {
|
||||
// 两处口径因此不可能分叉。会话标识从 ctx 解析(session_ctx.go):适配器没有会话
|
||||
// 概念,loop 是唯一把它播种进来的地方;未播种时为 "",落库写 NULL。
|
||||
func usageSampleOf(ctx context.Context, adapter string, req Request, resp Response, took time.Duration) UsageSample {
|
||||
return UsageSample{
|
||||
Adapter: adapter,
|
||||
SessionID: SessionIDFrom(ctx),
|
||||
Model: resp.Model,
|
||||
Purpose: req.Purpose,
|
||||
PromptTokens: resp.PromptTokens,
|
||||
@@ -86,9 +93,12 @@ func usageSampleOf(adapter string, req Request, resp Response, took time.Duratio
|
||||
//
|
||||
// log 可以为 nil(适配器在未接日志时用的是丢弃 logger),这里仍做一次防御:
|
||||
// 缺字段容错、绝不 panic。sink 为 nil 时**这一步是空操作** —— 逐字等于引入落库
|
||||
// 之前的行为。
|
||||
func logUsage(log *slog.Logger, sink UsageSink, adapter string, req Request, resp Response, took time.Duration) {
|
||||
sample := usageSampleOf(adapter, req, resp, took)
|
||||
// 之前的行为(日志字段集也不含会话:会话只进落库面,见 UsageSample.SessionID)。
|
||||
//
|
||||
// ctx 只为解析会话标识而来(session_ctx.go)—— 适配器的 record 拿不到会话,
|
||||
// 循环在发起调用时把它播种在 ctx 上。
|
||||
func logUsage(ctx context.Context, log *slog.Logger, sink UsageSink, adapter string, req Request, resp Response, took time.Duration) {
|
||||
sample := usageSampleOf(ctx, adapter, req, resp, took)
|
||||
if log != nil {
|
||||
log.Info(usageLogEvent,
|
||||
"adapter", sample.Adapter,
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
)
|
||||
|
||||
// 本文件钉住「会话标识端到端进 llm_usage.session_id」的前半程(模型层):
|
||||
// loop 经 ctx 播种会话(session_ctx.go)⇒ 适配器在**两条**调用路径上都把
|
||||
// 它读进 UsageSample ⇒ 没播种时是空串(落库据此写 NULL,而不是空串)。
|
||||
//
|
||||
// 后半程(真的写进列)由 usage_store_test.go 的
|
||||
// TestUsageWriter_PersistsSessionID 钉住,loop 侧的播种由
|
||||
// internal/runtime/loop/usage_session_test.go 端到端钉住。
|
||||
|
||||
// sessionUsageRequest 是最小可用请求(只关心用量样本里的会话字段)。
|
||||
func sessionUsageRequest() Request {
|
||||
return Request{Purpose: "chat", Messages: []Message{TextMessage(RoleUser, "hi")}}
|
||||
}
|
||||
|
||||
// TestUsageSink_CarriesSessionFromContext 钉住打标通道:会话挂在 ctx 上时,
|
||||
// 流式与非流式两条路径交给 sink 的样本都带着它。
|
||||
//
|
||||
// 两条路径都要覆盖:它们进入适配器的方式不同(ChatStream 由循环直接驱动,
|
||||
// Chat 由 agent.RunStep 驱动),漏任何一条都会让那半边的用量归不了档。
|
||||
func TestUsageSink_CarriesSessionFromContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
const session = "web:sess-usage"
|
||||
|
||||
jsonSrv := newCompletionServer(t, http.StatusOK, "application/json", okCompletion, nil)
|
||||
streamSrv := newCompletionServer(t, http.StatusOK, "text/event-stream", strings.Join([]string{
|
||||
`data: {"model":"gpt-test","choices":[{"index":0,"delta":{"content":"你好"}}]}`,
|
||||
`data: {"model":"gpt-test","choices":[],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}`,
|
||||
`data: [DONE]`,
|
||||
}, "\n")+"\n", nil)
|
||||
anthropicSrv := newAnthropicServer(t, http.StatusOK, "application/json", anthropicTextResponse, nil)
|
||||
|
||||
ctx := WithSessionID(context.Background(), session)
|
||||
probes := []struct {
|
||||
name string
|
||||
call func(sink UsageSink) error
|
||||
}{
|
||||
{"openai 非流式 Chat", func(sink UsageSink) error {
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: jsonSrv.URL, Model: "gpt-test"}, WithUsageSink(sink))
|
||||
_, err := ad.Chat(ctx, sessionUsageRequest(), Options{})
|
||||
return err
|
||||
}},
|
||||
{"openai 流式 ChatStream", func(sink UsageSink) error {
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: streamSrv.URL, Model: "gpt-test"}, WithUsageSink(sink))
|
||||
_, err := ad.ChatStream(ctx, sessionUsageRequest(), nil)
|
||||
return err
|
||||
}},
|
||||
{"anthropic 非流式 Chat", func(sink UsageSink) error {
|
||||
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: anthropicSrv.URL, AnthropicModel: "claude-test"},
|
||||
WithAnthropicUsageSink(sink))
|
||||
_, err := ad.Chat(ctx, sessionUsageRequest(), Options{})
|
||||
return err
|
||||
}},
|
||||
}
|
||||
|
||||
for _, p := range probes {
|
||||
sink := &fakeUsageSink{}
|
||||
if err := p.call(sink); err != nil {
|
||||
t.Fatalf("%s: 调用失败: %v", p.name, err)
|
||||
}
|
||||
if len(sink.samples) != 1 {
|
||||
t.Fatalf("%s: sink 收到 %d 条样本, want 1", p.name, len(sink.samples))
|
||||
}
|
||||
if got := sink.samples[0].SessionID; got != session {
|
||||
t.Fatalf("%s: 样本 SessionID = %q, want %q(会话打标在这一路上丢了)", p.name, got, session)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageSink_UnseededSessionStaysEmpty 钉住「没播种 ⇒ 空串」:管理面板试连、
|
||||
// 未接循环的后台流程本就不属于任何会话,样本里必须是空串 —— 写入侧据此写 NULL;
|
||||
// 一旦这里变成别的值(比如 "unknown"),列里就分不出「无归属」与「真会话」。
|
||||
func TestUsageSink_UnseededSessionStaysEmpty(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newCompletionServer(t, http.StatusOK, "application/json", okCompletion, nil)
|
||||
sink := &fakeUsageSink{}
|
||||
ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"}, WithUsageSink(sink))
|
||||
|
||||
if _, err := ad.Chat(context.Background(), sessionUsageRequest(), Options{}); err != nil {
|
||||
t.Fatalf("Chat: %v", err)
|
||||
}
|
||||
if len(sink.samples) != 1 {
|
||||
t.Fatalf("sink 收到 %d 条样本, want 1", len(sink.samples))
|
||||
}
|
||||
if got := sink.samples[0].SessionID; got != "" {
|
||||
t.Fatalf("未播种会话时样本 SessionID = %q, want 空串", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageWriter_PersistsSessionID 钉住落库侧:样本带会话 ⇒ 该列写会话;
|
||||
// 样本不带会话 ⇒ 写 NULL(**不是空串**),且读回来的会话就是写入的那个。
|
||||
func TestUsageWriter_PersistsSessionID(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
conn := usageTestDB(t)
|
||||
writer := NewUsageWriter(conn, discardLogger("model/usage_session_test"))
|
||||
writer.Start(ctx)
|
||||
|
||||
const session = "web:sess-persist"
|
||||
writer.RecordUsage(UsageSample{Adapter: "openai", Model: "gpt-test", Purpose: "chat", SessionID: session})
|
||||
writer.RecordUsage(UsageSample{Adapter: "openai", Model: "gpt-test", Purpose: "chat"})
|
||||
writer.Stop()
|
||||
|
||||
var stored string
|
||||
if err := conn.QueryRowContext(ctx,
|
||||
`SELECT session_id FROM llm_usage WHERE session_id IS NOT NULL`).Scan(&stored); err != nil {
|
||||
t.Fatalf("读回 session_id: %v", err)
|
||||
}
|
||||
if stored != session {
|
||||
t.Fatalf("落库 session_id = %q, want %q", stored, session)
|
||||
}
|
||||
|
||||
var nullRows, emptyRows int64
|
||||
if err := conn.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM llm_usage WHERE session_id IS NULL`).Scan(&nullRows); err != nil {
|
||||
t.Fatalf("count null session_id: %v", err)
|
||||
}
|
||||
if err := conn.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM llm_usage WHERE session_id = ''`).Scan(&emptyRows); err != nil {
|
||||
t.Fatalf("count empty session_id: %v", err)
|
||||
}
|
||||
if nullRows != 1 {
|
||||
t.Fatalf("session_id IS NULL 的行 = %d, want 1(无归属的调用必须写 NULL)", nullRows)
|
||||
}
|
||||
if emptyRows != 0 {
|
||||
t.Fatalf("session_id = '' 的行 = %d, want 0(空串会把「无归属」写成一个真实值)", emptyRows)
|
||||
}
|
||||
}
|
||||
@@ -44,13 +44,22 @@ const (
|
||||
usageFinalFlushTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
// insertUsageSQL 写入一条用量样本。**session_id 当前恒为 NULL**(模型适配器跑在
|
||||
// loop 之下、没有会话概念;见迁移 0017 的说明),列留着是为了接线时只改这一处。
|
||||
// insertUsageSQL 写入一条用量样本。session_id 取 ctx 播种的会话标识
|
||||
// (session_ctx.go),**没有会话归属时写 NULL**:管理面板试连、未接循环的
|
||||
// 后台流程本就不属于任何会话,写空串会让它们与真实会话在列里混成一个值。
|
||||
const insertUsageSQL = `INSERT INTO llm_usage
|
||||
(session_id, purpose, adapter, model, prompt_tokens, completion_tokens,
|
||||
cache_hit_tokens, cache_miss_tokens, reasoning_tokens, total_tokens, took_ms)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
|
||||
|
||||
// usageSessionArg 把会话标识转成可绑定的值:空串 ⇒ NULL。
|
||||
func usageSessionArg(sessionID string) any {
|
||||
if sessionID == "" {
|
||||
return nil
|
||||
}
|
||||
return sessionID
|
||||
}
|
||||
|
||||
// UsageWriter 把用量样本异步落进 `llm_usage`。它实现 UsageSink,因此在模型适配器
|
||||
// 里就是一个「可选出口」:注入即有、不注入则只记日志。
|
||||
//
|
||||
@@ -234,9 +243,8 @@ func insertUsageBatch(ctx context.Context, db *sql.DB, batch []UsageSample) erro
|
||||
defer stmt.Close()
|
||||
|
||||
for i, sample := range batch {
|
||||
// session_id 传 NULL:见 insertUsageSQL 的说明。
|
||||
if _, err := stmt.ExecContext(ctx,
|
||||
nil,
|
||||
usageSessionArg(sample.SessionID),
|
||||
sample.Purpose,
|
||||
sample.Adapter,
|
||||
sample.Model,
|
||||
|
||||
@@ -195,7 +195,9 @@ func TestUsageWriter_PersistsBatchAndReaderSummarizes(t *testing.T) {
|
||||
if rows := usageRowCount(t, conn); rows != int64(len(samples)) {
|
||||
t.Fatalf("落库行数 = %d, want %d(日志:%s)", rows, len(samples), buf.String())
|
||||
}
|
||||
// session_id 当前恒为 NULL(模型层无会话概念):列在、但序列化面已说明。
|
||||
// 这里直接喂写入器、样本里**没有**会话 ⇒ session_id 落 NULL:无归属的调用
|
||||
// (管理面板试连、未接循环的后台流程)与「属于某会话」在列里必须可区分,
|
||||
// 所以「没有」写 NULL 而不是空串(打标链见 usage_session_test.go)。
|
||||
var nullSessions int64
|
||||
if err := conn.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM llm_usage WHERE session_id IS NULL`).Scan(&nullSessions); err != nil {
|
||||
|
||||
Reference in new issue
Block a user