feat(llm): llm表查询按org_id多租户隔离(本机构+系统级)，角色agent与会话agent模型选择统一(RoleSpec.model_name→产线default_model→deepseek-v4-pro)

pipeline_service/agent_loop.py

@@ -96,6 +96,53 @@ async def _resolve_role(project_id, role):
     return norm, ROLE_SPECIFICS.get(norm, ROLE_SPECIFICS.get('develop', '')), ROLE_CHAIN.get(norm)
 
 
+async def _resolve_llm_context(sor, project_id, role, model_name=None):
+    """解析角色 agent 的 LLM 上下文：返回 (model_name, org_id)。
+
+    model_name 解析链（与会话 agent 的 load_agent_config 一致）：
+      1. 显式传入的 model_name
+      2. RoleSpec.model_name（角色专属模型）
+      3. 产线 default_model（pipelines.default_model）
+      4. 全局默认 "deepseek-v4-pro"
+
+    org_id：从 sd_projects.org_id 读，传给 llm_bridge 做多租户隔离
+    （llm 表查询只取「本机构 + 系统级 org_id='0'」的模型）。
+    """
+    org_id = ""
+    pipeline_id = ""
+    try:
+        recs = await sor.sqlExe(
+            "SELECT org_id, pipeline_id FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
+        if recs:
+            org_id = getattr(recs[0], "org_id", "") or ""
+            pipeline_id = getattr(recs[0], "pipeline_id", "") or ""
+    except Exception:
+        pass
+
+    if not model_name:
+        # 2. RoleSpec.model_name（角色专属模型）
+        try:
+            from pipeline_core import get_role_spec
+            spec = get_role_spec(pipeline_id or await _resolve_pipeline_id(project_id), role)
+            if spec and spec.model_name:
+                model_name = spec.model_name
+        except Exception:
+            pass
+    if not model_name and pipeline_id:
+        # 3. 产线 default_model
+        try:
+            recs = await sor.sqlExe(
+                "SELECT default_model FROM pipelines WHERE id=${pid}$", {"pid": pipeline_id})
+            if recs:
+                model_name = getattr(recs[0], "default_model", "") or ""
+        except Exception:
+            pass
+    if not model_name:
+        # 4. 全局默认
+        model_name = "deepseek-v4-pro"
+    return model_name, org_id
+
+
 def _get_db():
     from sqlor.dbpools import DBPools
     db = DBPools()
@@ -703,6 +750,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
 
     db = _get_db()
     async with db.sqlorContext("pipeline") as sor:
+        # 解析 LLM 上下文（model 一致性 + org_id 多租户隔离）
+        model_name, org_id = await _resolve_llm_context(sor, project_id, role, model_name)
         task = await _claim_task(sor, project_id, role)
         if not task:
             return {"status": "idle", "message": "没有待办任务"}
@@ -757,7 +806,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
                 {"tid": task_id})
             await sor.sqlExe("COMMIT", {})
             try:
-                resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4)
+                resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4, org_id=org_id)
             except Exception as e:
                 err_msg = f"{type(e).__name__}: {str(e)[:400]}"
                 from .task_capability import mark_failed
@@ -897,6 +946,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
 async def pm_review_run(project_id, agent_id=None, model_name=None):
     db = _get_db()
     async with db.sqlorContext("pipeline") as sor:
+        # PM 审核用产线 default_model（无角色专属模型），org_id 多租户隔离
+        model_name, org_id = await _resolve_llm_context(sor, project_id, 'pm', model_name)
         task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False, set_state='review')
         if not task:
             return {"status": "idle", "message": "没有待审核任务"}
@@ -951,7 +1002,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                     "只能是 review_approve / review_reject / review_complete 三者之一，"
                     "禁止再调用任何工具。"})
             try:
-                raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3)
+                raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3, org_id=org_id)
             except Exception as e:
                 err_msg = f"{type(e).__name__}: {str(e)[:400]}"
                 from .task_capability import mark_failed

pipeline_service/agent_loop_v2.py

@@ -376,6 +376,7 @@ class AgentExecutor:
                 [{"role": "user", "content": prompt}],
                 model=self.model_name,
                 temperature=0,
+                org_id=self.org_id,
             )
             m = _re.search(r"\[[^\]]*\]", content or "")
             if m:
@@ -480,6 +481,7 @@ class AgentExecutor:
                     tools=tools_schema,
                     model=self.model_name,
                     temperature=self.config.temperature,
+                    org_id=self.org_id,
                 )
             except Exception as e:
                 logger.error(f"native function calling failed, fallback to text: {e}")
@@ -490,6 +492,7 @@ class AgentExecutor:
             self._msgs,
             model=self.model_name,
             temperature=self.config.temperature,
+            org_id=self.org_id,
         )
         return {"content": content or "", "tool_calls": []}
 
@@ -749,7 +752,7 @@ class AgentExecutor:
                 "SELECT name FROM sd_projects ORDER BY created_at DESC LIMIT 20", {})
             pnames = [getattr(r, "name", "") for r in (all_recs or [])]
             classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。"
-            matched = await llm_call(classify_prompt, temperature=0.0)
+            matched = await llm_call(classify_prompt, temperature=0.0, org_id=self.org_id)
             matched = matched.strip().strip('"').strip("'")
 
             recs2 = await sor.sqlExe(
@@ -1113,6 +1116,7 @@ class AgentExecutor:
             summary = await llm_call(
                 f"请用3-5句话总结以下对话的关键信息：\n\n{text[:4000]}",
                 temperature=0.1,
+                org_id=self.org_id,
             )
             return summary[:500]
         except Exception:

pipeline_service/llm_bridge.py

@@ -30,12 +30,17 @@ def _decrypt_key(encrypted: str) -> str:
         return encrypted  # already plaintext or decrypt failed
 
 
-async def _get_model_config(model_name: str = None) -> dict:
+async def _get_model_config(model_name: str = None, org_id: str = None) -> dict:
-    """Look up model config from llm table. Returns dict with api_base, api_key, model_id."""
+    """Look up model config from llm table. Returns dict with api_base, api_key, model_id.
+
+    org_id 多租户隔离：非空时只取「本机构 + 系统级(org_id='0')」的模型，
+    且本机构模型优先（ORDER BY (org_id='0') 让系统级排后）；org_id 为空时不过滤（向后兼容）。
+    """
     global _model_cache
 
-    if model_name and model_name in _model_cache:
+    cache_key = f"{model_name or ''}:{org_id or ''}"
-        return _model_cache[model_name]
+    if model_name and cache_key in _model_cache:
+        return _model_cache[cache_key]
 
     try:
         from sqlor.dbpools import DBPools
@@ -43,11 +48,21 @@ async def _get_model_config(model_name: str = None) -> dict:
         dbname = "pipeline"
         async with db.sqlorContext(dbname) as sor:
             if model_name:
-                sql = "SELECT api_base, api_key, model_id FROM llm WHERE name=${name}$ AND status='active' LIMIT 1"
+                sql = "SELECT api_base, api_key, model_id FROM llm WHERE name=${name}$ AND status='active'"
-                recs = await sor.sqlExe(sql, {"name": model_name})
+                params = {"name": model_name}
+                if org_id:
+                    sql += " AND (org_id=${org}$ OR org_id='0')"
+                    params["org"] = org_id
+                sql += " ORDER BY (org_id='0') ASC LIMIT 1"
+                recs = await sor.sqlExe(sql, params)
             else:
-                sql = "SELECT api_base, api_key, model_id, name FROM llm WHERE status='active' ORDER BY id LIMIT 1"
+                sql = "SELECT api_base, api_key, model_id, name FROM llm WHERE status='active'"
-                recs = await sor.sqlExe(sql, {})
+                params = {}
+                if org_id:
+                    sql += " AND (org_id=${org}$ OR org_id='0')"
+                    params["org"] = org_id
+                sql += " ORDER BY (org_id='0') ASC, id LIMIT 1"
+                recs = await sor.sqlExe(sql, params)
             if recs:
                 r = recs[0]
                 cfg = {
@@ -55,9 +70,7 @@ async def _get_model_config(model_name: str = None) -> dict:
                     "api_key": _decrypt_key(getattr(r, "api_key", "") or ""),
                     "model_id": getattr(r, "model_id", "") or "",
                 }
-                cache_key = model_name or getattr(r, "name", "")
+                _model_cache[cache_key] = cfg
-                if cache_key:
-                    _model_cache[cache_key] = cfg
                 return cfg
     except Exception as e:
         logger.warning("llm_bridge: DB lookup failed: %s", e)
@@ -107,13 +120,15 @@ async def _post_chat_completion(url: str, headers: dict, payload: dict) -> dict:
     raise last_exc if last_exc else ValueError("LLM call failed")
 
 
-async def llm_call(prompt: str, model: str = None, temperature: float = 0.7) -> str:
+async def llm_call(prompt: str, model: str = None, temperature: float = 0.7, org_id: str = None) -> str:
     """Call LLM and return text response.
 
     Backend priority:
     1. harnessed_agent.llm_chat (if loaded in ServerEnv)
     2. DB llm table (api_base + api_key)
     3. Environment variables (LLM_API_BASE, LLM_API_KEY, LLM_MODEL)
+
+    org_id 多租户隔离：非空时 llm 表查询只取「本机构 + 系统级」模型。
     """
     # Priority 1: harnessed_agent
     try:
@@ -128,7 +143,7 @@ async def llm_call(prompt: str, model: str = None, temperature: float = 0.7) ->
         pass
 
     # Priority 2: DB llm table
-    cfg = await _get_model_config(model)
+    cfg = await _get_model_config(model, org_id=org_id)
     if cfg.get("api_key") and cfg.get("api_base"):
         api_base = cfg["api_base"]
         api_key = cfg["api_key"]
@@ -165,11 +180,11 @@ async def call_llm(tenant_id: str, prompt: str, model: str = None, temperature:
     return await llm_call(prompt, model=model, temperature=temperature)
 
 
-async def llm_call_msgs(messages: list, model: str = None, temperature: float = 0.7) -> str:
+async def llm_call_msgs(messages: list, model: str = None, temperature: float = 0.7, org_id: str = None) -> str:
     """Call LLM with full message array (system/user/assistant)."""
     import aiohttp
 
-    cfg = await _get_model_config(model)
+    cfg = await _get_model_config(model, org_id=org_id)
     if cfg.get("api_key") and cfg.get("api_base"):
         api_base = cfg["api_base"]
         api_key = cfg["api_key"]
@@ -190,7 +205,7 @@ async def llm_call_msgs(messages: list, model: str = None, temperature: float =
     return data["choices"][0]["message"]["content"]
 
 
-async def llm_call_msgs_native(messages: list, tools: list = None, model: str = None, temperature: float = 0.7) -> dict:
+async def llm_call_msgs_native(messages: list, tools: list = None, model: str = None, temperature: float = 0.7, org_id: str = None) -> dict:
     """Native function calling. 传入 tools JSON schema，返回 message dict。
 
     Returns:
@@ -199,7 +214,7 @@ async def llm_call_msgs_native(messages: list, tools: list = None, model: str =
     """
     import aiohttp
 
-    cfg = await _get_model_config(model)
+    cfg = await _get_model_config(model, org_id=org_id)
     if cfg.get("api_key") and cfg.get("api_base"):
         api_base = cfg["api_base"]
         api_key = cfg["api_key"]