diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index 5826bde..ad3e5cf 100644 --- a/pipeline_service/gateway.py +++ b/pipeline_service/gateway.py @@ -76,12 +76,14 @@ class Gateway: # ── 上下文解析 ── async def resolve_project(self, user_id: str, session_id: str = "") -> dict: - """解析用户当前项目 → {pid, pipeline_id, name, default_llm_id, default_llm_name}。 + """解析用户当前项目 → {pid, pipeline_id, name, default_llm_id, default_llm_name, project_model_name}。 pid 按会话隔离（session_id 非空时读 pipeline_session_settings，无记录回退全局）； - 同时读出个人选择的模型（default_llm_id → llm.name，仍按用户全局），供 run_message 覆盖 config.model_name。 + 同时读出个人选择的模型（default_llm_id → llm.name，仍按用户全局）与项目级模型 + （sd_projects.default_model，产线通用、跨会话持久），供 run_message 按优先级覆盖。 """ - ctx = {"pid": "", "pipeline_id": "", "name": "", "default_llm_id": "", "default_llm_name": ""} + ctx = {"pid": "", "pipeline_id": "", "name": "", "default_llm_id": "", + "default_llm_name": "", "project_model_name": ""} if not user_id: return ctx try: @@ -106,11 +108,12 @@ class Gateway: ctx["default_llm_name"] = getattr(llm_recs[0], "name", "") or "" if ctx["pid"]: proj = await sor.sqlExe( - "SELECT name, pipeline_id FROM sd_projects WHERE id=${p}$", + "SELECT name, pipeline_id, default_model FROM sd_projects WHERE id=${p}$", {"p": ctx["pid"]}) if proj: ctx["name"] = getattr(proj[0], "name", "") ctx["pipeline_id"] = getattr(proj[0], "pipeline_id", "") or "" + ctx["project_model_name"] = getattr(proj[0], "default_model", "") or "" except Exception as e: logger.warning(f"gateway resolve_project failed: {e}") return ctx @@ -120,7 +123,7 @@ class Gateway: async def run_message(self, channel: str, user_id: str, content: str, history=None, role: str = "", generic: bool = False, base_url: str = "", session_id: str = "", - pipeline_id: str = "") -> AsyncGenerator[str, None]: + pipeline_id: str = "", model_id: str = "") -> AsyncGenerator[str, None]: """统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。 Args: @@ -133,19 +136,31 @@ class Gateway: session_id: 会话内唯一标识（web 多 tab 独立会话用；空 = 单会话） pipeline_id: 产线默认能力（会话入口指定，如 bidding_general/sdlc_general）。 无当前项目时用此产线装载能力，替代硬回退 DEFAULT_ABILITY_ID。 + model_id: 前端模型下拉选中的模型（llm.id 或 llm.model_id）。非空时： + ① 校验模型存在且可用 → 解析出 llm.name； + ② 持久化到当前项目（sd_projects.default_model，产线通用）—— + 项目模型一经设置即生效，直到用户再次选择，不因新会话/换 tab 丢失； + ③ 本条消息直接使用该模型。 + 空 = 未重新选择，沿用项目已设模型。 """ from pipeline_core.agent_config import load_agent_config from .agent_loop_v2 import AgentExecutor # 1. 解析项目上下文（纯通用模式跳过，不挂任何产线插件） if generic: - ctx = {"pid": "", "pipeline_id": "", "name": ""} + ctx = {"pid": "", "pipeline_id": "", "name": "", "project_model_name": ""} else: ctx = await self.resolve_project(user_id, session_id) # 无当前项目 → 用入口指定的默认产线（投标/开发产线各自独立入口的关键） if not ctx.get("pipeline_id") and pipeline_id: ctx["pipeline_id"] = pipeline_id + # 1.5 模型解析与项目级持久化（优先级：本次显式选择 > 项目已设 > 个人全局 > 产线默认） + chosen_model = "" + if model_id: + chosen_model = await self._resolve_and_persist_model( + user_id, ctx.get("pid", ""), model_id) + # 2. 会话生命周期（跟踪 project 切换；session_id 区分多 tab 会话） key = self._session_key(channel, user_id, session_id) sess = self._sessions.get(key) @@ -166,8 +181,12 @@ class Gateway: # 3. 加载产线能力 + 创建 executor（generic 时不装产线能力） config = await load_agent_config( pipeline_id=ctx["pipeline_id"], project_id=ctx["pid"], generic=generic) - # 个人选择的模型优先（default_llm_id → llm.name，覆盖产线 default_model） - if ctx.get("default_llm_name"): + # 模型覆盖优先级：本次选择 > 项目模型 > 个人全局选择（个人全局只兜底，不再压过项目设置） + if chosen_model: + config.model_name = chosen_model + elif ctx.get("project_model_name"): + config.model_name = ctx["project_model_name"] + elif ctx.get("default_llm_name"): config.model_name = ctx["default_llm_name"] executor = AgentExecutor( config=config, project_id=ctx["pid"], user_id=user_id, role=role, @@ -178,6 +197,60 @@ class Gateway: async for chunk in executor.run(content, history=history): yield chunk + async def _resolve_and_persist_model(self, user_id: str, project_id: str, + model_id: str) -> str: + """校验前端选中的模型并持久化到项目。返回 llm.name（空 = 无效/未持久化，调用方走回退链）。 + + model_id 兼容两种取值：llm.id（主键）或 llm.model_id（API 模型名，如 deepseek-v4-pro）—— + 前端 UiCode 的 valueField 历史上两种都用过，这里统一解析，避免语义漂移。 + 多租户隔离：非系统级用户只能选本机构 + 系统级共享模型。 + """ + if not model_id: + return "" + try: + from sqlor.dbpools import DBPools + db = DBPools() + async with db.sqlorContext("pipeline") as sor: + org_id = "" + try: + urecs = await sor.sqlExe( + "SELECT orgid FROM users WHERE id=${u}$", {"u": user_id}) + if urecs: + org_id = getattr(urecs[0], "orgid", "") or "" + except Exception: + pass + recs = await sor.sqlExe( + "SELECT id, name, org_id FROM llm " + "WHERE status='active' AND (id=${m}$ OR model_id=${m}$) LIMIT 1", + {"m": model_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + logger.warning(f"run_message: model_id 无效（不存在或未启用）: {model_id}") + return "" + rec = recs[0] + llm_org = getattr(rec, "org_id", "") or "" + if org_id and org_id != "0" and llm_org not in ("", "0", org_id): + logger.warning(f"run_message: 模型不属于本机构: {model_id} (org={org_id})") + return "" + name = getattr(rec, "name", "") or "" + # 持久化到项目：项目模型一经设置即生效，直到用户再次选择 + if project_id and name: + cur = await sor.sqlExe( + "SELECT default_model FROM sd_projects WHERE id=${p}$", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + if cur and (getattr(cur[0], "default_model", "") or "") != name: + await sor.sqlExe( + "UPDATE sd_projects SET default_model=${m}$ WHERE id=${p}$", + {"m": name, "p": project_id}) + await sor.sqlExe("COMMIT", {}) + logger.info( + f"项目模型已更新: project={project_id} model={name} by={user_id}") + return name + except Exception as e: + logger.warning(f"_resolve_and_persist_model failed: {e}") + return "" + # ── 全局单例 ──