diff --git a/client/main.go b/client/main.go index 1f7e864..6d0c151 100644 --- a/client/main.go +++ b/client/main.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "flag" + "fmt" "log" "net/http" "os" @@ -20,14 +21,14 @@ import ( "xcontrol/server/proxy" ) -// main loads server RAG configuration and triggers a manual sync by -// calling the running API server's /api/rag/sync endpoint. When a file -// path is provided via -file, it parses the Markdown locally and emits -// chunk data as newline-delimited JSON. +// main either lists markdown files for ingestion or processes a single file. +// When invoked without -file, it synchronizes configured repositories and +// prints absolute paths of markdown files to stdout. When -file is provided, +// the file is parsed, embedded and sent to the /api/rag/upsert endpoint. func main() { configPath := flag.String("config", "", "Path to server RAG configuration file") - filePath := flag.String("file", "", "Markdown file to parse and chunk") + filePath := flag.String("file", "", "Markdown file to embed and upsert") flag.Parse() var cfg *rconfig.Config @@ -43,24 +44,8 @@ func main() { proxy.Set(cfg.Global.Proxy) - if *filePath != "" { - chunkCfg := cfg.ResolveChunking() - secs, err := ingest.ParseMarkdown(*filePath) - if err != nil { - log.Fatalf("parse markdown: %v", err) - } - chunks, err := ingest.BuildChunks(secs, chunkCfg) - if err != nil { - log.Fatalf("build chunks: %v", err) - } - enc := json.NewEncoder(os.Stdout) - for _, ch := range chunks { - if err := enc.Encode(ch); err != nil { - log.Fatalf("encode chunk: %v", err) - } - } - return - } + embCfg := cfg.ResolveEmbedding() + chunkCfg := cfg.ResolveChunking() baseURL := os.Getenv("SERVER_URL") if baseURL == "" { @@ -70,64 +55,57 @@ func main() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() - embCfg := cfg.ResolveEmbedding() - chunkCfg := cfg.ResolveChunking() - var embedder embed.Embedder - if embCfg.Model != "" { - embedder = embed.NewOpenAI(embCfg.BaseURL, embCfg.APIKey, embCfg.Model, embCfg.Dimension) - } else { - embedder = embed.NewBGE(embCfg.BaseURL, embCfg.APIKey, embCfg.Dimension) - } - var syncErrs []string - - for _, ds := range cfg.Global.Datasources { - workdir := filepath.Join(os.TempDir(), "xcontrol", ds.Name) - if _, err := rsync.SyncRepo(ctx, ds.Repo, workdir); err != nil { - log.Printf("sync repo %s: %v", ds.Name, err) - syncErrs = append(syncErrs, ds.Name) - continue + if *filePath != "" { + var ds *rconfig.DataSource + var workdir string + for i := range cfg.Global.Datasources { + wd := filepath.Join(os.TempDir(), "xcontrol", cfg.Global.Datasources[i].Name) + if strings.HasPrefix(*filePath, wd) { + ds = &cfg.Global.Datasources[i] + workdir = wd + break + } } - root := filepath.Join(workdir, ds.Path) - files, err := ingest.ListMarkdown(root, chunkCfg.IncludeExts, chunkCfg.IgnoreDirs, 0) + if ds == nil { + log.Fatalf("file %s not under any datasource", *filePath) + } + + var embedder embed.Embedder + if embCfg.Model != "" { + embedder = embed.NewOpenAI(embCfg.BaseURL, embCfg.APIKey, embCfg.Model, embCfg.Dimension) + } else { + embedder = embed.NewBGE(embCfg.BaseURL, embCfg.APIKey, embCfg.Dimension) + } + + secs, err := ingest.ParseMarkdown(*filePath) if err != nil { - log.Fatalf("list markdown: %v", err) + log.Fatalf("parse markdown: %v", err) } - var rows []store.DocRow - for _, f := range files { - secs, err := ingest.ParseMarkdown(f) - if err != nil { - log.Printf("parse %s: %v", f, err) - continue - } - chunks, err := ingest.BuildChunks(secs, chunkCfg) - if err != nil { - log.Printf("chunk %s: %v", f, err) - continue - } - texts := make([]string, len(chunks)) - rs := make([]store.DocRow, len(chunks)) - for i, ch := range chunks { - texts[i] = ch.Text - rs[i] = store.DocRow{ - Repo: ds.Repo, - Path: strings.TrimPrefix(f, workdir+"/"), - ChunkID: ch.ChunkID, - Content: ch.Text, - Metadata: ch.Meta, - ContentSHA: ch.SHA256, - } - } - vecs, _, err := embedder.Embed(ctx, texts) - if err != nil { - log.Printf("embed %s: %v", f, err) - continue - } - for i := range rs { - rs[i].Embedding = vecs[i] - } - rows = append(rows, rs...) + chunks, err := ingest.BuildChunks(secs, chunkCfg) + if err != nil { + log.Fatalf("build chunks: %v", err) + } + texts := make([]string, len(chunks)) + rows := make([]store.DocRow, len(chunks)) + rel := strings.TrimPrefix(*filePath, workdir+"/") + for i, ch := range chunks { + texts[i] = ch.Text + rows[i] = store.DocRow{ + Repo: ds.Repo, + Path: rel, + ChunkID: ch.ChunkID, + Content: ch.Text, + Metadata: ch.Meta, + ContentSHA: ch.SHA256, + } + } + vecs, _, err := embedder.Embed(ctx, texts) + if err != nil { + log.Fatalf("embed %s: %v", *filePath, err) + } + for i := range rows { + rows[i].Embedding = vecs[i] } - payload := struct { Docs []store.DocRow `json:"docs"` }{Docs: rows} @@ -145,7 +123,26 @@ func main() { if resp.StatusCode != http.StatusOK { log.Fatalf("upsert failed: %s", resp.Status) } - log.Printf("ingested %d rows for %s", len(rows), ds.Name) + log.Printf("ingested %d chunks for %s", len(rows), rel) + return + } + + var syncErrs []string + for _, ds := range cfg.Global.Datasources { + workdir := filepath.Join(os.TempDir(), "xcontrol", ds.Name) + if _, err := rsync.SyncRepo(ctx, ds.Repo, workdir); err != nil { + log.Printf("sync repo %s: %v", ds.Name, err) + syncErrs = append(syncErrs, ds.Name) + continue + } + root := filepath.Join(workdir, ds.Path) + files, err := ingest.ListMarkdown(root, chunkCfg.IncludeExts, chunkCfg.IgnoreDirs, 0) + if err != nil { + log.Fatalf("list markdown: %v", err) + } + for _, f := range files { + fmt.Println(f) + } } if len(syncErrs) > 0 { log.Fatalf("failed to sync repositories: %s", strings.Join(syncErrs, ", ")) diff --git a/docs/rag-system-design.md b/docs/rag-system-design.md index 92618fb..f5906d8 100644 --- a/docs/rag-system-design.md +++ b/docs/rag-system-design.md @@ -119,15 +119,17 @@ datasources: ### 8.3 运行同步 ```bash -./xcontrol-cli sync --config rag.yaml +# 列出所有需要处理的 Markdown 文件路径 +./xcontrol-cli --config rag.yaml > files.txt + +# 逐个处理文件,生成向量并写入 pgvector +while read -r f; do + ./xcontrol-cli --config rag.yaml --file "$f" +done < files.txt ``` -运行过程会克隆或更新数据源仓库,解析 Markdown 并分块,调用模型生成向量,并写入 pgvector 数据库。成功后会看到类似下面的日志: - -``` -path: /2025/08/08 00:30:00 loaded 3 datasource(s) -2025/08/08 00:30:00 sync triggered -``` +第一步会克隆或更新数据源仓库,并将扫描到的 Markdown 文件绝对路径打印到标准输出。 +第二步读取这些路径,逐个解析、分块、调用 BGE Embed 并通过 `/api/rag/upsert` 写入数据库。 ### 8.4 查询接口 diff --git a/server/config/server.yaml b/server/config/server.yaml new file mode 100644 index 0000000..2dfab3a --- /dev/null +++ b/server/config/server.yaml @@ -0,0 +1,10 @@ +global: + redis: + addr: 127.0.0.1:6379 + vectordb: + pgurl: postgres://user:password@127.0.0.1:5432/postgres + datasources: [] +api: + askai: + timeout: 100 + retries: 0