fix(host): 后台入库任务禁硬编码库名,改经 get_module_dbname 映射

ingest_doc 是 background_reco 后台任务(无 request/env),原用 db.sqlorContext('rag') 6 处。
在 ragserver 上库名恰为 rag 所以能跑,接入 pipeline-app(库名 pipeline)后
sqlorFactory 报 NoneType.get → 入库全程失败,status 永久 pending、向量丢失。
改为函数开头 _dbname = get_module_dbname('rag') 统一映射。
This commit is contained in:
ymq 2026-08-25 15:45:44 +08:00
parent 25c7ad357c
commit 51c9968093

View File

@ -47,11 +47,14 @@ ext_l = ext.lower()
# ============================================================
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=文本(走 /txte)clip-vith14=多媒体(走 /mme)
emb_engine = 'clip-vith14'
try:
async with db.sqlorContext('rag') as sor:
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()
@ -116,7 +119,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path):
# --- 文本知识库不支持媒体文件 ---
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('rag') as sor:
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)})
@ -193,7 +196,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path):
chunk_meta = {"start_time": 0}
if frame_bboxes:
chunk_meta["bboxes"] = frame_bboxes
async with db.sqlorContext('rag') as sor:
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())",
@ -273,7 +276,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path):
except:
pass
async with db.sqlorContext('rag') as sor:
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(
@ -284,7 +287,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path):
chunks_n = len(chunks)
except Exception as e:
try:
async with db.sqlorContext('rag') as sor:
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)})
@ -294,7 +297,7 @@ async def ingest_doc(doc_id, kb_id, file_name, ext_l, real_path):
# --- finalize: mark document done + update KB chunk counts ---
meta_json = json.dumps(meta_parts, ensure_ascii=False)
try:
async with db.sqlorContext('rag') as sor:
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})