mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-13 11:40:53 +08:00
feat: cost ledger (per-profile EUR pricing on token usage)
The token metering is real now (input, output, cache read/write per profile), but it answers "how many tokens" while the question an operator actually asks is "what did this run cost". Converting requires per-profile prices, which nowhere exist. Changes: - llm_profiles gains four optional price columns (EUR per 1M tokens: input, output, cache read, cache write; 0/NULL = unset), migrated idempotently and round-tripped through SaveProfile / the profile API in one small auditable write - the usage aggregation computes cost_eur per profile from the stored prices, in Go after the archived-task merge so cold-storage tokens are priced identically; profiles without prices keep cost_eur nil instead of showing a fake zero - web types carry cost_eur and the price fields; the dashboard card and profile-dialog inputs consume them in a follow-up (the API serves the data already) - integration test: priced profile + usage rows -> expected EUR to the cent (2M in @1.5 + 0.5M out @4 + 1M cache-read @0.15 = 5.15), plus a prices round-trip check Note: the Go side activates on the next build; the schema migration is idempotent at startup.
This commit is contained in:
BaaRai
+26
-3
@@ -64,6 +64,12 @@ type LLMProfile struct {
|
||||
// or sticky routing off a session-id header. "" = not sent. Maps to
|
||||
// agent.Config.SessionHeaderKey.
|
||||
SessionHeaderKey string `json:"session_header_key"`
|
||||
// Cost ledger: EUR per 1M tokens (0 = unset; cost is computed only when a
|
||||
// price is set). Set through the same profile API/SaveProfile path.
|
||||
PriceInputPerM float64 `json:"price_input_per_m,omitempty"`
|
||||
PriceOutputPerM float64 `json:"price_output_per_m,omitempty"`
|
||||
PriceCacheReadPerM float64 `json:"price_cache_read_per_m,omitempty"`
|
||||
PriceCacheWritePerM float64 `json:"price_cache_write_per_m,omitempty"`
|
||||
// Retry overrides this profile's share of the retry ladder. Zero value =
|
||||
// inherit the global policy (LLMRetryPolicy), so an untouched profile behaves
|
||||
// exactly as before. See RetryOverride.
|
||||
@@ -83,8 +89,8 @@ type RetryOverride struct {
|
||||
// profileCols is the read column list (hint variant, no api key) shared by the
|
||||
// list query; profileColsKey is the same with api_key for the single-row loads.
|
||||
const profileRetryCols = `COALESCE(retry_connect_attempts,0),COALESCE(retry_connect_interval_ms,0),COALESCE(retry_empty_attempts,0),COALESCE(retry_empty_interval_ms,0),COALESCE(retry_stream_attempts,0),COALESCE(retry_stream_interval_ms,0)`
|
||||
const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols
|
||||
const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols
|
||||
const profileCols = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key_hint,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols + `,COALESCE(price_input_per_m,0),COALESCE(price_output_per_m,0),COALESCE(price_cache_read_per_m,0),COALESCE(price_cache_write_per_m,0)`
|
||||
const profileColsKey = `id,name,format,COALESCE(base_url,''),COALESCE(proxy,''),model,COALESCE(api_key,''),rate_per_second,rate_per_minute,context_window_k,COALESCE(reasoning_effort,''),is_default,priority,pool_exclude,COALESCE(thinking_type,''),COALESCE(streaming,true),COALESCE(max_tokens,0),COALESCE(max_tokens_field,''),COALESCE(session_header_key,''),` + profileRetryCols + `,COALESCE(price_input_per_m,0),COALESCE(price_output_per_m,0),COALESCE(price_cache_read_per_m,0),COALESCE(price_cache_write_per_m,0)`
|
||||
|
||||
// scanProfile reads one row in the profileCols / profileColsKey column order. The
|
||||
// 7th column lands in APIKeyHint or APIKey depending on which list the caller used.
|
||||
@@ -94,7 +100,8 @@ func scanProfile(sc interface{ Scan(...any) error }, into *string, p *LLMProfile
|
||||
&p.MaxTokens, &p.MaxTokensField, &p.SessionHeaderKey,
|
||||
&p.Retry.Connect.Attempts, &p.Retry.Connect.IntervalMS,
|
||||
&p.Retry.Empty.Attempts, &p.Retry.Empty.IntervalMS,
|
||||
&p.Retry.Stream.Attempts, &p.Retry.Stream.IntervalMS)
|
||||
&p.Retry.Stream.Attempts, &p.Retry.Stream.IntervalMS,
|
||||
&p.PriceInputPerM, &p.PriceOutputPerM, &p.PriceCacheReadPerM, &p.PriceCacheWritePerM)
|
||||
}
|
||||
|
||||
func (d *DB) ListProfiles() ([]*LLMProfile, error) {
|
||||
@@ -176,20 +183,36 @@ func (d *DB) SaveProfile(p *LLMProfile) (int64, error) {
|
||||
VALUES ($1,$2,NULLIF($3,''),NULLIF($4,''),$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24) RETURNING id`,
|
||||
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey,
|
||||
r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS).Scan(&id)
|
||||
if err == nil {
|
||||
_, err = d.saveProfilePrices(id, p)
|
||||
}
|
||||
return id, err
|
||||
}
|
||||
if p.APIKey == "" {
|
||||
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,rate_per_second=$6,rate_per_minute=$7,context_window_k=$8,reasoning_effort=$9,priority=$10,pool_exclude=$11,thinking_type=$12,streaming=$13,max_tokens=$14,max_tokens_field=$15,session_header_key=$16,retry_connect_attempts=$17,retry_connect_interval_ms=$18,retry_empty_attempts=$19,retry_empty_interval_ms=$20,retry_stream_attempts=$21,retry_stream_interval_ms=$22 WHERE id=$23`,
|
||||
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey,
|
||||
r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS, p.ID)
|
||||
if err == nil {
|
||||
_, err = d.saveProfilePrices(p.ID, p)
|
||||
}
|
||||
return p.ID, err
|
||||
}
|
||||
_, err := d.Exec(`UPDATE llm_profiles SET name=$1,format=$2,base_url=NULLIF($3,''),proxy=NULLIF($4,''),model=$5,api_key=$6,api_key_hint=$7,rate_per_second=$8,rate_per_minute=$9,context_window_k=$10,reasoning_effort=$11,priority=$12,pool_exclude=$13,thinking_type=$14,streaming=$15,max_tokens=$16,max_tokens_field=$17,session_header_key=$18,retry_connect_attempts=$19,retry_connect_interval_ms=$20,retry_empty_attempts=$21,retry_empty_interval_ms=$22,retry_stream_attempts=$23,retry_stream_interval_ms=$24 WHERE id=$25`,
|
||||
p.Name, p.Format, p.BaseURL, p.Proxy, p.Model, p.APIKey, hint, p.RatePerSecond, p.RatePerMinute, p.ContextWindowK, p.ReasoningEffort, p.Priority, p.PoolExclude, p.ThinkingType, p.Streaming, p.MaxTokens, p.MaxTokensField, p.SessionHeaderKey,
|
||||
r.Connect.Attempts, r.Connect.IntervalMS, r.Empty.Attempts, r.Empty.IntervalMS, r.Stream.Attempts, r.Stream.IntervalMS, p.ID)
|
||||
if err == nil {
|
||||
_, err = d.saveProfilePrices(p.ID, p)
|
||||
}
|
||||
return p.ID, err
|
||||
}
|
||||
|
||||
// saveProfilePrices writes the cost-ledger columns (kept out of the three main
|
||||
// statements so the pricing surface stays one small, auditable write).
|
||||
func (d *DB) saveProfilePrices(id int64, p *LLMProfile) (sql.Result, error) {
|
||||
return d.Exec(`UPDATE llm_profiles SET price_input_per_m=($1),price_output_per_m=($2),price_cache_read_per_m=($3),price_cache_write_per_m=($4) WHERE id=$5`,
|
||||
p.PriceInputPerM, p.PriceOutputPerM, p.PriceCacheReadPerM, p.PriceCacheWritePerM, id)
|
||||
}
|
||||
|
||||
var (
|
||||
ErrActiveLLMProfileDelete = errors.New("cannot delete the active LLM profile; activate another profile first")
|
||||
ErrLLMProfileReferencesChanged = errors.New("LLM profile references changed while deleting; retry the request")
|
||||
|
||||
@@ -117,6 +117,8 @@ type ProfileUsage struct {
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
CacheReadTokens int `json:"cache_read_tokens"`
|
||||
CacheWriteTokens int `json:"cache_write_tokens"`
|
||||
// CostEur is set only when the profile carries at least one price; nil = not priced.
|
||||
CostEur *float64 `json:"cost_eur,omitempty"`
|
||||
}
|
||||
|
||||
// UsageByProfile returns global token spend grouped by profile name, most-used
|
||||
@@ -171,6 +173,9 @@ ORDER BY SUM(input_tokens) + SUM(output_tokens) DESC`)
|
||||
for _, current := range byName {
|
||||
out = append(out, current)
|
||||
}
|
||||
if err := applyCostEur(d, out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
left := out[i].InputTokens + out[i].OutputTokens
|
||||
right := out[j].InputTokens + out[j].OutputTokens
|
||||
@@ -332,3 +337,44 @@ func ParseExpID(s string) int64 {
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// applyCostEur fills CostEur on each profile bucket from the stored per-profile
|
||||
// prices (EUR per 1M tokens). Profiles without any price keep CostEur nil. The
|
||||
// join is by profile name (llm_usage records the name, not the id); the first
|
||||
// matching profile wins, and duplicate names are a configuration error the
|
||||
// operator sees as a weird number, not a crash.
|
||||
func applyCostEur(d *DB, out []ProfileUsage) error {
|
||||
rows, err := d.Query(`SELECT name, COALESCE(price_input_per_m,0), COALESCE(price_output_per_m,0), COALESCE(price_cache_read_per_m,0), COALESCE(price_cache_write_per_m,0) FROM llm_profiles`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
type prices struct{ in, outp, cr, cw float64 }
|
||||
byName := map[string]prices{}
|
||||
seen := map[string]bool{}
|
||||
for rows.Next() {
|
||||
var n string
|
||||
var pr prices
|
||||
if err := rows.Scan(&n, &pr.in, &pr.outp, &pr.cr, &pr.cw); err != nil {
|
||||
return err
|
||||
}
|
||||
if !seen[n] {
|
||||
byName[n] = pr
|
||||
seen[n] = true
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range out {
|
||||
pr, ok := byName[out[i].ProfileName]
|
||||
if !ok || (pr.in == 0 && pr.outp == 0 && pr.cr == 0 && pr.cw == 0) {
|
||||
continue
|
||||
}
|
||||
cost := float64(out[i].InputTokens)/1e6*pr.in + float64(out[i].OutputTokens)/1e6*pr.outp +
|
||||
float64(out[i].CacheReadTokens)/1e6*pr.cr + float64(out[i].CacheWriteTokens)/1e6*pr.cw
|
||||
c := cost
|
||||
out[i].CostEur = &c
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1425,3 +1425,10 @@ CREATE INDEX IF NOT EXISTS idx_notification_deliveries_batch
|
||||
ON notification_deliveries(batch_id) WHERE batch_id IS NOT NULL;
|
||||
CREATE INDEX IF NOT EXISTS idx_notification_deliveries_channel
|
||||
ON notification_deliveries(channel_id, id DESC);
|
||||
|
||||
-- Cost ledger (#70): optional per-profile pricing, EUR per 1M tokens. NULL/0 = unset;
|
||||
-- usage aggregations compute cost only when at least one price is set.
|
||||
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS price_input_per_m NUMERIC(12,6);
|
||||
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS price_output_per_m NUMERIC(12,6);
|
||||
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS price_cache_read_per_m NUMERIC(12,6);
|
||||
ALTER TABLE llm_profiles ADD COLUMN IF NOT EXISTS price_cache_write_per_m NUMERIC(12,6);
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/Autumn-27/artex/db"
|
||||
)
|
||||
|
||||
// The cost ledger: profile prices (EUR per 1M tokens) turn token buckets into
|
||||
// EUR on the usage aggregation; unpriced profiles stay nil.
|
||||
func TestCostLedgerEurOnUsage(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { m.Close() })
|
||||
if m.pg == nil {
|
||||
t.Skip("no postgres for this run")
|
||||
}
|
||||
|
||||
pid, err := m.pg.SaveProfile(&db.LLMProfile{
|
||||
Name: "cost-test", Format: "openai", Model: "m", APIKey: "k",
|
||||
PriceInputPerM: 1.5, PriceOutputPerM: 4.0, PriceCacheReadPerM: 0.15,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _, _ = m.pg.Exec("DELETE FROM llm_profiles WHERE id=$1", pid) })
|
||||
|
||||
stored, err := m.pg.ProfileByID(pid)
|
||||
if err != nil || stored == nil {
|
||||
t.Fatalf("ProfileByID: %v %v", stored, err)
|
||||
}
|
||||
if stored.PriceInputPerM != 1.5 || stored.PriceOutputPerM != 4.0 || stored.PriceCacheReadPerM != 0.15 || stored.PriceCacheWritePerM != 0 {
|
||||
t.Fatalf("prices did not round-trip: %+v", *stored)
|
||||
}
|
||||
|
||||
if _, err := m.pg.Exec(`INSERT INTO llm_usage(task_id, profile_name, model, input_tokens, output_tokens, cache_read, cache_write, ts)
|
||||
VALUES ('', 'cost-test', 'm', 2000000, 500000, 1000000, 0, now())`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _, _ = m.pg.Exec(`DELETE FROM llm_usage WHERE profile_name='cost-test'`) })
|
||||
|
||||
profiles, err := m.pg.UsageByProfile()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got *float64
|
||||
for _, p := range profiles {
|
||||
if p.ProfileName == "cost-test" {
|
||||
got = p.CostEur
|
||||
}
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("priced profile has no cost_eur")
|
||||
}
|
||||
want := 2.0*1.5 + 0.5*4.0 + 1.0*0.15 // 3.0 + 2.0 + 0.15 = 5.15
|
||||
if d := *got - want; d > 0.001 || d < -0.001 {
|
||||
t.Fatalf("cost_eur = %v, want %v", *got, want)
|
||||
}
|
||||
}
|
||||
@@ -818,6 +818,8 @@ export interface ProfileUsage {
|
||||
output_tokens: number;
|
||||
cache_read_tokens: number;
|
||||
cache_write_tokens: number;
|
||||
// EUR cost from the profile's stored prices; absent when the profile has no price set.
|
||||
cost_eur?: number | null;
|
||||
}
|
||||
|
||||
// One (profile, UTC day) token bucket for the dashboard's daily chart (new source).
|
||||
@@ -1045,6 +1047,11 @@ export interface LLMProfile {
|
||||
pool_exclude?: boolean;
|
||||
// true (default) = streaming (SSE) | false = genuinely non-streaming (stream:false, one-shot response).
|
||||
streaming?: boolean;
|
||||
// Cost ledger: EUR per 1M tokens (0/absent = unset).
|
||||
price_input_per_m?: number;
|
||||
price_output_per_m?: number;
|
||||
price_cache_read_per_m?: number;
|
||||
price_cache_write_per_m?: number;
|
||||
// Output cap per reply (tokens). 0 = do not send the field; the server default decides.
|
||||
// Note the distinction from context_window_k: that is the model's total capacity, used locally only for the compression threshold.
|
||||
max_tokens?: number;
|
||||
|
||||
Reference in New Issue
Block a user