From abba9d1273ff67eb6f84368e03b25ff1a44d8a4c Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 22 Jul 2026 15:25:42 +0800
Subject: [PATCH] fix: use standalone get_sor_context(env, modulename)

---
 init.py | 79 ++++++++++++++++++++-------------------------------------
 1 file changed, 28 insertions(+), 51 deletions(-)

diff --git a/init.py b/init.py
index 742f47b..269aaf2 100644
--- a/init.py
+++ b/init.py
@@ -6,12 +6,12 @@ from traceback import format_exc
 from ahserver.serverenv import ServerEnv
 from appPublic.registerfunction import RegisterFunction
 from appPublic.log import debug, exception
+from sqlor.dbpools import get_sor_context
 import json
 
 
 async def status_handler(request, params_kw, *args, **kwargs):
     """服务状态"""
-    env = request._run_ns
     return json.dumps({
         "service": "ragserver",
         "version": "0.1.0",
@@ -27,12 +27,12 @@ async def kb_list_handler(request, params_kw, *args, **kwargs):
     env = request._run_ns
     try:
         userorgid = await env.get_userorgid()
-        sor = await env.get_sor_context('rag')
-        recs = await sor.R("knowledge_bases", {"org_id": userorgid})
-        return json.dumps({
-            "status": "SUCCEEDED",
-            "knowledge_bases": [dict(r) for r in recs]
-        }, ensure_ascii=False, default=str)
+        async with get_sor_context(env, 'rag') as sor:
+            recs = await sor.R("knowledge_bases", {"org_id": userorgid})
+            return json.dumps({
+                "status": "SUCCEEDED",
+                "knowledge_bases": [dict(r) for r in recs]
+            }, ensure_ascii=False, default=str)
     except Exception as e:
         exception(f"kb_list: {e}, {format_exc()}")
         return json.dumps({"error": str(e)})
@@ -43,18 +43,17 @@ async def engines_handler(request, params_kw, *args, **kwargs):
     env = request._run_ns
     try:
         userorgid = await env.get_userorgid()
-        sor = await env.get_sor_context('rag')
-        # 返回全局默认 + 租户自定义，租户优先
-        sql = """
-            SELECT * FROM engine_configs 
-            WHERE status='active' AND (org_id IS NULL OR org_id=%s)
-            ORDER BY engine_type, priority DESC
-        """
-        recs = await sor.sqlExe(sql, {"org_id": userorgid})
-        return json.dumps({
-            "status": "SUCCEEDED",
-            "engines": [dict(r) for r in recs]
-        }, ensure_ascii=False, default=str)
+        async with get_sor_context(env, 'rag') as sor:
+            sql = """
+                SELECT * FROM engine_configs 
+                WHERE status='active' AND (org_id IS NULL OR org_id=%s)
+                ORDER BY engine_type, priority DESC
+            """
+            recs = await sor.sqlExe(sql, {"org_id": userorgid})
+            return json.dumps({
+                "status": "SUCCEEDED",
+                "engines": [dict(r) for r in recs]
+            }, ensure_ascii=False, default=str)
     except Exception as e:
         exception(f"engines: {e}, {format_exc()}")
         return json.dumps({"error": str(e)})
@@ -67,60 +66,38 @@ async def search_handler(request, params_kw, *args, **kwargs):
         query = params_kw.get("query", "")
         kb_id = params_kw.get("kb_id", "")
         top_k = int(params_kw.get("top_k", 5))
-
         if not query:
             return json.dumps({"error": "query required"})
-
-        # 尝试调用 rag-pipeline 引擎
         try:
             from pipeline import search as pipeline_search
-            result = pipeline_search(
-                query,
-                pipeline_name=params_kw.get("pipeline", "kg-rag-standard"),
-                collection=kb_id or "knowledge",
-                graph_name=kb_id or "knowledge",
-                top_k=top_k,
-                llm_func=None
-            )
+            result = pipeline_search(query, pipeline_name="kg-rag-standard",
+                                     collection=kb_id or "knowledge",
+                                     graph_name=kb_id or "knowledge",
+                                     top_k=top_k, llm_func=None)
             return json.dumps(result, ensure_ascii=False)
         except ImportError:
-            return json.dumps({
-                "status": "FALLBACK",
-                "message": "rag-pipeline not available, search via direct API calls",
-                "query": query
-            }, ensure_ascii=False)
-
+            return json.dumps({"status": "FALLBACK", "message": "pipeline not available"})
     except Exception as e:
         exception(f"search: {e}, {format_exc()}")
         return json.dumps({"error": str(e)})
 
 
 async def ingest_handler(request, params_kw, *args, **kwargs):
-    """文档入库 (文本直接调用 pipeline)"""
+    """文档入库"""
     env = request._run_ns
     try:
         document = params_kw.get("document", "")
         kb_id = params_kw.get("kb_id", "")
-
         if not document:
             return json.dumps({"error": "document text required"})
-
         try:
             from pipeline import ingest as pipeline_ingest
-            result = pipeline_ingest(
-                document,
-                pipeline_name=params_kw.get("pipeline", "kg-rag-standard"),
-                collection=kb_id or "knowledge",
-                graph_name=kb_id or "knowledge",
-                llm_func=None
-            )
+            result = pipeline_ingest(document, pipeline_name="kg-rag-standard",
+                                     collection=kb_id or "knowledge",
+                                     graph_name=kb_id or "knowledge", llm_func=None)
             return json.dumps(result, ensure_ascii=False)
         except ImportError:
-            return json.dumps({
-                "status": "FALLBACK",
-                "message": "rag-pipeline not available for ingest"
-            }, ensure_ascii=False)
-
+            return json.dumps({"status": "FALLBACK", "message": "pipeline not available"})
     except Exception as e:
         exception(f"ingest: {e}, {format_exc()}")
         return json.dumps({"error": str(e)})