From 03dbf0f2b58022efec0caff742edb257451dd3e7 Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 1 Sep 2026 15:47:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(rag):=20=E5=A4=9A=E5=AA=92=E4=BD=93?= =?UTF-8?q?=E7=9F=A5=E8=AF=86=E5=BA=93=E6=8E=A5=E5=9B=9E=E2=80=94=E2=80=94?= =?UTF-8?q?qwen3-vl=E5=9C=A8=E7=BA=BF(2560=E7=BB=B4)+CLIP=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E7=BB=99GPU;VDB=E5=9C=B0=E5=9D=80=E8=AF=BBupapp;?= =?UTF-8?q?=E5=9C=A8=E7=BA=BF=E9=87=8D=E6=8E=92;=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E7=95=8C=E9=9D=A24=E5=BC=95=E6=93=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rag/init.py | 38 ++++- rag/vl_online.py | 153 ++++++++++++++++++ wwwroot/engine_configs_list/engine_form.dspy | 8 +- wwwroot/knowledge_bases_list/create_kb.dspy | 6 +- .../embedding_options.dspy | 5 +- .../knowledge_bases_list/search_result.dspy | 65 ++++++-- wwwroot/knowledge_bases_list/upload_file.dspy | 104 +++++++++--- 7 files changed, 334 insertions(+), 45 deletions(-) create mode 100644 rag/vl_online.py diff --git a/rag/init.py b/rag/init.py index 2e132d9..977e245 100644 --- a/rag/init.py +++ b/rag/init.py @@ -41,10 +41,11 @@ async def kb_list_handler(request, params_kw, *args, **kwargs): size_str = f"{total_size/1024:.0f}KB" else: size_str = f"{total_size}B" - engine = {'bge-m3': '文本 bge-m3', 'clip-vith14': '多媒体 CLIP', 'CLIP ViT-H-14': '多媒体 CLIP'}.get((r.embedding_engine or '').strip(), (r.embedding_engine or 'CLIP')) + engine = {'bge-m3': '文本·在线', 'qwen3-vl-embedding': '多媒体·在线', 'clip-vith14': '多媒体·GPU CLIP', 'CLIP ViT-H-14': '多媒体·GPU CLIP'}.get((r.embedding_engine or '').strip(), (r.embedding_engine or '未配置')) + dim = 2560 if (r.embedding_engine or '').strip() == 'qwen3-vl-embedding' else 1024 card = {"widgettype":"VBox","options":{"cwidth":16,"cheight":12,"bgcolor":"#f0f7ff","padding":"16px","css":"card clickable","border":"1px solid #d0e4f7"},"subwidgets":[ {"widgettype":"Text","options":{"text":"📚 " + str(r.name),"cfontsize":16,"fontWeight":"bold"}}, - {"widgettype":"Text","options":{"text":str(engine)+" · 1024维","cfontsize":12,"color":"#888"}}, + {"widgettype":"Text","options":{"text":str(engine)+f" · {dim}维","cfontsize":12,"color":"#888"}}, {"widgettype":"HBox","options":{"spacing":"12px"},"subwidgets":[ {"widgettype":"Text","options":{"text":"📄 "+str(doc_count)+"文档","cfontsize":12,"color":"#666"}}, {"widgettype":"Text","options":{"text":"💾 "+size_str,"cfontsize":12,"color":"#666"}}]}], @@ -992,7 +993,7 @@ async def engine_cfg_get_handler(request, params_kw, *args, **kwargs): async with get_sor_context(env, 'rag') as sor: recs = await sor.sqlExe( "SELECT engine_type, model_name, endpoint_url, api_key, status FROM rag_engine_configs " - "WHERE engine_type IN ('embedding','rerank') ORDER BY engine_type", {}) + "WHERE engine_type IN ('embedding','rerank','mm_embedding','mm_rerank') ORDER BY engine_type", {}) rows = [] for r in recs: enc = getattr(r, "api_key", "") or "" @@ -1010,8 +1011,8 @@ async def engine_cfg_save_handler(request, params_kw, *args, **kwargs): env = request._run_ns try: engine_type = (params_kw.get("engine_type") or "").strip() - if engine_type not in ("embedding", "rerank"): - return json.dumps({"error": "engine_type must be embedding/rerank"}) + if engine_type not in ("embedding", "rerank", "mm_embedding", "mm_rerank"): + return json.dumps({"error": "engine_type must be embedding/rerank/mm_embedding/mm_rerank"}) model_id = (params_kw.get("model_id") or "").strip() api_base = (params_kw.get("api_base") or "").strip() api_key_plain = (params_kw.get("api_key") or "").strip() @@ -1071,6 +1072,33 @@ async def engine_cfg_test_handler(request, params_kw, *args, **kwargs): import aiohttp headers = {"Authorization": "Bearer " + api_key_plain, "Content-Type": "application/json"} base = api_base.rstrip("/") + # 在线多模态：DashScope 原生 API + if engine_type == "mm_embedding": + if "/api/v1" not in base: + base = base + "/api/v1" + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as s: + async with s.post(base + "/services/embeddings/multimodal-embedding/multimodal-embedding", + json={"model": model_id, "input": {"contents": [{"text": "连通性测试"}]}}, + headers=headers) as resp: + data = await resp.json() + embs = (data.get("output") or {}).get("embeddings") or [] + dim = len(embs[0]["embedding"]) if embs and embs[0].get("embedding") else 0 + if not dim: + return json.dumps({"error": str(data)[:200]}, ensure_ascii=False) + return json.dumps({"status": "SUCCEEDED", "message": f"mm_embedding OK，维度 {dim}"}, ensure_ascii=False) + if engine_type == "mm_rerank": + if "/api/v1" not in base: + base = base + "/api/v1" + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as s: + async with s.post(base + "/services/rerank/text-rerank/text-rerank", + json={"model": model_id, + "input": {"query": "测试", "documents": ["甲", "乙"]}, + "parameters": {"top_n": 2}}, + headers=headers) as resp: + data = await resp.json() + if not (data.get("output") or {}).get("results"): + return json.dumps({"error": str(data)[:200]}, ensure_ascii=False) + return json.dumps({"status": "SUCCEEDED", "message": "mm_rerank OK"}, ensure_ascii=False) async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) as s: if engine_type == "embedding": async with s.post(base + "/embeddings", json={"model": model_id, "input": ["连通性测试"]}, diff --git a/rag/vl_online.py b/rag/vl_online.py new file mode 100644 index 0000000..1b95384 --- /dev/null +++ b/rag/vl_online.py @@ -0,0 +1,153 @@ +# -*- coding:utf-8 -*- +"""在线多模态引擎：qwen3-vl-embedding / qwen3-vl-rerank（阿里 DashScope 原生 API）。 +与 CLIP（GPU 本地）并存：kb.embedding_engine=qwen3-vl-embedding 走在线， +clip-vith14 仍走 GPU /mme 端点（有 GPU 的环境保留本地部署）。 +配置来自 rag_engine_configs：engine_type=mm_embedding / mm_rerank，空=能力屏蔽。 +""" +import json +from traceback import format_exc +from ahserver.serverenv import ServerEnv +from appPublic.log import exception +from sqlor.dbpools import get_sor_context + +NATIVE_EMB_PATH = "/api/v1/services/embeddings/multimodal-embedding/multimodal-embedding" +NATIVE_RR_PATH = "/api/v1/services/rerank/text-rerank/text-rerank" +DEFAULT_MM_DIM = 2560 +TEXT_DIM = 1024 + + +async def get_mm_cfg(engine_type="mm_embedding"): + """读引擎配置，返回 {model_id, api_base, api_key 明文, dim}；未配置/空 → None（能力屏蔽）。""" + from appPublic.rc4 import unpassword + env = ServerEnv() + try: + async with get_sor_context(env, 'rag') as sor: + recs = await sor.sqlExe( + "SELECT model_name, endpoint_url, api_key, config_json FROM rag_engine_configs " + "WHERE engine_type=${t}$ AND status='active' ORDER BY is_default DESC, priority DESC LIMIT 1", + {"t": engine_type}) + if not recs: + return None + r = recs[0] + model_id = (getattr(r, "model_name", "") or "").strip() + api_base = (getattr(r, "endpoint_url", "") or "").strip() + enc = (getattr(r, "api_key", "") or "").strip() + if not (model_id and api_base and enc): + return None + try: + from appPublic.jsonConfig import getConfig + key = getConfig().password_key or 'QRIVSRHrthhwyjy176556332' + api_key = unpassword(enc, key) + except Exception: + api_key = enc + dim = DEFAULT_MM_DIM + try: + cj = json.loads(getattr(r, "config_json", "") or "{}") + dim = int(cj.get("dim", DEFAULT_MM_DIM)) + except Exception: + pass + return {"model_id": model_id, "api_base": api_base.rstrip("/"), "api_key": api_key, "dim": dim} + except Exception as e: + exception(f"mm cfg read failed ({engine_type}): {e}") + return None + + +async def _post_json(url, payload, headers, timeout=60): + import aiohttp + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as s: + async with s.post(url, json=payload, headers=headers) as resp: + return await resp.json() + + +async def vl_embed_contents(contents, engine_type="mm_embedding"): + """单条多模态向量化。contents 例：[{"text": "..."}] 或 [{"image": b64dataurl}] 或图文混合。""" + cfg = await get_mm_cfg(engine_type) + if not cfg: + return None + url = cfg["api_base"] + NATIVE_EMB_PATH + headers = {"Authorization": "Bearer " + cfg["api_key"], "Content-Type": "application/json"} + try: + data = await _post_json(url, {"model": cfg["model_id"], "input": {"contents": contents}}, headers) + embs = (data.get("output") or {}).get("embeddings") or [] + return embs[0]["embedding"] if embs and embs[0].get("embedding") else None + except Exception as e: + exception(f"vl embed failed: {e}, {format_exc()}") + return None + + +async def vl_embed_texts(texts, engine_type="mm_embedding"): + """批量纯文本向量化（走 vl 原生 API，contents=[{"text":t}]）。""" + cfg = await get_mm_cfg(engine_type) + if not cfg: + return [] + url = cfg["api_base"] + NATIVE_EMB_PATH + headers = {"Authorization": "Bearer " + cfg["api_key"], "Content-Type": "application/json"} + out = [] + for t in texts: + try: + data = await _post_json(url, {"model": cfg["model_id"], "input": {"contents": [{"text": t}]}}, headers) + embs = (data.get("output") or {}).get("embeddings") or [] + out.append(embs[0]["embedding"] if embs and embs[0].get("embedding") else None) + except Exception as e: + exception(f"vl embed text failed: {e}") + out.append(None) + return out + + +async def vl_embed_image_bytes(image_bytes, engine_type="mm_embedding"): + """图片 bytes → base64 data url → 在线向量化。""" + import base64 + b64 = base64.b64encode(image_bytes).decode() + return await vl_embed_contents([{"image": "data:image/jpeg;base64," + b64}], engine_type) + + +async def vl_rerank(query, documents): + """在线多模态重排。query: 文本 str；documents: list[str] 或 [{\"image\": ...}]。 + 未配置/失败返回 None（调用方回退召回分排序）。""" + cfg = await get_mm_cfg("mm_rerank") + if not cfg: + return None + base = cfg["api_base"] + if "compatible" in base: + base = base.replace("/compatible-mode", "").replace("/compatible-api", "") + if "/api/v1" not in base: + base = base + "/api/v1" + url = base + NATIVE_RR_PATH.replace("/api/v1", "") + headers = {"Authorization": "Bearer " + cfg["api_key"], "Content-Type": "application/json"} + payload = {"model": cfg["model_id"], + "input": {"query": query, "documents": documents}, + "parameters": {"top_n": len(documents), "return_documents": False}} + try: + data = await _post_json(url, payload, headers, timeout=30) + results = (data.get("output") or {}).get("results") or [] + if not results: + return None + scores = [0.0] * len(documents) + for it in results: + idx = it.get("index") + if isinstance(idx, int) and 0 <= idx < len(scores): + scores[idx] = it.get("relevance_score", 0) + return {"scores": scores} + except Exception as e: + exception(f"vl rerank failed: {e}") + return None + + +async def vdb_baseurl(): + """VDB 服务地址：读 upapp.rag-vdb（生产已切内网 9186），不再硬编码域名。""" + env = ServerEnv() + async with get_sor_context(env, 'rag') as sor: + recs = await sor.sqlExe("SELECT baseurl FROM upapp WHERE id='rag-vdb'", {}) + if not recs: + raise Exception("upapp rag-vdb not configured") + return (recs[0].baseurl or "").rstrip("/") + + +async def vdb_call(apiname, payload, timeout=20): + """直调 VDB HTTP 接口（createcollection/upsert），URL 来自 upapp。""" + import aiohttp + base = await vdb_baseurl() + url = f"{base}/v1/{apiname}" + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as s: + async with s.post(url, json=payload, headers={"Content-Type": "application/json"}) as resp: + return await resp.json() diff --git a/wwwroot/engine_configs_list/engine_form.dspy b/wwwroot/engine_configs_list/engine_form.dspy index aab4ce8..3954784 100644 --- a/wwwroot/engine_configs_list/engine_form.dspy +++ b/wwwroot/engine_configs_list/engine_form.dspy @@ -14,6 +14,8 @@ async def get_cfg(t): emb = await get_cfg("embedding") rr = await get_cfg("rerank") +mm_emb = await get_cfg("mm_embedding") +mm_rr = await get_cfg("mm_rerank") def card(title, desc, prefix, cfg): shielded = (not cfg["model_id"]) or cfg["status"] != "active" @@ -56,7 +58,9 @@ def card(title, desc, prefix, cfg): } result = {"widgettype": "VBox", "options": {"width": "100%", "spacing": "16px"}, "subwidgets": [ - card("🔤 文本向量化（Embedding）", "文本知识库必需。模型留空或停用 = 屏蔽（文档仍入库可浏览，检索不可用）。", "embedding", emb), - card("🎯 重排序（Rerank）", "检索结果精排，可选。未配置时按召回分排序。", "rerank", rr) + card("🔤 文本向量化（Embedding）", "文本知识库(bge-m3引擎)必需。模型留空或停用 = 屏蔽（文档仍入库可浏览，检索不可用）。", "embedding", emb), + card("🎯 文本重排（Rerank）", "文本检索结果精排，可选。未配置时按召回分排序。", "rerank", rr), + card("🖼️ 多模态向量化（qwen3-vl）", "多媒体知识库必需（图文跨模态）。支持以文搜图/以图搜文。留空=屏蔽多媒体检索。", "mm_embedding", mm_emb), + card("🖼️ 多模态重排（qwen3-vl-rerank）", "多媒体/混合检索精排，可选。未配置时按召回分排序。", "mm_rerank", mm_rr) ]} return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/knowledge_bases_list/create_kb.dspy b/wwwroot/knowledge_bases_list/create_kb.dspy index ebf4ca9..1444eb4 100644 --- a/wwwroot/knowledge_bases_list/create_kb.dspy +++ b/wwwroot/knowledge_bases_list/create_kb.dspy @@ -7,10 +7,10 @@ async with db.sqlorContext(dbname) as sor: kb_id = uuid() name = ns.get("name", "") desc = ns.get("description", "") - # 向量引擎：clip-vith14=多媒体(CLIP)，bge-m3=文本(bge-m3)。默认 clip-vith14 兼容旧数据 + # 向量引擎：bge-m3=文本(在线) / qwen3-vl-embedding=多媒体(在线) / clip-vith14=多媒体(GPU本地CLIP) emb_type = ns.get("embedding_type", "") or ns.get("embedding_engine", "") - if emb_type not in ("bge-m3", "clip-vith14"): - emb_type = "clip-vith14" + if emb_type not in ("bge-m3", "clip-vith14", "qwen3-vl-embedding"): + emb_type = "bge-m3" await sor.sqlExe( "INSERT INTO rag_knowledge_bases (id, name, description, org_id, embedding_engine, vdb_collection, doc_count, total_size, chunk_count, status, created_at) " "VALUES (${id}$, ${name}$, ${desc}$, ${org_id}$, ${emb}$, 'rag_collection', 0, 0, 0, 'active', NOW())", diff --git a/wwwroot/knowledge_bases_list/embedding_options.dspy b/wwwroot/knowledge_bases_list/embedding_options.dspy index 9a4bc33..8b9821f 100644 --- a/wwwroot/knowledge_bases_list/embedding_options.dspy +++ b/wwwroot/knowledge_bases_list/embedding_options.dspy @@ -1,5 +1,6 @@ import json return json.dumps([ - {"value": "clip-vith14", "text": "多媒体（图文音视）"}, - {"value": "bge-m3", "text": "文本（文档检索）"} + {"value": "bge-m3", "text": "文本（在线·文档检索）"}, + {"value": "qwen3-vl-embedding", "text": "多媒体（在线·图文跨模态）"}, + {"value": "clip-vith14", "text": "多媒体（GPU本地CLIP·需GPU）"} ], ensure_ascii=False) diff --git a/wwwroot/knowledge_bases_list/search_result.dspy b/wwwroot/knowledge_bases_list/search_result.dspy index 97ba739..4f77b94 100644 --- a/wwwroot/knowledge_bases_list/search_result.dspy +++ b/wwwroot/knowledge_bases_list/search_result.dspy @@ -75,7 +75,7 @@ hits = [] raw_rows = [] kw_rows = [] -# 读知识库向量引擎，决定用文本(bge-m3 /txte)还是多媒体(CLIP /mme) embedding +# 读知识库向量引擎：bge-m3(在线文本) / qwen3-vl-embedding(在线多模态) / clip-vith14(GPU本地) emb_engine = 'clip-vith14' try: async with get_sor_context(env, 'rag') as sor: @@ -83,6 +83,7 @@ try: if krecs: emb_engine = (getattr(krecs[0], 'embedding_engine', '') or 'clip-vith14').strip() except: pass +is_vl = emb_engine == 'qwen3-vl-embedding' if emb_engine == 'bge-m3': emb_url = 'https://embedding.opencomputing.net:10443/txte/api/embed' emb_model = 'bge-m3' @@ -90,25 +91,47 @@ else: emb_url = 'https://embedding.opencomputing.net:10443/mme/api/embed' emb_model = 'CLIP-ViT-H-14' +# VDB 服务地址：读 upapp.rag-vdb（生产已切内网），不硬编码 +try: + async with get_sor_context(env, 'rag') as sor: + _u = await sor.sqlExe("SELECT baseurl FROM upapp WHERE id='rag-vdb'", {}) + VDB_BASE = (_u[0].baseurl or '').rstrip('/') if _u else 'https://vectordb.opencomputing.net:10443' +except: + VDB_BASE = 'https://vectordb.opencomputing.net:10443' + if query: - # 1. Embed + # 1. Embed（在线：vl原生 / bge-m3兼容；GPU：CLIP /mme） vec = [] - try: - client = StreamHttpClient() - resp = await client.request('POST', emb_url, - json={"texts": [query], "model": emb_model}) - emb = json.loads(resp) - vec = emb.get("text_embeddings", emb.get("embeddings", [[]]))[0] - except: - pass + if is_vl: + try: + from rag.vl_online import vl_embed_contents + vec = await vl_embed_contents([{"text": query}]) or [] + except: + vec = [] + elif emb_engine == 'bge-m3': + try: + from rag.init import _online_embed + _vs = await _online_embed(env, [query]) + vec = _vs[0] if _vs else [] + except: + vec = [] + else: + try: + client = StreamHttpClient() + resp = await client.request('POST', emb_url, + json={"texts": [query], "model": emb_model}) + emb = json.loads(resp) + vec = emb.get("text_embeddings", emb.get("embeddings", [[]]))[0] + except: + pass if not vec: - return json.dumps({"widgettype": "Text", "options": {"text": "向量化失败", "cfontsize": 14, "color": "#e74c3c"}}, ensure_ascii=False) + return json.dumps({"widgettype": "Text", "options": {"text": "向量化失败（引擎未配置或不可用，请到引擎配置页检查）", "cfontsize": 14, "color": "#e74c3c"}}, ensure_ascii=False) # 2. VDB search try: recall_n = top_k * 3 client2 = StreamHttpClient() - resp2 = await client2.request('POST', 'https://vectordb.opencomputing.net:10443/v1/query', + resp2 = await client2.request('POST', VDB_BASE + '/v1/query', json={"colname": kb_id, "vector": vec, "pagerows": recall_n, "output_fields": ["*"]}) vdb = json.loads(resp2) raw_rows = vdb.get("data", {}).get("rows", []) @@ -172,7 +195,23 @@ if query: hits.append(kr) seen.add(kr["id"]) - hits.sort(key=lambda x: x.get("score", 0), reverse=True) + # 在线重排（mm_rerank；未配置/失败则保持召回分排序） + rr_applied = False + try: + from rag.vl_online import vl_rerank + cand = hits[:top_k * 3] + if cand: + rr = await vl_rerank(query, [h.get("text", "") for h in cand]) + if rr and isinstance(rr.get("scores"), list) and len(rr["scores"]) == len(cand): + for i, s in enumerate(rr["scores"]): + cand[i]["score"] = s + cand.sort(key=lambda x: x.get("score", 0), reverse=True) + rr_applied = True + except Exception as e: + info('[search_result] rerank failed: %s' % e) + + if not rr_applied: + hits.sort(key=lambda x: x.get("score", 0), reverse=True) hits = hits[:top_k] else: diff --git a/wwwroot/knowledge_bases_list/upload_file.dspy b/wwwroot/knowledge_bases_list/upload_file.dspy index 4cab01a..a5c0d3e 100644 --- a/wwwroot/knowledge_bases_list/upload_file.dspy +++ b/wwwroot/knowledge_bases_list/upload_file.dspy @@ -51,7 +51,10 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): # 后台任务无 request/env，用注入的全局函数解析；禁硬编码库名 _dbname = get_module_dbname('rag') - # 读知识库向量引擎：bge-m3=文本(走 /txte)，clip-vith14=多媒体(走 /mme) + # 读知识库向量引擎： + # bge-m3 → 在线文本(阿里, 1024维) + # qwen3-vl-embedding → 在线多模态(阿里, 2560维) + # clip-vith14 → GPU 本地 CLIP(1024维, 有GPU环境保留) emb_engine = 'clip-vith14' try: async with db.sqlorContext(_dbname) as sor: @@ -60,6 +63,17 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): emb_engine = (getattr(krecs[0], 'embedding_engine', '') or 'clip-vith14').strip() except: pass is_text = emb_engine == 'bge-m3' + is_vl = emb_engine == 'qwen3-vl-embedding' + vdb_dim = 2560 if is_vl else 1024 + + # VDB 服务地址：读 upapp.rag-vdb（生产已切内网），不硬编码 + try: + async with db.sqlorContext(_dbname) as sor: + _u = await sor.sqlExe("SELECT baseurl FROM upapp WHERE id='rag-vdb'", {}) + VDB_BASE = (_u[0].baseurl or '').rstrip('/') if _u else 'https://vectordb.opencomputing.net:10443' + except: + VDB_BASE = 'https://vectordb.opencomputing.net:10443' + if is_text: emb_url = 'https://embedding.opencomputing.net:10443/txte/api/embed' emb_model = 'bge-m3' @@ -67,9 +81,49 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): emb_url = 'https://embedding.opencomputing.net:10443/mme/api/embed' emb_model = 'CLIP-ViT-H-14' + # 在线引擎：配置读取 + key 解密（mm_embedding / mm_rerank） + _vl_cfg = None + if is_vl or not is_text: + try: + from rag.vl_online import get_mm_cfg + _vl_cfg = await get_mm_cfg('mm_embedding') + except: + _vl_cfg = None + + async def _online_text_embed(texts): + """文本向量化：vl 引擎走在线原生API；bge-m3 走在线兼容API（经 engine_configs）""" + if is_vl and _vl_cfg: + from rag.vl_online import vl_embed_texts + return await vl_embed_texts(texts) + try: + from rag.init import _online_embed + from ahserver.serverenv import ServerEnv + return await _online_embed(ServerEnv(), texts) + except: + return [] + + async def _media_embed_image(img_b64): + """图片向量化：vl 在线(base64) 或 GPU CLIP""" + if is_vl and _vl_cfg: + try: + from rag.vl_online import vl_embed_contents + v = await vl_embed_contents([{"image": "data:image/jpeg;base64," + img_b64}]) + return v + except: + return None + try: + client = StreamHttpClient() + resp = await client.request('POST', emb_url, + json={"images": [img_b64], "model": emb_model}) + emb_data = json.loads(resp) + es = emb_data.get("image_embeddings", emb_data.get("embeddings", [])) + return es[0] if es else None + except: + return None + async def ensure_vdb_collection(client, kb_id): - payload = {"colname": kb_id, "fields": [{"name": "id", "type": "str", "is_primary": True, "max_length": 64}, {"name": "vector", "type": "fvector", "dim": 1024}, {"name": "text", "type": "str", "max_length": 65535}], "description": "RAG kb", "metric": "COSINE"} - await client.request('POST', 'https://vectordb.opencomputing.net:10443/v1/createcollection', json=payload) + payload = {"colname": kb_id, "fields": [{"name": "id", "type": "str", "is_primary": True, "max_length": 64}, {"name": "vector", "type": "fvector", "dim": vdb_dim}, {"name": "text", "type": "str", "max_length": 65535}], "description": "RAG kb", "metric": "COSINE"} + await client.request('POST', VDB_BASE + '/v1/createcollection', json=payload) text = '' chunks_n = 0 @@ -126,7 +180,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): except: pass return - # --- IMAGE: face detection --- + # --- IMAGE: face detection + 图片向量化入库 --- if ext_l in image_exts: img_b64 = base64.b64encode(file_data).decode() try: @@ -141,6 +195,24 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): meta_parts['face_bboxes'] = [f.get("bbox", {}) for f in faces[:10]] meta_parts['face'] = face_count except: pass + # 图片本身向量化（vl在线 或 GPU CLIP），存入 VDB 供以文搜图 + try: + img_vec = await _media_embed_image(img_b64) + if img_vec: + client2 = StreamHttpClient() + await ensure_vdb_collection(client2, kb_id) + vdb_data = {"colname": kb_id, "data": [ + {"id": doc_id + "_c0", "vector": img_vec, "text": file_name}]} + await client2.request('POST', VDB_BASE + '/v1/upsert', json=vdb_data) + async with db.sqlorContext(_dbname) as sor: + await sor.sqlExe( + "INSERT INTO rag_document_chunks (id, doc_id, kb_id, chunk_index, content, vector_id, created_at) " + "VALUES (${id}$, ${doc_id}$, ${kb_id}$, 0, ${content}$, ${vid}$, NOW())", + {"id": doc_id + "_c0", "doc_id": doc_id, "kb_id": kb_id, + "content": file_name, "vid": doc_id + "_c0"}) + chunks_n = 1 + meta_parts['image'] = 'embedded' + except: pass # --- AUDIO: voiceprint --- if ext_l in audio_exts: @@ -176,13 +248,10 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): face_count = len(faces) frame_bboxes = [f.get("bbox", {}) for f in faces[:10]] if faces else [] except: pass - # --- CLIP image embedding for video frame --- + # --- 视频帧向量化（vl在线 或 GPU CLIP） --- try: - client2 = StreamHttpClient() - resp2 = await client2.request('POST', emb_url, - json={"images": [img_b64], "model": emb_model}) - emb_data = json.loads(resp2) - img_embeddings = emb_data.get("image_embeddings", emb_data.get("embeddings", [])) + frame_vec = await _media_embed_image(img_b64) + img_embeddings = [frame_vec] if frame_vec else [] except: img_embeddings = [] if img_embeddings: @@ -192,7 +261,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): vdb_data = {"colname": kb_id, "data": [ {"id": doc_id + "_c0", "vector": img_embeddings[0], "text": file_name} ]} - await client3.request('POST', 'https://vectordb.opencomputing.net:10443/v1/upsert', json=vdb_data) + await client3.request('POST', VDB_BASE + '/v1/upsert', json=vdb_data) chunk_meta = {"start_time": 0} if frame_bboxes: chunk_meta["bboxes"] = frame_bboxes @@ -253,14 +322,9 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): if cur: chunks.append(cur) if chunks: - try: - client = StreamHttpClient() - resp = await client.request('POST', emb_url, - json={"texts": chunks, "model": emb_model}) - emb_data = json.loads(resp) - embeddings = emb_data.get("text_embeddings", emb_data.get("embeddings", [])) - except: - embeddings = [] + # 在线文本向量化（vl引擎→qwen3-vl原生；bge-m3→在线兼容；GPU CLIP 时代的老路径已移除） + embeddings = await _online_text_embed(chunks) + embeddings = [e for e in embeddings if e] or [] vector_ids = [] if embeddings: @@ -270,7 +334,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): vdb_data = {"colname": kb_id, "data": [ {"id": doc_id + "_c" + str(i), "vector": emb, "text": chunks[i]} for i, emb in enumerate(embeddings)]} - resp3 = await client2.request('POST', 'https://vectordb.opencomputing.net:10443/v1/upsert', json=vdb_data) + resp3 = await client2.request('POST', VDB_BASE + '/v1/upsert', json=vdb_data) if json.loads(resp3).get('status') == 'SUCCEEDED': vector_ids = [doc_id + "_c" + str(i) for i in range(len(embeddings))] except: