diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 1416686..50ba5a3 100644 --- a/pipeline_service/agent_loop.py +++ b/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 diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index fe8cb40..70797e6 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/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: diff --git a/pipeline_service/llm_bridge.py b/pipeline_service/llm_bridge.py index 6efc055..93d00b1 100644 --- a/pipeline_service/llm_bridge.py +++ b/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: - """Look up model config from llm table. Returns dict with api_base, api_key, model_id.""" +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. + + org_id 多租户隔离：非空时只取「本机构 + 系统级(org_id='0')」的模型， + 且本机构模型优先（ORDER BY (org_id='0') 让系统级排后）；org_id 为空时不过滤（向后兼容）。 + """ global _model_cache - if model_name and model_name in _model_cache: - return _model_cache[model_name] + cache_key = f"{model_name or ''}:{org_id or ''}" + 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" - recs = await sor.sqlExe(sql, {"name": model_name}) + sql = "SELECT api_base, api_key, model_id FROM llm WHERE name=${name}$ AND status='active'" + 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" - recs = await sor.sqlExe(sql, {}) + sql = "SELECT api_base, api_key, model_id, name FROM llm WHERE status='active'" + 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", "") - if cache_key: - _model_cache[cache_key] = cfg + _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"]