feat: 实时入库监控 + 向量拆分可视化（异步入库 + 进度 SSE）

入库从同步改为异步流水线 + 进度回流（复用 token 流 NATS streaming）。
UI 实时看到 解析→切块→向量化(分批)→写入 各阶段 + 拆分块预览。

- shared: contract.IngestEvent(stage/done/total/chunks/error)
- mcp-go: rag.Ingest 加 onProgress + 分批向量化(10/批)逐批回报;kb_ingest 带 job_id
  把进度发到 sundynix.streams.<job_id> + CompleteStream
- gateway: 入库异步返回 job_id，后台 runIngest 发进度;GET /kb/ingest/:id/stream SSE
- frontend: streamIngest(EventSource);KbView 实时进度面板(阶段徽标+进度条+拆分列表)
- 验证: build✓+e2e PASS; 浏览器 12 行→6 阶段点亮+进度条 12/12+拆分 12 块逐条

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

sundynix-desktop/frontend/src/lib/api.ts

@@ -47,27 +47,57 @@ export function streamTokens(
   return () => es.close();
 }
 
-// ingestKb: POST /api/v1/kb/ingest，把文本入库（→ mcp-go kb_ingest：切块/embedding/Milvus）。
+// 入库进度事件（与后端 contract.IngestEvent 对应）。
+export interface IngestEvent {
+  stage: string;
+  msg?: string;
+  done?: number;
+  total?: number;
+  chunks?: string[];
+  error?: string;
+}
+
+// ingestKb: POST /api/v1/kb/ingest —— 文本入库（异步，返回 job_id）。
 export async function ingestKb(kb: string, text: string): Promise<string> {
   const res = await fetch(`${GATEWAY}/api/v1/kb/ingest`, {
     method: "POST",
     headers: { "Content-Type": "application/json" },
     body: JSON.stringify({ kb, text }),
   });
-  const data = (await res.json()) as { message?: string; error?: string };
-  if (!res.ok) throw new Error(data.error ?? `ingest failed: ${res.status}`);
-  return data.message ?? "ok";
+  const data = (await res.json()) as { job_id?: string; error?: string };
+  if (!res.ok || !data.job_id) throw new Error(data.error ?? `ingest failed: ${res.status}`);
+  return data.job_id;
 }
 
-// ingestFile: POST /api/v1/kb/ingest_file（multipart）—— 上传文件入库（docx/xlsx/pdf… → mcp-py 解析）。
+// ingestFile: POST /api/v1/kb/ingest_file（multipart）—— 文件入库（异步，返回 job_id）。
 export async function ingestFile(kb: string, file: File): Promise<string> {
   const fd = new FormData();
   fd.append("kb", kb);
   fd.append("file", file);
   const res = await fetch(`${GATEWAY}/api/v1/kb/ingest_file`, { method: "POST", body: fd });
-  const data = (await res.json()) as { message?: string; chars?: number; error?: string };
-  if (!res.ok) throw new Error(data.error ?? `ingest file failed: ${res.status}`);
-  return `${file.name}：解析 ${data.chars ?? 0} 字 → ${data.message ?? "ok"}`;
+  const data = (await res.json()) as { job_id?: string; error?: string };
+  if (!res.ok || !data.job_id) throw new Error(data.error ?? `ingest file failed: ${res.status}`);
+  return data.job_id;
+}
+
+// streamIngest: SSE 订阅入库进度（/kb/ingest/:id/stream）。返回关闭函数。
+export function streamIngest(
+  jobId: string,
+  onEvent: (ev: IngestEvent) => void,
+  onDone: () => void,
+  onError?: () => void,
+): () => void {
+  const es = new EventSource(`${GATEWAY}/api/v1/kb/ingest/${jobId}/stream`);
+  es.addEventListener("progress", (e) => onEvent(JSON.parse((e as MessageEvent).data) as IngestEvent));
+  es.addEventListener("done", () => {
+    es.close();
+    onDone();
+  });
+  es.onerror = () => {
+    es.close();
+    onError?.();
+  };
+  return () => es.close();
 }
 
 export interface KbHit {

sundynix-desktop/frontend/src/views/KbView.tsx

@@ -1,5 +1,5 @@
 import { useRef, useState } from "react";
-import { ingestKb, ingestFile, searchKb, type KbHit } from "../lib/api";
+import { ingestKb, ingestFile, streamIngest, searchKb, type IngestEvent, type KbHit } from "../lib/api";
 
 interface IngestLog {
   t: string;
@@ -7,13 +7,21 @@ interface IngestLog {
   ok: boolean;
 }
 
-// 知识库管理：入库监控（切块/embedding/Milvus）+ 检索调试台（带分数与来源）。
+interface Progress {
+  active: boolean;
+  stage: string;
+  done?: number;
+  total?: number;
+  chunks: string[];
+  error?: string;
+}
+
+// 知识库管理：实时入库监控（解析→切块→向量化→写入 + 拆分可视化）+ 检索调试台。
 export function KbView() {
   const [kb, setKb] = useState("docs");
   const [text, setText] = useState("");
   const [logs, setLogs] = useState<IngestLog[]>([]);
-  const [ingesting, setIngesting] = useState(false);
-
+  const [prog, setProg] = useState<Progress | null>(null);
   const fileRef = useRef<HTMLInputElement>(null);
 
   const [q, setQ] = useState("");
@@ -23,31 +31,54 @@ export function KbView() {
   const [err, setErr] = useState("");
 
   const stamp = () => new Date().toLocaleTimeString();
+  const ingesting = prog?.active ?? false;
+
+  // 订阅某入库 job 的进度流。
+  const follow = (job: string, label: string) => {
+    setProg({ active: true, stage: "提交", chunks: [] });
+    streamIngest(
+      job,
+      (ev: IngestEvent) =>
+        setProg((p) => ({
+          active: ev.stage !== "完成" && ev.stage !== "失败",
+          stage: ev.stage,
+          done: ev.done ?? p?.done,
+          total: ev.total ?? p?.total,
+          chunks: ev.chunks ?? p?.chunks ?? [],
+          error: ev.error,
+        })),
+      () =>
+        setProg((p) => {
+          const ok = p?.stage !== "失败";
+          setLogs((l) => [
+            { t: stamp(), msg: ok ? `${label}：${p?.total ?? 0} 块入库完成` : `${label}：${p?.error ?? "失败"}`, ok },
+            ...l,
+          ]);
+          return p ? { ...p, active: false } : null;
+        }),
+      () => setProg((p) => (p ? { ...p, active: false, stage: "连接中断" } : null)),
+    );
+  };
 
   const onIngest = async () => {
     if (!text.trim()) return;
-    setIngesting(true);
     try {
-      const msg = await ingestKb(kb, text);
-      setLogs((l) => [{ t: stamp(), msg, ok: true }, ...l]);
+      const job = await ingestKb(kb, text);
       setText("");
+      follow(job, "文本");
     } catch (e) {
       setLogs((l) => [{ t: stamp(), msg: (e as Error).message, ok: false }, ...l]);
-    } finally {
-      setIngesting(false);
     }
   };
 
   const onFile = async (file?: File) => {
     if (!file) return;
-    setIngesting(true);
     try {
-      const msg = await ingestFile(kb, file);
-      setLogs((l) => [{ t: stamp(), msg, ok: true }, ...l]);
+      const job = await ingestFile(kb, file);
+      follow(job, file.name);
     } catch (e) {
       setLogs((l) => [{ t: stamp(), msg: `${file.name}: ${(e as Error).message}`, ok: false }, ...l]);
     } finally {
-      setIngesting(false);
       if (fileRef.current) fileRef.current.value = "";
     }
   };
@@ -66,6 +97,8 @@ export function KbView() {
     }
   };
 
+  const pct = prog?.total ? Math.round(((prog.done ?? 0) / prog.total) * 100) : 0;
+
   return (
     <div className="flex h-full flex-col">
       <div className="flex items-center gap-2 border-b bg-white px-4 py-2">
@@ -77,18 +110,18 @@ export function KbView() {
           placeholder="知识库名"
           title="知识库（Milvus kb 字段分区）"
         />
-        <span className="text-[11px] text-gray-400">入库 → 切块 / embedding / Milvus；检索 → 向量召回</span>
+        <span className="text-[11px] text-gray-400">入库 → 解析 / 切块 / 向量化 / 写入；检索 → 混合召回</span>
       </div>
 
       <div className="flex min-h-0 flex-1">
-        {/* 左：入库 + 监控日志 */}
+        {/* 左：入库 + 实时监控 */}
         <section className="flex w-1/2 flex-col border-r p-4">
-          <h3 className="mb-2 text-xs font-semibold text-gray-600">入库（按行切块）</h3>
+          <h3 className="mb-2 text-xs font-semibold text-gray-600">入库</h3>
           <textarea
-            className="h-40 w-full resize-none rounded border p-2 text-sm"
+            className="h-24 w-full resize-none rounded border p-2 text-sm"
             value={text}
             onChange={(e) => setText(e.target.value)}
-            placeholder={"每行一条知识，例如：\nsundynix 用 Milvus 做向量库\nsundynix 用 NATS 做消息总线"}
+            placeholder={"每行一条知识，或上传文件"}
           />
           <div className="mt-2 flex items-center gap-2">
             <button
@@ -109,7 +142,59 @@ export function KbView() {
             />
           </div>
           <span className="mt-1 text-[10px] text-gray-400">支持 txt/md/csv/docx/xlsx/pdf（docx/xlsx/pdf 经 mcp-py 解析）</span>
-          <h3 className="mb-1 mt-4 text-xs font-semibold text-gray-600">入库监控</h3>
+
+          {/* 实时流水线进度 */}
+          {prog && (
+            <div className="mt-3 rounded border bg-gray-50 p-2">
+              <div className="flex items-center gap-2 text-xs">
+                {["解析", "切块", "向量化", "写Milvus", "写Bleve", "完成"].map((s) => {
+                  const active = prog.stage.startsWith(s) || (s === "完成" && prog.stage === "完成");
+                  const passed = stageOrder(prog.stage) > stageOrder(s);
+                  return (
+                    <span
+                      key={s}
+                      className={`rounded px-1.5 py-0.5 text-[10px] ${
+                        prog.stage === "失败"
+                          ? "bg-gray-100 text-gray-400"
+                          : active
+                            ? "bg-violet-600 text-white"
+                            : passed
+                              ? "bg-emerald-100 text-emerald-700"
+                              : "bg-gray-100 text-gray-400"
+                      }`}
+                    >
+                      {s}
+                    </span>
+                  );
+                })}
+              </div>
+              {prog.error && <p className="mt-1 text-[11px] text-rose-600">✗ {prog.error}</p>}
+              {prog.total ? (
+                <div className="mt-2">
+                  <div className="h-1.5 w-full overflow-hidden rounded bg-gray-200">
+                    <div className="h-full bg-violet-500 transition-all" style={{ width: `${pct}%` }} />
+                  </div>
+                  <div className="mt-0.5 text-[10px] text-gray-500">
+                    向量化 {prog.done ?? 0}/{prog.total} 块（{pct}%）
+                  </div>
+                </div>
+              ) : null}
+              {prog.chunks.length > 0 && (
+                <div className="mt-2">
+                  <div className="text-[10px] font-medium text-gray-500">拆分 {prog.chunks.length} 块：</div>
+                  <ul className="mt-1 max-h-24 space-y-0.5 overflow-auto">
+                    {prog.chunks.map((c, i) => (
+                      <li key={i} className="truncate rounded bg-white px-1.5 py-0.5 text-[10px] text-gray-600">
+                        <span className="text-gray-400">#{i + 1}</span> {c}
+                      </li>
+                    ))}
+                  </ul>
+                </div>
+              )}
+            </div>
+          )}
+
+          <h3 className="mb-1 mt-4 text-xs font-semibold text-gray-600">入库历史</h3>
           <ul className="flex-1 space-y-1 overflow-auto">
             {logs.length === 0 && <li className="text-xs text-gray-400">尚无入库记录。</li>}
             {logs.map((l, i) => (
@@ -122,7 +207,7 @@ export function KbView() {
 
         {/* 右：检索调试台 */}
         <section className="flex w-1/2 flex-col p-4">
-          <h3 className="mb-2 text-xs font-semibold text-gray-600">检索调试台（向量召回）</h3>
+          <h3 className="mb-2 text-xs font-semibold text-gray-600">检索调试台（混合召回 + rerank）</h3>
           <div className="flex gap-2">
             <input
               className="flex-1 rounded border px-2 py-1 text-sm"
@@ -157,7 +242,7 @@ export function KbView() {
             {hits?.map((h, i) => (
               <li key={i} className="rounded border bg-gray-50 p-2">
                 <div className="mb-1 flex items-center gap-2 text-[10px]">
-                  <span className="rounded bg-sky-100 px-1.5 py-0.5 text-sky-700">Milvus 向量</span>
+                  <span className="rounded bg-sky-100 px-1.5 py-0.5 text-sky-700">混合检索</span>
                   <span className="text-gray-400">#{i + 1}</span>
                   <span className="ml-auto font-mono text-violet-600">{h.score.toFixed(3)}</span>
                 </div>
@@ -170,3 +255,9 @@ export function KbView() {
     </div>
   );
 }
+
+const STAGES = ["解析", "解析完成", "切块", "向量化", "写Milvus", "写Bleve", "完成"];
+function stageOrder(stage: string): number {
+  const i = STAGES.findIndex((s) => stage.startsWith(s));
+  return i < 0 ? -1 : i;
+}

sundynix-gateway/internal/handler/kb.go

@@ -2,21 +2,23 @@ package handler
 
 import (
 	"context"
+	"crypto/rand"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/gin-gonic/gin"
 
 	"github.com/sundynix/sundynix-shared/contract"
 )
 
-// KbIngest: POST /api/v1/kb/ingest —— 把文本入库到知识库（→ mcp-go kb_ingest → 切块/embedding/Milvus）。
-// 供知识库管理页/脚本调用。
+// KbIngest: POST /api/v1/kb/ingest —— 文本入库（异步，返回 job_id；进度经 SSE 看）。
 func (h *Handler) KbIngest(c *gin.Context) {
 	var body struct {
 		KB   string `json:"kb"`
@@ -26,21 +28,13 @@ func (h *Handler) KbIngest(c *gin.Context) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "text required"})
 		return
 	}
-	res, err := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("kb_ingest"),
-		&contract.ToolCall{Tool: "kb_ingest", Args: map[string]any{"kb": body.KB, "text": body.Text}})
-	if err != nil {
-		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
-		return
-	}
-	if !res.OK {
-		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": res.Error})
-		return
-	}
-	c.JSON(http.StatusOK, gin.H{"status": "ok", "message": res.Content})
+	job := newJobID()
+	go h.runIngest(job, body.KB, "", nil, body.Text)
+	c.JSON(http.StatusAccepted, gin.H{"job_id": job})
 }
 
-// KbIngestFile: POST /api/v1/kb/ingest_file（multipart）—— 上传文件入库。
-// 按类型路由：文本直读；docx/xlsx/pdf/csv → mcp-py parse_document 解析为文本 → kb_ingest。
+// KbIngestFile: POST /api/v1/kb/ingest_file（multipart）—— 文件入库（异步，返回 job_id）。
+// 流水线（解析→切块→向量化→写入）的进度经 sundynix.streams.<job_id> 回流，UI 用 SSE 看。
 func (h *Handler) KbIngestFile(c *gin.Context) {
 	kb := c.PostForm("kb")
 	fh, err := c.FormFile("file")
@@ -59,22 +53,108 @@ func (h *Handler) KbIngestFile(c *gin.Context) {
 		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
 		return
 	}
-	text, err := h.parseFile(c.Request.Context(), fh.Filename, data)
-	if err != nil {
-		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "解析失败: " + err.Error()})
-		return
+	job := newJobID()
+	go h.runIngest(job, kb, fh.Filename, data, "")
+	c.JSON(http.StatusAccepted, gin.H{"job_id": job, "file": fh.Filename})
+}
+
+// runIngest 后台跑入库流水线，逐阶段把进度发到 sundynix.streams.<job>。
+// filename 非空表示文件入库（先经 mcp-py 解析）；否则用 rawText。
+func (h *Handler) runIngest(job, kb, filename string, data []byte, rawText string) {
+	ctx := context.Background()
+	emit := func(ev contract.IngestEvent) { _ = h.bus.PublishIngest(job, &ev) }
+	time.Sleep(400 * time.Millisecond) // 给 SSE 客户端订阅时间（core NATS 无缓冲）
+
+	text := rawText
+	if filename != "" {
+		emit(contract.IngestEvent{Stage: "解析", Msg: filename})
+		parsed, err := h.parseFile(ctx, filename, data)
+		if err != nil {
+			emit(contract.IngestEvent{Stage: "失败", Error: "解析失败: " + err.Error()})
+			_ = h.bus.CompleteStream(job)
+			return
+		}
+		emit(contract.IngestEvent{Stage: "解析完成", Msg: "解析出 " + itoa(len([]rune(parsed))) + " 字"})
+		text = parsed
 	}
-	res, err := h.bus.CallTool(c.Request.Context(), contract.ToolSubjectGo("kb_ingest"),
-		&contract.ToolCall{Tool: "kb_ingest", Args: map[string]any{"kb": kb, "text": text}})
+
+	// 调 mcp-go kb_ingest（带 job_id）：它会发 切块/向量化/写入/完成 事件 + CompleteStream。
+	res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("kb_ingest"),
+		&contract.ToolCall{Tool: "kb_ingest", Args: map[string]any{"kb": kb, "text": text, "job_id": job}})
+	if err != nil || res == nil || !res.OK {
+		msg := "kb_ingest 失败"
+		if err != nil {
+			msg = err.Error()
+		} else if res != nil {
+			msg = res.Error
+		}
+		emit(contract.IngestEvent{Stage: "失败", Error: msg})
+		_ = h.bus.CompleteStream(job)
+	}
+}
+
+// KbIngestStream: GET /api/v1/kb/ingest/:id/stream —— SSE 实时推送入库进度事件。
+func (h *Handler) KbIngestStream(c *gin.Context) {
+	job := c.Param("id")
+	c.Writer.Header().Set("Content-Type", "text/event-stream")
+	c.Writer.Header().Set("Cache-Control", "no-cache")
+	c.Writer.Header().Set("Connection", "keep-alive")
+
+	events := make(chan []byte, 64)
+	done := make(chan struct{})
+	unsub, err := h.bus.SubscribeTokens(job,
+		func(ev []byte) {
+			select {
+			case events <- ev:
+			default:
+			}
+		},
+		func() { close(done) },
+	)
 	if err != nil {
 		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
 		return
 	}
-	if !res.OK {
-		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": res.Error})
-		return
+	defer func() { _ = unsub() }()
+
+	c.Stream(func(w io.Writer) bool {
+		select {
+		case ev := <-events:
+			c.SSEvent("progress", string(ev))
+			return true
+		case <-done:
+			c.SSEvent("done", job)
+			return false
+		case <-c.Request.Context().Done():
+			return false
+		}
+	})
+}
+
+func newJobID() string {
+	var b [8]byte
+	_, _ = rand.Read(b[:])
+	return "ingest_" + hex.EncodeToString(b[:])
+}
+
+// itoa 简易整数转字符串（避免引入 strconv）。
+func itoa(n int) string {
+	if n == 0 {
+		return "0"
 	}
-	c.JSON(http.StatusOK, gin.H{"status": "ok", "file": fh.Filename, "chars": len([]rune(text)), "message": res.Content})
+	neg := n < 0
+	if neg {
+		n = -n
+	}
+	var b []byte
+	for n > 0 {
+		b = append([]byte{byte('0' + n%10)}, b...)
+		n /= 10
+	}
+	if neg {
+		b = append([]byte{'-'}, b...)
+	}
+	return string(b)
 }
 
 // parseFile 把文件字节转为纯文本：文本类直读，其余经 mcp-py parse_document（算法层）。

sundynix-gateway/internal/nats/publisher.go

@@ -3,6 +3,7 @@ package nats
 
 import (
 	"context"
+	"encoding/json"
 	"log"
 
 	sharedbus "github.com/sundynix/sundynix-shared/bus"
@@ -58,4 +59,16 @@ func (b *Bus) PublishConfigUpdated(kind string, cfg *contract.ModelConfig) error
 	return b.inner.PublishConfigUpdated(kind, cfg)
 }
 
+// PublishIngest 把一条入库进度事件发到 sundynix.streams.<jobID>。
+func (b *Bus) PublishIngest(jobID string, ev *contract.IngestEvent) error {
+	data, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	return b.inner.PublishToken(jobID, data)
+}
+
+// CompleteStream 发送入库流结束信号。
+func (b *Bus) CompleteStream(jobID string) error { return b.inner.CompleteStream(jobID) }
+
 func (b *Bus) Close() { b.inner.Close() }

sundynix-gateway/internal/router/router.go

@@ -24,8 +24,9 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus) *gin.Engine {
 		api.GET("/tasks/:id/stream", h.StreamTask) // 4. SSE/WS 回流 Token Stream
 		api.PUT("/memory", h.SetMemory)            // 偏好记忆登记（→ mcp-go memory_upsert）
 		api.POST("/kb/ingest", h.KbIngest)         // 知识库入库（文本，→ mcp-go kb_ingest）
-		api.POST("/kb/ingest_file", h.KbIngestFile) // 文件入库（docx/xlsx/pdf… → mcp-py 解析 → 入库）
-		api.POST("/kb/search", h.KbSearch)         // 知识库检索台（→ mcp-go kb_search）
+		api.POST("/kb/ingest_file", h.KbIngestFile)        // 文件入库（docx/xlsx/pdf… 异步）
+		api.GET("/kb/ingest/:id/stream", h.KbIngestStream) // 入库进度 SSE（实时监控）
+		api.POST("/kb/search", h.KbSearch)                 // 知识库检索台（→ mcp-go kb_search）
 		api.GET("/billing", h.Billing)
 
 		// 运维控制面：LLM 模型配置（独立运维控制台调用）。

sundynix-mcp-go/internal/mcp/gateway.go

@@ -160,14 +160,32 @@ func (g *Gateway) kbSearch(ctx context.Context, call *contract.ToolCall) *contra
 	return &contract.ToolResult{OK: true, Content: string(data)}
 }
 
-// kbIngest 把文本入库（切块→embedding→Milvus）。
+// kbIngest 把文本入库（切块→embedding→Milvus+Bleve）。
+// 带 job_id 时逐阶段把进度发到 sundynix.streams.<job_id>，供 UI 实时入库监控。
 func (g *Gateway) kbIngest(ctx context.Context, call *contract.ToolCall) *contract.ToolResult {
 	kb, _ := call.Args["kb"].(string)
 	text, _ := call.Args["text"].(string)
+	jobID, _ := call.Args["job_id"].(string)
 	if text == "" {
 		return &contract.ToolResult{OK: false, Error: "kb_ingest: text 必填"}
 	}
-	n, err := g.rag.Ingest(ctx, kb, text)
+	var onProgress func(contract.IngestEvent)
+	if jobID != "" {
+		onProgress = func(ev contract.IngestEvent) {
+			if data, err := json.Marshal(ev); err == nil {
+				_ = g.bus.PublishToken(jobID, data)
+			}
+		}
+	}
+	n, err := g.rag.Ingest(ctx, kb, text, onProgress)
+	if jobID != "" {
+		if err != nil {
+			onProgress(contract.IngestEvent{Stage: "失败", Error: err.Error()})
+		} else {
+			onProgress(contract.IngestEvent{Stage: "完成", Done: n, Total: n, Msg: fmt.Sprintf("已入库 %d 块", n)})
+		}
+		_ = g.bus.CompleteStream(jobID)
+	}
 	if err != nil {
 		return &contract.ToolResult{OK: false, Error: "kb_ingest: " + err.Error()}
 	}

sundynix-mcp-go/internal/rag/rag.go

@@ -8,8 +8,13 @@ import (
 	"log"
 	"strings"
 	"sync"
+
+	"github.com/sundynix/sundynix-shared/contract"
 )
 
+// embedBatch 是每批向量化的块数（让大文件的入库进度可观测）。
+const embedBatch = 10
+
 // Engine 聚合 embedding + Milvus(向量) + Bleve(全文) + RRF 融合 + 可选 rerank。
 // embedding 可热更新（控制面下发）。
 type Engine struct {
@@ -65,8 +70,14 @@ func Open(ctx context.Context, milvusAddr, embBase, embKey, embModel, rerankBase
 // Ready 报告 RAG 是否可用（embedding + Milvus 均就绪）。
 func (e *Engine) Ready() bool { return e.embed().ready() && e.mv != nil }
 
-// Ingest 把一段文本切块 → 向量化 → 写入 Milvus，返回块数。
-func (e *Engine) Ingest(ctx context.Context, kb, text string) (int, error) {
+// Ingest 把一段文本切块 → 分批向量化 → 写 Milvus + Bleve，返回块数。
+// onProgress 非空时逐阶段/逐批回调进度（用于实时入库监控）。
+func (e *Engine) Ingest(ctx context.Context, kb, text string, onProgress func(contract.IngestEvent)) (int, error) {
+	emit := func(ev contract.IngestEvent) {
+		if onProgress != nil {
+			onProgress(ev)
+		}
+	}
 	if !e.Ready() {
 		return 0, errors.New("rag 未配置（需 embedding + Milvus）")
 	}
@@ -74,17 +85,58 @@ func (e *Engine) Ingest(ctx context.Context, kb, text string) (int, error) {
 	if len(chunks) == 0 {
 		return 0, nil
 	}
-	vecs, err := e.embed().Embed(ctx, chunks)
-	if err != nil {
-		return 0, err
+	emit(contract.IngestEvent{Stage: "切块", Total: len(chunks), Chunks: previews(chunks), Msg: "拆为 " + itoa(len(chunks)) + " 块"})
+
+	// 分批向量化（逐批回报进度）。
+	vecs := make([][]float32, 0, len(chunks))
+	for i := 0; i < len(chunks); i += embedBatch {
+		end := min(i+embedBatch, len(chunks))
+		bv, err := e.embed().Embed(ctx, chunks[i:end])
+		if err != nil {
+			emit(contract.IngestEvent{Stage: "失败", Error: "向量化: " + err.Error()})
+			return 0, err
+		}
+		vecs = append(vecs, bv...)
+		emit(contract.IngestEvent{Stage: "向量化", Done: end, Total: len(chunks)})
 	}
+
+	emit(contract.IngestEvent{Stage: "写Milvus", Msg: "向量库写入中"})
 	if err := e.mv.insert(ctx, kb, chunks, vecs); err != nil {
+		emit(contract.IngestEvent{Stage: "失败", Error: "写Milvus: " + err.Error()})
 		return 0, err
 	}
+	emit(contract.IngestEvent{Stage: "写Bleve", Msg: "全文索引写入中"})
 	_ = e.bleve.index(kb, chunks) // 同步写全文索引（失败不阻断向量入库）
+
 	return len(chunks), nil
 }
 
+// previews 取每块的前若干字作为预览（供 UI 展示拆分情况）。
+func previews(chunks []string) []string {
+	out := make([]string, len(chunks))
+	for i, c := range chunks {
+		r := []rune(c)
+		if len(r) > 50 {
+			out[i] = string(r[:50]) + "…"
+		} else {
+			out[i] = c
+		}
+	}
+	return out
+}
+
+func itoa(n int) string {
+	if n == 0 {
+		return "0"
+	}
+	var b []byte
+	for n > 0 {
+		b = append([]byte{byte('0' + n%10)}, b...)
+		n /= 10
+	}
+	return string(b)
+}
+
 // Search 混合检索：Milvus(向量) + Bleve(全文) → RRF 融合 → 可选 rerank → topK。降级时返回空。
 func (e *Engine) Search(ctx context.Context, kb, query string, topK int) ([]Hit, error) {
 	if !e.Ready() {

sundynix-shared/contract/task.go

@@ -40,6 +40,16 @@ const (
 func ConfigGetSubject(kind string) string     { return "sundynix.config." + kind + ".get" }
 func ConfigUpdatedSubject(kind string) string { return "sundynix.config." + kind + ".updated" }
 
+// IngestEvent 是入库流水线的实时进度事件（经 sundynix.streams.<job_id> 回流给 UI）。
+type IngestEvent struct {
+	Stage  string   `json:"stage"`            // 解析/切块/向量化/写Milvus/写Bleve/完成/失败
+	Msg    string   `json:"msg,omitempty"`    // 文案
+	Done   int      `json:"done,omitempty"`   // 进度（如已向量化块数）
+	Total  int      `json:"total,omitempty"`  // 总数
+	Chunks []string `json:"chunks,omitempty"` // 切块预览（切块阶段发一次）
+	Error  string   `json:"error,omitempty"`
+}
+
 // ModelConfig 是一个模型后端的连接配置（provider 抽象，chat 与 embedding 同形）。
 // 开发期指向第三方在线 API（OpenAI 兼容）；生产期可换自部署或其它在线模型。
 type ModelConfig struct {