From 8603839711d18bd02d2d55c48aeef530ee5ca4b0 Mon Sep 17 00:00:00 2001 From: Aska Date: Thu, 20 Aug 2026 20:57:27 +0800 Subject: [PATCH] feat(ai-core): trace instrumentation, LLM thinker, decay scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - orchestrator: Tracer interface with spans for intent analysis, persona, subsession dispatch, context build, LLM streaming and each tool call; traces finish when the stream completes — the admin Trace waterfall finally receives data - background: Thinker generates real LLM thoughts (post-chat + silence triggers, recent history as context, empty/overlong outputs discarded); thoughts broadcast to web clients as 'thinking' WS messages; pipeline postChatHook triggers post-chat thinking after each round - memory: ApplyDecayAll + ListUserIDs with a 6h decay/purge cron in main (was implemented but never scheduled) --- backend/cmd/cyrene/main.go | 39 ++++++++ backend/cmd/cyrene/trace_adapter.go | 6 ++ .../services/ai-core/background/thinker.go | 90 ++++++++++++++++-- .../ai-core/background/thinker_test.go | 75 +++++++++++++++ .../ai-core/orchestrator/orchestrator.go | 95 ++++++++++++++++++- backend/services/memory/repository.go | 18 ++++ backend/services/memory/service.go | 17 ++++ frontend/web/src/hooks/useWebSocket.ts | 8 ++ 8 files changed, 338 insertions(+), 10 deletions(-) create mode 100644 backend/services/ai-core/background/thinker_test.go diff --git a/backend/cmd/cyrene/main.go b/backend/cmd/cyrene/main.go index 58ec3a5..dea3312 100644 --- a/backend/cmd/cyrene/main.go +++ b/backend/cmd/cyrene/main.go @@ -32,6 +32,7 @@ import ( webfetch "git.yeij.top/AskaEth/Cyrene_v2/plugins/web-fetch" websearch "git.yeij.top/AskaEth/Cyrene_v2/plugins/web-search" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/background" "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm" llmlog "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm" tracepkg "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/trace" @@ -244,6 +245,37 @@ func main() { perfMon := perfpkg.NewMonitor(120, 10*time.Second) perfMon.Start() + // Wire tracing into the dialogue pipeline + orch.SetTracer(pipelineTracer{traceLogger}) + + // === 12.5 Background Thinker === + thinker := background.New(true, 5*time.Minute, 30*time.Minute) + thinker.SetLLM(&memory.LLMAdapterBridge{Adapter: llmSelector.DefaultAdapter()}) + thinker.SetHistoryProvider(func(sessionID string, limit int) []model.Message { + msgs := ctxStore.Get(sessionID) + if len(msgs) > limit { + msgs = msgs[len(msgs)-limit:] + } + return msgs + }) + thinker.Start(context.Background()) + orch.SetPostChatHook(thinker.TriggerPostChat) + + // === 12.6 Memory decay scheduler === + if memSvc != nil { + go func() { + ticker := time.NewTicker(6 * time.Hour) + defer ticker.Stop() + for range ticker.C { + if n, err := memSvc.ApplyDecayAll(context.Background()); err != nil { + logger.Infof("[memory] decay run failed: %v", err) + } else if n > 0 { + logger.Infof("[memory] decay applied (%d memories updated)", n) + } + } + }() + } + gwDeps := api.Dependencies{ AuthMgr: authMgr, Orchestrator: orch, @@ -267,6 +299,13 @@ func main() { wsHub := ws.NewHub(orch) gwDeps.WSHub = wsHub + // Push background thoughts to connected web clients + go func() { + for th := range thinker.ThoughtChan() { + wsHub.Broadcast(map[string]any{"type": "thinking", "content": th.Content}) + } + }() + apiRouter := api.New(gwDeps) wsMux := http.NewServeMux() diff --git a/backend/cmd/cyrene/trace_adapter.go b/backend/cmd/cyrene/trace_adapter.go index 2ed7895..cccd335 100644 --- a/backend/cmd/cyrene/trace_adapter.go +++ b/backend/cmd/cyrene/trace_adapter.go @@ -40,6 +40,12 @@ func (a *traceAdapter) Recent(n int) []map[string]any { func (a *traceAdapter) Clear() { a.Logger.Clear() } +// pipelineTracer adapts trace.Logger to orchestrator.Tracer +// (StartTrace's return value is discarded at the call site). +type pipelineTracer struct{ *tracepkg.Logger } + +func (p pipelineTracer) StartTrace(id, userMsg string) { p.Logger.StartTrace(id, userMsg) } + func spanToMap(s tracepkg.Span) map[string]any { m := map[string]any{ "name": s.Name, "label": s.Label, diff --git a/backend/services/ai-core/background/thinker.go b/backend/services/ai-core/background/thinker.go index a58bc1f..b195280 100644 --- a/backend/services/ai-core/background/thinker.go +++ b/backend/services/ai-core/background/thinker.go @@ -4,11 +4,23 @@ package background import ( "context" + "fmt" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "strings" "sync" "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" ) +// LLMClient is the minimal LLM interface needed for thought generation. +type LLMClient interface { + Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) +} + +// HistoryProvider returns recent conversation messages for a session. +type HistoryProvider func(sessionID string, limit int) []model.Message + // Thinker manages autonomous background thinking. type Thinker struct { mu sync.Mutex @@ -20,6 +32,9 @@ type Thinker struct { thoughts []Thought stopCh chan struct{} postChatCh chan string + llm LLMClient + history HistoryProvider + out chan Thought } // Thought represents an autonomous thought generated by the AI. @@ -39,9 +54,16 @@ func New(enabled bool, thinkInterval, silenceTimeout time.Duration) *Thinker { thoughts: make([]Thought, 0), stopCh: make(chan struct{}), postChatCh: make(chan string, 10), + out: make(chan Thought, 10), } } +// SetLLM injects the LLM used to generate thoughts. +func (t *Thinker) SetLLM(c LLMClient) { t.llm = c } + +// SetHistoryProvider injects a way to fetch recent conversation context. +func (t *Thinker) SetHistoryProvider(fn HistoryProvider) { t.history = fn } + // Start begins the background thinking loops. func (t *Thinker) Start(ctx context.Context) { if !t.enabled { @@ -80,12 +102,9 @@ func (t *Thinker) UpdateActivity() { t.mu.Unlock() } -// ThoughtChan returns a channel that emits new thoughts. -// Not yet wired — will connect to NATS for event broadcasting. -func (t *Thinker) ThoughtChan() chan Thought { - ch := make(chan Thought, 10) - // In future: push thoughts to this channel - return ch +// ThoughtChan returns the channel that emits new thoughts. +func (t *Thinker) ThoughtChan() <-chan Thought { + return t.out } func (t *Thinker) silenceLoop(ctx context.Context) { @@ -126,9 +145,13 @@ func (t *Thinker) postChatLoop(ctx context.Context) { } func (t *Thinker) generateThought(trigger string, sessionID string) { - // Placeholder: in production this calls the LLM + content := t.think(trigger, sessionID) + if content == "" { + return + } + thought := Thought{ - Content: "[思考] 当前无主动想法", + Content: content, SessionID: sessionID, CreatedAt: time.Now(), } @@ -140,9 +163,60 @@ func (t *Thinker) generateThought(trigger string, sessionID string) { } t.mu.Unlock() + select { + case t.out <- thought: + default: + } + logger.Infof("[thinker] %s thought generated for session %s", trigger, sessionID) } +// think calls the LLM to produce a short inner thought. Returns "" when +// there is nothing worth saying (or no LLM is configured). +func (t *Thinker) think(trigger, sessionID string) string { + if t.llm == nil { + return "" + } + + cue := "刚结束和主人的一段对话。" + if trigger == "silence" { + cue = "主人已经安静了一段时间。" + } + + var sb strings.Builder + if t.history != nil && sessionID != "" { + for _, m := range t.history(sessionID, 6) { + role := "主人" + if m.Role == "assistant" { + role = "你" + } + fmt.Fprintf(&sb, "%s: %s\n", role, m.Content) + } + } + + messages := []model.LLMMessage{ + {Role: "system", Content: `你是昔涟的内心独白模块。根据最近的对话，生成一条简短的内心想法或想对主人说的话。 +要求：一两句话，温柔自然的少女口吻；如果没有值得说的内容，只输出一个空字符串。 +不要解释，不要加引号，直接输出想法本身。`}, + {Role: "user", Content: fmt.Sprintf("情境：%s\n最近对话：\n%s", cue, sb.String())}, + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + resp, err := t.llm.Chat(ctx, messages) + if err != nil { + logger.Infof("[thinker] llm error: %v", err) + return "" + } + + content := strings.TrimSpace(resp.Content) + // Discard empty or overly long outputs + if content == "" || len([]rune(content)) > 100 { + return "" + } + return content +} + // RecentThoughts returns the most recent thoughts. func (t *Thinker) RecentThoughts(n int) []Thought { t.mu.Lock() diff --git a/backend/services/ai-core/background/thinker_test.go b/backend/services/ai-core/background/thinker_test.go new file mode 100644 index 0000000..d8d2435 --- /dev/null +++ b/backend/services/ai-core/background/thinker_test.go @@ -0,0 +1,75 @@ +package background + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +type fakeLLM struct{ content string } + +func (f *fakeLLM) Chat(_ context.Context, _ []model.LLMMessage) (*model.LLMResponse, error) { + return &model.LLMResponse{Content: f.content}, nil +} + +func TestGenerateThoughtWithLLM(t *testing.T) { + th := New(false, time.Minute, time.Minute) + th.SetLLM(&fakeLLM{content: "主人今天好像有点累呢。"}) + + th.generateThought("post_chat", "s1") + + select { + case thought := <-th.ThoughtChan(): + if thought.Content != "主人今天好像有点累呢。" { + t.Errorf("unexpected thought: %q", thought.Content) + } + case <-time.After(time.Second): + t.Fatal("no thought emitted") + } + + if got := th.RecentThoughts(1); len(got) != 1 { + t.Errorf("expected 1 stored thought, got %d", len(got)) + } +} + +func TestGenerateThoughtEmptyResponse(t *testing.T) { + th := New(false, time.Minute, time.Minute) + th.SetLLM(&fakeLLM{content: " "}) + + th.generateThought("silence", "s1") + + select { + case thought := <-th.ThoughtChan(): + t.Errorf("empty thought should be discarded, got %q", thought.Content) + case <-time.After(50 * time.Millisecond): + } + + if got := len(th.RecentThoughts(10)); got != 0 { + t.Errorf("expected 0 stored thoughts, got %d", got) + } +} + +func TestGenerateThoughtWithoutLLM(t *testing.T) { + th := New(false, time.Minute, time.Minute) + th.generateThought("post_chat", "s1") + + select { + case thought := <-th.ThoughtChan(): + t.Errorf("no LLM means no thought, got %q", thought.Content) + case <-time.After(50 * time.Millisecond): + } +} + +func TestTriggerPostChat(t *testing.T) { + th := New(false, time.Minute, time.Minute) + th.TriggerPostChat("s1") + th.UpdateActivity() + + th.mu.Lock() + if th.activeSessionID != "s1" { + t.Errorf("activeSessionID = %q, want s1", th.activeSessionID) + } + th.mu.Unlock() +} diff --git a/backend/services/ai-core/orchestrator/orchestrator.go b/backend/services/ai-core/orchestrator/orchestrator.go index bfcbbdd..869d0b1 100644 --- a/backend/services/ai-core/orchestrator/orchestrator.go +++ b/backend/services/ai-core/orchestrator/orchestrator.go @@ -28,6 +28,13 @@ type ToolExecutor interface { Execute(ctx context.Context, name string, params map[string]any) (string, error) } +// Tracer records full-chain trace spans for each conversation round. +type Tracer interface { + StartTrace(id, userMsg string) + AddSpan(traceID, name, label, status string, durationMs int64, detail string) + FinishTrace(id, status, errMsg string) +} + // maxToolRounds caps consecutive tool-call rounds to avoid runaway loops. const maxToolRounds = 3 @@ -47,6 +54,8 @@ type Pipeline struct { personaInjector *persona.Injector memoryExtractor MemoryExtractor toolExecutor ToolExecutor + tracer Tracer + postChatHook func(sessionID string) } // NewPipeline creates a new dialogue pipeline. @@ -77,49 +86,115 @@ func (p *Pipeline) SetToolExecutor(t ToolExecutor) { p.toolExecutor = t } +// SetTracer injects full-chain tracing. +func (p *Pipeline) SetTracer(t Tracer) { + p.tracer = t +} + +// SetPostChatHook registers a callback fired after each completed round +// (used by the background thinker). +func (p *Pipeline) SetPostChatHook(fn func(sessionID string)) { + p.postChatHook = fn +} + +// span records a trace span when tracing is enabled. +func (p *Pipeline) span(traceID, name, label, status string, start time.Time, detail string) { + if p.tracer == nil || traceID == "" { + return + } + p.tracer.AddSpan(traceID, name, label, status, time.Since(start).Milliseconds(), detail) +} + +// traceStream wraps a token channel to record the LLM span and finish the +// trace when the stream completes. +func (p *Pipeline) traceStream(traceID, modelName string, in <-chan model.StreamToken) <-chan model.StreamToken { + out := make(chan model.StreamToken, 64) + go func() { + defer close(out) + start := time.Now() + status, errMsg := "success", "" + for tok := range in { + out <- tok + if tok.Done { + if strings.HasPrefix(tok.Content, "[Error:") { + status, errMsg = "error", tok.Content + } + p.tracer.AddSpan(traceID, "llm_call", "LLM 生成", status, time.Since(start).Milliseconds(), modelName) + p.tracer.FinishTrace(traceID, status, errMsg) + } + } + }() + return out +} + // ProcessInput processes a user message through the full pipeline. // Returns a channel of StreamToken for the LLM response. func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userID, userMessage string) (<-chan model.StreamToken, *model.Intent, error) { + // 0. Start trace + traceID := "" + if p.tracer != nil { + traceID = fmt.Sprintf("t-%d", time.Now().UnixNano()) + p.tracer.StartTrace(traceID, userMessage) + } + // 1. Cache check if cached := p.cache.Get(userMessage); cached != nil { ch := make(chan model.StreamToken, 1) ch <- model.StreamToken{Content: cached.Content, Done: true} close(ch) + p.span(traceID, "cache_hit", "语义缓存命中", "success", time.Now(), "") + if p.tracer != nil { + p.tracer.FinishTrace(traceID, "success", "") + } intent := &model.Intent{Type: model.IntentGreeting, Confidence: 1.0, RawText: userMessage} return ch, intent, nil } // 2. Intent analysis + t0 := time.Now() intent := p.analyzeIntent(userMessage) + p.span(traceID, "intent_analysis", "意图分析", "success", t0, intent.Type) // 3. Build system prompt + t1 := time.Now() systemPrompt := p.personaInjector.BuildSystemPrompt() + p.span(traceID, "persona", "人设注入", "success", t1, "") // 4. Sub-session dispatch (parallel) + t2 := time.Now() subInput := subsession.ProviderInput{ UserMessage: userMessage, History: p.ctxStore.Get(sessionID), Intent: intent, } subResults := p.subMgr.Dispatch(ctx, subInput, 500*time.Millisecond) + p.span(traceID, "subsession_dispatch", "子会话分派", "success", t2, fmt.Sprintf("%d 个结果", len(subResults))) // 5. Build LLM context + t3 := time.Now() llmMessages := p.ctxBuilder.Build(systemPrompt, sessionID, subResults, "") llmMessages = append(llmMessages, model.LLMMessage{Role: "user", Content: userMessage}) + p.span(traceID, "context_build", "上下文拼装", "success", t3, fmt.Sprintf("%d 条消息", len(llmMessages))) // 6. LLM synthesis (streaming, with tool calling when available) adapter := p.llmSelector.Select(model.PurposeChat) var tokenCh <-chan model.StreamToken if p.toolExecutor != nil && len(p.toolExecutor.Tools()) > 0 { - tokenCh = p.streamWithTools(ctx, adapter, llmMessages) + tokenCh = p.streamWithTools(ctx, adapter, llmMessages, traceID) } else { var err error tokenCh, err = adapter.ChatStream(ctx, llmMessages) if err != nil { + if p.tracer != nil { + p.tracer.FinishTrace(traceID, "error", err.Error()) + } return nil, intent, err } } + if p.tracer != nil { + tokenCh = p.traceStream(traceID, adapter.ModelName(), tokenCh) + } // 7. Store user message p.ctxStore.AddWithUser(sessionID, userID, model.Message{Role: "user", Content: userMessage}) @@ -132,7 +207,7 @@ func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userID, userMess // calls, the tools are executed and their results are fed back for another // round (up to maxToolRounds). The final round omits tools to force a text // answer. -func (p *Pipeline) streamWithTools(ctx context.Context, adapter *llm.Adapter, messages []model.LLMMessage) <-chan model.StreamToken { +func (p *Pipeline) streamWithTools(ctx context.Context, adapter *llm.Adapter, messages []model.LLMMessage, traceID string) <-chan model.StreamToken { out := make(chan model.StreamToken, 64) go func() { defer close(out) @@ -170,10 +245,14 @@ func (p *Pipeline) streamWithTools(ctx context.Context, adapter *llm.Adapter, me // and append the results for the next round. msgs = append(msgs, model.LLMMessage{Role: "assistant", ToolCalls: calls}) for _, call := range calls { + tcStart := time.Now() result, err := p.toolExecutor.Execute(ctx, call.Name, call.Arguments) + status := "success" if err != nil { result = "Error: " + err.Error() + status = "error" } + p.span(traceID, "tool_call", call.Name, status, tcStart, truncateDetail(result, 80)) msgs = append(msgs, model.LLMMessage{ Role: "tool", ToolCallID: call.ID, @@ -196,6 +275,9 @@ func (p *Pipeline) FinishRound(sessionID, userID, assistantReply string) { p.ctxStore.AddWithUser(sessionID, userID, model.Message{Role: "assistant", Content: assistantReply}) } p.PostProcess(sessionID, userID) + if p.postChatHook != nil { + p.postChatHook(sessionID) + } } // PostProcess triggers async memory extraction after a conversation round completes. @@ -206,6 +288,15 @@ func (p *Pipeline) PostProcess(sessionID, userID string) { p.memoryExtractor.ExtractFromConversation(nil, userID, history) } +// truncateDetail shortens a string for trace span details. +func truncateDetail(s string, n int) string { + runes := []rune(s) + if len(runes) <= n { + return s + } + return string(runes[:n]) + "…" +} + // --- Intent Analysis --- // controlWords are common IoT control verbs. diff --git a/backend/services/memory/repository.go b/backend/services/memory/repository.go index 7a41d32..bcd7b7c 100644 --- a/backend/services/memory/repository.go +++ b/backend/services/memory/repository.go @@ -221,6 +221,24 @@ func (r *Repository) RecordAccess(ctx context.Context, id string) error { return err } +// ListUserIDs returns distinct user IDs that have memories. +func (r *Repository) ListUserIDs(ctx context.Context) ([]string, error) { + rows, err := r.db.QueryContext(ctx, `SELECT DISTINCT user_id FROM memories`) + if err != nil { + return nil, fmt.Errorf("list user ids: %w", err) + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + // PurgeDecayed removes memories below the decay threshold. func (r *Repository) PurgeDecayed(ctx context.Context, userID string) (int64, error) { result, err := r.db.ExecContext(ctx, diff --git a/backend/services/memory/service.go b/backend/services/memory/service.go index f704d62..bd0f374 100644 --- a/backend/services/memory/service.go +++ b/backend/services/memory/service.go @@ -161,6 +161,23 @@ func (s *Service) ApplyDecay(ctx context.Context, userID string) (int, error) { return updated, nil } +// ApplyDecayAll applies decay for every user that has memories. +func (s *Service) ApplyDecayAll(ctx context.Context) (int, error) { + users, err := s.repo.ListUserIDs(ctx) + if err != nil { + return 0, err + } + total := 0 + for _, uid := range users { + n, err := s.ApplyDecay(ctx, uid) + if err != nil { + return total, err + } + total += n + } + return total, nil +} + // --- Keyword extraction (sliding window n-grams) --- // extractNGrams extracts all n-gram substrings from text. diff --git a/frontend/web/src/hooks/useWebSocket.ts b/frontend/web/src/hooks/useWebSocket.ts index eeaef54..d948de0 100644 --- a/frontend/web/src/hooks/useWebSocket.ts +++ b/frontend/web/src/hooks/useWebSocket.ts @@ -44,6 +44,14 @@ export function useWebSocket() { case 'device_update': if (msg.devices) setIotDevices(msg.devices) break + case 'thinking': + if (msg.content) { + addMessage({ + id: crypto.randomUUID(), session_id: currentSessionId || '', + role: 'system', content: `💭 ${msg.content}`, created_at: new Date().toISOString(), + }) + } + break case 'error': addMessage({ id: crypto.randomUUID(), session_id: '', role: 'system', content: `Error: ${msg.error}`, created_at: new Date().toISOString() })