diff --git a/pipeline_service/rag_client.py b/pipeline_service/rag_client.py
index b74d4ac..fbd45d9 100644
--- a/pipeline_service/rag_client.py
+++ b/pipeline_service/rag_client.py
@@ -222,6 +222,30 @@ async def rag_kb_list(project_id):
     return await _rag_call(project_id, "kb_list.dspy", payload={})
 
 
+async def rag_embed_texts(project_id, texts, batch_size=10):
+    """批量文本向量化（rag 对外 API：/rag/api/embed.dspy，凭据单点在 rag）。
+
+    以项目 owner 身份调用；内部按 batch_size 分块循环，返回
+    (vectors 列表, '') 或 (None, 错误)。任一分块失败立即整体失败并带块号——
+    调用方（挖掘批次）需要可行动的报错，禁止部分成功静默丢数据。
+    """
+    if not texts:
+        return None, "texts 为空"
+    all_vecs = []
+    bs = max(1, min(int(batch_size or 10), 10))
+    for i in range(0, len(texts), bs):
+        chunk = texts[i:i + bs]
+        data, err = await _rag_call(project_id, "embed.dspy", payload={"texts": chunk})
+        if err:
+            return None, "embedding 失败@块%d-%d: %s" % (i, i + len(chunk) - 1, err)
+        vecs = (data or {}).get("vectors") if isinstance(data, dict) else None
+        if not vecs or len(vecs) != len(chunk):
+            return None, "embedding 返回数量不符@块%d-%d（%s/%d）" % (
+                i, i + len(chunk) - 1, len(vecs or []), len(chunk))
+        all_vecs.extend(vecs)
+    return all_vecs, ""
+
+
 async def rag_doc_upload(project_id, kb_id, file_path, file_name=""):
     """以 owner 身份上传文件入库（API 模式：原始字节 + query 参数）。"""
     import os