Class: SmartBrain::MemoryStore::Postgres

Inherits:
Object
  • Object
show all
Defined in:
lib/smart_brain/memory_store/postgres.rb

Overview

Postgres-backed MemoryStore. Drop-in replacement for MemoryStore::InMemory that additionally:

- persists memory_items to Postgres (durable across restarts),
- writes memory_chunks with a `simple`-config TSVECTOR so FTS works,
- serves #search_memory via ts_rank-ranked full-text search,
- persists working summaries to the summaries table.

Constant Summary collapse

OVERWRITE_TYPES = 
%w[preferences goals tasks].freeze
RANK_SCALE =

ts_rank is unbounded and typically tiny (~0.06); scale so FTS scores land in the same band as ExactRetriever's overlap+confidence scores.

8.0
TOKEN_RE =

Match an ASCII word OR a single CJK character. The word class is ASCII- only on purpose: Ruby's [[:alnum:]] is Unicode-aware and would greedy- match a whole CJK run (默认存储) as one token, defeating per-character indexing. Splitting CJK into unigrams lets the 'simple' text-search config (whitespace-only splitter) match at character granularity: indexing 默认存储 as 默 认 存 储 lets a query for 存储 (存 储) hit.

/[A-Za-z0-9_\-]+|[\p{Han}\p{Hiragana}\p{Katakana}\p{Hangul}]/.freeze

Instance Method Summary collapse

Constructor Details

#initialize(db:, config:) ⇒ Postgres

Returns a new instance of Postgres.



28
29
30
31
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 28

def initialize(db:, config:)
  @db = db
  @config = config
end

Instance Method Details

#active_items(session_id: nil, scope_ids: nil) ⇒ Object 



63
64
65
66
67
68
69
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 63

def active_items(session_id: nil, scope_ids: nil)
  dataset = db[:memory_items]
    .where(status: 'active', lifecycle_status: Governance::Tiers::CONTEXT_LIFECYCLE)
    .order(:updated_at)
  dataset = scope_filter(dataset, session_id, scope_ids)
  dataset.map { |row| item_from_row(row) }
end

#add_edge(session_id:, edge:, scope_id: nil, scope: nil, source_session_id: nil) ⇒ Object

--- knowledge graph -----------------------------------------------------



247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 247

def add_edge(session_id:, edge:, scope_id: nil, scope: nil, source_session_id: nil)
  resolved_scope_id = scope_id || edge[:scope_id] || legacy_scope_id(session_id)
  id = SecureRandom.uuid
  db[:kg_edges].insert(
    id: id,
    session_id: session_id,
    source_session_id: source_session_id || edge[:source_session_id] || session_id,
    scope_id: resolved_scope_id,
    subject: edge[:subject].to_s,
    predicate: edge[:predicate].to_s,
    object: edge[:object].to_s,
    subject_entity_id: resolve_entity_id(nil, edge[:subject], scope_ids: [resolved_scope_id]),
    object_entity_id: resolve_entity_id(nil, edge[:object], scope_ids: [resolved_scope_id]),
    valid_from: edge[:valid_from] ? time_from(edge[:valid_from]) : Time.now.utc,
    valid_to: edge[:valid_to] ? time_from(edge[:valid_to]) : nil,
    source_turn_id: edge[:source_turn_id],
    source_memory_item_id: edge[:source_memory_item_id],
    confidence: edge[:confidence] || 0.6,
    status: edge[:status] || 'active',
    meta_json: Sequel.pg_jsonb(symbolizable(edge[:meta] || {})),
    created_at: Time.now.utc
  )
  find_edge(id: id)
end

#all_summaries ⇒ Object 



172
173
174
175
176
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 172

def all_summaries
  db[:summaries].all.each_with_object({}) do |row, h|
    h[row[:session_id]] = row_to_summary(row)
  end
end

#create_item(session_id:, item:, scope_id: nil, scope: nil) ⇒ Object

--- knowledge lifecycle -------------------------------------------------



179
180
181
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 179

def create_item(session_id:, item:, scope_id: nil, scope: nil)
  persist_item(session_id: session_id, item: item.merge(scope_id: scope_id, scope: scope).compact)
end

#edge_from_row(row) ⇒ Object 



324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 324

def edge_from_row(row)
  {
    id: row[:id],
    session_id: row[:session_id],
    source_session_id: row[:source_session_id] || row[:session_id],
    scope_id: row[:scope_id],
    scope: scope_ref(row[:scope_id]),
    subject: row[:subject],
    predicate: row[:predicate],
    object: row[:object],
    subject_entity_id: row[:subject_entity_id],
    object_entity_id: row[:object_entity_id],
    valid_from: iso8601(row[:valid_from]),
    valid_to: row[:valid_to] ? iso8601(row[:valid_to]) : nil,
    source_turn_id: row[:source_turn_id],
    source_memory_item_id: row[:source_memory_item_id],
    confidence: row[:confidence].to_f,
    status: row[:status],
    meta: symbolize(row[:meta_json]),
    created_at: iso8601(row[:created_at])
  }
end

#edge_stats(session_id: nil, scope_ids: nil) ⇒ Object 



300
301
302
303
304
305
306
307
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 300

def edge_stats(session_id: nil, scope_ids: nil)
  scoped = scope_filter(db[:kg_edges], session_id, scope_ids)
  {
    total: scoped.count,
    active: scoped.where(status: 'active').count,
    invalidated: scoped.where(status: 'invalidated').count
  }
end

#edges_for_subject(session_id: nil, scope_ids: nil, subject:) ⇒ Object 



281
282
283
284
285
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 281

def edges_for_subject(session_id: nil, scope_ids: nil, subject:)
  scope_filter(db[:kg_edges], session_id, scope_ids)
    .where(Sequel.ilike(:subject, subject.to_s))
    .order(:valid_from).map { |row| edge_from_row(row) }
end

#entities(session_id: nil, scope_ids: nil) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 71

def entities(session_id: nil, scope_ids: nil)
  dataset = scope_filter(db[:memory_items].where(type: 'entities', status: 'active'), session_id, scope_ids)
  dataset.map do |row|
    value = symbolize(row[:value_json])
    canonical = value[:canonical] || value[:name]
    scope = scope_ref(row[:scope_id])
    {
      id: row[:id],
      name: value[:name] || canonical,
      kind: value[:kind] || 'other',
      canonical: canonical,
      memory_item_id: row[:id],
      scope_id: row[:scope_id],
      scope: scope,
      source_session_id: row[:source_session_id] || row[:session_id]
    }
  end
end

#events_for(memory_item_id:) ⇒ Object 



229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 229

def events_for(memory_item_id:)
  db[:knowledge_events].where(memory_item_id: memory_item_id).order(:created_at).map do |row|
    {
      id: row[:id],
      memory_item_id: row[:memory_item_id],
      event_type: row[:event_type],
      from_lifecycle: row[:from_lifecycle],
      to_lifecycle: row[:to_lifecycle],
      reason: row[:reason],
      reason_type: row[:reason_type],
      reviewer: row[:reviewer],
      evidence_refs: symbolize(row[:evidence_refs]),
      created_at: iso8601(row[:created_at])
    }
  end
end

#find_edge(id:) ⇒ Object 



309
310
311
312
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 309

def find_edge(id:)
  row = db[:kg_edges].where(id: id).first
  row ? edge_from_row(row) : nil
end

#find_item(id:) ⇒ Object 



183
184
185
186
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 183

def find_item(id:)
  row = db[:memory_items].where(id: id).first
  row ? item_from_row(row) : nil
end

#invalidate_edge(id:, reason: nil) ⇒ Object 



287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 287

def invalidate_edge(id:, reason: nil)
  edge = find_edge(id: id)
  return nil unless edge

  meta = (edge[:meta] || {}).merge(invalidated_reason: reason)
  db[:kg_edges].where(id: id).update(
    valid_to: Time.now.utc,
    status: 'invalidated',
    meta_json: Sequel.pg_jsonb(symbolizable(meta))
  )
  find_edge(id: id)
end

#latest_summary(session_id:) ⇒ Object 



165
166
167
168
169
170
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 165

def latest_summary(session_id:)
  row = db[:summaries].where(session_id: session_id).first
  return nil unless row

  row_to_summary(row)
end

#query_edges(session_id: nil, scope_ids: nil, subject: nil, predicate: nil, object: nil, include_invalid: false) ⇒ Object 



272
273
274
275
276
277
278
279
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 272

def query_edges(session_id: nil, scope_ids: nil, subject: nil, predicate: nil, object: nil, include_invalid: false)
  ds = scope_filter(db[:kg_edges], session_id, scope_ids)
  ds = ds.where(status: 'active') unless include_invalid
  ds = ds.where(Sequel.ilike(:subject, subject.to_s)) if subject
  ds = ds.where(Sequel.ilike(:predicate, predicate.to_s)) if predicate
  ds = ds.where(Sequel.ilike(:object, object.to_s)) if object
  ds.order(:valid_from).map { |row| edge_from_row(row) }
end

#record_event(memory_item_id:, event_type:, from_lifecycle: nil, to_lifecycle: nil, reason: nil, reason_type: nil, reviewer: nil, evidence_refs: []) ⇒ Object 



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 211

def record_event(memory_item_id:, event_type:, from_lifecycle: nil, to_lifecycle: nil,
                 reason: nil, reason_type: nil, reviewer: nil, evidence_refs: [])
  id = SecureRandom.uuid
  db[:knowledge_events].insert(
    id: id,
    memory_item_id: memory_item_id,
    event_type: event_type,
    from_lifecycle: from_lifecycle,
    to_lifecycle: to_lifecycle,
    reason: reason,
    reason_type: reason_type,
    reviewer: reviewer,
    evidence_refs: Sequel.pg_jsonb(Array(evidence_refs)),
    created_at: Time.now.utc
  )
  events_for(memory_item_id: memory_item_id).last
end

#resolve_entity_id(session_id, name, scope_ids: nil) ⇒ Object 



314
315
316
317
318
319
320
321
322
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 314

def resolve_entity_id(session_id, name, scope_ids: nil)
  return nil if name.nil? || name.to_s.empty?

  match = entities(session_id: session_id, scope_ids: scope_ids).find do |e|
    e[:canonical].to_s.downcase == name.to_s.downcase ||
      e[:name].to_s.downcase == name.to_s.downcase
  end
  match && match[:id]
end

#save_summary(session_id:, summary:) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 146

def save_summary(session_id:, summary:)
  db[:summaries].insert_conflict(
    target: :session_id,
    update: {
      summary_text: summary[:text].to_s,
      summary_version: summary[:summary_version],
      summary_source_turn_range: Sequel.pg_jsonb(symbolizable(summary[:summary_source_turn_range] || {})),
      summary_generated_at: time_from(summary[:summary_generated_at])
    }
  ).insert(
    session_id: session_id,
    summary_text: summary[:text].to_s,
    summary_version: summary[:summary_version],
    summary_source_turn_range: Sequel.pg_jsonb(symbolizable(summary[:summary_source_turn_range] || {})),
    summary_generated_at: time_from(summary[:summary_generated_at])
  )
  summary
end

#search_memory(query:, session_id: nil, scope_ids: nil, limit:) ⇒ Object

Full-text search over memory_chunks (joined to active memory_items), ranked by ts_rank. Returns evidence hashes shaped like ExactRetriever's output so the fusion layer is unchanged.

Query terms are OR'd (not AND'd): a CJK query like "数据库 持久化" tokenizes to 数 据 库 持 久 化, and we want any-matching docs recalled (ts_rank then favours docs matching more terms). AND semantics would drop a persistence decision just because it lacks the literal 数据库.



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 98

def search_memory(query:, session_id: nil, scope_ids: nil, limit:)
  or_expr = or_tsquery_expression(query)
  return [] if or_expr.empty?

  fts = fts_config
  ids = Array(scope_ids).compact
  scope_clause = if ids.empty?
                   'm.session_id = ?'
                 else
                   "m.scope_id IN (#{Array.new(ids.length, '?').join(', ')})"
                 end
  sql = <<~SQL
    SELECT c.text, m.id AS item_id, m.type AS memory_type, m.key, m.value_json, m.confidence, m.tier,
           m.scope_id, m.source_session_id, ms.scope_type, ms.external_id,
           ts_rank(c.tsv, to_tsquery(?, ?)) AS rank
    FROM memory_chunks c
    JOIN memory_items m ON m.id = c.memory_item_id
    JOIN memory_scopes ms ON ms.id = m.scope_id
    WHERE #{scope_clause} AND m.status = 'active'
      AND m.lifecycle_status IN ('raw', 'promoted')
      AND c.tsv @@ to_tsquery(?, ?)
    ORDER BY rank DESC
    LIMIT ?
  SQL

  filter_values = ids.empty? ? [session_id] : ids
  db.fetch(sql, fts, or_expr, *filter_values, fts, or_expr, limit).map do |row|
    value = symbolize(row[:value_json])
    confidence = row[:confidence].to_f
    {
      id: row[:item_id],
      source: 'memory',
      source_uri: "smartbrain://memory/#{row[:item_id]}",
      title: row[:key],
      snippet: flatten_value(value),
      mode: 'fts',
      score: (row[:rank].to_f * RANK_SCALE) + confidence,
      tier: row[:tier] || 'evidence',
      memory_type: row[:memory_type],
      memory_key: row[:key],
      scope_id: row[:scope_id],
      scope: { type: row[:scope_type], id: row[:external_id] },
      source_session_id: row[:source_session_id],
      ref: { memory_item_id: row[:item_id] }
    }
  end
end

#set_lifecycle(id:, lifecycle_status:, merge_value: nil) ⇒ Object 



188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 188

def set_lifecycle(id:, lifecycle_status:, merge_value: nil)
  row = db[:memory_items].where(id: id).first
  return nil unless row

  updates = { lifecycle_status: lifecycle_status, updated_at: Time.now.utc }
  if merge_value
    merged = symbolize(row[:value_json]).merge(merge_value)
    updates[:value_json] = Sequel.pg_jsonb(symbolizable(merged))
  end
  db[:memory_items].where(id: id).update(updates)
  find_item(id: id)
end

#set_status(id:, status:, merge_value: nil) ⇒ Object 



201
202
203
204
205
206
207
208
209
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 201

def set_status(id:, status:, merge_value: nil)
  row = db[:memory_items].where(id: id).first
  return nil unless row

  updates = { status: status, updated_at: Time.now.utc }
  updates[:value_json] = Sequel.pg_jsonb(symbolizable(symbolize(row[:value_json]).merge(merge_value))) if merge_value
  db[:memory_items].where(id: id).update(updates)
  find_item(id: id)
end

#upsert(extracted) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/smart_brain/memory_store/postgres.rb', line 33

def upsert(extracted)
  session_id = extracted.fetch(:session_id)
  items = extracted.fetch(:items, [])
  written = []
  conflicts = []

  db.transaction do
    items.each do |item|
      scope_id = item[:scope_id] || legacy_scope_id(session_id)
      existing = active_item(scope_id: scope_id, type: item[:type], key: item[:key])

      if existing && item[:status] == 'retracted'
        db[:memory_items].where(id: existing[:id]).update(status: 'retracted', updated_at: Sequel::CURRENT_TIMESTAMP)
        conflicts << { type: 'retract', key: item[:key], previous_memory_item_id: existing[:id] }
        next
      end

      if existing && OVERWRITE_TYPES.include?(item[:type])
        db[:memory_items].where(id: existing[:id]).update(status: 'superseded', updated_at: Sequel::CURRENT_TIMESTAMP)
        conflicts << { type: 'overwrite', key: item[:key], previous_memory_item_id: existing[:id] }
      end

      record = persist_item(session_id: session_id, item: item.merge(scope_id: scope_id))
      written << record.slice(:id, :type, :key, :status, :confidence, :scope_id, :scope, :source_session_id)
    end
  end

  { count: written.size, items: written, conflicts: conflicts }
end