feat(rag): auto sync and git watch

docs/rag-system-design.md

@@ -26,25 +26,33 @@
 
 ## 4. 数据库设计
 
-使用 PostgreSQL + [pgvector](https://github.com/pgvector/pgvector)。建表及索引 SQL 如下：
+使用 PostgreSQL + [pgvector](https://github.com/pgvector/pgvector)。初始化步骤：
 
-```sql
-CREATE TABLE documents (
-    id BIGSERIAL PRIMARY KEY,
-    repo TEXT NOT NULL,         -- 来源仓库
-    path TEXT NOT NULL,         -- 文件路径
-    chunk_id INT NOT NULL,
-    content TEXT NOT NULL,
-    embedding VECTOR(1536),     -- 向量
-    metadata JSONB              -- 额外信息：标签/更新时间等
-);
+1. 在目标数据库中启用扩展：
+   ```sql
+   CREATE EXTENSION IF NOT EXISTS vector;
+   ```
+2. 按以下 SQL 创建存储向量的表及索引：
 
--- 向量索引
-CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
+   ```sql
+   CREATE TABLE documents (
+       id BIGSERIAL PRIMARY KEY,
+       repo TEXT NOT NULL,         -- 来源仓库
+       path TEXT NOT NULL,         -- 文件路径
+       chunk_id INT NOT NULL,
+       content TEXT NOT NULL,
+       embedding VECTOR(1536),     -- 向量
+       metadata JSONB              -- 额外信息：标签/更新时间等
+   );
 
--- 元数据索引
-CREATE INDEX idx_documents_metadata ON documents USING gin (metadata);
-```
+   -- 向量索引
+   CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
+
+   -- 元数据索引
+   CREATE INDEX idx_documents_metadata ON documents USING gin (metadata);
+   ```
+
+连接字符串示例：`postgres://user:password@127.0.0.1:5432`。
 
 ## 5. 检索与问答流程
 

server/api/rag.go

@@ -30,11 +30,15 @@ func initRAG() *rag.Service {
 		return nil
 	}
 	key := os.Getenv("OPENAI_API_KEY")
-	if key == "" {
-		return rag.New(nil, st, nil)
+	svcCfg := cfg.ToConfig()
+	var emb embed.Embedder
+	if key != "" {
+		emb = embed.NewOpenAI("text-embedding-3-small", key)
 	}
-	emb := embed.NewOpenAI("text-embedding-3-small", key)
-	return rag.New(nil, st, emb)
+	svc := rag.New(svcCfg, st, emb)
+	go svc.Sync(context.Background())
+	go svc.Watch(context.Background())
+	return svc
 }
 
 // registerRAGRoutes wires the /api/rag endpoints.

server/rag/config/runtime.go

@@ -8,6 +8,12 @@ import (
 )
 
 // Runtime holds runtime configuration for RAG features.
+type Datasource struct {
+	Name string `yaml:"name"`
+	Repo string `yaml:"repo"`
+	Path string `yaml:"path"`
+}
+
 type Runtime struct {
 	Redis struct {
 		Addr     string `yaml:"addr"`
@@ -17,7 +23,7 @@ type Runtime struct {
 	VectorDB struct {
 		PGURL string `yaml:"pgurl"`
 	} `yaml:"vectordb"`
-	Datasources []string `yaml:"datasources"`
+	Datasources []Datasource `yaml:"datasources"`
 }
 
 // LoadServer loads RAG configuration from server/config/server.yaml.
@@ -35,3 +41,19 @@ func LoadServer() (*Runtime, error) {
 	}
 	return &cfg.RAG, nil
 }
+
+// ToConfig converts runtime configuration into service configuration.
+func (rt *Runtime) ToConfig() *Config {
+	if rt == nil {
+		return nil
+	}
+	var c Config
+	for _, ds := range rt.Datasources {
+		c.Repos = append(c.Repos, Repo{
+			URL:   ds.Repo,
+			Paths: []string{ds.Path},
+			Local: filepath.Join("server", "rag", ds.Name),
+		})
+	}
+	return &c
+}

server/rag/rag.go

@@ -2,6 +2,7 @@ package rag
 
 import (
 	"context"
+	"time"
 
 	"xcontrol/server/rag/config"
 	"xcontrol/server/rag/embed"
@@ -32,30 +33,63 @@ func (s *Service) Sync(ctx context.Context) error {
 		return nil
 	}
 	for _, repo := range s.cfg.Repos {
-		files, err := rsync.Repo(repo)
-		if err != nil {
+		if err := s.syncRepo(ctx, repo, true); err != nil {
 			return err
 		}
-		for _, f := range files {
-			docs, err := ingest.File(repo.URL, f)
-			if err != nil {
-				continue
-			}
-			for i := range docs {
-				vec, err := s.emb.Embed(ctx, docs[i].Content)
-				if err != nil {
-					continue
-				}
-				docs[i].Embedding = vec
-			}
-			if err := s.st.Upsert(ctx, docs); err != nil {
-				return err
-			}
-		}
 	}
 	return nil
 }
 
+// syncRepo pulls markdown files for a single repo and ingests them.
+// If force is false, ingestion is skipped when the repo has no changes.
+func (s *Service) syncRepo(ctx context.Context, repo config.Repo, force bool) error {
+	files, changed, err := rsync.Repo(repo)
+	if err != nil {
+		return err
+	}
+	if !force && !changed {
+		return nil
+	}
+	for _, f := range files {
+		docs, err := ingest.File(repo.URL, f)
+		if err != nil {
+			continue
+		}
+		for i := range docs {
+			vec, err := s.emb.Embed(ctx, docs[i].Content)
+			if err != nil {
+				continue
+			}
+			docs[i].Embedding = vec
+		}
+		if err := s.st.Upsert(ctx, docs); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Watch monitors configured repositories and triggers sync on updates.
+func (s *Service) Watch(ctx context.Context) {
+	if s == nil || s.cfg == nil {
+		return
+	}
+	for _, repo := range s.cfg.Repos {
+		go func(r config.Repo) {
+			ticker := time.NewTicker(time.Minute)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					_ = s.syncRepo(ctx, r, false)
+				}
+			}
+		}(repo)
+	}
+}
+
 // Query embeds the question and searches the store for similar documents.
 // If the service is not fully configured, Query returns nil without error.
 func (s *Service) Query(ctx context.Context, question string, limit int) ([]store.Document, error) {

server/rag/sync/sync.go

@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"errors"
 	"io/fs"
 	"path/filepath"
 
@@ -9,21 +10,30 @@ import (
 )
 
 // Repo synchronizes the configured repository and returns markdown file paths.
-func Repo(c config.Repo) ([]string, error) {
+// The returned boolean indicates whether new commits were pulled.
+func Repo(c config.Repo) ([]string, bool, error) {
+	changed := false
 	if _, err := git.PlainOpen(c.Local); err != nil {
 		if _, err := git.PlainClone(c.Local, false, &git.CloneOptions{URL: c.URL}); err != nil {
-			return nil, err
+			return nil, false, err
 		}
+		changed = true
 	} else {
 		r, err := git.PlainOpen(c.Local)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		w, err := r.Worktree()
 		if err != nil {
-			return nil, err
+			return nil, false, err
+		}
+		if err := w.Pull(&git.PullOptions{RemoteName: "origin"}); err != nil {
+			if !errors.Is(err, git.NoErrAlreadyUpToDate) {
+				return nil, false, err
+			}
+		} else {
+			changed = true
 		}
-		_ = w.Pull(&git.PullOptions{RemoteName: "origin"})
 	}
 	var files []string
 	for _, p := range c.Paths {
@@ -41,5 +51,5 @@ func Repo(c config.Repo) ([]string, error) {
 			return nil
 		})
 	}
-	return files, nil
+	return files, changed, nil
 }