270 lines
13 KiB
Python
270 lines
13 KiB
Python
"""
|
||
pipeline-service: gateway — agent 网关（统一消息入口 + 会话生命周期 + 通道注册）
|
||
|
||
gateway 的核心价值：通道无关的统一消息入口。
|
||
Web(AgentIO) / 微信 等通道都通过 gateway.run_message() 触达 agent，
|
||
共享同一套「解析项目上下文 → 加载产线能力 → AgentExecutor 执行」逻辑。
|
||
|
||
标准化事件流（channel-neutral NDJSON），所有通道可消费：
|
||
{"type":"progress","message":...}
|
||
{"type":"tool_call","tool":...,"params":...}
|
||
{"type":"tool_result","result":...}
|
||
{"type":"reply","message":...}
|
||
{"type":"ask_user","message":...}
|
||
{"type":"error","message":...}
|
||
|
||
安全模型（微信通道）：机构级公众号 + 用户 openid 绑定 + 匿名拒绝，
|
||
通道身份（openid）→ 系统 user_id 的映射由各通道适配器负责，gateway 只认 user_id。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from typing import AsyncGenerator, Dict, Optional
|
||
|
||
logger = logging.getLogger("pipeline.gateway")
|
||
|
||
|
||
@dataclass
|
||
class GatewaySession:
|
||
"""一个会话 = (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 = ""
|
||
created_at: float = 0.0
|
||
updated_at: float = 0.0
|
||
approve_all: bool = False # 会话级"全部确认"：用户评估后授权，本会话不再询问危险命令
|
||
pending_confirm: Optional[dict] = None # 待确认工具 {"tool":..., "params":...}（危险命令等待用户确认）
|
||
|
||
|
||
class Gateway:
|
||
"""agent 网关。"""
|
||
|
||
def __init__(self):
|
||
self._sessions: Dict[str, GatewaySession] = {} # key = f"{channel}:{user_id}"
|
||
self._channels: Dict[str, object] = {}
|
||
|
||
# ── 通道注册 ──
|
||
|
||
def register_channel(self, name: str, channel) -> None:
|
||
"""注册一个通道适配器（如 web/wechat）。"""
|
||
self._channels[name] = channel
|
||
logger.info(f"Gateway channel registered: {name}")
|
||
|
||
def list_channels(self) -> list:
|
||
return list(self._channels.keys())
|
||
|
||
# ── 会话生命周期 ──
|
||
|
||
@staticmethod
|
||
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, 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, session_id: str = "") -> None:
|
||
"""清除会话（/reset 时用）。"""
|
||
self._sessions.pop(self._session_key(channel, user_id, session_id), None)
|
||
|
||
# ── 上下文解析 ──
|
||
|
||
async def resolve_project(self, user_id: str, session_id: str = "") -> dict:
|
||
"""解析用户当前项目 → {pid, pipeline_id, name, default_llm_id, default_llm_name, project_model_name}。
|
||
|
||
pid 按会话隔离（session_id 非空时读 pipeline_session_settings，无记录回退全局）；
|
||
同时读出个人选择的模型（default_llm_id → llm.name，仍按用户全局）与项目级模型
|
||
（sd_projects.default_model，产线通用、跨会话持久），供 run_message 按优先级覆盖。
|
||
"""
|
||
ctx = {"pid": "", "pipeline_id": "", "name": "", "default_llm_id": "",
|
||
"default_llm_name": "", "project_model_name": ""}
|
||
if not user_id:
|
||
return ctx
|
||
try:
|
||
from sqlor.dbpools import DBPools
|
||
from .workspace import get_session_project_id
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
# 当前项目：按会话隔离（多 tab 各自项目，互不覆盖）
|
||
ctx["pid"] = await get_session_project_id(sor, user_id, session_id) or ""
|
||
# 个人模型选择：仍按用户全局读
|
||
recs = await sor.sqlExe(
|
||
"SELECT default_llm_id FROM pipeline_agent_settings WHERE user_id=${u}$",
|
||
{"u": user_id})
|
||
if recs:
|
||
ctx["default_llm_id"] = getattr(recs[0], "default_llm_id", "") or ""
|
||
# 单独查 llm.name（避免 JOIN 触发两表字段 collation 不一致）
|
||
if ctx["default_llm_id"]:
|
||
llm_recs = await sor.sqlExe(
|
||
"SELECT name FROM llm WHERE id=${id}$ AND status='active'",
|
||
{"id": ctx["default_llm_id"]})
|
||
if llm_recs:
|
||
ctx["default_llm_name"] = getattr(llm_recs[0], "name", "") or ""
|
||
if ctx["pid"]:
|
||
proj = await sor.sqlExe(
|
||
"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
|
||
|
||
# ── 统一消息入口 ──
|
||
|
||
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 = "", model_id: str = "") -> AsyncGenerator[str, None]:
|
||
"""统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。
|
||
|
||
Args:
|
||
channel: 通道名（web / wechat / ...）
|
||
user_id: 系统用户 ID（通道适配器负责把 openid 映射到 user_id）
|
||
content: 用户消息内容
|
||
history: 可选历史消息
|
||
role: 可选产线内角色（驾驶舱 agent 为空）
|
||
generic: True = 纯通用会话（不解析项目、不加载产线能力，只用 GENERAL_TOOLS + 通用心智）
|
||
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": "", "project_model_name": ""}
|
||
else:
|
||
ctx = await self.resolve_project(user_id, session_id)
|
||
# 产线隔离：入口指定产线时，会话项目必须同产线才生效——
|
||
# 防止"元景项目"(sdlc_general)劫持商机/投标产线会话。
|
||
if pipeline_id and ctx.get("pipeline_id") and ctx["pipeline_id"] != pipeline_id:
|
||
ctx["pid"] = ""
|
||
ctx["pipeline_id"] = pipeline_id
|
||
# 无当前项目 → 用入口指定的默认产线
|
||
elif 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)
|
||
now = time.time()
|
||
if sess is None or sess.project_id != ctx["pid"]:
|
||
sess = GatewaySession(
|
||
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,
|
||
)
|
||
self._sessions[key] = sess
|
||
else:
|
||
sess.project_id = ctx["pid"]
|
||
sess.pipeline_id = ctx["pipeline_id"]
|
||
sess.project_name = ctx["name"]
|
||
sess.updated_at = now
|
||
|
||
# 3. 加载产线能力 + 创建 executor（generic 时不装产线能力）
|
||
config = await load_agent_config(
|
||
pipeline_id=ctx["pipeline_id"], project_id=ctx["pid"], generic=generic)
|
||
# 模型覆盖优先级：本次选择 > 项目模型 > 个人全局选择（个人全局只兜底，不再压过项目设置）
|
||
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,
|
||
base_url=base_url, generic=generic, session=sess, session_id=session_id,
|
||
default_pipeline_id=pipeline_id)
|
||
|
||
# 4. 转发标准化事件流
|
||
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 ""
|
||
|
||
|
||
# ── 全局单例 ──
|
||
|
||
_gateway: Optional[Gateway] = None
|
||
|
||
|
||
def get_gateway() -> Gateway:
|
||
global _gateway
|
||
if _gateway is None:
|
||
_gateway = Gateway()
|
||
return _gateway
|