Skip to content

symfonic.memory.backends.postgres_candidates

postgres_candidates

Native eligible top-k selection; no unbounded row materialization in Python.

statement 

statement(scope, request)

Return parameterized SQL; all user cue/scope/session values are bound.

Source code in src/symfonic/memory/backends/postgres_candidates.py 
27
28
29
30
31
32
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def statement(scope, request):
    """Return parameterized SQL; all user cue/scope/session values are bound."""
    params = [scope.tenant_id, list(request.layers), request.session_id,
              sorted(set(request.cue.lower().split())), request.limit, request.profile_slots]
    prefix, values = sql_prefix_condition(scope, "scope_path", 7)
    if request.exact_scope:
        params.extend([materialise_scope_path(scope), values[1]])
        prefix = "COALESCE(scope_path,$8)=$7"
    else:
        params.extend(values)
    params.append(scope.ancestor_prefix_paths())
    flags = []
    for key in ("__symfonic_pending__", "__symfonic_retracted__"):
        flags.append(f"COALESCE(properties->'{key}','null'::jsonb) "
                     "IN ('null'::jsonb,'false'::jsonb,'0'::jsonb,"
                     "'\"\"'::jsonb,'[]'::jsonb,'{}'::jsonb)")
    event = _identity("past_event")
    profile = _identity("profile", "user")
    absent = " AND ".join(
        f"({part})->'{key}' IS NULL"
        for part in ("properties", "properties->'record_metadata'")
        for key in ("memory_category", "subject")
    )
    marker = ("upper(btrim(COALESCE(properties->'record_metadata'->>'label_prefix',"
              "properties->>'label_prefix','')))='SOUL'")
    profile = (f"(layer='semantic' AND ({profile} OR "
               f"({absent} AND (ltrim(label) ~* '^SOUL([ :]|$)' OR {marker}))))")
    sql = f"""
WITH eligible AS (
 SELECT *, {profile} AS profile,
 (SELECT count(*) FROM unnest($4::text[]) token
  WHERE token=ANY(regexp_split_to_array(
    lower(label || ' ' || COALESCE(properties->>'content','')), '[[:space:]]+'))) AS hits
 FROM memory_nodes WHERE tenant_id=$1 AND layer=ANY($2::text[]) AND {prefix}
 AND {' AND '.join(flags)}
 AND ($3='' OR layer NOT IN ('working','episodic')
      OR (layer='episodic' AND {event})
      OR COALESCE(properties->'record_metadata'->>'session_id',properties->>'session_id')=$3)
), scored AS (
 SELECT *, ((LEAST(GREATEST(importance,1),10)-1)/9.0) *
 (0.25 + 0.75 * CASE WHEN cardinality($4::text[])=0 THEN 1.0
                   ELSE hits::float/cardinality($4::text[]) END) *
 power(0.5,cardinality($9::text[])-array_position($9::text[],COALESCE(scope_path,$8)))
 AS cue_score FROM eligible
), ranked AS (
 SELECT *, row_number() OVER (
 PARTITION BY profile ORDER BY cue_score DESC,id) AS ordinal,
 count(*) OVER () AS eligible_count FROM scored
)
SELECT * FROM ranked
ORDER BY CASE WHEN profile AND ordinal <= $6 THEN 0 ELSE 1 END,cue_score DESC,id
LIMIT $5
"""
    return sql, params