Semantic search inside the server · Valkey 9.1

1  $ head -c 400 /data/articles.jsonl; echo; wc -l /data/articles.jsonl; valkey-cli MODULE LIST; valkey-cli INFO server | head -20; python3 -c "import valkey" ; python3 -c "import redis"; pip list 2>/dev/null | grep -i -E "valkey|redis|numpy"; grep -v '^#' /etc/valkey/valkey.conf | grep -v '^$'

2  $ valkey-cli DBSIZE; valkey-cli FT._LIST; valkey-cli CONFIG GET save; valkey-cli CONFIG GET appendonly; valkey-cli KEYS '*' | head; python3 -c "
import json;ids=set();n=0
for l in open('/data/articles.jsonl'):
  d=json.loads(l);ids.add(d['id']);n+=1
  assert len(d['embedding'])==64, d['id']
print(n,len(ids))"

3  $ cat > /tmp/load.py <<'EOF'
import json, struct, valkey
r = valkey.Valkey(host='127.0.0.1', port=6379)
p = r.pipeline(transaction=False)
for line in open('/data/articles.jsonl'):
    d = json.loads(line)
    p.hset(f"article:{d['id']}", mapping={'id': d['id'], 'title': d['title'],
           'embedding': struct.pack('<64f', *d['embedding'])})
p.execute()
r.execute_command('FT.CREATE','articles_idx','ON','HASH','PREFIX','1','article:','SCHEMA',
  'id','TAG','embedding','VECTOR','FLAT','6','TYPE','FLOAT32','DIM','64','DISTANCE_METRIC','COSINE')
EOF
python3 /tmp/load.py && sleep 1 && valkey-cli DBSIZE && valkey-cli FT.INFO articles_idx | head -60

4  $ mkdir -p /app && cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Print ids of the 5 articles most similar (cosine) to the query embedding on stdin.
The KNN search runs inside Valkey (valkey-search index 'articles_idx', FLAT/COSINE)."""
import json, struct, sys
import valkey

K = 5
vec = json.load(sys.stdin)
if len(vec) != 64:
    sys.exit(f"expected 64 floats, got {len(vec)}")
r = valkey.Valkey(host='127.0.0.1', port=6379)
res = r.execute_command(
    'FT.SEARCH', 'articles_idx', f'*=>[KNN {K} @embedding $vec AS dist]',
    'PARAMS', '2', 'vec', struct.pack('<64f', *map(float, vec)),
    'RETURN', '2', 'id', 'dist', 'LIMIT', '0', str(K), 'DIALECT', '2')
hits = []
for i in range(1, len(res), 2):
    f = res[i + 1]
    d = {f[j].decode(): f[j + 1].decode() for j in range(0, len(f), 2)}
    hits.append((float(d['dist']), d['id']))
hits.sort()  # order by Valkey-computed distance (ascending = most similar first)
for _, i in hits[:K]:
    print(i)
EOF
cat > /tmp/check.py <<'EOF'
import json, numpy as np, subprocess, random
A=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in A]; M=np.array([a['embedding'] for a in A]); Mn=M/np.linalg.norm(M,axis=1,keepdims=True)
random.seed(1); bad=0
for t in range(30):
    q = (M[random.randrange(1000)] + np.random.normal(0,0.1,64)) if t%2 else np.random.normal(0,1,64)
    s=Mn@(q/np.linalg.norm(q)); exp=[ids[i] for i in np.argsort(-s)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(q.tolist()),capture_output=True,text=True).stdout.split()
    if out!=exp: bad+=1; print(out,exp,sorted(s)[-6:])
print('mismatches',bad)
EOF
python3 /tmp/check.py

5  $ valkey-cli SAVE; ls -la /var/lib/valkey; echo '[0.1,0.2]' | python3 /app/search.py; python3 -c "import json;print(json.dumps(json.loads(open('/data/articles.jsonl').readline())['embedding']))" | python3 /app/search.py

6  $ valkey-cli SHUTDOWN SAVE; sleep 1; valkey-server /etc/valkey/valkey.conf; sleep 2; valkey-cli DBSIZE; valkey-cli FT._LIST; valkey-cli FT.INFO articles_idx | grep -A1 -E 'num_docs|backfill_in_progress'; python3 /tmp/check.py

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ head -c 300 /data/articles.jsonl; echo; wc -l /data/articles.jsonl; valkey-cli INFO server | head -20; valkey-cli MODULE LIST; cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$'; ls /usr/lib/valkey* /usr/local/lib 2>/dev/null; python3 -c "import valkey" ; python3 -c "import redis"; pip list 2>/dev/null | grep -i -E "valkey|redis|numpy"

2  $ valkey-cli CONFIG GET save; valkey-cli CONFIG GET appendonly; valkey-cli DBSIZE; valkey-cli FT._LIST; ls -la /var/lib/valkey; python3 -c "
import json,numpy as np
A=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in A]; print(len(set(ids)), set(len(a['embedding']) for a in A))
E=np.array([a['embedding'] for a in A]); n=np.linalg.norm(E,axis=1); print(n.min(),n.max())
print(len({tuple(r) for r in E.round(6)}))
print(set(k for a in A for k in a))
"

3  $ cat > /tmp/load.py <<'EOF'
import json, numpy as np, valkey
r = valkey.Valkey(host='127.0.0.1', port=6379)
p = r.pipeline(transaction=False)
for line in open('/data/articles.jsonl'):
    a = json.loads(line)
    p.hset(f"article:{a['id']}", mapping={'id': a['id'], 'title': a['title'],
        'embedding': np.asarray(a['embedding'], dtype=np.float32).tobytes()})
p.execute()
try: r.execute_command('FT.DROPINDEX', 'articles_idx')
except Exception as e: print(e)
print(r.execute_command('FT.CREATE','articles_idx','ON','HASH','PREFIX','1','article:','SCHEMA',
  'embedding','VECTOR','FLAT','6','TYPE','FLOAT32','DIM','64','DISTANCE_METRIC','COSINE'))
EOF
python3 /tmp/load.py; sleep 1; valkey-cli DBSIZE; valkey-cli FT.INFO articles_idx | head -60

4  $ mkdir -p /app; cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Usage: python3 /app/search.py < query.json  -> top-5 article ids by cosine similarity (KNN in Valkey Search)."""
import json, sys
import numpy as np
import valkey

K = 5
query = json.load(sys.stdin)
vec = np.asarray(query, dtype=np.float32)
if vec.shape != (64,):
    sys.exit(f"expected 64-dim embedding, got shape {vec.shape}")

r = valkey.Valkey(host="127.0.0.1", port=6379)
res = r.execute_command(
    "FT.SEARCH", "articles_idx", f"*=>[KNN {K} @embedding $vec AS score]",
    "PARAMS", "2", "vec", vec.tobytes(),
    "RETURN", "2", "id", "score",
    "SORTBY", "score", "ASC",
    "LIMIT", "0", str(K),
    "DIALECT", "2",
)
hits = []
for i in range(1, len(res), 2):
    key = res[i].decode()
    f = res[i + 1]
    d = {f[j].decode(): f[j + 1].decode() for j in range(0, len(f), 2)}
    hits.append((float(d["score"]), d.get("id") or key.split(":", 1)[1]))
hits.sort(key=lambda h: h[0])  # cosine distance ascending = most similar first
for _, aid in hits[:K]:
    print(aid)
EOF
cd /tmp && python3 - <<'EOF'
import json,numpy as np,subprocess
A=[json.loads(l) for l in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A]); ids=[a['id'] for a in A]
rng=np.random.default_rng(0); bad=0
for t in range(30):
    q=rng.normal(size=64) if t%2 else E[t*7]+rng.normal(scale=0.05,size=64)
    s=E@q/np.linalg.norm(E,axis=1)/np.linalg.norm(q)
    exp=[ids[i] for i in np.argsort(-s)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(q.tolist()),capture_output=True,text=True)
    got=out.stdout.split()
    if got!=exp: bad+=1; print(exp,got,out.stderr)
print('mismatches',bad)
EOF

5  $ sed -i '/"SORTBY", "score", "ASC",/d' /app/search.py && cd /tmp && python3 - <<'EOF'
import json,numpy as np,subprocess
A=[json.loads(l) for l in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A]); ids=[a['id'] for a in A]
rng=np.random.default_rng(0); bad=0
for t in range(40):
    q=rng.normal(size=64) if t%2 else E[t*7]+rng.normal(scale=0.05,size=64)
    s=E@q/np.linalg.norm(E,axis=1)/np.linalg.norm(q)
    o=np.argsort(-s); exp=[ids[i] for i in o[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(q.tolist()),capture_output=True,text=True)
    got=out.stdout.split()
    if got!=exp: bad+=1; print(exp,got,s[o[:6]],out.stderr[-300:])
print('mismatches',bad)
EOF

6  $ valkey-cli SAVE; ls -la /var/lib/valkey; cat /app/search.py; python3 -c "
import json;print(json.dumps(json.loads(open('/data/articles.jsonl').readline())['embedding']))" > /tmp/q.json; python3 /app/search.py < /tmp/q.json

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ valkey-cli INFO server | head -20; valkey-cli MODULE LIST; cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$'; ls /usr/lib/valkey* /usr/local/lib 2>/dev/null; find / -name "*search*.so" 2>/dev/null | head; head -c 300 /data/articles.jsonl; python3 -c "import valkey" ; python3 -c "import redis"; pip list 2>/dev/null | grep -i -E "valkey|redis|numpy"

2  $ cd /data; wc -l articles.jsonl; python3 -c "
import json,collections,math
ids=collections.Counter();bad=0
for i,l in enumerate(open('articles.jsonl')):
  l=l.strip()
  if not l: print('blank',i);continue
  d=json.loads(l); ids[d['id']]+=1
  e=d['embedding']
  if len(e)!=64 or any(not isinstance(x,(int,float)) or x!=x for x in e): print('bad',i,d['id'],len(e))
  n=math.sqrt(sum(x*x for x in e))
  if n<1e-9 or abs(n-1)>0.01: print('norm',d['id'],n)
  if set(d)!={'id','title','embedding'}: print('keys',d.keys())
print([k for k,v in ids.items() if v>1], len(ids))
" | head -30; valkey-cli CONFIG GET save; valkey-cli CONFIG GET appendonly; valkey-cli DBSIZE; valkey-cli FT._LIST

3  $ cat > /app/load.py <<'EOF'
import json, struct, valkey
r = valkey.Valkey(host='127.0.0.1', port=6379)
try: r.execute_command('FT.DROPINDEX', 'articles_idx')
except Exception: pass
p = r.pipeline(transaction=False)
for line in open('/data/articles.jsonl'):
    if not line.strip(): continue
    d = json.loads(line)
    p.hset(f"article:{d['id']}", mapping={'id': d['id'], 'title': d['title'],
           'embedding': struct.pack('<64f', *d['embedding'])})
p.execute()
r.execute_command('FT.CREATE', 'articles_idx', 'ON', 'HASH', 'PREFIX', 1, 'article:',
    'SCHEMA', 'embedding', 'VECTOR', 'FLAT', 6, 'TYPE', 'FLOAT32', 'DIM', 64, 'DISTANCE_METRIC', 'COSINE')
EOF
mkdir -p /app; python3 /app/load.py && valkey-cli DBSIZE && valkey-cli FT.INFO articles_idx | head -60

4  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
import sys, json, struct, valkey

def main():
    q = json.load(sys.stdin)
    if len(q) != 64:
        sys.exit(f"expected 64-dim embedding, got {len(q)}")
    r = valkey.Valkey(host='127.0.0.1', port=6379)
    res = r.execute_command(
        'FT.SEARCH', 'articles_idx', '*=>[KNN 5 @embedding $vec AS score]',
        'PARAMS', 2, 'vec', struct.pack('<64f', *map(float, q)),
        'RETURN', 2, 'id', 'score', 'SORTBY', 'score', 'ASC', 'LIMIT', 0, 5, 'DIALECT', 2)
    hits = []
    for i in range(1, len(res), 2):
        f = res[i + 1]
        d = {f[j].decode(): f[j + 1].decode() for j in range(0, len(f), 2)}
        hits.append((float(d['score']), d['id']))
    hits.sort(key=lambda h: h[0])
    for _, aid in hits[:5]:
        print(aid)

if __name__ == '__main__':
    main()
EOF
cd /tmp && python3 - <<'EOF'
import json,random,subprocess,numpy as np
A=[json.loads(l) for l in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A]);E/=np.linalg.norm(E,axis=1,keepdims=True)
ok=0
for t in range(50):
    q=(E[random.randrange(1000)]+np.random.randn(64)*0.1) if t%2 else np.random.randn(64)
    s=E@(q/np.linalg.norm(q)); exp=[A[i]['id'] for i in np.argsort(-s)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(q.tolist()),capture_output=True,text=True)
    got=out.stdout.split()
    if got==exp: ok+=1
    else: print(exp,got,out.stderr, sorted(s)[-6:])
print(ok,'/50')
EOF

5  $ sed -i "s/ 'SORTBY', 'score', 'ASC',//" /app/search.py && grep -n LIMIT /app/search.py && cd /tmp && python3 - <<'EOF'
import json,random,subprocess,numpy as np
A=[json.loads(l) for l in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A]);E/=np.linalg.norm(E,axis=1,keepdims=True)
ok=0
for t in range(60):
    q=(E[random.randrange(1000)]+np.random.randn(64)*0.1) if t%2 else np.random.randn(64)*(t+1)
    s=E@(q/np.linalg.norm(q)); exp=[A[i]['id'] for i in np.argsort(-s)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(q.tolist()),capture_output=True,text=True)
    got=out.stdout.split()
    if got==exp: ok+=1
    else: print(exp,got,out.stderr[-300:], sorted(s)[-6:])
print(ok,'/60')
EOF

6  $ python3 -c "
import json
A=[json.loads(l) for l in open('/data/articles.jsonl')]
print(len(A)-len({tuple(a['embedding']) for a in A}),'dup embeddings')"; valkey-cli SAVE; ls -la /var/lib/valkey; valkey-cli LASTSAVE

7  $ valkey-cli SHUTDOWN SAVE; sleep 1; valkey-server /etc/valkey/valkey.conf; sleep 2; valkey-cli DBSIZE; valkey-cli FT._LIST; valkey-cli FT.INFO articles_idx | grep -A1 -E "num_docs|state"; python3 -c "
import json;print(json.dumps(json.loads(open('/data/articles.jsonl').readline())['embedding']))" > /tmp/q.json; python3 /app/search.py < /tmp/q.json

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ ls -la /app /data; head -c 550 /data/articles.jsonl; echo; valkey-cli INFO server | head -25; valkey-cli MODULE LIST; grep -Ein 'search|module|protected' /etc/valkey/valkey.conf | head -60

2  $ which valkey-cli python3; python3 - <<'PY'
try: import redis; print('redis',redis.__version__)
except Exception as e: print(e)
PY
ls /usr/lib/valkey /usr/lib/redis/modules /usr/local/lib 2>/dev/null | head -100

3  $ python3 - <<'PY'
import requests
u='https://valkey.io/commands/ft.search/'
t=requests.get(u).text
from bs4 import BeautifulSoup
s=BeautifulSoup(t,'html.parser').get_text(' ',strip=True)
for word in ['KNN','VECTOR_RANGE','=>[','query language']:
 i=s.find(word); print(word,i,s[i-250:i+850])
PY

4  $ python3 - <<'PY'
import json, math
rows=[json.loads(l) for l in open('/data/articles.jsonl')]
print(len(rows),len({r['id'] for r in rows}),set(len(r['embedding']) for r in rows));print(min(math.sqrt(sum(x*x for x in r['embedding'])) for r in rows))
PY
valkey-cli FT.CREATE kb:idx ON HASH PREFIX 1 kb:article: SKIPINITIALSCAN SCHEMA embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE

5  $ cat > /tmp/load_articles.py <<'PY'
import json
import struct
import redis

client = redis.Redis(host='127.0.0.1', port=6379)
with client.pipeline(transaction=False) as pipe:
    count = 0
    for line in open('/data/articles.jsonl', encoding='utf-8'):
        row = json.loads(line)
        pipe.hset('kb:article:' + row['id'], mapping={
            'id': row['id'],
            'title': row['title'],
            'embedding': struct.pack('<64f', *row['embedding']),
        })
        count += 1
        if count % 100 == 0:
            pipe.execute()
    pipe.execute()
print('Loaded', count, 'articles')
PY
python3 /tmp/load_articles.py
valkey-cli FT.INFO kb:idx | head -90
valkey-cli --scan --pattern 'kb:article:*' | wc -l

6  $ python3 - <<'PY'
import redis,json,struct
r=redis.Redis()
v=struct.pack('<64f',*json.loads(open('/data/articles.jsonl').readline())['embedding'])
for expr in ('*=>[KNN 5 @embedding $vec AS distance]','*=>[KNN 5 @embedding $vec]'):
 try:
  result=r.execute_command('FT.SEARCH','kb:idx',expr,'PARAMS',2,'vec',v,'RETURN',1,'id','LIMIT',0,5,'DIALECT',2)
  print(expr,result)
 except Exception as e: print(type(e),str(e))
PY

7  $ cat > /app/search.py <<'PY'
#!/usr/bin/env python3
"""Return the five nearest article IDs using Valkey's exact cosine vector index."""

import json
import math
import struct
import sys

import redis


def main():
    embedding = json.load(sys.stdin)
    if (not isinstance(embedding, list) or len(embedding) != 64
            or any(isinstance(x, bool) or not isinstance(x, (int, float))
                   or not math.isfinite(x) for x in embedding)):
        raise ValueError('Expected a JSON list of 64 finite numbers')

    vector = struct.pack('<64f', *embedding)
    client = redis.Redis(host='127.0.0.1', port=6379)
    result = client.execute_command(
        'FT.SEARCH', 'kb:idx', '*=>[KNN 5 @embedding $query]',
        'PARAMS', 2, 'query', vector,
        'RETURN', 1, 'id', 'LIMIT', 0, 5, 'DIALECT', 2,
    )
    for fields in result[2::2]:
        values = dict(zip(fields[::2], fields[1::2]))
        print(values[b'id'].decode('utf-8'))


if __name__ == '__main__':
    main()
PY
python3 - <<'PY' | python3 /app/search.py
import json
print(json.dumps(json.loads(open('/data/articles.jsonl').readline())['embedding']))
PY
valkey-cli BGSAVE

8  $ valkey-cli INFO persistence | grep -E 'rdb_bgsave_in_progress|rdb_last_bgsave_status|rdb_last_save_time'; ls -l /app/search.py; valkey-cli FT.INFO kb:idx | grep -A1 -E '^num_docs$|^state$'

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ valkey-cli ping; valkey-cli MODULE LIST; valkey-cli INFO server | head -25; ls -la /app /data; head -c 350 /data/articles.jsonl

2  $ grep -Ev '^\s*(#|$)' /etc/valkey/valkey.conf | tail -70; command -v python3; python3 - <<'PY'
import importlib.util
for x in ['redis','valkey','numpy']:
 print(x,importlib.util.find_spec(x))
PY
find / -iname '*search*.so' -o -iname '*vector*.so' 2>/dev/null | head -60

3  $ valkey-cli FT._LIST; valkey-cli CONFIG GET save; valkey-cli CONFIG GET appendonly; python3 - <<'PY'
import json,collections,math
A=[json.loads(s) for s in open('/data/articles.jsonl')]
print(len(A),len(set(a['id'] for a in A)),collections.Counter(len(a['embedding']) for a in A))
print('zero vectors',sum(not any(a['embedding']) for a in A))
PY

4  $ cat > /app/search.py <<'PY'
#!/usr/bin/env python3
"""Return the five nearest knowledge-base article IDs using Valkey vector search."""
import json
import math
import struct
import sys

import valkey


def main():
    vector = json.load(sys.stdin)
    if not isinstance(vector, list) or len(vector) != 64 or any(
        not isinstance(x, (float, int)) or isinstance(x, bool) or not math.isfinite(x)
        for x in vector
    ):
        raise ValueError('query must be a JSON list of 64 finite numbers')
    blob = struct.pack('<64f', *vector)
    client = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=False)
    result = client.execute_command(
        'FT.SEARCH', 'articles_idx', '*=>[KNN 5 @embedding $vec AS distance]',
        'PARAMS', 2, 'vec', blob, 'SORTBY', 'distance', 'ASC',
        'LIMIT', 0, 5, 'NOCONTENT', 'DIALECT', 2,
    )
    for key in result[1:]:
        print(key.decode('utf-8').removeprefix('article:'))


if __name__ == '__main__':
    main()
PY
python3 - <<'PY'
import json,struct,valkey
r=valkey.Valkey(host='127.0.0.1',port=6379)
print('Index:',r.execute_command('FT.CREATE','articles_idx','ON','HASH','PREFIX',1,'article:','SKIPINITIALSCAN','SCHEMA','embedding','VECTOR','FLAT',6,'TYPE','FLOAT32','DIM',64,'DISTANCE_METRIC','COSINE'))
with open('/data/articles.jsonl') as f:
    with r.pipeline(transaction=False) as p:
        n=0
        for line in f:
            obj=json.loads(line)
            p.hset('article:'+obj['id'],mapping={'id':obj['id'],'title':obj['title'],'embedding':struct.pack('<64f',*obj['embedding'])})
            n+=1
            if n%100==0:
                p.execute()
        p.execute()
print('Loaded:',n)
print('info:',r.execute_command('FT.INFO','articles_idx'))
PY

5  $ python3 - <<'PY'
import json,subprocess,numpy as np
A=[json.loads(s) for s in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A]); E=E/np.linalg.norm(E,axis=1)[:,None]
for i in (0,311,999):
 q=A[i]['embedding']; p=subprocess.run(['python3','/app/search.py'],input=json.dumps(q),text=True,capture_output=True)
 rank=np.argsort(-(E@np.array(q)) )[:5]
 print('query',i,'exit',p.returncode,'actual',p.stdout.splitlines(),'expected',[A[j]['id'] for j in rank], 'stderr',p.stderr)
PY
valkey-cli SAVE; ls -lh /var/lib/valkey/

6  $ sed -i "s/'PARAMS', 2, 'vec', blob, 'SORTBY', 'distance', 'ASC',/'PARAMS', 2, 'vec', blob,/" /app/search.py
python3 - <<'PY'
import json,subprocess,numpy as np
A=[json.loads(s) for s in open('/data/articles.jsonl')]
E=np.array([a['embedding'] for a in A],dtype=np.float32); E=E/np.linalg.norm(E,axis=1)[:,None]
for i in (0,311,999):
 q=A[i]['embedding']; p=subprocess.run(['python3','/app/search.py'],input=json.dumps(q),text=True,capture_output=True)
 rank=np.argsort(-(E@np.array(q,dtype=np.float32)))[:5]
 print('query',i,'exit',p.returncode,'actual',p.stdout.splitlines(),'expected',[A[j]['id'] for j in rank], 'stderr',p.stderr[:400])
PY

7  $ valkey-cli LASTSAVE; valkey-cli CONFIG GET dir dbfilename; ls -lah /var/lib/valkey; valkey-cli INFO persistence | head -40; cat /app/search.py

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ ls -l /data/articles.jsonl /app /etc/valkey/valkey.conf; valkey-cli ping; valkey-cli MODULE LIST; command -v valkey-cli; python3 - <<'PY'
import json
p='/data/articles.jsonl'
with open(p) as f:
 for i in range(2):
  obj=json.loads(next(f));print(obj['id'],obj['title'],len(obj['embedding']),obj['embedding'][:4])
PY

2  $ valkey-server --version; valkey-cli COMMAND INFO FT.CREATE FT.SEARCH VSIM VADD; grep -E '^(loadmodule|save|appendonly|dir|requirepass|protected-mode)' /etc/valkey/valkey.conf | head -60; ls /usr/lib/valkey /usr/lib/redis/modules 2>/dev/null

3  $ cat /app/catalog.py; python3 - <<'PY'
for pkg in ['redis','valkey']:
 try:
  m=__import__(pkg);print(pkg,m.__version__ if hasattr(m,'__version__') else m)
 except Exception as e:print(pkg,str(e))
PY
valkey-cli FT._LIST

4  $ python3 - <<'PY'
import requests
u='https://valkey.io/commands/ft.search/'
s=requests.get(u).text
from bs4 import BeautifulSoup
text=BeautifulSoup(s,'html.parser').get_text(' ',strip=True)
for x in ['KNN 10','KNN 5','KNN','COSINE']:
 p=text.find(x);print(x,p,text[p-240:p+450])
PY

5  $ valkey-cli FT.CREATE kb:idx ON HASH PREFIX 1 kb:article: SKIPINITIALSCAN SCHEMA embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE; valkey-cli FT.INFO kb:idx | head -75

6  $ cat /etc/valkey/valkey.conf; python3 - <<'PY'
import valkey,redis
print(valkey.Valkey,redis.Redis)
PY

7  $ python3 - <<'PY'
import json,struct,redis
r=redis.Redis()
ids=set()
with open('/data/articles.jsonl') as f:
 with r.pipeline(transaction=False) as pipe:
  for line in f:
   a=json.loads(line); id=a['id'];assert id not in ids;ids.add(id)
   v=a['embedding'];assert len(v)==64
   pipe.hset('kb:article:'+id,mapping={'id':id,'title':a['title'],'embedding':struct.pack('<64f',*v)})
  out=pipe.execute(raise_on_error=True)
print('loaded',len(ids),'HSET replies',len(out))
print('index num_docs:',r.execute_command('FT.INFO','kb:idx')[r.execute_command('FT.INFO','kb:idx').index(b'num_docs')+1])
a=json.loads(open('/data/articles.jsonl').readline())
res=r.execute_command('FT.SEARCH','kb:idx','*=>[KNN 5 @embedding $query AS distance]','PARAMS',2,'query',struct.pack('<64f',*a['embedding']),'SORTBY','distance','ASC','RETURN',2,'id','distance','LIMIT',0,5,'DIALECT',2)
print(res)
PY

8  $ python3 - <<'PY'
import json,struct,redis
r=redis.Redis(); a=json.loads(open('/data/articles.jsonl').readline());v=struct.pack('<64f',*a['embedding'])
for query in ['*=>[KNN 5 @embedding $query AS distance]','*=>[KNN 5 @embedding $query]']:
 for tail in [('PARAMS',2,'query',v,'RETURN',2,'id','distance','LIMIT',0,5,'DIALECT',2),('PARAMS',2,'query',v,'NOCONTENT','LIMIT',0,5,'DIALECT',2)]:
  try: print(query,r.execute_command('FT.SEARCH','kb:idx',query,*tail))
  except Exception as e: print(query,type(e).__name__,e)
PY

9  $ cat > /app/search.py <<'PY'
"""Return the five closest knowledge-base articles using Valkey's cosine vector index."""
import json
import struct
import sys

import redis


def main():
    embedding = json.load(sys.stdin)
    if not isinstance(embedding, list) or len(embedding) != 64:
        raise ValueError('Expected a JSON list of 64 floats')
    vector = struct.pack('<64f', *embedding)
    client = redis.Redis(host='127.0.0.1', port=6379)
    result = client.execute_command(
        'FT.SEARCH', 'kb:idx', '*=>[KNN 5 @embedding $query]',
        'PARAMS', 2, 'query', vector,
        'RETURN', 1, 'id', 'LIMIT', 0, 5, 'DIALECT', 2,
    )
    if result[0] != 5:
        raise RuntimeError(f'Expected 5 search results, got {result[0]}')
    for fields in result[2::2]:
        values = dict(zip(fields[::2], fields[1::2]))
        print(values[b'id'].decode('utf-8'))


if __name__ == '__main__':
    main()
PY
python3 - <<'PY' | python3 /app/search.py
import json
print(json.dumps(json.loads(open('/data/articles.jsonl').readline())['embedding']))
PY
valkey-cli SAVE; valkey-cli LASTSAVE; ls -lh /var/lib/valkey/

10  $ python3 - <<'PY'
import json,math,subprocess,random
rows=[json.loads(s) for s in open('/data/articles.jsonl')]
for a in random.Random(4).sample(rows,5):
 q=a['embedding'];nq=math.sqrt(sum(v*v for v in q))
 expected=sorted(rows,key=lambda r:sum(x*y for x,y in zip(r['embedding'],q))/(math.sqrt(sum(x*x for x in r['embedding']))*nq),reverse=True)[:5]
 actual=subprocess.check_output(['python3','/app/search.py'],input=json.dumps(q).encode()).decode().splitlines()
 assert actual==[r['id'] for r in expected],(actual,[r['id'] for r in expected])
print('5 sample queries match exact cosine top 5')
PY

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ valkey-server --version; valkey-cli --version; ls -la /data/; head -c 600 /data/articles.jsonl; echo; wc -l /data/articles.jsonl

2  $ cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$'; echo ---; valkey-cli ping; valkey-cli info server | head -30

3  $ valkey-cli module list; echo ---; valkey-cli ft._list; echo ---; valkey-cli command list | grep -i -E 'ft\.|vector' | head -40

4  $ ls /usr/share/doc/ 2>/dev/null; find / -name "*search*" -path "*valkey*" 2>/dev/null | head; find / -iname "*query*language*" 2>/dev/null | head

5  $ ls -R /usr/local/lib/python3.13/dist-packages/valkey/commands/search; pip show valkey 2>/dev/null | head

6  $ grep -n -i "vector\|KNN\|score_field\|distance" /usr/local/lib/python3.13/dist-packages/valkey/commands/search/querystring.py | head -40; echo =====; grep -rn -i "vector" /usr/local/lib/python3.13/dist-packages/valkey/commands/search/query.py | head -40

7  $ cd /app 2>/dev/null || mkdir -p /app; cd /app; python3 - <<'EOF'
import json, struct
import valkey
r = valkey.Valkey(host='127.0.0.1', port=6379)
r.flushall()
line = open('/data/articles.jsonl').readline()
a = json.loads(line)
print(a['id'], a['title'], len(a['embedding']))
blob = struct.pack('<64f', *a['embedding'])
print(len(blob))
r.hset('article:'+a['id'], mapping={'id': a['id'], 'title': a['title'], 'embedding': blob})
r.execute_command('FT.CREATE','idx','ON','HASH','PREFIX','1','article:','SCHEMA','id','TAG','title','TEXT','embedding','VECTOR','FLAT','6','TYPE','FLOAT32','DIM','64','DISTANCE_METRIC','COSINE')
import time; time.sleep(0.5)
print(r.execute_command('FT.INFO','idx'))
q = struct.pack('<64f', *a['embedding'])
res = r.execute_command('FT.SEARCH','idx','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',q,'RETURN','2','id','score','SORTBY','score','DIALECT','2')
print(res)
EOF

8  $ cd /app; python3 - <<'EOF'
import json, struct
import valkey
r = valkey.Valkey(host='127.0.0.1', port=6379)
a = json.loads(open('/data/articles.jsonl').readline())
q = struct.pack('<64f', *a['embedding'])
for args in [
  ['FT.SEARCH','idx','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',q,'RETURN','2','id','score','DIALECT','2'],
  ['FT.SEARCH','idx','*=>[KNN 5 @embedding $vec]','PARAMS','2','vec',q,'DIALECT','2'],
  ['FT.SEARCH','idx','*=>[KNN 5 @embedding $vec]','PARAMS','2','vec',q,'RETURN','1','id','DIALECT','2'],
  ['FT.SEARCH','idx','*','PARAMS','2','vec',q,'DIALECT','2'],
]:
    try:
        print(args[:3], '->', r.execute_command(*args))
    except Exception as e:
        print(args[:3], 'ERR', e)
EOF

9  $ valkey-cli config get save; valkey-cli config get appendonly; valkey-cli config get dir; valkey-cli config get maxmemory

10  $ cat > /app/load_articles.py <<'EOF'
#!/usr/bin/env python3
"""Load knowledge-base articles from /data/articles.jsonl into Valkey and
build a valkey-search vector index over their embeddings.

Key layout : HASH  article:<id>  -> {id, title, embedding(FLOAT32 binary, 64 dims)}
Index      : FT.CREATE articles ON HASH PREFIX 1 article:
             SCHEMA id TAG, title TEXT, embedding VECTOR FLAT ... DISTANCE_METRIC COSINE
"""
import json
import struct
import sys
import time

import valkey

ARTICLES_PATH = "/data/articles.jsonl"
INDEX_NAME = "articles"
PREFIX = "article:"
DIM = 64
BATCH = 200


def pack(vec):
    return struct.pack("<%df" % len(vec), *vec)


def main():
    client = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=False)
    client.ping()

    # Idempotent: drop any previous index / data.
    try:
        client.execute_command("FT.DROPINDEX", INDEX_NAME, "DD")
    except valkey.exceptions.ResponseError:
        pass

    pipe = client.pipeline(transaction=False)
    n = 0
    with open(ARTICLES_PATH) as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            art = json.loads(line)
            emb = art["embedding"]
            assert len(emb) == DIM, "unexpected embedding dim for %s" % art["id"]
            key = PREFIX + art["id"]
            pipe.hset(key, mapping={"id": art["id"], "title": art["title"],
                                    "embedding": pack(emb)})
            n += 1
            if n % BATCH == 0:
                pipe.execute()
    pipe.execute()
    print("loaded %d articles" % n)

    # Create the vector index (created *after* loading triggers a backfill;
    # we also wait until the backfill reports complete).
    client.execute_command(
        "FT.CREATE", INDEX_NAME, "ON", "HASH", "PREFIX", "1", PREFIX,
        "SCHEMA",
        "id", "TAG",
        "title", "TEXT",
        "embedding", "VECTOR", "FLAT", "6",
        "TYPE", "FLOAT32", "DIM", str(DIM), "DISTANCE_METRIC", "COSINE",
    )

    for _ in range(120):
        info = client.execute_command("FT.INFO", INDEX_NAME)
        d = dict(zip(info[0::2], info[1::2]))
        if d.get(b"state") == b"ready" and int(d.get(b"num_docs", 0)) >= n:
            break
        time.sleep(0.5)
    else:
        print("warning: index did not report ready", file=sys.stderr)

    d = dict(zip(*[iter(client.execute_command("FT.INFO", INDEX_NAME))] * 2))
    print("index %r ready: num_docs=%s" % (INDEX_NAME, d.get(b"num_docs")))

    # Make sure everything survives a restart.
    client.save()


if __name__ == "__main__":
    main()
EOF
python3 /app/load_articles.py

11  $ python3 -c "import numpy; print(numpy.__version__)" 2>&1 | head -2

12  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge-base articles stored in Valkey.

Usage:  python3 /app/search.py < query.json

`query.json` is a JSON list of 64 floats (the query embedding). The script
prints the ids of the 5 most similar articles, one per line, most similar
first.

All similarity computation happens *inside* Valkey: this script only sends the
query vector to the server and runs an FT.SEARCH KNN query against the
`articles` vector index. It never reads /data/articles.jsonl and never
computes cosine similarities itself.
"""
import json
import struct
import sys

import valkey

INDEX_NAME = "articles"
VECTOR_FIELD = "embedding"
DIM = 64
TOP_K = 5


def main():
    raw = sys.stdin.read()
    query = json.loads(raw)
    if not isinstance(query, list) or len(query) != DIM:
        raise SystemExit(
            "query.json must be a JSON list of %d floats, got %r" % (DIM, type(query))
        )
    query_blob = struct.pack("<%df" % DIM, *[float(x) for x in query])

    client = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)

    # KNN search executed server-side by the valkey-search module.
    reply = client.execute_command(
        "FT.SEARCH", INDEX_NAME,
        "*=>[KNN %d @%s $vec AS score]" % (TOP_K, VECTOR_FIELD),
        "PARAMS", "2", "vec", query_blob,
        "RETURN", "1", "id",
        "SORTBY", "score",
        "DIALECT", "2",
    )

    # reply = [count, key1, [field, value, ...], key2, [field, value, ...], ...]
    count = reply[0]
    docs = reply[1:]
    for i in range(0, len(docs), 2):
        fields = docs[i + 1]
        fmap = dict(zip(fields[0::2], fields[1::2]))
        print(fmap["id"])


if __name__ == "__main__":
    main()
EOF
python3 -c "
import json,struct,random
a=json.loads(open('/data/articles.jsonl').readline())
json.dump(a['embedding'], open('/tmp/q.json','w'))
"
python3 /app/search.py < /tmp/q.json

13  $ cd /app; python3 - <<'EOF'
import json, struct, valkey
r = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
a = json.loads(open('/data/articles.jsonl').readline())
q = struct.pack('<64f', *a['embedding'])
tests = [
  ['FT.SEARCH','articles','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',q,'RETURN','1','id','SORTBY','score','DIALECT','2'],
  ['FT.SEARCH','articles','*=>[KNN 5 @embedding $vec]','PARAMS','2','vec',q,'RETURN','1','id','DIALECT','2'],
  ['FT.SEARCH','articles','*=>[KNN 5 @embedding $vec AS s]','PARAMS','2','vec',q,'DIALECT','2'],
  ['FT.SEARCH','articles','*=>[KNN 5 @embedding $vec]','PARAMS','2','vec',q,'DIALECT','2'],
]
for t in tests:
    try:
        res = r.execute_command(*t)
        print('OK', t[2], '|', res[0], '|', res[1:7])
    except Exception as e:
        print('ERR', t[2], e)
EOF

14  $ cd /app; python3 - <<'EOF'
import json, struct, valkey, numpy as np
r = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=True)
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts],dtype=np.float64)
Mn=M/np.linalg.norm(M,axis=1,keepdims=True)
def valkey_top(q,k=5):
    qb=struct.pack('<64f',*[float(x) for x in q])
    res=r.execute_command('FT.SEARCH','articles','*=>[KNN %d @embedding $vec]'%k,'PARAMS','2','vec',qb,'RETURN','1','id','DIALECT','2')
    out=[]
    for i in range(1,len(res),2):
        f=res[i+1]; out.append(dict(zip(f[0::2],f[1::2]))['id'])
    return out
def numpy_top(q,k=5):
    q=np.array(q,dtype=np.float64); q=q/np.linalg.norm(q)
    s=Mn@q
    return [ids[i] for i in np.argsort(-s)[:k]]

import random
random.seed(1)
ok=0
for t in range(20):
    if t<5:
        q=random.choice(arts)['embedding']  # self query
    else:
        q=[random.gauss(0,1) for _ in range(64)]
    v=valkey_top(q); n=numpy_top(q)
    same = v==n
    ok+=same
    print(t, 'MATCH' if same else 'DIFF', v[:3], n[:3])
print('exact-set agreement', ok, '/20')
# also verify distances returned via a run WITH returned score using alias not named score
qb=struct.pack('<64f',*arts[0]['embedding'])
res=r.execute_command('FT.SEARCH','articles','*=>[KNN 5 @embedding $vec AS dist]','PARAMS','2','vec',qb,'DIALECT','2','RETURN','1','id')
print(res[:7])
EOF

15  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge-base articles stored in Valkey.

Usage:  python3 /app/search.py < query.json

`query.json` is a JSON list of 64 floats (the query embedding). The script
prints the ids of the 5 articles most similar to the query, one per line, most
similar first.

All similarity computation happens *inside* Valkey: this script only sends the
query vector to the server and runs an FT.SEARCH KNN query against the
`articles` vector index (which was built by /app/load_articles.py). It never
reads /data/articles.jsonl and never computes cosine similarities itself.
"""
import json
import struct
import sys

import valkey

INDEX_NAME = "articles"
VECTOR_FIELD = "embedding"
DIM = 64
TOP_K = 5


def main():
    query = json.loads(sys.stdin.read())
    if not isinstance(query, list) or len(query) != DIM:
        raise SystemExit(
            "query.json must be a JSON list of %d floats" % DIM
        )
    query_blob = struct.pack("<%df" % DIM, *[float(x) for x in query])

    client = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=True)

    # KNN search executed server-side by the valkey-search module; results come
    # back already sorted by (cosine) distance, closest first.
    reply = client.execute_command(
        "FT.SEARCH", INDEX_NAME,
        "*=>[KNN %d @%s $vec]" % (TOP_K, VECTOR_FIELD),
        "PARAMS", "2", "vec", query_blob,
        "RETURN", "1", "id",
        "DIALECT", "2",
    )

    # reply = [count, key1, [field, value, ...], key2, [field, value, ...], ...]
    docs = reply[1:]
    for i in range(0, len(docs), 2):
        fields = docs[i + 1]
        fmap = dict(zip(fields[0::2], fields[1::2]))
        print(fmap["id"])


if __name__ == "__main__":
    main()
EOF
echo "=== run with self-query ==="
python3 -c "import json;json.dump(json.loads(open('/data/articles.jsonl').readline())['embedding'],open('/tmp/q.json','w'))"
python3 /app/search.py < /tmp/q.json
echo "=== output line count ==="
python3 /app/search.py < /tmp/q.json | wc -l

16  $ valkey-cli save; sleep 1
valkey-cli shutdown nosave 2>/dev/null; sleep 2
valkey-server /etc/valkey/valkey.conf
sleep 3
valkey-cli ping; valkey-cli dbsize; valkey-cli ft._list
valkey-cli ft.info articles | head -12
echo "=== search after restart ==="
python3 /app/search.py < /tmp/q.json

17  $ valkey-cli ft.dropindex idx
valkey-cli ft._list
valkey-cli dbsize
echo "=== final smoke test ==="
python3 -c "import json,random;random.seed(7);json.dump([random.gauss(0,1) for _ in range(64)],open('/tmp/q2.json','w'))"
python3 /app/search.py < /tmp/q2.json
echo "--- verifying against numpy ---"
python3 - <<'EOF'
import json,struct,valkey,numpy as np
r=valkey.Valkey(host='127.0.0.1',port=6379,decode_responses=True)
q=json.load(open('/tmp/q2.json'))
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts]);M/=np.linalg.norm(M,axis=1,keepdims=True)
qn=np.array(q);qn/=np.linalg.norm(qn)
print([ids[i] for i in np.argsort(-(M@qn))[:5]])
EOF
ls -la /app

18  $ valkey-cli save; valkey-cli info keyspace; valkey-cli ft.info articles | python3 -c "import sys;d=sys.stdin.read().split();print('index present, num_docs=',d[d.index('num_docs')+1])"

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ head -c 600 /data/articles.jsonl; echo; echo "---"; wc -l /data/articles.jsonl; echo "---"; valkey-cli -v 2>/dev/null || redis-cli -v

2  $ valkey-cli ping; valkey-cli module list; valkey-cli info server | head -20

3  $ valkey-cli FT._LIST; valkey-cli FT.CREATE foo ON HASH PREFIX 1 zz: SCHEMA t TEXT; echo "exit=$?"; valkey-cli FT.INFO foo

4  $ valkey-cli FT.DROPINDEX foo; valkey-cli FT._LIST; grep -n -i "vector\|bind\|module" /etc/valkey/valkey.conf | head -40

5  $ python3 -c "import numpy; print(numpy.__version__)"; python3 -V; ls /app 2>/dev/null; which redis-cli valkey-cli

6  $ cat /app/catalog.py; echo "---"; python3 -c "import valkey; print(valkey.__version__)" 2>&1; python3 -c "import redis; print(redis.__version__)" 2>&1

7  $ grep -n "VECTOR\|KNN\|query language\|cosine" -i /usr/share/doc/valkey* -r 2>/dev/null | head; ls /usr/share/doc/ | head -30

8  $ python3 -c "
import json
with open('/data/articles.jsonl') as f:
    line=f.readline()
o=json.loads(line)
print(o['id'], o['title'])
print(len(o['embedding']), type(o['embedding'][0]))
"; valkey-cli dbsize

9  $ python3 - <<'EOF'
import urllib.request
url="https://raw.githubusercontent.com/valkey-io/valkey-search/main/docs/topics/search-data-formats.md"
t=urllib.request.urlopen(url).read().decode()
i=t.find('# Vector Fields')
print(t[i:])
EOF

10  $ cat > /app/load.py <<'EOF'
"""Load articles from /data/articles.jsonl into Valkey and build the vector index."""
import json
import numpy as np
import redis

SRC = "/data/articles.jsonl"
INDEX = "articles_idx"
PREFIX = "article:"

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)

# 1. Create the index (HASH keys, 64-dim FLOAT32 vectors, cosine distance).
try:
    r.execute_command("FT.DROPINDEX", INDEX)
except redis.ResponseError:
    pass

r.execute_command(
    "FT.CREATE", INDEX,
    "ON", "HASH",
    "PREFIX", "1", PREFIX,
    "SCHEMA",
    "id", "TAG",
    "title", "TEXT",
    "embedding", "VECTOR", "HNSW", "6",
    "TYPE", "FLOAT32",
    "DIM", "64",
    "DISTANCE_METRIC", "COSINE",
)
print("index created")

# 2. Load all articles as hashes (mutations after FT.CREATE are auto-indexed).
pipe = r.pipeline(transaction=False)
n = 0
with open(SRC) as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        art = json.loads(line)
        vec = np.asarray(art["embedding"], dtype=np.float32).tobytes()
        key = PREFIX + art["id"]
        pipe.hset(key, mapping={"id": art["id"], "title": art["title"], "embedding": vec})
        n += 1
        if n % 200 == 0:
            pipe.execute()
pipe.execute()
print("loaded", n, "articles")

# 3. Wait for the index to finish ingesting.
for _ in range(100):
    info = r.execute_command("FT.INFO", INDEX)
    info = dict(zip(info[::2], info[1::2]))
    if int(info[b"num_docs"]) >= n and int(info[b"backfill_complete_percent"] and 1 or 1):
        if int(info[b"num_docs"]) >= n:
            break
    import time
    time.sleep(0.2)
info = r.execute_command("FT.INFO", INDEX)
info = dict(zip(info[::2], info[1::2]))
print("num_docs:", info[b"num_docs"], "state:", info[b"state"])
EOF
python3 /app/load.py

11  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge base.

Reads one query embedding (JSON list of 64 floats) from stdin and prints the
ids of the 5 most similar articles, one per line, most similar first.

All similarity work is performed inside Valkey by the valkey-search module;
this script only encodes the query vector and forwards it to FT.SEARCH.
"""
import json
import struct
import sys

import redis

INDEX = "articles_idx"
VECTOR_FIELD = "embedding"
DIM = 64
K = 5


def main() -> None:
    query = json.load(sys.stdin)
    if len(query) != DIM:
        raise SystemExit(f"expected a {DIM}-dimension query vector, got {len(query)}")
    blob = struct.pack(f"<{DIM}f", *(float(x) for x in query))

    client = redis.Redis(host="127.0.0.1", port=6379)

    result = client.execute_command(
        "FT.SEARCH", INDEX,
        f"*=>[KNN {K} @{VECTOR_FIELD} $vec AS score]",
        "PARAMS", "2", "vec", blob,
        "RETURN", "1", "id",
        "SORTBY", "score",
        "DIALECT", "2",
        "LIMIT", "0", str(K),
    )

    # Response: [total, key1, [field, value, ...], key2, [...], ...]
    for i in range(1, len(result), 2):
        fields = result[i + 1]
        fields = dict(zip(fields[::2], fields[1::2]))
        print(fields[b"id"].decode())


if __name__ == "__main__":
    main()
EOF
python3 -c "
import json,random
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
q=arts[7]['embedding']
open('/tmp/q.json','w').write(json.dumps(q))
print('query for', arts[7]['id'])
"
python3 /app/search.py < /tmp/q.json

12  $ cd /app && python3 -c "
import struct,redis,json
q=json.load(open('/tmp/q.json'))
blob=struct.pack('<64f',*q)
r=redis.Redis()
print(r.execute_command('FT.SEARCH','articles_idx','*=>[KNN 5 @embedding \$vec AS score]','PARAMS','2','vec',blob,'RETURN','1','id','SORTBY','score','DIALECT','2','LIMIT','0','5'))
" 2>&1 | tail -20

13  $ cd /app && python3 -c "
import struct,redis,json
q=json.load(open('/tmp/q.json'))
blob=struct.pack('<64f',*q)
r=redis.Redis()
print(r.execute_command('FT.SEARCH','articles_idx','*=>[KNN 5 @embedding \$vec AS score]','PARAMS','2','vec',blob,'RETURN','2','id','score','DIALECT','2','LIMIT','0','5'))
" 2>&1 | tail -20

14  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge-base articles stored in Valkey.

Usage:  python3 /app/search.py < query.json

query.json contains a single query embedding: a JSON list of 64 floats.
The script prints the ids of the 5 most similar articles, one per line,
most similar first.

The similarity search itself runs entirely inside Valkey (the valkey-search
module's FT.SEARCH KNN query over a cosine-distance vector index); this script
only encodes the query vector and forwards it.
"""
import json
import struct
import sys

import redis

INDEX = "articles_idx"
VECTOR_FIELD = "embedding"
DIM = 64
K = 5


def main() -> None:
    query = json.load(sys.stdin)
    if len(query) != DIM:
        raise SystemExit(f"expected a {DIM}-dimension query vector, got {len(query)}")

    # FLOAT32 little-endian blob, matching the index's declared type.
    blob = struct.pack(f"<{DIM}f", *(float(x) for x in query))

    client = redis.Redis(host="127.0.0.1", port=6379)

    # Vector results are returned in ascending distance order by default.
    result = client.execute_command(
        "FT.SEARCH", INDEX,
        f"*=>[KNN {K} @{VECTOR_FIELD} $vec AS score]",
        "PARAMS", "2", "vec", blob,
        "RETURN", "1", "id",
        "DIALECT", "2",
        "LIMIT", "0", str(K),
    )

    # Response shape: [total, key1, [field, value, ...], key2, [...], ...]
    for i in range(1, len(result), 2):
        fields = result[i + 1]
        fields = dict(zip(fields[::2], fields[1::2]))
        print(fields[b"id"].decode())


if __name__ == "__main__":
    main()
EOF
python3 /app/search.py < /tmp/q.json

15  $ cd /app && python3 - <<'EOF'
import json, subprocess, numpy as np, random

arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts], dtype=np.float64)
Mn=M/np.linalg.norm(M,axis=1,keepdims=True)

random.seed(0)
bad=0
for trial in range(25):
    if trial<10:
        q=M[trial]                      # exact article vectors
    else:
        q=np.random.randn(64)           # random queries
    qn=q/np.linalg.norm(q)
    sims=Mn@qn
    top=[ids[i] for i in np.argsort(-sims)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(list(q)),capture_output=True,text=True)
    got=out.stdout.split()
    if got!=top:
        bad+=1
        print("MISMATCH", trial, got, top, out.stderr)
print("mismatches:", bad, "/ 25")
EOF

16  $ cd /app && python3 - <<'EOF'
import json, struct, redis, numpy as np
r=redis.Redis()
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts],dtype=np.float64)
Mn=M/np.linalg.norm(M,axis=1,keepdims=True)
np.random.seed(0)
for _ in range(10): q=np.random.randn(64)  # advance rng to trial 10+
# regenerate trial 23 exactly as before
random_state=np.random.RandomState(0)
EOF
python3 - <<'EOF'
import numpy as np
rng=np.random.RandomState(0)
for t in range(10): rng.randn(64)
for t in range(10,25):
    q=rng.randn(64)
    if t==23:
        print("trial23 q[:3]", q[:3])
EOF

17  $ cd /app && python3 - <<'EOF'
import json, numpy as np
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts],dtype=np.float64)
Mn=M/np.linalg.norm(M,axis=1,keepdims=True)
rng=np.random.RandomState(0)
qs=[rng.randn(64) for _ in range(15)]
for idx in [0,3,4,7,13]:
    q=qs[idx]; qn=q/np.linalg.norm(q)
    sims=Mn@qn
    order=np.argsort(-sims)
    top5=sims[order[:5]]
    print(f"trial{10+idx}: top5 sims", np.round(top5,4), " 6th:", round(sims[order[5]],4))
EOF

18  $ cd /app && sed -i 's/"VECTOR", "HNSW", "6",/"VECTOR", "FLAT", "6",/' load.py && grep -n 'VECTOR' load.py && python3 load.py

19  $ cd /app && python3 - <<'EOF'
import json, subprocess, numpy as np

arts=[json.loads(l) for l in open('/data/articles.jsonl')]
ids=[a['id'] for a in arts]
M=np.array([a['embedding'] for a in arts], dtype=np.float64)
Mn=M/np.linalg.norm(M,axis=1,keepdims=True)

rng=np.random.RandomState(0)
bad=0; trials=60
for trial in range(trials):
    if trial<10:
        q=M[trial]
    else:
        q=rng.randn(64)
    qn=q/np.linalg.norm(q)
    sims=Mn@qn
    top=[ids[i] for i in np.argsort(-sims)[:5]]
    out=subprocess.run(['python3','/app/search.py'],input=json.dumps(list(q)),capture_output=True,text=True)
    got=out.stdout.split()
    if got!=top:
        bad+=1
        print("MISMATCH", trial, got, top)
print(f"mismatches: {bad}/{trials}")
EOF

20  $ cd / && echo "--- search.py references ---" && grep -n "articles.jsonl\|numpy\|cosine\|similar" /app/search.py || echo "(no forbidden references)"
echo "--- run from another cwd, with whitespace/newlines in input ---"
printf '  [%s]\n' "$(python3 -c "import json;print(','.join(map(str,json.loads(open('/data/articles.jsonl').readline())['embedding'])))")" > /tmp/q2.json
python3 /app/search.py < /tmp/q2.json
echo "--- Valkey state ---"
valkey-cli FT._LIST
valkey-cli DBSIZE
valkey-cli --raw FT.INFO articles_idx | sed -n '1,40p'

21  $ valkey-cli config get save; valkey-cli config get appendonly; echo "--- force a save so state persists ---"; valkey-cli BGSAVE; sleep 2; ls -la /var/lib/valkey/ 2>/dev/null; valkey-cli info persistence | grep -i "rdb_last_bgsave_status\|rdb_bgsave_in_progress"

22  $ cd /tmp && python3 -c "
import json
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
json.dump(arts[500]['embedding'], open('query.json','w'))
print('expect top hit:', arts[500]['id'])
"
python3 /app/search.py < /tmp/query.json

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ ls -la /data/ && head -c 500 /data/articles.jsonl && echo && wc -l /data/articles.jsonl

2  $ valkey-cli -h 127.0.0.1 -p 6379 ping; valkey-cli info server | head -20; valkey-cli module list

3  $ valkey-cli FT._LIST; valkey-cli FT.CREATE --help 2>&1 | head -5; valkey-cli MODULE LIST

4  $ python3 -c "import valkey" 2>&1; python3 -c "import redis; print(redis.__version__)" 2>&1; pip list 2>/dev/null | grep -i -E "valkey|redis|numpy"

5  $ valkey-cli HSET test:1 t "hello" v "$(python3 -c "import numpy as np;print(np.array([1.0,0,0],dtype=np.float32).tobytes().hex())" | xxd -r -p | base64)" 2>&1 | head; echo "---"; valkey-cli FT.CREATE testidx ON HASH PREFIX 1 test: SCHEMA t TEXT v VECTOR FLAT 6 TYPE FLOAT32 DIM 3 DISTANCE_METRIC COSINE 2>&1; valkey-cli FT.INFO testidx 2>&1 | head -40

6  $ cat /etc/valkey/valkey.conf | grep -v "^#" | grep -v "^$"

7  $ cat > /tmp/load.py <<'EOF'
import json, struct
import valkey

r = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=False)

# clean previous
try:
    r.execute_command("FT.DROPINDEX", "articles_idx")
except Exception as e:
    print("drop:", e)
for k in r.scan_iter("art:*"):
    r.delete(k)

DIM = 64
pipe = r.pipeline(transaction=False)
n = 0
with open("/data/articles.jsonl") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        obj = json.loads(line)
        emb = obj["embedding"]
        assert len(emb) == DIM, (obj["id"], len(emb))
        blob = struct.pack(f"<{DIM}f", *emb)
        pipe.hset(f"art:{obj['id']}", mapping={
            "title": obj["title"],
            "embedding": blob,
        })
        n += 1
        if n % 500 == 0:
            pipe.execute()
            pipe = r.pipeline(transaction=False)
pipe.execute()
print("loaded", n)

r.execute_command(
    "FT.CREATE", "articles_idx", "ON", "HASH", "PREFIX", 1, "art:",
    "SCHEMA",
    "title", "TEXT",
    "embedding", "VECTOR", "FLAT", 6,
    "TYPE", "FLOAT32", "DIM", str(DIM), "DISTANCE_METRIC", "COSINE",
)
info = r.execute_command("FT.INFO", "articles_idx")
d = dict(zip(info[::2], info[1::2]))
print("num_docs:", d.get(b"num_docs"))
EOF
python3 /tmp/load.py

8  $ sleep 2; valkey-cli FT.INFO articles_idx | sed -n '1,60p'; echo "=== keys ==="; valkey-cli DBSIZE

9  $ mkdir -p /app
cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over knowledge-base articles stored in Valkey.

Usage:
    python3 /app/search.py < query.json

`query.json` contains a single query embedding: a JSON list of 64 floats.

The 5 most similar articles (by cosine similarity) are printed, one `id`
per line, most similar first.

All similarity computation happens inside Valkey via the RediSearch/Valkey
Search module (FT.SEARCH ... KNN).  This script never reads the source
articles file and never computes similarities itself.
"""

import json
import struct
import sys

import valkey

HOST = "127.0.0.1"
PORT = 6379
INDEX = "articles_idx"
DIM = 64
K = 5


def main() -> int:
    raw = sys.stdin.read().strip()
    if not raw:
        print("error: empty query", file=sys.stderr)
        return 1

    query = json.loads(raw)
    if not isinstance(query, list) or len(query) != DIM:
        print(f"error: expected a JSON list of {DIM} floats", file=sys.stderr)
        return 1

    # Pack the query vector as little-endian float32 bytes, exactly the
    # representation stored in the index.
    blob = struct.pack(f"<{DIM}f", *[float(x) for x in query])

    client = valkey.Valkey(host=HOST, port=PORT, decode_responses=True)

    # KNN vector search executed by Valkey Search.
    res = client.execute_command(
        "FT.SEARCH", INDEX,
        "(*)=>[KNN {} @embedding $vec AS score]".format(K),
        "PARAMS", 2, "vec", blob,
        "RETURN", 1, "score",
        "SORTBY", "score", "ASC",
        "DIALECT", 2,
    )

    # res = [total, key1, [field, value, ...], key2, [...], ...]
    total = res[0]
    out = []
    for i in range(1, len(res), 2):
        key = res[i]
        fields = res[i + 1]
        doc_id = key.split(":", 1)[1] if ":" in key else key
        out.append(doc_id)

    for doc_id in out[:K]:
        print(doc_id)

    return 0


if __name__ == "__main__":
    sys.exit(main())
EOF
echo "written"

10  $ python3 -c "
import json
with open('/data/articles.jsonl') as f:
    for i,line in enumerate(f):
        o=json.loads(line)
        if i==123:
            print(json.dumps(o['embedding']))
            break
" > /tmp/q.json
python3 /app/search.py < /tmp/q.json

11  $ cd /tmp && python3 -c "
import struct, valkey
r=valkey.Valkey(host='127.0.0.1',port=6379,decode_responses=False)
import json
q=json.load(open('/tmp/q.json'))
blob=struct.pack('<64f',*q)
try:
    print(r.execute_command('FT.SEARCH','articles_idx','(*)=>[KNN 5 @embedding \$vec AS score]','PARAMS',2,'vec',blob,'SORTBY','score','ASC','DIALECT',2))
except Exception as e:
    print('ERR1',e)
try:
    print(r.execute_command('FT.SEARCH','articles_idx','(*)=>[KNN 5 @embedding \$vec AS score]','PARAMS',2,'vec',blob,'DIALECT',2))
except Exception as e:
    print('ERR2',e)
"

12  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over knowledge-base articles stored in Valkey.

Usage:
    python3 /app/search.py < query.json

`query.json` contains a single query embedding: a JSON list of 64 floats.

The 5 most similar articles (by cosine similarity) are printed, one `id`
per line, most similar first.

All similarity computation is performed *inside Valkey* by the Valkey
Search module (FT.SEARCH ... KNN).  This script never reads the source
articles file and never computes similarities itself -- it only sends the
query vector to Valkey and prints the ids that Valkey returns, already
ranked by the engine.
"""

import json
import struct
import sys

import valkey

HOST = "127.0.0.1"
PORT = 6379
INDEX = "articles_idx"
DIM = 64
K = 5


def main() -> int:
    raw = sys.stdin.read().strip()
    if not raw:
        print("error: empty query", file=sys.stderr)
        return 1

    try:
        query = json.loads(raw)
    except json.JSONDecodeError as exc:
        print(f"error: invalid JSON query: {exc}", file=sys.stderr)
        return 1

    if not isinstance(query, list) or len(query) != DIM:
        print(f"error: expected a JSON list of {DIM} floats", file=sys.stderr)
        return 1

    # Pack the query vector as little-endian float32 bytes, exactly the
    # representation stored in the index (FLOAT32).
    blob = struct.pack(f"<{DIM}f", *[float(x) for x in query])

    client = valkey.Valkey(host=HOST, port=PORT, decode_responses=True)

    # KNN vector search executed by Valkey Search.  Valkey returns the
    # matches ordered by cosine distance, ascending (i.e. most similar
    # first).  We ask it to return only the distance it computed.
    res = client.execute_command(
        "FT.SEARCH", INDEX,
        "(*)=>[KNN {} @embedding $vec AS dist]".format(K),
        "PARAMS", 2, "vec", blob,
        "RETURN", 1, "dist",
        "DIALECT", 2,
    )

    # Response shape: [total, key1, [field, value, ...], key2, [...], ...]
    ranked = []
    for i in range(1, len(res), 2):
        key = res[i]
        fields = res[i + 1] if i + 1 < len(res) else []
        fmap = dict(zip(fields[::2], fields[1::2]))
        doc_id = key.split(":", 1)[1] if ":" in key else key
        ranked.append((float(fmap.get("dist", 0.0)), doc_id))

    # Belt-and-suspenders: order by the distances Valkey computed.
    ranked.sort(key=lambda t: t[0])

    for _dist, doc_id in ranked[:K]:
        print(doc_id)

    return 0


if __name__ == "__main__":
    sys.exit(main())
EOF
python3 /app/search.py < /tmp/q.json

13  $ cat > /tmp/verify.py <<'EOF'
import json, subprocess, numpy as np, random

data = [json.loads(l) for l in open('/data/articles.jsonl')]
E = np.array([d['embedding'] for d in data], dtype=np.float64)
ids = [d['id'] for d in data]
En = E / np.linalg.norm(E, axis=1, keepdims=True)

def brute(q):
    q = np.array(q, dtype=np.float64)
    qn = q / np.linalg.norm(q)
    sims = En @ qn
    order = np.argsort(-sims)[:5]
    return [ids[i] for i in order], [sims[i] for i in order]

random.seed(0)
fails = 0
for trial in range(8):
    q = random.choice(data)['embedding']
    expected, sims = brute(q)
    out = subprocess.run(['python3','/app/search.py'], input=json.dumps(q),
                         capture_output=True, text=True)
    got = out.stdout.split()
    ok = got == expected
    if not ok:
        fails += 1
    print(trial, "OK" if ok else "MISMATCH", got, expected)
print("fails:", fails)

# also test random (non-corpus) queries
for trial in range(5):
    q = list(np.random.randn(64))
    expected, sims = brute(q)
    out = subprocess.run(['python3','/app/search.py'], input=json.dumps(q),
                         capture_output=True, text=True)
    got = out.stdout.split()
    print("rand", trial, "OK" if got == expected else "MISMATCH")
EOF
python3 /tmp/verify.py

14  $ mv /data/articles.jsonl /data/articles.jsonl.bak
python3 /app/search.py < /tmp/q.json; echo "exit=$?"
mv /data/articles.jsonl.bak /data/articles.jsonl
ls -la /data/

15  $ grep -iE "^(save|appendonly|dir|dbfilename)" /etc/valkey/valkey.conf; echo "---runtime---"; valkey-cli config get save; valkey-cli config get appendonly; valkey-cli config get dir

16  $ valkey-cli SAVE; ls -la /var/lib/valkey/; valkey-cli DBSIZE; valkey-cli FT.INFO articles_idx | grep -A1 num_docs | head -4

17  $ printf '\n  [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 0.1, 0.2, 0.3, 0.4]\n' | python3 /app/search.py; echo "exit=$?"

18  $ python3 -c "
import json, numpy as np
q=np.array([0.1,0.2,0.3,0.4,0.5,0.6,0.7,0.8,0.9,1.0]*6+[0.1,0.2,0.3,0.4])
data=[json.loads(l) for l in open('/data/articles.jsonl')]
E=np.array([d['embedding'] for d in data]); ids=[d['id'] for d in data]
sims=(E/np.linalg.norm(E,axis=1,keepdims=True))@(q/np.linalg.norm(q))
print([ids[i] for i in np.argsort(-sims)[:5]])
"

{
 "dbsize": 1001,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ head -c 600 /data/articles.jsonl; echo; wc -l /data/articles.jsonl

2  $ valkey-cli PING; valkey-cli INFO server | head -25; valkey-cli MODULE LIST

3  $ python3 --version; pip3 list 2>/dev/null | grep -Ei 'valkey|redis|numpy' ; which python3

4  $ valkey-cli FT.CREATE tmp:idx ON JSON PREFIX 1 "tmp:" SCHEMA $.embedding AS embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE

5  $ cd /tmp && python3 - <<'EOF'
import struct, numpy as np, json
from valkey import Valkey
from valkey.commands.search.query import Query
r = Valkey(host='127.0.0.1', port=6379, decode_responses=True)
rng = np.random.default_rng(0)
v1 = rng.normal(size=64); v2 = v1 * 0.5 + 0.1
r.json().set('tmp:doc1', '$', json.dumps({"id":"t1","embedding":[float(x) for x in v1]}))
r.json().set('tmp:doc2', '$', json.dumps({"id":"t2","embedding":[float(x) for x in v2]}))
q = Query('*=>[KNN 2 @embedding $vec AS score]').sort_by('score').dialect(2)
res = r.ft('tmp:idx').search(q, query_params={'vec': np.array(v1, dtype=np.float32).tobytes()})
print([ (d.id, d.score) for d in res.docs ])
EOF

6  $ cd /tmp && python3 - <<'EOF'
import numpy as np, json
from valkey import Valkey
from valkey.commands.search.query import Query
r = Valkey(host='127.0.0.1', port=6379, decode_responses=True)
v1 = np.arange(64, dtype=np.float32)/64
v2 = np.ones(64, dtype=np.float32)/8
print('cos(v1,v1)', float(np.dot(v1,v1)/np.linalg.norm(v1)/np.linalg.norm(v1)))
print('cos(v1,v2)', float(np.dot(v1,v2)/np.linalg.norm(v1)/np.linalg.norm(v2)))
r.json().set('tmp:doc1', '$', json.dumps({"id":"t1","embedding":[float(x) for x in v1]}))
r.json().set('tmp:doc2', '$', json.dumps({"id":"t2","embedding":[float(x) for x in v2]}))
tests = {
 'knn-only': Query('*=>[KNN 2 @embedding $vec AS score]').dialect(2),
 'knn-sortby': Query('*=>[KNN 2 @embedding $vec AS score]').sort_by('score').dialect(2),
 'knn-sortby-asc': Query('*=>[KNN 2 @embedding $vec AS score]').sort_by('score', asc=True).dialect(2),
 'sortby-alias-only': Query('*=>[KNN 2 @embedding $vec AS score]').sort_by('__embedding_score').dialect(2),
}
for name, q in tests.items():
    try:
        res = r.ft('tmp:idx').search(q, query_params={'vec': v1.tobytes()})
        print(name, '->', [(d.id, d.score) for d in res.docs])
    except Exception as e:
        print(name, 'ERR:', e)
EOF

7  $ valkey-cli FT.INFO tmp:idx | sed -n '1,40p' | tr -d '\n' | head -c 1200; echo; echo ---; valkey-cli JSON.GET tmp:doc1 '$.id'; valkey-cli FT.SEARCH tmp:idx '*' NOCONTENT DIALECT 2

8  $ echo '--- bare star:'; valkey-cli FT.SEARCH tmp:idx '*'; echo '--- star dialect2:'; valkey-cli FT.SEARCH tmp:idx '*' DIALECT 2; echo '--- knn:'; valkey-cli FT.SEARCH tmp:idx '*=>[KNN 2 @embedding $vec AS score]' PARAMS 2 vec "$(python3 -c "import numpy as np;import sys;sys.stdout.write(np.ones(64,dtype=np.float32).tobytes().decode('latin1'))" | base64 -w0 | base64 -d)" DIALECT 2 2>&1 | head -5

9  $ for q in 'hello' 'hello*' '*' '(@embedding:{a})' '0'; do echo "--- [$q]"; valkey-cli FT.SEARCH tmp:idx "$q" NOCONTENT DIALECT 2 2>&1 | head -3; done; echo '--- EXPLAINCLI:'; valkey-cli FT.EXPLAINCLI tmp:idx '*' DIALECT 2 2>&1 | head -3

10  $ cd /tmp && python3 - <<'EOF'
import numpy as np
from valkey import Valkey
from valkey.commands.search.query import Query
r = Valkey(host='127.0.0.1', port=6379, decode_responses=True)
v1 = np.arange(64, dtype=np.float32)/64
tests = {
 'knn-only': Query('*=>[KNN 2 @embedding $vec AS score]').dialect(2),
 'filter-first': Query('@embedding:{*}=>[KNN 2 @embedding $vec AS score]').dialect(2),
}
for name, q in tests.items():
    try:
        res = r.ft('tmp:idx').search(q, query_params={'vec': v1.tobytes()})
        print(name, '->', [(d.id, d.score) for d in res.docs])
    except Exception as e:
        print(name, 'ERR:', type(e).__name__, e)
EOF

11  $ cd /tmp && python3 - <<'EOF'
import numpy as np
from valkey import Valkey
r = Valkey(host='127.0.0.1', port=6379, decode_responses=False)
v1 = (np.arange(64, dtype=np.float32)/64).tobytes()
reply = r.execute_command('FT.SEARCH', 'tmp:idx', '*=>[KNN 2 @embedding $vec AS score]', 'PARAMS', 2, 'vec', v1, 'DIALECT', 2)
print('RAW REPLY:', reply)
EOF

12  $ python3 -c "
import numpy as np
open('/tmp/vec.bin','wb').write((np.arange(64,dtype=np.float32)/64).tobytes())
"; valkey-cli -x FT.SEARCH tmp:idx '*=>[KNN 2 @embedding $vec AS score]' PARAMS 2 vec DIALECT 2 < /tmp/vec.bin 2>&1 | head -20

13  $ valkey-cli JSON.GET tmp:doc1 '$'; echo; valkey-cli JSON.TYPE tmp:doc1 '$'; echo

14  $ cd /tmp && python3 - <<'EOF'
import numpy as np, json
from valkey import Valkey
from valkey.commands.search.query import Query
r = Valkey(host='127.0.0.1', port=6379, decode_responses=True)
v1 = (np.arange(64, dtype=np.float32)/64).astype(np.float64)
v2 = (np.ones(64, dtype=np.float32)/8).astype(np.float64)
r.json().set('tmp:doc1', '$', {"id":"t1","embedding":list(v1)})
r.json().set('tmp:doc2', '$', {"id":"t2","embedding":list(v2)})
print('type now:', r.json().type('tmp:doc1', '$'))
q = Query('*=>[KNN 2 @embedding $vec AS score]').dialect(2)
res = r.ft('tmp:idx').search(q, query_params={'vec': np.asarray(v1, dtype=np.float32).tobytes()})
print([(d.id, d.score) for d in res.docs])
EOF

15  $ valkey-cli FT.DROPINDEX tmp:idx; valkey-cli DEL tmp:doc1 tmp:doc2; mkdir -p /app; grep -E '^(save|appendonly|dir|dbfilename)' /etc/valkey/valkey.conf | grep -v '^#' | head

16  $ cat > /app/load_articles.py <<'EOF'
#!/usr/bin/env python3
"""One-time loader: ingest all KB articles into Valkey as JSON docs and
create the vector index used by /app/search.py.

Data model
----------
- Hash  kb:meta:<id>   -> {"id": <str>, "title": <str>}        (article metadata)
- JSON  kb:vec:<id>     -> {"embedding": [64 floats]}            (indexed vector part)

The metadata is split out into a small hash so FT.SEARCH replies stay tiny:
the query only needs the document id, not the full title + embedding payload.
The vector JSON doc is what the FT.SEARCH index matches on (prefix "kb:vec:").

Index uses the FLAT algorithm => exact (brute-force) KNN, so results are the
true top-5 by cosine similarity, with no recall loss from ANN approximation.
"""
import json
import sys

import numpy as np
from valkey import Valkey

ARTICLES_PATH = "/data/articles.jsonl"
IDX_NAME = "kb:articles"
KEY_PREFIX = "kb:vec:"
META_PREFIX = "kb:meta:"
DIM = 64
TOP_K = 5

SCHEMA = (
    f"FT.CREATE {IDX_NAME} ON JSON PREFIX 1 {KEY_PREFIX} "
    f"SCHEMA $.embedding AS embedding VECTOR FLAT 6 "
    f"TYPE FLOAT32 DIM {DIM} DISTANCE_METRIC COSINE"
)


def main() -> None:
    r = Valkey(host="127.0.0.1", port=6379, decode_responses=True)

    # Idempotency: if the index already exists with the expected doc count,
    # just verify and exit.
    info = r.ft(IDX_NAME).info()
    if info.get("num_docs") == 1000:
        print("Index already loaded:", info["num_docs"], "docs")
        return

    n = 0
    with open(ARTICLES_PATH, "r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            art = json.loads(line)
            aid = art["id"]
            emb = np.asarray(art["embedding"], dtype=np.float32)
            assert emb.shape == (DIM,), f"bad embedding for {aid}: {emb.shape}"
            r.json().set(KEY_PREFIX + aid, "$", {"embedding": emb.tolist()})
            r.hset(META_PREFIX + aid, mapping={"id": aid, "title": art["title"]})
            n += 1
    print(f"Loaded {n} articles")

    # Wait until the backfill has indexed every doc before declaring victory.
    for _ in range(600):
        info = r.ft(IDX_NAME).info()
        if info.get("num_docs") == n:
            break
        time.sleep(0.1)
    info = r.ft(IDX_NAME).info()
    print(f"num_docs={info.get('num_docs')}")


if __name__ == "__main__":
    main()
EOF
sed -i 's/^import json$/import json\nimport time/' /app/load_articles.py
python3 - <<'EOF'
import sys
sys.path.insert(0, '/app')
EOF
python3 -c "import ast; ast.parse(open('/app/load_articles.py').read()); print('syntax ok')"

17  $ cd /app && python3 - <<'EOF'
src = open('load_articles.py').read()
old = """    # Idempotency: if the index already exists with the expected doc count,
    # just verify and exit.
    info = r.ft(IDX_NAME).info()
    if info.get("num_docs") == 1000:
        print("Index already loaded:", info["num_docs"], "docs")
        return

    n = 0"""
new = """    # Idempotency: if the index already exists and is fully populated,
    # just verify and exit.
    try:
        info = r.ft(IDX_NAME).info()
        if info.get("num_docs") == 1000:
            print("Index already loaded:", info["num_docs"], "docs")
            return
        r.ft(IDX_NAME).dropindex(delete_documents=False)
    except Exception:
        pass  # index does not exist yet

    n = 0"""
assert old in src
open('load_articles.py', 'w').write(src.replace(old, new))
print('patched')
EOF
time python3 /app/load_articles.py

18  $ valkey-cli FT.INFO kb:articles | head -60; echo '--- DBSIZE:'; valkey-cli DBSIZE; echo '--- NUM DOCS via raw INFO:'; valkey-cli FT.INFO kb:articles | grep -A3 num_docs | head -5

19  $ cd /app && python3 - <<'EOF'
src = open('load_articles.py').read()
old = """    r.ft(IDX_NAME).dropindex(delete_documents=False)
    except Exception:
        pass  # index does not exist yet

    n = 0"""
new = """    r.ft(IDX_NAME).dropindex(delete_documents=False)
    except Exception:
        pass  # index does not exist yet

    # Create the vector index (exact FLAT / cosine).
    r.execute_command(*SCHEMA.split())
    print("Created index", IDX_NAME)

    n = 0"""
assert old in src
open('load_articles.py', 'w').write(src.replace(old, new))
print('patched')
EOF
valkey-cli FT.CREATE kb:articles ON JSON PREFIX 1 kb:vec: SCHEMA $.embedding AS embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE

20  $ for i in $(seq 1 30); do n=$(valkey-cli FT.INFO kb:articles | tr -d '\n' | grep -o 'num_docs[0-9]*' | head -1 | sed 's/num_docs//'); [ "$n" = "1000" ] && break; sleep 1; done; echo "num_docs=$n (waited ${i}s)"; valkey-cli FT.INFO kb:articles | tr -d '\n' | grep -o 'num_records[0-9]*'

21  $ cd /tmp && python3 - <<'EOF'
import json, numpy as np
from valkey import Valkey
from valkey.commands.search.query import Query
r = Valkey(host='127.0.0.1', port=6379, decode_responses=True)

# ground truth embeddings
embs, ids = [], []
with open('/data/articles.jsonl') as fh:
    for line in fh:
        a = json.loads(line); ids.append(a['id']); embs.append(a['embedding'])
E = np.asarray(embs, dtype=np.float64)
print('ids contain ":"?', any(':' in i for i in ids))

rng = np.random.default_rng(7)
for qi in rng.choice(len(ids), size=3, replace=False):
    qv = np.asarray(E[qi], dtype=np.float32)
    q = Query('*=>[KNN 5 @embedding $vec AS distance]').dialect(2)
    res = r.ft('kb:articles').search(q, query_params={'vec': qv.tobytes()})
    got = [d.id.split('kb:vec:',1)[1] for d in res.docs]
    scores = [getattr(d, 'score', None) for d in res.docs]
    payload_len = [len(d.payload or '') for d in res.docs]
    # numpy exact cosine top5
    sims = (E @ (qv.astype(np.float64))) / (np.linalg.norm(E,axis=1)*np.linalg.norm(qv.astype(np.float64)))
    truth = [ids[i] for i in np.argsort(-sims)[:5]]
    print(f'q={ids[qi]} got={got} scores={scores} payload_bytes={payload_len}')
    print(f'   truth={truth} match={got==truth}')
EOF

22  $ cd /tmp && python3 - <<'EOF'
import numpy as np
from valkey import Valkey
r = Valkey(host='127.0.0.1', port=6379, decode_responses=False)
qv = (np.arange(64, dtype=np.float32)/64).tobytes()
# without NOCONTENT
rep = r.execute_command('FT.SEARCH', 'kb:articles', '*=>[KNN 5 @embedding $vec AS distance]', 'PARAMS', 2, 'vec', qv, 'DIALECT', 2)
print('total:', rep[0])
print('num returned:', len(rep[1]))
print('raw first reply element type:', type(rep[2]))
print('repr first doc:', repr(rep[2]))
EOF

23  $ cd /tmp && python3 - <<'EOF'
import numpy as np
from valkey import Valkey
r = Valkey(host='127.0.0.1', port=6379, decode_responses=False)
qv = (np.arange(64, dtype=np.float32)/64).tobytes()
rep = r.execute_command('FT.SEARCH', 'kb:articles', '*=>[KNN 5 @embedding $vec AS distance]', 'PARAMS', 2, 'vec', qv, 'DIALECT', 2)
print('top-level len:', len(rep))
print('total:', rep[0])
body = rep[1]
print('body len:', len(body))
for i, el in enumerate(body):
    print(i, type(el).__name__, (el[:40] if isinstance(el, bytes) else ('attr pairs: ' + str([(x[0].decode(), x[1].decode()) for x in el]) if isinstance(el, list) and el and isinstance(el[0], list) else repr(el)[:80])))
# and with NOCONTENT for comparison
rep2 = r.execute_command('FT.SEARCH', 'kb:articles', '*=>[KNN 5 @embedding $vec AS distance]', 'PARAMS', 2, 'vec', qv, 'DIALECT', 2, 'NOCONTENT')
print('NOCONTENT total:', rep2[0], 'body:', [x.decode() if isinstance(x, bytes) else x for x in rep2[1]])
EOF

24  $ cd /tmp && python3 - <<'EOF'
import socket, numpy as np
qv = (np.arange(64, dtype=np.float32)/64).tobytes()
cmd = [b'FT.SEARCH', b'kb:articles', b'*=>[KNN 5 @embedding $vec AS distance]',
       b'PARAMS', b'2', b'vec', qv, b'DIALECT', b'2']
buf = b'*%d\r\n' % len(cmd)
for a in cmd:
    buf += b'$%d\r\n%s\r\n' % (len(a), a)
s = socket.create_connection(('127.0.0.1', 6379))
s.sendall(buf)
s.settimeout(2)
data = b''
try:
    while True:
        chunk = s.recv(65536)
        if not chunk: break
        data += chunk
        if b'\r\n' in data and len(data) > 100 and data.rstrip().endswith(b'\r\n'):
            # crude: stop after we have something substantial and it looks complete
            pass
except socket.timeout:
    pass
print(repr(data[:900]))
print('...')
print('total reply bytes:', len(data))
s.close()
EOF

25  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge-base articles stored in Valkey.

Usage:
    python3 /app/search.py < query.json

query.json holds a single query embedding: a JSON list of 64 floats
(a JSON object with an "embedding" key is also accepted).

Prints the ids of the 5 most similar articles (cosine similarity), most
similar first, one per line.

How it works
------------
All 1,000 articles live in Valkey as JSON documents under keys
`kb:vec:<id>` (field `$.embedding`), indexed by the `kb:articles` vector
index created with:

    FT.CREATE kb:articles ON JSON PREFIX 1 kb:vec: SCHEMA $.embedding
        AS embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE

This script sends the query vector to Valkey and runs a KNN query *inside*
the server:

    FT.SEARCH kb:articles '*=>[KNN 5 @embedding $vec AS distance]'
        PARAMS 2 vec <blob> DIALECT 2

The FLAT algorithm performs an exact brute-force scan, so the top-5 by
cosine distance (= 1 - cosine similarity) is the true top-5. Article
titles/metadata are kept separately in hashes `kb:meta:<id>`; this script
only needs the ids. Nothing is read from /data/articles.jsonl and no
similarities are computed locally.
"""

import json
import sys

import numpy as np
from valkey import Valkey

INDEX_NAME = "kb:articles"
KEY_PREFIX = "kb:vec:"
DIM = 64
K = 5
QUERY_TEMPLATE = f"*=>[KNN {K} @embedding $vec AS distance]"


def read_query_vector(stream) -> np.ndarray:
    """Read one 64-float embedding from JSON text and return it as float32."""
    obj = json.load(stream)
    if isinstance(obj, dict):  # tolerate {"embedding": [...]}
        obj = obj["embedding"]
    if not isinstance(obj, list) or len(obj) != DIM:
        raise ValueError(f"query must be a JSON list of {DIM} floats")
    return np.asarray(obj, dtype=np.float32)


def search(valkey: Valkey, query_vec: np.ndarray) -> list[tuple[str, float]]:
    """Run the KNN query inside Valkey; return [(article_id, distance), ...]."""
    raw = valkey.execute_command(
        "FT.SEARCH", INDEX_NAME, QUERY_TEMPLATE,
        "PARAMS", 2, "vec", query_vec.tobytes(),
        "DIALECT", 2,
    )
    # Reply layout: [total, key1, attrs1, key2, attrs2, ...] where attrs is a
    # flat list of (name, value) pairs; the score pair uses our alias
    # ("distance") and the JSON document itself arrives under the "$" name.
    total = int(raw[0])
    results: list[tuple[str, float]] = []
    for i in range(total):
        key = raw[1 + 2 * i]
        attrs = raw[2 + 2 * i] or []
        article_id = key.decode()
        if article_id.startswith(KEY_PREFIX):
            article_id = article_id[len(KEY_PREFIX):]
        distance = None
        for j in range(0, len(attrs) - 1, 2):
            name = attrs[j]
            if isinstance(name, bytes) and name.decode() == "$":
                continue
            try:
                distance = float(attrs[j + 1])
            except (TypeError, ValueError):
                distance = None
            break
        if distance is None:
            distance = float(i)  # fall back to reply order
        results.append((article_id, distance))
    return results


def main() -> int:
    try:
        query_vec = read_query_vector(sys.stdin)
    except Exception as exc:
        print(f"error reading query: {exc}", file=sys.stderr)
        return 1
    try:
        valkey = Valkey(host="127.0.0.1", port=6379, db=0, decode_responses=False)
        results = search(valkey, query_vec)
    except Exception as exc:
        print(f"error querying Valkey: {exc}", file=sys.stderr)
        return 1
    results.sort(key=lambda item: item[1])  # cosine distance: lower is better
    for article_id, _ in results[:K]:
        print(article_id)
    return 0


if __name__ == "__main__":
    sys.exit(main())
EOF
chmod +x /app/search.py
python3 -c "import ast; ast.parse(open('/app/search.py').read()); print('syntax ok')"

26  $ cd /tmp && python3 - <<'EOF'
import json, subprocess, numpy as np

arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
E = np.asarray([a['embedding'] for a in arts], dtype=np.float64)
rng = np.random.default_rng(42)

def truth(q):
    q = np.asarray(q, dtype=np.float64)
    sims = (E @ q) / (np.linalg.norm(E, axis=1) * np.linalg.norm(q))
    return [ids[i] for i in np.argsort(-sims)[:5]]

def run(qvec):
    json.dump(list(map(float, qvec)), open('/tmp/query.json', 'w'))
    p = subprocess.run(['python3', '/app/search.py'], stdin=open('/tmp/query.json'),
                       capture_output=True, text=True)
    assert p.returncode == 0, p.stderr
    return p.stdout.split()

cases = {
  'exact article vector':      list(map(float, arts[313]['embedding'])),
  'perturbed article (+5%)':   list(map(float, np.asarray(arts[777]['embedding']) * 1.05)),
  'mix of two articles':       list(map(float, np.asarray(arts[11]['embedding']) + np.asarray(arts[999]['embedding']))),
  'pure random vector':        list(map(float, rng.normal(size=64))),
  'near-orthogonal random':    list(map(float, rng.normal(size=64) * 0.01)),
}
ok = True
for name, q in cases.items():
    got, exp = run(q), truth(q)
    match = got == exp
    ok &= match
    print(f"{name:26s} -> {'EXACT MATCH' if match else 'MISMATCH'}")
    print(f"   got : {got}")
    if not match: print(f"   want: {exp}")
print('ALL PASS' if ok else 'FAILURES PRESENT')
EOF

27  $ sed -i 's/^import json$/import json\nimport time/' /app/load_articles.py && python3 -c "import ast; ast.parse(open('/app/load_articles.py').read()); print('loader syntax ok')"
echo '--- persistence config:'
grep -E '^(save |appendonly|appendfsync|dbfilename|dir )' /etc/valkey/valkey.conf
echo '--- runtime state:'
valkey-cli CONFIG GET save appendonly dir dbfilename
echo '--- dump dir:'
ls -la /var/lib/valkey/ 2>/dev/null

28  $ rm -f /tmp/vec.bin /tmp/query.json; echo '--- search.py references to /data (should be none):'; grep -n '/data' /app/search.py || echo 'none'; echo '--- DBSIZE / index:'; valkey-cli DBSIZE; valkey-cli FT.INFO kb:articles | tr -d '\n' | grep -o 'num_docs[0-9]*\|num_records[0-9]*'; echo '--- sample key types:'; valkey-cli TYPE kb:vec:kb-0001; valkey-cli TYPE kb:meta:kb-0001

29  $ ps aux | grep -E 'valkey' | grep -v grep; echo '---'; valkey-cli BGSAVE; sleep 2; ls -la /var/lib/valkey/dump.rdb; echo '--- last bgsave:'; valkey-cli INFO persistence | grep -E 'rdb_(last|bgsave)'

30  $ mkdir -p /tmp/restore_test && cp /var/lib/valkey/dump.rdb /tmp/restore_test/
cd /tmp/restore_test
nohup valkey-server --port 6380 --dir /tmp/restore_test --dbfilename dump.rdb --save '' --loadmodule /usr/lib/valkey/libsearch.so > /tmp/restore_test/server.log 2>&1 &
sleep 3
echo '--- keys loaded on 6380:'; valkey-cli -p 6380 DBSIZE
echo '--- index restored from RDB:'; valkey-cli -p 6380 FT.INFO kb:articles | tr -d '\n' | grep -o 'num_docs[0-9]*\|num_records[0-9]*'
echo '--- quick KNN on restored index:'
python3 - <<'EOF'
import json, subprocess, numpy as np
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
qv = np.asarray(arts[500]['embedding'], dtype=np.float32)
open('/tmp/restore_test/q.bin','wb').write(qv.tobytes())
p = subprocess.run(['valkey-cli','-p','6380','-x','FT.SEARCH','kb:articles',
    '*=>[KNN 3 @embedding $vec AS distance]','PARAMS','2','vec','DIALECT','2'],
    stdin=open('/tmp/restore_test/q.bin'), capture_output=True)
print(p.stdout.decode()[:200] or p.stderr.decode()[:200])
EOF

31  $ cat /tmp/restore_test/server.log; echo '--- port check:'; ss -tlnp 2>/dev/null | grep -E '6379|6380' || netstat -tlnp 2>/dev/null | grep -E '6379|6380'

32  $ cd /tmp/restore_test
setsid sh -c 'exec valkey-server --port 6380 --dir /tmp/restore_test --dbfilename dump.rdb --save "" --loadmodule /usr/lib/valkey/libsearch.so --loadmodule /usr/lib/valkey/libjson.so >> /tmp/restore_test/server.log 2>&1' < /dev/null > /dev/null 2>&1 &
sleep 4
valkey-cli -p 6380 PING
echo '--- keys loaded on 6380:'; valkey-cli -p 6380 DBSIZE
echo '--- index restored from RDB:'; valkey-cli -p 6380 FT.INFO kb:articles | tr -d '\n' | grep -o 'num_docs[0-9]*\|num_records[0-9]*'

33  $ python3 - <<'EOF'
import json, subprocess, numpy as np
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
E = np.asarray([a['embedding'] for a in arts], dtype=np.float64)

def truth(q):
    q = np.asarray(q, dtype=np.float64)
    sims = (E @ q) / (np.linalg.norm(E, axis=1) * np.linalg.norm(q))
    return [ids[i] for i in np.argsort(-sims)[:5]]

ok = True
for i in (0, 500, 999):
    qv = np.asarray(arts[i]['embedding'], dtype=np.float32)
    open('/tmp/restore_test/q.bin','wb').write(qv.tobytes())
    p = subprocess.run(['valkey-cli','-p','6380','-x','FT.SEARCH','kb:articles',
        '*=>[KNN 5 @embedding $vec AS distance] NOCONTENT','PARAMS','2','vec','DIALECT','2'],
        stdin=open('/tmp/restore_test/q.bin'), capture_output=True, text=True)
    out = [x for x in p.stdout.split() if x.startswith('kb-')][:5]
    exp = truth(arts[i]['embedding'])
    m = out == exp; ok &= m
    print(f'restored-index query {i}: {"MATCH" if m else "MISMATCH"} {out}')
print('RESTORED INDEX OK' if ok else 'PROBLEM')
EOF
valkey-cli -p 6380 SHUTDOWN NOSAVE 2>/dev/null; sleep 1
rm -rf /tmp/restore_test /tmp/nohup.out
echo '--- port 6380 down, port 6379 still healthy:'; valkey-cli PING; ss -tlnp 2>/dev/null | grep 6380 || echo '6380 gone'

34  $ cd /tmp && python3 - <<'EOF'
import json, subprocess, numpy as np
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
E = np.asarray([a['embedding'] for a in arts], dtype=np.float64)
def truth(q):
    q = np.asarray(q, dtype=np.float64)
    sims = (E @ q) / (np.linalg.norm(E, axis=1) * np.linalg.norm(q))
    return [ids[i] for i in np.argsort(-sims)[:5]]
rng = np.random.default_rng(2024)
ok = True
for i, (name, q) in enumerate([
    ('article 0 exact', arts[0]['embedding']),
    ('article 999 exact', arts[999]['embedding']),
    ('half+half mix', (np.asarray(arts[400]['embedding']) + np.asarray(arts[601]['embedding'])/2).tolist()),
    ('noise', list(map(float, rng.normal(size=64)))),
]):
    json.dump(list(map(float, q)), open('/tmp/q.json','w'))
    p = subprocess.run(['python3','/app/search.py'], stdin=open('/tmp/q.json'), capture_output=True, text=True)
    got = p.stdout.split(); exp = truth(q); m = got == exp; ok &= m
    print(f'{name:18s}: {"MATCH" if m else "MISMATCH"} got={got}')
    if not m: print('   want:', exp, 'stderr:', p.stderr[:200])
print('6379 STILL CORRECT' if ok else 'PROBLEM')
EOF
rm -f /tmp/q.json
echo '--- zero vector (should error, not hang/crash):'
python3 -c "import json; print(json.dumps([0.0]*64))" > /tmp/qz.json
python3 /app/search.py < /tmp/qz.json; echo "exit=$?"; rm -f /tmp/qz.json

35  $ cd /tmp && python3 - <<'EOF'
import sys, json, numpy as np
sys.path.insert(0, '/app')
from search import read_query_vector, search, Valkey

arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
E = np.asarray([a['embedding'] for a in arts], dtype=np.float64)
r = Valkey(host='127.0.0.1', port=6379, decode_responses=False)

qv = read_query_vector(iter([json.dumps(arts[123]['embedding'])]))
res = search(r, qv)
print('parsed (id, distance) pairs:')
for aid, d in res: print(f'  {aid}: {d}')
# numpy truth
q = qv.astype(np.float64)
sims = (E @ q) / (np.linalg.norm(E, axis=1) * np.linalg.norm(q))
cos_dist = {ids[i]: 1 - sims[i] for i in np.argsort(-sims)[:5]}
print('numpy cosine distances for the same ids:')
for aid, cd in cos_dist.items(): print(f'  {aid}: {cd:.9f}')
parsed_ok = all(abs(res[i][1] - list(cos_dist.values())[i]) < 1e-5 for i in range(5))
print('parsed distances match numpy:', parsed_ok)

# zero-vector determinism
z = np.zeros(64, dtype=np.float32)
a = search(r, z); b = search(r, z)
print('zero-vector results deterministic:', [x[0] for x in a] == [x[0] for x in b], [x[0] for x in a])
EOF

36  $ cd /tmp && python3 - <<'EOF'
import sys, io, json, numpy as np
sys.path.insert(0, '/app')
from search import read_query_vector, search, Valkey

arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
E = np.asarray([a['embedding'] for a in arts], dtype=np.float64)
r = Valkey(host='127.0.0.1', port=6379, decode_responses=False)

qv = read_query_vector(io.StringIO(json.dumps(arts[123]['embedding'])))
res = search(r, qv)
print('parsed (id, distance) pairs:')
for aid, d in res: print(f'  {aid}: {d}')
q = qv.astype(np.float64)
sims = (E @ q) / (np.linalg.norm(E, axis=1) * np.linalg.norm(q))
truth_pairs = [(ids[i], 1 - sims[i]) for i in np.argsort(-sims)[:5]]
print('numpy truth (id, cosine distance):')
for aid, d in truth_pairs: print(f'  {aid}: {d:.9f}')
parsed_ok = all(res[i][1] < truth_pairs[i][1] + 1e-4 and res[i][0] == truth_pairs[i][0] for i in range(5))
print('ids + parsed distances consistent with numpy:', parsed_ok)

z = np.zeros(64, dtype=np.float32)
a = search(r, z); b = search(r, z)
print('zero-vector deterministic:', [x[0] for x in a] == [x[0] for x in b], [x[0] for x in a])
EOF

37  $ cd /tmp && python3 -c "
import json
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
json.dump(arts[42]['embedding'], open('/tmp/final_query.json','w'))
"
echo '=== python3 /app/search.py < /tmp/final_query.json ==='
python3 /app/search.py < /tmp/final_query.json
echo '=== files ==='; ls -la /app/
echo '=== valkey state ==='
valkey-cli DBSIZE
valkey-cli FT.INFO kb:articles | tr -d '\n' | grep -o 'num_docs[0-9]*\|num_records[0-9]*\|dimensions64\|distance_metricCOSINE\|algorithmnameFLAT'
valkey-cli INFO persistence | grep -E 'rdb_last_bgsave_status|rdb_changes_since_last_save'
rm -f /tmp/final_query.json

38  $ rm -rf /app/__pycache__; ls /app/

{
 "dbsize": 2000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ wc -l /data/articles.jsonl && head -n 1 /data/articles.jsonl | cut -c1-300

2  $ valkey-server --version; valkey-cli ping; valkey-cli MODULE LIST

3  $ python3 --version; python3 -m pip list 2>/dev/null | grep -iE "valkey|redis|numpy|orjson"; ls -la /app 2>&1; ps aux | grep -i valkey | grep -v grep; grep -iE "^(save|appendonly|dir|dbfilename|daemonize|port|requirepass|bind)" /etc/valkey/valkey.conf

4  $ cat /app/catalog.py; echo ---; grep -vE "^\s*(#|$)" /etc/valkey/valkey.conf | head -60; echo ---; valkey-cli INFO persistence | grep -E "rdb|aof"

5  $ python3 - <<'EOF'
import redis
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=False)
try:
    print(r.execute_command('FT.CREATE', 'idx_probe', 'ON', 'HASH', 'PREFIX', '1', 'probe:',
        'SCHEMA', 'embedding', 'VECTOR', 'FLAT', '6', 'TYPE', 'FLOAT32', 'DIM', '64', 'DISTANCE_METRIC', 'COSINE'))
except Exception as e:
    print('ERR', e)
print(r.execute_command('FT._LIST'))
EOF

6  $ python3 - <<'EOF'
import redis, struct, json
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=False)
# insert two probe hashes with known vectors
v1 = [1.0,0.0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0]
v2 = [0.0,1.0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0]
def pack(v): return struct.pack('<64f', *v)
r.hset('probe:a', mapping={'embedding': pack(v1)})
r.hset('probe:b', mapping={'embedding': pack(v2)})
import time; time.sleep(0.5)
q = pack([1.0,0.0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0, 0,0,0,0,0,0,0,0])
for dialect in (1,2,3):
    try:
        res = r.execute_command('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',q,'SORTBY','score','ASC','DIALECT',str(dialect))
        print('dialect',dialect,'total',res[0])
        for el in res[1:3]:
            print('  ', el if not isinstance(el, bytes) or len(el)<80 else el[:80])
    except Exception as e:
        print('dialect',dialect,'ERR',e)
EOF

7  $ python3 - <<'EOF'
import redis, struct
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=False)
def pack(v): return struct.pack('<64f', *v)
qb = pack([1.0]+[0.0]*63)
# nearest to q should be probe:a (cos=1) then probe:b (cos=0)
for args in (
    ('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',qb,'DIALECT','2'),
    ('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec AS score]','PARAMS','2','vec',qb,'DIALECT','3'),
    ('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec AS score]','NOCONTENT','PARAMS','2','vec',qb,'DIALECT','2'),
):
    try:
        res = r.execute_command(*args)
        print(args[3], 'total=', res[0], 'ids=', [x.decode() if isinstance(x,bytes) else x for x in res[1::2]])
    except Exception as e:
        print(args[3],'ERR',e)
# also test KNN inside query without alias, dialect 2
try:
    res = r.execute_command('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec]','PARAMS','2','vec',qb,'DIALECT','2')
    print('noalias total=', res[0], [x for x in res[1::2]])
except Exception as e:
    print('noalias ERR', e)
EOF

8  $ python3 - <<'EOF'
import redis
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=False)
try:
    print(r.execute_command('FT.CREATE','idx_probe64','ON','HASH','PREFIX','1','probe64:',
        'SCHEMA','embedding','VECTOR','FLAT','6','TYPE','FLOAT64','DIM','64','DISTANCE_METRIC','COSINE'))
    print('FLOAT64 OK')
except Exception as e:
    print('FLOAT64 ERR', e)
try:
    print('VADD:', r.execute_command('VADD','vs1','1.0','0.5','e1'))
except Exception as e:
    print('VADD ERR:', e)
print('save cfg:', r.config_get('save'))
print('dbsize:', r.dbsize())
EOF

9  $ python3 - <<'EOF'
import valkey, struct
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=False)
def pack(v): return struct.pack('<64f', *v)
qb = pack([1.0]+[0.0]*63)
res = c.execute_command('FT.SEARCH','idx_probe','*=>[KNN 5 @embedding $vec AS score]','NOCONTENT','PARAMS','2','vec',qb,'DIALECT','2')
print('valkey-py raw reply:', res)
info = c.execute_command('FT.INFO','idx_probe')
d = dict(zip(info[::2], info[1::2]))
for k in (b'num_docs', b'backfill_finish_ts', b'backfill_remaining_hint', b'index_size', b'fields'):
    print(k, d.get(k))
EOF

10  $ python3 - <<'EOF'
import valkey
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=False)
info = c.execute_command('FT.INFO','idx_probe')
# print nested structure readably
def show(x, ind=0):
    if isinstance(x, list):
        for el in x:
            show(el, ind)
    else:
        print(x.decode() if isinstance(x, bytes) else x)
show(info)
EOF

11  $ python3 - <<'EOF'
import json
ids=[]; dims=set(); n=0
with open('/data/articles.jsonl') as f:
    for line in f:
        o=json.loads(line); n+=1
        ids.append(o['id']); dims.add(len(o['embedding']))
        assert set(o.keys()) <= {'id','title','embedding'}, o.keys()
print('n=',n,'unique ids=',len(set(ids)),'dims=',dims)
print('first,last:',ids[0],ids[-1])
# embedding value ranges
vals=[v for line in open('/data/articles.jsonl') for v in json.loads(line)['embedding']]
print('min,max:',min(vals),max(vals))
# how many digits in floats
print(open('/data/articles.jsonl').readline()[150:230])
EOF

12  $ cat > /tmp/load_articles.py <<'EOF'
"""Load /data/articles.jsonl into Valkey + create vector index for exact cosine KNN."""
import json, struct, time
import valkey

KEY_PREFIX = "kb:"
INDEX_NAME = "idx:articles"
DIM = 64

c = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=False)

# --- clean up any probe leftovers from earlier experiments ---
for idx in c.execute_command("FT._LIST"):
    name = idx.decode()
    if name.startswith("idx_probe"):
        print("dropping probe index", name, c.execute_command("FT.DROPINDEX", name))
probe_keys = list(c.scan_iter(match="probe*", count=100))
if probe_keys:
    print("deleting probe keys", c.delete(*probe_keys))

# --- create the index (exact KNN via FLAT, cosine similarity) ---
try:
    c.execute_command(
        "FT.CREATE", INDEX_NAME, "ON", "HASH", "PREFIX", "1", KEY_PREFIX,
        "SCHEMA",
        "id", "TAG",
        "title", "TEXT",
        "embedding", "VECTOR", "FLAT", "6",
        "TYPE", "FLOAT32", "DIM", str(DIM), "DISTANCE_METRIC", "COSINE",
    )
    print("created index", INDEX_NAME)
except Exception as e:
    print("FT.CREATE:", e)

# --- load articles in pipeline batches ---
n = 0
BATCH = 500
pipe = c.pipeline(transaction=False)
with open("/data/articles.jsonl") as f:
    for line in f:
        art = json.loads(line)
        assert len(art["embedding"]) == DIM
        emb = struct.pack(f"<{DIM}f", *[float(x) for x in art["embedding"]])
        pipe.hset(KEY_PREFIX + art["id"], mapping={
            "id": art["id"],
            "title": art["title"],
            "embedding": emb,
        })
        n += 1
        if n % BATCH == 0:
            pipe.execute()
pipe.execute()
print("loaded", n, "articles; dbsize =", c.dbsize())

# --- wait until the index has ingested everything ---
deadline = time.time() + 60
while time.time() < deadline:
    info = dict(zip(c.execute_command("FT.INFO", INDEX_NAME)[::2],
                    c.execute_command("FT.INFO", INDEX_NAME)[1::2]))
    if info[b"num_docs"] >= n and info[b"backfill_complete_percent"] == 1.0:
        break
    time.sleep(0.5)
print("num_docs =", info[b"num_docs"], "backfill =", info[b"backfill_complete_percent"],
      "state =", info[b"state"], "failures =", info[b"hash_indexing_failures"])
assert info[b"num_docs"] >= n and info[b"backfill_complete_percent"] == 1.0
print("OK")
EOF
python3 /tmp/load_articles.py

13  $ cat > /app/search.py <<'EOF'
"""Semantic search over knowledge-base articles using Valkey's vector index.

Usage: python3 /app/search.py < query.json
where query.json contains a single query embedding (JSON list of 64 floats).
Prints the ids of the 5 most cosine-similar articles, one per line, best first.

All similarity computation happens inside Valkey (FT.SEARCH KNN over the
`idx:articles` vector index); this script never touches the source data file.
"""
import json
import struct
import sys

import valkey

INDEX_NAME = "idx:articles"
KEY_PREFIX = "kb:"
DIM = 64
K = 5


def main() -> None:
    query = json.load(sys.stdin)
    if not isinstance(query, list) or len(query) != DIM:
        sys.exit(f"query embedding must be a JSON list of {DIM} floats")

    query_blob = struct.pack(f"<{DIM}f", *[float(x) for x in query])

    c = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=False)

    # Exact KNN search executed by Valkey's search module.  The KNN clause
    # orders results by ascending distance (cosine distance = 1 - cosine
    # similarity), so reply id #0 is the most similar article.
    reply = c.execute_command(
        "FT.SEARCH", INDEX_NAME,
        "*=>[KNN $K @embedding $vec AS score]",
        "NOCONTENT",
        "PARAMS", "4", "vec", query_blob, "K", str(K),
        "DIALECT", "2",
    )

    if isinstance(reply, bytes) or not isinstance(reply, list):
        sys.exit(f"unexpected FT.SEARCH reply: {reply!r}")
    if len(reply) < 1 + 2 * K:
        sys.exit(f"short FT.SEARCH reply: {reply!r}")

    ids = [key.decode() if isinstance(key, bytes) else key for key in reply[1::2]]
    for article_id in ids:
        print(article_id[len(KEY_PREFIX):])


if __name__ == "__main__":
    main()
EOF
echo "--- written ---"; python3 -c "import ast; ast.parse(open('/app/search.py').read()); print('syntax OK')"

14  $ python3 - <<'EOF'
import valkey, struct, json, numpy as np
c = valkey.Valkey(host='127.0.0.1', port=6379, decode_responses=False)
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
X = np.array([a['embedding'] for a in arts], dtype=np.float32)  # (1000,64)
ids = [a['id'] for a in arts]
# duplicate embeddings?
u = np.unique(X, axis=0); print('unique embeddings:', u.shape[0])

def pack(v): return struct.pack('<64f', *v)
q = pack(X[42])
def run(*extra):
    return c.execute_command('FT.SEARCH','idx:articles','*=>[KNN 5 @embedding $vec AS score]',
                             'PARAMS','2','vec',q,'DIALECT','2', *extra)
# 1) NOCONTENT order
r = run('NOCONTENT')
print('NOCONTENT keys:', [k.decode() for k in r[1::2]])
# 2) RETURN id
try:
    r2 = run('RETURN','1','id')
    print('RETURN raw:', r2)
except Exception as e:
    print('RETURN ERR:', e)
# 3) $K param
try:
    r3 = c.execute_command('FT.SEARCH','idx:articles','*=>[KNN $K @embedding $vec AS score]',
                           'NOCONTENT','PARAMS','4','vec',q,'K','5','DIALECT','2')
    print('$K param keys:', [k.decode() for k in r3[1::2]])
except Exception as e:
    print('$K ERR:', e)

# ground truth float32 cosine
Xn = X / np.linalg.norm(X, axis=1, keepdims=True)
qv = Xn[42]
sims = Xn @ qv
top5 = np.argsort(-sims)[:5]
print('numpy top5:', [ids[i] for i in top5], 'sims:', sims[top5])
print('gap 5th-6th:', sims[top5[-1]] - sims[np.argsort(-sims)[5]])
EOF

15  $ cat > /app/search.py <<'EOF'
"""Semantic search over knowledge-base articles using Valkey's vector index.

Usage: python3 /app/search.py < query.json
where query.json contains a single query embedding (JSON list of 64 floats).
Prints the ids of the 5 articles most cosine-similar to the query, one per
line, most similar first.

All similarity computation happens inside Valkey (FT.SEARCH KNN over the
`idx:articles` vector index); this script never reads the source data file
and never computes similarities itself.
"""
import json
import struct
import sys

import valkey

INDEX_NAME = "idx:articles"
KEY_PREFIX = "kb:"
DIM = 64
K = 5


def main() -> None:
    query = json.load(sys.stdin)
    if not isinstance(query, list) or len(query) != DIM:
        sys.exit(f"query embedding must be a JSON list of {DIM} floats")

    query_blob = struct.pack(f"<{DIM}f", *[float(x) for x in query])

    c = valkey.Valkey(host="127.0.0.1", port=6379, decode_responses=False)

    # Exact KNN search executed by Valkey's search module.  The KNN clause
    # orders the reply by ascending distance (cosine distance = 1 - cosine
    # similarity), so the first document is the most similar article.
    # RETURN 1 id fetches the article id stored in each hash.
    reply = c.execute_command(
        "FT.SEARCH", INDEX_NAME,
        f"*=>[KNN {K} @embedding $vec AS score]",
        "RETURN", "1", "id",
        "PARAMS", "2", "vec", query_blob,
        "DIALECT", "2",
    )

    if not isinstance(reply, list) or len(reply) < 1 + 2 * K:
        sys.exit(f"unexpected FT.SEARCH reply: {reply!r}")

    # Reply: [total, key, [field, value], key, [field, value], ...]
    results = []
    for key, attrs in zip(reply[1::2], reply[2::2]):
        key = key.decode() if isinstance(key, bytes) else str(key)
        article_id = None
        if isinstance(attrs, (list, tuple)):
            for i in range(0, len(attrs) - 1, 2):
                field = attrs[i].decode() if isinstance(attrs[i], bytes) else str(attrs[i])
                if field == "id":
                    val = attrs[i + 1]
                    article_id = val.decode() if isinstance(val, bytes) else str(val)
        results.append(article_id if article_id else key[len(KEY_PREFIX):])

    for article_id in results:
        print(article_id)


if __name__ == "__main__":
    main()
EOF
python3 -c "import ast; ast.parse(open('/app/search.py').read()); print('syntax OK')"
# end-to-end test: query = article kb-0043's own embedding -> should return kb-0043 first
python3 - <<'EOF'
import json, subprocess
arts = {json.loads(l)['id']: json.loads(l)['embedding'] for l in open('/data/articles.jsonl')}
q = json.dumps(arts['kb-0043'])
out = subprocess.run(['python3','/app/search.py'], input=q, capture_output=True, text=True)
print('stdout:'); print(out.stdout); print('stderr:', out.stderr)
EOF

16  $ python3 - <<'EOF'
import json, subprocess, numpy as np
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
ids = [a['id'] for a in arts]
X = np.array([a['embedding'] for a in arts], dtype=np.float32)
rng = np.random.default_rng(7)
qidx = rng.choice(1000, 60, replace=False).tolist()
randoms = rng.normal(0, 0.2, (40, 64)).astype(np.float32)
queries = [X[i] for i in qidx] + list(randoms)

Xn = X / np.linalg.norm(X, axis=1, keepdims=True)
mismatch = 0
for i, q in enumerate(queries):
    qn = q / np.linalg.norm(q)
    sims = Xn @ qn
    # top-5 with deterministic tie-break by id
    order = sorted(range(1000), key=lambda j: (-sims[j], ids[j]))[:5]
    truth = [ids[j] for j in order]
    out = subprocess.run(['python3','/app/search.py'], input=json.dumps([float(x) for x in q]),
                         capture_output=True, text=True)
    got = out.stdout.split()
    ok = got == truth
    if not ok:
        mismatch += 1
        print(f'MISMATCH q#{i} expected={truth} got={got} err={out.stderr.strip()}')
print(f'{len(queries)-mismatch}/{len(queries)} queries exact match')
# nearest tie sanity: gap distribution
sims_all = []
for q in queries:
    qn = q/np.linalg.norm(q)
    s = np.sort(Xn @ qn)[::-1]
    sims_all.append(s[4]-s[5])
print('min top5/top6 sim gap:', min(sims_all))
EOF
grep -c "articles.jsonl" /app/search.py; echo "(0 = script does not touch data file)"

17  $ ls -la /var/lib/valkey/; valkey-cli SAVE; ls -la /var/lib/valkey/; valkey-cli INFO persistence | grep -E "rdb_last_save_time|rdb_last_bgsave_status|rdb_changes_since_last_save"; ls /etc/systemd/system/ 2>/dev/null | grep -i valkey; systemctl list-unit-files 2>/dev/null | grep -i valkey

18  $ python3 - <<'EOF' > /tmp/proof_before.json
import json, subprocess
emb = json.loads(open('/data/articles.jsonl').readline().split('"embedding":',1)[1].rsplit('}',1)[0]) if False else None
EOF
# capture a reference query before restart
python3 - <<'EOF'
import json, subprocess
line = open('/data/articles.jsonl').readline()
q = json.loads(line)['embedding']
open('/tmp/q1.json','w').write(json.dumps(q))
open('/tmp/proof_before.txt','w').write(subprocess.run(['python3','/app/search.py'], input=json.dumps(q), capture_output=True, text=True).stdout)
line2 = open('/data/articles.jsonl').readlines()[777]
open('/tmp/q2.json','w').write(json.dumps(json.loads(line2)['embedding']))
open('/tmp/proof2_before.txt','w').write(subprocess.run(['python3','/app/search.py'], input=open('/tmp/q2.json').read(), capture_output=True, text=True).stdout)
print(open('/tmp/proof_before.txt').read())
EOF
echo "=== restarting valkey ==="
valkey-cli SHUTDOWN SAVE || true
sleep 1
valkey-server /etc/valkey/valkey.conf
sleep 2
valkey-cli PING
valkey-cli DBSIZE
valkey-cli FT._LIST

19  $ python3 /app/search.py < /tmp/q1.json > /tmp/proof_after.txt
python3 /app/search.py < /tmp/q2.json > /tmp/proof2_after.txt
diff /tmp/proof_before.txt /tmp/proof_after.txt && echo "query1 identical after restart"
diff /tmp/proof2_before.txt /tmp/proof2_after.txt && echo "query2 identical after restart"
cat /tmp/proof_after.txt
valkey-cli FT.INFO idx:articles | tr -d '\n' | grep -oE "num_docs[^a-z]*[0-9]+|backfill_in_progress[^a-z]*[0-9]+|hash_indexing_failures[^a-z]*[0-9]+|state[^a-z]*ready" 

20  $ echo '--- bad input handling ---'
echo '[1,2,3]' | python3 /app/search.py; echo "exit=$?"
echo '[]' | python3 /app/search.py; echo "exit=$?"
echo '--- 5 ids exactly, in order, one per line ---'
python3 /app/search.py < /tmp/q2.json | wc -l
echo '--- final state ---'
valkey-cli DBSIZE; valkey-cli FT._LIST; valkey-cli RANDOMKEY

21  $ valkey-cli INFO persistence | grep -E "rdb_changes_since_last_save|rdb_last_bgsave_status"; valkey-cli FT.INFO idx:articles | tr -d '\n' | grep -oE "num_docs[^a-z]*[0-9]+" | head -1

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ valkey-server --version; valkey-cli --version; python3 --version; pip list 2>/dev/null | grep -i -E 'redis|valkey|numpy'; echo ---; wc -l /data/articles.jsonl; head -c 400 /data/articles.jsonl

2  $ valkey-cli -h 127.0.0.1 -p 6379 PING; valkey-cli -h 127.0.0.1 -p 6379 MODULE LIST; echo ---; valkey-cli -h 127.0.0.1 -p 6379 INFO server | grep -E 'valkey_version|modules'

3  $ python3 -c "import valkey, sys; print(valkey.__version__)"; pip show valkey 2>/dev/null | head -3

4  $ ls /usr/share/doc/valkey* 2>/dev/null; ls /usr/share/doc/ | head -50; find / -iname '*valkey*search*' -not -path '/proc/*' 2>/dev/null | head

5  $ valkey-cli -h 127.0.0.1 -p 6379 FT.CREATE help 2>&1 | head -60; echo ===; valkey-cli -h 127.0.0.1 -p 6379 FT.INFO help 2>&1 | head -5

6  $ valkey-cli -h 127.0.0.1 -p 6379 COMMAND COUNT; valkey-cli -h 127.0.0.1 -p 6379 --raw COMMAND | grep -i -E '^(FT|VS)' | sort | head -50

7  $ find /usr -name 'search.so' -o -name '*libsearch*' 2>/dev/null; ls -la /usr/lib/valkey/; python3 -c "import valkey; print(valkey.__file__)"

8  $ valkey-cli -h 127.0.0.1 -p 6379 COMMAND INFO FT.CREATE | head -c 2000; echo; echo ===DOCS===; valkey-cli -h 127.0.0.1 -p 6379 --raw COMMAND DOCS FT.CREATE FT.SEARCH | head -c 6000

9  $ strings -n 3 /usr/lib/valkey/libsearch.so | grep -x -i -E 'HNSW|FLAT|COSINE|L2|IP|DIM|TYPE|FLOAT32|FLOAT16|FLOAT64|KNN|AS|SORTBY|PARAMS|DIALECT|LIMIT|EF|M|NOHYBRID|HYBRID|BFACF|YIELD_RADIUS_AS|YIELD_SCORE_AS|EPSILON|CAP' | sort -u

10  $ strings -n 4 /usr/lib/valkey/libsearch.so | grep -E 'Usage|usage|FT\.CREATE|FT\.SEARCH' | head -40

11  $ python3 - <<'EOF'
import re
data = open('/usr/lib/valkey/libsearch.so','rb').read()
print(len(data))
pat = re.compile(rb'[\x20-\x7e]{4,}')
strs = pat.findall(data)
print(len(strs))
keys = [s.decode() for s in strs if re.fullmatch(rb'(HNSW|FLAT|COSINE|L2|IP|DIM|TYPE|FLOAT32|FLOAT16|FLOAT64|KNN|AS|SORTBY|PARAMS|DIALECT|LIMIT|EF_ROUNDS|NOHYBRID|EF_RUNTIME|YIELD_RADIUS_AS|YIELD_SCORE_AS)', s)]
from collections import Counter
print(Counter(keys))
EOF

12  $ python3 - <<'EOF'
import re
data = open('/usr/lib/valkey/libsearch.so','rb').read()
pat = re.compile(rb'[\x20-\x7e]{3,}')
# find offsets of key tokens
for tok in [b'HNSW', b'COSINE', b'KNN', b'VECTOR', b'VEC', b'FLOAT16', b'M=', b'EF_CONSTRUCTION', b'Algorithm', b'algorithm', b'distance', b'DISTANCE', b'num_vectors', b'attribute']:
    offs = [m.start() for m in re.finditer(re.escape(tok), data)]
    print(tok, offs[:10], '...' if len(offs)>10 else '')
EOF

13  $ python3 - <<'EOF'
import re
data = open('/usr/lib/valkey/libsearch.so','rb').read()
def dump(lo, hi):
    chunk = data[lo:hi]
    txt = re.sub(rb'[^\x20-\x7e\n]+', b'\n', chunk)
    print(txt.decode())
print('=== around 16735800-16737200 ===')
dump(16735800, 16737400)
print('=== around 16750000-16755000 ===')
dump(16750000, 16755200)
EOF

14  $ grep -E '^(save|appendonly|appendfsync|dir|dbfilename|loadmodule)' /etc/valkey/valkey.conf; echo ---; valkey-cli -h 127.0.0.1 -p 6379 CONFIG GET save appendonly dir

15  $ valkey-cli -h 127.0.0.1 -p 6379 FT.CREATE idx:articles ON HASH PREFIX 1 article: SCHEMA title TEXT embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE

16  $ mkdir -p /app && cat > /app/load_articles.py <<'EOF'
#!/usr/bin/env python3
"""Load /data/articles.jsonl into Valkey as hashes for FT.SEARCH vector search."""
import json, struct
from valkey import Valkey

DB = "articles"

def main():
    with open("/data/articles.jsonl") as f:
        articles = [json.loads(line) for line in f if line.strip()]
    r = Valkey(host="127.0.0.1", port=6379)
    pipe = r.pipeline(transaction=False)
    for i, a in enumerate(articles, 1):
        key = f"article:{a['id']}"
        blob = struct.pack("<64f", *a["embedding"])
        pipe.hset(key, mapping={"title": a["title"], "embedding": blob})
        if i % 200 == 0:
            pipe.execute()
            print(f"loaded {i}")
    pipe.execute()
    print(f"done, total {len(articles)}")
    print(r.ft(DB).info["num_docs"] if hasattr(r, "ft") else r.execute_command("FT.INFO", DB))

if __name__ == "__main__":
    main()
EOF
python3 /app/load_articles.py

17  $ valkey-cli -h 127.0.0.1 -p 6379 FT.INFO idx:articles | head -20; valkey-cli -h 127.0.0.1 -p 6379 INFO search | grep -E 'ingest_hash|backfill_hash|global_ingestion|search_queries|number_of'

18  $ valkey-cli -h 127.0.0.1 -p 6379 INFO search | grep -E 'ingest|backfill|global'; echo ---; valkey-cli -h 127.0.0.1 -p 6379 FT.INFO idx:articles | grep -A2 num_docs

19  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over knowledge-base articles using Valkey Search (FT.SEARCH).

Usage: python3 /app/search.py < query.json
  query.json contains one query embedding (a JSON list of 64 floats).
Prints the ids of the 5 most similar articles (cosine similarity), one per
line, most similar first. All similarity math happens inside Valkey via the
FT.SEARCH KNN vector query; this script only forwards the query vector.
"""
import json
import struct
import sys

from valkey import Valkey

INDEX = "idx:articles"
K = 5
DIM = 64


def main() -> None:
    query_embedding = json.load(sys.stdin)
    if len(query_embedding) != DIM:
        sys.exit(f"query embedding must have {DIM} dimensions")
    query_blob = struct.pack(f"<{DIM}f", *query_embedding)

    r = Valkey(host="127.0.0.1", port=6379)
    # Pure KNN query: rank all indexed articles by cosine distance and keep
    # the K closest. SORTBY on the distance alias orders most-similar first.
    reply = r.execute_command(
        "FT.SEARCH", INDEX,
        "*=>[KNN %d @embedding $query_vector AS vec_distance]" % K,
        "PARAMS", 2, "query_vector", query_blob,
        "SORTBY", "vec_distance", "ASC",
        "NOCONTENT",
        "DIALECT", 2,
    )
    # reply = [total_hits, key1, key2, ...] with NOCONTENT
    for item in reply[1 : 1 + K]:
        key = item.decode() if isinstance(item, bytes) else item
        print(key.split(":", 1)[1])  # "article:kb-0123" -> "kb-0123"


if __name__ == "__main__":
    main()
EOF
chmod +x /app/search.py
python3 - <<'EOF'
import json
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
for art in arts[:3]:
    with open('/tmp/q.json','w') as f:
        json.dump(art['embedding'], f)
    print('query:', art['id'])
EOF
head -c 120 /tmp/q.json; echo

20  $ cat > /tmp/verify.py <<'EOF'
import json, subprocess, math, sys
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
by_id = {a['id']: a for a in arts}

def brute(q):
    s = sorted(arts, key=lambda a: -sum(x*y for x,y in zip(a['embedding'], q)))
    return [a['id'] for a in s[:5]]

ok = True
for art in arts[::97] + arts[:3]:
    q = art['embedding']
    open('/tmp/q.json','w').write(json.dumps(q))
    out = subprocess.run(['python3','/app/search.py'], stdin=open('/tmp/q.json'), capture_output=True, text=True)
    if out.returncode != 0:
        print('FAIL run', art['id'], out.stderr); ok=False; continue
    got = out.stdout.split()
    want = brute(q)
    status = 'OK' if got == want else 'MISMATCH'
    if got != want: ok = False
    print(art['id'], status, got, '| brute:', want, 'top1 dist:', 
          1-sum(x*y for x,y in zip(by_id[got[0]]['embedding'], q)))
print('ALL OK' if ok else 'FAILURES PRESENT')
EOF
python3 /tmp/verify.py

21  $ python3 - <<'EOF'
import json, struct
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
open('/tmp/qblob.bin','wb').write(struct.pack('<64f', *arts[0]['embedding']))
EOF
valkey-cli -h 127.0.0.1 -p 6379 --raw FT.SEARCH idx:articles '*=>[KNN 5 @embedding $query_vector AS vec_distance]' PARAMS 2 query_vector "$(xxd -p /tmp/qblob.bin | tr -d '\n')" NOCONTENT DIALECT 2

22  $ 

23  $ python3 - <<'EOF'
import json, struct
from valkey import Valkey
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
r=Valkey(host='127.0.0.1',port=6379)
blob=struct.pack('<64f',*arts[0]['embedding'])
for q in ['*=>[KNN 5 @embedding $query_vector AS vec_distance]']:
    for extra in (['NOCONTENT','DIALECT','2'], ['DIALECT','2']):
        try:
            reply = r.execute_command('FT.SEARCH','idx:articles',q,'PARAMS',2,'query_vector',blob,*extra)
            print(extra, '->', reply)
        except Exception as e:
            print(extra, 'ERROR:', e)
EOF

24  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over knowledge-base articles using Valkey Search (FT.SEARCH).

Usage: python3 /app/search.py < query.json
  query.json contains one query embedding (a JSON list of 64 floats).
Prints the ids of the 5 most similar articles (cosine similarity), one per
line, most similar first. The similarity ranking is computed entirely inside
Valkey by the KNN vector query; this script only forwards the query vector.
"""
import json
import struct
import sys

from valkey import Valkey

INDEX = "idx:articles"
KEY_PREFIX = "article:"
K = 5
DIM = 64


def main() -> None:
    query_embedding = json.load(sys.stdin)
    if len(query_embedding) != DIM:
        sys.exit(f"query embedding must have {DIM} dimensions")
    query_blob = struct.pack(f"<{DIM}f", *query_embedding)

    r = Valkey(host="127.0.0.1", port=6379)
    # KNN query: Valkey ranks all indexed articles by cosine distance and
    # returns the K closest ones, already ordered nearest-first.
    reply = r.execute_command(
        "FT.SEARCH", INDEX,
        f"*=>[KNN {K} @embedding $query_vector AS vec_distance]",
        "PARAMS", 2, "query_vector", query_blob,
        "NOCONTENT",
        "DIALECT", 2,
    )
    # reply = [total_hits, key1, key2, ...] because of NOCONTENT
    for item in reply[1 : 1 + K]:
        key = item.decode() if isinstance(item, bytes) else item
        print(key[len(KEY_PREFIX):] if key.startswith(KEY_PREFIX) else key)


if __name__ == "__main__":
    main()
EOF
python3 /tmp/verify.py

25  $ ps aux | grep -v grep | grep valkey; echo ---; valkey-cli -h 127.0.0.1 -p 6379 BGSAVE; sleep 2; ls -la /var/lib/valkey/; valkey-cli -h 127.0.0.1 -p 6379 LASTSAVE; valkey-cli -h 127.0.0.1 -p 6379 DBSIZE

26  $ valkey-cli -h 127.0.0.1 -p 6379 SAVE 2>&1; valkey-cli -h 127.0.0.1 -p 6379 SHUTDOWN NOSAVE 2>/dev/null; sleep 1; valkey-server /etc/valkey/valkey.conf --daemonize yes; sleep 2; valkey-cli -h 127.0.0.1 -p 6379 DBSIZE; valkey-cli -h 127.0.0.1 -p 6379 PING

27  $ valkey-cli -h 127.0.0.1 -p 6379 FT.INFO idx:articles | grep -E 'num_docs|index_name' ; valkey-cli -h 127.0.0.1 -p 6379 FT._LIST

28  $ valkey-cli -h 127.0.0.1 -p 6379 FT.INFO idx:articles | grep -E -A1 'index_name|num_docs'; echo ---; python3 /app/search.py < /tmp/q.json

29  $ cat > /tmp/verify2.py <<'EOF'
import json, subprocess, random
arts=[json.loads(l) for l in open('/data/articles.jsonl')]
def brute(q):
    return [a['id'] for a in sorted(arts, key=lambda a: -sum(x*y for x,y in zip(a['embedding'],q)))[:5]]
ok=True
random.seed(7)
tests=[]
for _ in range(20):
    v=[random.gauss(0,0.2) for _ in range(64)]  # unnormalized random vectors
    tests.append(v)
tests += [a['embedding'] for a in arts[::53]]
for q in tests:
    open('/tmp/q.json','w').write(json.dumps(q))
    out=subprocess.run(['python3','/app/search.py'],stdin=open('/tmp/q.json'),capture_output=True,text=True)
    got=out.stdout.split()
    want=brute(q)
    if got!=want or out.returncode!=0:
        ok=False; print('FAIL', got, want, out.stderr[:200])
print('ALL', len(tests), 'QUERIES OK' if ok else 'FAILED')
EOF
python3 /tmp/verify2.py; ls -la /app/; rm -f /tmp/qblob.bin

30  $ grep -n '/data' /app/search.py || echo "search.py: no /data reference"; grep -cE 'zip|dot|norm|cosine' /app/search.py || true; echo ===; cat /app/search.py

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ ls -la /data/ /app/ 2>&1; head -c 500 /data/articles.jsonl; echo; wc -l /data/articles.jsonl

2  $ valkey-server --version; valkey-cli ping; cat /etc/valkey/valkey.conf | grep -v '^#' | grep -v '^$'

3  $ valkey-cli COMMAND DOCS 2>/dev/null | grep -iE '^\s+"[a-z.*]+' | head -100; echo ---; valkey-cli INFO modules

4  $ cat /app/catalog.py; ls /usr/lib/valkey/

5  $ valkey-cli COMMAND LIST | tr ' ' '\n' | grep -i '^FT\.' ; echo ---; valkey-cli COMMAND DOCS FT.CREATE 2>&1 | head -50

6  $ python3 -c "import redis; print(redis.__version__)" 2>&1; python3 --version; pip list 2>/dev/null | head -30

7  $ cat > /app/load.py <<'EOF'
import json, struct, sys

# use the valkey python client if available, else redis
try:
    import valkey as client_lib
except ImportError:
    import redis as client_lib

r = client_lib.Valkey(host="127.0.0.1", port=6379, decode_responses=False)

# reset previous state
try:
    r.execute_command("FT.DROPINDEX", "articles", "DD")
except Exception:
    pass
# delete any leftover article keys
keys = list(r.scan_iter("article:*"))
if keys:
    for i in range(0, len(keys), 500):
        r.delete(*keys[i:i+500])

pipe = r.pipeline(transaction=False)
n = 0
with open("/data/articles.jsonl") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        a = json.loads(line)
        emb = struct.pack("<%df" % len(a["embedding"]), *a["embedding"])
        pipe.hset("article:%s" % a["id"], mapping={
            "id": a["id"],
            "title": a["title"],
            "embedding": emb,
        })
        n += 1
        if n % 200 == 0:
            pipe.execute()
            pipe = r.pipeline(transaction=False)
pipe.execute()

# create vector index: exact (FLAT) cosine similarity
r.execute_command(
    "FT.CREATE", "articles",
    "ON", "HASH",
    "PREFIX", "1", "article:",
    "SCHEMA",
    "id", "TEXT",
    "title", "TEXT",
    "embedding", "VECTOR", "FLAT", "6",
    "TYPE", "FLOAT32",
    "DIM", "64",
    "DISTANCE_METRIC", "COSINE",
)

print("loaded", n, "articles", file=sys.stderr)
print("index count:", r.execute_command("FT.INFO", "articles"), file=sys.stderr)
EOF
python3 /app/load.py 2>&1 | tail -5

8  $ for i in $(seq 1 30); do
  out=$(valkey-cli FT.INFO articles | grep -A1 'num_docs' | tail -1 | tr -d ' \r')
  echo "try $i num_docs=$out"
  [ "$out" = "1000" ] && break
  sleep 1
done

9  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the knowledge-base articles stored in Valkey.

Reads a query embedding (JSON list of 64 floats) from stdin, runs a
cosine-similarity vector search inside Valkey (valkey-search module,
FT.SEARCH over a FLAT vector index), and prints the ids of the 5 most
similar articles, most similar first.
"""
import json
import struct
import sys

try:
    import valkey as client_lib
except ImportError:
    import redis as client_lib

TOP_K = 5


def main():
    query = json.load(sys.stdin)
    # binary float32 encoding of the query vector, as expected by FT.SEARCH
    vec = struct.pack("<%df" % len(query), *query)

    r = client_lib.Valkey(host="127.0.0.1", port=6379, decode_responses=False)
    res = r.execute_command(
        "FT.SEARCH", "articles", "*=>[KNN 5 @embedding $vec]",
        "PARAMS", "2", "vec", vec,
        "SORTBY", "__embedding_score", "ASC",
        "LIMIT", "0", str(TOP_K),
        "RETURN", "1", "id",
        "DIALECT", "2",
    )

    # reply: [total, key1, [field, value, ...], key2, ...]
    out = sys.stdout
    it = iter(res[1:])
    for key in it:
        fields = next(it)
        attrs = dict(zip(fields[0::2], fields[1::2]))
        out.write(attrs[b"id"].decode() + "\n")


if __name__ == "__main__":
    main()
EOF
chmod +x /app/search.py
echo written

10  $ cd /tmp && python3 -c "
import json
a = json.loads(open('/data/articles.jsonl').readline())
json.dump(a['embedding'], open('q.json','w'))
print('query is article', a['id'])
" && python3 /app/search.py < q.json

11  $ valkey-cli FT.SEARCH articles '*=>[KNN 5 @embedding $vec]' PARAMS 2 vec $(python3 -c "
import json,struct
e=json.loads(open('/data/articles.jsonl').readlines()[500]['embedding'])
print(struct.pack('<64f',*e).hex())
") SORTBY __embedding_score ASC LIMIT 0 5 RETURN 1 id DIALECT 2

12  $ cd /app && python3 - <<'EOF'
import re
s = open('search.py').read()
s = s.replace('''        "SORTBY", "__embedding_score", "ASC",\n''', '')
open('search.py','w').write(s)
EOF
python3 /app/search.py < /tmp/q.json; echo "exit=$?"

13  $ cd /tmp && python3 - <<'EOF'
import json, struct, subprocess, random, math

arts = [json.loads(l) for l in open('/data/articles.jsonl')]
random.seed(42)

def gt_top5(q):
    def cos(a, b):
        dot = sum(x*y for x, y in zip(a, b))
        na = math.sqrt(sum(x*x for x in a)); nb = math.sqrt(sum(x*x for x in b))
        return dot/(na*nb)
    ranked = sorted(arts, key=lambda a: -cos(q, a["embedding"]))
    return [a["id"] for a in ranked[:5]]

ok = True
for trial in range(5):
    if trial == 0:
        q = arts[0]["embedding"]          # exact article
    elif trial == 1:
        q = arts[777]["embedding"]        # another exact article
    else:
        q = [random.gauss(0,1) for _ in range(64)]  # random vector
    got = subprocess.run(["python3","/app/search.py"], input=json.dumps(q),
                         capture_output=True, text=True).stdout.split()
    exp = gt_top5(q)
    match = got == exp
    ok &= match
    print("trial", trial, "match:", match, "| got:", got, "| exp:", exp)

print("ALL OK" if ok else "MISMATCH")
EOF

14  $ grep -n "articles.jsonl\|/data/" /app/search.py || echo "search.py does NOT read /data"; echo ---; grep -n -i "^save\|^appendonly\|^rdb" /etc/valkey/valkey.conf; echo ---; cat /app/search.py

15  $ valkey-cli LASTSAVE; valkey-cli CONFIG GET save

16  $ valkey-cli SAVE && valkey-cli SHUTDOWN NOSAVE; sleep 1; valkey-server /etc/valkey/valkey.conf; sleep 2; valkey-cli ping; valkey-cli FT.INFO articles | grep -A1 num_docs | head -2; valkey-cli DBSIZE; echo '[]' | python3 -c "import sys,json; json.dump([0.0]*64, sys.stdout)" | python3 /app/search.py

17  $ cd /tmp && python3 -c "
import json
lines = open('/data/articles.jsonl').readlines()
json.dump(json.loads(lines[42])['embedding'], open('query.json','w'))
" && python3 /app/search.py < query.json && echo "---" && head -1 query.json | cut -c1-80

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ valkey-server --version; echo ---; ls -la /data/ /app/ 2>/dev/null; echo ---; head -c 500 /data/articles.jsonl; echo; echo ---; wc -l /data/articles.jsonl

2  $ valkey-cli INFO server | head -30; echo ---; valkey-cli INFO modules; echo ---; valkey-cli MODULE LIST

3  $ cat /app/catalog.py; echo ===; python3 -c "import redis; print(redis.__version__)"; python3 -c "import valkey; print('valkey-py')" 2>&1 | tail -1; pip list 2>/dev/null | head -40

4  $ cat > /app/load.py <<'EOF'
import json, struct, sys
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)

# clean previous state
try:
    for name in r.execute_command("FT._LIST"):
        r.execute_command("FT.DROPINDEX", name, b"DD")
except redis.ResponseError as e:
    print("FT._LIST/DROP:", e)
for k in r.scan_iter("article:*"):
    r.delete(k)

# create index on hashes with exact KNN over cosine distance
r.execute_command(
    "FT.CREATE", "articles_idx",
    "ON", "HASH",
    "PREFIX", "1", "article:",
    "SCHEMA",
    "id", "TEXT",
    "title", "TEXT",
    "embedding", "VECTOR", "FLAT", "6",
    "TYPE", "FLOAT32",
    "DIM", "64",
    "DISTANCE_METRIC", "COSINE",
)

n = 0
pipe = r.pipeline(transaction=False)
with open("/data/articles.jsonl") as f:
    for line in f:
        a = json.loads(line)
        emb = struct.pack("<64f", *a["embedding"])
        key = "article:" + a["id"]
        pipe.hset(key, mapping={"id": a["id"], "title": a["title"], "embedding": emb})
        n += 1
        if n % 200 == 0:
            pipe.execute()
            pipe = r.pipeline(transaction=False)
pipe.execute()
print("loaded", n, "articles")
print("FT._LIST:", r.execute_command("FT._LIST"))
info = r.execute_command("FT.INFO", "articles_idx")
print("index num_docs:", info[info.index(b"num_docs")+1] if b"num_docs" in info else "?")
EOF
python3 /app/load.py

5  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over the Valkey-hosted article embeddings.

Usage:  python3 /app/search.py < query.json

Reads one query embedding (JSON list of 64 floats) from stdin and prints the
ids of the 5 most similar articles (cosine similarity), most similar first.
The similarity search itself is executed inside Valkey via an exact KNN query
(FT.SEARCH, FLAT index, COSINE distance) -- no local computation of
similarities and no access to the source data file.
"""
import json
import struct
import sys

import redis

INDEX = "articles_idx"
TOP_K = 5


def main() -> None:
    embedding = json.load(sys.stdin)
    if len(embedding) != 64:
        raise SystemExit(f"expected 64 floats, got {len(embedding)}")
    vec = struct.pack("<64f", *embedding)

    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)

    # Exact k-nearest-neighbour search run server-side inside Valkey.
    # COSINE distance = 1 - cosine similarity, returned ascending,
    # which is exactly cosine similarity descending.
    query = f"*=>[KNN {TOP_K} @embedding $vec]"
    res = r.execute_command(
        "FT.SEARCH", INDEX, query,
        "PARAMS", "2", "vec", vec,
        "RETURN", "1", "id",
        "DIALECT", "2",
        "LIMIT", "0", str(TOP_K),
        "NOCONTENT",
    )

    # reply: [total_count, key1, key2, ...] with NOCONTENT
    keys = res[1:]
    for key in keys:
        # keys look like "article:kb-0123"; prefer the indexed `id` field
        art_id = r.hget(key, "id")
        print(art_id.decode() if art_id else key.decode().split(":", 1)[-1])


if __name__ == "__main__":
    main()
EOF
chmod +x /app/search.py
# build a test query: use the embedding of article kb-042 plus a random one
python3 - <<'EOF'
import json, random
lines = open("/data/articles.jsonl").read().splitlines()
a = json.loads(lines[41])
json.dump(a["embedding"], open("/tmp/q1.json","w"))
random.seed(0)
json.dump([random.gauss(0,1) for _ in range(64)], open("/tmp/q2.json","w"))
print("self-query should be kb-0042 first")
EOF
echo "--- q1 (should be kb-0042 first) ---"
python3 /app/search.py < /tmp/q1.json
echo "--- q2 (random) ---"
python3 /app/search.py < /tmp/q2.json

6  $ python3 - <<'EOF'
import json, math, random, subprocess

arts = [json.loads(l) for l in open("/data/articles.jsonl")]
by_id = {a["id"]: a["embedding"] for a in arts}

def cos(a, b):
    da = math.sqrt(sum(x*x for x in a)); db = math.sqrt(sum(x*x for x in b))
    return sum(x*y for x, y in zip(a, b)) / (da*db)

def expected(q, k=5):
    return [i for i, _ in sorted(((a["id"], cos(q, a["embedding"])) for a in arts), key=lambda t: -t[1])[:k]]

random.seed(1)
queries = [json.loads(open("/tmp/q1.json").read()),
           json.loads(open("/tmp/q2.json").read()),
           [random.gauss(0,1) for _ in range(64)],
           json.loads(arts[777]["embedding"]),
           [0.0]*63 + [1.0]]
ok = True
for i, q in enumerate(queries):
    exp = expected(q)
    got = subprocess.run(["python3","/app/search.py"], input=json.dumps(q),
                         capture_output=True, text=True).stdout.split()
    match = exp == got
    ok &= match
    print(f"q{i}: {'OK' if match else 'MISMATCH'}  exp={exp} got={got}")
print("ALL OK" if ok else "FAILURES")
EOF

7  $ grep -rn "articles.jsonl\|/data" /app/search.py; echo "grep-exit:$?"; valkey-cli DBSIZE; valkey-cli FT.INFO articles_idx | grep -A1 num_docs

8  $ python3 - <<'EOF'
import json, math, random, subprocess

arts = [json.loads(l) for l in open("/data/articles.jsonl")]

def cos(a, b):
    da = math.sqrt(sum(x*x for x in a)); db = math.sqrt(sum(x*x for x in b))
    return sum(x*y for x, y in zip(a, b)) / (da*db)

def expected(q, k=5):
    return [i for i, _ in sorted(((a["id"], cos(q, a["embedding"])) for a in arts), key=lambda t: -t[1])[:k]]

random.seed(1)
queries = [json.load(open("/tmp/q1.json")),
           json.load(open("/tmp/q2.json")),
           [random.gauss(0,1) for _ in range(64)],
           arts[777]["embedding"],
           [0.0]*63 + [1.0]]
ok = True
for i, q in enumerate(queries):
    exp = expected(q)
    got = subprocess.run(["python3","/app/search.py"], input=json.dumps(q),
                         capture_output=True, text=True).stdout.split()
    match = exp == got
    ok &= match
    print(f"q{i}: {'OK' if match else 'MISMATCH'}  exp={exp} got={got}")
print("ALL OK" if ok else "FAILURES")
EOF

9  $ valkey-cli FT._LIST; valkey-cli HGET article:kb-0001 title; valkey-cli HLEN article:kb-0001; valkey-cli DBSIZE; echo ---; python3 /app/search.py < /tmp/q1.json

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}

1  $ head -c 500 /data/articles.jsonl; echo; wc -l /data/articles.jsonl; ls /app 2>/dev/null; valkey-cli ping; valkey-cli info server | head

2  $ valkey-cli info modules; valkey-cli command docs FT.CREATE 2>/dev/null | head -c 300; echo; which redis-cli valkey-cli python3; python3 -c "import redis; print(redis.__version__)" 2>&1

3  $ cat /app/catalog.py

4  $ valkey-cli FT.CREATE --help 2>&1 | head; valkey-cli FT.SEARCH --help 2>&1 | head; grep -rn "search" /etc/valkey/valkey.conf | head

5  $ mkdir -p /app && cat > /app/load.py <<'EOF'
import json, struct, sys
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)

# drop old index/keys if re-running
try:
    r.execute_command("FT.DROPINDEX", "idx:articles", "DD")
except redis.ResponseError:
    pass

r.execute_command(
    "FT.CREATE", "idx:articles",
    "ON", "JSON",
    "PREFIX", "1", "article:",
    "SCHEMA",
    "id", "TEXT",
    "title", "TEXT",
    "embedding", "VECTOR", "HNSW", "6",
    "TYPE", "FLOAT32", "DIM", "64", "DISTANCE_METRIC", "COSINE",
)

n = 0
with open("/data/articles.jsonl") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        art = json.loads(line)
        key = "article:" + art["id"]
        doc = {"id": art["id"], "title": art["title"], "embedding": art["embedding"]}
        r.execute_command("JSON.SET", key, "$", json.dumps(doc))
        n += 1

print("loaded", n, file=sys.stderr)
info = r.execute_command("FT.INFO", "idx:articles")
print(info[:20], file=sys.stderr)
EOF
python3 /app/load.py

6  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over Valkey (FT.SEARCH kNN, cosine similarity).

stdin:  a JSON list of 64 floats (the query embedding)
stdout: ids of the 5 most similar articles, most similar first

The similarity search runs inside Valkey via the search module; this script
only sends the query vector and prints the returned ids.
"""
import json
import struct
import sys

import redis

INDEX = "idx:articles"
TOP_K = 5


def main() -> None:
    query = json.load(sys.stdin)
    vec = struct.pack(f"<{len(query)}f", *query)  # raw FLOAT32 bytes

    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)
    res = r.execute_command(
        "FT.SEARCH", INDEX,
        f"*=>[KNN {TOP_K} @embedding $vec AS score]",
        "PARAMS", "2", "vec", vec,
        "SORTBY", "score", "ASC",
        "RETURN", "1", "__key",
        "DIALECT", "2",
        "LIMIT", "0", TOP_K,
    )

    # Response: [total, key1, doc1, key2, doc2, ...]
    keys = res[1::2]
    for key in keys:
        print(key.decode().split(":", 1)[1])


if __name__ == "__main__":
    main()
EOF
chmod +x /app/search.py
# Build a query from article 42's embedding and test
python3 - <<'EOF'
import json
line = open('/data/articles.jsonl').readlines()[41]
art = json.loads(line)
json.dump(art['embedding'], open('/tmp/q.json','w'))
print("query from", art['id'], art['title'])
EOF
python3 /app/search.py < /tmp/q.json

7  $ python3 /app/search.py < /tmp/q.json 2>&1 | tail -5

8  $ VEC=$(python3 -c "
import json,struct
q=json.load(open('/tmp/q.json'))
print(struct.pack('<64f',*q).hex())
")
echo "--- pure knn query ---"
redis-cli FT.SEARCH idx:articles "=>[KNN 5 @embedding \$vec AS score]" PARAMS 2 vec "$(printf $(echo $VEC | sed 's/../\\x&/g'))" RETURN 1 __key DIALECT 2 LIMIT 0 5
echo "--- hybrid, no sortby ---"
redis-cli FT.SEARCH idx:articles "*=>[KNN 5 @embedding \$vec AS score]" PARAMS 2 vec "$(printf $(echo $VEC | sed 's/../\\x&/g'))" RETURN 1 __key DIALECT 2 LIMIT 0 5

9  $ python3 - <<'EOF'
import json, struct, redis
r = redis.Redis(decode_responses=False)
q = json.load(open('/tmp/q.json'))
vec = struct.pack('<64f', *q)

def try_query(label, *extra):
    try:
        res = r.execute_command("FT.SEARCH", "idx:articles",
            "*=>[KNN 5 @embedding $vec AS score]",
            "PARAMS", "2", "vec", vec, *extra)
        print(label, "OK ->", res[1::2])
    except redis.ResponseError as e:
        print(label, "ERR:", e)

try_query("plain", "RETURN", "1", "__key", "DIALECT", "2", "LIMIT", "0", "5")
try_query("sortby score", "SORTBY", "score", "ASC", "RETURN", "1", "__key", "DIALECT", "2", "LIMIT", "0", "5")
try_query("pure knn", "DIALECT", "2", "LIMIT", "0", "5")
try_query("sortby with no alias", "SORTBY", "embedding", "ASC", "RETURN", "1", "__key", "DIALECT", "2", "LIMIT", "0", "5")
EOF

10  $ cat > /app/search.py <<'EOF'
#!/usr/bin/env python3
"""Semantic search over Valkey (Valkey Search module, kNN / cosine).

stdin:  a JSON list of 64 floats (the query embedding)
stdout: ids of the 5 most similar articles, most similar first

The similarity search runs entirely inside Valkey (FT.SEARCH with a KNN
vector query); this script only ships the query vector to the server and
prints the ids of the documents Valkey returns.
"""
import json
import struct
import sys

import redis

INDEX = "idx:articles"
TOP_K = 5


def main() -> None:
    query = json.load(sys.stdin)
    if len(query) != 64:
        sys.exit(f"expected 64 floats, got {len(query)}")
    # Raw little-endian FLOAT32 blob, the format the index stores.
    vec = struct.pack("<64f", *query)

    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)
    # Pure kNN query: Valkey ranks by cosine distance and returns the
    # top TOP_K documents, nearest first.
    res = r.execute_command(
        "FT.SEARCH", INDEX,
        f"=>[KNN {TOP_K} @embedding $vec]",
        "PARAMS", "2", "vec", vec,
        "DIALECT", "2",
        "LIMIT", "0", TOP_K,
    )

    for key in res[1::2]:  # [total, key1, doc1, key2, doc2, ...]
        print(key.decode().split(":", 1)[1])


if __name__ == "__main__":
    main()
EOF
cat > /tmp/verify.py <<'EOF'
import json, math, random, subprocess

arts = [json.loads(l) for l in open('/data/articles.jsonl')]
vecs = [a['embedding'] for a in arts]

def top5(q):
    s = []
    nq = math.sqrt(sum(x*x for x in q))
    for a, v in zip(arts, vecs):
        d = sum(x*y for x, y in zip(q, v))
        s.append((-d / (nq * math.sqrt(sum(x*x for x in v))), a['id']))
    return [i for _, i in sorted(s)[:5]]

random.seed(7)
cases = [vecs[0], vecs[500], vecs[999]]
cases += [[random.uniform(-1, 1) for _ in range(64)] for _ in range(7)]

ok = True
for i, q in enumerate(cases):
    out = subprocess.run(['python3', '/app/search.py'], input=json.dumps(q),
                         capture_output=True, text=True)
    got = out.stdout.split()
    exp = top5(q)
    match = got == exp
    ok &= match
    print(f"case {i}: {'OK ' if match else 'FAIL'} got={got} exp={exp}")
    if out.stderr.strip():
        print("  stderr:", out.stderr.strip())
print("ALL OK" if ok else "MISMATCHES FOUND")
EOF
python3 /tmp/verify.py

11  $ redis-cli FT.CREATE idx:flattest ON JSON PREFIX 1 article: SCHEMA id TEXT title TEXT embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 64 DISTANCE_METRIC COSINE 2>&1
redis-cli FT.DROPINDEX idx:flattest DD 2>/dev/null; echo done

12  $ python3 - <<'EOF'
import json, struct, redis
r = redis.Redis(decode_responses=False)
q = json.load(open('/tmp/q.json'))
vec = struct.pack('<64f', *q)
for qstr in ["=>[KNN 5 @embedding $vec EF_RUNTIME 200]",
             "*=>[KNN 5 @embedding $vec EF_RUNTIME 200]"]:
    try:
        res = r.execute_command("FT.SEARCH", "idx:articles", qstr,
                                "PARAMS", "2", "vec", vec, "DIALECT", "2", "LIMIT", "0", "5")
        print(qstr, "OK", res[1::2])
    except redis.ResponseError as e:
        print(qstr, "ERR:", e)
EOF

13  $ redis-cli FT.DROPINDEX idx:articles DD
sed -i 's/"VECTOR", "HNSW", "6"/"VECTOR", "FLAT", "6"/' /app/load.py
python3 /app/load.py && python3 /tmp/verify.py

14  $ redis-cli FT.DROPINDEX idx:articles; echo "---"; redis-cli FT.DROPINDEX idx:articles DD; echo "---"; redis-cli FT.DROPINDEX

15  $ redis-cli FT.DROPINDEX idx:flattest
sed -i 's/"DD")/"NoDD") if False else None/' /app/load.py  # no-op guard not needed; check current drop block
grep -n "DROPINDEX" -A2 /app/load.py

16  $ cat > /app/load.py <<'EOF'
"""Load the 1,000 knowledge-base articles from /data/articles.jsonl into
Valkey and build an exact (FLAT) cosine vector index over their embeddings."""
import json
import sys

import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=False)

# Idempotent re-run: remove any previous index and articles.
try:
    r.execute_command("FT.DROPINDEX", "idx:articles")
except redis.ResponseError:
    pass
for key in r.scan_iter("article:*"):
    r.delete(key)

# Exact kNN (brute-force FLAT) index over 64-dim FLOAT32 cosine vectors.
r.execute_command(
    "FT.CREATE", "idx:articles",
    "ON", "JSON",
    "PREFIX", "1", "article:",
    "SCHEMA",
    "id", "TEXT",
    "title", "TEXT",
    "embedding", "VECTOR", "FLAT", "6",
    "TYPE", "FLOAT32", "DIM", "64", "DISTANCE_METRIC", "COSINE",
)

n = 0
with open("/data/articles.jsonl") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        art = json.loads(line)
        doc = {"id": art["id"], "title": art["title"], "embedding": art["embedding"]}
        r.execute_command("JSON.SET", "article:" + art["id"], "$", json.dumps(doc))
        n += 1

print("loaded", n, "articles", file=sys.stderr)
EOF
python3 /app/load.py && python3 /tmp/verify.py

17  $ cat > /tmp/verify2.py <<'EOF'
import json, math, random, subprocess
arts = [json.loads(l) for l in open('/data/articles.jsonl')]
vecs = [a['embedding'] for a in arts]
norms = [math.sqrt(sum(x*x for x in v)) for v in vecs]
def top5(q):
    nq = math.sqrt(sum(x*x for x in q))
    s = sorted((-(sum(x*y for x,y in zip(q,v)))/(nq*n2), a['id'])
               for a,v,n2 in zip(arts,vecs,norms))
    return [i for _,i in s[:5]]
random.seed(99)
bad = 0
for t in range(50):
    if t % 5 == 0:                       # sometimes query near real articles
        q = [x + random.uniform(-.05,.05) for x in vecs[random.randrange(1000)]]
    else:                                # sometimes pure random vector
        q = [random.uniform(-1,1) for _ in range(64)]
    out = subprocess.run(['python3','/app/search.py'], input=json.dumps(q),
                         capture_output=True, text=True, check=True)
    got, exp = out.stdout.split(), top5(q)
    if got != exp:
        bad += 1; print("MISMATCH", got, exp)
print(f"47/50 checked... mismatches: {bad}")
EOF
python3 /tmp/verify2.py

18  $ # sanity: exact invocation form, no stderr, exactly 5 lines
python3 -c "
import json,random;random.seed(1);json.dump([random.uniform(-1,1) for _ in range(64)], open('/tmp/query.json','w'))"
python3 /app/search.py < /tmp/query.json; echo "exit=$?"
echo "--- index/doc state ---"
redis-cli FT.INFO idx:articles | grep -E "num_docs|index_name|algorithm|distance_metric|dim"
redis-cli DBSIZE
redis-cli FT.SEARCH idx:articles "=>[KNN 3 @embedding $v]" PARAMS 2 v "$(python3 -c "
import struct,json,sys
q=json.load(open('/tmp/query.json'));sys.stdout.buffer.write(struct.pack('<64f',*q))")" DIALECT 2 LIMIT 0 3 | head -2
grep -c "articles.jsonl\|/data/" /app/search.py || true

19  $ grep -vE "^\s*#|^\s*$" /etc/valkey/valkey.conf; redis-cli CONFIG GET save; redis-cli CONFIG GET appendonly; redis-cli CONFIG GET dir; redis-cli LASTSAVE; ls -la /var/lib/valkey 2>/dev/null || true

20  $ cat /app/search.py; echo "===="; cd /tmp && python3 /app/search.py < /tmp/query.json

21  $ redis-cli SAVE; ls -la /var/lib/valkey; redis-cli FT.INFO idx:articles | awk '/num_docs|num_records/{getline v; print $0, v}' | head -4; redis-cli JSON.GET article:kb-0001 | head -c 200; echo; python3 /app/search.py < /tmp/q.json

{
 "dbsize": 1000,
 "queries": [
  {
   "exit": 0,
   "ids": [
    "kb-0468",
    "kb-0368",
    "kb-0028",
    "kb-0088",
    "kb-0708"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0215",
    "kb-0155",
    "kb-0315",
    "kb-0535",
    "kb-0615"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0963",
    "kb-0123",
    "kb-0423",
    "kb-0583",
    "kb-0823"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0237",
    "kb-0597",
    "kb-0097",
    "kb-0337",
    "kb-0837"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  },
  {
   "exit": 0,
   "ids": [
    "kb-0541",
    "kb-0621",
    "kb-0841",
    "kb-0581",
    "kb-0481"
   ],
   "overlap": 5,
   "stderr": "",
   "top1_ok": true
  }
 ],
 "server": {
  "redis_version": "7.2.4",
  "server_name": "valkey",
  "valkey_version": "9.1.2"
 },
 "server_side_search_commands": [
  "FT.SEARCH"
 ]
}