diff --git a/docs/rag-system-design.md b/docs/rag-system-design.md index 7d19962..44e7efa 100644 --- a/docs/rag-system-design.md +++ b/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. 检索与问答流程 diff --git a/server/api/rag.go b/server/api/rag.go index 69cf53e..c9412ef 100644 --- a/server/api/rag.go +++ b/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. diff --git a/server/rag/config/runtime.go b/server/rag/config/runtime.go index 04e8823..54611e5 100644 --- a/server/rag/config/runtime.go +++ b/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 +} diff --git a/server/rag/rag.go b/server/rag/rag.go index c97ab8c..fd5e9cc 100644 --- a/server/rag/rag.go +++ b/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) { diff --git a/server/rag/sync/sync.go b/server/rag/sync/sync.go index 8adf6e1..ab05e4e 100644 --- a/server/rag/sync/sync.go +++ b/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 }