diff --git a/models/pipeline_user_memory.json b/models/pipeline_user_memory.json index 66ae6a3..7115dce 100644 --- a/models/pipeline_user_memory.json +++ b/models/pipeline_user_memory.json @@ -54,6 +54,22 @@ "nullable": "no", "default": "" }, + { + "name": "org_id", + "title": "归属机构ID(空=平台种子)", + "type": "str", + "length": 32, + "nullable": "no", + "default": "" + }, + { + "name": "user_id", + "title": "归属用户ID(user域必填)", + "type": "str", + "length": 32, + "nullable": "no", + "default": "" + }, { "name": "priority", "title": "优先级", diff --git a/pipeline_core/agent_config.py b/pipeline_core/agent_config.py index 1d7b627..c4a5824 100644 --- a/pipeline_core/agent_config.py +++ b/pipeline_core/agent_config.py @@ -27,6 +27,10 @@ class ToolDefinition: enabled: bool = True category: str = "general" # project / task / repo / shell / agent requires_confirmation: bool = False # 是否需要用户确认 + # 显式必填参数列表（2026-09-10）：None = 旧行为（全部参数必填）。 + # 含可选参数的工具必须显式声明，否则 native FC schema 把可选参数 + # 也标成 required，LLM 被迫为每个参数编值（v1 同款坑，机制层根治）。 + required: Optional[list] = None @dataclass @@ -132,6 +136,8 @@ class AgentConfig: "parameters": t.parameters, "enabled": t.enabled, "category": t.category, + "requires_confirmation": t.requires_confirmation, + "required": t.required, } for t in self.tools ], @@ -241,8 +247,9 @@ GENERAL_TOOLS = [ ), ToolDefinition( name="run_command", - description="在工作空间中执行shell命令", - parameters={"command": "命令"}, + description="在工作空间中执行shell命令。长任务（装依赖/构建/跑测试）用 background=true 转后台：立即返回 job_id，之后用 process 工具 poll/log/wait/kill 跟进，不阻塞对话", + parameters={"command": "命令", "background": "可选：true=后台执行返回job_id（默认前台，60秒超时）", "timeout": "可选：前台超时秒数（默认60，最大300）"}, + required=["command"], category="shell", requires_confirmation=True, ), @@ -255,8 +262,16 @@ GENERAL_TOOLS = [ ), ToolDefinition( name="delegate_subtask", - description="派生子agent调查子任务（并行执行）", - parameters={"goal": "子任务目标", "context": "背景信息"}, + description="派生子agent执行子任务。background=false（默认）同步等待返回结果；background=true 后台并行执行（同一会话最多3个并行），立即返回 subagent_id，之后用 subagent 工具跟进。子agent看不到本会话上下文，goal/context 必须自包含", + parameters={"goal": "子任务目标（自包含）", "context": "可选：背景信息", "background": "可选：true=后台并行（默认false同步等待）"}, + required=["goal"], + category="agent", + ), + ToolDefinition( + name="subagent", + description="管理后台子agent：list=列出本会话的子agent及状态；steer=给运行中的子agent追加指示（下轮生效）；stop=提前终止（返回已有部分结果）；result=取最终结果", + parameters={"action": "list|steer|stop|result", "subagent_id": "子agent ID（list 时可空）", "message": "steer 时的追加指示"}, + required=["action"], category="agent", ), # ── 通用工具集（Hermes CLI 能力子集：文件/搜索/会话/规划）── @@ -264,12 +279,14 @@ GENERAL_TOOLS = [ name="read_file", description="读取工作空间中的文件内容。支持 docx/pdf/txt/md/json 等格式，docx/pdf 会自动解析提取正文文本，直接调用即可读取。单次最多 30000 字符；返回带截断提示时，用 offset 参数分段续读后文，逐段读完全文（切勿只读开头就以为读全了）", parameters={"path": "相对路径", "offset": "可选：从第几个字符开始读（分段读大文件）"}, + required=["path"], category="file", ), ToolDefinition( name="load_skill", description="加载指定技能的完整内容（具体步骤/规范/陷阱）。先在系统提示的『可用技能』目录里找到技能名，需要时再调用本工具加载正文。技能带子文件（模板库/参考文档）时，正文末尾会列出可用子文件，用 file_path 参数逐个加载", parameters={"name": "技能名称", "file_path": "可选：子文件相对路径（如 references/uapi-templates.md），只允许 references/scripts/templates/assets 下的文件"}, + required=["name"], category="skill", ), ToolDefinition( @@ -288,6 +305,7 @@ GENERAL_TOOLS = [ name="propose_skill", description="沉淀技能：把值得复用的流程/经验/坑写成技能并实时发布到你所属机构的技能目录（同名覆盖通用技能，机构内立即生效，其他机构不受影响）。用户要求沉淀经验、总结技能，或你发现反复出现的流程/坑/规范时调用。平台缺省机构(org 0)的提议会转人工审核（不实时生效）", parameters={"name": "技能名（字母数字._-，≤64字符）", "description": "技能描述", "content": "SKILL.md 正文（frontmatter 可选，系统自动补全）"}, + required=["name", "content"], category="skill", ), ToolDefinition( @@ -300,12 +318,14 @@ GENERAL_TOOLS = [ name="list_files", description="列出工作空间目录内容", parameters={"path": "相对路径(可选,默认工作空间根)"}, + required=[], category="file", ), ToolDefinition( name="search_files", description="在工作空间中搜索文件内容（grep）", parameters={"pattern": "搜索关键词或正则", "path": "相对路径(可选,默认整个工作空间)"}, + required=["pattern"], category="file", ), ToolDefinition( @@ -314,10 +334,56 @@ GENERAL_TOOLS = [ parameters={"query": "搜索关键词"}, category="memory", ), + # ── 持久记忆写入（2026-09-10，对齐 Hermes memory 工具）── + # 多租户门禁在 agent_loop_v2._t_memory：agent 只能写自己机构/自己名下的 + # user/project/pipeline 域，global/org 平台种子域禁写；org_id 强制注入。 + ToolDefinition( + name="memory", + description="持久记忆（跨会话保留）：add=记住一条事实/偏好；list=列出你可见的记忆；remove=删除一条。用户表达偏好/纠正/要求记住某事，或你发现值得跨会话保留的事实时调用。scope 选择：user=关于用户本人的偏好，project=当前项目的约定/事实，pipeline=当前产线的通用经验", + parameters={ + "action": "add|list|remove", + "content": "记忆内容（add 必填，一句话陈述事实，勿写指令式）", + "scope": "可选：user|project|pipeline（默认 user；无项目时 project 不可用）", + "category": "可选：user=用户偏好|memory=一般事实（默认 memory）", + "key": "remove 时必填：记忆键（list 返回）", + }, + required=["action"], + category="memory", + ), + # ── 技能管理（2026-09-10，对齐 Hermes skill_manage：create/patch/write_file/remove_file/delete）── + # 隔离由 skill_live.resolve_target 保证：只能落在本机构 orgs/{org}/ 或 + # org 0 降级 users/{uid}/，global 与他机构目录物理不可达。 + ToolDefinition( + name="manage_skill", + description="管理你所属机构的技能库：create=新建技能（整篇 SKILL.md）；patch=定向修改已有技能的片段（old_string 须唯一）；write_file=给技能添加子文件（限 references/scripts/templates/assets 下）；remove_file=删技能子文件；delete=删除整个技能。只能操作本机构（或本人）目录下的技能——覆盖自通用技能的同名副本可改可删，但删不掉通用(global)原版。propose_skill 等价于 create", + parameters={ + "action": "create|patch|write_file|remove_file|delete", + "name": "技能名（字母数字._-，≤64字符）", + "description": "create 时的一句话描述", + "content": "create 时的 SKILL.md 正文；patch 时不用", + "old_string": "patch 必填：要替换的原文片段（须在目标文件中唯一）", + "new_string": "patch 必填：替换后的新文本", + "file_path": "write_file/remove_file 必填、patch 可选：子文件相对路径（如 references/api.md，首段限 references/scripts/templates/assets）", + "file_content": "write_file 必填：子文件内容", + }, + required=["action", "name"], + category="skill", + ), + # ── 后台进程管理（2026-09-10，配套 run_command background=true）── + # 状态文件化（workspace/.bg/{job_id}/），跨 worker 进程可读； + # 隔离靠 workspace 路径（用户/项目工作空间天然隔离）。 + ToolDefinition( + name="process", + description="管理 run_command 启动的后台任务：poll=查状态+新增输出；log=取完整输出（可分页）；wait=阻塞等待结束（最多120秒，超时返回部分输出）；kill=终止", + parameters={"action": "poll|log|wait|kill", "job_id": "后台任务 ID（run_command background=true 返回）", "offset": "log 可选：从第几个字符开始读"}, + required=["action", "job_id"], + category="shell", + ), ToolDefinition( name="todo", description="管理当前会话的任务清单（list/add/done）", parameters={"action": "list|add|done", "content": "任务内容(add时必填)"}, + required=["action"], category="agent", ), # ── 平台模型调用（2026-09-07）：可用模型 = 平台 owner 机构 + 本机构的 @@ -330,6 +396,7 @@ GENERAL_TOOLS = [ name="list_platform_models", description="列出平台当前可用的模型（本机构+平台owner机构的模型，含能力类型：t2t对话/i2t图像理解/t2i文生图/t2v文生视频/i2v图生视频/tts语音合成/asr语音识别等）。用户问「有哪些模型/能做什么」，或你要调用非对话能力（生图/视频/语音）前需要选模型时调用。capability 参数可按能力过滤（如 t2i）", parameters={"capability": "可选：能力类型过滤（如 t2i/t2v/tts），空=全部"}, + required=[], category="model", ), ToolDefinition( @@ -341,6 +408,7 @@ GENERAL_TOOLS = [ "capability": "可选：能力类型（t2i/t2v/i2v/tts/asr等，自动匹配时用于过滤候选）", "params": "可选：业务参数 JSON 字符串。媒体输入用三数组契约：image_files/audio_files/video_files（值为公网URL或base64的数组）；生成参数如 resolution/duration/size 按模型文档", }, + required=["task"], category="model", ), # ── 联网检索与网页抓取（2026-09-08，对齐 Hermes web_search/web_extract）── @@ -350,6 +418,7 @@ GENERAL_TOOLS = [ name="web_search", description="联网检索信息（搜索引擎）。需要外部资料/时事/文档/依据而知识库与本地文件没有时调用。返回标题+URL+摘要列表；需要某条结果的完整内容时再用 fetch_url 抓取", parameters={"query": "搜索关键词", "limit": "可选：返回条数(默认8,最大10)"}, + required=["query"], category="web", ), ToolDefinition( @@ -370,6 +439,7 @@ GENERAL_TOOLS = [ "order_by": "可选：排序列（可带 ASC/DESC）", "limit": "可选：行数上限（默认/最大100）", }, + required=["table"], category="data", ), ] diff --git a/pipeline_core/memory_store.py b/pipeline_core/memory_store.py index b42dbc3..80b1bf2 100644 --- a/pipeline_core/memory_store.py +++ b/pipeline_core/memory_store.py @@ -1,5 +1,5 @@ -""" -pipeline-core: Memory Store — 跨会话记忆系统 +# -*- coding: utf-8 -*- +"""pipeline-core: Memory Store — 跨会话记忆系统（多租户版，2026-09-10） 对照 Hermes Agent 的 memory 工具 + MEMORY.md / USER.md： - 用户偏好存储（user profile） @@ -8,25 +8,44 @@ pipeline-core: Memory Store — 跨会话记忆系统 - 自动去重和淘汰 存储方式：MySQL pipeline_user_memory 表 + 内存缓存 + +多租户隔离（2026-09-10，服务多机构多用户）： +- 每条记忆带 org_id / user_id 归属列（org_id='' 且 user_id='' = 平台种子记忆） +- memory_key = hash(content + org_id + user_id)——同内容不同租户 key 不同， + 唯一键永不跨租户碰撞 +- 可见性规则（visible_to）： + global 种子（org_id=''） → 所有人可见（平台预置，agent 禁写） + org 归属（org_id=O） → 仅机构 O 的用户可见 + user 归属（user_id=U） → 仅用户 U 本人可见 + pipeline/project 域带 org_id=O → 机构 O 内、且产线/项目匹配才可见 +- 写入约束由调用方（agent handler）把关：agent 只能写自己 org_id 名下、 + scope ∈ {user, project, pipeline}；global/org 域是平台种子域，禁写 +- 淘汰分区（_evict）：只淘汰「同 org_id + 同 user 分区」内的低优先级旧条目， + 绝不删平台种子（org_id=''）或其他机构的记忆 """ -import json +import hashlib import logging import time -from dataclasses import dataclass, field -from typing import Dict, List, Optional +from dataclasses import dataclass +from typing import Dict, List, Optional, Tuple logger = logging.getLogger("pipeline.memory_store") +# 缓存分区键 = (category, scope, scope_id, org_id, user_id) +_PartitionKey = Tuple[str, str, str, str, str] + @dataclass class MemoryEntry: """一条记忆""" - key: str # 唯一标识 + key: str # 唯一标识（掺入租户维度的内容哈希） content: str # 记忆内容 category: str = "memory" # "user" | "memory" - scope: str = "global" # "global" | "pipeline" | "project" | "user"（归属范围） - scope_id: str = "" # pipeline_id / project_id / user_id + scope: str = "global" # "global" | "pipeline" | "project" | "org" | "user" + scope_id: str = "" # pipeline_id / project_id / org_id / user_id + org_id: str = "" # 归属机构（'' = 平台种子） + user_id: str = "" # 归属用户（user 域必填；'' = 非个人） priority: int = 0 # 优先级（越高越重要） created_at: float = 0.0 updated_at: float = 0.0 @@ -39,6 +58,8 @@ class MemoryEntry: "category": self.category, "scope": self.scope, "scope_id": self.scope_id, + "org_id": self.org_id, + "user_id": self.user_id, "priority": self.priority, "created_at": self.created_at, "updated_at": self.updated_at, @@ -53,19 +74,38 @@ class MemoryEntry: category=data.get("category", "memory"), scope=data.get("scope", "global"), scope_id=data.get("scope_id", ""), + org_id=data.get("org_id", ""), + user_id=data.get("user_id", ""), priority=data.get("priority", 0), created_at=data.get("created_at", time.time()), updated_at=data.get("updated_at", time.time()), access_count=data.get("access_count", 0), ) + @property + def partition(self) -> _PartitionKey: + return (self.category, self.scope, self.scope_id, self.org_id, self.user_id) + + +def _make_key(content: str, org_id: str = "", user_id: str = "", + scope: str = "", scope_id: str = "") -> str: + """从内容 + 租户 + 作用域维度生成稳定 key。 + + 掺入 org_id/user_id/scope/scope_id： + - 同一段内容在机构 A 和机构 B 各自存储时 key 不同，唯一索引 + (memory_key, category) 永不跨租户碰撞、互不覆盖； + - 同租户同内容写到不同作用域（项目域/个人域）也是不同条目。 + """ + raw = f"{content}|org={org_id}|user={user_id}|scope={scope}|sid={scope_id}" + return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16] + class MemoryStore: - """跨会话记忆存储。 + """跨会话记忆存储（多租户）。 两层架构： 1. MySQL pipeline_user_memory（持久化） - 2. 内存缓存（加速读取） + 2. 内存缓存（按分区键组织，加速读取） """ # 优先级分组 @@ -73,12 +113,18 @@ class MemoryStore: PRIORITY_MEDIUM = 5 # 项目约定、工作流 PRIORITY_LOW = 1 # 临时备注 + # agent 可写的 scope 白名单（global/org 是平台种子域，禁写） + WRITABLE_SCOPES = ("user", "project", "pipeline") + def __init__(self): - self._cache: Dict[str, Dict[str, MemoryEntry]] = {} # {category: {key: entry}} + # {partition_key: {key: MemoryEntry}} + self._cache: Dict[_PartitionKey, Dict[str, MemoryEntry]] = {} self._cache_loaded = False + # ── 缓存 ── + async def _ensure_cache(self): - """从 DB 加载缓存""" + """从 DB 加载缓存（全表按优先级取前 2000 条；可见性在读取时过滤）""" if self._cache_loaded: return @@ -88,8 +134,9 @@ class MemoryStore: db = DBPools() async with db.sqlorContext("pipeline") as sor: recs = await sor.sqlExe( - "SELECT memory_key, content, category, scope, scope_id, priority, created_at, updated_at, access_count " - "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 500", + "SELECT memory_key, content, category, scope, scope_id, " + "org_id, user_id, priority, created_at, updated_at, access_count " + "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 2000", {}, ) for r in (recs or []): @@ -99,30 +146,82 @@ class MemoryStore: category=getattr(r, "category", "memory"), scope=getattr(r, "scope", "global"), scope_id=getattr(r, "scope_id", ""), + org_id=getattr(r, "org_id", "") or "", + user_id=getattr(r, "user_id", "") or "", priority=getattr(r, "priority", 0), created_at=_ts(getattr(r, "created_at", None)), updated_at=_ts(getattr(r, "updated_at", None)), access_count=getattr(r, "access_count", 0), ) - if entry.category not in self._cache: - self._cache[entry.category] = {} - self._cache[entry.category][entry.key] = entry + pk = entry.partition + if pk not in self._cache: + self._cache[pk] = {} + self._cache[pk][entry.key] = entry except Exception as e: logger.warning(f"MemoryStore cache load failed: {e}") self._cache_loaded = True + def _invalidate_cache(self): + """多 worker 共享 DB：写路径失效缓存，下次读重新加载（防跨进程读到旧缓存）。""" + self._cache = {} + self._cache_loaded = False + + # ── 可见性 ── + + @staticmethod + def visible_to(entry: MemoryEntry, org_id: str, user_id: str, + pipeline_id: str = "", project_id: str = "") -> bool: + """多租户可见性判定（唯一权威规则，读取端全部走这里）。 + + - 平台种子（org_id='' 且 user_id=''）→ 所有人可见（只读预置） + - user 归属 → 仅本人 + - org 归属（scope=org）→ 仅本机构 + - pipeline 域 → 本机构（或种子）且 pipeline_id 匹配 + - project 域 → 本机构（或种子）且 project_id 匹配 + - global 域机构记忆（org_id=O）→ 仅机构 O + """ + seed = (entry.org_id == "" and entry.user_id == "") + if seed: + # 种子记忆：pipeline/project 域还要求当前上下文匹配（或域值为空=通用种子） + if entry.scope == "pipeline" and entry.scope_id and entry.scope_id != pipeline_id: + return False + if entry.scope == "project" and entry.scope_id and entry.scope_id != project_id: + return False + return True + # 归属记忆：机构必须匹配（跨机构一律不可见） + if entry.org_id and entry.org_id != org_id: + return False + # user 域：仅本人可见（user_id 记录时恒为归属者） + if entry.scope == "user": + return entry.user_id == user_id + # project/pipeline 域：机构内共享（user_id 只记创建者，供删除过滤， + # 不影响可见性）；上下文须匹配对应项目/产线 + if entry.scope == "pipeline": + return entry.scope_id == pipeline_id + if entry.scope == "project": + return entry.scope_id == project_id + if entry.scope == "org": + return True # org_id 已在上面匹配 + # 其余（global 域带 org_id 的机构通用记忆）：机构匹配即可见 + return True + + # ── 写入 ── + async def add(self, content: str, category: str = "memory", priority: int = PRIORITY_MEDIUM, key: str = None, - scope: str = "global", scope_id: str = ""): - """添加一条记忆。自动生成 key（取内容前80字符的哈希）。 + scope: str = "global", scope_id: str = "", + org_id: str = "", user_id: str = ""): + """添加一条记忆（自动生成掺租户维度的 key）。 - scope 分域：global（通用）/ pipeline（产线）/ project（项目）/ user（用户个人）。 + ⚠️ 本方法不做写入权限校验（平台内部/种子导入也用它）。 + agent 侧的写入门禁（禁 global/org 域、org_id 强制注入）在 + agent_loop_v2 的 _t_memory handler 里把关，勿绕过 handler 直调。 """ await self._ensure_cache() if not key: - key = _make_key(content) + key = _make_key(content, org_id, user_id, scope, scope_id) now = time.time() entry = MemoryEntry( @@ -131,87 +230,154 @@ class MemoryStore: category=category, scope=scope, scope_id=scope_id, + org_id=org_id, + user_id=user_id, priority=priority, created_at=now, updated_at=now, ) - # 更新缓存 - if category not in self._cache: - self._cache[category] = {} - self._cache[category][key] = entry + pk = entry.partition + if pk not in self._cache: + self._cache[pk] = {} + self._cache[pk][key] = entry - # 持久化 try: await self._db_upsert(entry) except Exception as e: logger.error(f"MemoryStore add failed: {e}") - # 淘汰低优先级旧条目 - await self._evict(max_entries=100) + # 分区内淘汰（只动同 org+user 分区，绝不碰种子和他租户） + await self._evict_partition(org_id, user_id, max_entries=100) - async def get(self, category: str = None, key: str = None, - scope: str = None, scope_id: str = None) -> List[MemoryEntry]: - """获取记忆。可按分类/key/scope 筛选。 + # 多 worker 一致性：写后失效本进程缓存，下次读取重载全表—— + # 否则其他 worker 进程写入的记忆本进程永远看不到（旧版同款缺陷） + self._invalidate_cache() - scope 过滤：scope=None 返回全部；scope 指定时返回该 scope 且 scope_id 匹配（或 scope_id=None 时该 scope 全部）。 + # ── 读取 ── + + async def get(self, category: Optional[str] = None, key: Optional[str] = None, + scope: Optional[str] = None, scope_id: Optional[str] = None, + org_id: Optional[str] = None, user_id: Optional[str] = None) -> List[MemoryEntry]: + """获取记忆（**不带可见性过滤的底层查询**，管理/巡检用）。 + + agent 读取一律走 get_visible()——本方法会把所有租户的条目都返回， + 直接暴露给 LLM 就是跨机构泄漏。 """ await self._ensure_cache() results = [] - cats = [category] if category else list(self._cache.keys()) - for cat in cats: - if cat in self._cache: - for k, entry in self._cache[cat].items(): - if key and k != key: - continue - if scope is not None: - if entry.scope != scope: - continue - if scope_id is not None and entry.scope_id != scope_id: - continue - results.append(entry) - entry.access_count += 1 + for pk, entries in self._cache.items(): + cat, sc, sid, oid, uid = pk + if category and cat != category: + continue + if scope is not None and sc != scope: + continue + if scope_id is not None and sid != scope_id: + continue + if org_id is not None and oid != org_id: + continue + if user_id is not None and uid != user_id: + continue + for k, entry in entries.items(): + if key and k != key: + continue + results.append(entry) + entry.access_count += 1 results.sort(key=lambda e: e.priority, reverse=True) return results - async def remove(self, key: str, category: str = "memory"): - """删除一条记忆""" + async def get_visible(self, org_id: str, user_id: str, + pipeline_id: str = "", project_id: str = "", + category: str = None) -> List[MemoryEntry]: + """按租户可见性取记忆（agent 读取唯一入口）。""" + all_entries = await self.get(category=category) + visible = [e for e in all_entries + if self.visible_to(e, org_id or "", user_id or "", + pipeline_id or "", project_id or "")] + visible.sort(key=lambda e: e.priority, reverse=True) + return visible + + # ── 删除 ── + + async def remove(self, key: str, category: str = "memory", + scope: Optional[str] = None, scope_id: str = "", + org_id: Optional[str] = None, user_id: Optional[str] = None): + """删除一条记忆。 + + 多租户安全：org_id/user_id 传 None 表示不过滤（内部淘汰用，已分区）； + agent 侧删除必须显式传 org_id/user_id，防止按 key 误删他租户同名条目。 + """ await self._ensure_cache() - if category in self._cache: - self._cache[category].pop(key, None) + removed = 0 + for pk in list(self._cache.keys()): + cat, sc, sid, oid, uid = pk + if cat != category: + continue + if scope is not None and sc != scope: + continue + if scope_id and sid != scope_id: + continue + if org_id is not None and oid != org_id: + continue + if user_id is not None and uid != user_id: + continue + if key in self._cache[pk]: + self._cache[pk].pop(key, None) + removed += 1 - try: - from sqlor.dbpools import DBPools + if removed: + try: + from sqlor.dbpools import DBPools - db = DBPools() - async with db.sqlorContext("pipeline") as sor: - await sor.sqlExe( - "DELETE FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$", - {"k": key, "c": category}, - ) - except Exception as e: - logger.error(f"MemoryStore remove failed: {e}") + db = DBPools() + async with db.sqlorContext("pipeline") as sor: + sql = "DELETE FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$" + params = {"k": key, "c": category} + if scope is not None: + sql += " AND scope=${s}$" + params["s"] = scope + if scope_id: + sql += " AND scope_id=${sid}$" + params["sid"] = scope_id + if org_id is not None: + sql += " AND org_id=${oid}$" + params["oid"] = org_id + if user_id is not None: + sql += " AND user_id=${uid}$" + params["uid"] = user_id + await sor.sqlExe(sql, params) + self._invalidate_cache() # 同 add：写后失效，跨进程可见 + except Exception as e: + logger.error(f"MemoryStore remove failed: {e}") + return removed - async def build_prompt_block(self, category: str = None, max_entries: int = 20, - scope: str = None, scope_id: str = None) -> str: - """构建注入 system prompt 的记忆段落。 + # ── prompt 注入 ── + + async def build_prompt_block(self, category: Optional[str] = None, max_entries: int = 20, + scope: Optional[str] = None, scope_id: Optional[str] = None, + org_id: str = "", user_id: str = "", + pipeline_id: str = "", project_id: str = "") -> str: + """构建注入 system prompt 的记忆段落（租户可见性过滤版）。 + + 传入 org_id/user_id 时按 visible_to 过滤（agent 会话注入的唯一正确用法）； + 两者都不传 = 旧行为（global 种子 + 指定 scope），仅供无租户上下文的 + 内部场景（如平台巡检），agent 链路禁用。 - 指定 scope 时加载「global（通用）+ 指定 scope（专属）」的记忆， - 实现通用记忆 + 产线/项目专属记忆的叠加注入。 高优先级记忆注入完整内容，低优先级只注入摘要。 """ - if scope: - # 通用 + 专属叠加 - global_entries = await self.get(category, scope="global") - scoped_entries = await self.get(category, scope=scope, scope_id=scope_id) - # 去重（同 key 专属覆盖通用） - merged = {e.key: e for e in global_entries} - merged.update({e.key: e for e in scoped_entries}) - entries = list(merged.values()) + if org_id or user_id: + entries = await self.get_visible(org_id, user_id, pipeline_id, project_id, + category=category) else: - entries = await self.get(category) + # 兼容旧签名：无租户上下文时只给平台种子（绝不给归属记忆） + all_entries = await self.get(category=category) + entries = [e for e in all_entries if e.org_id == "" and e.user_id == ""] + if scope: + entries = [e for e in entries + if e.scope in ("global", scope) + and (not scope_id or e.scope != scope or e.scope_id == scope_id)] if not entries: return "" @@ -231,16 +397,19 @@ class MemoryStore: return "\n".join(blocks) if blocks else "" + # ── DB ── + async def _db_upsert(self, entry: MemoryEntry): - """写入/更新 DB（同 key + category + scope + scope_id 才 upsert）。""" + """写入/更新 DB（同 key + category + scope + scope_id + org + user 才 upsert）。""" from sqlor.dbpools import DBPools db = DBPools() async with db.sqlorContext("pipeline") as sor: existing = await sor.sqlExe( "SELECT id FROM pipeline_user_memory WHERE memory_key=${k}$ AND category=${c}$ " - "AND scope=${s}$ AND scope_id=${sid}$", - {"k": entry.key, "c": entry.category, "s": entry.scope, "sid": entry.scope_id}, + "AND scope=${s}$ AND scope_id=${sid}$ AND org_id=${oid}$ AND user_id=${uid}$", + {"k": entry.key, "c": entry.category, "s": entry.scope, "sid": entry.scope_id, + "oid": entry.org_id, "uid": entry.user_id}, ) if existing: rid = existing[0].id @@ -261,36 +430,40 @@ class MemoryStore: "category": entry.category, "scope": entry.scope, "scope_id": entry.scope_id, + "org_id": entry.org_id, + "user_id": entry.user_id, "priority": entry.priority, }) - async def _evict(self, max_entries: int = 100): - """淘汰低优先级的旧记忆（LRU）""" - total = sum(len(v) for v in self._cache.values()) - if total <= max_entries: + async def _evict_partition(self, org_id: str, user_id: str, max_entries: int = 100): + """分区淘汰：只淘汰「同 org_id + 同 user_id」分区内的低优先级旧记忆。 + + 平台种子（org_id='' 且 user_id=''）永不被 agent 写入触发淘汰； + 机构 A 的写入永远不会删机构 B / 其他用户 / 种子的记忆。 + """ + partition_entries: List[MemoryEntry] = [] + for pk, entries in self._cache.items(): + _cat, _sc, _sid, oid, uid = pk + if oid == (org_id or "") and uid == (user_id or ""): + partition_entries.extend(entries.values()) + + if len(partition_entries) <= max_entries: return - # 找出可淘汰的条目（低优先级 + 低访问次数） - all_entries = [] - for cat_entries in self._cache.values(): - all_entries.extend(cat_entries.values()) - - all_entries.sort(key=lambda e: (e.priority, e.access_count, e.updated_at)) - - to_remove = all_entries[: (total - max_entries)] + partition_entries.sort(key=lambda e: (e.priority, e.access_count, e.updated_at)) + to_remove = partition_entries[: (len(partition_entries) - max_entries)] for entry in to_remove: - await self.remove(entry.key, entry.category) + await self.remove(entry.key, entry.category, + scope=entry.scope, scope_id=entry.scope_id, + org_id=entry.org_id, user_id=entry.user_id) + + # 兼容旧接口名（旧调用点 _evict(max_entries) 语义已不安全，转发到分区版） + async def _evict(self, max_entries: int = 100, org_id: str = "", user_id: str = ""): + await self._evict_partition(org_id, user_id, max_entries) # ── 辅助 ── -def _make_key(content: str) -> str: - """从内容生成稳定 key""" - import hashlib - - return hashlib.md5(content.encode("utf-8")).hexdigest()[:16] - - def _ts(val) -> float: """转换时间戳""" if val is None: diff --git a/pipeline_core/tool_registry.py b/pipeline_core/tool_registry.py index 6d49626..47d1c44 100644 --- a/pipeline_core/tool_registry.py +++ b/pipeline_core/tool_registry.py @@ -88,7 +88,8 @@ class ToolRegistry: k: {"type": "string", "description": v} for k, v in (t.parameters or {}).items() }, - "required": list(t.parameters.keys()) if t.parameters else [], + "required": (t.required if t.required is not None + else list(t.parameters.keys())) if t.parameters else [], }, }, }