Initial commit: DeepRes Go backend — bot + worker + embed service

.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

.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/

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:

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"]

embed-service/main.py

@@ -0,0 +1,50 @@
+"""
+E5-small-v2 embedding server — OpenAI-compatible /v1/embeddings endpoint.
+Minimal, CPU-only, ~100MB RAM.
+"""
+from fastapi import FastAPI
+from pydantic import BaseModel
+from sentence_transformers import SentenceTransformer
+import numpy as np
+
+app = FastAPI()
+model = SentenceTransformer("intfloat/e5-small-v2", device="cpu")
+
+
+class EmbedRequest(BaseModel):
+    model: str = "intfloat/e5-small-v2"
+    input: list[str]
+
+
+class EmbedData(BaseModel):
+    object: str = "embedding"
+    index: int
+    embedding: list[float]
+
+
+class EmbedResponse(BaseModel):
+    object: str = "list"
+    data: list[EmbedData]
+    model: str
+
+
+@app.post("/v1/embeddings")
+def embed(req: EmbedRequest):
+    # E5 requires "query: " or "passage: " prefix
+    prefixed = [f"passage: {t}" if "\n" not in t else t for t in req.input]
+    embs = model.encode(prefixed, normalize_embeddings=True, show_progress_bar=False)
+    data = [
+        EmbedData(index=i, embedding=emb.tolist())
+        for i, emb in enumerate(embs)
+    ]
+    return EmbedResponse(data=data, model=req.model)
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok"}
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8081)

embed-service/requirements.txt

@@ -0,0 +1,3 @@
+sentence-transformers>=3.0
+fastapi>=0.115
+uvicorn>=0.30

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
+)

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
+}

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
+}

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"`
+}

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
+}

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
+}

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)))
+}

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]) + "..."
+}

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
+}

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),
+		},
+	}
+}

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);

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

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

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