feat(llm): add LLM call recording with admin panel integration
- llm.Adapter: SetCallLogger hook for recording every API call - llm.CallLogger: ring-buffer with Recent+Stats for admin API - main.go: wire DeepSeek adapter to CallLogger - Fix: single CallLogger instance for both recording and display - Metrics: model name, duration, success/fail, tokens - All 16 packages pass
This commit is contained in:
1 parent
5b2bcc9441
commit
6a4887e52a
6 files changed
+131
-67
No files matched your search
@@ -149,6 +149,11 @@ func main() {
|
|||||||
llmCfg.Model = cfg.LLMModel
|
llmCfg.Model = cfg.LLMModel
|
||||||
llmCfg.FallbackModel = cfg.LLMFallbackModel
|
llmCfg.FallbackModel = cfg.LLMFallbackModel
|
||||||
llmSelector := llm.NewModelSelector(llmCfg)
|
llmSelector := llm.NewModelSelector(llmCfg)
|
||||||
|
// Wire LLM call logging to admin panel
|
||||||
|
llmCallLogger := llm.NewCallLogger(200)
|
||||||
|
llmSelector.DefaultAdapter().SetCallLogger(func(c llm.CallRecord) {
|
||||||
|
llmCallLogger.Record(c)
|
||||||
|
})
|
||||||
log.Printf("[llm] adapter ready (model=%s)", cfg.LLMModel)
|
log.Printf("[llm] adapter ready (model=%s)", cfg.LLMModel)
|
||||||
|
|
||||||
// === 9. AI-Core: Context Store ===
|
// === 9. AI-Core: Context Store ===
|
||||||
@@ -211,7 +216,7 @@ func main() {
|
|||||||
authMgr := api.NewRealAuthManager(cfg.JWTSecret, cfg.JWTExpiryHours, userStore)
|
authMgr := api.NewRealAuthManager(cfg.JWTSecret, cfg.JWTExpiryHours, userStore)
|
||||||
|
|
||||||
// Build dependencies for gateway handlers
|
// Build dependencies for gateway handlers
|
||||||
llmLogger := llmlog.NewLogger(200)
|
llmLogger := &llmLoggerAdapter{CallLogger: llmCallLogger}
|
||||||
traceLogger := tracepkg.NewLogger(500)
|
traceLogger := tracepkg.NewLogger(500)
|
||||||
perfMon := perfpkg.NewMonitor(120, 10*time.Second)
|
perfMon := perfpkg.NewMonitor(120, 10*time.Second)
|
||||||
perfMon.Start()
|
perfMon.Start()
|
||||||
@@ -219,7 +224,7 @@ func main() {
|
|||||||
gwDeps := api.Dependencies{
|
gwDeps := api.Dependencies{
|
||||||
AuthMgr: authMgr,
|
AuthMgr: authMgr,
|
||||||
Orchestrator: orch,
|
Orchestrator: orch,
|
||||||
LLMLogger: &llmLoggerAdapter{Logger: llmLogger},
|
LLMLogger: llmLogger,
|
||||||
TraceLog: &traceAdapter{Logger: traceLogger},
|
TraceLog: &traceAdapter{Logger: traceLogger},
|
||||||
PerfMon: &perfAdapter{Monitor: perfMon},
|
PerfMon: &perfAdapter{Monitor: perfMon},
|
||||||
ModelConfig: api.NewModelConfigStore(),
|
ModelConfig: api.NewModelConfigStore(),
|
||||||
@@ -269,9 +274,9 @@ func main() {
|
|||||||
_ = orch
|
_ = orch
|
||||||
}
|
}
|
||||||
// llmLoggerAdapter adapts llm.Logger to api.LLMCallLogger interface.
|
// llmLoggerAdapter adapts llm.Logger to api.LLMCallLogger interface.
|
||||||
type llmLoggerAdapter struct{ *llmlog.Logger }
|
type llmLoggerAdapter struct{ *llmlog.CallLogger }
|
||||||
|
|
||||||
func (a *llmLoggerAdapter) Recent(n int) []map[string]any { return a.Logger.RecentMaps(n) }
|
func (a *llmLoggerAdapter) Recent(n int) []map[string]any { return a.CallLogger.RecentMaps(n) }
|
||||||
|
|
||||||
func pluginNamesToMap(names []string) []map[string]string {
|
func pluginNamesToMap(names []string) []map[string]string {
|
||||||
result := make([]map[string]string, len(names))
|
result := make([]map[string]string, len(names))
|
||||||
|
|||||||
@@ -0,0 +1,53 @@
|
|||||||
|
package logger
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"os"
|
||||||
|
)
|
||||||
|
|
||||||
|
// L is the global logger instance. Configure via Init().
|
||||||
|
var L = New(os.Stderr, LevelInfo, "text")
|
||||||
|
|
||||||
|
// Init configures the global logger.
|
||||||
|
func Init(level Level, format string) {
|
||||||
|
L = New(os.Stderr, level, format)
|
||||||
|
}
|
||||||
|
|
||||||
|
// InitNop sets the global logger to discard all output.
|
||||||
|
func InitNop() {
|
||||||
|
L = New(io.Discard, LevelDebug, "text")
|
||||||
|
}
|
||||||
|
|
||||||
|
// G returns the global logger.
|
||||||
|
func G() *Logger { return L }
|
||||||
|
|
||||||
|
// --- Global convenience methods ---
|
||||||
|
|
||||||
|
func Debug(msg string, args ...any) { L.Debug(msg, args...) }
|
||||||
|
func Info(msg string, args ...any) { L.Info(msg, args...) }
|
||||||
|
func Warn(msg string, args ...any) { L.Warn(msg, args...) }
|
||||||
|
func Error(msg string, args ...any) { L.Error(msg, args...) }
|
||||||
|
|
||||||
|
func Debugf(format string, args ...any) {
|
||||||
|
L.Debug(fmt.Sprintf(format, args...))
|
||||||
|
}
|
||||||
|
func Infof(format string, args ...any) {
|
||||||
|
L.Info(fmt.Sprintf(format, args...))
|
||||||
|
}
|
||||||
|
func Warnf(format string, args ...any) {
|
||||||
|
L.Warn(fmt.Sprintf(format, args...))
|
||||||
|
}
|
||||||
|
func Errorf(format string, args ...any) {
|
||||||
|
L.Error(fmt.Sprintf(format, args...))
|
||||||
|
}
|
||||||
|
|
||||||
|
func Fatalf(format string, args ...any) {
|
||||||
|
L.Error(fmt.Sprintf(format, args...))
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
|
|
||||||
|
func Fatal(msg string, args ...any) {
|
||||||
|
L.Error(msg, args...)
|
||||||
|
os.Exit(1)
|
||||||
|
}
|
||||||
@@ -4,7 +4,7 @@ package background
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"log"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@@ -45,19 +45,19 @@ func New(enabled bool, thinkInterval, silenceTimeout time.Duration) *Thinker {
|
|||||||
// Start begins the background thinking loops.
|
// Start begins the background thinking loops.
|
||||||
func (t *Thinker) Start(ctx context.Context) {
|
func (t *Thinker) Start(ctx context.Context) {
|
||||||
if !t.enabled {
|
if !t.enabled {
|
||||||
log.Println("[thinker] background thinking disabled")
|
logger.Info("[thinker] background thinking disabled")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
go t.silenceLoop(ctx)
|
go t.silenceLoop(ctx)
|
||||||
go t.postChatLoop(ctx)
|
go t.postChatLoop(ctx)
|
||||||
log.Printf("[thinker] started (interval=%v, silence=%v)", t.thinkInterval, t.silenceTimeout)
|
logger.Infof("[thinker] started (interval=%v, silence=%v)", t.thinkInterval, t.silenceTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop shuts down the thinker.
|
// Stop shuts down the thinker.
|
||||||
func (t *Thinker) Stop() {
|
func (t *Thinker) Stop() {
|
||||||
close(t.stopCh)
|
close(t.stopCh)
|
||||||
log.Println("[thinker] stopped")
|
logger.Info("[thinker] stopped")
|
||||||
}
|
}
|
||||||
|
|
||||||
// TriggerPostChat schedules a post-chat think after a conversation.
|
// TriggerPostChat schedules a post-chat think after a conversation.
|
||||||
@@ -140,7 +140,7 @@ func (t *Thinker) generateThought(trigger string, sessionID string) {
|
|||||||
}
|
}
|
||||||
t.mu.Unlock()
|
t.mu.Unlock()
|
||||||
|
|
||||||
log.Printf("[thinker] %s thought generated for session %s", trigger, sessionID)
|
logger.Infof("[thinker] %s thought generated for session %s", trigger, sessionID)
|
||||||
}
|
}
|
||||||
|
|
||||||
// RecentThoughts returns the most recent thoughts.
|
// RecentThoughts returns the most recent thoughts.
|
||||||
|
|||||||
@@ -39,8 +39,9 @@ func DefaultConfig() Config {
|
|||||||
|
|
||||||
// Adapter handles communication with an OpenAI-compatible LLM API.
|
// Adapter handles communication with an OpenAI-compatible LLM API.
|
||||||
type Adapter struct {
|
type Adapter struct {
|
||||||
cfg Config
|
cfg Config
|
||||||
client *http.Client
|
client *http.Client
|
||||||
|
onCall func(call CallRecord) // Hook for logging each LLM call
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewAdapter creates a new LLM adapter.
|
// NewAdapter creates a new LLM adapter.
|
||||||
@@ -58,11 +59,29 @@ func NewAdapter(cfg Config) *Adapter {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetCallLogger injects a callback that is invoked after each LLM call.
|
||||||
|
func (a *Adapter) SetCallLogger(fn func(call CallRecord)) {
|
||||||
|
a.onCall = fn
|
||||||
|
}
|
||||||
|
|
||||||
|
// CallRecord holds metadata about a single LLM API call.
|
||||||
|
type CallRecord struct {
|
||||||
|
Time time.Time
|
||||||
|
Model string
|
||||||
|
Purpose string
|
||||||
|
PromptTokens int
|
||||||
|
CompletionTokens int
|
||||||
|
DurationMs int64
|
||||||
|
Success bool
|
||||||
|
Error string
|
||||||
|
}
|
||||||
|
|
||||||
// ModelName returns the configured model name.
|
// ModelName returns the configured model name.
|
||||||
func (a *Adapter) ModelName() string { return a.cfg.Model }
|
func (a *Adapter) ModelName() string { return a.cfg.Model }
|
||||||
|
|
||||||
// Chat sends a chat completion request and returns the full response.
|
// Chat sends a chat completion request and returns the full response.
|
||||||
func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
|
func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
|
||||||
|
start := time.Now()
|
||||||
body := map[string]any{
|
body := map[string]any{
|
||||||
"model": a.cfg.Model,
|
"model": a.cfg.Model,
|
||||||
"messages": messages,
|
"messages": messages,
|
||||||
@@ -73,27 +92,23 @@ func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model
|
|||||||
if choices, ok := data["choices"].([]any); ok && len(choices) > 0 {
|
if choices, ok := data["choices"].([]any); ok && len(choices) > 0 {
|
||||||
choice := choices[0].(map[string]any)
|
choice := choices[0].(map[string]any)
|
||||||
if msg, ok := choice["message"].(map[string]any); ok {
|
if msg, ok := choice["message"].(map[string]any); ok {
|
||||||
if content, ok := msg["content"].(string); ok {
|
if content, ok := msg["content"].(string); ok { resp.Content = content }
|
||||||
resp.Content = content
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if reason, ok := choice["finish_reason"].(string); ok {
|
|
||||||
resp.FinishReason = reason
|
|
||||||
}
|
}
|
||||||
|
if reason, ok := choice["finish_reason"].(string); ok { resp.FinishReason = reason }
|
||||||
}
|
}
|
||||||
if usage, ok := data["usage"].(map[string]any); ok {
|
if usage, ok := data["usage"].(map[string]any); ok {
|
||||||
if total, ok := usage["total_tokens"].(float64); ok {
|
if total, ok := usage["total_tokens"].(float64); ok { resp.TokensUsed = int(total) }
|
||||||
resp.TokensUsed = int(total)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|
||||||
|
a.recordCall(start, "chat", resp.TokensUsed, 0, err)
|
||||||
return resp, err
|
return resp, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// ChatStream sends a chat completion request and returns a token channel for streaming.
|
// ChatStream sends a chat completion request and returns a token channel.
|
||||||
func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (<-chan model.StreamToken, error) {
|
func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (<-chan model.StreamToken, error) {
|
||||||
|
start := time.Now()
|
||||||
body := map[string]any{
|
body := map[string]any{
|
||||||
"model": a.cfg.Model,
|
"model": a.cfg.Model,
|
||||||
"messages": messages,
|
"messages": messages,
|
||||||
@@ -101,14 +116,13 @@ func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (
|
|||||||
}
|
}
|
||||||
|
|
||||||
ch := make(chan model.StreamToken, 64)
|
ch := make(chan model.StreamToken, 64)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
defer close(ch)
|
defer close(ch)
|
||||||
|
|
||||||
err := a.doStreamRequest(ctx, "POST", "/chat/completions", body, ch)
|
err := a.doStreamRequest(ctx, "POST", "/chat/completions", body, ch)
|
||||||
|
a.recordCall(start, "chat_stream", 0, 0, err)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
select {
|
select {
|
||||||
case ch <- model.StreamToken{Content: fmt.Sprintf("[LLM error: %v]", err), Done: true}:
|
case ch <- model.StreamToken{Content: fmt.Sprintf("[Error: %v]", err), Done: true}:
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -117,6 +131,21 @@ func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (
|
|||||||
return ch, nil
|
return ch, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (a *Adapter) recordCall(start time.Time, purpose string, promptTok, compTok int, err error) {
|
||||||
|
if a.onCall != nil {
|
||||||
|
a.onCall(CallRecord{
|
||||||
|
Time: start,
|
||||||
|
Model: a.cfg.Model,
|
||||||
|
Purpose: purpose,
|
||||||
|
PromptTokens: promptTok,
|
||||||
|
CompletionTokens: compTok,
|
||||||
|
DurationMs: time.Since(start).Milliseconds(),
|
||||||
|
Success: err == nil,
|
||||||
|
Error: func() string { if err != nil { return err.Error() }; return "" }(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// doRequest sends a non-streaming request and parses the response.
|
// doRequest sends a non-streaming request and parses the response.
|
||||||
func (a *Adapter) doRequest(ctx context.Context, method, path string, body map[string]any, parser func(map[string]any) error) error {
|
func (a *Adapter) doRequest(ctx context.Context, method, path string, body map[string]any, parser func(map[string]any) error) error {
|
||||||
jsonBody, _ := json.Marshal(body)
|
jsonBody, _ := json.Marshal(body)
|
||||||
|
|||||||
@@ -5,19 +5,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
type CallRecord struct {
|
type CallLogger struct {
|
||||||
ID string `json:"id"`
|
|
||||||
Time time.Time `json:"time"`
|
|
||||||
Model string `json:"model"`
|
|
||||||
Purpose string `json:"purpose"`
|
|
||||||
PromptTokens int `json:"prompt_tokens"`
|
|
||||||
CompTokens int `json:"completion_tokens"`
|
|
||||||
DurationMs int64 `json:"duration_ms"`
|
|
||||||
Success bool `json:"success"`
|
|
||||||
Error string `json:"error,omitempty"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type Logger struct {
|
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
records []CallRecord
|
records []CallRecord
|
||||||
max int
|
max int
|
||||||
@@ -25,24 +13,20 @@ type Logger struct {
|
|||||||
total int64
|
total int64
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewLogger(max int) *Logger { return &Logger{records: make([]CallRecord, max), max: max} }
|
func NewCallLogger(max int) *CallLogger { return &CallLogger{records: make([]CallRecord, max), max: max} }
|
||||||
|
|
||||||
func (l *Logger) Record(r CallRecord) {
|
func (l *CallLogger) Record(r CallRecord) {
|
||||||
l.mu.Lock()
|
l.mu.Lock()
|
||||||
defer l.mu.Unlock()
|
defer l.mu.Unlock()
|
||||||
r.ID = time.Now().Format("150405") + "-" + r.Purpose
|
|
||||||
l.records[l.index%l.max] = r
|
l.records[l.index%l.max] = r
|
||||||
l.index++
|
l.index++
|
||||||
l.total++
|
l.total++
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *Logger) Recent(n int) []CallRecord {
|
func (l *CallLogger) Recent(n int) []CallRecord {
|
||||||
l.mu.RLock()
|
l.mu.RLock(); defer l.mu.RUnlock()
|
||||||
defer l.mu.RUnlock()
|
|
||||||
if n > l.max { n = l.max }
|
if n > l.max { n = l.max }
|
||||||
count := l.index
|
count := l.index; if count > l.max { count = l.max }; if n > count { n = count }
|
||||||
if count > l.max { count = l.max }
|
|
||||||
if n > count { n = count }
|
|
||||||
result := make([]CallRecord, n)
|
result := make([]CallRecord, n)
|
||||||
for i := 0; i < n; i++ {
|
for i := 0; i < n; i++ {
|
||||||
idx := (l.index - 1 - i + l.max) % l.max
|
idx := (l.index - 1 - i + l.max) % l.max
|
||||||
@@ -51,34 +35,27 @@ func (l *Logger) Recent(n int) []CallRecord {
|
|||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *Logger) RecentMaps(n int) []map[string]any {
|
func (l *CallLogger) RecentMaps(n int) []map[string]any {
|
||||||
records := l.Recent(n)
|
records := l.Recent(n); result := make([]map[string]any, len(records))
|
||||||
result := make([]map[string]any, len(records))
|
|
||||||
for i, r := range records {
|
for i, r := range records {
|
||||||
result[i] = map[string]any{
|
result[i] = map[string]any{
|
||||||
"id": r.ID, "time": r.Time.Format(time.RFC3339), "model": r.Model,
|
"time": r.Time.Format(time.RFC3339), "model": r.Model, "purpose": r.Purpose,
|
||||||
"purpose": r.Purpose, "prompt_tokens": r.PromptTokens, "completion_tokens": r.CompTokens,
|
"prompt_tokens": r.PromptTokens, "completion_tokens": r.CompletionTokens,
|
||||||
"duration_ms": r.DurationMs, "success": r.Success, "error": r.Error,
|
"duration_ms": r.DurationMs, "success": r.Success, "error": r.Error,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
func (l *Logger) Stats() map[string]any {
|
func (l *CallLogger) Stats() map[string]any {
|
||||||
l.mu.RLock()
|
l.mu.RLock(); defer l.mu.RUnlock()
|
||||||
defer l.mu.RUnlock()
|
var calls, succ int64; var dur, tok int64
|
||||||
var calls, succ int64
|
count := l.index; if count > l.max { count = l.max }
|
||||||
var dur, tok int64
|
|
||||||
count := l.index
|
|
||||||
if count > l.max { count = l.max }
|
|
||||||
for i := 0; i < count; i++ {
|
for i := 0; i < count; i++ {
|
||||||
r := l.records[(l.index-1-i+l.max)%l.max]
|
r := l.records[(l.index-1-i+l.max)%l.max]
|
||||||
calls++
|
calls++; if r.Success { succ++ }
|
||||||
if r.Success { succ++ }
|
dur += r.DurationMs; tok += int64(r.PromptTokens + r.CompletionTokens)
|
||||||
dur += r.DurationMs
|
|
||||||
tok += int64(r.PromptTokens + r.CompTokens)
|
|
||||||
}
|
}
|
||||||
avg := 0
|
avg := 0; if calls > 0 { avg = int(dur / calls) }
|
||||||
if calls > 0 { avg = int(dur / calls) }
|
|
||||||
return map[string]any{"total_calls": calls, "success_calls": succ, "avg_duration_ms": avg, "total_tokens": tok}
|
return map[string]any{"total_calls": calls, "success_calls": succ, "avg_duration_ms": avg, "total_tokens": tok}
|
||||||
}
|
}
|
||||||
@@ -6,7 +6,7 @@ package bridge
|
|||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -83,7 +83,7 @@ func (s *ConfigStore) StartWatching(interval time.Duration) {
|
|||||||
lastMod = info.ModTime()
|
lastMod = info.ModTime()
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
if s.load() {
|
if s.load() {
|
||||||
log.Printf("[bridge] config hot-reloaded (%d adapters)", len(s.config.Adapters))
|
logger.Infof("[bridge] config hot-reloaded (%d adapters)", len(s.config.Adapters))
|
||||||
if s.onChange != nil {
|
if s.onChange != nil {
|
||||||
s.onChange(s.config)
|
s.onChange(s.config)
|
||||||
}
|
}
|
||||||
@@ -184,7 +184,7 @@ func (s *ConfigStore) load() bool {
|
|||||||
}
|
}
|
||||||
var cfg BridgeConfig
|
var cfg BridgeConfig
|
||||||
if err := json.Unmarshal(data, &cfg); err != nil {
|
if err := json.Unmarshal(data, &cfg); err != nil {
|
||||||
log.Printf("[bridge] config parse error: %v", err)
|
logger.Infof("[bridge] config parse error: %v", err)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
if cfg.Adapters == nil {
|
if cfg.Adapters == nil {
|
||||||
|
|||||||
Reference in new issue
Block a user