ground

Your sources for grounding, and the links from each concept to your own work.

The tutor writes better questions when it can see what you read and what you did. This module gives it that view.

• A source finds short passages for a query. It reads only.
• ground asks every source and merges the passages.
• Learning.connect keeps the best passages for a concept as its links: where the concept shows up in your work.
• Two indexers put your ramabana sessions and leela notebooks on vault shelves, so every tool can search them.

These are the imports of the module.

Passages

A passage is one short piece of text and where it came from. src names the source, ref points into it, and url opens it when there is a page for it.

---

source

cite

def cite(
    p
)->str:

A short citation for a passage: its source and its title.

---

source

passage

def passage(
    src, ref, title, text, url:str='', **kw
)->fastcore.basics.AttrDict:

One passage from a source.

Query words

The FTS5 sources need a safe query. fts_query keeps the words of three letters or more and joins them with OR. Marks such as ā and ṣ stay, because vedicreader stores IAST.

---

source

fts_query

def fts_query(
    q:str, n:int=8
)->str:

An FTS5 query that matches any of the first n words of q with three letters or more.

Words of three letters or more stay, with their marks. With only short words, the query is empty.

test_eq(fts_query('Vowel sandhi: a + a → ā'), '"vowel" OR "sandhi"')
test_eq(fts_query('pramāṇa and anumāna'), '"pramāṇa" OR "and" OR "anumāna"')
test_eq(fts_query('a b'), '')

Vault shelves

VaultSource searches every shelf of the vault except the learning shelf. It is the widest source: your documents, leela’s notes, and the session shelves that the indexers below make.

---

source

VaultSource

def VaultSource(
    learning:abhyasa.store.Learning, shelves:bool=True
):

Passages from every vault shelf except the learning shelf.

The vedicreader library

VedicSource reads the verse databases of the vedicreader repo, data/db/shlokas/*.db, in read-only mode. It searches the English meaning of each section through the sections_fts table, and names the text from the titles table.

---

source

VedicSource

def VedicSource(
    root:NoneType=None
):

Passages from the vedicreader library: the meaning of each section of each text.

When the repo is on this machine, this live check searches it:

vs = VedicSource()
if vs.dbs():
    ps = vs.search('inference from a sign, as smoke shows fire')
    assert ps and ps[0].src == 'vedicreader'
    print(cite(ps[0]), ps[0].text[:120])

Ground a query

ground asks each source for passages and takes them in turn, so one source cannot fill the list.

---

source

ground_text

def ground_text(
    ps
)->str:

Passages as prompt text, one line each, with citations.

---

source

ground

def ground(
    q:str, sources, k:int=3, must:tuple=()
)->fastcore.foundation.L:

Up to k passages for q, taken from the sources in turn. With must, a passage must name one of those terms.

---

source

names

def names(
    p, terms
)->bool:

Whether passage p names one of terms as a whole word, in its title or its text.

---

source

fold

def fold(
    s:str
)->str:

Lower case without marks, so Anumāna and anumana match.

Two stand-in sources give numbered passages. ground takes them in turn.

class Src:
    def __init__(self, name, n): self.name, self.n = name, n
    def search(self, q, k=4): return L(passage(self.name, str(i), f'{self.name}{i}', q) for i in range(self.n))[:k]
test_eq(ground('x', [Src('a', 3), Src('b', 1)], k=3).attrgot('title'), ['a0', 'b0', 'a1'])

A link must name the concept. names compares whole words without marks, so a passage on anumāna links to Anumana, and a passage on rank fusion does not link to rank.

test_eq(fold('Anumāna'), 'anumana')
assert names(passage('s', '1', 'Nyāya notes', 'On anumāna and vyāpti.'), ['Anumana'])
assert not names(passage('s', '1', 'Notes', 'ranked lists'), ['rank'])
test_eq(ground('x', [Src('alpha', 3)], k=3, must=['alpha1']).attrgot('title'), ['alpha1'])

Links from a concept to your work

connect searches the sources for one concept and keeps the best five passages in the seen_in meta of the concept. It does not keep the text, only the place of the text, so the links stay small.

---

source

Learning.stale_links

def stale_links(
    c, age:float=86400.0
)->bool:

Whether the links of concept c are missing or older than age seconds.

---

source

Learning.connect

def connect(
    slug:str, sources, k:int=5
)->list:

Find where concept slug shows up in your sources, and keep the links on the concept.

Index ramabana sessions

ramabana writes one JSON line for each turn to agent-history.jsonl, and the session titles to agent-sessions.json. index_ramabana makes one vault document for each session on the ramabana shelf. It skips a session whose text did not change, so a second run is fast. It only adds documents to its own shelf.

ramabana also keeps memory: one MEMORY.md for each project, and what dhrona learned (learned/AGENTS.md and its skills). ramabana_memory reads them. index_ramabana puts each one on the same shelf, with kind memory.

---

source

ramabana_memory

def ramabana_memory(
    cfg:NoneType=None
):

ramabana’s memory as (ref, title, text): each project’s MEMORY.md, and the learned AGENTS.md and skills.

ramabana_sessions reads the history. index_ramabana puts the sessions and the memory on the shelf. RAMA_CFG is the ramabana config folder: $RAMABANA_CFG, else ~/.config/ramabana.

---

source

index_ramabana

def index_ramabana(
    vault, cfg:NoneType=None, shelf:str='ramabana', dry_run:bool=False
)->dict:

Put each ramabana session on the vault shelf shelf, one document each. Returns counts.

---

source

ramabana_sessions

def ramabana_sessions(
    cfg:NoneType=None, history:str='agent', max_turn:int=2000, max_doc:int=60000
):

Each ramabana session as (id, title, text, meta), from the history file.

This test makes a small history in a temporary folder, reads it and indexes it two times:

import tempfile
from abhyasa.store import temp_learning
d = Path(tempfile.mkdtemp())
(d/'memory/abc').mkdir(parents=True); (d/'memory/abc/MEMORY.md').write_text('- Use RRF for hybrid search.')
(d/'agent-sessions.json').write_text(json.dumps(dict(sessions={'s1': dict(title='Fix the RRF merge', roots=['/code'])})))
(d/'agent-history.jsonl').write_text('\n'.join(json.dumps(r) for r in [
    dict(session='s1', at=1., prompt='<user-request>why does rrf use ranks</user-request>', reply='Because scores differ in scale.', model='m'),
    dict(session='s1', at=2., prompt='and k?', reply='k damps the top ranks.', model='m'),
    dict(session='s2', at=3., prompt='sandhi of a + i', reply='It gives e.', model='m')]) + '\nnot json\n')
ss = list(ramabana_sessions(d))
test_eq([s[1] for s in ss], ['Fix the RRF merge', 'sandhi of a + i'])
assert ss[0][2].startswith('## why does rrf use ranks')
lr = temp_learning()
test_eq(index_ramabana(lr.vault, d), dict(sessions=2, memory=1, changed=3))
test_eq(index_ramabana(lr.vault, d), dict(sessions=2, memory=1, changed=0))     # nothing changed: nothing written
test_eq(index_ramabana(lr.vault, d, dry_run=True), dict(sessions=2, memory=1, changed=0))

With no history file yet, there are no sessions, and no error.

test_eq(list(ramabana_sessions(Path(tempfile.mkdtemp()))), [])

Index leela notebooks

leela keeps agent sessions as notebooks in _leela_agent_sessions folders, and you keep your own notebooks next to your code. index_notebooks puts each notebook on the notebooks shelf: its markdown, its code and its text outputs. It skips .venv, node_modules and other build folders.

---

source

index_notebooks

def index_notebooks(
    vault, roots, shelf:str='notebooks', only_agent:bool=False, dry_run:bool=False
)->dict:

Put each notebook under roots that is new or changed on the vault shelf shelf. Returns counts.

---

source

nb_text

def nb_text(
    p, max_out:int=1000, max_doc:int=60000
)->str:

The readable text of a notebook: markdown, code in fences, and text outputs.

---

source

find_notebooks

def find_notebooks(
    roots, only_agent:bool=False
):

Every .ipynb under roots, or only those in _leela_agent_sessions folders.

Two notebooks in a temporary folder: one in _leela_agent_sessions and one beside it. A third notebook in .venv is skipped. The first run indexes both.

from fastcore.nbio import new_nb, mk_cell, write_nb
root = Path(tempfile.mkdtemp())
(root/'proj/_leela_agent_sessions').mkdir(parents=True); (root/'proj/.venv').mkdir()
write_nb(new_nb([mk_cell('# RRF notes', 'markdown'), mk_cell('1 + 1')]), root/'proj/_leela_agent_sessions/a.ipynb')
write_nb(new_nb([mk_cell('x = 1')]), root/'proj/b.ipynb')
write_nb(new_nb([mk_cell('y = 2')]), root/'proj/.venv/c.ipynb')
test_eq(sorted(p.name for p in find_notebooks([root])), ['a.ipynb', 'b.ipynb'])
test_eq([p.name for p in find_notebooks([root], only_agent=True)], ['a.ipynb'])
assert nb_text(root/'proj/_leela_agent_sessions/a.ipynb').startswith('# RRF notes')
test_eq(index_notebooks(lr.vault, [root]), dict(notebooks=2, changed=2, failed=0, same=0))

A second run reads no notebook again. A notebook that changes is read again.

test_eq(index_notebooks(lr.vault, [root]), dict(notebooks=2, changed=0, failed=0, same=2))
write_nb(new_nb([mk_cell('x = 2')]), root/'proj/b.ipynb'); os.utime(root/'proj/b.ipynb', (1e9, 1e9))
test_eq(index_notebooks(lr.vault, [root]), dict(notebooks=2, changed=1, failed=0, same=1))

The sources together

With the two shelves in the vault, VaultSource finds sessions and notebooks too. connect then links a concept to them.

lr.put('Reciprocal rank fusion', 'Merge ranked lists by summing 1/(k + rank).', 'ml', aliases=['RRF'])
links = lr.connect('reciprocal-rank-fusion', [VaultSource(lr)])
assert any(l['src'] == 'vault:ramabana' for l in links), links
assert not lr.stale_links(lr.concept('reciprocal-rank-fusion'))

A ramabana tick

The ramabana beat runs every five minutes. It runs each pobblebonk schedule that is due, through the ramabana.ticks entry point with the same name. abhyasa gives the schedule abhyasa-index. Each fire indexes new and changed sessions, memory and leela agent notebooks. For a session that did not change, the tick only reads it. Thus a fire takes seconds.

The tick leaves a note for the next ramabana session only when something changed.

---

source

tick

def tick(
    fire, vault:NoneType=None
)->str | None:

One fire of the abhyasa-index schedule: index sessions, memory and leela agent notebooks. A note when something changed.

A fire on the temporary history from above: the first one writes and leaves a note, the second finds nothing new.

v2 = temp_learning().vault
assert 'indexed 3 changed' in tick(dict(cfg=d, roots=[root]), vault=v2)
test_is(tick(dict(cfg=d, roots=[root]), vault=v2), None)

schedule writes the schedule into pobblebonk once. Run it again to change how often it fires. It needs pobblebonk, which ramabana installs.

---

source

schedule

def schedule(
    every:str='15m', db:NoneType=None, **payload
)->dict:

Add or change the abhyasa-index schedule in pobblebonk, so ramabana’s beat indexes your sessions.

A test on a temporary pobblebonk database. A second call changes the schedule, and does not add another one.

pdb = Path(tempfile.mkdtemp())/'pob.db'
test_eq(schedule('15m', db=pdb)['name'], TICK)
schedule('1h', db=pdb)
from pobblebonk.core import Pob
test_eq([r['name'] for r in Pob(str(pdb)).all()], [TICK])

Command line

abhyasa-index ramabana and abhyasa-index notebooks fill the shelves of your vault. --dry-run counts and writes nothing.

---

source

index_main

def index_main(
    argv:NoneType=None
):

abhyasa-index: put ramabana sessions or notebooks on vault shelves.