feat(ai-core): trace instrumentation, LLM thinker, decay scheduler
Cyrene CI / backend (push) Has been cancelled
Cyrene CI / frontend (push) Has been cancelled

- 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)
This commit is contained in:
Aska committed 2026-08-20 20:57:27 +08:00
1 parent 8495a4b055
commit 8603839711
8 files changed
+338 -10

No files matched your search

+39
View File
@@ -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()
+6
View File
@@ -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,
+82 -8
View File
@@ -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()
@@ -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()
}
@@ -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.
+18
View File
@@ -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,
+17
View File
@@ -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.
+8
View File
@@ -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() })