389 lines
20 KiB
Python
389 lines
20 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"]:
|
||
# 2026-09-04 切换到模型治理新表（旧 llm 表停用）
|
||
llm_recs = await sor.sqlExe(
|
||
"SELECT name FROM llm_model 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 = "",
|
||
image_paths=None) -> 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
|
||
|
||
# 0. 空输入守卫（2026-09-07 实测安全缺陷）：空/纯空白消息绝不进 LLM。
|
||
# 实测根因：curl --data-urlencode 'prompt@不存在的文件' 发出空 prompt，
|
||
# 空用户消息下 LLM 自由发挥——一次幻觉出 delete_project 要求确认，
|
||
# 一次直接调 run_command 执行 python3 heredoc（非危险清单命令，
|
||
# 不触发确认门即执行）。空输入没有任何合法意图，必须入口拦截。
|
||
if not (content or "").strip():
|
||
yield json.dumps({
|
||
"type": "error",
|
||
"message": "收到空消息（可能前端输入丢失或调用方 prompt 为空），已拒绝处理。请输入内容后重试。",
|
||
}, ensure_ascii=False) + "\n"
|
||
return
|
||
|
||
# 0.5 用户敏感信息入站门禁（2026-09-17）：自动侦测明文凭据 → 密文入库 →
|
||
# 原文替换为占位符。放在所有上下文解析之前，因为**净化后的 content 才是
|
||
# 该进模型上下文与落库的版本**（明文一旦进 executor 就会每轮上行供应商）。
|
||
# 失败绝不阻断会话（降级为原文继续，安全增强不牺牲可用性）。
|
||
try:
|
||
content, secret_notes = await self._secret_inbound_gate(user_id, content)
|
||
except Exception as e:
|
||
logger.warning(f"secret inbound gate failed (continue with raw content): {e}")
|
||
content, secret_notes = content, []
|
||
for note in secret_notes:
|
||
yield json.dumps({"type": "progress", "message": note + "\n"}, ensure_ascii=False) + "\n"
|
||
|
||
# 0.6 PII 入站门禁（2026-09-17 用户定案：合规，敏感信息不传第三方）：
|
||
# 手机号/身份证/银行卡/邮箱等个人敏感信息 → AES 密文入 Redis（24h ephemeral，
|
||
# **不进机构密码表**）→ 原文替换 @@pii:占位符。与 0.5 凭据门禁同构不同库。
|
||
# 放在模型/落库之前：净化后的 content 才是上行与落库版本。失败降级原文。
|
||
pii_key = self._session_key(channel, user_id, session_id)
|
||
try:
|
||
from . import pii_guard
|
||
_pr = await pii_guard.mask_inbound(content, pii_key)
|
||
content = _pr.get("text") or content
|
||
pii_notes = list(_pr.get("notes") or [])
|
||
except Exception as e:
|
||
logger.warning(f"pii inbound gate failed (continue with raw content): {e}")
|
||
pii_notes = []
|
||
for note in pii_notes:
|
||
yield json.dumps({"type": "progress", "message": note + "\n"}, ensure_ascii=False) + "\n"
|
||
|
||
# 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, pii_session_key=pii_key)
|
||
|
||
# 3.5 原生视觉（2026-09-10）：上传图片构造多模态 parts。
|
||
# 上限/超限策略在 build_image_parts（8MB/张、4张/条，超限显式告知不静默丢）。
|
||
# 构造失败不阻断会话（降级为纯文本 + 说明）。
|
||
image_parts, image_notes = [], []
|
||
if image_paths:
|
||
try:
|
||
from pipeline_core.upload_tools import build_image_parts
|
||
image_parts, image_notes = build_image_parts(image_paths)
|
||
except Exception as e:
|
||
logger.warning(f"build_image_parts failed: {e}")
|
||
image_notes = [f"图片处理失败（{str(e)[:100]}），本条消息按纯文本处理。"]
|
||
if image_notes:
|
||
content = content + "\n\n[系统说明·本次上传图片] " + "；".join(image_notes)
|
||
|
||
# 4. 转发标准化事件流
|
||
# PII 用户侧还原（2026-09-17）：模型/落库只见占位符，回用户的 reply/progress
|
||
# 还原原文（用户自己贴的信息要能看见）；tool_call/tool_result **不还原**——
|
||
# 它们会进模型上下文与 llm_call_trace 上行。还原失败降级为占位符原样。
|
||
from . import pii_guard as _pii
|
||
async for chunk in executor.run(content, history=history,
|
||
image_parts=image_parts or None):
|
||
if pii_notes and chunk.strip():
|
||
try:
|
||
_ev = json.loads(chunk.strip())
|
||
if _ev.get("type") in ("reply", "progress") and "@@pii:" in str(_ev.get("message", "")):
|
||
_ev["message"] = await _pii.restore_outbound(
|
||
str(_ev["message"]), pii_key)
|
||
chunk = json.dumps(_ev, ensure_ascii=False) + "\n"
|
||
except Exception:
|
||
pass
|
||
yield chunk
|
||
|
||
async def _secret_inbound_gate(self, user_id: str, content: str):
|
||
"""用户敏感信息入站门禁：侦测明文凭据 → 密文入库 → 替换为占位符。
|
||
|
||
返回 (净化后的 content, 动作说明列表)。动作说明以 progress 事件回显给用户
|
||
——「体感不变但安全增强」：用户照常打字，只多看到一条「已自动保存」的提示。
|
||
|
||
安全边界：
|
||
- 无 user_id（无人值守/内部调用）→ 不做任何处理，原文返回（不给无主凭据建库）。
|
||
- 只处理**高置信度**候选（已知凭据前缀，或关键词+熵≥3.5）；中低置信度不自动入库，
|
||
避免把哈希/base64 数据块当凭据吞掉（误报会静默毁数据，比漏报更糟）。
|
||
- 任何异常都降级为原文继续，绝不阻断会话。
|
||
"""
|
||
if not user_id or not content:
|
||
return content, []
|
||
from . import secret_vault
|
||
from sqlor.dbpools import DBPools
|
||
|
||
db = DBPools()
|
||
async with db.sqlorContext("pipeline") as sor:
|
||
org_id = ""
|
||
try:
|
||
recs = await sor.sqlExe(
|
||
"SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": user_id})
|
||
if recs:
|
||
org_id = getattr(recs[0], "orgid", "") or ""
|
||
except Exception as e:
|
||
logger.warning(f"secret gate: resolve org_id failed: {e}")
|
||
r = await secret_vault.scan_and_capture(
|
||
sor, content, org_id=org_id, user_id=user_id,
|
||
who=user_id, auto_store=True)
|
||
try:
|
||
await sor.sqlExe("COMMIT", {})
|
||
except Exception:
|
||
pass
|
||
return r.get("text") or content, list(r.get("actions") or [])
|
||
|
||
async def _resolve_and_persist_model(self, user_id: str, project_id: str,
|
||
model_id: str) -> str:
|
||
"""校验前端选中的模型并持久化到项目。返回 llm.name（空 = 无效/未持久化，调用方走回退链）。
|
||
|
||
model_id 兼容两种取值：llm_model.id（主键）或 llm_model.vendor_model_id（API 模型名）——
|
||
前端 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
|
||
# 2026-09-04 切换到模型治理新表（旧 llm 表停用）
|
||
# 2026-09-06 会话 agent 只接受对话能力（t2t/i2t/m2t）：
|
||
# 能力白名单唯一事实源 = pipeline_llm.selection.CHAT_CAPS
|
||
# （模块未装时兜底同值）；存量 capability 空串按 t2t 语义归一；
|
||
# 非对话模型（embedding/视频生成等）选中即无效，走回退链，
|
||
# 不落项目 default_model
|
||
try:
|
||
from pipeline_llm.selection import CHAT_CAPS as _chat_caps
|
||
except ImportError:
|
||
_chat_caps = ("t2t", "i2t", "m2t")
|
||
_cap_ph = ",".join("${c%d}$" % i for i in range(len(_chat_caps)))
|
||
_cap_params = {"c%d" % i: c for i, c in enumerate(_chat_caps)}
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, org_id FROM llm_model "
|
||
"WHERE status='active' AND (id=${m}$ OR vendor_model_id=${m}$) "
|
||
"AND COALESCE(NULLIF(capability,''),'t2t') IN (%s) LIMIT 1" % _cap_ph,
|
||
dict({"m": model_id}, **_cap_params))
|
||
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
|