From 8bc1b4e571a45bbcaeccc06d97f6fd9e1a411c24 Mon Sep 17 00:00:00 2001 From: ymq Date: Thu, 20 Aug 2026 17:38:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(session):=20gateway/executor=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=20session=5Fid=20=E5=A4=9A=E4=BC=9A=E8=AF=9D=E9=9A=94?= =?UTF-8?q?=E7=A6=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit web 多 tab 独立会话：run_message 加 session_id，会话 key 从 channel:user_id 扩展为 channel:user_id:session_id（空则向后兼容单会话）；AgentExecutor 历史 加载/保存按 session_id 隔离，空则回退 project 级隔离。 --- pipeline_service/agent_loop_v2.py | 42 ++++++++++++++++++++++--------- pipeline_service/gateway.py | 26 +++++++++++-------- 2 files changed, 45 insertions(+), 23 deletions(-) diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 5f1c665..0c780fd 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -65,6 +65,7 @@ class AgentExecutor: base_url: str = "", # 请求 base_url（scheme://host/），供 slash 命令生成 widget 绝对 URL generic: bool = False, # True = 纯通用会话（不解析项目、不挂产线能力） session=None, # GatewaySession（会话级状态：approve_all/pending_confirm） + session_id: str = "", # 会话内唯一标识（web 多 tab 独立会话；历史隔离键） ): self.config = config self.project_id = project_id @@ -74,6 +75,7 @@ class AgentExecutor: self.role = role # 产线内角色（技能/工具/prompt 按角色加载） self.generic = generic # 是否纯通用会话 self._session = session # GatewaySession（approve_all/pending_confirm 会话级状态） + self.session_id = session_id # 会话内唯一标识（历史消息按此隔离） self.space = GENERAL_SPACE # 项目空间键：generic→'general'，产线→真实 pipeline_id self.workspace_dir = workspace_dir or WORKSPACE_BASE self.model_name = model_name or config.model_name @@ -1196,7 +1198,11 @@ class AgentExecutor: # ═══════════════════════════════════════════════════════ async def _load_history(self, external_history: List[dict] = None) -> List[dict]: - """加载历史消息。隔离策略：project 级别。""" + """加载历史消息。隔离策略：session_id > project 级别。 + + 多 tab 独立会话：session_id 非空时按 session_id 隔离（每个 tab 独立历史）； + 否则退回 config.session_isolation（project/user/none）。 + """ if external_history: return external_history @@ -1208,22 +1214,32 @@ class AgentExecutor: db = DBPools() async with db.sqlorContext("pipeline") as sor: - # 项目隔离 - pid = self.project_id - if self.config.session_isolation == "project" and pid: + # 会话级隔离（web 多 tab 独立会话） + if self.session_id: recs = await sor.sqlExe( "SELECT role, content FROM pipeline_conversations " - "WHERE iteration_id=${pid}$ AND created_by=${uid}$ " + "WHERE session_id=${sid}$ AND created_by=${uid}$ " "ORDER BY created_at ASC LIMIT ${lim}$", - {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit}, + {"sid": self.session_id, "uid": self.user_id or "", + "lim": self.config.history_limit}, ) else: - recs = await sor.sqlExe( - "SELECT role, content FROM pipeline_conversations " - "WHERE created_by=${uid}$ " - "ORDER BY created_at ASC LIMIT ${lim}$", - {"uid": self.user_id or "", "lim": self.config.history_limit}, - ) + # 项目隔离 + pid = self.project_id + if self.config.session_isolation == "project" and pid: + recs = await sor.sqlExe( + "SELECT role, content FROM pipeline_conversations " + "WHERE iteration_id=${pid}$ AND created_by=${uid}$ " + "ORDER BY created_at ASC LIMIT ${lim}$", + {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit}, + ) + else: + recs = await sor.sqlExe( + "SELECT role, content FROM pipeline_conversations " + "WHERE created_by=${uid}$ " + "ORDER BY created_at ASC LIMIT ${lim}$", + {"uid": self.user_id or "", "lim": self.config.history_limit}, + ) if not recs: return [] @@ -1263,6 +1279,7 @@ class AgentExecutor: "content": user_input, "created_by": self.user_id or "", "iteration_id": self.project_id or "", + "session_id": self.session_id or "", }) # 助手回复（不含 tool_call JSON！） reply_clean = reply @@ -1274,6 +1291,7 @@ class AgentExecutor: "content": reply_clean[:4000], "created_by": self.user_id or "", "iteration_id": self.project_id or "", + "session_id": self.session_id or "", }) except Exception as e: logger.warning(f"Session save failed: {e}") diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index ed35a04..5b35779 100644 --- a/pipeline_service/gateway.py +++ b/pipeline_service/gateway.py @@ -28,9 +28,10 @@ logger = logging.getLogger("pipeline.gateway") @dataclass class GatewaySession: - """一个会话 = (channel, user_id) 的 agent 上下文（元数据，executor 每次新建）。""" + """一个会话 = (channel, user_id, session_id) 的 agent 上下文（元数据，executor 每次新建）。""" channel: str user_id: str + session_id: str = "" # 会话内唯一标识（web 多 tab 独立会话）；空 = 单会话（向后兼容） project_id: str = "" pipeline_id: str = "" project_name: str = "" @@ -60,15 +61,17 @@ class Gateway: # ── 会话生命周期 ── @staticmethod - def _session_key(channel: str, user_id: str) -> str: + def _session_key(channel: str, user_id: str, session_id: str = "") -> str: + if session_id: + return f"{channel}:{user_id}:{session_id}" return f"{channel}:{user_id}" - def get_session(self, channel: str, user_id: str) -> Optional[GatewaySession]: - return self._sessions.get(self._session_key(channel, user_id)) + def get_session(self, channel: str, user_id: str, session_id: str = "") -> Optional[GatewaySession]: + return self._sessions.get(self._session_key(channel, user_id, session_id)) - def clear_session(self, channel: str, user_id: str) -> None: + def clear_session(self, channel: str, user_id: str, session_id: str = "") -> None: """清除会话（/reset 时用）。""" - self._sessions.pop(self._session_key(channel, user_id), None) + self._sessions.pop(self._session_key(channel, user_id, session_id), None) # ── 上下文解析 ── @@ -112,7 +115,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 = "") -> AsyncGenerator[str, None]: + base_url: str = "", session_id: str = "") -> AsyncGenerator[str, None]: """统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。 Args: @@ -122,6 +125,7 @@ class Gateway: history: 可选历史消息 role: 可选产线内角色（驾驶舱 agent 为空） generic: True = 纯通用会话（不解析项目、不加载产线能力，只用 GENERAL_TOOLS + 通用心智） + session_id: 会话内唯一标识（web 多 tab 独立会话用；空 = 单会话） """ from pipeline_core.agent_config import load_agent_config from .agent_loop_v2 import AgentExecutor @@ -132,13 +136,13 @@ class Gateway: else: ctx = await self.resolve_project(user_id) - # 2. 会话生命周期（跟踪 project 切换） - key = self._session_key(channel, user_id) + # 2. 会话生命周期（跟踪 project 切换；session_id 区分多 tab 会话） + key = self._session_key(channel, user_id, session_id) sess = self._sessions.get(key) now = time.time() if sess is None or sess.project_id != ctx["pid"]: sess = GatewaySession( - channel=channel, user_id=user_id, + channel=channel, user_id=user_id, session_id=session_id, project_id=ctx["pid"], pipeline_id=ctx["pipeline_id"], project_name=ctx["name"], created_at=now, updated_at=now, ) @@ -157,7 +161,7 @@ class Gateway: config.model_name = ctx["default_llm_name"] executor = AgentExecutor( config=config, project_id=ctx["pid"], user_id=user_id, role=role, - base_url=base_url, generic=generic, session=sess) + base_url=base_url, generic=generic, session=sess, session_id=session_id) # 4. 转发标准化事件流 async for chunk in executor.run(content, history=history):