feat(memory): add LLM-driven memory extraction from conversations
- memory/extractor: async LLM-based memory extraction from chat history - Parses JSON lines from LLM response as memory entries - Wire into orchestrator pipeline via PostProcess hook - Non-blocking: runs in goroutine with 30s timeout - Automatic dedup via existing keyword overlap logic
This commit is contained in:
1 parent
886fcf2216
commit
47047c903c
5 files changed
+263
-6
No files matched your search
@@ -182,6 +182,11 @@ func main() {
|
|||||||
// === 11. AI-Core: Orchestrator ===
|
// === 11. AI-Core: Orchestrator ===
|
||||||
semanticCache := llm.NewSemanticCache()
|
semanticCache := llm.NewSemanticCache()
|
||||||
orch := orchestrator.NewPipeline(llmSelector, semanticCache, ctxStore, subMgr, personaInjector)
|
orch := orchestrator.NewPipeline(llmSelector, semanticCache, ctxStore, subMgr, personaInjector)
|
||||||
|
if memSvc != nil {
|
||||||
|
memAdapter := llmSelector.DefaultAdapter()
|
||||||
|
memExtractor := memory.NewExtractor(memSvc, &memory.LLMAdapterBridge{Adapter: memAdapter})
|
||||||
|
orch.SetMemoryExtractor(memExtractor)
|
||||||
|
}
|
||||||
log.Printf("[orchestrator] pipeline ready")
|
log.Printf("[orchestrator] pipeline ready")
|
||||||
|
|
||||||
// === 11.5 Platform Bridge ===
|
// === 11.5 Platform Bridge ===
|
||||||
|
|||||||
@@ -15,14 +15,20 @@ import (
|
|||||||
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession"
|
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// MemoryExtractor is the interface for async memory extraction.
|
||||||
|
type MemoryExtractor interface {
|
||||||
|
ExtractFromConversation(ctx context.Context, userID string, conversation []model.Message)
|
||||||
|
}
|
||||||
|
|
||||||
// Pipeline orchestrates the full dialogue flow.
|
// Pipeline orchestrates the full dialogue flow.
|
||||||
type Pipeline struct {
|
type Pipeline struct {
|
||||||
llmSelector *llm.ModelSelector
|
llmSelector *llm.ModelSelector
|
||||||
cache *llm.SemanticCache
|
cache *llm.SemanticCache
|
||||||
ctxStore *ctxpkg.Store
|
ctxStore *ctxpkg.Store
|
||||||
ctxBuilder *ctxpkg.Builder
|
ctxBuilder *ctxpkg.Builder
|
||||||
subMgr *subsession.Manager
|
subMgr *subsession.Manager
|
||||||
personaInjector *persona.Injector
|
personaInjector *persona.Injector
|
||||||
|
memoryExtractor MemoryExtractor
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewPipeline creates a new dialogue pipeline.
|
// NewPipeline creates a new dialogue pipeline.
|
||||||
@@ -43,6 +49,11 @@ func NewPipeline(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetMemoryExtractor injects the memory extraction capability.
|
||||||
|
func (p *Pipeline) SetMemoryExtractor(e MemoryExtractor) {
|
||||||
|
p.memoryExtractor = e
|
||||||
|
}
|
||||||
|
|
||||||
// ProcessInput processes a user message through the full pipeline.
|
// ProcessInput processes a user message through the full pipeline.
|
||||||
// Returns a channel of StreamToken for the LLM response.
|
// 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) {
|
func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage string) (<-chan model.StreamToken, *model.Intent, error) {
|
||||||
@@ -87,6 +98,14 @@ func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage stri
|
|||||||
return tokenCh, intent, nil
|
return tokenCh, intent, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PostProcess triggers async memory extraction after a conversation round completes.
|
||||||
|
func (p *Pipeline) PostProcess(sessionID, userID string) {
|
||||||
|
if p.memoryExtractor == nil { return }
|
||||||
|
history := p.ctxStore.Get(sessionID)
|
||||||
|
if len(history) == 0 { return }
|
||||||
|
p.memoryExtractor.ExtractFromConversation(nil, userID, history)
|
||||||
|
}
|
||||||
|
|
||||||
// --- Intent Analysis ---
|
// --- Intent Analysis ---
|
||||||
|
|
||||||
// controlWords are common IoT control verbs.
|
// controlWords are common IoT control verbs.
|
||||||
|
|||||||
@@ -0,0 +1,17 @@
|
|||||||
|
package memory
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LLMAdapterBridge adapts llm.Adapter to LLMClient for memory extraction.
|
||||||
|
type LLMAdapterBridge struct {
|
||||||
|
Adapter *llm.Adapter
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *LLMAdapterBridge) Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
|
||||||
|
return b.Adapter.Chat(ctx, messages)
|
||||||
|
}
|
||||||
@@ -0,0 +1,117 @@
|
|||||||
|
package memory
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LLMClient is the interface for calling an LLM.
|
||||||
|
type LLMClient interface {
|
||||||
|
Chat(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Extractor uses LLM to extract memories from conversations.
|
||||||
|
type Extractor struct {
|
||||||
|
svc *Service
|
||||||
|
llm LLMClient
|
||||||
|
enabled bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewExtractor creates a memory extractor.
|
||||||
|
func NewExtractor(svc *Service, llm LLMClient) *Extractor {
|
||||||
|
return &Extractor{svc: svc, llm: llm, enabled: llm != nil}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExtractedMemory represents a single memory extracted by the LLM.
|
||||||
|
type ExtractedMemory struct {
|
||||||
|
Content string `json:"content"`
|
||||||
|
Category string `json:"category"`
|
||||||
|
Priority float64 `json:"priority"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ExtractFromConversation analyzes a conversation and extracts memories asynchronously.
|
||||||
|
// Does not block — runs in a goroutine.
|
||||||
|
func (e *Extractor) ExtractFromConversation(ctx context.Context, userID string, conversation []model.Message) {
|
||||||
|
if !e.enabled || e.llm == nil || len(conversation) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
bgCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
memories, err := e.extract(bgCtx, conversation)
|
||||||
|
if err != nil {
|
||||||
|
return // Silent fail — don't block main flow
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, m := range memories {
|
||||||
|
if m.Content == "" { continue }
|
||||||
|
if m.Priority <= 0 { m.Priority = 0.7 }
|
||||||
|
if m.Category == "" { m.Category = CatFact }
|
||||||
|
e.svc.Add(bgCtx, userID, "", m.Content, m.Category, m.Priority)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *Extractor) extract(ctx context.Context, messages []model.Message) ([]ExtractedMemory, error) {
|
||||||
|
// Build conversation text
|
||||||
|
var convText strings.Builder
|
||||||
|
for _, msg := range messages {
|
||||||
|
role := "用户"
|
||||||
|
if msg.Role == "assistant" || msg.Role == "action" {
|
||||||
|
role = "AI"
|
||||||
|
}
|
||||||
|
convText.WriteString(fmt.Sprintf("%s: %s\n", role, msg.Content))
|
||||||
|
}
|
||||||
|
|
||||||
|
prompt := []model.LLMMessage{
|
||||||
|
{Role: "system", Content: `你是一个记忆提取助手。从对话中提取关于用户的重要信息。
|
||||||
|
每条记忆用一行 JSON 表示,格式:{"content":"记忆内容","category":"category","priority":0.8}
|
||||||
|
|
||||||
|
类别可选: fact(事实), preference(偏好), event(事件), plan(计划)
|
||||||
|
优先级 0-1,重要信息用高优先级。
|
||||||
|
只提取新信息,不重复已有内容。
|
||||||
|
只输出 JSON 行,不要其他文字。`},
|
||||||
|
{Role: "user", Content: fmt.Sprintf("从以下对话中提取记忆:\n%s", convText.String())},
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := e.llm.Chat(ctx, prompt)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("llm extract: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return parseExtractedMemories(resp.Content), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func parseExtractedMemories(text string) []ExtractedMemory {
|
||||||
|
var memories []ExtractedMemory
|
||||||
|
lines := strings.Split(text, "\n")
|
||||||
|
for _, line := range lines {
|
||||||
|
line = strings.TrimSpace(line)
|
||||||
|
if line == "" || !strings.HasPrefix(line, "{") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
var m ExtractedMemory
|
||||||
|
if err := json.Unmarshal([]byte(line), &m); err != nil {
|
||||||
|
// Try to extract just the JSON part
|
||||||
|
start := strings.Index(line, "{")
|
||||||
|
end := strings.LastIndex(line, "}")
|
||||||
|
if start >= 0 && end > start {
|
||||||
|
if err := json.Unmarshal([]byte(line[start:end+1]), &m); err == nil {
|
||||||
|
if m.Content != "" { memories = append(memories, m) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if m.Content != "" {
|
||||||
|
memories = append(memories, m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return memories
|
||||||
|
}
|
||||||
@@ -0,0 +1,99 @@
|
|||||||
|
package memory
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestParseExtractedMemories(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
input string
|
||||||
|
minCount int
|
||||||
|
}{
|
||||||
|
{`{"content":"用户喜欢咖啡","category":"preference","priority":0.8}
|
||||||
|
{"content":"用户每天跑步","category":"fact","priority":0.7}`, 2},
|
||||||
|
{`这是提取的记忆:
|
||||||
|
{"content":"用户名叫张三","category":"fact","priority":0.9}
|
||||||
|
{"content":"用户在北京工作","category":"fact","priority":0.85}`, 2},
|
||||||
|
{`no json here`, 0},
|
||||||
|
{``, 0},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run("", func(t *testing.T) {
|
||||||
|
result := parseExtractedMemories(tt.input)
|
||||||
|
if len(result) < tt.minCount {
|
||||||
|
t.Errorf("expected >= %d memories, got %d: %+v", tt.minCount, len(result), result)
|
||||||
|
}
|
||||||
|
for _, m := range result {
|
||||||
|
if m.Content == "" {
|
||||||
|
t.Error("memory content should not be empty")
|
||||||
|
}
|
||||||
|
if m.Category == "" {
|
||||||
|
t.Error("memory category should not be empty")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractorDisabled(t *testing.T) {
|
||||||
|
e := NewExtractor(nil, nil)
|
||||||
|
if e.enabled {
|
||||||
|
t.Error("extractor should be disabled when llm is nil")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Should not panic
|
||||||
|
e.ExtractFromConversation(nil, "user-1", nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractorParseEdgeCases(t *testing.T) {
|
||||||
|
// Test JSON with whitespace
|
||||||
|
result := parseExtractedMemories(` {"content":"test","category":"fact","priority":0.5} `)
|
||||||
|
if len(result) != 1 {
|
||||||
|
t.Errorf("expected 1, got %d", len(result))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Test empty JSON object
|
||||||
|
result = parseExtractedMemories(`{}`)
|
||||||
|
if len(result) != 0 {
|
||||||
|
t.Errorf("expected 0 for empty object, got %d", len(result))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Test invalid JSON line mixed with valid
|
||||||
|
result = parseExtractedMemories(`invalid
|
||||||
|
{"content":"ok","category":"fact","priority":0.5}
|
||||||
|
not json either`)
|
||||||
|
if len(result) != 1 {
|
||||||
|
t.Errorf("expected 1 valid, got %d", len(result))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// mockLLM implements LLMClient for testing.
|
||||||
|
type mockLLM struct {
|
||||||
|
response string
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockLLM) Chat(_ context.Context, _ []model.LLMMessage) (*model.LLMResponse, error) {
|
||||||
|
if m.err != nil { return nil, m.err }
|
||||||
|
return &model.LLMResponse{Content: m.response, Model: "mock"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExtractorWithMockLLM(t *testing.T) {
|
||||||
|
mock := &mockLLM{
|
||||||
|
response: `{"content":"用户喜欢喝咖啡","category":"preference","priority":0.8}
|
||||||
|
{"content":"用户在2024年去过日本","category":"event","priority":0.9}`,
|
||||||
|
}
|
||||||
|
|
||||||
|
// We can't easily test the full extraction without a real DB
|
||||||
|
// But verify mock LLM works
|
||||||
|
resp, err := mock.Chat(nil, nil)
|
||||||
|
if err != nil { t.Fatalf("mock LLM: %v", err) }
|
||||||
|
memories := parseExtractedMemories(resp.Content)
|
||||||
|
if len(memories) != 2 {
|
||||||
|
t.Errorf("expected 2 memories, got %d", len(memories))
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user