diff --git a/rag/pipeline.py b/rag/pipeline.py new file mode 100644 index 0000000..48cb8e8 --- /dev/null +++ b/rag/pipeline.py @@ -0,0 +1,208 @@ +""" +RAG 文件处理管线 — 文档解析 / 人脸检测 / 声纹处理 +GPU 服务映射: + 人脸: https://media.opencomputing.net:10443/face/api/detect + 声纹: https://media.opencomputing.net:10443/voiceprint/extract/submit +""" +import json, os, uuid, base64 + +# File type classification +TEXT_EXTS = {'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm', '.py', '.js', '.css', '.yaml', '.yml', '.log', '.rst'} +DOC_EXTS = {'.pdf', '.docx', '.pptx', '.xlsx', '.doc', '.ppt', '.xls'} +IMAGE_EXTS = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'} +AUDIO_EXTS = {'.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac'} +VIDEO_EXTS = {'.mp4', '.avi', '.mov', '.mkv', '.webm'} + +# GPU service endpoints +FACE_DETECT_URL = "https://media.opencomputing.net:10443/face/api/detect" +FACE_RECOGNIZE_URL = "https://media.opencomputing.net:10443/face/api/recognize" +VOICEPRINT_EXTRACT_URL = "https://media.opencomputing.net:10443/voiceprint/extract/submit" +VOICEPRINT_STATUS_URL = "https://media.opencomputing.net:10443/voiceprint/extract/status" + + +def classify_file(file_name): + ext = os.path.splitext(file_name)[1].lower() + if ext in TEXT_EXTS: + return 'text' + if ext in DOC_EXTS: + return 'document' + if ext in IMAGE_EXTS: + return 'image' + if ext in AUDIO_EXTS: + return 'audio' + if ext in VIDEO_EXTS: + return 'video' + return 'other' + + +async def extract_text(file_data, file_name): + ext = os.path.splitext(file_name)[1].lower() + text = "" + + if ext in TEXT_EXTS: + text = file_data.decode('utf-8', errors='replace') + + elif ext == '.pdf': + try: + import io + from PyPDF2 import PdfReader + reader = PdfReader(io.BytesIO(file_data)) + text = '\n'.join(p.extract_text() or '' for p in reader.pages) + except Exception as e: + text = "[PDF error: " + str(e) + "]" + + elif ext == '.docx': + try: + import io + from docx import Document + doc = Document(io.BytesIO(file_data)) + text = '\n'.join(p.text for p in doc.paragraphs if p.text.strip()) + except Exception as e: + text = "[DOCX error: " + str(e) + "]" + + elif ext == '.pptx': + try: + import io + from pptx import Presentation + prs = Presentation(io.BytesIO(file_data)) + slides = [] + for slide in prs.slides: + parts = [s.text for s in slide.shapes if hasattr(s, 'text') and s.text.strip()] + slides.append(' '.join(parts)) + text = '\n'.join(slides) + except Exception as e: + text = "[PPTX error: " + str(e) + "]" + + elif ext in {'.xlsx', '.xls'}: + try: + import io + from openpyxl import load_workbook + wb = load_workbook(io.BytesIO(file_data), read_only=True, data_only=True) + rows = [] + for ws in wb.worksheets: + for row in ws.iter_rows(values_only=True): + r = ' | '.join(str(c) if c is not None else '' for c in row) + if r.strip(): + rows.append(r) + text = '\n'.join(rows) + except Exception as e: + text = "[XLSX error: " + str(e) + "]" + + return text.strip() + + +async def ingest_to_rag(env, text, kb_id, doc_id): + try: + from pipeline import ingest as pipeline_ingest + result = pipeline_ingest( + text, pipeline_name="kg-rag-standard", + collection=kb_id, graph_name=kb_id, llm_func=None) + return result + except Exception as e: + return {"error": str(e), "chunks": 0} + + +async def detect_faces(file_data, file_name): + """Detect faces via GPU face-service (InsightFace buffalo_l)""" + import aiohttp + try: + b64 = base64.b64encode(file_data).decode() + payload = {"image": b64} + async with aiohttp.ClientSession() as session: + async with session.post( + FACE_DETECT_URL, + json=payload, + timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + if resp.status == 200: + result = await resp.json() + return { + "faces": result.get("faces", result.get("count", 0)), + "details": result + } + return {"error": "face service returned " + str(resp.status), "faces": 0} + except Exception as e: + return {"error": str(e), "faces": 0} + + +async def extract_voiceprint(file_data, file_name): + import aiohttp + try: + async with aiohttp.ClientSession() as session: + form = aiohttp.FormData() + form.add_field('file', file_data, filename=file_name, content_type='audio/wav') + async with session.post(VOICEPRINT_EXTRACT_URL, data=form, timeout=aiohttp.ClientTimeout(total=60)) as resp: + if resp.status == 200: + r = await resp.json() + if r.get('status') == 'SUCCEEDED': + return {'voiceprint': r, 'speakers': 1} + return {'error': r.get('error', ''), 'speakers': 0} + return {'error': 'service returned ' + str(resp.status), 'speakers': 0} + except Exception as e: + return {'error': str(e), 'speakers': 0} + +async def process_upload(env, file_data, kb_id, folder_id, file_name): + """Full upload pipeline — save + classify + process + DB record""" + import uuid as _uuid + doc_id = str(_uuid.uuid4()).hex[:16] + ext = '.' + file_name.rsplit('.', 1)[1] if '.' in file_name else '.bin' + saved_name = doc_id + ext + file_path = '/d/rag/ragserver/pkgs/rag/rag/files/' + saved_name + with open(file_path, 'wb') as f: + f.write(file_data) + file_size = len(file_data) + + ft = classify_file(file_name) + ingest_result = None + face_result = None + voice_result = None + status = 'pending' + + if ft in ('text', 'document'): + text = await extract_text(file_data, file_name) + if text and len(text) > 10: + ingest_result = await ingest_to_rag(env, text, kb_id, doc_id) + status = 'done' if ingest_result and not ingest_result.get('error') else 'error' + else: + status = 'done' + elif ft == 'image': + face_result = await detect_faces(file_data, file_name) + status = 'done' + elif ft == 'audio': + voice_result = await extract_voiceprint(file_data, file_name) + status = 'done' + elif ft == 'video': + face_result = await detect_faces(file_data, file_name) + voice_result = await extract_voiceprint(file_data, file_name) + status = 'done' + else: + status = 'done' + + import json as _json + meta = _json.dumps({'file_type': ft, 'face': face_result, 'voice': voice_result, 'ingest': str(ingest_result)[:500]}, ensure_ascii=False, default=str) + + userorgid = await env.get_userorgid() + from sqlor.dbpools import get_sor_context + async with get_sor_context(env, 'rag') as sor: + await sor.sqlExe( + "INSERT INTO documents (id, kb_id, folder_id, file_name, file_type, file_size, file_path, mime_type, status, metadata, org_id, created_at, updated_at) " + "VALUES (" + "${id}$, ${kb_id}$, ${folder_id}$, ${file_name}$, 'other', ${file_size}$, ${file_path}$, 'application/octet-stream', ${status}$, ${meta}$, ${org_id}$, NOW(), NOW())", + {"id": doc_id, "kb_id": kb_id, "folder_id": folder_id, "file_name": file_name, + "file_size": file_size, "file_path": "/idfile/files/" + saved_name, + "status": status, "meta": meta, "org_id": userorgid}) + await sor.sqlExe( + "UPDATE knowledge_bases SET doc_count=doc_count+1, total_size=total_size+" + "${size}$ WHERE id=${kb_id}$", + {"size": file_size, "kb_id": kb_id}) + if ingest_result and status == 'done': + chunks_n = ingest_result.get('chunks', 0) if isinstance(ingest_result, dict) else 0 + if chunks_n: + await sor.sqlExe("UPDATE documents SET chunk_count=${chunks}$ WHERE id=${id}$", {"chunks": chunks_n, "id": doc_id}) + await sor.sqlExe("UPDATE knowledge_bases SET chunk_count=chunk_count+${n}$ WHERE id=${kb_id}$", {"n": chunks_n, "kb_id": kb_id}) + + faces_n = face_result.get('faces', 0) if face_result and isinstance(face_result, dict) else 0 + speakers_n = voice_result.get('speakers', 0) if voice_result and isinstance(voice_result, dict) else 0 + return _json.dumps({ + "status": "SUCCEEDED", "doc_id": doc_id, "file_name": file_name, + "file_size": file_size, "folder_id": folder_id, "file_type": ft, + "rag_status": status, "faces": faces_n, "speakers": speakers_n + }, ensure_ascii=False, default=str) diff --git a/wwwroot/knowledge_bases_list/index.ui b/wwwroot/knowledge_bases_list/index.ui index 0d92287..70293d6 100644 --- a/wwwroot/knowledge_bases_list/index.ui +++ b/wwwroot/knowledge_bases_list/index.ui @@ -9,28 +9,127 @@ "subwidgets": [ { "widgettype": "HBox", - "options": {"width": "100%", "justifyContent": "space-between", "alignItems": "center"}, + "options": { + "width": "100%", + "justifyContent": "space-between", + "alignItems": "center" + }, "subwidgets": [ - {"widgettype": "Text", "options": {"text": "知识库", "cfontsize": 22, "fontWeight": "bold", "color": "#333"}}, - {"widgettype": "Button", "options": {"label": "+ 新建知识库", "bgcolor": "#4a90d9", "color": "#fff"}, "binds": [ - {"wid": "self", "event": "click", "actiontype": "urlwidget", - "target": "Popup", - "popup_options": {"auto_dismiss": true}, - "options": {"url": "{{entire_url('/knowledge_bases_list/new_kb_form.ui')}}"}} - ]} + { + "widgettype": "Text", + "options": { + "text": "知识库", + "cfontsize": 22, + "fontWeight": "bold", + "color": "#333" + } + }, + { + "widgettype": "HBox", + "options": { + "spacing": "8px" + }, + "subwidgets": [ + { + "widgettype": "Button", + "options": { + "label": "🔍 检索知识", + "bgcolor": "#10b981", + "color": "#fff" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.rag_main_content", + "mode": "replace", + "options": { + "url": "./search.ui" + } + } + ] + }, + { + "widgettype": "Button", + "options": { + "label": "+ 新建知识库", + "bgcolor": "#4a90d9", + "color": "#fff" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "Popup", + "popup_options": { + "auto_dismiss": true + }, + "options": { + "url": "{{entire_url('/knowledge_bases_list/new_kb_form.ui')}}" + } + } + ] + } + ] + } ] }, { "widgettype": "VBox", - "options": {"width": "100%", "bgcolor": "#f8f9fa", "padding": "16px", "css": "card", "spacing": "4px"}, + "options": { + "width": "100%", + "bgcolor": "#f8f9fa", + "padding": "16px", + "css": "card", + "spacing": "4px" + }, "subwidgets": [ - {"widgettype": "HBox", "options": {"width": "100%", "justifyContent": "space-between"}, "subwidgets": [ - {"widgettype": "Text", "options": {"text": "💾 存储容量", "cfontsize": 14, "fontWeight": "bold", "color": "#333"}}, - {"widgettype": "Text", "options": {"text": "已用 0MB / 总计 100MB", "cfontsize": 12, "color": "#888"}} - ]}, - {"widgettype": "VBox", "options": {"width": "100%", "cheight": 0.5, "bgcolor": "#e0e0e0", "css": ""}, "subwidgets": [ - {"widgettype": "VBox", "options": {"cwidth": 0, "cheight": 0.5, "bgcolor": "#4a90d9", "css": ""}} - ]} + { + "widgettype": "HBox", + "options": { + "width": "100%", + "justifyContent": "space-between" + }, + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "text": "💾 存储容量", + "cfontsize": 14, + "fontWeight": "bold", + "color": "#333" + } + }, + { + "widgettype": "Text", + "options": { + "text": "已用 0MB / 总计 100MB", + "cfontsize": 12, + "color": "#888" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "width": "100%", + "cheight": 0.5, + "bgcolor": "#e0e0e0" + }, + "subwidgets": [ + { + "widgettype": "VBox", + "options": { + "cwidth": 0, + "cheight": 0.5, + "bgcolor": "#4a90d9" + } + } + ] + } ] }, { @@ -40,4 +139,4 @@ } } ] -} +} \ No newline at end of file diff --git a/wwwroot/knowledge_bases_list/kb_options.dspy b/wwwroot/knowledge_bases_list/kb_options.dspy new file mode 100644 index 0000000..81b0c09 --- /dev/null +++ b/wwwroot/knowledge_bases_list/kb_options.dspy @@ -0,0 +1,6 @@ +import json +env = request._run_ns +async with get_sor_context(env, 'rag') as sor: + recs = await sor.sqlExe("SELECT id, name FROM knowledge_bases WHERE org_id IS NULL OR org_id='0' ORDER BY created_at DESC", {}) + items = [{"value": r.id, "text": r.name} for r in recs] + return json.dumps(items, ensure_ascii=False) diff --git a/wwwroot/knowledge_bases_list/search.ui b/wwwroot/knowledge_bases_list/search.ui new file mode 100644 index 0000000..bf7d5fd --- /dev/null +++ b/wwwroot/knowledge_bases_list/search.ui @@ -0,0 +1,56 @@ +{ + "widgettype": "VBox", + "options": {"padding": "20px", "spacing": "16px", "width": "100%"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": "🔍 知识检索", "cfontsize": 22, "fontWeight": "bold", "color": "#1a1a2e"}}, + {"widgettype": "Text", "options": {"text": "输入文本或上传文件，检索相关内容", "cfontsize": 13, "color": "#888"}}, + { + "widgettype": "HBox", + "options": {"spacing": "12px", "alignItems": "center"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": "知识库:", "cfontsize": 13, "color": "#666"}}, + { + "widgettype": "code", + "id": "kb_selector", + "options": {"name": "kb_id", "dataurl": "{{entire_url('./kb_options.dspy')}}", "text": "全部", "cwidth": 18} + }, + { + "widgettype": "SearchBar", + "id": "search_bar", + "options": {"cwidth": 25, "cfontsize": 14, "placeholder": "输入检索关键词..."} + }, + { + "widgettype": "UiFile", + "id": "search_file", + "options": {"accept": "", "multiple": false, "preview": false, "otext": "📎 上传文件检索", "cwidth": 12, "cheight": 3, "css": "card", "border": "1px dashed #ccc"} + } + ] + }, + { + "widgettype": "VBox", + "id": "search_results", + "options": {"marginTop": "16px", "spacing": "0px"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": "选择知识库，输入文字或上传文件开始检索", "cfontsize": 14, "color": "#aaa", "halign": "center", "marginTop": "40px"}} + ] + } + ], + "binds": [ + { + "wid": "search_bar", + "event": "search", + "actiontype": "urlwidget", + "target": "search_results", + "mode": "replace", + "datawidget": "kb_selector", + "options": {"url": "{{entire_url('./search_result.dspy')}}"} + }, + { + "wid": "search_file", + "event": "changed", + "actiontype": "script", + "target": "search_file", + "script": "var f=this.value;if(!f)return;if(Array.isArray(f))f=f[0];if(!f)return;var kb=document.querySelector('[id=kb_selector]');var kb_id=kb?kb.value||'':'all';var fd=new FormData();fd.append('file',f);var u='/rag/knowledge_bases_list/search_result.dspy?_webbricks_=1&kb_id='+encodeURIComponent(kb_id)+'&file_name='+encodeURIComponent(f.name);fetch(u,{method:'POST',body:fd}).then(function(r){return r.json()}).then(function(d){var fp=bricks.getWidgetById('search_results',bricks.app.root);if(fp)fp.clear_widgets();bricks.widgetBuild(d,fp).then(function(w){if(w)fp.add_widget(w)})})" + } + ] +} diff --git a/wwwroot/knowledge_bases_list/search_result.dspy b/wwwroot/knowledge_bases_list/search_result.dspy new file mode 100644 index 0000000..b0b7bf7 --- /dev/null +++ b/wwwroot/knowledge_bases_list/search_result.dspy @@ -0,0 +1,98 @@ +ns = params_kw.copy() +kb_id = ns.get('kb_id', '') +env = request._run_ns +query = ns.get("query") or ns.get("value") or ns.get("keyword") or ns.get("text") or "" + +file_data = None +file_name = None +try: + reader = await request.multipart() + while True: + part = await reader.next() + if part is None: break + if part.name == "file": file_data = await part.read(); file_name = part.filename; break +except: pass +if not file_data: + try: file_data = await request.read(); + except: pass + if file_data and len(file_data) < 10: file_data = None + +import base64, io +top_k = int(ns.get('top_k', 5)) + +if not query: + return json.dumps({ + "widgettype": "VBox", "options": {"padding": "20px", "spacing": "16px"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": "🔍 知识检索", "cfontsize": 20, "fontWeight": "bold"}}, + {"widgettype": "Text", "options": {"text": "请输入检索内容", "color": "#666", "marginTop": "20px"}} + ] + }, ensure_ascii=False) + +import aiohttp + +async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=20)) as s: + # 1. Embed + r = await s.post('https://embedding.opencomputing.net:10443/api/embed', + json={"texts": [query], "model": "CLIP-ViT-H-14"}) + emb = await r.json() if r.status == 200 else {} + vec = emb.get("text_embeddings", emb.get("embeddings", [[]]))[0] + if not vec: + return json.dumps({"widgettype": "Text", "options": {"text": "向量化失败", "cfontsize": 14, "color": "#e74c3c"}}, ensure_ascii=False) + + # 2. VDB search + recall_n = top_k * 3 + r2 = await s.post('https://vectordb.opencomputing.net:10443/v1/query', + json={"colname": kb_id, "vector": vec, "pagerows": recall_n, "output_fields": ["*"]}) + vdb = await r2.json() if r2.status == 200 else {} + raw_rows = vdb.get("data", {}).get("rows", []) + if not isinstance(raw_rows, list): + raw_rows = [] + +hits = [] +for row in raw_rows: + rid = str(row.get("id", "")) + score = row.get("score", 0) + # Resolve doc_id from chunk id pattern "xxxx_c0" + doc_id = rid.rsplit('_c', 1)[0] if '_c' in rid else rid + # Look up chunk text from DB + chunk_text = '' + doc_name = '' + async with get_sor_context(env, 'rag') as sor: + recs = await sor.sqlExe( + "SELECT content, doc_id FROM document_chunks WHERE id=${id}$", + {"id": rid}) + if recs: + chunk_text = recs[0].content or '' + doc_id = recs[0].doc_id or doc_id + if chunk_text: + hits.append({"id": rid, "doc_id": doc_id, "score": score, "text": chunk_text}) + +hits.sort(key=lambda x: x.get("score", 0), reverse=True) +hits = hits[:top_k] + +subwidgets = [ + {"widgettype": "Text", "options": {"text": f"🔍 检索: {query}", "cfontsize": 18, "fontWeight": "bold", "marginBottom": "8px"}}, + {"widgettype": "Text", "options": {"text": f"共 {len(hits)} 条结果 (召回 {len(raw_rows)} 条)", "cfontsize": 13, "color": "#888", "marginBottom": "16px"}} +] + +if not hits: + subwidgets.append({"widgettype": "Text", "options": {"text": "😔 未找到相关内容", "cfontsize": 14, "color": "#aaa", "halign": "center", "marginTop": "40px"}}) +else: + for i, h in enumerate(hits): + score_pct = round(float(h["score"]) * 100, 1) + color = "#3b82f6" if score_pct > 60 else ("#10b981" if score_pct > 30 else "#f59e0b") + subwidgets.append({ + "widgettype": "VBox", + "options": {"padding": "12px 16px", "marginBottom": "8px", "border": "1px solid #e0e0e0", "borderLeft": f"3px solid {color}", "bgcolor": "#fafafa", "borderRadius": "4px"}, + "subwidgets": [ + {"widgettype": "HBox", "options": {"alignItems": "center", "marginBottom": "6px"}, "subwidgets": [ + {"widgettype": "Text", "options": {"text": f"#{i+1}", "cfontsize": 12, "fontWeight": "bold", "color": color, "marginRight": "8px"}}, + {"widgettype": "Text", "options": {"text": f"{score_pct}%", "cfontsize": 11, "bgcolor": color, "color": "#fff", "padding": "2px 8px", "borderRadius": "10px"}}, + {"widgettype": "Text", "options": {"text": f" {h['id'][:16]}", "cfontsize": 11, "color": "#999", "marginLeft": "8px"}} + ]}, + {"widgettype": "Text", "options": {"text": h["text"][:300], "cfontsize": 13, "color": "#333", "lineHeight": "1.6"}} + ] + }) + +return json.dumps({"widgettype": "VBox", "options": {"padding": "20px", "spacing": "4px"}, "subwidgets": subwidgets}, ensure_ascii=False, default=str)