diff --git a/models/rag_api_keys.json b/models/rag_api_keys.json new file mode 100644 index 0000000..937fbf5 --- /dev/null +++ b/models/rag_api_keys.json @@ -0,0 +1,98 @@ +{ + "summary": [ + { + "name": "rag_api_keys", + "title": "对外API密钥", + "primary": [ + "id" + ] + } + ], + "fields": [ + { + "name": "id", + "title": "ID", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "org_id", + "title": "所属机构", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "name", + "title": "密钥名称", + "type": "str", + "length": 100, + "nullable": "yes" + }, + { + "name": "key_hash", + "title": "密钥哈希(SHA256)", + "type": "str", + "length": 64, + "nullable": "no" + }, + { + "name": "prefix", + "title": "密钥前缀(展示)", + "type": "str", + "length": 16, + "nullable": "yes" + }, + { + "name": "scopes", + "title": "权限范围", + "type": "str", + "length": 500, + "nullable": "yes", + "default": "" + }, + { + "name": "status", + "title": "状态", + "type": "str", + "length": 16, + "nullable": "yes", + "default": "active" + }, + { + "name": "expires_at", + "title": "过期时间", + "type": "datetime", + "nullable": "yes" + }, + { + "name": "last_used_at", + "title": "最近使用", + "type": "datetime", + "nullable": "yes" + }, + { + "name": "created_at", + "title": "创建时间", + "type": "datetime", + "nullable": "yes" + } + ], + "indexes": [ + { + "name": "idx_ak_hash", + "idxtype": "unique", + "idxfields": [ + "key_hash" + ] + }, + { + "name": "idx_ak_org", + "idxtype": "index", + "idxfields": [ + "org_id" + ] + } + ] +} diff --git a/rag/api_core.py b/rag/api_core.py new file mode 100644 index 0000000..16414b6 --- /dev/null +++ b/rag/api_core.py @@ -0,0 +1,551 @@ +# -*- coding:utf-8 -*- +"""RAG 对外 API 核心（B2B 机器接口）— Bearer Key 鉴权 + org 注入。 + +与 UI 通道的关系： +- UI 通道（wwwroot/knowledge_bases_list/*.dspy）：RBAC 登录会话鉴权，org/user 取自会话； +- API 通道（wwwroot/api/*.dspy → 本模块）：路由/权限照常由 RBAC 控制 + （端点授 any），业务身份由 rag_api_keys 的 Bearer Key 决定（绑定 org_id）。 + +org 注入方式：构造一个轻量 env（DictObject），get_userorgid/get_user 返回 key 绑定的 +机构，get_module_dbname 等透传全局 ServerEnv —— 复用 init.py 的底层能力 +（_resolve_search_kbs/_build_search_vector/_call_uapi 等），杜绝双实现分叉。 + +权限语义（v1）： +- key 即机构级凭据：org 隔离强制（只能操作本机构知识库），KB 级 maintain_roles/ + search_roles 是人类通道的门槛，对 API key 不生效（key 的授权面由 scopes 控制）； +- scopes 逗号分隔：kb.create / kb.delete / doc.upload / doc.delete / tag.write / search； + '*' 全通。 + +2026-09-03 新增（配合 wwwroot/api/ 六个对外端点）。 +""" +import json +import os +import time +import uuid as _uuid +from hashlib import sha256 + +from ahserver.serverenv import ServerEnv +from appPublic.dictObject import DictObject +from sqlor.dbpools import get_sor_context + + +# ────────────────────────── API Key 鉴权 ────────────────────────── + +def hash_key(plain): + return sha256((plain or '').encode('utf-8')).hexdigest() + + +async def verify_api_key(request): + """从 Authorization: Bearer 或 x-api-key: 验证调用方。 + + 返回 (ctx_dict, None) 或 (None, error_message)。 + ctx = {org_id, key_id, name, scopes, user_id} + """ + token = '' + auth = request.headers.get('Authorization', '') or '' + if auth.startswith('Bearer '): + token = auth[7:].strip() + if not token: + token = (request.headers.get('x-api-key', '') or '').strip() + if not token: + return None, "missing api key（Authorization: Bearer 或 x-api-key 头）" + + env = ServerEnv() + async with get_sor_context(env, 'rag') as sor: + recs = await sor.sqlExe( + "SELECT id, org_id, name, scopes, status, expires_at FROM rag_api_keys " + "WHERE key_hash=${h}$ LIMIT 1", {"h": hash_key(token)}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return None, "invalid api key" + r = recs[0] + if (getattr(r, 'status', '') or '') != 'active': + return None, "api key disabled" + exp = getattr(r, 'expires_at', None) + if exp is not None and str(exp).strip() and time.strftime('%Y-%m-%d %H:%M:%S') >= str(exp): + return None, "api key expired" + scopes = [s.strip() for s in str(getattr(r, 'scopes', '') or '').split(',') if s.strip()] + key_id = getattr(r, 'id', '') + # 最近使用时间（best-effort，失败不影响调用） + try: + await sor.sqlExe( + "UPDATE rag_api_keys SET last_used_at=NOW() WHERE id=${i}$", {"i": key_id}) + await sor.sqlExe("COMMIT", {}) + except Exception: + try: + await sor.sqlExe("COMMIT", {}) + except Exception: + pass + return {"org_id": str(getattr(r, 'org_id', '') or ''), "key_id": key_id, + "name": str(getattr(r, 'name', '') or ''), "scopes": scopes, + "user_id": ''}, None + + +def require_scope(ctx, scope): + sc = ctx.get('scopes') or [] + return '*' in sc or scope in sc + + +async def read_json_body(request, params_kw): + """三级兜底取请求体：JSON body → raw text JSON → query/form params_kw。""" + payload = None + try: + payload = await request.json() + except Exception: + payload = None + if not isinstance(payload, dict): + try: + raw = await request.text() + payload = json.loads(raw) if raw and raw.strip().startswith('{') else None + except Exception: + payload = None + if isinstance(payload, dict) and payload: + return payload + return dict(params_kw or {}) + + +# ────────────────────────── org 注入 env ────────────────────────── + +def make_api_env(ctx): + """轻量 env：org 取自 API key，其余透传全局 ServerEnv。""" + g = ServerEnv() + + async def _orgid(): + return ctx['org_id'] + + async def _user(): + return ctx.get('user_id') or '' + + return DictObject( + get_userorgid=_orgid, + get_user=_user, + get_module_dbname=g.get_module_dbname, + ) + + +def _ok(**kw): + """统一成功格式：{"status":"ok","data":{...}}""" + return json.dumps({"status": "ok", "data": kw}, ensure_ascii=False, default=str) + + +def _err(message, **kw): + """统一错误格式：{"status":"error","message":...,"data":null}""" + out = {"status": "error", "message": message, "data": None} + if kw: + out.update(kw) + return json.dumps(out, ensure_ascii=False, default=str) + + +# ────────────────────────── 业务核心（org 来自 key） ────────────────────────── + +_ENGINES = ("bge-m3", "clip-vith14", "qwen3-vl-embedding") + + +async def kb_create(env, ns): + """创建知识库（embedding_engine 创建时定死，知识库级选：bge-m3文本 / clip-vith14多媒体 / qwen3-vl-embedding多模态在线）。""" + name = (ns.get('name') or '').strip() + if not name: + return _err("name required") + emb = (ns.get('embedding_engine') or ns.get('embedding_type') or 'bge-m3').strip() + if emb not in _ENGINES: + return _err("embedding_engine must be one of " + "/".join(_ENGINES)) + org_id = await env.get_userorgid() + kb_id = _uuid.uuid4().hex + from rag.init import get_rags_base, ensure_kb_dir + async with get_sor_context(env, 'rag') as sor: + dup = await sor.sqlExe( + "SELECT id FROM rag_knowledge_bases WHERE org_id=${o}$ AND name=${n}$ LIMIT 1", + {"o": org_id, "n": name}) + await sor.sqlExe("COMMIT", {}) + if dup: + return _err("知识库名称已存在: " + name) + await sor.sqlExe( + "INSERT INTO rag_knowledge_bases (id, name, description, org_id, embedding_engine, " + "vdb_collection, doc_count, total_size, chunk_count, status, maintain_roles, search_roles, created_at) " + "VALUES (${id}$, ${name}$, ${desc}$, ${org_id}$, ${emb}$, 'rag_collection', 0, 0, 0, 'active', '', '', NOW())", + {"id": kb_id, "name": name, "desc": ns.get('description') or '', + "org_id": org_id, "emb": emb}) + await sor.sqlExe("COMMIT", {}) + rags_base = await get_rags_base(sor) + ensure_kb_dir(rags_base, org_id, kb_id) + return _ok(kb_id=kb_id, name=name, embedding_engine=emb) + + +async def _kb_owned(sor, env, kb_id): + org_id = await env.get_userorgid() + recs = await sor.sqlExe( + "SELECT * FROM rag_knowledge_bases WHERE id=${k}$ LIMIT 1", {"k": kb_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return None, "知识库不存在" + kb = recs[0] + if str(getattr(kb, 'org_id', '') or '') != str(org_id or ''): + return None, "无权操作其他机构的知识库" + return kb, None + + +async def kb_delete(env, ns): + """删除知识库：VDB 集合级删除 + 图 + DB 记录 + 磁盘文件 全清理。""" + kb_id = (ns.get('kb_id') or '').strip() + if not kb_id: + return _err("kb_id required") + from rag.init import _call_uapi, get_rags_base + vdb_err = graph_err = None + async with get_sor_context(env, 'rag') as sor: + kb, err = await _kb_owned(sor, env, kb_id) + if err: + return _err(err) + docs = await sor.sqlExe( + "SELECT id, file_path, file_size FROM rag_documents WHERE kb_id=${k}$", {"k": kb_id}) + docs = list(docs or []) + doc_ids = [d.id for d in docs] + + # VDB：整集合清理（集合名=kb_id，见 create 与 ingest 约定） + try: + chunks = await sor.sqlExe( + "SELECT vector_id FROM rag_document_chunks WHERE kb_id=${k}$ " + "AND vector_id IS NOT NULL AND vector_id != ''", {"k": kb_id}) + vids = [c.vector_id for c in (chunks or [])] + if vids: + await _call_uapi("rag-vdb", "delete", {"colname": kb_id, "ids": vids}) + except Exception as e: + vdb_err = str(e)[:200] + + # 图 + try: + await _call_uapi("rag-graph", "delete", {"graph": kb_id}) + except Exception as e: + graph_err = str(e)[:200] + + # DB 记录 + if doc_ids: + await sor.sqlExe("DELETE FROM rag_document_chunks WHERE doc_id IN (${ids}$)", {"ids": doc_ids}) + await sor.sqlExe("DELETE FROM rag_entities WHERE kb_id=${k}$", {"k": kb_id}) + await sor.sqlExe("DELETE FROM rag_entity_relations WHERE kb_id=${k}$", {"k": kb_id}) + await sor.sqlExe("DELETE FROM rag_media_tags WHERE kb_id=${k}$", {"k": kb_id}) + await sor.sqlExe("DELETE FROM rag_tags WHERE kb_id=${k}$", {"k": kb_id}) + await sor.sqlExe("DELETE FROM rag_documents WHERE kb_id=${k}$", {"k": kb_id}) + await sor.sqlExe("DELETE FROM rag_knowledge_bases WHERE id=${k}$", {"k": kb_id}) + await sor.sqlExe("COMMIT", {}) + rags_base = await get_rags_base(sor) + + # 磁盘文件（/rags/ 新布局 + /idfile/ 旧布局兼容删除） + files_removed = 0 + for d in docs: + fp = getattr(d, 'file_path', '') or '' + try: + if fp.startswith('/rags/'): + parts = [p for p in fp.split('/') if p] + if len(parts) >= 4: + real = os.path.join(rags_base, parts[1], 'rags', parts[2], parts[3]) + if os.path.isfile(real): + os.remove(real) + files_removed += 1 + else: + g = ServerEnv() + real = g.realpath(fp) if callable(getattr(g, 'realpath', None)) else None + if real and os.path.isfile(real): + os.remove(real) + files_removed += 1 + except Exception: + pass + # 知识库目录整体清理（rags/{org}/{kb}/ 空壳） + try: + import shutil + kdir = os.path.join(rags_base, str(await env.get_userorgid() or '0'), 'rags', kb_id) + if os.path.isdir(kdir): + shutil.rmtree(kdir, ignore_errors=True) + except Exception: + pass + return _ok(kb_id=kb_id, documents=len(docs), files_removed=files_removed, + vdb_cleanup=vdb_err or 'done', graph_cleanup=graph_err or 'done') + + +async def doc_delete(env, ns): + """删除文档：VDB 向量 → chunks/entities DB → 磁盘文件 → KB 统计回退。""" + doc_id = (ns.get('doc_id') or '').strip() + if not doc_id: + return _err("doc_id required") + from rag.init import _call_uapi, get_rags_base + async with get_sor_context(env, 'rag') as sor: + recs = await sor.R("rag_documents", {"id": doc_id}) + if not recs: + return _err("document not found") + doc = recs[0] + _kb, err = await _kb_owned(sor, env, doc.kb_id) + if err: + return _err(err) + chunks = await sor.R("rag_document_chunks", {"doc_id": doc_id}) + vids = [c.vector_id for c in (chunks or []) if getattr(c, 'vector_id', '')] + if vids: + try: + await _call_uapi("rag-vdb", "delete", {"colname": doc.kb_id, "ids": vids}) + except Exception: + pass + await sor.sqlExe("DELETE FROM rag_document_chunks WHERE doc_id=${i}$", {"i": doc_id}) + await sor.sqlExe("DELETE FROM rag_media_tags WHERE media_type='document' AND media_id=${i}$", {"i": doc_id}) + await sor.sqlExe("DELETE FROM rag_documents WHERE id=${i}$", {"i": doc_id}) + await sor.sqlExe( + "UPDATE rag_knowledge_bases SET doc_count=GREATEST(doc_count-1,0), " + "total_size=GREATEST(total_size-${s}$,0), chunk_count=GREATEST(chunk_count-${n}$,0) " + "WHERE id=${k}$", + {"s": getattr(doc, 'file_size', 0) or 0, "n": len(chunks or []), "k": doc.kb_id}) + await sor.sqlExe("COMMIT", {}) + rags_base = await get_rags_base(sor) + kb_id, file_path = doc.kb_id, getattr(doc, 'file_path', '') or '' + + removed = False + if file_path.startswith('/rags/'): + parts = [p for p in file_path.split('/') if p] + if len(parts) >= 4: + real = os.path.join(rags_base, parts[1], 'rags', parts[2], parts[3]) + if os.path.isfile(real): + os.remove(real) + removed = True + else: + try: + g = ServerEnv() + real = g.realpath(file_path) if callable(getattr(g, 'realpath', None)) else None + if real and os.path.isfile(real): + os.remove(real) + removed = True + except Exception: + pass + return _ok(doc_id=doc_id, kb_id=kb_id, chunks_deleted=len(chunks or []), file_removed=removed) + + +async def tag_create(env, ns): + """创建标签（同名幂等返回已有标签）。""" + kb_id = (ns.get('kb_id') or '').strip() + name = (ns.get('name') or '').strip() + color = (ns.get('color') or '#3b82f6').strip() + if not kb_id or not name: + return _err("kb_id and name required") + async with get_sor_context(env, 'rag') as sor: + _kb, err = await _kb_owned(sor, env, kb_id) + if err: + return _err(err) + org_id = await env.get_userorgid() + existing = await sor.sqlExe( + "SELECT id, name, color FROM rag_tags WHERE kb_id=${k}$ AND name=${n}$ AND org_id=${o}$", + {"k": kb_id, "n": name, "o": org_id}) + if existing: + r = existing[0] + return _ok(tag_id=r.id, name=r.name, color=r.color, duplicate=True) + tag_id = _uuid.uuid4().hex + await sor.sqlExe( + "INSERT INTO rag_tags (id, kb_id, name, color, org_id, created_at) " + "VALUES (${id}$, ${k}$, ${n}$, ${c}$, ${o}$, NOW())", + {"id": tag_id, "k": kb_id, "n": name, "c": color, "o": org_id}) + await sor.sqlExe("COMMIT", {}) + return _ok(tag_id=tag_id, name=name, color=color) + + +def _as_list(v): + if isinstance(v, (list, tuple)): + return [str(x).strip() for x in v if str(x).strip()] + return [s.strip() for s in str(v or '').split(',') if s.strip()] + + +async def doc_set_tags(env, ns): + """文件设置标签（全量语义：传入的即最终集合，多余的删、缺的补）。 + + 入参：kb_id, doc_id, tags（名称数组或逗号串）或 tag_ids（id 数组或逗号串）。 + 标签不存在时自动在本知识库创建。 + """ + kb_id = (ns.get('kb_id') or '').strip() + doc_id = (ns.get('doc_id') or '').strip() + if not kb_id or not doc_id: + return _err("kb_id and doc_id required") + tag_ids_in = _as_list(ns.get('tag_ids')) + tag_names = _as_list(ns.get('tags')) + async with get_sor_context(env, 'rag') as sor: + _kb, err = await _kb_owned(sor, env, kb_id) + if err: + return _err(err) + docs = await sor.sqlExe( + "SELECT id FROM rag_documents WHERE id=${d}$ AND kb_id=${k}$", {"d": doc_id, "k": kb_id}) + if not docs: + return _err("document not found in this kb") + org_id = await env.get_userorgid() + + # 名称 → id（不存在自动建） + for nm in tag_names: + recs = await sor.sqlExe( + "SELECT id FROM rag_tags WHERE kb_id=${k}$ AND name=${n}$", {"k": kb_id, "n": nm}) + if recs: + tag_ids_in.append(recs[0].id) + else: + tid = _uuid.uuid4().hex + await sor.sqlExe( + "INSERT INTO rag_tags (id, kb_id, name, color, org_id, created_at) " + "VALUES (${id}$, ${k}$, ${n}$, '#3b82f6', ${o}$, NOW())", + {"id": tid, "k": kb_id, "n": nm, "o": org_id}) + tag_ids_in.append(tid) + wanted = list(dict.fromkeys(tag_ids_in)) # 去重保序 + + # 校验 tag_id 属于本知识库（wanted 为空=清空全部标签，合法） + if wanted: + nsmap = {("t%d" % i): t for i, t in enumerate(wanted)} + placeholders = ",".join("${" + k + "}$" for k in nsmap) + recs = await sor.sqlExe( + "SELECT id FROM rag_tags WHERE id IN (" + placeholders + ")", nsmap) + valid = {r.id for r in (recs or [])} + bad = [t for t in wanted if t not in valid] + if bad: + return _err("tag not found in this kb: " + ",".join(bad)) + + cur = await sor.sqlExe( + "SELECT id, tag_id FROM rag_media_tags WHERE media_type='document' AND media_id=${d}$", + {"d": doc_id}) + current = {r.tag_id: r.id for r in (cur or [])} + removed = added = 0 + for tid, mt_id in current.items(): + if tid not in wanted: + await sor.sqlExe("DELETE FROM rag_media_tags WHERE id=${i}$", {"i": mt_id}) + removed += 1 + for tid in wanted: + if tid not in current: + await sor.sqlExe( + "INSERT INTO rag_media_tags (id, kb_id, media_type, media_id, tag_id, created_at) " + "VALUES (${id}$, ${k}$, 'document', ${d}$, ${t}$, NOW())", + {"id": _uuid.uuid4().hex, "k": kb_id, "d": doc_id, "t": tid}) + added += 1 + await sor.sqlExe("COMMIT", {}) + # 回读最终标签列表 + recs = await sor.sqlExe( + "SELECT t.id, t.name, t.color FROM rag_media_tags mt JOIN rag_tags t ON mt.tag_id=t.id " + "WHERE mt.media_type='document' AND mt.media_id=${d}$", {"d": doc_id}) + tags = [{"id": r.id, "name": r.name, "color": r.color} for r in (recs or [])] + return _ok(doc_id=doc_id, added=added, removed=removed, tags=tags) + + +async def doc_upload(env, ns, file_data, file_name): + """文件上传：落盘 {workspace_base}/{org}/rags/{kb}/ + 文档记录 + 后台入库（与 UI 同一引擎 rag.ingest）。""" + kb_id = (ns.get('kb_id') or '').strip() + if not kb_id: + return _err("kb_id required") + if not file_data: + return _err("empty file body") + file_name = file_name or 'upload.bin' + org_id = await env.get_userorgid() + from rag.init import get_rags_base, ensure_kb_dir, _detect_file_type, _get_param_value, _fmt_bytes + from ahserver.globalEnv import background_reco + from rag.ingest import ingest_one + + async with get_sor_context(env, 'rag') as sor: + _kb, err = await _kb_owned(sor, env, kb_id) + if err: + return _err(err) + check_quota = str(await _get_param_value(sor, 'rag_check_storage_quota', '0')).strip().lower() + quota_msg = None + if check_quota in ('1', 'true', 'yes', 'on'): + rec = await sor.sqlExe( + "SELECT COALESCE(SUM(file_size),0) AS used FROM rag_documents WHERE org_id=${o}$", + {"o": org_id}) + used = int(rec[0].used) if rec else 0 + lim = await sor.sqlExe( + "SELECT limit_bytes FROM rag_org_storage_limits WHERE org_id=${o}$", {"o": org_id}) + quota = int(lim[0].limit_bytes) if lim else 104857600 + if used + len(file_data) > quota: + quota_msg = ("存储配额超限：机构已用 " + _fmt_bytes(used) + "，限额 " + + _fmt_bytes(quota) + "，本文件 " + _fmt_bytes(len(file_data))) + if quota_msg: + return _err(quota_msg, code="storage_quota_exceeded") + rags_base = await get_rags_base(sor) + + doc_id = _uuid.uuid4().hex + kb_dir = ensure_kb_dir(rags_base, org_id, kb_id) + safe_name = file_name.replace('/', '_').replace('\\', '_') + disk_name = doc_id[:8] + '_' + safe_name + with open(os.path.join(kb_dir, disk_name), 'wb') as f: + f.write(file_data) + web_path = '/rags/' + str(org_id or '0') + '/' + str(kb_id) + '/' + disk_name + file_type = _detect_file_type(file_name, "application/octet-stream") + + async with get_sor_context(env, 'rag') as sor: + await sor.sqlExe( + "INSERT INTO rag_documents (id, kb_id, folder_id, file_name, file_type, file_size, " + "file_path, mime_type, status, chunk_count, metadata, org_id, created_at, updated_at) " + "VALUES (${id}$, ${k}$, '', ${fn}$, 'other', ${sz}$, ${fp}$, 'application/octet-stream', " + "'pending', 0, '{}', ${o}$, NOW(), NOW())", + {"id": doc_id, "k": kb_id, "fn": safe_name, "sz": len(file_data), + "fp": web_path, "o": org_id}) + await sor.sqlExe( + "UPDATE rag_knowledge_bases SET doc_count=doc_count+1, total_size=total_size+${s}$ WHERE id=${k}$", + {"s": len(file_data), "k": kb_id}) + await sor.sqlExe("COMMIT", {}) + + ext = ('.' + file_name.rsplit('.', 1)[1]) if '.' in file_name else '.bin' + background_reco(ingest_one, doc_id, kb_id, safe_name, ext.lower(), + os.path.join(kb_dir, disk_name)) + return _ok(doc_id=doc_id, kb_id=kb_id, file_name=safe_name, file_size=len(file_data), + status='pending', ingest='running_in_background') + + +async def search(env, ns): + """知识库检索：query → embedding → 多 KB VDB 召回 → 重排 → 富化元数据。 + + 与 UI 检索共用底层能力（_resolve_search_kbs/_build_search_vector/_online_rerank）， + kb_id 缺省检索本机构全部知识库。 + """ + from rag.init import (_resolve_search_kbs, _build_search_vector, _parse_vdb_hits, + _apply_rerank, _enrich_search_results, _online_rerank, _call_uapi) + org_id = await env.get_userorgid() + query = (ns.get('query') or '').strip() + kb_id = (ns.get('kb_id') or '').strip() + try: + top_k = int(ns.get('top_k') or 10) + except (TypeError, ValueError): + top_k = 10 + try: + recall_k = int(ns.get('recall_k') or top_k * 3) + except (TypeError, ValueError): + recall_k = top_k * 3 + if not query: + return _err("query required") + + kb_ids = await _resolve_search_kbs(env, org_id, kb_id) + if not kb_ids: + return _ok(results=[], total=0, message="no knowledge bases visible to this key") + + query_vec = await _build_search_vector(query, None, None, env, kb_id) + if not query_vec: + return _err("向量化失败：检查 embedding 引擎配置", code="embed_unavailable") + + all_hits = [] + for kid in kb_ids: + try: + vdb_resp = await _call_uapi("rag-vdb", "search", { + "collection": kid, "vector": query_vec, "topK": recall_k}) + all_hits.extend(_parse_vdb_hits(vdb_resp, kid)) + except Exception: + pass + + seen = set() + unique = [] + for h in sorted(all_hits, key=lambda x: x.get("score", 0), reverse=True): + hid = h.get("id", h.get("text", "")) + if hid not in seen: + seen.add(hid) + unique.append(h) + + if unique: + documents = [h.get("text", h.get("content", "")) for h in unique[:recall_k]] + rerank_resp = await _online_rerank(env, query, documents) + if rerank_resp: + unique = _apply_rerank(unique[:recall_k], rerank_resp) + + final = unique[:top_k] + enriched = await _enrich_search_results(env, final) + # 输出瘦身：分片正文 + 分数 + 来源文档 + results = [{ + "chunk_id": h.get("id", ""), + "text": h.get("text", h.get("content", "")), + "score": h.get("rerank_score", h.get("score", 0)), + "kb_id": h.get("kb_id", ""), + "document": {k: v for k, v in (h.get("document") or {}).items() + if k in ("id", "file_name", "file_type", "kb_id")}, + } for h in enriched] + return _ok(results=results, total=len(results), + recall=len(all_hits), kbs_searched=len(kb_ids)) diff --git a/rag/ingest.py b/rag/ingest.py new file mode 100644 index 0000000..f5a92e0 --- /dev/null +++ b/rag/ingest.py @@ -0,0 +1,348 @@ +# -*- coding:utf-8 -*- +"""RAG 文档入库管线（共享模块）— 从 upload_file.dspy 抽出，UI dspy 与对外 API dspy 共用。 + +ingest_one(doc_id, kb_id, file_name, ext_l, real_path): + 后台 asyncio 任务（background_reco 派发）。只接原始类型参数， + 不触碰 env/request（响应返回后已失效），自建 DBPools 连接。 + 解析文本/图片/音频/视频 → 向量化 → VDB upsert → chunks 落库 → 文档置 done。 + +2026-09-03 从 wwwroot/knowledge_bases_list/upload_file.dspy 抽取（逻辑零改动）， +避免 UI 上传与 /rag/api 对外上传两份入库实现分叉。 +""" +import json, os, base64, subprocess +from ahserver.serverenv import ServerEnv +from sqlor.dbpools import DBPools +from appPublic.streamhttpclient import StreamHttpClient + +_env = ServerEnv() + + +def _dbname(): + return _env.get_module_dbname('rag') + + +async def ingest_one(doc_id, kb_id, file_name, ext_l, real_path): + db = DBPools() + # 库名必须经宿主的 get_module_dbname 映射（ragserver→'rag'，pipeline-app→'pipeline'） + # 后台任务无 request/env，用注入的全局函数解析；禁硬编码库名 + _dbn = _dbname() + + # 读知识库向量引擎： + # bge-m3 → 在线文本(阿里, 1024维) + # qwen3-vl-embedding → 在线多模态(阿里, 2560维) + # clip-vith14 → GPU 本地 CLIP(1024维, 有GPU环境保留) + emb_engine = 'clip-vith14' + try: + async with db.sqlorContext(_dbn) as sor: + krecs = await sor.sqlExe("SELECT embedding_engine FROM rag_knowledge_bases WHERE id=${kb_id}$", {"kb_id": kb_id}) + if krecs: + 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(_dbn) 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' + else: + 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": 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 + face_count = 0 + voice_speakers = 0 + meta_parts = {} + try: + with open(real_path, 'rb') as f: + file_data = f.read() + + text_exts = {'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm', '.py', '.js', '.css', '.yaml', '.yml', '.log', '.rst'} + image_exts = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'} + audio_exts = {'.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac'} + video_exts = {'.mp4', '.avi', '.mov', '.mkv', '.webm'} + + # --- OFFICE DOCS: text extraction (PDF/DOCX/PPTX/XLSX) --- + if ext_l == '.pdf' and not text: + import io; from PyPDF2 import PdfReader + reader = PdfReader(io.BytesIO(file_data)) + text = '\n'.join(p.extract_text() or '' for p in reader.pages) + elif ext_l == '.docx' and not text: + import io; from docx import Document + doc = Document(io.BytesIO(file_data)) + text = '\n'.join(p.text for p in doc.paragraphs) + elif ext_l == '.pptx' and not text: + import io; from pptx import Presentation + prs = Presentation(io.BytesIO(file_data)) + parts = [] + for slide in prs.slides: + for shape in slide.shapes: + if hasattr(shape, 'text') and shape.text: + parts.append(shape.text) + text = '\n'.join(parts) + elif ext_l == '.xlsx' and not text: + import io; from openpyxl import load_workbook + wb = load_workbook(io.BytesIO(file_data), data_only=True) + parts = [] + for sheet in wb.worksheets: + for row in sheet.iter_rows(values_only=True): + parts.append('\t'.join(str(c or '') for c in row)) + text = '\n'.join(parts) + + # --- TEXT EXTRACTION --- + if ext_l in text_exts: + text = file_data.decode('utf-8', errors='replace') + + # --- 文本知识库不支持媒体文件 --- + if is_text and (ext_l in image_exts or ext_l in audio_exts or ext_l in video_exts): + try: + async with db.sqlorContext(_dbn) as sor: + await sor.sqlExe( + "UPDATE rag_documents SET status='failed', metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", + {"id": doc_id, "meta": json.dumps({"error": "文本知识库不支持媒体文件，请上传文本类文件(txt/md/pdf/docx等)或改用多媒体知识库"}, ensure_ascii=False)}) + except: pass + return + + # --- IMAGE: face detection + 图片向量化入库 --- + if ext_l in image_exts: + img_b64 = base64.b64encode(file_data).decode() + try: + client = StreamHttpClient() + resp = await client.request('POST', 'https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) + fd = json.loads(resp) + results = fd.get("results", []) + if results and isinstance(results[0], dict): + faces = results[0].get("faces", results[0].get("detections", [])) + face_count = len(faces) + if faces and isinstance(faces[0], dict): + 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(_dbn) 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: + try: + client = StreamHttpClient() + resp = await client.request('POST', 'https://media.opencomputing.net:10443/voiceprint/extract/submit', + files={'file': (file_name, file_data)}) + vd = json.loads(resp) + voice_speakers = vd.get('speakers', 1) if vd.get('status') == 'SUCCEEDED' else (1 if vd.get('embedding') else 0) + meta_parts['voiceprint'] = voice_speakers + except: pass + + # --- VIDEO: frame extraction + voiceprint --- + if ext_l in video_exts: + meta_parts['video'] = 'pending' + video_ok = False + try: + tmp_img = '/tmp/' + doc_id + '_frame.jpg' + subprocess.run(['ffmpeg', '-y', '-i', real_path, '-vframes', '1', '-q:v', '2', tmp_img], + capture_output=True, timeout=30) + if os.path.exists(tmp_img): + with open(tmp_img, 'rb') as fi: + frame_data = fi.read() + img_b64 = base64.b64encode(frame_data).decode() + frame_bboxes = [] + try: + client = StreamHttpClient() + resp = await client.request('POST', 'https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) + fd = json.loads(resp) + results = fd.get("results", []) + if results and isinstance(results[0], dict): + faces = results[0].get("faces", results[0].get("detections", [])) + face_count = len(faces) + frame_bboxes = [f.get("bbox", {}) for f in faces[:10]] if faces else [] + except: pass + # --- 视频帧向量化（vl在线 或 GPU CLIP） --- + try: + frame_vec = await _media_embed_image(img_b64) + img_embeddings = [frame_vec] if frame_vec else [] + except: + img_embeddings = [] + if img_embeddings: + try: + client3 = StreamHttpClient() + await ensure_vdb_collection(client3, kb_id) + vdb_data = {"colname": kb_id, "data": [ + {"id": doc_id + "_c0", "vector": img_embeddings[0], "text": file_name} + ]} + await client3.request('POST', VDB_BASE + '/v1/upsert', json=vdb_data) + chunk_meta = {"start_time": 0} + if frame_bboxes: + chunk_meta["bboxes"] = frame_bboxes + async with db.sqlorContext(_dbn) as sor: + await sor.sqlExe( + "INSERT INTO rag_document_chunks (id, doc_id, kb_id, chunk_index, content, vector_id, metadata, created_at) " + "VALUES (${id}$, ${doc_id}$, ${kb_id}$, 0, ${content}$, ${vid}$, ${meta}$, NOW())", + {"id": doc_id + "_c0", "doc_id": doc_id, "kb_id": kb_id, + "content": file_name, "vid": doc_id + "_c0", + "meta": json.dumps(chunk_meta, ensure_ascii=False)}) + except: + pass + os.remove(tmp_img) + video_ok = True + except: pass + + # --- Voiceprint: extract audio from video --- + if video_ok: + try: + tmp_wav = '/tmp/' + doc_id + '_audio.wav' + subprocess.run(['ffmpeg', '-y', '-i', real_path, '-vn', '-acodec', 'pcm_s16le', + '-ar', '16000', '-ac', '1', tmp_wav], + capture_output=True, timeout=60) + if os.path.exists(tmp_wav) and os.path.getsize(tmp_wav) > 1000: + with open(tmp_wav, 'rb') as fa: + audio_data = fa.read() + try: + client4 = StreamHttpClient() + resp4 = await client4.request('POST', + 'https://media.opencomputing.net:10443/voiceprint/extract/submit', + files={'file': (file_name.rsplit('.', 1)[0] + '.wav', audio_data)}) + vd = json.loads(resp4) + voice_speakers = vd.get('speakers', 1) if vd.get('status') == 'SUCCEEDED' else (1 if vd.get('embedding') else 0) + meta_parts['voiceprint'] = voice_speakers + except: pass + if os.path.exists(tmp_wav): + os.remove(tmp_wav) + except: pass + + if video_ok: + meta_parts['video'] = 'done' + + # --- RAG INGEST for text --- + if text and len(text.strip()) > 10: + paragraphs = text.split('\n') + chunks = [] + cur = '' + for p in paragraphs: + p = p.strip() + if not p: + if cur: chunks.append(cur); cur = '' + continue + if len(cur) + len(p) < 500: + cur = (cur + '\n' + p).strip() + else: + if cur: chunks.append(cur) + cur = p + if cur: chunks.append(cur) + + if chunks: + # 在线文本向量化（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: + try: + client2 = StreamHttpClient() + await ensure_vdb_collection(client2, kb_id) + 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', 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: + pass + + async with db.sqlorContext(_dbn) as sor: + for i, chunk_text in enumerate(chunks): + vid = vector_ids[i] if i < len(vector_ids) else '' + 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}$, ${idx}$, ${content}$, ${vid}$, NOW())", + {"id": doc_id + "_c" + str(i), "doc_id": doc_id, "kb_id": kb_id, + "idx": i, "content": chunk_text[:2000], "vid": vid}) + chunks_n = len(chunks) + except Exception as e: + try: + async with db.sqlorContext(_dbn) as sor: + await sor.sqlExe( + "UPDATE rag_documents SET status='failed', metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", + {"id": doc_id, "meta": json.dumps({"error": str(e)[:300]}, ensure_ascii=False)}) + except: pass + return + + # --- finalize: mark document done + update KB chunk counts --- + meta_json = json.dumps(meta_parts, ensure_ascii=False) + try: + async with db.sqlorContext(_dbn) as sor: + await sor.sqlExe( + "UPDATE rag_documents SET status='done', chunk_count=${chunks}$, metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", + {"id": doc_id, "chunks": chunks_n, "meta": meta_json}) + if chunks_n: + await sor.sqlExe( + "UPDATE rag_knowledge_bases SET chunk_count=chunk_count+${n}$ WHERE id=${kb_id}$", + {"n": chunks_n, "kb_id": kb_id}) + except: pass diff --git a/rag/tools.py b/rag/tools.py new file mode 100644 index 0000000..053af01 --- /dev/null +++ b/rag/tools.py @@ -0,0 +1,186 @@ +# -*- coding:utf-8 -*- +"""RAG tools — 与对外 API 相同功能的函数包装，供内部助手（agent）调用。 + +对外 API（wwwroot/api/*.dspy → api_core.py）是给其他系统的 HTTP 机器接口； +内部助手（本进程/同生态的 LLM agent）不必绕 HTTP + API Key，直接调本模块： + + from rag.tools import RAG_TOOL_SCHEMAS, exec_rag_tool + # 助手注册工具时注入 schema；执行时： + result = await exec_rag_tool("rag_search", {"query": "...", "kb_id": "..."}, org_id=my_org) + +约定： +- schema 为 OpenAI function-calling 格式（RAG_TOOL_SCHEMAS 列表）； +- exec_rag_tool 返回 dict：成功 {"status":"ok","data":{...}}，失败 {"status":"error","message":...,"data":null} + —— 与对外 API 的 JSON 完全同构，助手侧处理逻辑可共用； +- org 隔离与 api_core 一致：以传入的 org_id 为身份边界（内部助手由宿主注入可信 org，不走 key）； +- rag_doc_upload 与 HTTP 版的差异：HTTP 收原始字节；tools 收 file_path（服务端可读路径）或 file_base64。 + +2026-09-03 新增（配合对外 API，避免"API 一套、助手又写一套"的分叉）。 +""" +import base64 +import json +import os + +from rag import api_core as C + + +# ────────────────────────── OpenAI tool schema ────────────────────────── + +RAG_TOOL_SCHEMAS = [ + { + "type": "function", + "function": { + "name": "rag_kb_create", + "description": "创建知识库。embedding_engine 创建时定死：bge-m3(纯文本) / clip-vith14(多媒体) / qwen3-vl-embedding(多模态在线)。", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "知识库名称"}, + "description": {"type": "string", "description": "描述（可选）"}, + "embedding_engine": {"type": "string", "enum": ["bge-m3", "clip-vith14", "qwen3-vl-embedding"], "description": "向量引擎，默认 bge-m3"}, + }, + "required": ["name"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_kb_delete", + "description": "删除知识库（级联清理向量、分块、文件，不可恢复）。", + "parameters": { + "type": "object", + "properties": {"kb_id": {"type": "string", "description": "知识库ID"}}, + "required": ["kb_id"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_doc_upload", + "description": "上传文件入库（异步：返回后后台解析+向量化，可稍后用 rag_search 验证）。file_path 与 file_base64 二选一。", + "parameters": { + "type": "object", + "properties": { + "kb_id": {"type": "string", "description": "目标知识库ID"}, + "file_name": {"type": "string", "description": "文件名（含扩展名）"}, + "file_path": {"type": "string", "description": "服务端可读的文件绝对路径（与 file_base64 二选一）"}, + "file_base64": {"type": "string", "description": "文件内容 base64（与 file_path 二选一）"}, + }, + "required": ["kb_id", "file_name"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_doc_delete", + "description": "删除文档（级联清理向量、分块、磁盘文件）。", + "parameters": { + "type": "object", + "properties": {"doc_id": {"type": "string", "description": "文档ID"}}, + "required": ["doc_id"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_tag_create", + "description": "创建标签（同名幂等，已存在则返回原标签）。", + "parameters": { + "type": "object", + "properties": { + "kb_id": {"type": "string", "description": "知识库ID"}, + "name": {"type": "string", "description": "标签名"}, + "color": {"type": "string", "description": "颜色 #rrggbb（可选）"}, + }, + "required": ["kb_id", "name"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_doc_set_tags", + "description": "给文档设置标签（全量语义：传入的集合即最终标签，未传的解绑；空集合=清空）。标签名不存在时自动创建。", + "parameters": { + "type": "object", + "properties": { + "kb_id": {"type": "string", "description": "知识库ID"}, + "doc_id": {"type": "string", "description": "文档ID"}, + "tags": {"type": "array", "items": {"type": "string"}, "description": "标签名列表"}, + }, + "required": ["kb_id", "doc_id"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "rag_search", + "description": "知识库检索（向量召回+重排）。不传 kb_id 则检索本机构全部知识库。", + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "检索问题/关键词"}, + "kb_id": {"type": "string", "description": "限定知识库ID（可选）"}, + "top_k": {"type": "integer", "description": "返回条数，默认10"}, + }, + "required": ["query"], + }, + }, + }, +] + +RAG_TOOL_NAMES = [t["function"]["name"] for t in RAG_TOOL_SCHEMAS] + + +# ────────────────────────── 分发执行 ────────────────────────── + +async def exec_rag_tool(tool_name: str, params: dict, org_id: str) -> dict: + """执行一个 rag tool。params 与 schema 对齐；org_id 为调用方可操作机构（宿主注入）。 + + 返回 dict：{"status":"ok","data":{...}} 或 {"status":"error","message":...,"data":null} + """ + if tool_name not in RAG_TOOL_NAMES: + return {"status": "error", "message": f"unknown rag tool: {tool_name}", "data": None} + if not org_id: + return {"status": "error", "message": "org_id required", "data": None} + ns = dict(params or {}) + env = C.make_api_env({"org_id": org_id, "user_id": ns.pop("_user_id", "") or ""}) + + if tool_name == "rag_doc_upload": + file_data = None + if ns.get("file_path"): + try: + with open(ns["file_path"], "rb") as f: + file_data = f.read() + except Exception as e: + return {"status": "error", "message": f"read file_path failed: {e}", "data": None} + elif ns.get("file_base64"): + try: + file_data = base64.b64decode(ns["file_base64"]) + except Exception as e: + return {"status": "error", "message": f"base64 decode failed: {e}", "data": None} + else: + return {"status": "error", "message": "file_path or file_base64 required", "data": None} + ns.pop("file_path", None) + ns.pop("file_base64", None) + raw = await C.doc_upload(env, ns, file_data, ns.get("file_name", "upload.bin")) + else: + fn = { + "rag_kb_create": C.kb_create, + "rag_kb_delete": C.kb_delete, + "rag_doc_delete": C.doc_delete, + "rag_tag_create": C.tag_create, + "rag_doc_set_tags": C.doc_set_tags, + "rag_search": C.search, + }[tool_name] + raw = await fn(env, ns) + + try: + return json.loads(raw) + except Exception: + return {"status": "error", "message": f"tool result not JSON: {str(raw)[:200]}", "data": None} diff --git a/scripts/load_path.py b/scripts/load_path.py index 66474e3..e59f5bd 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -13,6 +13,18 @@ PATHS_ANY = [ f"/{MOD}/knowledge_bases_list/upload.js", ] +# 对外 B2B API 端点（路由权限授 any；业务鉴权靠 rag_api_keys 的 Bearer Key， +# org 隔离与 scopes 在 rag/api_core.py 内强制——见模块 README「对外 API」） +PATHS_API_ANY = [ + f"/{MOD}/api/kb_create.dspy", + f"/{MOD}/api/kb_delete.dspy", + f"/{MOD}/api/doc_upload.dspy", + f"/{MOD}/api/doc_delete.dspy", + f"/{MOD}/api/tag_create.dspy", + f"/{MOD}/api/doc_set_tags.dspy", + f"/{MOD}/api/search.dspy", +] + # 登录用户可访问（知识库管理全部端点） PATHS_LOGINED = [ f"/{MOD}", @@ -51,6 +63,10 @@ PATHS_LOGINED = [ # 存储用量 f"/{MOD}/knowledge_bases_list/storage_card.dspy", f"/{MOD}/knowledge_bases_list/storage_stats.dspy", + # 对外 API Key 管理（管理端，登录会话鉴权） + f"/{MOD}/knowledge_bases_list/api_key_create.dspy", + f"/{MOD}/knowledge_bases_list/api_key_list.dspy", + f"/{MOD}/knowledge_bases_list/api_key_revoke.dspy", # CRUD 管理页 f"/{MOD}/documents_list/index.ui", f"/{MOD}/engine_configs_list/index.ui", diff --git a/wwwroot/api/doc_delete.dspy b/wwwroot/api/doc_delete.dspy new file mode 100644 index 0000000..5507efe --- /dev/null +++ b/wwwroot/api/doc_delete.dspy @@ -0,0 +1,12 @@ +# 对外 API：文件删除（级联清理向量/chunks/文件） +# POST /rag/api/doc_delete.dspy Authorization: Bearer +# body/query: {doc_id} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "doc.delete"): + return C._err("api key lacks scope: doc.delete", code="forbidden") +env = C.make_api_env(_ctx) +return await C.doc_delete(env, _ns) diff --git a/wwwroot/api/doc_set_tags.dspy b/wwwroot/api/doc_set_tags.dspy new file mode 100644 index 0000000..3008121 --- /dev/null +++ b/wwwroot/api/doc_set_tags.dspy @@ -0,0 +1,12 @@ +# 对外 API：文件设置标签（全量语义，标签名不存在自动创建） +# POST /rag/api/doc_set_tags.dspy Authorization: Bearer +# body/query: {kb_id, doc_id, tags:[名称...] 或 "a,b" | tag_ids:[id...] 或 "id1,id2"} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "tag.write"): + return C._err("api key lacks scope: tag.write", code="forbidden") +env = C.make_api_env(_ctx) +return await C.doc_set_tags(env, _ns) diff --git a/wwwroot/api/doc_upload.dspy b/wwwroot/api/doc_upload.dspy new file mode 100644 index 0000000..23a6b11 --- /dev/null +++ b/wwwroot/api/doc_upload.dspy @@ -0,0 +1,14 @@ +# 对外 API：文件上传（原始字节 body） +# POST /rag/api/doc_upload.dspy?kb_id=X&file_name=Y Authorization: Bearer +# Content-Type: application/octet-stream，请求体即文件原始字节 +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +if not C.require_scope(_ctx, "doc.upload"): + return C._err("api key lacks scope: doc.upload", code="forbidden") +_ns = dict(params_kw or {}) +_file_data = await request.read() +_file_name = _ns.get('file_name', '') or 'upload.bin' +env = C.make_api_env(_ctx) +return await C.doc_upload(env, _ns, _file_data, _file_name) diff --git a/wwwroot/api/kb_create.dspy b/wwwroot/api/kb_create.dspy new file mode 100644 index 0000000..45359b9 --- /dev/null +++ b/wwwroot/api/kb_create.dspy @@ -0,0 +1,12 @@ +# 对外 API：创建知识库 +# POST /rag/api/kb_create.dspy Authorization: Bearer +# body(JSON 或 query): {name, description?, embedding_engine?} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "kb.create"): + return C._err("api key lacks scope: kb.create", code="forbidden") +env = C.make_api_env(_ctx) +return await C.kb_create(env, _ns) diff --git a/wwwroot/api/kb_delete.dspy b/wwwroot/api/kb_delete.dspy new file mode 100644 index 0000000..ee9f7b2 --- /dev/null +++ b/wwwroot/api/kb_delete.dspy @@ -0,0 +1,12 @@ +# 对外 API：删除知识库（全量级联清理） +# POST /rag/api/kb_delete.dspy Authorization: Bearer +# body/query: {kb_id} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "kb.delete"): + return C._err("api key lacks scope: kb.delete", code="forbidden") +env = C.make_api_env(_ctx) +return await C.kb_delete(env, _ns) diff --git a/wwwroot/api/search.dspy b/wwwroot/api/search.dspy new file mode 100644 index 0000000..13ebe20 --- /dev/null +++ b/wwwroot/api/search.dspy @@ -0,0 +1,12 @@ +# 对外 API：知识库检索（向量召回 + 重排；kb_id 缺省=本机构全部知识库） +# POST /rag/api/search.dspy Authorization: Bearer +# body/query: {query, kb_id?, top_k?(默认10), recall_k?(默认top_k*3)} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "search"): + return C._err("api key lacks scope: search", code="forbidden") +env = C.make_api_env(_ctx) +return await C.search(env, _ns) diff --git a/wwwroot/api/tag_create.dspy b/wwwroot/api/tag_create.dspy new file mode 100644 index 0000000..240c9fe --- /dev/null +++ b/wwwroot/api/tag_create.dspy @@ -0,0 +1,12 @@ +# 对外 API：创建标签（同名幂等） +# POST /rag/api/tag_create.dspy Authorization: Bearer +# body/query: {kb_id, name, color?} +from rag import api_core as C +_ctx, _e = await C.verify_api_key(request) +if _e: + return C._err(_e, code="unauthorized") +_ns = await C.read_json_body(request, params_kw) +if not C.require_scope(_ctx, "tag.write"): + return C._err("api key lacks scope: tag.write", code="forbidden") +env = C.make_api_env(_ctx) +return await C.tag_create(env, _ns) diff --git a/wwwroot/knowledge_bases_list/api_key_create.dspy b/wwwroot/knowledge_bases_list/api_key_create.dspy new file mode 100644 index 0000000..0632568 --- /dev/null +++ b/wwwroot/knowledge_bases_list/api_key_create.dspy @@ -0,0 +1,42 @@ +# 管理端：创建对外 API Key（登录用户；org 取自会话） +# POST /rag/knowledge_bases_list/api_key_create.dspy +# params: {name, scopes?, expires_days?} +# 返回 widget：一次性展示明文 key（库里只存 SHA256） +import secrets +from rag import api_core as C + +ns = params_kw.copy() +env = request._run_ns +org_id = await env.get_userorgid() +name = (ns.get('name') or '默认密钥').strip() +scopes = ns.get('scopes') or '' +if isinstance(scopes, (list, tuple)): + scopes = ','.join(str(x) for x in scopes) +scopes = str(scopes).strip() or 'kb.create,kb.delete,doc.upload,doc.delete,tag.write,search' +try: + exp_days = int(ns.get('expires_days') or 0) +except (TypeError, ValueError): + exp_days = 0 + +plain = 'rag-' + secrets.token_hex(24) +kid = str(uuid()).replace('-', '')[:16] +db = DBPools() +async with db.sqlorContext(get_module_dbname('rag')) as sor: + if exp_days > 0: + await sor.sqlExe( + "INSERT INTO rag_api_keys (id, org_id, name, key_hash, prefix, scopes, status, expires_at, created_at) " + "VALUES (${id}$, ${o}$, ${n}$, ${h}$, ${p}$, ${s}$, 'active', DATE_ADD(NOW(), INTERVAL ${d}$ DAY), NOW())", + {"id": kid, "o": org_id, "n": name, "h": C.hash_key(plain), + "p": plain[:12], "s": scopes, "d": exp_days}) + else: + await sor.sqlExe( + "INSERT INTO rag_api_keys (id, org_id, name, key_hash, prefix, scopes, status, created_at) " + "VALUES (${id}$, ${o}$, ${n}$, ${h}$, ${p}$, ${s}$, 'active', NOW())", + {"id": kid, "o": org_id, "n": name, "h": C.hash_key(plain), "p": plain[:12], "s": scopes}) + await sor.sqlExe("COMMIT", {}) + +return {"widgettype": "VBox", "options": {"padding": "12px", "spacing": "8px"}, "subwidgets": [ + {"widgettype": "Text", "options": {"text": "✅ API Key 已创建（仅此一次展示，请立即保存）", "cfontsize": 15, "color": "#10b981"}}, + {"widgettype": "Text", "options": {"text": plain, "cfontsize": 13, "bgcolor": "#f5f7fa", "padding": "8px", "css": "selectable monospace"}}, + {"widgettype": "Text", "options": {"text": "名称: " + name + " · 范围: " + scopes, "cfontsize": 12, "color": "#888"}}, +]} diff --git a/wwwroot/knowledge_bases_list/api_key_list.dspy b/wwwroot/knowledge_bases_list/api_key_list.dspy new file mode 100644 index 0000000..3babf50 --- /dev/null +++ b/wwwroot/knowledge_bases_list/api_key_list.dspy @@ -0,0 +1,35 @@ +# 管理端：API Key 列表（本机构；只展示前缀，不回显明文） +# GET /rag/knowledge_bases_list/api_key_list.dspy +ns = params_kw.copy() +env = request._run_ns +org_id = await env.get_userorgid() +rows = [] +db = DBPools() +async with db.sqlorContext(get_module_dbname('rag')) as sor: + recs = await sor.sqlExe( + "SELECT id, name, prefix, scopes, status, last_used_at, expires_at, created_at " + "FROM rag_api_keys WHERE org_id=${o}$ ORDER BY created_at DESC", {"o": org_id}) + await sor.sqlExe("COMMIT", {}) + for r in (recs or []): + st = getattr(r, 'status', '') + badge = '🟢' if st == 'active' else '⛔' + exp = str(getattr(r, 'expires_at', '') or '') + exp_s = ('至 ' + exp[:10]) if exp and exp != 'None' else '永久' + used = str(getattr(r, 'last_used_at', '') or '') + used_s = used[:16] if used and used != 'None' else '从未使用' + rows.append({"widgettype": "HBox", "options": {"spacing": "12px", "padding": "8px 12px", "borderBottom": "1px solid #eee"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": badge + " " + str(getattr(r, 'name', '') or ''), "cfontsize": 13}}, + {"widgettype": "Text", "options": {"text": str(getattr(r, 'prefix', '')) + '…', "cfontsize": 12, "color": "#888"}}, + {"widgettype": "Text", "options": {"text": str(getattr(r, 'scopes', '') or ''), "cfontsize": 11, "color": "#aaa"}}, + {"widgettype": "Text", "options": {"text": exp_s + " · " + used_s, "cfontsize": 11, "color": "#aaa"}}, + {"widgettype": "Button", "options": {"label": ("禁用" if st == 'active' else "启用"), "cfontsize": 11}, + "binds": [{"wid": "self", "event": "click", "actiontype": "ajax", + "options": {"url": entire_url('./api_key_revoke.dspy'), + "params": {"key_id": str(getattr(r, 'id', '')), + "action": ("disable" if st == 'active' else "enable")}, + "success": "refresh"}}]}, + ]}) +if not rows: + rows = [{"widgettype": "Text", "options": {"text": "暂无 API Key", "color": "#aaa"}}] +return {"widgettype": "VBox", "options": {"width": "100%"}, "subwidgets": rows} diff --git a/wwwroot/knowledge_bases_list/api_key_revoke.dspy b/wwwroot/knowledge_bases_list/api_key_revoke.dspy new file mode 100644 index 0000000..fe0a679 --- /dev/null +++ b/wwwroot/knowledge_bases_list/api_key_revoke.dspy @@ -0,0 +1,24 @@ +# 管理端：启用/禁用 API Key（只能操作本机构） +# POST /rag/knowledge_bases_list/api_key_revoke.dspy params: {key_id, action=disable|enable|delete} +ns = params_kw.copy() +env = request._run_ns +org_id = await env.get_userorgid() +key_id = (ns.get('key_id') or '').strip() +action = (ns.get('action') or 'disable').strip() +if not key_id: + return json.dumps({"status": "error", "error": "key_id required"}) +db = DBPools() +async with db.sqlorContext(get_module_dbname('rag')) as sor: + recs = await sor.sqlExe( + "SELECT id FROM rag_api_keys WHERE id=${k}$ AND org_id=${o}$", {"k": key_id, "o": org_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return json.dumps({"status": "error", "error": "key not found in your org"}) + if action == 'delete': + await sor.sqlExe("DELETE FROM rag_api_keys WHERE id=${k}$", {"k": key_id}) + elif action == 'enable': + await sor.sqlExe("UPDATE rag_api_keys SET status='active' WHERE id=${k}$", {"k": key_id}) + else: + await sor.sqlExe("UPDATE rag_api_keys SET status='disabled' WHERE id=${k}$", {"k": key_id}) + await sor.sqlExe("COMMIT", {}) +return json.dumps({"status": "SUCCEEDED", "action": action, "key_id": key_id}) diff --git a/wwwroot/knowledge_bases_list/upload_file.dspy b/wwwroot/knowledge_bases_list/upload_file.dspy index a5c0d3e..efafe83 100644 --- a/wwwroot/knowledge_bases_list/upload_file.dspy +++ b/wwwroot/knowledge_bases_list/upload_file.dspy @@ -39,339 +39,9 @@ doc_id = str(uuid()).replace('-', '')[:16] ext = '.' + file_name.rsplit('.', 1)[1] if '.' in file_name else '.bin' ext_l = ext.lower() -# ============================================================ -# BACKGROUND INGESTION — runs in a separate asyncio task after -# the response is returned (background_reco = create_task wrapper). -# SELF-CONTAINED: never touches env/request (invalid after response), -# creates its own DBPools connection. All args are primitives. -# ============================================================ -async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path): - db = DBPools() - # 库名必须经宿主的 get_module_dbname 映射（ragserver→'rag'，pipeline-app→'pipeline'） - # 后台任务无 request/env，用注入的全局函数解析；禁硬编码库名 - _dbname = get_module_dbname('rag') - - # 读知识库向量引擎： - # 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: - krecs = await sor.sqlExe("SELECT embedding_engine FROM rag_knowledge_bases WHERE id=${kb_id}$", {"kb_id": kb_id}) - if krecs: - 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' - else: - 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": 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 - face_count = 0 - voice_speakers = 0 - meta_parts = {} - try: - with open(real_path, 'rb') as f: - file_data = f.read() - - text_exts = {'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm', '.py', '.js', '.css', '.yaml', '.yml', '.log', '.rst'} - image_exts = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'} - audio_exts = {'.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac'} - video_exts = {'.mp4', '.avi', '.mov', '.mkv', '.webm'} - - # --- OFFICE DOCS: text extraction (PDF/DOCX/PPTX/XLSX) --- - if ext_l == '.pdf' and not text: - import io; from PyPDF2 import PdfReader - reader = PdfReader(io.BytesIO(file_data)) - text = '\n'.join(p.extract_text() or '' for p in reader.pages) - elif ext_l == '.docx' and not text: - import io; from docx import Document - doc = Document(io.BytesIO(file_data)) - text = '\n'.join(p.text for p in doc.paragraphs) - elif ext_l == '.pptx' and not text: - import io; from pptx import Presentation - prs = Presentation(io.BytesIO(file_data)) - parts = [] - for slide in prs.slides: - for shape in slide.shapes: - if hasattr(shape, 'text') and shape.text: - parts.append(shape.text) - text = '\n'.join(parts) - elif ext_l == '.xlsx' and not text: - import io; from openpyxl import load_workbook - wb = load_workbook(io.BytesIO(file_data), data_only=True) - parts = [] - for sheet in wb.worksheets: - for row in sheet.iter_rows(values_only=True): - parts.append('\t'.join(str(c or '') for c in row)) - text = '\n'.join(parts) - - # --- TEXT EXTRACTION --- - if ext_l in text_exts: - text = file_data.decode('utf-8', errors='replace') - - # --- 文本知识库不支持媒体文件 --- - if is_text and (ext_l in image_exts or ext_l in audio_exts or ext_l in video_exts): - try: - async with db.sqlorContext(_dbname) as sor: - await sor.sqlExe( - "UPDATE rag_documents SET status='failed', metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", - {"id": doc_id, "meta": json.dumps({"error": "文本知识库不支持媒体文件，请上传文本类文件(txt/md/pdf/docx等)或改用多媒体知识库"}, ensure_ascii=False)}) - except: pass - return - - # --- IMAGE: face detection + 图片向量化入库 --- - if ext_l in image_exts: - img_b64 = base64.b64encode(file_data).decode() - try: - client = StreamHttpClient() - resp = await client.request('POST', 'https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) - fd = json.loads(resp) - results = fd.get("results", []) - if results and isinstance(results[0], dict): - faces = results[0].get("faces", results[0].get("detections", [])) - face_count = len(faces) - if faces and isinstance(faces[0], dict): - 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: - try: - client = StreamHttpClient() - resp = await client.request('POST', 'https://media.opencomputing.net:10443/voiceprint/extract/submit', - files={'file': (file_name, file_data)}) - vd = json.loads(resp) - voice_speakers = vd.get('speakers', 1) if vd.get('status') == 'SUCCEEDED' else (1 if vd.get('embedding') else 0) - meta_parts['voiceprint'] = voice_speakers - except: pass - - # --- VIDEO: frame extraction + voiceprint --- - if ext_l in video_exts: - meta_parts['video'] = 'pending' - video_ok = False - try: - tmp_img = '/tmp/' + doc_id + '_frame.jpg' - subprocess.run(['ffmpeg', '-y', '-i', real_path, '-vframes', '1', '-q:v', '2', tmp_img], - capture_output=True, timeout=30) - if os.path.exists(tmp_img): - with open(tmp_img, 'rb') as fi: - frame_data = fi.read() - img_b64 = base64.b64encode(frame_data).decode() - frame_bboxes = [] - try: - client = StreamHttpClient() - resp = await client.request('POST', 'https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) - fd = json.loads(resp) - results = fd.get("results", []) - if results and isinstance(results[0], dict): - faces = results[0].get("faces", results[0].get("detections", [])) - face_count = len(faces) - frame_bboxes = [f.get("bbox", {}) for f in faces[:10]] if faces else [] - except: pass - # --- 视频帧向量化（vl在线 或 GPU CLIP） --- - try: - frame_vec = await _media_embed_image(img_b64) - img_embeddings = [frame_vec] if frame_vec else [] - except: - img_embeddings = [] - if img_embeddings: - try: - client3 = StreamHttpClient() - await ensure_vdb_collection(client3, kb_id) - vdb_data = {"colname": kb_id, "data": [ - {"id": doc_id + "_c0", "vector": img_embeddings[0], "text": file_name} - ]} - await client3.request('POST', VDB_BASE + '/v1/upsert', json=vdb_data) - chunk_meta = {"start_time": 0} - if frame_bboxes: - chunk_meta["bboxes"] = frame_bboxes - 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, metadata, created_at) " - "VALUES (${id}$, ${doc_id}$, ${kb_id}$, 0, ${content}$, ${vid}$, ${meta}$, NOW())", - {"id": doc_id + "_c0", "doc_id": doc_id, "kb_id": kb_id, - "content": file_name, "vid": doc_id + "_c0", - "meta": json.dumps(chunk_meta, ensure_ascii=False)}) - except: - pass - os.remove(tmp_img) - video_ok = True - except: pass - - # --- Voiceprint: extract audio from video --- - if video_ok: - try: - tmp_wav = '/tmp/' + doc_id + '_audio.wav' - subprocess.run(['ffmpeg', '-y', '-i', real_path, '-vn', '-acodec', 'pcm_s16le', - '-ar', '16000', '-ac', '1', tmp_wav], - capture_output=True, timeout=60) - if os.path.exists(tmp_wav) and os.path.getsize(tmp_wav) > 1000: - with open(tmp_wav, 'rb') as fa: - audio_data = fa.read() - try: - client4 = StreamHttpClient() - resp4 = await client4.request('POST', - 'https://media.opencomputing.net:10443/voiceprint/extract/submit', - files={'file': (file_name.rsplit('.', 1)[0] + '.wav', audio_data)}) - vd = json.loads(resp4) - voice_speakers = vd.get('speakers', 1) if vd.get('status') == 'SUCCEEDED' else (1 if vd.get('embedding') else 0) - meta_parts['voiceprint'] = voice_speakers - except: pass - if os.path.exists(tmp_wav): - os.remove(tmp_wav) - except: pass - - if video_ok: - meta_parts['video'] = 'done' - - # --- RAG INGEST for text --- - if text and len(text.strip()) > 10: - paragraphs = text.split('\n') - chunks = [] - cur = '' - for p in paragraphs: - p = p.strip() - if not p: - if cur: chunks.append(cur); cur = '' - continue - if len(cur) + len(p) < 500: - cur = (cur + '\n' + p).strip() - else: - if cur: chunks.append(cur) - cur = p - if cur: chunks.append(cur) - - if chunks: - # 在线文本向量化（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: - try: - client2 = StreamHttpClient() - await ensure_vdb_collection(client2, kb_id) - 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', 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: - pass - - async with db.sqlorContext(_dbname) as sor: - for i, chunk_text in enumerate(chunks): - vid = vector_ids[i] if i < len(vector_ids) else '' - 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}$, ${idx}$, ${content}$, ${vid}$, NOW())", - {"id": doc_id + "_c" + str(i), "doc_id": doc_id, "kb_id": kb_id, - "idx": i, "content": chunk_text[:2000], "vid": vid}) - chunks_n = len(chunks) - except Exception as e: - try: - async with db.sqlorContext(_dbname) as sor: - await sor.sqlExe( - "UPDATE rag_documents SET status='failed', metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", - {"id": doc_id, "meta": json.dumps({"error": str(e)[:300]}, ensure_ascii=False)}) - except: pass - return - - # --- finalize: mark document done + update KB chunk counts --- - meta_json = json.dumps(meta_parts, ensure_ascii=False) - try: - async with db.sqlorContext(_dbname) as sor: - await sor.sqlExe( - "UPDATE rag_documents SET status='done', chunk_count=${chunks}$, metadata=${meta}$, updated_at=NOW() WHERE id=${id}$", - {"id": doc_id, "chunks": chunks_n, "meta": meta_json}) - if chunks_n: - await sor.sqlExe( - "UPDATE rag_knowledge_bases SET chunk_count=chunk_count+${n}$ WHERE id=${kb_id}$", - {"n": chunks_n, "kb_id": kb_id}) - except: pass - -# ============================================================ +# ── 后台入库：共享模块 rag.ingest（UI 上传与对外 API 上传同一实现）── +# background_reco 派发 asyncio 任务；ingest_one 只接原始类型参数、自建连接。 +from rag.ingest import ingest_one # SYNC PART — record document as 'pending', update KB counts, # fire background ingestion, return immediately. # ============================================================ @@ -386,7 +56,7 @@ async with get_sor_context(env, 'rag') as sor: {"size": file_size, "kb_id": kb_id}) # Fire background ingestion — pass primitives only (no env/request/proxy objects) -background_reco(ingest_doc, doc_id, kb_id, file_name, ext_l, real_path) +background_reco(ingest_one, doc_id, kb_id, file_name, ext_l, real_path) result = {"status": "SUCCEEDED", "doc_id": doc_id, "file_name": file_name, "file_size": file_size, "folder_id": folder_id, "ingest": "pending"}