diff --git a/backend/services/ai-core/background/thinker.go b/backend/services/ai-core/background/thinker.go new file mode 100644 index 0000000..efb20a3 --- /dev/null +++ b/backend/services/ai-core/background/thinker.go @@ -0,0 +1,155 @@ +// Package background provides the autonomous thinking engine for Cyrene. +// It generates thoughts after conversations and during idle periods. +package background + +import ( + "context" + "log" + "sync" + "time" +) + +// Thinker manages autonomous background thinking. +type Thinker struct { + mu sync.Mutex + enabled bool + thinkInterval time.Duration + silenceTimeout time.Duration + lastActivity time.Time + activeSessionID string + thoughts []Thought + stopCh chan struct{} + postChatCh chan string +} + +// Thought represents an autonomous thought generated by the AI. +type Thought struct { + Content string `json:"content"` + SessionID string `json:"session_id,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// New creates a new background thinker. +func New(enabled bool, thinkInterval, silenceTimeout time.Duration) *Thinker { + return &Thinker{ + enabled: enabled, + thinkInterval: thinkInterval, + silenceTimeout: silenceTimeout, + lastActivity: time.Now(), + thoughts: make([]Thought, 0), + stopCh: make(chan struct{}), + postChatCh: make(chan string, 10), + } +} + +// Start begins the background thinking loops. +func (t *Thinker) Start(ctx context.Context) { + if !t.enabled { + log.Println("[thinker] background thinking disabled") + return + } + + go t.silenceLoop(ctx) + go t.postChatLoop(ctx) + log.Printf("[thinker] started (interval=%v, silence=%v)", t.thinkInterval, t.silenceTimeout) +} + +// Stop shuts down the thinker. +func (t *Thinker) Stop() { + close(t.stopCh) + log.Println("[thinker] stopped") +} + +// TriggerPostChat schedules a post-chat think after a conversation. +func (t *Thinker) TriggerPostChat(sessionID string) { + t.mu.Lock() + t.activeSessionID = sessionID + t.lastActivity = time.Now() + t.mu.Unlock() + + select { + case t.postChatCh <- sessionID: + default: + } +} + +// UpdateActivity updates the last activity timestamp (for silence detection). +func (t *Thinker) UpdateActivity() { + t.mu.Lock() + t.lastActivity = time.Now() + 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 +} + +func (t *Thinker) silenceLoop(ctx context.Context) { + ticker := time.NewTicker(t.thinkInterval) + defer ticker.Stop() + + for { + select { + case <-t.stopCh: + return + case <-ticker.C: + t.mu.Lock() + elapsed := time.Since(t.lastActivity) + t.mu.Unlock() + + if elapsed >= t.silenceTimeout { + t.generateThought("silence", t.activeSessionID) + } + case <-ctx.Done(): + return + } + } +} + +func (t *Thinker) postChatLoop(ctx context.Context) { + for { + select { + case <-t.stopCh: + return + case sessionID := <-t.postChatCh: + // Delay post-chat thinking to avoid immediate noise + time.Sleep(5 * time.Second) + t.generateThought("post_chat", sessionID) + case <-ctx.Done(): + return + } + } +} + +func (t *Thinker) generateThought(trigger string, sessionID string) { + // Placeholder: in production this calls the LLM + thought := Thought{ + Content: "[思考] 当前无主动想法", + SessionID: sessionID, + CreatedAt: time.Now(), + } + + t.mu.Lock() + t.thoughts = append(t.thoughts, thought) + if len(t.thoughts) > 100 { + t.thoughts = t.thoughts[1:] + } + t.mu.Unlock() + + log.Printf("[thinker] %s thought generated for session %s", trigger, sessionID) +} + +// RecentThoughts returns the most recent thoughts. +func (t *Thinker) RecentThoughts(n int) []Thought { + t.mu.Lock() + defer t.mu.Unlock() + + if n > len(t.thoughts) { + n = len(t.thoughts) + } + return t.thoughts[len(t.thoughts)-n:] +} diff --git a/backend/services/ai-core/context/context.go b/backend/services/ai-core/context/context.go new file mode 100644 index 0000000..3ed355d --- /dev/null +++ b/backend/services/ai-core/context/context.go @@ -0,0 +1,178 @@ +// Package context manages conversation context with sliding window and token budgeting. +package context + +import ( + "fmt" + "strings" + "sync" + "unicode/utf8" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +// Store holds conversation history with a sliding window. +type Store struct { + mu sync.RWMutex + sessions map[string]*SessionHistory + maxTokens int // Maximum tokens to include in context + maxTurns int // Maximum message turns to keep +} + +// SessionHistory holds messages for a single session. +type SessionHistory struct { + Messages []model.Message + mu sync.RWMutex +} + +// NewStore creates a new conversation store. +func NewStore(maxTokens, maxTurns int) *Store { + if maxTokens <= 0 { + maxTokens = 4000 + } + if maxTurns <= 0 { + maxTurns = 50 + } + return &Store{ + sessions: make(map[string]*SessionHistory), + maxTokens: maxTokens, + maxTurns: maxTurns, + } +} + +// Add appends a message to the session history. +func (s *Store) Add(sessionID string, msg model.Message) { + s.mu.Lock() + sh, ok := s.sessions[sessionID] + if !ok { + sh = &SessionHistory{} + s.sessions[sessionID] = sh + } + s.mu.Unlock() + + sh.mu.Lock() + sh.Messages = append(sh.Messages, msg) + // Trim to max turns + if len(sh.Messages) > s.maxTurns { + sh.Messages = sh.Messages[len(sh.Messages)-s.maxTurns:] + } + sh.mu.Unlock() +} + +// Get retrieves recent messages for a session, trimmed to fit token budget. +func (s *Store) Get(sessionID string) []model.Message { + s.mu.RLock() + sh, ok := s.sessions[sessionID] + s.mu.RUnlock() + if !ok { + return nil + } + + sh.mu.RLock() + defer sh.mu.RUnlock() + + // Return all if within budget + tokens := countTokens(sh.Messages) + if tokens <= s.maxTokens { + // Return a copy + result := make([]model.Message, len(sh.Messages)) + copy(result, sh.Messages) + return result + } + + // Slide window from the end, keeping most recent messages + // Always keep at least the last 4 messages (2 turns) + minKeep := 4 + if len(sh.Messages) < minKeep { + minKeep = len(sh.Messages) + } + + result := make([]model.Message, 0, minKeep) + tokens = 0 + for i := len(sh.Messages) - 1; i >= 0; i-- { + msgTokens := tokenEstimate(sh.Messages[i].Content) + if tokens+msgTokens > s.maxTokens && len(result) >= minKeep { + break + } + result = append([]model.Message{sh.Messages[i]}, result...) + tokens += msgTokens + } + return result +} + +// Clear removes all messages for a session. +func (s *Store) Clear(sessionID string) { + s.mu.Lock() + delete(s.sessions, sessionID) + s.mu.Unlock() +} + +// Builder assembles the full context for an LLM call. +type Builder struct { + store *Store +} + +// NewBuilder creates a context builder. +func NewBuilder(store *Store) *Builder { + return &Builder{store: store} +} + +// Build creates the full message array for an LLM request. +// systemPrompt is injected as the first system message. +// session history, subsession results, and IoT context are merged. +func (b *Builder) Build(systemPrompt string, sessionID string, subsessionResults []model.SubSessionResult, iotContext string) []model.LLMMessage { + messages := make([]model.LLMMessage, 0) + + // 1. System prompt + if systemPrompt != "" { + messages = append(messages, model.LLMMessage{Role: "system", Content: systemPrompt}) + } + + // 2. Subsession results (injected as system context) + for _, r := range subsessionResults { + if r.Error != "" { + continue + } + if r.Content != "" { + msg := model.LLMMessage{ + Role: "system", + Content: fmt.Sprintf("[%s context]\n%s", r.Provider, r.Content), + } + messages = append(messages, msg) + } + } + + // 3. IoT device context + if iotContext != "" { + messages = append(messages, model.LLMMessage{ + Role: "system", + Content: fmt.Sprintf("[IoT devices]\n%s", iotContext), + }) + } + + // 4. Conversation history + history := b.store.Get(sessionID) + for _, m := range history { + messages = append(messages, model.LLMMessage{Role: m.Role, Content: m.Content}) + } + + return messages +} + +// --- Helpers --- + +// tokenEstimate estimates the number of tokens in a string (rough: 1 token ≈ 2 Chinese chars ≈ 0.75 English words). +func tokenEstimate(text string) int { + words := strings.Fields(text) + chars := utf8.RuneCountInString(text) + // Approximate: Chinese-heavy text ~ chars/2, English-heavy ~ words*1.3 + return (chars/2 + int(float64(len(words))*1.3)) / 2 +} + +// countTokens estimates total tokens in a message slice. +func countTokens(msgs []model.Message) int { + total := 0 + for _, m := range msgs { + total += tokenEstimate(m.Content) + } + return total +} diff --git a/backend/services/ai-core/llm/llm.go b/backend/services/ai-core/llm/llm.go new file mode 100644 index 0000000..d502aa4 --- /dev/null +++ b/backend/services/ai-core/llm/llm.go @@ -0,0 +1,297 @@ +// Package llm provides LLM abstraction with streaming, model routing, and response caching. +package llm + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +// Config holds LLM connection configuration. +type Config struct { + BaseURL string + APIKey string + Model string + FallbackModel string + MaxRetries int + Timeout time.Duration +} + +// DefaultConfig returns a sensible default Config. +func DefaultConfig() Config { + return Config{ + BaseURL: "https://api.openai.com/v1", + Model: "gpt-4o", + FallbackModel: "gpt-4o-mini", + MaxRetries: 3, + Timeout: 180 * time.Second, + } +} + +// Adapter handles communication with an OpenAI-compatible LLM API. +type Adapter struct { + cfg Config + client *http.Client +} + +// NewAdapter creates a new LLM adapter. +func NewAdapter(cfg Config) *Adapter { + return &Adapter{ + cfg: cfg, + client: &http.Client{ + Timeout: cfg.Timeout, + Transport: &http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 20, + IdleConnTimeout: 90 * time.Second, + }, + }, + } +} + +// ModelName returns the configured model name. +func (a *Adapter) ModelName() string { return a.cfg.Model } + +// Chat sends a chat completion request and returns the full response. +func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) { + body := map[string]any{ + "model": a.cfg.Model, + "messages": messages, + } + + resp := &model.LLMResponse{Model: a.cfg.Model} + err := a.doRequest(ctx, "POST", "/chat/completions", body, func(data map[string]any) error { + if choices, ok := data["choices"].([]any); ok && len(choices) > 0 { + choice := choices[0].(map[string]any) + if msg, ok := choice["message"].(map[string]any); ok { + if content, ok := msg["content"].(string); ok { + resp.Content = content + } + } + if reason, ok := choice["finish_reason"].(string); ok { + resp.FinishReason = reason + } + } + if usage, ok := data["usage"].(map[string]any); ok { + if total, ok := usage["total_tokens"].(float64); ok { + resp.TokensUsed = int(total) + } + } + return nil + }) + + return resp, err +} + +// ChatStream sends a chat completion request and returns a token channel for streaming. +func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (<-chan model.StreamToken, error) { + body := map[string]any{ + "model": a.cfg.Model, + "messages": messages, + "stream": true, + } + + ch := make(chan model.StreamToken, 64) + + go func() { + defer close(ch) + + err := a.doStreamRequest(ctx, "POST", "/chat/completions", body, ch) + if err != nil { + select { + case ch <- model.StreamToken{Content: fmt.Sprintf("[LLM error: %v]", err), Done: true}: + default: + } + } + }() + + return ch, nil +} + +// 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 { + jsonBody, _ := json.Marshal(body) + + req, err := http.NewRequestWithContext(ctx, method, a.cfg.BaseURL+path, bytes.NewReader(jsonBody)) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+a.cfg.APIKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := a.client.Do(req) + if err != nil { + return fmt.Errorf("http request: %w", err) + } + defer resp.Body.Close() + + respBytes, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read response: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("API error %d: %s", resp.StatusCode, string(respBytes[:min(len(respBytes), 200)])) + } + + var data map[string]any + if err := json.Unmarshal(respBytes, &data); err != nil { + return fmt.Errorf("parse response: %w", err) + } + return parser(data) +} + +// doStreamRequest sends a streaming request and sends tokens to the channel. +func (a *Adapter) doStreamRequest(ctx context.Context, method, path string, body map[string]any, ch chan<- model.StreamToken) error { + jsonBody, _ := json.Marshal(body) + + req, err := http.NewRequestWithContext(ctx, method, a.cfg.BaseURL+path, bytes.NewReader(jsonBody)) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+a.cfg.APIKey) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + + resp, err := a.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + errMsg, _ := io.ReadAll(resp.Body) + return fmt.Errorf("stream API error %d: %s", resp.StatusCode, string(errMsg[:min(len(errMsg), 200)])) + } + + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || !strings.HasPrefix(line, "data: ") { + continue + } + data := strings.TrimPrefix(line, "data: ") + if data == "[DONE]" { + ch <- model.StreamToken{Done: true} + return nil + } + + var frame map[string]any + if err := json.Unmarshal([]byte(data), &frame); err != nil { + continue + } + + if choices, ok := frame["choices"].([]any); ok && len(choices) > 0 { + choice := choices[0].(map[string]any) + if delta, ok := choice["delta"].(map[string]any); ok { + if content, ok := delta["content"].(string); ok { + ch <- model.StreamToken{Content: content} + } + } + if reason, ok := choice["finish_reason"].(string); ok && reason != "" { + ch <- model.StreamToken{FinishReason: reason, Done: true} + return nil + } + } + } + return scanner.Err() +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} + +// ModelSelector routes requests to different models based on purpose. +type ModelSelector struct { + adapters map[model.LLMPurpose]*Adapter + default_ *Adapter +} + +// NewModelSelector creates a selector with purpose-specific adapters. +func NewModelSelector(cfg Config) *ModelSelector { + base := NewAdapter(cfg) + return &ModelSelector{ + adapters: map[model.LLMPurpose]*Adapter{}, + default_: base, + } +} + +// SetAdapter assigns an adapter for a specific purpose. +func (s *ModelSelector) SetAdapter(purpose model.LLMPurpose, a *Adapter) { + s.adapters[purpose] = a +} + +// Select returns the adapter for the given purpose, falling back to default. +func (s *ModelSelector) Select(purpose model.LLMPurpose) *Adapter { + if a, ok := s.adapters[purpose]; ok { + return a + } + return s.default_ +} + +// DefaultAdapter returns the default adapter. +func (s *ModelSelector) DefaultAdapter() *Adapter { + return s.default_ +} + +// SemanticCache caches common greetings to avoid redundant LLM calls. +type SemanticCache struct { + entries map[string]CachedResponse + mu sync.RWMutex +} + +// CachedResponse holds a pre-computed response for a specific input. +type CachedResponse struct { + Content string `json:"content"` + Segments []model.ReviewSegment `json:"segments"` + ExpireAt time.Time `json:"expire_at"` +} + +// NewSemanticCache creates a cache preloaded with common greetings. +func NewSemanticCache() *SemanticCache { + c := &SemanticCache{entries: make(map[string]CachedResponse)} + ttl := 24 * time.Hour + + c.Set("你好", "你好呀~ 有什么可以帮你的吗?", nil, ttl) + c.Set("早", "早上好♪ 今天有什么计划吗?", nil, ttl) + c.Set("晚安", "晚安~ 做个好梦!", nil, ttl) + c.Set("谢谢", "不客气~ 随时可以找我帮忙哦", nil, ttl) + c.Set("hello", "Hello~ How can I help you today?", nil, ttl) + + return c +} + +// Set adds a cached response. +func (c *SemanticCache) Set(input, content string, segments []model.ReviewSegment, ttl time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.entries[strings.ToLower(strings.TrimSpace(input))] = CachedResponse{ + Content: content, + Segments: segments, + ExpireAt: time.Now().Add(ttl), + } +} + +// Get retrieves a cached response, or nil if not found/expired. +func (c *SemanticCache) Get(input string) *CachedResponse { + c.mu.RLock() + defer c.mu.RUnlock() + + key := strings.ToLower(strings.TrimSpace(input)) + if entry, ok := c.entries[key]; ok && time.Now().Before(entry.ExpireAt) { + return &entry + } + return nil +} diff --git a/backend/services/ai-core/model/model.go b/backend/services/ai-core/model/model.go new file mode 100644 index 0000000..803847b --- /dev/null +++ b/backend/services/ai-core/model/model.go @@ -0,0 +1,95 @@ +// Package model defines shared data types for the AI-Core engine. +package model + +import "time" + +// Message represents a single chat message in a conversation. +type Message struct { + ID string `json:"id"` + SessionID string `json:"session_id"` + Role string `json:"role"` // "user" | "assistant" | "action" | "system" + Content string `json:"content"` + CreatedAt time.Time `json:"created_at"` +} + +// Session represents a conversation session. +type Session struct { + ID string `json:"id"` + UserID string `json:"user_id"` + Title string `json:"title"` + State string `json:"state"` // "active" | "idle" + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// SubSessionResult holds the result from a single subsession provider. +type SubSessionResult struct { + Provider string `json:"provider"` // "general" | "memory" | "iot" | "review" + Content string `json:"content"` + Error string `json:"error,omitempty"` +} + +// Intent represents the detected intent of a user message. +type Intent struct { + Type string `json:"type"` // "chat" | "iot_control" | "iot_query" | "greeting" | "memory_trigger" | "knowledge" | "task" + Confidence float64 `json:"confidence"` // 0.0 - 1.0 + Entities []string `json:"entities"` // Detected entities (device names, etc.) + RawText string `json:"raw_text"` +} + +// LLMMessage is the standardized format sent to LLM APIs. +type LLMMessage struct { + Role string `json:"role"` // "system" | "user" | "assistant" | "tool" + Content string `json:"content"` + Name string `json:"name,omitempty"` +} + +// LLMResponse represents a non-streaming LLM response. +type LLMResponse struct { + Content string `json:"content"` + FinishReason string `json:"finish_reason"` // "stop" | "length" | "tool_calls" + Model string `json:"model"` + TokensUsed int `json:"tokens_used"` +} + +// StreamToken is a single token from a streaming LLM response. +type StreamToken struct { + Content string `json:"content"` + FinishReason string `json:"finish_reason,omitempty"` + Done bool `json:"done"` +} + +// ReviewSegment represents a parsed action/chat segment from LLM output. +type ReviewSegment struct { + Type string `json:"type"` // "action" | "chat" + Content string `json:"content"` +} + +// ToolCall represents a tool call requested by the LLM. +type ToolCall struct { + ID string `json:"id"` + Name string `json:"name"` + Arguments map[string]any `json:"arguments"` +} + +// Intent types +const ( + IntentChat = "chat" + IntentIoTControl = "iot_control" + IntentIoTQuery = "iot_query" + IntentGreeting = "greeting" + IntentMemoryTrigger = "memory_trigger" + IntentKnowledge = "knowledge" + IntentTask = "task" +) + +// LLM purposes for model routing +type LLMPurpose string + +const ( + PurposeChat LLMPurpose = "chat" + PurposeIntentAnalysis LLMPurpose = "intent_analysis" + PurposeDeepThinking LLMPurpose = "deep_thinking" + PurposeMemoryExtraction LLMPurpose = "memory_extraction" + PurposeToolCalling LLMPurpose = "tool_calling" +) diff --git a/backend/services/ai-core/orchestrator/orchestrator.go b/backend/services/ai-core/orchestrator/orchestrator.go new file mode 100644 index 0000000..7aad318 --- /dev/null +++ b/backend/services/ai-core/orchestrator/orchestrator.go @@ -0,0 +1,230 @@ +// Package orchestrator implements the main dialogue pipeline for Cyrene. +// It coordinates intent analysis, subsession dispatch, LLM synthesis, and review parsing. +package orchestrator + +import ( + "context" + "fmt" + "strings" + "time" + + ctxpkg "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/context" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/persona" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession" +) + +// Pipeline orchestrates the full dialogue flow. +type Pipeline struct { + llmSelector *llm.ModelSelector + cache *llm.SemanticCache + ctxStore *ctxpkg.Store + ctxBuilder *ctxpkg.Builder + subMgr *subsession.Manager + personaInjector *persona.Injector +} + +// NewPipeline creates a new dialogue pipeline. +func NewPipeline( + llmSel *llm.ModelSelector, + cache *llm.SemanticCache, + ctxStore *ctxpkg.Store, + subMgr *subsession.Manager, + personaInjector *persona.Injector, +) *Pipeline { + return &Pipeline{ + llmSelector: llmSel, + cache: cache, + ctxStore: ctxStore, + ctxBuilder: ctxpkg.NewBuilder(ctxStore), + subMgr: subMgr, + personaInjector: personaInjector, + } +} + +// 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, userMessage string) (<-chan model.StreamToken, *model.Intent, error) { + // 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) + + intent := &model.Intent{Type: model.IntentGreeting, Confidence: 1.0, RawText: userMessage} + return ch, intent, nil + } + + // 2. Intent analysis + intent := p.analyzeIntent(userMessage) + + // 3. Build system prompt + systemPrompt := p.personaInjector.BuildSystemPrompt() + + // 4. Sub-session dispatch (parallel) + subInput := subsession.ProviderInput{ + UserMessage: userMessage, + History: p.ctxStore.Get(sessionID), + Intent: intent, + } + subResults := p.subMgr.Dispatch(ctx, subInput, 500*time.Millisecond) + + // 5. Build LLM context + llmMessages := p.ctxBuilder.Build(systemPrompt, sessionID, subResults, "") + llmMessages = append(llmMessages, model.LLMMessage{Role: "user", Content: userMessage}) + + // 6. LLM synthesis (streaming) + adapter := p.llmSelector.Select(model.PurposeChat) + tokenCh, err := adapter.ChatStream(ctx, llmMessages) + if err != nil { + return nil, intent, err + } + + // 7. Store user message + p.ctxStore.Add(sessionID, model.Message{Role: "user", Content: userMessage}) + + return tokenCh, intent, nil +} + +// --- Intent Analysis --- + +// controlWords are common IoT control verbs. +var controlWords = []string{"打开", "关闭", "关掉", "关上", "调到", "设置", "开关", "调节", "调高", "调低"} + +// deviceWords are common IoT device names. +var deviceWords = []string{"灯", "空调", "窗帘", "电视", "风扇", "加湿器", "插座", "门锁", "传感器"} + +// greetingWords are common greetings. +var greetingWords = []string{"你好", "早", "晚安", "嗨", "hello", "hi", "在吗", "在不在"} + +func (p *Pipeline) analyzeIntent(message string) *model.Intent { + msg := strings.ToLower(message) + + // Fast path: IoT command + hasControl := false + for _, w := range controlWords { + if strings.Contains(msg, w) { + hasControl = true + break + } + } + hasDevice := false + var entities []string + for _, w := range deviceWords { + if strings.Contains(msg, w) { + hasDevice = true + entities = append(entities, w) + } + } + if hasControl && hasDevice { + // Check for "开" ambiguity: "开心", "开始", "离开" should NOT trigger IoT + if strings.Contains(msg, "开") { + ambiguous := false + for _, prefix := range []string{"开心", "开始", "离开", "展开", "打开"} { + if prefix == "打开" { + continue + } + if strings.Contains(msg, prefix) { + ambiguous = true + break + } + } + if !ambiguous { + return &model.Intent{Type: model.IntentIoTControl, Confidence: 0.95, Entities: entities, RawText: message} + } + } + return &model.Intent{Type: model.IntentIoTControl, Confidence: 0.9, Entities: entities, RawText: message} + } + if hasDevice { + return &model.Intent{Type: model.IntentIoTQuery, Confidence: 0.85, Entities: entities, RawText: message} + } + + // Fast path: Greeting + for _, w := range greetingWords { + if strings.Contains(msg, w) || msg == w { + return &model.Intent{Type: model.IntentGreeting, Confidence: 0.95, RawText: message} + } + } + + // Fast path: Memory trigger + memoryKeywords := []string{"记得", "还记得", "以前", "上次", "之前", "过去"} + for _, w := range memoryKeywords { + if strings.Contains(msg, w) { + return &model.Intent{Type: model.IntentMemoryTrigger, Confidence: 0.7, RawText: message} + } + } + + // Default: chat + return &model.Intent{Type: model.IntentChat, Confidence: 0.6, RawText: message} +} + +// --- Helpers --- + +// BuildStreamResponse accumulates streaming tokens and stores the final response. +func BuildStreamResponse(ctx context.Context, tokenCh <-chan model.StreamToken, store *ctxpkg.Store, sessionID string) (string, error) { + var fullContent strings.Builder + + for { + select { + case <-ctx.Done(): + return fullContent.String(), ctx.Err() + case token, ok := <-tokenCh: + if !ok { + return fullContent.String(), nil + } + if token.Done { + // Store the complete response + content := fullContent.String() + store.Add(sessionID, model.Message{Role: "assistant", Content: content}) + return content, nil + } + fullContent.WriteString(token.Content) + } + } +} + +// IsStrongIoTCommand checks if the message is unambiguously an IoT command. +func IsStrongIoTCommand(message string) bool { + hasControl := false + for _, cw := range controlWords { + if strings.Contains(message, cw) { + hasControl = true + break + } + } + hasDevice := false + for _, dw := range deviceWords { + if strings.Contains(message, dw) { + hasDevice = true + break + } + } + return hasControl && hasDevice +} + +// BuildIoTContext creates a device context string for the LLM prompt. +func BuildIoTContext(devices []DeviceState) string { + if len(devices) == 0 { + return "" + } + var sb strings.Builder + sb.WriteString("当前智能家居设备状态:\n") + for _, d := range devices { + state := "关" + if power, ok := d.State["power"]; ok { + if p, ok := power.(bool); ok && p { + state = "开" + } + } + sb.WriteString(fmt.Sprintf("- %s: %s\n", d.Name, state)) + } + return sb.String() +} + +// DeviceState holds a simplified device state for context building. +type DeviceState struct { + ID string + Name string + State map[string]any +} diff --git a/backend/services/ai-core/orchestrator/orchestrator_test.go b/backend/services/ai-core/orchestrator/orchestrator_test.go new file mode 100644 index 0000000..16fbe67 --- /dev/null +++ b/backend/services/ai-core/orchestrator/orchestrator_test.go @@ -0,0 +1,147 @@ +package orchestrator + +import ( + "testing" + + ctxpkg "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/context" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/persona" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession" +) + +func setupTestPipeline() *Pipeline { + return NewPipeline( + llm.NewModelSelector(llm.DefaultConfig()), + llm.NewSemanticCache(), + ctxpkg.NewStore(4000, 50), + subsession.NewManager(&subsession.GeneralProvider{}, &subsession.MemoryProvider{}), + persona.NewInjector(persona.BuiltinDefault()), + ) +} + +func TestIntentAnalysis_IoT(t *testing.T) { + p := setupTestPipeline() + + tests := []struct { + input string + wantType string + }{ + {"打开客厅灯", model.IntentIoTControl}, + {"关掉卧室灯", model.IntentIoTControl}, + {"把空调调到26度", model.IntentIoTControl}, + {"关闭窗帘", model.IntentIoTControl}, + {"灯开了吗", model.IntentIoTQuery}, + {"传感器温度多少", model.IntentIoTQuery}, + } + + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + intent := p.analyzeIntent(tt.input) + if intent.Type != tt.wantType { + t.Errorf("analyzeIntent(%q).Type = %q, want %q", tt.input, intent.Type, tt.wantType) + } + }) + } +} + +func TestIntentAnalysis_Greeting(t *testing.T) { + p := setupTestPipeline() + + tests := []string{"你好", "早", "晚安", "在吗"} + + for _, input := range tests { + t.Run(input, func(t *testing.T) { + intent := p.analyzeIntent(input) + if intent.Type != model.IntentGreeting { + t.Errorf("analyzeIntent(%q).Type = %q, want %s", input, intent.Type, model.IntentGreeting) + } + }) + } +} + +func TestIntentAnalysis_Chat(t *testing.T) { + p := setupTestPipeline() + + intent := p.analyzeIntent("今天天气怎么样") + if intent.Type != model.IntentChat { + t.Errorf("expected chat intent, got %s", intent.Type) + } +} + +func TestIntentAnalysis_MemoryTrigger(t *testing.T) { + p := setupTestPipeline() + + intent := p.analyzeIntent("你还记得我喜欢什么吗") + if intent.Type != model.IntentMemoryTrigger { + t.Errorf("expected memory_trigger, got %s", intent.Type) + } +} + +func TestIntentAnalysis_AmbiguousOpen(t *testing.T) { + p := setupTestPipeline() + + // "开心" and "开始" should NOT trigger IoT intent + intent := p.analyzeIntent("我今天很开心") + if intent.Type == model.IntentIoTControl { + t.Errorf("'开心' should not trigger IoT, got %s", intent.Type) + } + + intent = p.analyzeIntent("开始工作") + if intent.Type == model.IntentIoTControl { + t.Errorf("'开始' should not trigger IoT, got %s", intent.Type) + } +} + +func TestIsStrongIoTCommand(t *testing.T) { + if !IsStrongIoTCommand("打开客厅灯") { + t.Error("expected true for '打开客厅灯'") + } + if !IsStrongIoTCommand("关闭空调") { + t.Error("expected true for '关闭空调'") + } + if IsStrongIoTCommand("今天天气真好") { + t.Error("expected false for non-IoT message") + } +} + +func TestBuildIoTContext(t *testing.T) { + devices := []DeviceState{ + {Name: "客厅灯", State: map[string]any{"power": true}}, + {Name: "卧室灯", State: map[string]any{"power": false}}, + } + + ctx := BuildIoTContext(devices) + if ctx == "" { + t.Error("expected non-empty context") + } + if len(ctx) < 10 { + t.Errorf("context too short: %s", ctx) + } +} + +func TestSemanticCache(t *testing.T) { + p := setupTestPipeline() + + cached := p.cache.Get("你好") + if cached == nil { + t.Error("'你好' should be cached") + } + if cached != nil && cached.Content == "" { + t.Error("cached content should not be empty") + } + + notCached := p.cache.Get("随机字符串不会命中缓存") + if notCached != nil { + t.Error("random string should not be cached") + } +} + +func TestModelConstants(t *testing.T) { + if model.IntentChat != "chat" { + t.Errorf("IntentChat = %s", model.IntentChat) + } + if model.IntentIoTControl != "iot_control" { + t.Errorf("IntentIoTControl = %s", model.IntentIoTControl) + } +} diff --git a/backend/services/ai-core/persona/persona.go b/backend/services/ai-core/persona/persona.go new file mode 100644 index 0000000..2ae7255 --- /dev/null +++ b/backend/services/ai-core/persona/persona.go @@ -0,0 +1,185 @@ +// Package persona manages AI character loading and system prompt injection. +package persona + +import ( + "fmt" + "os" + "path/filepath" + "strings" +) + +// Persona defines an AI character's configuration. +type Persona struct { + Name string `yaml:"name" json:"name"` + Description string `yaml:"description" json:"description"` + Greeting string `yaml:"greeting" json:"greeting"` + Traits []string `yaml:"traits" json:"traits"` + Style string `yaml:"style" json:"style"` + Constraints []string `yaml:"constraints" json:"constraints"` + + // IoT-specific personality + IoTResponseStyle string `yaml:"iot_response_style" json:"iot_response_style,omitempty"` +} + +// Loader loads persona configurations from a directory. +type Loader struct { + personas map[string]*Persona +} + +// NewLoader loads all YAML persona files from the given directory. +func NewLoader(dir string) (*Loader, error) { + l := &Loader{personas: make(map[string]*Persona)} + + entries, err := os.ReadDir(dir) + if err != nil { + // Return empty loader if directory doesn't exist + if os.IsNotExist(err) { + return l, nil + } + return nil, fmt.Errorf("read persona dir %s: %w", dir, err) + } + + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".yaml") && !strings.HasSuffix(entry.Name(), ".yml") { + continue + } + path := filepath.Join(dir, entry.Name()) + p, err := loadFile(path) + if err != nil { + return nil, fmt.Errorf("load persona %s: %w", entry.Name(), err) + } + name := strings.TrimSuffix(entry.Name(), filepath.Ext(entry.Name())) + p.Name = name + l.personas[name] = p + } + + return l, nil +} + +// Get returns a persona by name, or nil if not found. +func (l *Loader) Get(name string) *Persona { + return l.personas[name] +} + +// List returns all loaded persona names. +func (l *Loader) List() []string { + names := make([]string, 0, len(l.personas)) + for name := range l.personas { + names = append(names, name) + } + return names +} + +// Count returns the number of loaded personas. +func (l *Loader) Count() int { + return len(l.personas) +} + +// Default returns the first persona, or a built-in default. +func (l *Loader) Default() *Persona { + for _, p := range l.personas { + return p + } + return BuiltinDefault() +} + +// loadFile reads a single persona YAML file. +func loadFile(path string) (*Persona, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + return parseYAML(data) +} + +// parseYAML parses persona YAML content (simple key-value parser). +func parseYAML(data []byte) (*Persona, error) { + p := &Persona{} + lines := strings.Split(string(data), "\n") + var currentList *[]string + + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + + if strings.HasPrefix(line, "- ") && currentList != nil { + *currentList = append(*currentList, strings.TrimPrefix(line, "- ")) + continue + } + currentList = nil + + parts := strings.SplitN(line, ":", 2) + if len(parts) != 2 { + continue + } + key := strings.TrimSpace(parts[0]) + val := strings.TrimSpace(parts[1]) + + switch key { + case "name": + p.Name = val + case "description": + p.Description = val + case "greeting": + p.Greeting = val + case "style": + p.Style = val + case "iot_response_style": + p.IoTResponseStyle = val + case "traits": + currentList = &p.Traits + case "constraints": + currentList = &p.Constraints + } + } + return p, nil +} + +// BuiltinDefault returns a hardcoded default persona. +func BuiltinDefault() *Persona { + return &Persona{ + Name: "default", + Description: "一个友好的 AI 助手", + Greeting: "你好呀~ 有什么可以帮你的吗?", + Traits: []string{"友善", "细心", "幽默"}, + Style: "用轻松自然的中文交流,偶尔加一点俏皮表情", + Constraints: []string{"不编造事实", "不提供医疗法律建议", "保持礼貌"}, + } +} + +// Injector builds the system prompt from a persona. +type Injector struct { + persona *Persona +} + +// NewInjector creates a system prompt injector for a persona. +func NewInjector(p *Persona) *Injector { + if p == nil { + p = BuiltinDefault() + } + return &Injector{persona: p} +} + +// BuildSystemPrompt constructs the full system prompt string. +func (inj *Injector) BuildSystemPrompt() string { + var sb strings.Builder + + sb.WriteString(fmt.Sprintf("你是%s,%s。\n", inj.persona.Name, inj.persona.Description)) + + if len(inj.persona.Traits) > 0 { + sb.WriteString(fmt.Sprintf("性格特点: %s\n", strings.Join(inj.persona.Traits, "、"))) + } + if inj.persona.Style != "" { + sb.WriteString(fmt.Sprintf("说话风格: %s\n", inj.persona.Style)) + } + if len(inj.persona.Constraints) > 0 { + sb.WriteString("遵守规则:\n") + for _, c := range inj.persona.Constraints { + sb.WriteString(fmt.Sprintf("- %s\n", c)) + } + } + + return sb.String() +} diff --git a/backend/services/ai-core/subsession/subsession.go b/backend/services/ai-core/subsession/subsession.go new file mode 100644 index 0000000..526f614 --- /dev/null +++ b/backend/services/ai-core/subsession/subsession.go @@ -0,0 +1,163 @@ +// Package subsession manages parallel sub-session providers for dialogue enrichment. +package subsession + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +// Provider executes a specific sub-session task in parallel with others. +type Provider interface { + Name() string + Execute(ctx context.Context, input ProviderInput) (*model.SubSessionResult, error) +} + +// ProviderInput holds the context for sub-session execution. +type ProviderInput struct { + UserMessage string + History []model.Message + Intent *model.Intent +} + +// Manager orchestrates parallel sub-session execution. +type Manager struct { + providers []Provider +} + +// NewManager creates a sub-session manager. +func NewManager(providers ...Provider) *Manager { + return &Manager{providers: providers} +} + +// Dispatch runs all providers in parallel and returns results within the timeout. +func (m *Manager) Dispatch(ctx context.Context, input ProviderInput, timeout time.Duration) []model.SubSessionResult { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + var wg sync.WaitGroup + results := make([]model.SubSessionResult, 0, len(m.providers)) + var mu sync.Mutex + + for _, p := range m.providers { + wg.Add(1) + go func(provider Provider) { + defer wg.Done() + result, err := provider.Execute(ctx, input) + mu.Lock() + defer mu.Unlock() + if err != nil { + results = append(results, model.SubSessionResult{ + Provider: provider.Name(), + Error: err.Error(), + }) + } else if result != nil { + results = append(results, *result) + } + }(p) + } + + wg.Wait() + return results +} + +// --- Built-in Providers --- + +// GeneralProvider handles general conversation intent. +type GeneralProvider struct{} + +func (p *GeneralProvider) Name() string { return "general" } +func (p *GeneralProvider) Execute(_ context.Context, input ProviderInput) (*model.SubSessionResult, error) { + return &model.SubSessionResult{ + Provider: "general", + Content: fmt.Sprintf("Intent: %s (confidence: %.2f)", input.Intent.Type, input.Intent.Confidence), + }, nil +} + +// MemoryProvider retrieves relevant memories (placeholder — wired to Memory Service later). +type MemoryProvider struct{} + +func (p *MemoryProvider) Name() string { return "memory" } +func (p *MemoryProvider) Execute(_ context.Context, _ ProviderInput) (*model.SubSessionResult, error) { + return &model.SubSessionResult{Provider: "memory", Content: "no relevant memories"}, nil +} + +// IoTProvider executes IoT device queries and commands. +type IoTProvider struct { + // GetDeviceState is injected by the main app to query IoT devices. + GetDeviceState func(ctx context.Context) (string, error) + // ControlDevice is injected to execute IoT commands. + ControlDevice func(ctx context.Context, device, action string) (string, error) +} + +func (p *IoTProvider) Name() string { return "iot" } +func (p *IoTProvider) Execute(ctx context.Context, input ProviderInput) (*model.SubSessionResult, error) { + if input.Intent == nil || (input.Intent.Type != model.IntentIoTControl && input.Intent.Type != model.IntentIoTQuery) { + return &model.SubSessionResult{Provider: "iot", Content: ""}, nil + } + + if p.GetDeviceState != nil { + state, err := p.GetDeviceState(ctx) + if err == nil { + return &model.SubSessionResult{Provider: "iot", Content: state}, nil + } + } + return &model.SubSessionResult{Provider: "iot", Content: "iot devices available"}, nil +} + +// ReviewProvider parses LLM output into action/chat segments. +type ReviewProvider struct{} + +func (p *ReviewProvider) Name() string { return "review" } +func (p *ReviewProvider) Execute(_ context.Context, _ ProviderInput) (*model.SubSessionResult, error) { + return &model.SubSessionResult{Provider: "review", Content: ""}, nil +} + +// ParseReviewMessages splits a message with action markers into segments. +// Input: "(歪着头看) 客厅灯已经开啦♪" +// Output: [{type:"action", content:"歪着头看"}, {type:"chat", content:"客厅灯已经开啦♪"}] +func ParseReviewMessages(content string) []model.ReviewSegment { + var segments []model.ReviewSegment + var current strings.Builder + inAction := false + + runes := []rune(content) + for i := 0; i < len(runes); i++ { + r := runes[i] + + // Chinese bracket detection + if r == '(' || r == '(' { + if current.Len() > 0 && !inAction { + segments = append(segments, model.ReviewSegment{Type: "chat", Content: current.String()}) + current.Reset() + } + inAction = true + continue + } + + if r == ')' || r == ')' { + if inAction && current.Len() > 0 { + segments = append(segments, model.ReviewSegment{Type: "action", Content: current.String()}) + current.Reset() + } + inAction = false + continue + } + + current.WriteRune(r) + } + + if current.Len() > 0 { + segType := "chat" + if inAction { + segType = "action" + } + segments = append(segments, model.ReviewSegment{Type: segType, Content: current.String()}) + } + + return segments +} diff --git a/backend/services/ai-core/subsession/subsession_test.go b/backend/services/ai-core/subsession/subsession_test.go new file mode 100644 index 0000000..c96de61 --- /dev/null +++ b/backend/services/ai-core/subsession/subsession_test.go @@ -0,0 +1,79 @@ +package subsession + +import ( + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model" +) + +func TestParseReviewMessages(t *testing.T) { + tests := []struct { + input string + expected []model.ReviewSegment + }{ + { + "(歪着头看你) 叶酱,客厅灯早就开着啦♪", + []model.ReviewSegment{ + {Type: "action", Content: "歪着头看你"}, + {Type: "chat", Content: " 叶酱,客厅灯早就开着啦♪"}, + }, + }, + { + "你好呀~ 今天有什么需要帮忙的吗?", + []model.ReviewSegment{ + {Type: "chat", Content: "你好呀~ 今天有什么需要帮忙的吗?"}, + }, + }, + { + "(思考中)嗯...这个问题比较复杂", + []model.ReviewSegment{ + {Type: "action", Content: "思考中"}, + {Type: "chat", Content: "嗯...这个问题比较复杂"}, + }, + }, + { + "", + nil, + }, + { + "(only action)", + []model.ReviewSegment{ + {Type: "action", Content: "only action"}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := ParseReviewMessages(tt.input) + if len(got) != len(tt.expected) { + t.Fatalf("expected %d segments, got %d: %+v", len(tt.expected), len(got), got) + } + for i, seg := range got { + if seg.Type != tt.expected[i].Type { + t.Errorf("segment[%d].Type = %q, want %q", i, seg.Type, tt.expected[i].Type) + } + if seg.Content != tt.expected[i].Content { + t.Errorf("segment[%d].Content = %q, want %q", i, seg.Content, tt.expected[i].Content) + } + } + }) + } +} + +func TestProviderNames(t *testing.T) { + p := &GeneralProvider{} + if p.Name() != "general" { + t.Errorf("got %s", p.Name()) + } + + m := &MemoryProvider{} + if m.Name() != "memory" { + t.Errorf("got %s", m.Name()) + } + + i := &IoTProvider{} + if i.Name() != "iot" { + t.Errorf("got %s", i.Name()) + } +}