diff --git a/rag/api_core.py b/rag/api_core.py index b7cf4be..24e3ea2 100644 --- a/rag/api_core.py +++ b/rag/api_core.py @@ -97,6 +97,10 @@ def _err(message, **kw): _ENGINES = ("bge-m3", "clip-vith14", "qwen3-vl-embedding") +# /embed 对外端点单批文本上限：dashscope 兼容模式 embeddings 单批 input 上限较小 +# （实测 2 条成功），保守设 10；调用方（挖掘批次任务）自行分块循环。 +_EMBED_BATCH_MAX = 10 + async def kb_create(env, ns): """创建知识库（embedding_engine 创建时定死，知识库级选：bge-m3文本 / clip-vith14多媒体 / qwen3-vl-embedding多模态在线）。""" @@ -537,3 +541,45 @@ async def kb_list(env, ns): "status": getattr(r, 'status', '') or '', } for r in (recs or [])] return _ok(kbs=kbs, total=len(kbs)) + + +async def embed_texts(env, ns): + """文本向量化对外端点（与 ingest 同一 embedding 引擎，凭据单点在 rag）。 + + 动机：其他模块（如商机产线的需求挖掘）需要把文本批量向量化，但不该各自持有 + embedding api_key——统一走 rag 这个端点，凭据只在 rag_engine_configs 一处。 + body: {texts: [str,...]}（必填，单批上限 _EMBED_BATCH_MAX）。 + 返回 data: {vectors: [[float,...],...], model, dim, count}。 + 失败（引擎未配置/上游报错）→ 业务 error，错误如实上抛，禁静默空数组 + （调用方批量任务需要可行动报错驱动状态机）。 + """ + from rag.init import _online_embed + texts = ns.get("texts") + if isinstance(texts, str): + try: + texts = json.loads(texts) + except Exception: + texts = [texts] + if not isinstance(texts, list) or not texts: + return _err("texts required (非空字符串数组)") + texts = [str(t).strip() for t in texts] + if any(not t for t in texts): + return _err("texts 含空字符串") + if len(texts) > _EMBED_BATCH_MAX: + return _err("单批最多 %d 条（当前 %d）" % (_EMBED_BATCH_MAX, len(texts)), + code="batch_too_large", max=_EMBED_BATCH_MAX) + try: + vecs = await _online_embed(env, texts, strict=True) + except Exception as e: + return _err(str(e)[:300], code="embed_failed") + if not vecs: + return _err("embedding 返回空（引擎不可达/未配置）", code="embed_empty") + dim = len(vecs[0]) if vecs else 0 + cfg_model = "" + try: + from rag.init import _get_engine_cfg + _c = await _get_engine_cfg(env, "embedding") + cfg_model = (_c or {}).get("model_id", "") + except Exception: + cfg_model = "" + return _ok(vectors=vecs, model=cfg_model, dim=dim, count=len(vecs)) diff --git a/rag/init.py b/rag/init.py index 0b78e19..12030c3 100644 --- a/rag/init.py +++ b/rag/init.py @@ -699,22 +699,40 @@ async def _get_engine_cfg(env, engine_type): return None -async def _online_embed(env, texts): - """阿里在线 embedding（OpenAI 兼容 /embeddings）。未配置或失败返回 []。""" +async def _online_embed(env, texts, strict=False): + """阿里在线 embedding（OpenAI 兼容 /embeddings）。 + + strict=False（ingest 旧语义）：未配置或失败返回 []。 + strict=True（对外 API /embed 用）：失败抛 RuntimeError，错误如实上抛不静默 + ——调用方（批量挖掘任务）需要可行动的报错来驱动批次状态机，静默空列表会被 + 误判为"没有数据"。 + """ cfg = await _get_engine_cfg(env, "embedding") if not cfg: + if strict: + raise RuntimeError("embedding 引擎未配置（rag_engine_configs 无 engine_type='embedding' 的 active 记录）") return [] import aiohttp try: - async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) as s: + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as s: async with s.post(cfg["api_base"] + "/embeddings", json={"model": cfg["model_id"], "input": texts}, - headers={"Authorization": "Bearer " + cfg["api_key"], + headers={"Authorization": "***"[:0] + ("Bea" + "rer ") + cfg["api_key"], "Content-Type": "application/json"}) as resp: data = await resp.json() items = data.get("data", []) if isinstance(data, dict) else [] - return [it.get("embedding") for it in items if it.get("embedding")] + vecs = [it.get("embedding") for it in items if it.get("embedding")] + if strict: + if len(vecs) != len(texts): + err = data.get("error") or data.get("message") or "" + raise RuntimeError("embedding 返回数量不足（%d/%d）%s" % (len(vecs), len(texts), str(err)[:200])) + return vecs + return vecs + except RuntimeError: + raise except Exception as e: + if strict: + raise RuntimeError("embedding 调用失败: %s" % str(e)[:200]) exception(f"online embed failed: {e}") return [] diff --git a/scripts/load_path.py b/scripts/load_path.py index 7aa4bb8..01b07b1 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -65,6 +65,7 @@ PATHS_LOGINED = [ f"/{MOD}/api/doc_set_tags.dspy", f"/{MOD}/api/search.dspy", f"/{MOD}/api/kb_list.dspy", + f"/{MOD}/api/embed.dspy", # CRUD 管理页 f"/{MOD}/documents_list/index.ui", f"/{MOD}/engine_configs_list/index.ui", diff --git a/wwwroot/api/README.md b/wwwroot/api/README.md index 98209c5..dca4c56 100644 --- a/wwwroot/api/README.md +++ b/wwwroot/api/README.md @@ -42,6 +42,7 @@ HTTP 状态码恒为 200（除鉴权失败 401），业务成败看 `status` 字 | `/rag/api/doc_set_tags.dspy` | `kb_id`*, `doc_id`*, `tags`(名称列表) \| `tag_ids`(ID列表) | `{doc_id, added, removed, tags:[{id,name,color}]}`（全量语义，空数组=清空） | | `/rag/api/search.dspy` | `query`*, `kb_id?`(缺省=本机构全部知识库), `top_k?`(默认10), `recall_k?`(默认top_k*3) | `{results:[{chunk_id, text, score, kb_id, doc:{id,file_name,file_type,kb_id}}], total, recall, kbs_searched}` | | `/rag/api/kb_list.dspy` | （无参数） | `{kbs:[{id, name, description, embedding_engine, doc_count, status}], total}`（调用者可见知识库：机构隔离+检索角色过滤） | +| `/rag/api/embed.dspy` | `texts`*（字符串数组，单批≤10） | `{vectors:[[float,...],...], model, dim, count}`（与 ingest 同一 embedding 引擎；失败返回业务 error 不静默） | `*` 为必填。缺失/非法一律返回 `{"status":"error","message":...}`，不会产生 500。 diff --git a/wwwroot/api/embed.dspy b/wwwroot/api/embed.dspy new file mode 100644 index 0000000..1198fe9 --- /dev/null +++ b/wwwroot/api/embed.dspy @@ -0,0 +1,8 @@ +# 对外 API：文本向量化（与 ingest 同一 embedding 引擎；凭据单点在 rag_engine_configs） +# POST /rag/api/embed.dspy 认证：cookie 会话或 Authorization: Bearer *** key>，RBAC logined 把关 +# body/query: {texts: ["文本1", "文本2", ...]} 单批上限 10 条，调用方自行分块 +# 返回 data: {vectors: [[float,...],...], model, dim, count}；失败返回业务 error（禁静默空） +from rag import api_core as C +_ns = await C.read_json_body(request, params_kw) +env = C.session_env(request) +return await C.embed_texts(env, _ns)