grail/gf-lt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Chore: remove old rag

Browse Source
This commit is contained in:
Grail Finder
2025-10-19 13:14:56 +03:00
parent dfa164e871
commit 60ccaed200
8 changed files with 101 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
bot.go
View File

@@ -9,7 +9,7 @@ import (
"gf-lt/config"
"gf-lt/extra"
"gf-lt/models"
"gf-lt/rag_new"
"gf-lt/rag"
"gf-lt/storage"
"io"
"log/slog"
@@ -41,7 +41,7 @@ var (
defaultStarter = []models.RoleMsg{}
defaultStarterBytes = []byte{}
interruptResp = false
ragger *rag_new.RAG
ragger *rag.RAG
chunkParser ChunkParser
lastToolCall *models.FuncCall
//nolint:unused // TTS_ENABLED conditionally uses this
@@ -571,7 +571,7 @@ func init() {
if store == nil {
os.Exit(1)
}
ragger = rag_new.New(logger, store, cfg)
ragger = rag.New(logger, store, cfg)
// https://github.com/coreydaley/ggerganov-llama.cpp/blob/master/examples/server/README.md
// load all chats in memory
if _, err := loadHistoryChats(); err != nil {

5
rag_new/embedder.go → rag/embedder.go
View File

@@ -1,10 +1,10 @@
package rag_new
package rag
import (
"bytes"
"gf-lt/config"
"encoding/json"
"fmt"
"gf-lt/config"
"log/slog"
"net/http"
)
@@ -96,3 +96,4 @@ func (a *APIEmbedder) EmbedSingle(text string) ([]float32, error) {
//
// For now, we'll focus on the API implementation which is already working in the current system,
// and can be extended later when we have ONNX runtime integration

265
rag/main.go
View File

@@ -1,265 +0,0 @@
package rag
import (
"bytes"
"gf-lt/config"
"gf-lt/models"
"gf-lt/storage"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"os"
"path"
"strings"
"sync"
"github.com/neurosnap/sentences/english"
)
var (
LongJobStatusCh = make(chan string, 1)
// messages
FinishedRAGStatus = "finished loading RAG file; press Enter"
LoadedFileRAGStatus = "loaded file"
ErrRAGStatus = "some error occured; failed to transfer data to vector db"
)
type RAG struct {
logger *slog.Logger
store storage.FullRepo
cfg *config.Config
}
func New(l *slog.Logger, s storage.FullRepo, cfg *config.Config) *RAG {
return &RAG{
logger: l,
store: s,
cfg: cfg,
}
}
func wordCounter(sentence string) int {
return len(strings.Split(sentence, " "))
}
func (r *RAG) LoadRAG(fpath string) error {
data, err := os.ReadFile(fpath)
if err != nil {
return err
}
r.logger.Debug("rag: loaded file", "fp", fpath)
LongJobStatusCh <- LoadedFileRAGStatus
fileText := string(data)
tokenizer, err := english.NewSentenceTokenizer(nil)
if err != nil {
return err
}
sentences := tokenizer.Tokenize(fileText)
sents := make([]string, len(sentences))
for i, s := range sentences {
sents[i] = s.Text
}
var (
maxChSize = 1000
left = 0
right = r.cfg.RAGBatchSize
batchCh = make(chan map[int][]string, maxChSize)
vectorCh = make(chan []models.VectorRow, maxChSize)
errCh = make(chan error, 1)
doneCh = make(chan bool, 1)
lock = new(sync.Mutex)
)
defer close(doneCh)
defer close(errCh)
defer close(batchCh)
// group sentences
paragraphs := []string{}
par := strings.Builder{}
for i := 0; i < len(sents); i++ {
par.WriteString(sents[i])
if wordCounter(par.String()) > int(r.cfg.RAGWordLimit) {
paragraphs = append(paragraphs, par.String())
par.Reset()
}
}
if len(paragraphs) < int(r.cfg.RAGBatchSize) {
r.cfg.RAGBatchSize = len(paragraphs)
}
// fill input channel
ctn := 0
for {
if int(right) > len(paragraphs) {
batchCh <- map[int][]string{left: paragraphs[left:]}
break
}
batchCh <- map[int][]string{left: paragraphs[left:right]}
left, right = right, right+r.cfg.RAGBatchSize
ctn++
}
finishedBatchesMsg := fmt.Sprintf("finished batching batches#: %d; paragraphs: %d; sentences: %d\n", len(batchCh), len(paragraphs), len(sents))
r.logger.Debug(finishedBatchesMsg)
LongJobStatusCh <- finishedBatchesMsg
for w := 0; w < int(r.cfg.RAGWorkers); w++ {
go r.batchToVectorHFAsync(lock, w, batchCh, vectorCh, errCh, doneCh, path.Base(fpath))
}
// wait for emb to be done
<-doneCh
// write to db
return r.writeVectors(vectorCh)
}
func (r *RAG) writeVectors(vectorCh chan []models.VectorRow) error {
for {
for batch := range vectorCh {
for _, vector := range batch {
if err := r.store.WriteVector(&vector); err != nil {
r.logger.Error("failed to write vector", "error", err, "slug", vector.Slug)
LongJobStatusCh <- ErrRAGStatus
continue // a duplicate is not critical
// return err
}
}
r.logger.Debug("wrote batch to db", "size", len(batch), "vector_chan_len", len(vectorCh))
if len(vectorCh) == 0 {
r.logger.Debug("finished writing vectors")
LongJobStatusCh <- FinishedRAGStatus
defer close(vectorCh)
return nil
}
}
}
}
func (r *RAG) batchToVectorHFAsync(lock *sync.Mutex, id int, inputCh <-chan map[int][]string,
vectorCh chan<- []models.VectorRow, errCh chan error, doneCh chan bool, filename string) {
for {
lock.Lock()
if len(inputCh) == 0 {
if len(doneCh) == 0 {
doneCh <- true
}
lock.Unlock()
return
}
select {
case linesMap := <-inputCh:
for leftI, v := range linesMap {
r.fecthEmbHF(v, errCh, vectorCh, fmt.Sprintf("%s_%d", filename, leftI), filename)
}
lock.Unlock()
case err := <-errCh:
r.logger.Error("got an error", "error", err)
lock.Unlock()
return
}
r.logger.Debug("to vector batches", "batches#", len(inputCh), "worker#", id)
LongJobStatusCh <- fmt.Sprintf("converted to vector; batches: %d, worker#: %d", len(inputCh), id)
}
}
func (r *RAG) fecthEmbHF(lines []string, errCh chan error, vectorCh chan<- []models.VectorRow, slug, filename string) {
payload, err := json.Marshal(
map[string]any{"inputs": lines, "options": map[string]bool{"wait_for_model": true}},
)
if err != nil {
r.logger.Error("failed to marshal payload", "err:", err.Error())
errCh <- err
return
}
// nolint
req, err := http.NewRequest("POST", r.cfg.EmbedURL, bytes.NewReader(payload))
if err != nil {
r.logger.Error("failed to create new req", "err:", err.Error())
errCh <- err
return
}
req.Header.Add("Authorization", "Bearer "+r.cfg.HFToken)
resp, err := http.DefaultClient.Do(req)
if err != nil {
r.logger.Error("failed to embedd line", "err:", err.Error())
errCh <- err
return
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
r.logger.Error("non 200 resp", "code", resp.StatusCode)
return
}
emb := [][]float32{}
if err := json.NewDecoder(resp.Body).Decode(&emb); err != nil {
r.logger.Error("failed to embedd line", "err:", err.Error())
errCh <- err
return
}
if len(emb) == 0 {
r.logger.Error("empty emb")
err = errors.New("empty emb")
errCh <- err
return
}
vectors := make([]models.VectorRow, len(emb))
for i, e := range emb {
vector := models.VectorRow{
Embeddings: e,
RawText: lines[i],
Slug: fmt.Sprintf("%s_%d", slug, i),
FileName: filename,
}
vectors[i] = vector
}
vectorCh <- vectors
}
func (r *RAG) LineToVector(line string) ([]float32, error) {
lines := []string{line}
payload, err := json.Marshal(
map[string]any{"inputs": lines, "options": map[string]bool{"wait_for_model": true}},
)
if err != nil {
r.logger.Error("failed to marshal payload", "err:", err.Error())
return nil, err
}
// nolint
req, err := http.NewRequest("POST", r.cfg.EmbedURL, bytes.NewReader(payload))
if err != nil {
r.logger.Error("failed to create new req", "err:", err.Error())
return nil, err
}
req.Header.Add("Authorization", "Bearer "+r.cfg.HFToken)
resp, err := http.DefaultClient.Do(req)
if err != nil {
r.logger.Error("failed to embedd line", "err:", err.Error())
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
err = fmt.Errorf("non 200 resp; code: %v", resp.StatusCode)
r.logger.Error(err.Error())
return nil, err
}
emb := [][]float32{}
if err := json.NewDecoder(resp.Body).Decode(&emb); err != nil {
r.logger.Error("failed to embedd line", "err:", err.Error())
return nil, err
}
if len(emb) == 0 || len(emb[0]) == 0 {
r.logger.Error("empty emb")
err = errors.New("empty emb")
return nil, err
}
return emb[0], nil
}
func (r *RAG) SearchEmb(emb *models.EmbeddingResp) ([]models.VectorRow, error) {
return r.store.SearchClosest(emb.Embedding)
}
func (r *RAG) ListLoaded() ([]string, error) {
return r.store.ListFiles()
}
func (r *RAG) RemoveFile(filename string) error {
return r.store.RemoveEmbByFileName(filename)
}

5
rag_new/rag.go → rag/rag.go
View File

@@ -1,10 +1,10 @@
package rag_new
package rag
import (
"fmt"
"gf-lt/config"
"gf-lt/models"
"gf-lt/storage"
"fmt"
"log/slog"
"os"
"path"
@@ -258,3 +258,4 @@ func (r *RAG) ListLoaded() ([]string, error) {
func (r *RAG) RemoveFile(filename string) error {
return r.storage.RemoveEmbByFileName(filename)
}

7
rag_new/storage.go → rag/storage.go
View File

@@ -1,10 +1,10 @@
package rag_new
package rag
import (
"gf-lt/models"
"gf-lt/storage"
"encoding/binary"
"fmt"
"gf-lt/models"
"gf-lt/storage"
"log/slog"
"sort"
"strings"
@@ -298,3 +298,4 @@ func sqrt(f float32) float32 {
}
return guess
}

6
storage/migrate.go
View File

@@ -5,8 +5,6 @@ import (
"fmt"
"io/fs"
"strings"
_ "github.com/asg017/sqlite-vec-go-bindings/ncruces"
)
//go:embed migrations/*
@@ -53,8 +51,8 @@ func (p *ProviderSQL) executeMigration(migrationsDir fs.FS, fileName string) err
}
func (p *ProviderSQL) executeSQL(sqlContent []byte) error {
// Connect to the database (example using a simple connection)
err := p.s3Conn.Exec(string(sqlContent))
// Execute the migration content using standard database connection
_, err := p.db.Exec(string(sqlContent))
if err != nil {
return fmt.Errorf("failed to execute SQL: %w", err)
}

10
storage/storage.go
View File

@@ -6,7 +6,6 @@ import (
_ "github.com/glebarez/go-sqlite"
"github.com/jmoiron/sqlx"
"github.com/ncruces/go-sqlite3"
)
type FullRepo interface {
@@ -28,7 +27,6 @@ type ChatHistory interface {
type ProviderSQL struct {
db *sqlx.DB
s3Conn *sqlite3.Conn
logger *slog.Logger
}
@@ -97,7 +95,7 @@ func (p ProviderSQL) ChatGetMaxID() (uint32, error) {
return id, err
}
// opens two connections
// opens database connection
func NewProviderSQL(dbPath string, logger *slog.Logger) FullRepo {
db, err := sqlx.Open("sqlite", dbPath)
if err != nil {
@@ -105,11 +103,7 @@ func NewProviderSQL(dbPath string, logger *slog.Logger) FullRepo {
return nil
}
p := ProviderSQL{db: db, logger: logger}
p.s3Conn, err = sqlite3.Open(dbPath)
if err != nil {
logger.Error("failed to open vecdb connection", "error", err)
return nil
}
p.Migrate()
return p
}

60
storage/vector.go
View File

@@ -66,35 +66,13 @@ func (p ProviderSQL) WriteVector(row *models.VectorRow) error {
if err != nil {
return err
}
stmt, _, err := p.s3Conn.Prepare(
fmt.Sprintf("INSERT INTO %s(embedding, slug, raw_text, filename) VALUES (?, ?, ?, ?)", tableName))
if err != nil {
p.logger.Error("failed to prep a stmt", "error", err)
return err
}
defer stmt.Close()
serializedEmbeddings := SerializeVector(row.Embeddings)
if err := stmt.BindBlob(1, serializedEmbeddings); err != nil {
p.logger.Error("failed to bind", "error", err)
query := fmt.Sprintf("INSERT INTO %s(embedding, slug, raw_text, filename) VALUES (?, ?, ?, ?)", tableName)
_, err = p.db.Exec(query, serializedEmbeddings, row.Slug, row.RawText, row.FileName)
return err
}
if err := stmt.BindText(2, row.Slug); err != nil {
p.logger.Error("failed to bind", "error", err)
return err
}
if err := stmt.BindText(3, row.RawText); err != nil {
p.logger.Error("failed to bind", "error", err)
return err
}
if err := stmt.BindText(4, row.FileName); err != nil {
p.logger.Error("failed to bind", "error", err)
return err
}
err = stmt.Exec()
if err != nil {
return err
}
return nil
}
func decodeUnsafe(bs []byte) []float32 {
@@ -110,30 +88,30 @@ func (p ProviderSQL) SearchClosest(q []float32) ([]models.VectorRow, error) {
func (p ProviderSQL) ListFiles() ([]string, error) {
q := fmt.Sprintf("SELECT filename FROM %s GROUP BY filename", vecTableName384)
stmt, _, err := p.s3Conn.Prepare(q)
rows, err := p.db.Query(q)
if err != nil {
return nil, err
}
defer stmt.Close()
defer rows.Close()
resp := []string{}
for stmt.Step() {
resp = append(resp, stmt.ColumnText(0))
}
if err := stmt.Err(); err != nil {
for rows.Next() {
var filename string
if err := rows.Scan(&filename); err != nil {
return nil, err
}
resp = append(resp, filename)
}
if err := rows.Err(); err != nil {
return nil, err
}
return resp, nil
}
func (p ProviderSQL) RemoveEmbByFileName(filename string) error {
q := fmt.Sprintf("DELETE FROM %s WHERE filename = ?", vecTableName384)
stmt, _, err := p.s3Conn.Prepare(q)
if err != nil {
_, err := p.db.Exec(q, filename)
return err
}
defer stmt.Close()
if err := stmt.BindText(1, filename); err != nil {
return err
}
return stmt.Exec()
}