Merge pull request #115 from svc-design/codex/adjust-xcontrol-cli-workflow
feat(cli): support per-file ingestion workflow
This commit is contained in:
shenlan
GitHub
commit
c686e60e25
153
client/main.go
153
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, ", "))
|
||||
|
||||
@ -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 查询接口
|
||||
|
||||
|
||||
10
server/config/server.yaml
Normal file
10
server/config/server.yaml
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user