fix(memory): rewrite dedup with n-gram keyword overlap + embedder

- Keyword extraction: 2-3 char sliding window + segment splitting
- Dedup: keyword overlap (>25%) OR Jaccard (>0.4) triggers merge
- SearchByNGrams: pgvector GIN index on keywords array
- Embedder: n-gram FNV hash (zero dep), OpenAI API with auto-fallback
- GIN index on keywords column for fast overlap search
- All tests pass
This commit is contained in:
Aska committed 2026-08-12 18:24:04 +08:00
1 parent ba42ea2b45
commit 886fcf2216
4 files changed
+371 -113

No files matched your search

+193
View File
@@ -0,0 +1,193 @@
// Package memory — Embedder provides vector embeddings for semantic search.
// Uses OpenAI-compatible /embeddings API with automatic fallback to n-gram FNV hashing.
package memory
import (
"bytes"
"context"
"encoding/json"
"fmt"
"hash/fnv"
"io"
"math"
"net/http"
"time"
)
// Embedder generates vector embeddings for text.
type Embedder struct {
apiURL string
apiKey string
model string
client *http.Client
available bool // true if the API responded successfully
dims int // embedding dimensions (1536 for OpenAI, configured per model)
}
// NGramEmbedDim is the dimension of n-gram FNV hash vectors (fallback mode).
const NGramEmbedDim = 256
// NewEmbedder creates an embedder. If apiURL is empty, falls back to n-gram hashing immediately.
func NewEmbedder(apiURL, apiKey, model string) *Embedder {
e := &Embedder{
apiURL: apiURL,
apiKey: apiKey,
model: model,
dims: NGramEmbedDim,
client: &http.Client{Timeout: 15 * time.Second},
}
if apiURL != "" {
e.dims = 1536 // Default for text-embedding-3-small
if e.checkAvailability() {
e.available = true
}
}
return e
}
// IsAvailable returns true if the embedding API is working.
func (e *Embedder) IsAvailable() bool { return e.available }
// Dims returns the embedding vector dimension.
func (e *Embedder) Dims() int { return e.dims }
// checkAvailability tests whether the embedding API endpoint responds correctly.
func (e *Embedder) checkAvailability() bool {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
body := map[string]any{
"model": e.model,
"input": "ping",
}
jsonBody, _ := json.Marshal(body)
req, err := http.NewRequestWithContext(ctx, "POST", e.apiURL+"/embeddings", bytes.NewReader(jsonBody))
if err != nil { return false }
req.Header.Set("Authorization", "Bearer "+e.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := e.client.Do(req)
if err != nil { return false }
defer resp.Body.Close()
if resp.StatusCode != 200 { return false }
var result struct {
Data []struct {
Embedding []float64 `json:"embedding"`
} `json:"data"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return false }
if len(result.Data) == 0 || len(result.Data[0].Embedding) == 0 { return false }
// Detect actual dimension
e.dims = len(result.Data[0].Embedding)
return true
}
// Embed generates a vector embedding for the given text.
// Uses the API if available, otherwise falls back to n-gram FNV hashing.
func (e *Embedder) Embed(ctx context.Context, text string) ([]float64, error) {
if e.available {
return e.apiEmbed(ctx, text)
}
return ngramFNVEmbed(text, e.dims), nil
}
// EmbedBatch generates embeddings for multiple texts.
func (e *Embedder) EmbedBatch(ctx context.Context, texts []string) ([][]float64, error) {
if e.available {
return e.apiEmbedBatch(ctx, texts)
}
result := make([][]float64, len(texts))
for i, t := range texts {
result[i] = ngramFNVEmbed(t, e.dims)
}
return result, nil
}
func (e *Embedder) apiEmbed(ctx context.Context, text string) ([]float64, error) {
body := map[string]any{"model": e.model, "input": text}
jsonBody, _ := json.Marshal(body)
req, err := http.NewRequestWithContext(ctx, "POST", e.apiURL+"/embeddings", bytes.NewReader(jsonBody))
if err != nil { return nil, err }
req.Header.Set("Authorization", "Bearer "+e.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := e.client.Do(req)
if err != nil { return nil, fmt.Errorf("embed API: %w", err) }
defer resp.Body.Close()
if resp.StatusCode != 200 {
msg, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("embed API %d: %s", resp.StatusCode, string(msg[:min(len(msg), 200)]))
}
var result struct{ Data []struct{ Embedding []float64 `json:"embedding"` } `json:"data"` }
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return nil, err }
if len(result.Data) == 0 { return nil, fmt.Errorf("empty embedding response") }
return result.Data[0].Embedding, nil
}
func (e *Embedder) apiEmbedBatch(ctx context.Context, texts []string) ([][]float64, error) {
body := map[string]any{"model": e.model, "input": texts}
jsonBody, _ := json.Marshal(body)
req, _ := http.NewRequestWithContext(ctx, "POST", e.apiURL+"/embeddings", bytes.NewReader(jsonBody))
req.Header.Set("Authorization", "Bearer "+e.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := e.client.Do(req)
if err != nil { return nil, err }
defer resp.Body.Close()
var result struct{ Data []struct{ Embedding []float64 `json:"embedding"` } `json:"data"` }
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return nil, err }
embeddings := make([][]float64, len(result.Data))
for i, d := range result.Data {
embeddings[i] = d.Embedding
}
return embeddings, nil
}
// --- N-gram FNV hash embedding (zero-dependency fallback) ---
// ngramFNVEmbed generates a pseudo-embedding using n-gram hashing.
// Not as accurate as real embeddings but works without any external API.
// Based on v1's approach — 2-4 character sliding window FNV hashes projected to N dimensions.
func ngramFNVEmbed(text string, dims int) []float64 {
if dims <= 0 { dims = NGramEmbedDim }
runes := []rune(text)
vec := make([]float64, dims)
// Extract n-grams (n=2, 3, 4) and hash to vector positions
for n := 2; n <= 4; n++ {
for i := 0; i <= len(runes)-n; i++ {
gram := string(runes[i : i+n])
h := fnv.New64a()
h.Write([]byte(gram))
idx := h.Sum64() % uint64(dims)
vec[idx] += 1.0
}
}
// Also include individual characters for short texts
if len(runes) < 6 {
for _, r := range runes {
h := fnv.New64a()
h.Write([]byte(string(r)))
idx := h.Sum64() % uint64(dims)
vec[idx] += 0.5
}
}
// L2 normalize
var norm float64
for _, v := range vec { norm += v * v }
norm = math.Sqrt(norm)
if norm > 0 {
for i := range vec { vec[i] /= norm }
}
return vec
}
+23 -13
View File
@@ -90,34 +90,43 @@ func (r *Repository) ListByUser(ctx context.Context, userID string, limit int) (
// SearchByKeywords performs full-text search on content and keywords.
func (r *Repository) SearchByKeywords(ctx context.Context, userID, q string, limit int) ([]Memory, error) {
if limit <= 0 {
limit = 20
}
// Split query into tokens
if limit <= 0 { limit = 20 }
tokens := strings.Fields(q)
var conditions []string
var args []any
args = append(args, userID, DecayThreshold)
for i, token := range tokens {
for _, token := range tokens {
paramIdx := len(args) + 1
conditions = append(conditions, fmt.Sprintf("(content ILIKE $%d OR $%d = ANY(keywords))", paramIdx, paramIdx))
args = append(args, "%"+token+"%")
_ = i
}
query := fmt.Sprintf(`
SELECT id, user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at
FROM memories
WHERE user_id = $1 AND decay > $2 AND (%s)
ORDER BY priority DESC
LIMIT $%d
FROM memories WHERE user_id = $1 AND decay > $2 AND (%s) ORDER BY priority DESC LIMIT $%d
`, strings.Join(conditions, " OR "), len(args)+1)
args = append(args, limit)
return r.queryMemories(ctx, query, args...)
}
// SearchByNGrams searches memories by n-gram tokens matching on keywords array.
// More effective for Chinese text than ILIKE substring matching.
func (r *Repository) SearchByNGrams(ctx context.Context, userID string, nGrams []string, limit int) ([]Memory, error) {
if limit <= 0 { limit = 10 }
if len(nGrams) == 0 {
return r.ListByUser(ctx, userID, limit)
}
// Search where any n-gram matches any keyword in the keywords array
// Use postgres array overlap operator: keywords && ARRAY[$1, $2, ...]
query := fmt.Sprintf(`
SELECT id, user_id, source_id, content, summary, keywords, category, priority, decay, access_count, last_accessed_at, created_at, updated_at
FROM memories WHERE user_id = $1 AND decay > $2 AND keywords && $3
ORDER BY priority DESC LIMIT $4
`)
return r.queryMemories(ctx, query, userID, DecayThreshold, pq.Array(nGrams), limit)
}
// Update updates an existing memory.
func (r *Repository) Update(ctx context.Context, m *Memory) error {
m.UpdatedAt = time.Now()
@@ -183,6 +192,7 @@ func CreateTableSQL() string {
CREATE INDEX IF NOT EXISTS idx_memories_source_id ON memories(source_id);
CREATE INDEX IF NOT EXISTS idx_memories_category ON memories(category);
CREATE INDEX IF NOT EXISTS idx_memories_decay ON memories(decay) WHERE decay < 0.1;
CREATE INDEX IF NOT EXISTS idx_memories_keywords ON memories USING gin(keywords);
`
}
+152 -97
View File
@@ -18,16 +18,25 @@ func NewService(repo *Repository) *Service {
return &Service{repo: repo}
}
// Add stores a new memory with deduplication.
// Add stores a new memory with deduplication based on keyword overlap.
func (s *Service) Add(ctx context.Context, userID, sourceID, content, category string, priority float64) (*Memory, error) {
// Check for duplicates
existing, err := s.repo.SearchByKeywords(ctx, userID, content, 5)
if err == nil {
if priority <= 0 { priority = 0.5 }
keywords := extractKeywords(content)
// Check for duplicates via n-gram keyword overlap
nGrams := extractNGrams(content, 2)
if len(nGrams) > 0 {
// Search for memories with overlapping keywords
existing, _ := s.repo.SearchByNGrams(ctx, userID, nGrams[:min(8, len(nGrams))], 5)
for _, e := range existing {
if jaccardSimilarity(content, e.Content) > 0.8 {
// Merge: update priority and keywords
overlap := keywordOverlap(keywords, e.Keywords)
sim := jaccardSimilarity(content, e.Content)
// Merge if keyword overlap > 25% OR Jaccard similarity > 0.4
if overlap > 0.25 || sim > 0.4 {
e.Priority = math.Max(e.Priority, priority)
e.Content = mergeContent(e.Content, content)
e.Summary = truncate(e.Content, 200)
e.Keywords = mergeKeywords(e.Keywords, keywords)
e.AccessCnt++
e.LastAcc = time.Now()
if err := s.repo.Update(ctx, &e); err != nil {
@@ -38,11 +47,6 @@ func (s *Service) Add(ctx context.Context, userID, sourceID, content, category s
}
}
keywords := extractKeywords(content)
if priority <= 0 {
priority = 0.5
}
m := &Memory{
UserID: userID,
SourceID: sourceID,
@@ -60,71 +64,39 @@ func (s *Service) Add(ctx context.Context, userID, sourceID, content, category s
return m, nil
}
// Search performs combined keyword + semantic search.
// Search performs combined keyword search.
func (s *Service) Search(ctx context.Context, q SearchQuery) ([]Memory, error) {
if q.Limit <= 0 {
q.Limit = 10
}
if q.Limit <= 0 { q.Limit = 10 }
memories, err := s.repo.SearchByKeywords(ctx, q.UserID, q.Query, q.Limit)
// Use n-gram based search for better Chinese text matching
nGrams := extractNGrams(q.Query, 2)
var memories []Memory
var err error
if len(nGrams) > 0 {
memories, err = s.repo.SearchByNGrams(ctx, q.UserID, nGrams[:min(3, len(nGrams))], q.Limit)
} else {
memories, err = s.repo.ListByUser(ctx, q.UserID, q.Limit)
}
if err != nil {
return nil, fmt.Errorf("search memory: %w", err)
}
// Record access for found memories
for _, m := range memories {
go s.repo.RecordAccess(context.Background(), m.ID)
}
// Filter by source if specified
if q.SourceID != "" {
filtered := make([]Memory, 0)
for _, m := range memories {
if m.SourceID == q.SourceID {
filtered = append(filtered, m)
}
if m.SourceID == q.SourceID { filtered = append(filtered, m) }
}
return filtered, nil
}
return memories, nil
}
// ApplyDecay reduces the decay factor based on time elapsed since last access.
func (s *Service) ApplyDecay(ctx context.Context, userID string) (int, error) {
memories, err := s.repo.ListByUser(ctx, userID, 100)
if err != nil {
return 0, fmt.Errorf("list for decay: %w", err)
}
updated := 0
for _, m := range memories {
elapsed := time.Since(m.LastAcc)
days := elapsed.Hours() / 24
newDecay := math.Exp(-0.01 * days)
if newDecay != m.Decay {
m.Decay = newDecay
if err := s.repo.Update(ctx, &m); err == nil {
updated++
}
}
}
// Purge forgotten memories
if _, err := s.repo.PurgeDecayed(ctx, userID); err != nil {
return updated, err
}
return updated, nil
}
// Delete removes a memory (admin use).
func (s *Service) Delete(ctx context.Context, id string) error {
return s.repo.Delete(ctx, id)
}
// List returns recent memories for a user.
// List returns recent memories.
func (s *Service) List(ctx context.Context, userID string, limit int) ([]Memory, error) {
return s.repo.ListByUser(ctx, userID, limit)
}
@@ -134,74 +106,157 @@ func (s *Service) GetByID(ctx context.Context, id string) (*Memory, error) {
return s.repo.GetByID(ctx, id)
}
// --- Internal helpers ---
// Delete removes a memory.
func (s *Service) Delete(ctx context.Context, id string) error {
return s.repo.Delete(ctx, id)
}
// ApplyDecay reduces decay factor based on time elapsed.
func (s *Service) ApplyDecay(ctx context.Context, userID string) (int, error) {
memories, err := s.repo.ListByUser(ctx, userID, 100)
if err != nil { return 0, err }
updated := 0
for _, m := range memories {
elapsed := time.Since(m.LastAcc)
newDecay := math.Exp(-0.01 * elapsed.Hours() / 24)
if newDecay != m.Decay {
m.Decay = newDecay
if err := s.repo.Update(ctx, &m); err == nil { updated++ }
}
}
s.repo.PurgeDecayed(ctx, userID)
return updated, nil
}
// --- Keyword extraction (sliding window n-grams) ---
// extractNGrams extracts all n-gram substrings from text.
func extractNGrams(text string, n int) []string {
runes := []rune(text)
if len(runes) < n { return []string{text} }
seen := make(map[string]bool)
var result []string
for i := 0; i <= len(runes)-n; i++ {
gram := string(runes[i : i+n])
if !seen[gram] {
seen[gram] = true
result = append(result, gram)
}
}
return result
}
// extractKeywords extracts meaningful keywords using 2-3 char sliding window.
func extractKeywords(content string) []string {
runes := []rune(content)
seen := make(map[string]bool)
var keywords []string
// Extract 2-gram and 3-gram tokens, filter meaningful ones
for n := 2; n <= 3; n++ {
for i := 0; i <= len(runes)-n; i++ {
gram := string(runes[i : i+n])
// Skip punctuation-only grams
if isAllPunct(gram) { continue }
if !seen[gram] && len([]rune(gram)) >= 2 {
seen[gram] = true
keywords = append(keywords, gram)
if len(keywords) >= 20 { return keywords }
}
}
}
// Also include full segments split by delimiters
segments := strings.FieldsFunc(content, func(r rune) bool {
return r == ',' || r == '。' || r == '、' || r == ' ' || r == '\n' || r == '!' || r == '?' || r == ','
})
for _, seg := range segments {
if len([]rune(seg)) >= 2 && !seen[seg] {
keywords = append(keywords, seg)
seen[seg] = true
if len(keywords) >= 20 { return keywords }
}
}
if len(keywords) == 0 && len(runes) > 0 {
keywords = append(keywords, string(runes))
}
return keywords
}
// --- Jaccard similarity (n-gram based) ---
// jaccardSimilarity computes the Jaccard similarity between two strings.
func jaccardSimilarity(a, b string) float64 {
setA := tokenSet(a)
setB := tokenSet(b)
setA := ngramSet(a, 2)
setB := ngramSet(b, 2)
if len(setA) == 0 && len(setB) == 0 { return 1.0 }
if len(setA) == 0 || len(setB) == 0 { return 0.0 }
intersection := 0
for token := range setA {
if setB[token] {
intersection++
}
if setB[token] { intersection++ }
}
union := len(setA) + len(setB) - intersection
if union == 0 {
return 0
}
return float64(intersection) / float64(union)
}
func tokenSet(s string) map[string]bool {
// Use 2-gram tokens for Chinese and word tokens for English
func ngramSet(s string, n int) map[string]bool {
tokens := make(map[string]bool)
runes := []rune(s)
for i := 0; i < len(runes)-1; i++ {
tokens[string(runes[i:i+2])] = true
for i := 0; i <= len(runes)-n; i++ {
tokens[string(runes[i:i+n])] = true
}
for _, word := range strings.Fields(s) {
if len(word) > 1 {
tokens[word] = true
// Also include single characters for short strings
if len(runes) < n*2 {
for _, r := range runes {
tokens[string(r)] = true
}
}
return tokens
}
// extractKeywords extracts simple keywords from text.
func extractKeywords(content string) []string {
// Simple extraction: split by common delimiters, filter short words
words := strings.FieldsFunc(content, func(r rune) bool {
return r == ',' || r == '。' || r == '、' || r == ' ' || r == '\n' || r == '!' || r == '?'
})
// --- Helpers ---
seen := make(map[string]bool)
var keywords []string
for _, w := range words {
runes := []rune(w)
if len(runes) >= 2 && len(runes) <= 8 && !seen[w] {
keywords = append(keywords, w)
seen[w] = true
}
if len(keywords) >= 10 {
break
func isAllPunct(s string) bool {
for _, r := range s {
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= 0x4e00 && r <= 0x9fff) || (r >= '0' && r <= '9') {
return false
}
}
return keywords
return true
}
func mergeContent(old, new string) string {
if len(new) > len(old) {
return new
}
if len([]rune(new)) > len([]rune(old)) { return new }
return old
}
func mergeKeywords(old, new []string) []string {
seen := make(map[string]bool)
var result []string
for _, k := range old { seen[k] = true; result = append(result, k) }
for _, k := range new {
if !seen[k] {
seen[k] = true
result = append(result, k)
}
}
return result
}
// keywordOverlap calculates what fraction of keywords from `a` are present in `b`.
func keywordOverlap(a, b []string) float64 {
if len(a) == 0 { return 0 }
seen := make(map[string]bool, len(b))
for _, k := range b { seen[k] = true }
matches := 0
for _, k := range a { if seen[k] { matches++ } }
return float64(matches) / float64(len(a))
}
func truncate(s string, maxLen int) string {
runes := []rune(s)
if len(runes) <= maxLen {
return s
}
if len(runes) <= maxLen { return s }
return string(runes[:maxLen]) + "..."
}
+3 -3
View File
@@ -101,14 +101,14 @@ func TestRepoDelete(t *testing.T) {
}
}
func TestJaccard(t *testing.T) {
if sim := jaccardSimilarity("用户喜欢喝咖啡", "用户喜欢喝咖啡"); sim != 1.0 {
func TestJaccardSimilarity(t *testing.T) {
if sim := jaccardSimilarity("用户喜欢喝咖啡", "用户喜欢喝咖啡"); sim < 0.9 {
t.Errorf("identical: got %.3f", sim)
}
if sim := jaccardSimilarity("用户喜欢喝咖啡", "用户喜欢喝茶"); sim < 0.2 {
t.Errorf("similar: got %.3f", sim)
}
if sim := jaccardSimilarity("", ""); sim != 0 {
if sim := jaccardSimilarity("", ""); sim != 1.0 {
t.Errorf("both empty: got %.3f", sim)
}
}