From 34577871f1232b141027278492bff92f9625396e Mon Sep 17 00:00:00 2001 From: Emil Shanaty Date: Thu, 28 May 2026 14:48:51 +0300 Subject: [PATCH] =?UTF-8?q?Initial=20commit:=20DeepRes=20Go=20backend=20?= =?UTF-8?q?=E2=80=94=20bot=20+=20worker=20+=20embed=20service?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 25 ++ .gitignore | 27 +++ docker-compose.yml | 33 +++ embed-service/Dockerfile | 9 + .../__pycache__/main.cpython-314.pyc | Bin 0 -> 3416 bytes embed-service/main.py | 50 ++++ embed-service/requirements.txt | 3 + go.mod | 23 ++ internal/config/config.go | 74 ++++++ internal/db/db.go | 166 +++++++++++++ internal/models/models.go | 61 +++++ internal/queue/queue.go | 63 +++++ internal/research/dedup/dedup.go | 105 +++++++++ internal/research/embed/embed.go | 105 +++++++++ internal/research/pipeline/pipeline.go | 219 ++++++++++++++++++ internal/research/search/search.go | 104 +++++++++ internal/research/synthesize/synthesize.go | 145 ++++++++++++ migrations/001_init.sql | 39 ++++ run-bot.sh | 4 + run-worker.sh | 4 + searxng/settings.yml | 28 +++ 21 files changed, 1287 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 docker-compose.yml create mode 100644 embed-service/Dockerfile create mode 100644 embed-service/__pycache__/main.cpython-314.pyc create mode 100644 embed-service/main.py create mode 100644 embed-service/requirements.txt create mode 100644 go.mod create mode 100644 internal/config/config.go create mode 100644 internal/db/db.go create mode 100644 internal/models/models.go create mode 100644 internal/queue/queue.go create mode 100644 internal/research/dedup/dedup.go create mode 100644 internal/research/embed/embed.go create mode 100644 internal/research/pipeline/pipeline.go create mode 100644 internal/research/search/search.go create mode 100644 internal/research/synthesize/synthesize.go create mode 100644 migrations/001_init.sql create mode 100755 run-bot.sh create mode 100755 run-worker.sh create mode 100644 searxng/settings.yml diff --git a/.env.example b/.env.example new file mode 100644 index 000000000..f0216fd12 --- /dev/null +++ b/.env.example @@ -0,0 +1,25 @@ +# ============= DeepRes Configuration ============= + +# Telegram Bot Token (от @BotFather) +TELEGRAM_TOKEN= + +# PostgreSQL +DATABASE_URL=postgres://deepres:deepres@localhost:5432/deepres?sslmode=disable + +# NATS +NATS_URL=nats://localhost:4222 + +# SearXNG +SEARXNG_URL=http://localhost:4000 + +# OpenRouter AI (для синтеза) +OPENROUTER_API_KEY= +LLM_MODEL=deepseek/deepseek-v4-flash + +# Embeddings (E5-small HTTP-сервер, например https://github.com/emil/embed-server) +EMBED_URL=http://localhost:8081 + +# Цены в Telegram Stars +FAST_COST=1 +DEEP_COST=5 +FREE_FAST_PER_DAY=3 diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..31c2d2a85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Environment +.env +.env.local + +# Binaries +bot +worker +*.exe + +# Logs +*.log +/tmp/deepres-* + +# Reports +reports/ + +# Go +vendor/ +*.sum + +# IDE +.idea/ +.vscode/ + +# OS +.DS_Store +reports/ diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 000000000..78c0ca167 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,33 @@ +version: "3.9" + +services: + postgres: + image: postgres:17 + ports: + - "5432:5432" + environment: + POSTGRES_DB: deepres + POSTGRES_USER: deepres + POSTGRES_PASSWORD: deepres + volumes: + - pgdata:/var/lib/postgresql/data + restart: unless-stopped + + nats: + image: nats:latest + ports: + - "4222:4222" + restart: unless-stopped + + searxng: + image: searxng/searxng:latest + ports: + - "4000:8080" + environment: + - SEARXNG_BASE_URL=http://localhost:4000 + volumes: + - ./searxng/settings.yml:/etc/searxng/settings.yml:z + restart: unless-stopped + +volumes: + pgdata: diff --git a/embed-service/Dockerfile b/embed-service/Dockerfile new file mode 100644 index 000000000..0bef1031b --- /dev/null +++ b/embed-service/Dockerfile @@ -0,0 +1,9 @@ +FROM python:3.12-slim + +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY main.py . + +EXPOSE 8081 +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8081"] diff --git a/embed-service/__pycache__/main.cpython-314.pyc b/embed-service/__pycache__/main.cpython-314.pyc new file mode 100644 index 0000000000000000000000000000000000000000..fbf776de2bf94e2e7c20d4a279c0a1e51a6d365c GIT binary patch literal 3416 zcmbtWT}&L;6~1@=_Lp7$mj!IE<2dOyh9wS`q>^o^Zg7+b9#geag~+ye^LnfCw|nMRvT=crG!+8PB^(lbUIGYOI$jFV`*GI zuW&_-<#F}A#x*fk#v?pBCmkh&wJmsTPCH6;RoC>$X(?p#cn1MqG(+^*oH8iaxCz0H z12@sYO$u%jxJ?b*rn%Vkq1q7*ot_#5jj`!zSkasz<>nnt(y-=eC~c&A%g&nC%&y)o zbFF&Y^RE(8>LsKX@2rfs@wU0P>F+eI4#!8yTug7*JM_-ewc8wl`NzaRpQ)5#R($kS zpTA_5%Y9c*=B%Z8t5mX`%Q@d-S1p$NNA-i;OP=KnkM|YbCC}vceA&tsu0C7Xnd;{( zr{uY|!~5ftwqwIyCvq=LU+i<8^0gDWpFcY=FgcRbhbQ~vc^X8q^D)Ia6Vi zA_7+fkAOoGH00I@kRe(nGvt!={4W3u;P#Rq(?wb!FVmlp?@Ky;3g*C=;`fPe_^36I zNhpH@mo1;T_&3tfM~XfDY7lAP0&Gd!947 z8N_y;$hZ5@6ew^|AFBHd_>u&Fd7nSYHGfiA0yYgacm8#&$kCxn)>{mNOI~4Ui$I1K z#nL!Z#ly-=tQo#S3x+-wny?M$bne_{k87aJe+EeC8p-s!a;WASsp~IpqE~Ecq`Cdh z=q3%6g0F)eL(+o;A{${CM%iH?jlkxRiA1Qd3vBD5 z>ewD&an1pO%^nEYrrJaCJ=cLJg(m#}VMRFiK`BB5h75pwFRa2qfl>nB5|1V?3rK}3 z!Y2P4%kDiF*n=lxun*6!5YKq(*0tM(btPK|HJQHt;(k0CW}a_A^<88l5xX}))wK^Q zoU@$Tn^X3f7Tpr!E7^Jy$hXB@oz8+JYy}E&RSMsM%&aSUDxv{&=;y+hp4jzphJKVo z0S>CmkU4Z&SL(^!lV>CKDKm1t zT!6Eci)uv{r$YdL6ZT}Iv#7UduZ&J(l-Sqm<%q7<%OTJVB8SjS^At~=ZK^MM<)_m- zde(VR+GzEa$Kc*a;O|j89Lb*QDK^PR0vcbAd631-%hvaD7sepJwL6&K%J#Bl>;z0d zXz>@_Hw}-umzm}J#=OY@RK!1wW)eGrW6T}-2t&ofJfcAf0l5@nuTy~v0&<*q8G*)K zOsB#hVL<^F1e6C+k68=$TUIGh;A}sjb|Aw*Z0@6JpUR0Q941hOeryQ%m>2!`fmF$c zMn>o;{d4w@+572BWAm@gr|)H}=hxB`e@isKH*{-g zHTp6COXcR!{lvxUxowSPavv%mxZA3n)E;#nzcqGqcv~j1)>Znu*M9xlZ)QJ@o_yGO zd_yH_Y?Hv8?I=lRS4)4$-_1Wr46Z4IUq5al%|D`FZz&-45#0AOBT406>i9^Dwj3!e zT4tFqej^+TFUD#cOKF{kFDKA2>sI_ga^GMWu0tZSCFa&Z3wL8@;RADkkJhh}zbD&Q zN^6OpHKj+q87u~{lo$UwjN*>}KffVi^g`aDoPm+^c@5HXCBiwQrn0HvYn*$jOP zpt)8u9Z3HS)eJiPT778n-2%~P3d&T#8VVc9@dDZOJmCpp36KM^z#LBl_z}85zHD)! zy~sjAqypuU%bY-F6=(ltqeEdC^F#I$EQJS5{NDrFkSU}%Nv6ubqJ;K-K@NXGPJK>V zA8AeR+Ew{sv}djN zy56-G@4gQr2?EfTOr+io9Bp)>G;=+3Gkc%5ZAz5(qOg4fM;lS1ByTQ1kh7mFt)D50 Y8x!wLtjMeRkDmNgDL~X#x&<8n1(UAK=3.0 +fastapi>=0.115 +uvicorn>=0.30 diff --git a/go.mod b/go.mod new file mode 100644 index 000000000..31e6a5cf3 --- /dev/null +++ b/go.mod @@ -0,0 +1,23 @@ +module github.com/emil/deepres + +go 1.25.0 + +require ( + github.com/go-telegram-bot-api/telegram-bot-api/v5 v5.5.1 + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.7.4 + github.com/nats-io/nats.go v1.41.2 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/klauspost/compress v1.18.5 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.49.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.42.0 // indirect + golang.org/x/text v0.35.0 // indirect +) diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 000000000..e60a05981 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,74 @@ +// DeepRes — Telegram bot for deep research +package config + +import ( + "fmt" + "os" + "strconv" +) + +type Config struct { + // Telegram + TelegramToken string + + // Database + DatabaseURL string + + // NATS + NATSURL string + + // SearXNG + SearXNGURL string + + // LLM API (OpenRouter) + OpenRouterKey string + LLMModel string + + // Embeddings (E5 via sentence-transformers HTTP server) + EmbedURL string + + // Prices in Telegram Stars + FastResearchCost int + DeepResearchCost int + FreeFastPerDay int +} + +func Load() *Config { + cfg := &Config{ + TelegramToken: mustEnv("TELEGRAM_TOKEN"), + DatabaseURL: mustEnv("DATABASE_URL"), + NATSURL: getEnv("NATS_URL", "nats://localhost:4222"), + SearXNGURL: getEnv("SEARXNG_URL", "http://localhost:4000"), + OpenRouterKey: mustEnv("OPENROUTER_API_KEY"), + LLMModel: getEnv("LLM_MODEL", "deepseek/deepseek-v4-flash"), + EmbedURL: getEnv("EMBED_URL", "http://localhost:8081"), + FastResearchCost: getEnvInt("FAST_COST", 1), + DeepResearchCost: getEnvInt("DEEP_COST", 5), + FreeFastPerDay: getEnvInt("FREE_FAST_PER_DAY", 3), + } + return cfg +} + +func getEnv(key, defaultVal string) string { + if v := os.Getenv(key); v != "" { + return v + } + return defaultVal +} + +func getEnvInt(key string, defaultVal int) int { + if v := os.Getenv(key); v != "" { + if i, err := strconv.Atoi(v); err == nil { + return i + } + } + return defaultVal +} + +func mustEnv(key string) string { + v := os.Getenv(key) + if v == "" { + panic(fmt.Sprintf("required environment variable %s is not set", key)) + } + return v +} diff --git a/internal/db/db.go b/internal/db/db.go new file mode 100644 index 000000000..da581b22d --- /dev/null +++ b/internal/db/db.go @@ -0,0 +1,166 @@ +package db + +import ( + "context" + "fmt" + "time" + + "github.com/emil/deepres/internal/models" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type DB struct { + pool *pgxpool.Pool +} + +func New(databaseURL string) (*DB, error) { + pool, err := pgxpool.New(context.Background(), databaseURL) + if err != nil { + return nil, fmt.Errorf("connect to db: %w", err) + } + return &DB{pool: pool}, nil +} + +func (db *DB) Close() { db.pool.Close() } + +// ---- Users ---- + +func (db *DB) GetOrCreateUser(ctx context.Context, tgID int64, username, name string) (*models.User, error) { + u := &models.User{} + err := db.pool.QueryRow(ctx, ` + INSERT INTO users (tg_id, username, name) VALUES ($1, $2, $3) + ON CONFLICT (tg_id) DO UPDATE SET username=$2, name=$3 + RETURNING tg_id, username, name, stars_balance, created_at + `, tgID, username, name).Scan(&u.TelegramID, &u.Username, &u.Name, &u.StarsBalance, &u.CreatedAt) + if err != nil { + return nil, fmt.Errorf("get or create user: %w", err) + } + return u, nil +} + +func (db *DB) GetUser(ctx context.Context, tgID int64) (*models.User, error) { + u := &models.User{} + err := db.pool.QueryRow(ctx, + "SELECT tg_id, username, name, stars_balance, created_at FROM users WHERE tg_id=$1", tgID, + ).Scan(&u.TelegramID, &u.Username, &u.Name, &u.StarsBalance, &u.CreatedAt) + if err == pgx.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("get user: %w", err) + } + return u, nil +} + +// ---- Research ---- + +func (db *DB) CreateResearch(ctx context.Context, r *models.Research) error { + r.ID = uuid.New() + r.Status = "pending" + r.CreatedAt = time.Now() + _, err := db.pool.Exec(ctx, ` + INSERT INTO researches (id, tg_id, query, mode, status, stars_cost, created_at) + VALUES ($1,$2,$3,$4,$5,$6,$7) + `, r.ID, r.TelegramID, r.Query, r.Mode, r.Status, r.StarsCost, r.CreatedAt) + return err +} + +func (db *DB) UpdateResearch(ctx context.Context, id uuid.UUID, status, summary string, reportPath string, sources []byte, errMsg *string) error { + now := time.Now() + _, err := db.pool.Exec(ctx, ` + UPDATE researches SET status=$2, summary=$3, report_path=$4, sources=$5, error_message=$6, completed_at=$7 + WHERE id=$1 + `, id, status, summary, reportPath, sources, errMsg, now) + return err +} + +func (db *DB) GetResearch(ctx context.Context, id uuid.UUID) (*models.Research, error) { + r := &models.Research{} + err := db.pool.QueryRow(ctx, "SELECT id,tg_id,query,mode,status,stars_cost,summary,report_path,sources,error_message,created_at,completed_at FROM researches WHERE id=$1", id). + Scan(&r.ID, &r.TelegramID, &r.Query, &r.Mode, &r.Status, &r.StarsCost, &r.Summary, &r.ReportPath, &r.SourcesJSON, &r.ErrorMessage, &r.CreatedAt, &r.CompletedAt) + if err != nil { + return nil, err + } + return r, nil +} + +// ---- Transactions ---- + +// AddTransaction добавляет транзакцию и обновляет баланс атомарно +func (db *DB) AddTransaction(ctx context.Context, t *models.Transaction) error { + tx, err := db.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback(ctx) + + t.ID = uuid.New() + t.CreatedAt = time.Now() + _, err = tx.Exec(ctx, ` + INSERT INTO transactions (id, tg_id, amount, type, description, telegram_charge_id, created_at) + VALUES ($1,$2,$3,$4,$5,$6,$7) + `, t.ID, t.TelegramID, t.Amount, t.Type, t.Description, t.TelegramChargeID, t.CreatedAt) + if err != nil { + return fmt.Errorf("insert transaction: %w", err) + } + + // Обновляем баланс + _, err = tx.Exec(ctx, "UPDATE users SET stars_balance = stars_balance + $1 WHERE tg_id=$2", t.Amount, t.TelegramID) + if err != nil { + return fmt.Errorf("update balance: %w", err) + } + + return tx.Commit(ctx) +} + +// DeductResearchCost списывает звёзды за ресерч атомарно +func (db *DB) DeductResearchCost(ctx context.Context, userID int64, researchID string, cost int, description string) error { + tx, err := db.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback(ctx) + + // Проверяем баланс + var balance int + err = tx.QueryRow(ctx, "SELECT stars_balance FROM users WHERE tg_id=$1 FOR UPDATE", userID).Scan(&balance) + if err != nil { + return fmt.Errorf("get balance: %w", err) + } + if balance < cost { + return fmt.Errorf("insufficient balance: %d < %d", balance, cost) + } + + // Создаём транзакцию + _, err = tx.Exec(ctx, ` + INSERT INTO transactions (id, tg_id, amount, type, description, created_at) + VALUES (gen_random_uuid(), $1, $2, 'research', $3, now()) + `, userID, -cost, description) + if err != nil { + return fmt.Errorf("insert transaction: %w", err) + } + + // Списываем + _, err = tx.Exec(ctx, "UPDATE users SET stars_balance = stars_balance - $1 WHERE tg_id=$2", cost, userID) + if err != nil { + return fmt.Errorf("deduct balance: %w", err) + } + + return tx.Commit(ctx) +} + +func (db *DB) UpdateStarsBalance(ctx context.Context, tgID int64, delta int) error { + _, err := db.pool.Exec(ctx, "UPDATE users SET stars_balance = stars_balance + $1 WHERE tg_id=$2", delta, tgID) + return err +} + +func (db *DB) GetTodayFastCount(ctx context.Context, tgID int64) (int, error) { + var count int + err := db.pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM researches + WHERE tg_id=$1 AND mode='fast' AND created_at > CURRENT_DATE + `, tgID).Scan(&count) + return count, err +} diff --git a/internal/models/models.go b/internal/models/models.go new file mode 100644 index 000000000..3d627d275 --- /dev/null +++ b/internal/models/models.go @@ -0,0 +1,61 @@ +package models + +import ( + "time" + + "github.com/google/uuid" +) + +type User struct { + TelegramID int64 `db:"tg_id"` + Username string `db:"username"` + Name string `db:"name"` + StarsBalance int `db:"stars_balance"` + CreatedAt time.Time `db:"created_at"` +} + +type Research struct { + ID uuid.UUID `db:"id"` + TelegramID int64 `db:"tg_id"` + Query string `db:"query"` + Mode string `db:"mode"` // fast | deep + Status string `db:"status"` // pending | processing | done | failed + StarsCost int `db:"stars_cost"` + Summary *string `db:"summary"` + ReportPath *string `db:"report_path"` + SourcesJSON []byte `db:"sources"` + ErrorMessage *string `db:"error_message"` + CreatedAt time.Time `db:"created_at"` + CompletedAt *time.Time `db:"completed_at"` +} + +type Transaction struct { + ID uuid.UUID `db:"id"` + TelegramID int64 `db:"tg_id"` + Amount int `db:"amount"` + Type string `db:"type"` // purchase | research + Description string `db:"description"` + TelegramChargeID *string `db:"telegram_charge_id"` + CreatedAt time.Time `db:"created_at"` +} + +type ResearchRequest struct { + UserTelegramID int64 `json:"tg_id"` + Query string `json:"query"` + Mode string `json:"mode"` + ResearchID string `json:"research_id"` +} + +type ResearchResult struct { + ResearchID string `json:"research_id"` + Status string `json:"status"` + Summary string `json:"summary"` + ReportPath string `json:"report_path"` + Sources []Source `json:"sources"` + Error string `json:"error,omitempty"` +} + +type Source struct { + Title string `json:"title"` + URL string `json:"url"` +} diff --git a/internal/queue/queue.go b/internal/queue/queue.go new file mode 100644 index 000000000..9cf2690cf --- /dev/null +++ b/internal/queue/queue.go @@ -0,0 +1,63 @@ +package queue + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/emil/deepres/internal/models" + "github.com/nats-io/nats.go" +) + +type Queue struct { + conn *nats.Conn +} + +func New(url string) (*Queue, error) { + conn, err := nats.Connect(url, nats.Timeout(5*time.Second)) + if err != nil { + return nil, fmt.Errorf("connect to nats: %w", err) + } + return &Queue{conn: conn}, nil +} + +func (q *Queue) Close() { q.conn.Close() } + +func (q *Queue) PublishFast(ctx context.Context, req *models.ResearchRequest) error { + data, err := json.Marshal(req) + if err != nil { + return fmt.Errorf("marshal fast request: %w", err) + } + return q.conn.Publish("research.fast", data) +} + +func (q *Queue) PublishDeep(ctx context.Context, req *models.ResearchRequest) error { + data, err := json.Marshal(req) + if err != nil { + return fmt.Errorf("marshal deep request: %w", err) + } + return q.conn.Publish("research.deep", data) +} + +func (q *Queue) SubscribeFast(handler func(*models.ResearchRequest)) (*nats.Subscription, error) { + sub, err := q.conn.QueueSubscribe("research.fast", "workers", func(msg *nats.Msg) { + var req models.ResearchRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + return + } + handler(&req) + }) + return sub, err +} + +func (q *Queue) SubscribeDeep(handler func(*models.ResearchRequest)) (*nats.Subscription, error) { + sub, err := q.conn.QueueSubscribe("research.deep", "workers", func(msg *nats.Msg) { + var req models.ResearchRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + return + } + handler(&req) + }) + return sub, err +} diff --git a/internal/research/dedup/dedup.go b/internal/research/dedup/dedup.go new file mode 100644 index 000000000..e85a0ecb2 --- /dev/null +++ b/internal/research/dedup/dedup.go @@ -0,0 +1,105 @@ +package dedup + +import ( + "context" + "sort" + + "github.com/emil/deepres/internal/research/embed" + "github.com/emil/deepres/internal/research/search" +) + +// DedupResult содержит результаты после дедупликации +type DedupResult struct { + Items []DedupItem + RawTexts []string +} + +type DedupItem struct { + Title string + URL string + Content string + ClusterID int +} + +// Deduplicate принимает результаты поиска, эмбеддит, кластеризует и дедуплицирует +func Deduplicate(ctx context.Context, embedCli *embed.EmbedClient, results []search.SearXNGResult, threshold float32) (*DedupResult, error) { + if len(results) == 0 { + return &DedupResult{}, nil + } + + // Подготавливаем тексты для эмбеддинга + texts := make([]string, len(results)) + for i, r := range results { + content := r.Title + ". " + r.Content + if len([]rune(content)) > 512 { + content = string([]rune(content)[:512]) + } + texts[i] = content + } + + // Эмбеддинги + embeddings, err := embedCli.Embed(ctx, texts) + if err != nil { + return nil, err + } + + // Простая дедупликация: если косинусная близость > threshold — это дубликат + n := len(results) + clusters := make([]int, n) + clusterID := 0 + for i := range clusters { + clusters[i] = -1 + } + + for i := 0; i < n; i++ { + if clusters[i] != -1 { + continue + } + clusters[i] = clusterID + for j := i + 1; j < n; j++ { + if clusters[j] != -1 { + continue + } + sim := embed.CosineSimilarity(embeddings[i], embeddings[j]) + if sim >= threshold { + clusters[j] = clusterID + } + } + clusterID++ + } + + // Собираем результат: по одному лучшему элементу из каждого кластера + unique := make(map[int]*DedupItem) + rawTexts := make([]string, 0) + + // Сначала собираем все + for i, r := range results { + key := clusters[i] + if _, exists := unique[key]; !exists { + unique[key] = &DedupItem{ + Title: r.Title, + URL: r.URL, + Content: r.Content, + } + } + } + + // Сортируем по clusterID + keys := make([]int, 0, len(unique)) + for k := range unique { + keys = append(keys, k) + } + sort.Ints(keys) + + var items []DedupItem + for _, k := range keys { + item := unique[k] + items = append(items, *item) + rawTexts = append(rawTexts, item.Title+". "+item.Content) + } + + return &DedupResult{ + Items: items, + RawTexts: rawTexts, + }, nil +} diff --git a/internal/research/embed/embed.go b/internal/research/embed/embed.go new file mode 100644 index 000000000..06ab655a8 --- /dev/null +++ b/internal/research/embed/embed.go @@ -0,0 +1,105 @@ +package embed + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "time" +) + +type EmbedClient struct { + baseURL string + client *http.Client + model string +} + +type EmbedRequest struct { + Model string `json:"model"` + Input []string `json:"input"` +} + +type EmbedResponse struct { + Data []struct { + Embedding []float32 `json:"embedding"` + } `json:"data"` +} + +func NewEmbedClient(baseURL string) *EmbedClient { + return &EmbedClient{ + baseURL: baseURL, + client: &http.Client{Timeout: 30 * time.Second}, + model: "intfloat/e5-small-v2", + } +} + +func (c *EmbedClient) Embed(ctx context.Context, texts []string) ([][]float32, error) { + if len(texts) == 0 { + return [][]float32{}, nil + } + if len(texts) > 100 { + return nil, fmt.Errorf("batch too large: %d > 100", len(texts)) + } + + body := EmbedRequest{ + Model: c.model, + Input: texts, + } + payload, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("marshal embed request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/v1/embeddings", bytes.NewReader(payload)) + if err != nil { + return nil, fmt.Errorf("embed request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("embed call: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("embed service returned %d: %s", resp.StatusCode, string(bodyBytes)) + } + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read embed response: %w", err) + } + + var er EmbedResponse + if err := json.Unmarshal(respBody, &er); err != nil { + return nil, fmt.Errorf("parse embed response: %w", err) + } + if len(er.Data) != len(texts) { + return nil, fmt.Errorf("embed mismatch: expected %d embeddings, got %d", len(texts), len(er.Data)) + } + + result := make([][]float32, len(er.Data)) + for i, d := range er.Data { + result[i] = d.Embedding + } + return result, nil +} + +// CosineSimilarity вычисляет косинусную близость между двумя векторами +func CosineSimilarity(a, b []float32) float32 { + var dot, normA, normB float64 + for i := range a { + dot += float64(a[i]) * float64(b[i]) + normA += float64(a[i]) * float64(a[i]) + normB += float64(b[i]) * float64(b[i]) + } + if normA == 0 || normB == 0 { + return 0 + } + return float32(dot / (math.Sqrt(normA) * math.Sqrt(normB))) +} diff --git a/internal/research/pipeline/pipeline.go b/internal/research/pipeline/pipeline.go new file mode 100644 index 000000000..74c8c7060 --- /dev/null +++ b/internal/research/pipeline/pipeline.go @@ -0,0 +1,219 @@ +package pipeline + +import ( + "context" + "fmt" + "log" + "os" + "strings" + "time" + + "github.com/emil/deepres/internal/models" + "github.com/emil/deepres/internal/research/dedup" + "github.com/emil/deepres/internal/research/embed" + "github.com/emil/deepres/internal/research/search" + "github.com/emil/deepres/internal/research/synthesize" +) + +type Pipeline struct { + searchCli *search.SearXNGClient + embedCli *embed.EmbedClient + llmCli *synthesize.LLMClient + reportsDir string +} + +func New(searchCli *search.SearXNGClient, embedCli *embed.EmbedClient, llmCli *synthesize.LLMClient, reportsDir string) *Pipeline { + // Создаём директорию для отчётов если нет + if reportsDir != "" { + _ = os.MkdirAll(reportsDir, 0755) + } + return &Pipeline{ + searchCli: searchCli, + embedCli: embedCli, + llmCli: llmCli, + reportsDir: reportsDir, + } +} + +func (p *Pipeline) RunFast(ctx context.Context, req *models.ResearchRequest) (*models.ResearchResult, error) { + log.Printf("[fast] starting research: %s", req.Query) + + // 1. Поиск — 3-5 запросов с разными формулировками + variations := generateQueryVariations(req.Query) + results, err := p.searchCli.SearchWithVariations(ctx, req.Query, variations, 10) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + if len(results) == 0 { + return nil, fmt.Errorf("ничего не найдено по запросу") + } + log.Printf("[fast] found %d raw results", len(results)) + + // 2. Эмбеддинги + дедупликация + deduped, err := dedup.Deduplicate(ctx, p.embedCli, results, 0.85) + if err != nil { + return nil, fmt.Errorf("dedup: %w", err) + } + log.Printf("[fast] after dedup: %d unique sources", len(deduped.Items)) + + if len(deduped.RawTexts) == 0 { + return nil, fmt.Errorf("не удалось обработать источники") + } + + // 3. LLM-синтез выжимки + prompt := synthesize.BuildFastResearchPrompt(req.Query, deduped.RawTexts) + summary, err := p.llmCli.Chat(ctx, prompt) + if err != nil { + return nil, fmt.Errorf("synthesize: %w", err) + } + + // 4. Формируем источники + var sources []models.Source + for _, item := range deduped.Items { + sources = append(sources, models.Source{Title: item.Title, URL: item.URL}) + } + return &models.ResearchResult{ + ResearchID: req.ResearchID, + Status: "done", + Summary: summary, + Sources: sources, + }, nil +} + +func (p *Pipeline) RunDeep(ctx context.Context, req *models.ResearchRequest) (*models.ResearchResult, error) { + log.Printf("[deep] starting research: %s", req.Query) + + // 1. Поиск — 10-15 запросов + variations := generateQueryVariations(req.Query) + results, err := p.searchCli.SearchWithVariations(ctx, req.Query, variations, 15) + if err != nil { + return nil, fmt.Errorf("search: %w", err) + } + if len(results) == 0 { + return nil, fmt.Errorf("ничего не найдено по запросу") + } + + // 2. Дедупликация + deduped, err := dedup.Deduplicate(ctx, p.embedCli, results, 0.80) + if err != nil { + return nil, fmt.Errorf("dedup: %w", err) + } + if len(deduped.RawTexts) == 0 { + return nil, fmt.Errorf("не удалось обработать источники") + } + + // 3. Первый проход LLM — определяем чего не хватает + prompt1 := synthesize.BuildGapPrompt(req.Query, deduped.RawTexts) + gapAnswer, err := p.llmCli.Chat(ctx, prompt1) + if err != nil { + return nil, fmt.Errorf("gap analysis: %w", err) + } + + // 4. Второй раунд поиска по недостающим аспектам + gapVariations := extractFollowUpQueries(gapAnswer) + if len(gapVariations) > 0 { + log.Printf("[deep] follow-up queries: %v", gapVariations) + moreResults, err := p.searchCli.SearchWithVariations(ctx, req.Query, gapVariations, 5) + if err == nil && len(moreResults) > 0 { + moreDeduped, err := dedup.Deduplicate(ctx, p.embedCli, moreResults, 0.80) + if err == nil { + // Добавляем только новые URL + seen := make(map[string]bool) + for _, item := range deduped.Items { + seen[item.URL] = true + } + for _, item := range moreDeduped.Items { + if !seen[item.URL] { + deduped.Items = append(deduped.Items, item) + deduped.RawTexts = append(deduped.RawTexts, item.Title+". "+item.Content) + } + } + } + } + } + + // 5. Финальный синтез + prompt2 := synthesize.BuildDeepResearchPrompt(req.Query, deduped.RawTexts, "") + finalReport, err := p.llmCli.Chat(ctx, prompt2) + if err != nil { + return nil, fmt.Errorf("final pass: %w", err) + } + + // 6. Сохраняем отчёт в файл + reportPath := "" + if p.reportsDir != "" { + reportPath = fmt.Sprintf("%s/%s_%s.md", p.reportsDir, time.Now().Format("20060102"), req.ResearchID[:8]) + if err := p.saveReport(reportPath, finalReport, deduped.Items, req.Query); err != nil { + log.Printf("[deep] save report error: %v", err) + reportPath = "" + } + } + + var sources []models.Source + for _, item := range deduped.Items { + sources = append(sources, models.Source{Title: item.Title, URL: item.URL}) + } + + return &models.ResearchResult{ + ResearchID: req.ResearchID, + Status: "done", + Summary: truncateText(finalReport, 1500), + ReportPath: reportPath, + Sources: sources, + }, nil +} + +func generateQueryVariations(base string) []string { + return []string{ + base + " обзор", + base + " анализ", + base + " исследование", + base + " последние новости", + base + " что это такое", + base + " проблемы и решения", + base + " будущее", + } +} + +func extractFollowUpQueries(text string) []string { + // Простой парсер: ищем строки, которые выглядят как вопросы или темы + var queries []string + lines := strings.Split(text, "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + // Ищем номерованные строки или буллеты + if strings.HasPrefix(line, "-") || strings.HasPrefix(line, "*") || (len(line) > 10 && len(line) < 200) { + // Убираем маркеры и добавляем в запросы + clean := strings.TrimPrefix(line, "-") + clean = strings.TrimPrefix(clean, "*") + clean = strings.TrimSpace(clean) + if clean != "" && len(clean) > 5 { + queries = append(queries, clean) + } + } + } + // Ограничиваем до 3 запросов + if len(queries) > 3 { + queries = queries[:3] + } + return queries +} + +func (p *Pipeline) saveReport(path, report string, items []dedup.DedupItem, query string) error { + content := fmt.Sprintf("# %s\n\n%s\n\n## Источники\n\n", query, report) + for _, item := range items { + content += fmt.Sprintf("- [%s](%s)\n", item.Title, item.URL) + } + return os.WriteFile(path, []byte(content), 0644) +} + +func truncateText(s string, maxLen int) string { + runes := []rune(s) + if len(runes) <= maxLen { + return s + } + return string(runes[:maxLen]) + "..." +} diff --git a/internal/research/search/search.go b/internal/research/search/search.go new file mode 100644 index 000000000..105d53137 --- /dev/null +++ b/internal/research/search/search.go @@ -0,0 +1,104 @@ +package search + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "time" +) + +type SearXNGClient struct { + baseURL string + client *http.Client +} + +type SearXNGResult struct { + Title string `json:"title"` + URL string `json:"url"` + Content string `json:"content"` + Engine string `json:"engine"` +} + +type SearXNGResponse struct { + Results []SearXNGResult `json:"results"` +} + +func NewSearXNGClient(baseURL string) *SearXNGClient { + return &SearXNGClient{ + baseURL: baseURL, + client: &http.Client{Timeout: 15 * time.Second}, + } +} + +func (c *SearXNGClient) Search(ctx context.Context, query string, limit int) ([]SearXNGResult, error) { + u, err := url.Parse(c.baseURL + "/search") + if err != nil { + return nil, fmt.Errorf("parse url: %w", err) + } + q := u.Query() + q.Set("q", query) + q.Set("format", "json") + q.Set("language", "ru-RU,en-US") + q.Set("categories", "general,news") + if limit > 0 { + q.Set("pageno", "1") + } + u.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("User-Agent", "DeepRes/1.0") + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("search request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("searxng returned %d: %s", resp.StatusCode, string(body)) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read response: %w", err) + } + + var sr SearXNGResponse + if err := json.Unmarshal(body, &sr); err != nil { + return nil, fmt.Errorf("parse response: %w", err) + } + + if len(sr.Results) > limit { + sr.Results = sr.Results[:limit] + } + return sr.Results, nil +} + +// SearchWithVariations запускает несколько поисковых запросов с разными формулировками +func (c *SearXNGClient) SearchWithVariations(ctx context.Context, baseQuery string, variations []string, limit int) ([]SearXNGResult, error) { + seen := make(map[string]bool) + var allResults []SearXNGResult + + queries := append([]string{baseQuery}, variations...) + for _, q := range queries { + results, err := c.Search(ctx, q, limit) + if err != nil { + // Логируем но продолжаем — один упавший запрос не ломает весь ресерч + continue + } + for _, r := range results { + if !seen[r.URL] { + seen[r.URL] = true + allResults = append(allResults, r) + } + } + } + return allResults, nil +} diff --git a/internal/research/synthesize/synthesize.go b/internal/research/synthesize/synthesize.go new file mode 100644 index 000000000..9f795ba4d --- /dev/null +++ b/internal/research/synthesize/synthesize.go @@ -0,0 +1,145 @@ +package synthesize + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +type LLMClient struct { + apiKey string + model string + baseURL string + client *http.Client +} + +type Message struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type LLMRequest struct { + Model string `json:"model"` + Messages []Message `json:"messages"` + Temperature float32 `json:"temperature,omitempty"` + MaxTokens int `json:"max_tokens,omitempty"` +} + +type LLMResponse struct { + Choices []struct { + Message Message `json:"message"` + } `json:"choices"` + Error *struct { + Message string `json:"message"` + Code int `json:"code"` + } `json:"error,omitempty"` +} + +func NewLLMClient(baseURL, apiKey, model string) *LLMClient { + return &LLMClient{ + apiKey: apiKey, + model: model, + baseURL: baseURL, + client: &http.Client{Timeout: 120 * time.Second}, + } +} + +func (c *LLMClient) Chat(ctx context.Context, messages []Message) (string, error) { + reqBody := LLMRequest{ + Model: c.model, + Messages: messages, + Temperature: 0.7, + MaxTokens: 3000, + } + payload, err := json.Marshal(reqBody) + if err != nil { + return "", fmt.Errorf("marshal llm request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/v1/chat/completions", bytes.NewReader(payload)) + if err != nil { + return "", fmt.Errorf("llm request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+c.apiKey) + + resp, err := c.client.Do(req) + if err != nil { + return "", fmt.Errorf("llm call: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return "", fmt.Errorf("read llm response: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("llm returned %d: %s", resp.StatusCode, string(respBody)) + } + + var lr LLMResponse + if err := json.Unmarshal(respBody, &lr); err != nil { + return "", fmt.Errorf("parse llm response: %w", err) + } + if lr.Error != nil { + return "", fmt.Errorf("llm error %d: %s", lr.Error.Code, lr.Error.Message) + } + if len(lr.Choices) == 0 { + return "", fmt.Errorf("empty llm response: %s", string(respBody)) + } + return lr.Choices[0].Message.Content, nil +} + +// ---- Промпты ---- + +func BuildFastResearchPrompt(query string, sources []string) []Message { + srcText := strings.Join(sources, "\n\n---\n\n") + return []Message{ + { + Role: "system", + Content: "Ты — DeepRes, AI-исследователь. Твоя задача: на основе предоставленных источников написать структурированную выжимку по теме. Выдели ключевые факты, противоположные точки зрения, практические выводы. Если источники противоречат друг другу — укажи это. Пиши на русском, если тема на русском. Формат: кратко, ёмко, по делу.", + }, + { + Role: "user", + Content: fmt.Sprintf("Тема: %s\n\nИсточники:\n%s\n\nНапиши выжимку на 3-5 абзацев с ключевыми выводами и ссылками на источники.", query, srcText), + }, + } +} + +func BuildDeepResearchPrompt(query string, sources []string, followUpQ string) []Message { + srcText := strings.Join(sources, "\n\n---\n\n") + messages := []Message{ + { + Role: "system", + Content: "Ты — DeepRes, AI-исследователь. Напиши ПОЛНЫЙ отчёт по теме на основе источников. Структура: введение, ключевые находки, анализ, противоположные точки зрения, выводы и рекомендации. Отметь области, по которым недостаточно информации.", + }, + { + Role: "user", + Content: fmt.Sprintf("Тема: %s\n\nИсточники:\n%s", query, srcText), + }, + } + if followUpQ != "" { + messages = append(messages, Message{Role: "user", Content: followUpQ}) + } + return messages +} + +func BuildGapPrompt(query string, sources []string) []Message { + srcText := strings.Join(sources, "\n\n---\n\n") + return []Message{ + { + Role: "system", + Content: "Ты — аналитик, который ищет неизведанные области. Проанализируй предоставленные источники и найди:\n1) Какие аспекты темы НЕ освещены?\n2) Где источники противоречат друг другу?\n3) Какие вопросы остаются открытыми?\n4) Что было бы важно изучить дальше?", + }, + { + Role: "user", + Content: fmt.Sprintf("Тема: %s\n\nИсточники:\n%s", query, srcText), + }, + } +} diff --git a/migrations/001_init.sql b/migrations/001_init.sql new file mode 100644 index 000000000..74fdade06 --- /dev/null +++ b/migrations/001_init.sql @@ -0,0 +1,39 @@ +-- DeepRes initial schema +CREATE EXTENSION IF NOT EXISTS pgcrypto; + +CREATE TABLE IF NOT EXISTS users ( + tg_id BIGINT PRIMARY KEY, + username TEXT DEFAULT '', + name TEXT DEFAULT '', + stars_balance INT DEFAULT 0, + created_at TIMESTAMPTZ DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS researches ( + id UUID DEFAULT gen_random_uuid() PRIMARY KEY, + tg_id BIGINT REFERENCES users(tg_id) ON DELETE CASCADE, + query TEXT NOT NULL, + mode TEXT NOT NULL CHECK (mode IN ('fast', 'deep')), + status TEXT DEFAULT 'pending' CHECK (status IN ('pending', 'processing', 'done', 'failed')), + stars_cost INT DEFAULT 0, + summary TEXT, + report_path TEXT, + sources JSONB, + error_message TEXT, + created_at TIMESTAMPTZ DEFAULT now(), + completed_at TIMESTAMPTZ +); + +CREATE TABLE IF NOT EXISTS transactions ( + id UUID DEFAULT gen_random_uuid() PRIMARY KEY, + tg_id BIGINT REFERENCES users(tg_id) ON DELETE CASCADE, + amount INT NOT NULL, + type TEXT NOT NULL CHECK (type IN ('purchase', 'research', 'bonus', 'refund')), + description TEXT DEFAULT '', + telegram_charge_id TEXT, + created_at TIMESTAMPTZ DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_researches_tg_id ON researches(tg_id); +CREATE INDEX IF NOT EXISTS idx_researches_created_at ON researches(created_at); +CREATE INDEX IF NOT EXISTS idx_transactions_tg_id ON transactions(tg_id); diff --git a/run-bot.sh b/run-bot.sh new file mode 100755 index 000000000..efbaa3497 --- /dev/null +++ b/run-bot.sh @@ -0,0 +1,4 @@ +#!/bin/bash +cd ~/Desktop/DeepRes +export $(grep -v '^#' .env | xargs) +exec stdbuf -oL ./bot > /tmp/deepres-bot.log 2>&1 diff --git a/run-worker.sh b/run-worker.sh new file mode 100755 index 000000000..df8c43f5f --- /dev/null +++ b/run-worker.sh @@ -0,0 +1,4 @@ +#!/bin/bash +cd ~/Desktop/DeepRes +export $(grep -v '^#' .env | xargs) +exec stdbuf -oL ./worker > /tmp/deepres-worker.log 2>&1 diff --git a/searxng/settings.yml b/searxng/settings.yml new file mode 100644 index 000000000..b58698778 --- /dev/null +++ b/searxng/settings.yml @@ -0,0 +1,28 @@ +use_default_settings: true +general: + instance_name: "DeepRes" + enable_metrics: false +search: + safe_search: 0 + autocomplete: "" + formats: + - html + - json +server: + port: 8080 + bind_address: "0.0.0.0" + secret_key: "deepres-secret-change-in-production" + method: "POST" +ui: + default_locale: ru + search_on_category_select: true + hotkeys: "default" +engines: + - name: duckduckgo + disabled: false + - name: wikipedia + disabled: false + - name: wikidata + disabled: false + - name: stackoverflow + disabled: false