INPROGRESS Review and refigure this
well here it is, a demonstration of using my document search to distill my previous Privacy works in to some key themes in my Self Hosting work and the Concept Operating System.
This is a Literate Programming document showing the implementation of a vector search and generative text pipeline with RAG built on top of the document vector search. With it, one can ask interesting questions of their org-roam Knowledge Base and along the way construct a technopagan monolith to an AI that will kill you if you don't give your money to my AI research buddies.
INPROGRESS vector search in The Arcology Project
INPROGRESS what does vector search get us
we can use ollama to generate vectors from arbitrary text embedded in to an LLM model's internal vector space. You can do this to all your docs, and then to queries, and find other documents "nearby" your queries
DONE package sqlite-vec and write a django middleware for it
sqlite-vec provides C library modules with the vector functions and a virtual table for storing pk/vector tuples. I wrote a small Django middleware for it in django-sqlite-vec that exposes the table column types and functions. I oughta expose that work sooner or later, i don't think I've upstreamed it yet.
DONE vector search models
from django.db import modelsDocuments track metadata, a document type, and generated summaries
from typing import Optional
from oracle.ollama_embed import embed, embed_model
from oracle.summarizer import OllamaContextualSummarizer, OllamaRefineSummarizer
CHUNK_SIZE = 6000
CHUNK_OVERLAP = 400
from roam.models import calculate_hash
import logging
logger = logging.getLogger(__name__)
DOC_TYPES = [
("org", "Notes and Journal"),
("book", "Ebook or PDF"),
("chat", "Chat Log Snippet"),
]
class Document(models.Model):
path = models.TextField(primary_key=True)
dtype = models.CharField(max_length=32, choices=DOC_TYPES)
summary = models.TextField(default="", blank=True)
embed_hash = models.CharField(max_length=256, blank=True)
@classmethod
def create_from_arroyo(cls, doc):
try:
cls.objects.get(path=doc.path).delete()
except cls.DoesNotExist:
pass
doc = cls.objects.create(
path=doc.path,
dtype="org",
)
# doc.embed()
return docDeletion of Files when the Arcology ingestfiles Command runs won't get cascade to this model by a foreign-key relationship with roam.models.File so we have to try to delete it before recreating it.
From playing with langchain I learned that the chunks are overlapping, so you for example take 2000 chars and then go forward 1800 and take another 2000, and embed each of those chunks. this gets you some good idea that ideas won't be cut in half and lost.
each chunk is fed to an ollama embedding model, a vector is returned and stored somewhere (in a sqlite virtual table)
def embed(self):
def chunk_overlap(seq, chunk_size, overlap):
for i in range(0, len(seq) - overlap, chunk_size - overlap):
yield i, seq[i:i+chunk_size]
self.embed_hash = calculate_hash(self.path)
with open(self.path, "r") as f:
text = f.read()
for start_idx, snippet in chunk_overlap(text, CHUNK_SIZE, CHUNK_OVERLAP):
c = Chunk(
document=self,
embedded_in=embed_model(),
chunk_length=CHUNK_SIZE,
chunk_start=start_idx,
)
c.save()
ce = ChunkEmbedding(
chunk=c,
embedding=embed(snippet)
)
ce.save()There is also a summarizer which will be explored below. You can search for relevant Chunks using the knn below, go from the ChunkEmbedding to the Chunk to the Document and call a summarizer defined even further in this document with that document's text. You can pass it a context string to include that in the Summarizer's prompts
def summarize(self, force=False, persist=False, ctx: Optional[str]=None):
# if not force and self.summary != "":
# return self.summary
if ctx is not None:
summarizer = OllamaContextualSummarizer(self, ctx)
else:
summarizer = OllamaRefineSummarizer(self)
self.summary = summarizer.summarize()
if persist:
self.save()
return self.summarythey have a file-path to point to the original source, but do not necessary have foreignkey relationships to Arcology Roam Models, as nice as that would be. I'd rather have an un-enforced file path foreignkey that can be used in my ebook library and whatnot too
Chunks belong to documents
and they have a vector field, and probably should track which model is used to embed it, also the index in the document and the length of tokens used to generate the embedding, you should be able to call a chunk.raw_text() to load it from the document
that django module i wrote adds a VectorField and then we can access the functions, but you have to do Some Bullshit in the migrations to get that database table set up and used, the model is not fully managed by Django's ORM and migration system since it's a virtual table.
from asgiref.sync import sync_to_async
from django_sqlite_vec.field import VectorField, serialize_f32
MODELS = [
("gemma3:4b", "Gemma3 Small"),
("gemma3:12b", "Gemma3 Large"),
("llama3:8b", "Llama3 8B"),
("llama3:instruct", "Llama3 8B Instruct"),
("llama3.2", "Llama3.2"),
("llama3.3", "Llama3.3"),
("deepseek-r1:7b", "DeepSeek Reasoning (Qwen)"),
("deepseek-r1:14b", "DeepSeek Reasoning (Qwen2)"),
("hermes3:8b", "Hermes 3"),
("phi4", "MS Phi-4"),
("nomic-embed-text", "Nomic Text Embedding"),
]
class Chunk(models.Model):
document = models.ForeignKey(
to=Document,
on_delete=models.CASCADE
)
embedded_in = models.CharField(max_length=32, choices=MODELS)
chunk_length = models.IntegerField()
chunk_start = models.IntegerField()
summary = models.TextField()
def raw_text(self) -> str:
doc_path = self.document.path
with open(doc_path, "r") as f:
try:
f.seek(self.chunk_start)
return f.read(self.chunk_length)
except UnicodeDecodeError: # yes! haha! yes!!!
try:
f.seek(self.chunk_start-1)
return f.read(self.chunk_length)
except UnicodeDecodeError: # yes! haha! yes!!!
f.seek(self.chunk_start-2)
return f.read(self.chunk_length)
class ChunkEmbedding(models.Model):
embedding = VectorField()
chunk = models.OneToOneField(
to=Chunk,
primary_key=True,
on_delete=models.CASCADE,
db_column="id",
db_constraint=False,
)
class Meta:
managed = False
db_table = "oracle_chunk_embedding"
@classmethod
def knn(cls, query, k=20):
query_vec = embed(query)
serialized = serialize_f32(query_vec)
res = cls.objects.raw("""
SELECT id, distance
FROM oracle_chunk_embedding
WHERE embedding match %s AND k = %s
ORDER BY distance;
""", [serialized, k])
return [(obj.chunk.document, obj.distance, obj.chunk.chunk_start) for obj in res]
@classmethod
async def aknn(cls, query, k=20):
return await sync_to_async(cls.knn)(query, k)these should go in django-sqlite-vec...
class VecToJson(models.Func):
function = "vec_to_json"
@models.Field.register_lookup
class VectorMatch(models.Lookup):
lookup_name = "vector_match"
def as_sql(self, compiler, connection):
lhs, lhs_params = self.process_lhs(compiler, connection)
rhs, rhs_params = self.process_rhs(compiler, connection)
params = lhs_params + rhs_params
return f"{lhs} match {rhs}", paramsNEXT add squashed migration here...
it's in git! sorry
you need to run this in a migration:
CREATE VIRTUAL TABLE IF NOT EXISTS "oracle_chunk_embedding" using vec0(
id integer primary key,
embedding float[768] distance_metric=cosine,
)NEXT expose a similarity search API
now with a similarity search some neat things can happen, an org-roam buffer can have similarity searches applied to it to show semantically similar nodes, for example.
There is a knn on the ChunkEmbedding model that you can use to query for documents, I oughta expose that in the Localhost API for the Arcology
INPROGRESS your own personal stochastic parrot
i would create an embedding database w/ your own local data, a simple oracle that one can consult to have a search of all personally collected knowledge and notes and synthesis; 3-8 similar, time-weighted results, which are then summarized / semantically compressed by an LLM in to small snippets, each of these are added to another LLM prompt
Read:
NEXT how do you train an llm from scratch, can you train an llm from scratch
how much text data do you need to create a model that can do compelling similarity search and response generation like this?
what if i just started goign to the library and checking out every book i can and OCRing them one by one? how many books would i need access to?
NEXT how do you justify this work
dark laughter
INPROGRESS let's build one that ever so slowly runs in the living room
let's build an oracular parrot for +0$+ 430$ that runs on commodity hardware. this thing is useful in talking to yourself, via your journal and digital detritus. the arcology captures much of mine.
what if you could ask your computer a complicated question about your inner life, and a few minutes later that computer will have cross referenced your notes and a journal and about a zillion other things people downloaded from the internet without asking, and then generated words that would be likely to be relevant to your question from within the set of all the words on the internet? what a hack. it's not prophetic but perhaps it could be thought provoking.
here's what we're gonna do
a user has a text string that is basically a question they want to ask about things from things like their org-mode notes, wikipedia, their epub book library
we'll search the vector DB for documents that might be relevant, and summarize them
use those summaries to prime the response from an LLM to then go off and return some probable texts
from django.apps import AppConfig
class OracleConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "oracle"We're gonna make some ORM Data Models to store the Chats
the prototype of this thing just used an in-memory model that was passed between the client and the server when the client connected, but it was kind of janky. this is also kind of janky but it forced me to understand a bit better how the channels stuff works by forcing me to put write actions in the right place.
SystemPrompt stores base prompts. These are inserted as =ChatMessage='s in to new =Chat='s and can then be edited as a message.
import arrow
class SystemPrompt(models.Model):
description = models.TextField()
prompt_text = models.TextField()
# XXX how to handle formatting arguments
# need a bunch of bullshit for this probably
def __str__(self):
return self.prompt_text.format(
date=arrow.now().strftime("%A %B %-m %Y, week %W")
)Chat contains messages and lets you route to collections of them and query new text from them.
import arrow
import ollama
from itertools import islice
class Chat(models.Model):
slug = models.CharField(max_length=32, blank=True, primary_key=True)
title = models.TextField(blank=True)
system_prompt = models.ForeignKey(
to=SystemPrompt,
on_delete=models.CASCADE,
)
model = models.CharField(max_length=32, choices=MODELS)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
prompt = kwargs.get("system_prompt", None)
if prompt is not None:
self.add_message(
role="system",
message=str(prompt)
)
def messages_qs(self, filtered):
message_qs = self.chatmessage_set.order_by("ts")
if filtered:
message_qs = message_qs.filter(
role__in=["system", "user", "assistant", "tool"]
)
return message_qs
def messages(self, filtered=True):
return [
message.to_dict()
for message in self.messages_qs(filtered).all()
]
def truncate(self, n=0):
messages = self.messages_qs(filtered=False)
for message in islice(messages, n, None):
message.delete()
return messages[:n]
def add_message(self, role, message):
self.chatmessage_set.create(
role=role,
message=message,
ts=arrow.now().strftime("%Y-%m-%d %H:%M:%S%z")
)
def query(self):
resp = ollama.chat(
model=self.model,
messages=self.messages(filtered=True)
)
return resp['message']
async def aquery(self):
c = ollama.AsyncClient()
messages = await sync_to_async(self.messages)(filtered=True)
async for part in await c.chat(
model=self.model,
messages=messages,
stream=True
):
yield partThe ChatMessage is a simple object to store a message, a sender, and a time and associate those with a Chat to populate their collection. The to_dict method turns the message in to something suitable to send to ollama.
class ChatMessage(models.Model):
role = models.CharField(max_length=64)
message = models.TextField()
ts = models.DateTimeField()
chat = models.ForeignKey(
to=Chat,
on_delete=models.CASCADE,
)
def to_dict(self):
return dict(
id=self.id,
role=self.role,
content=self.message.strip(),
)Stick the Chat stuff in the Admin UI:
from django.contrib import admin
import oracle.models
class ChatMessageInline(admin.TabularInline):
model = oracle.models.ChatMessage
@admin.register(oracle.models.ChatMessage)
class ChatMessageAdmin(admin.ModelAdmin):
list_display=["ts", "role", "message", "chat"]
@admin.register(oracle.models.SystemPrompt)
class PromptAdmin(admin.ModelAdmin):
list_display=["description", "prompt_text"]
@admin.register(oracle.models.Chat)
class ChatAdmin(admin.ModelAdmin):
list_display=["slug", "title", "model", "system_prompt"]
inlines = [
ChatMessageInline
]NEXT minimize and fold in migrations
Some "test" Management Commands to interact with the data models and Ollama
Embed a document's text in to an LLM and store the resulting vectors for search.
from django.core.management.base import BaseCommand
from django.db.models import Q
from django.db import transaction
import oracle.models
import roam.models
import logging
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = 'Generate embeddings'
output_transaction = True
requires_migrations_checks = True
def handle(self, *args, **kwargs):
all_embedded_docs = oracle.models.Document.objects.all().values("embed_hash")
document_digests = [d['embed_hash'] for d in all_embedded_docs]
missing = roam.models.File.objects.filter(~Q(digest__in=document_digests))
logger.info(f"Need to embed {len(missing)} docs.")
for file in missing:
with transaction.atomic():
path = file.path
logger.debug(f"Embedding {path}")
d = oracle.models.Document(path=path, dtype="text")
d.save()
d.embed()
d.save()
all_indexed_files = roam.models.File.objects.all().values("digest")
file_hashes = [d['digest'] for d in all_indexed_files]
stale_docs = oracle.models.Document.objects.filter(~Q(embed_hash__in=file_hashes))
logger.info(f"Need to delete {len(stale_docs)} docs.")
with transaction.atomic():
for doc in stale_docs:
doc.delete()Query your documents with Ollama
from django.core.management.base import BaseCommand
import oracle.models
from oracle.ollama_embed import gen_model
import oracle.summarizer
import logging
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = 'Query the embedding DB for context and ask an LLM'
output_transaction = True
requires_migrations_checks = True
def add_arguments(self, parser):
parser.add_argument('query', type=str)
parser.add_argument('-k', type=int,
help="number of documents to return in context query")
parser.add_argument('-s', type=str, default="oracle",
help="chat slug/identifier")
parser.add_argument('-p', type=str, default="oracle",
help="system prompt descriptor")
def handle(self, *args, **kwargs):
query = kwargs['query']
k = kwargs['k']
slug = kwargs['s']
descriptor = kwargs['p']
def format_docs(docs):
return "\n\n".join([f"File: {k}\n{v}" for k,v in docs.items()])
prompt = oracle.models.SystemPrompt.objects.get(description=descriptor)
chat, created = oracle.models.Chat.objects.get_or_create(
slug=slug,
defaults=dict(
title="",
system_prompt=prompt,
model=gen_model(),
)
)
if not created:
logger.warn(f"Truncating existing chat '{slug}'")
chat.truncate()
# generate summaries of each doc...?
relevants = oracle.models.ChunkEmbedding.knn(query)[0:k]
logger.info(f"found paths: {relevants}")
summaries = dict()
for (doc, _, _) in relevants:
logger.info(f"Summarize {doc.path}")
summary = doc.summarize(persist=True, ctx=query)
summaries[doc.path] = summary
logger.debug(summary)
chat.add_message("assistant", format_docs(summaries))
chat.add_message("user", query)
resp = chat.query()
chat.add_message("assistant", resp)
print(resp['content'])Summarize a document with the Summarizer
import os
from django.core.management.base import BaseCommand
import oracle.models
import oracle.summarizer
import logging
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = 'Summarize a doc'
output_transaction = True
requires_migrations_checks = True
def add_arguments(self, parser):
parser.add_argument('file', type=str, default="~/org/concept_operating_system.org", nargs='?')
def handle(self, *args, **kwargs):
path = os.path.expanduser(kwargs['file'])
doc = oracle.models.Document.objects.get(path=path)
ctx = input("Conextual Query? ")
# summarizer = oracle.summarizer.OllamaRefineSummarizer(doc)
summarizer = oracle.summarizer.OllamaContextualSummarizer(doc, ctx)
summarizer.summarize()
if input("Store Summary? ").startswith("y"):
summarizer.persist()the chat text is fed in to an embedding model, a vector is persisted to the DB
API Surface to Ollama
Here, have some helpers. I could get rid of this Chat class if I rewrote the generate function... I should make the models configurable:
import ollama
import functools
@functools.lru_cache(maxsize=512)
def embed(text):
return ollama.embeddings(
prompt=text,
model=embed_model(),
).get("embedding")
def gen_model():
return "deepseek-r1:7b"
# return "llama3.2"
# return "hermes3:8b"
def instruct_model():
return "llama3.2:3b-instruct-q8_0"
def embed_model():
return "nomic-embed-text"
def generate(system, assistant=None, user=None):
messages = [ m for m in
[
dict(role="system", message=system),
dict(role="assistant", message=assistant),
dict(role="user", message=user),
]
if m["message"] is not None
]
resp = ollama.chat(
model=gen_model(),
messages=messages
)Document Summarizer
You can get chunks that match the query using knn, use the orm to find the documents, get those documents' chunks' raw text and summarize them. Neat!
from typing import Optional
from inspect import cleandoc
from oracle.ollama_embed import generate
import logging
logger = logging.getLogger(__name__)summarize each chunk by asking the llm nicely, recursively
there are some prompt chains in langchain which claimed to do this but i didn't get far playing with langchain except that i realized i could implement things like this myself. https://python.langchain.com/v0.2/docs/tutorials/summarization/#refine
so you can get a summary of each of the docs, and then cache those; hell we could pre-generate them when you generate the embeddings dark laughter
take each chunk and ask a model to output text summarizing it.
compare whether the summary of the two of them are different enough to append together, and keep a running summary of the doc and then at the end, ask for a summary of each chunk's summary
Summarizer Prompts
You are responsible for summarizing documents for the user.
These notes are included via the prompt below.
The notes are a mix of the inquirer's personal notes, highlights from important documents and reports, and his personal journal. consider dates and names in documents to be important.
there may be code samples included, if so summarize the names of classes, and the names of functions and their arguments.
the document may indicate that it is a personal Journal, first attempt to discern this, whether it's a summarization of web/document content or original work, and if it's the latter pick the most important entries (prefixed with multiple * characters) to summarize, and be very careful of maintaining any dates and person names and location names in those summaries, as well as the context that is a journal entry. In other contexts lines which are prefixed with multiple * characters are sub-headings of the document, where a related but subsident concept, process, event, or idea is described. It is important that the overall context of the document should be retained.try to take the following text and describe the core point or argument in it in four or five sentences.try to take the following text and describe the core point or argument in it in one or two sentences. This is the first part of a multi-part document you'll be summarizing so it may itself include a summary, which you can re-use or emphasize yourself. If there is a definition or factual salient point, return it verbatim and nothing else. If there is a reference or quote from Wikipedia, include it.you are returning to this document's summarization task. this is the {idx} of {total} chunks of a document you have been summarizing.
As context, you have been provided a summary of the previous section of text up to where this new text begins, along with the next text below. integrate these two pieces of texts in to one while retaining the most important aspects of the both. If the prompt is similar enough to the prior summary that it does not add to it, just return the that prior summary verbatim and nothing else.you have been summarizing a document and have finished reading and summarizing each piece of the document. each section's summary will be included below, concisely rephrase all of these summaries in to one, in 2-4 paragraphs and return it verbatim with nothing else. Do not repeat points, be sure to look at previous paragraphs to make sure the summary is not repetive.You are responsible for summarizing documents for the user.
These notes are included via the prompt below.
The notes are a mix of the inquirer's personal notes, highlights from important documents and reports, and his personal journal. consider dates and names in documents to be important.
there may be code samples included, if so summarize the names of
classes, and the names of functions and their arguments.
the document may indicate that it is a personal Journal, if so pick the most important entries (prefixed with multiple * characters) to summarize, and be very careful of maintaining any dates and person names and location names in those summaries, as well as the context that is a journal entry. In other contexts lines which are prefixed with multiple * characters are sub-headings of the document, where a related but subsident concept, process, event, or idea is described. It is important that the overall context of the document should be retained.
The user is asking the following question about their documents, and this document you're summarizing is one that may be relevant to this query, please consider summarizing it with this query mind: {ctx}
Do not respond with anything other that the text of the summary. Do not prefix your message with anything like "here is a summary of the text" or anyting like that. Do not mention the summary itself.The Recursive Prompter
This isn't quite that thing i described above, but it just uses the rolling summary and tries to append new information to that and retain the original summary information. I have some other ideas on how to do this and i'll try them at some point, it's pretty easy to iterate on this component and there is a lot to gain here
Sorry for using noweb syntax here, it makes working with the tangled file a pain, but makes working with the base prompts less painful.
class OllamaRefineSummarizer():
def __init__(self, doc):
self.document = doc
self.chunks = doc.chunk_set.all()
if self.document.summary != '':
self.summary = self.document.summary
def base_prompt(self):
return """
<<base_prompt>>
"""
def single_chunk_prompt(self):
return cleandoc(self.base_prompt() + """
<<single_chunk_prompt>>
""")
def first_chunk_prompt(self):
return cleandoc(self.base_prompt() + """
<<first_chunk_prompt>>
""")
def rolling_summary_prompt(self, idx, rolling_summary_prev):
return cleandoc(self.base_prompt() + """
<<rolling_summary_prompt>>
""").format(
idx=idx+1,
total=len(self.chunks),
rolling_summary=rolling_summary_prev,
)
def final_rollup_summary(self):
return cleandoc(self.base_prompt() + """
<<final_chunk_prompt>>
""") def summarize(self) -> str:
rolling_summary = ""
if len(self.chunks) == 0:
logger.warn("No chunks???")
pass
elif len(self.chunks) == 1:
rolling_summary = self.single_chunk_summarize()
else:
rolling_summary = self.recursive_summarize()
self.summary = rolling_summary
logger.info(self.summary)
return self.summary
def single_chunk_summarize(self):
logger.info("SINGLE CHUNK DOC")
rolling_summary = generate(
system=self.single_chunk_prompt(),
user=self.chunks[0].raw_text()
)
logger.info("summary: %s", rolling_summary)
return rolling_summary
def recursive_summarize(self):
logger.info(f"BEGIN DOCUMENT: len({len(self.chunks)})")
rolling_summaries = list()
rolling_summaries = [generate(
system=self.first_chunk_prompt(),
user=self.chunks[0].raw_text()
)]
logger.debug("summary: %s", rolling_summaries[0])
for idx, chunk in enumerate(self.chunks[1:]):
logger.info(f"BEGIN CHUNK {idx+1}")
following_prompt = self.rolling_summary_prompt(idx, rolling_summaries[idx-1])
rolling_summaries += [generate(
system=following_prompt,
# assistant=rolling_summaries[idx-1],
user=chunk.raw_text()
)]
logger.debug("intermediate summary: %s", rolling_summaries[idx])
chunk.summary = rolling_summaries[idx]
rolling_summary = generate(
system=self.final_rollup_summary(),
user="\n---\n\n---\n".join(rolling_summaries),
)
logger.info("final summary: %s", rolling_summary)
return rolling_summaryWorth keeping generic summaries around if they're decent...
def persist(self):
self.document.summary = self.summary
self.document.save()
return self.documentBut including the original query in the base prompt of the summarizer might be helpful, though those probably can't be persisted. let's try some alchemy:
class OllamaContextualSummarizer(OllamaRefineSummarizer):
def __init__(self, doc, ctx: Optional[str] = None):
super().__init__(doc)
self.context = ctx
def base_prompt(self):
return """
<<contextual_base_prompt>>
""".format(ctx=self.context)
def persist(self):
raise NotImplementedthe summaries of those documents can be used to add high-cardinality context to the question
The bot will answer those for you
The django channels setup
Channels allows you to hook Django views up to arbitrary messaging protocols like WebSockets or MQTT or even Mastadan bots. You can receive WebSockets and respond with JSON or HTML, I use HTMX here to generate and swap HTML elements using Django views.
Channels also allow you to have standalone workers that can respond to messages and join broadcast groups to have autonomous worker processes that can react to user input or other sensor data. This is a very cool technology to add to The Arcology Project, i often wonder when I work on the Elixir iterations where i can build GenServer agents to streamline parts of the design, and this is in some ways similar. open threads
Multiple clients will listen to the same "channel layer" group, routed by the WebSocket router below, and they exchange messages between the clients largely to signal "hey refresh your state from the database" -- this is a dangerous little distributed system we've made here! When a user is connected, it starts up an LLM Bot that will respond to the user's messages by passing messages between these agents:
chat.user_connectis emitted when a new websocket client is set up, this allows the LLM bot to start itself up and seed a prompt.chat.bot_messageandchat.user_messageare emitted when either of those are committed to the database.chat.user_messageincludes the user query, but also some metadata like whether the document search and summarizer should run.the bot will emit
chat.streaming_messageto stream incomplete results to the frontend as they are generated byollama.chat.refill_historywill be used to signal that the history of a chat has been truncated, the frontend will use this to re-draw.
Every consumer on the channel group needs to implement all these messages, even if they are "no-op" pass functions. But you can see below how each client responds to and emits theses messages in the send_message functions, for example.
The LLM Bot
The LLM bot sits on that channel group that is passed to it when a user connects and responds to user messages with LLM generated text, and optionally handles the document search.
from asgiref.sync import sync_to_async
from channels.utils import asyncio
from django.template.loader import get_template
from oracle.ollama_embed import gen_model
import logging
logger = logging.getLogger(__name__)One thing I do not like about how this django-channels thing works but maybe it's mostly just user error; if I import oracle.models in the topmatter, Django complains that this consumer is being imported before the Django ORM and the rest of the system has "started up". This kills the application. So we do ugly imports in the functions, I hope the module caching that Python stuff is good enough.
from channels.consumer import AsyncConsumer
class OracleLLMConsumer(AsyncConsumer):
chats = dict()
def chat(self, group_name):
return self.chats.get(group_name)
async def chat_user_connect(self, event):
import oracle.models
group_name = event["message"]["group_name"]
new_user = event["message"]["client_name"]
await self.channel_layer.group_add(
group_name, self.channel_name
)^ The bot and the client join a group named from the Websocket URL created by the Websocket client below and included in the chat.user_connect method.
if self.chat(group_name) is None:
# try to fetch matching system prompt, fall back to the default oracle prompt
try:
prompt = await oracle.models.SystemPrompt.objects.aget(description=group_name)
except oracle.models.SystemPrompt.DoesNotExist:
prompt = await oracle.models.SystemPrompt.objects.aget(description="oracle")
self.chats[group_name], _created = await oracle.models.Chat.objects.aget_or_create(
slug=group_name,
defaults=dict(
model=gen_model(),
title="",
system_prompt=prompt,
)
)
logger.info(f"bot spawned: {group_name} {self.channel_name}")
logger.info(f"conn from: {new_user}")^ A chat is found or created by both the consumers, this is a pretty gnarly race condition in theory but the LLM bot is started after the websocket bot initializes its Chat object, so it's Probably Fine™. Flop-Sweating and repeating Personal Software Can Be Shitty as a careful mantra...
chat.user_message is processed by the bot to kick off the RAG and the ollama query.
async def chat_user_message(self, event):
logger.debug(event)
message = event["message"]
group_name = event["group_name"]
do_rag = message.get("search_rag", False)
if do_rag:
import oracle.models
logger.debug("querying RAG...")
await self.send_message(
group_name,
dict(role="annotation", content="Searching your documents now...")
)
query = event["message"]["content"]
relevants = list(set(await oracle.models.ChunkEmbedding.aknn(query, k=5)))
logger.info(f"found paths: {relevants}")
def format_doc(path, doc):
return f"- File: {path}\n {doc}"
for (doc, _distance, _chunk_start) in relevants:
logger.info(f"Contextually summarize {doc.path}")
await self.send_message(
group_name,
dict(role="assistant", content=f"Summarizing {doc.path}")
)
summary = await sync_to_async(doc.summarize)(persist=False, ctx=query)
await self.send_message(
group_name,
dict(role="user", content=format_doc(doc.path, summary))
)
logger.info("querying...")
new_response = ""
streamed_message = dict(role="", content="")
async for chunk in self.chat(group_name).aquery():
logger.debug(streamed_message)
logger.debug(chunk)
new_response += chunk['message']['content']
streamed_message = dict(role=chunk["message"]["role"], content=new_response)
if chunk["done"] == True:
logger.debug("stream done, propagate")
await self.channel_layer.group_send(
group_name, {"type": "chat.streaming_message", "message": ""}
)
else:
await self.channel_layer.group_send(
group_name, {"type": "chat.streaming_message", "message": streamed_message}
)
logger.debug("done, propagate", streamed_message)
return await self.send_message(group_name, streamed_message)
async def collect_summaries(self, group_name, query, k=5):
import oracle.models
for (doc, _, _) in relevants:
summary = await sync_to_async(doc.summarize)(persist=False, ctx=query)
yield summary
async def send_message(self, group_name, message):
await sync_to_async(self.chat(group_name).add_message)(message["role"], message["content"])
return await self.channel_layer.group_send(
group_name, dict(type="chat.bot_message", message=message)
)Stubs for messages the bot doesn't need to process:
async def chat_bot_message(self, event):
pass
async def chat_streaming_message(self, event):
pass
async def chat_refill_history(self, event):
passthe Websocket client
The Websocket client receives events from the HTMX form frontend view set up below.
See django-channels docs.
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class OracleWSConsumer(AsyncWebsocketConsumer):
streaming_message = None
awaiting_messages = False
chat = None
async def connect(self):
self.user = self.scope["user"]
if not self.user.is_superuser:
logger.warn(f"deny un-authenticated user: {self.user}")
return
import oracle.models
await self.accept()
self.group_name = self.scope["url_route"]["kwargs"].get("chat_id", "default")
await self.channel_layer.group_add(
self.group_name, self.channel_name
)
await self.channel_layer.send(
"llm-bot", dict(
type="chat.user_connect", message=dict(
client_name=self.channel_name,
group_name=self.group_name
)
)
)
logger.info(f"user connected: {self.group_name} {self.channel_name}")
if self.chat is None:
# try to fetch matching system prompt, fall back to the default oracle prompt
try:
prompt = await oracle.models.SystemPrompt.objects.aget(description=self.group_name)
except oracle.models.SystemPrompt.DoesNotExist:
prompt = await oracle.models.SystemPrompt.objects.aget(description="oracle")
self.chat, _created = await oracle.models.Chat.objects.aget_or_create(
slug=self.group_name,
defaults=dict(
model="hermes3:8b",
system_prompt=prompt,
)
)
return await self.render()connecthandles authentication, it sets up the group channel, spawns an instance of the LLM bot on that channel, and sets up its Chat when the user loads the page.receivehandles messages from the HTMX form, the query and some other parameters or form buttons, basically where they are handled and emit other events. When a message is received, it's stored and
async def receive(self, text_data):
jason = json.loads(text_data)
if jason.get("clear_messages") == "1":
await sync_to_async(self.chat.truncate)(1)
self.streaming_message = ""
await self.channel_layer.group_send(
self.group_name,
dict(
type="chat.refill_history",
messages=await sync_to_async(self.chat.messages)(),
),
)
elif jason.get("content") is not None:
query = jason["content"]
do_rag_search = jason.get("consult-tomes", "off") == "on"
msg = dict(role="user", content=query, search_rag=do_rag_search)
await self.send_message(msg)
async def disconnect(self, close_code):
await self.channel_layer.group_discard(
self.group_name, self.channel_name
)
async def send_message(self, message):
await sync_to_async(self.chat.add_message)(message["role"], message["content"])
return await self.channel_layer.group_send(
self.group_name, dict(
type="chat.user_message",
message=message,
group_name=self.group_name,
)
)When the websocket client receives messages from itself and from other clients and from the LLM, and uses that to reach in to the database to get the ChatMessages and render them in to an HTML template defined below.
async def render(self):
html = await sync_to_async(self.render_messages)()
return await self.send(text_data=html)
def render_messages(self):
context = dict(
messages=self.chat.messages(filtered=False),
waiting=self.awaiting_messages,
streaming_message=self.streaming_message
)
return get_template("oracle/partial_message.html").render(
context=context
)Basically when the front-end gets a notification for any of these events it's because there are new or fewer messages in the database. the chat.streaming_message event is emitted by the LLM bot and fed to the frontend below in the HTMX response swapping logic.
async def chat_user_message(self, event):
message = event["message"]
self.awaiting_messages = True
return await self.render()
async def chat_bot_message(self, event):
self.awaiting_messages = False
return await self.render()
async def chat_refill_history(self, event):
return await self.render()
async def chat_streaming_message(self, event):
self.streaming_message = event.get("message", None)
return await self.render()The HTMX-powered dynamic chat pages
the Oracle exposes an "index" that for now joins a default chat (but could/should list an index of them too), and slugged URLs for chats that can be maintained for long term or with certain prompts. The websockets URLs are generated by the views based on that slug, and are used to plumb messages to particular instances of the consumers.
from django.urls import path, re_path
from . import views
from . import consumers
urlpatterns = [
path("", views.index, name="index"),
path("<slug:chat_id>", views.chat, name="chat"),
]
websocket_urlpatterns = [
re_path(r"^ws/oracle/(?P<chat_id>\w+)$", consumers.OracleWSConsumer.as_asgi()),
re_path(r"^ws/oracle/", consumers.OracleWSConsumer.as_asgi()),
]The view decorators make sure that only an authenticated Django Admin user is able to log in to the site backend, this will handle the login page redirect and all that which is quite nice. Django really is quite nice with these batteries!
from django.shortcuts import render
from django.http import HttpRequest
from django.contrib.admin.views.decorators import staff_member_required
def ws_url(request: HttpRequest, id=""):
if request.is_secure():
return f"wss://{request.headers.get('Host')}/ws/oracle/{id}"
return f"ws://{request.headers.get('Host')}/ws/oracle/{id}"
@staff_member_required
def index(request: HttpRequest):
return render(request, "oracle/index.html",
dict(ws_url=ws_url(request),
chat_slug="default",
))
@staff_member_required
def chat(request: HttpRequest, chat_id):
return render(request, "oracle/index.html",
dict(ws_url=ws_url(request, chat_id),
chat_slug=chat_id,
))Each of these renders a simple index.html template that has HTMX loaded in to it with the WebSocket extension, and a WebSocket URL embedded in to it, and an otherwise "un-hydrated" form. That is to say: it has the form elements but no text from the bots until the HTMX WebSocket connection is established.
{% extends "arcology/app.html" %}
{% block title %}Ask the Arcology{% endblock %}
{% block h1 %}<h1>Ask the Arcology</h1>{% endblock %}
{% load static %}
{% block extra_head %}
<script src="{% static 'sitemap/js/htmx.js' %}"></script>
<script src="{% static 'sitemap/js/htmx-ws.js' %}"></script>
<link rel="stylesheet" href="{% static 'oracle/css/oracle.css' %}"/>
<script defer>
htmx.logger = function (elt, event, data) {
if (console) {
console.log(event, elt, data);
}
};
htmx.on("htmx:wsAfterSend", (evt) => {
document.querySelector("#spinner").classList.add("htmx-request");
});
htmx.on("htmx:wsAfterMessage", (evt) => {
document.querySelector("#spinner").classList.remove("htmx-request");
});
</script>
{% endblock %}When the WebSocket client above sees the user connect it eventually calls render_messages which pushes HTML to the client over the websocket, it's swapped by referring to the =#chat and #chatinput= elements in the chat messages partial below, along with the "spinner"'s visibility or invisibility.
It's important to understand here how HTMX's "out of band" swaps work. It lets you just cram arbitrary element updates in to the form responses, see that the chat is "out of band" of the form. Other stuff can be included in here like the partial messages or sidebar links to the summoned documents, or things of that nature.
{% block content %}
<section>
<div id="chat" hx-swap-oob="true scroll:bottom">
...
</div>
<div id="spinner" hx-swap-oob="true">Waiting...</div>
<form hx-ext="ws" ws-connect="{{ws_url}}" class="oracle-box" id="chatinput"
hx-target="#chat" hx-swap="beforeend"
hx-indicator="#spinner" hx-on:submit="this.reset()" ws-send>
<div>
<input class="" autofocus
type="text" placeholder="Ponder the orb..."
name="content" id="user-message" />
<input type="checkbox" name="consult-tomes" id="consult-tomes">
<label for="consult-tomes">Consult the tomes?</label>
</div>
<div>
<button class="invoke" type="submit">Ask the Arcology</button>
<button class="reset" value="1" name="clear_messages" ws-send>Let's try again...</button>
</div>
</form>
<ul>
<li><a href="{% url 'admin:oracle_chat_change' chat_slug %}" target="_blank">Scry Settings</a> (change system prompts, etc)</li>
</ul>
</section>
{% endblock %}This thing has only a few features right now, as you can see:
A textbox and button to add a message to the chat and generate after it using
ollamaA check mark to enable/disable RAG search
A button to truncate the history back to the initial "system" prompt
A link to the admin settings to change prompts, etc
<div id="chat" hx-swap-oob="true">
{% for message in messages %}
<span id="message-{{message.id}}" class="message role-{{ message.role }}">{{ message.content }}</span>
{% endfor %}
{% if streaming_message %}
<span class="message role-{{ streaming_message.role }}">{{ streaming_message.content }}</span>
{% endif %}
</div>Oracle CSS
Message rendering:
span.message {
display: block;
margin: 0.5em;
padding: 0.5em;
border: 1pt solid;
border-radius: 0.5ch;
word-wrap: break-word;
white-space: pre-wrap;
}
span.role-assistant {
border-color: var(--success);
}
span.role-user {
font-size: smaller;
background-color: var(--light-gray);
}
span.role-system {
font-size: smaller;
height: 10ch;
overflow: scroll;
background-color: var(--light-gray);
border-color: var(--warning);
}Form rendering:
form.oracle-box div {
display: flex;
padding: 0.5em;
}
form input[type="text"] {
flex: fit-content;
}
form > * {
margin: 0.5em;
}
form button,input,textarea {
font-family: "Vulf Mono", monospace;
font-style: italic;
}
form button {
width: 100%;
margin: 0.5em;
background-color: var(--light-gray);
color: var(--dark-gray);
border: 1pt var(--warning) solid;
}
form button.reset {
border-color: var(--alert);
}Spinner visibility/invisility:
#spinner{
display:none;
}
.htmx-request#spinner{
display: block;
}DONE Deployment changes
OLLAMA_HOSTenv varchannel worker for
llm-botredis server for channel layer
NEXT can also use wikipedia to provide context for many pages which have external ROAM_REFS
need to move from chunking pages to chunking headings for this to be most effective...
NEXT move to headings instead of pages for RAG search
this requires changes to The arroyo_rs Native Org Parser, an org->org converter that can extract subheadings...
What you end up with
it works at like a handful of tokens per second, and it makes the laptop very hot, and is at risk of bullshitting, but it does work and can integrate different information and concepts from those contexts in fascinating ways.
you could build a PC with a GPU, a small shrine in the corner of your room to the inner daemon which talks back. I stuck a 400$ GPU in to my desktop and installed Ollama on that and made it available to my server over Tailscale, that runs llama3.1:8b at like 60tok/s
you can use a more complex model where you can consult the oracle in the corner of your room, trade patience for complexity or money on a fancy gpu inference engine; for now, for cheap, you can submit a question and come back in an hour, ask again later.
NEXT Figure about how this interacts with the arcology, what the UX is
INPROGRESS i want to have something that will let me put questions in to the arcology on my phone, to my laptop or my server over tailscale.
NEXT multi-user handling
right now the embedding DB doesn't have any way to limit to a given user id... could add a user ID to chats and JOIN the chunk embedding to that but that's a pretty explosive join expansion...
almost need per-user directories of org files and start to think about things like that instead of one big syncthing dir...
NEXT show related/similar articles or things from my Archive in interesting ways
NEXT clustering???
NEXT include links as higher-weighted/"more important" RAG context, not just text
NEXT need to carefully segregate published/unpublished/private content, this is obviously a huge risk!
NEXT i want to be able to tag a handful of documents on the site and ask questions of them
NEXT i want to be able to do that in the Localhost API for the Arcology
NEXT Figure about how this interacts with the CCE, what local tools can be added?
NEXT being able to make queries of an open buffer, and its related buffers
NEXT use an org-mode buffer as a chat-like or conversational interface
query the model to take the above parts of the document and linked/similar documents and output an org heading in to the doc to review/discard
NEXT generate cce code that matches the description i write in a heading
NEXT write an gptel backend or wrapper that can use the arcology's prompts and history stuff...
(use-package gptel
:config
(evil-leader/set-key (kbd "f") #'consult-omni-gptel)
(setq
gptel-model "deepseek-r1:14b"
gptel-backend (gptel-make-ollama "Ollama"
:host "100.82.104.74:11434"
:stream t
:models '("deepseek-r1:14b" "deepseek-r1:7b" "hermes3:8b" "llama3.3"))))
(use-package ellama
:config
(require 'llm-ollama)
(setopt ellama-naming-scheme 'ellama-generate-name-by-llm)
(setopt ellama-providers `(("r1-7b" . ,(make-llm-ollama
:host "windows"
:chat-model "deepseek-r1:7b"
:embedding-model "nomic-embed-text"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("r1-14b" . ,(make-llm-ollama
:host "windows"
:chat-model "deepseek-r1:14b"
:embedding-model "nomic-embed-text"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("l3" . ,(make-llm-ollama
:host "windows"
:chat-model "llama3.2"
:embedding-model "nomic-embed-text"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("g3" . ,(make-llm-ollama
:host "windows"
:chat-model "gemma3:4b"
:embedding-model "nomic-embed-text"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("g3-12" . ,(make-llm-ollama
:host "windows"
:chat-model "gemma3:12b"
:embedding-model "nomic-embed-text"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("qwen35" . ,(make-llm-ollama
:host "localhost"
:chat-model "qwen3.5:cloud"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
("qwen3-coder-next" . ,(make-llm-ollama
:host "localhost"
:chat-model "qwen3-coder-next:cloud"
:default-chat-non-standard-params `(("num_ctx" . ,(* 1024 128)))))
))
(setopt ellama-provider (alist-get "qwen3.5" ellama-providers nil nil #'equal))
(setopt ellama-summarization-provider (alist-get "r1-14b" ellama-providers nil nil #'equal))
(setopt ellama-naming-provider (alist-get "l3" ellama-providers nil nil #'equal))
;; ellama-coding-provider
;; ellama-translation-provider
(evil-leader/set-key "e" #'ellama-transient-main-menu))
(provide 'cce/arcology-oracle)