From b9b6403de35820411e2d507d082f47566bee4e97 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 15 Aug 2026 09:46:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A7=92=E8=89=B2=E5=AE=9A=E4=B9=89?= =?UTF-8?q?=E5=8F=AF=E6=8F=92=E6=8B=94(RoleSpec+=E8=BE=85=E5=8A=A9?= =?UTF-8?q?=E5=87=BD=E6=95=B0)+=E8=AE=B0=E5=BF=86=E5=88=86=E5=9F=9F(scope:?= =?UTF-8?q?global/pipeline/project/user)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_core/__init__.py | 5 +++ pipeline_core/ability.py | 53 ++++++++++++++++++++++++++++- pipeline_core/memory_store.py | 63 +++++++++++++++++++++++++++++------ 3 files changed, 109 insertions(+), 12 deletions(-) diff --git a/pipeline_core/__init__.py b/pipeline_core/__init__.py index 60ad6fd..8fd530d 100644 --- a/pipeline_core/__init__.py +++ b/pipeline_core/__init__.py @@ -30,11 +30,16 @@ from .agent_config import ( ) from .ability import ( PipelineAbility, + RoleSpec, register_ability, get_ability, list_abilities, get_ability_tools, get_ability_handler, + get_role_spec, + normalize_role, + get_next_role, + list_roles, ) from .slash import ( SlashCommand, diff --git a/pipeline_core/ability.py b/pipeline_core/ability.py index 81a8e56..eac3b89 100644 --- a/pipeline_core/ability.py +++ b/pipeline_core/ability.py @@ -19,9 +19,25 @@ from typing import Callable, Dict, List, Optional logger = logging.getLogger("pipeline.ability") +@dataclass +class RoleSpec: + """产线内角色定义（可插拔）。 + + 一个角色 = 工具白名单 + 专属 prompt + 专属模型 + 任务链位置 + 别名。 + 产线能力包通过 roles 字段定义自己的角色集（如 SDLC 的 requirement/design/develop/test/deploy）。 + """ + name: str # "develop" + description: str = "" # 角色职责描述 + aliases: List[str] = field(default_factory=list) # ["dev", "developer"] + system_prompt: str = "" # 角色专属 prompt（职责 + 产出要求） + next_role: str = "" # 任务链下一角色（空 = 终结） + model_name: str = "" # 角色专属模型（空 = 继承产线） + tools: List[str] = field(default_factory=list) # 工具白名单（空 = 产线全部工具） + + @dataclass class PipelineAbility: - """产线能力包：一个产线 = 一组工具 + 一段 prompt + 一组 handler。 + """产线能力包：一个产线 = 一组工具 + 一段 prompt + 一组 handler + 一组角色。 handlers 签名统一为：async def handler(sor, params, ctx) -> str - sor: 已打开的 pipeline DB context（由 AgentExecutor 统一管理） @@ -33,6 +49,7 @@ class PipelineAbility: tools: List = field(default_factory=list) # [ToolDefinition] system_prompt: str = "" # 产线专属 prompt 片段（追加到通用心智后） handlers: Dict[str, Callable] = field(default_factory=dict) # {tool_name: handler} + roles: List[RoleSpec] = field(default_factory=list) # 产线角色集 # ── 全局注册表 ── @@ -71,3 +88,37 @@ def get_ability_handler(pipeline_id: str, tool_name: str) -> Optional[Callable]: if not a: return None return a.handlers.get(tool_name) + + +# ── 角色查询 ── + +def get_role_spec(pipeline_id: str, role: str) -> Optional[RoleSpec]: + """取某产线某角色的定义（按 name 或 alias 匹配）。""" + a = get_ability(pipeline_id) + if not a or not role: + return None + r = (role or "").strip().lower() + for spec in a.roles: + if spec.name.lower() == r or r in [x.lower() for x in spec.aliases]: + return spec + return None + + +def normalize_role(pipeline_id: str, role: str) -> str: + """角色标准化：别名 → 规范名。能力包未定义时原样返回（lower）。""" + spec = get_role_spec(pipeline_id, role) + if spec: + return spec.name + return (role or "").strip().lower() + + +def get_next_role(pipeline_id: str, role: str) -> str: + """任务链下一角色（空 = 终结）。""" + spec = get_role_spec(pipeline_id, role) + return spec.next_role if spec else "" + + +def list_roles(pipeline_id: str) -> List[RoleSpec]: + """列出某产线的全部角色。""" + a = get_ability(pipeline_id) + return a.roles if a else [] diff --git a/pipeline_core/memory_store.py b/pipeline_core/memory_store.py index e92f72a..b42dbc3 100644 --- a/pipeline_core/memory_store.py +++ b/pipeline_core/memory_store.py @@ -25,6 +25,8 @@ class MemoryEntry: 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 priority: int = 0 # 优先级（越高越重要） created_at: float = 0.0 updated_at: float = 0.0 @@ -35,6 +37,8 @@ class MemoryEntry: "key": self.key, "content": self.content, "category": self.category, + "scope": self.scope, + "scope_id": self.scope_id, "priority": self.priority, "created_at": self.created_at, "updated_at": self.updated_at, @@ -47,6 +51,8 @@ class MemoryEntry: key=data.get("key", ""), content=data.get("content", ""), category=data.get("category", "memory"), + scope=data.get("scope", "global"), + scope_id=data.get("scope_id", ""), priority=data.get("priority", 0), created_at=data.get("created_at", time.time()), updated_at=data.get("updated_at", time.time()), @@ -82,8 +88,8 @@ class MemoryStore: db = DBPools() async with db.sqlorContext("pipeline") as sor: recs = await sor.sqlExe( - "SELECT memory_key, content, category, priority, created_at, updated_at, access_count " - "FROM pipeline_user_memory ORDER BY priority DESC, updated_at DESC LIMIT 200", + "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", {}, ) for r in (recs or []): @@ -91,6 +97,8 @@ class MemoryStore: key=getattr(r, "memory_key", ""), content=getattr(r, "content", ""), category=getattr(r, "category", "memory"), + scope=getattr(r, "scope", "global"), + scope_id=getattr(r, "scope_id", ""), priority=getattr(r, "priority", 0), created_at=_ts(getattr(r, "created_at", None)), updated_at=_ts(getattr(r, "updated_at", None)), @@ -105,8 +113,12 @@ class MemoryStore: self._cache_loaded = True async def add(self, content: str, category: str = "memory", - priority: int = PRIORITY_MEDIUM, key: str = None): - """添加一条记忆。自动生成 key（取内容前80字符的哈希）。""" + priority: int = PRIORITY_MEDIUM, key: str = None, + scope: str = "global", scope_id: str = ""): + """添加一条记忆。自动生成 key（取内容前80字符的哈希）。 + + scope 分域：global（通用）/ pipeline（产线）/ project（项目）/ user（用户个人）。 + """ await self._ensure_cache() if not key: @@ -117,6 +129,8 @@ class MemoryStore: key=key, content=content, category=category, + scope=scope, + scope_id=scope_id, priority=priority, created_at=now, updated_at=now, @@ -136,8 +150,12 @@ class MemoryStore: # 淘汰低优先级旧条目 await self._evict(max_entries=100) - async def get(self, category: str = None, key: str = None) -> List[MemoryEntry]: - """获取记忆。可按分类和 key 筛选。""" + async def get(self, category: str = None, key: str = None, + scope: str = None, scope_id: str = None) -> List[MemoryEntry]: + """获取记忆。可按分类/key/scope 筛选。 + + scope 过滤：scope=None 返回全部；scope 指定时返回该 scope 且 scope_id 匹配（或 scope_id=None 时该 scope 全部）。 + """ await self._ensure_cache() results = [] @@ -147,6 +165,11 @@ class MemoryStore: 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 @@ -171,15 +194,30 @@ class MemoryStore: except Exception as e: logger.error(f"MemoryStore remove failed: {e}") - async def build_prompt_block(self, category: str = None, max_entries: int = 20) -> str: + async def build_prompt_block(self, category: str = None, max_entries: int = 20, + scope: str = None, scope_id: str = None) -> str: """构建注入 system prompt 的记忆段落。 + 指定 scope 时加载「global（通用）+ 指定 scope（专属）」的记忆， + 实现通用记忆 + 产线/项目专属记忆的叠加注入。 高优先级记忆注入完整内容，低优先级只注入摘要。 """ - entries = await self.get(category) + 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()) + else: + entries = await self.get(category) + if not entries: return "" + entries.sort(key=lambda e: e.priority, reverse=True) + blocks = [] high_priority = [e for e in entries if e.priority >= self.PRIORITY_MEDIUM] low_priority = [e for e in entries if e.priority < self.PRIORITY_MEDIUM] @@ -194,14 +232,15 @@ class MemoryStore: return "\n".join(blocks) if blocks else "" async def _db_upsert(self, entry: MemoryEntry): - """写入/更新 DB""" + """写入/更新 DB（同 key + category + scope + scope_id 才 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}$", - {"k": entry.key, "c": entry.category}, + "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}, ) if existing: rid = existing[0].id @@ -220,6 +259,8 @@ class MemoryStore: "memory_key": entry.key, "content": entry.content, "category": entry.category, + "scope": entry.scope, + "scope_id": entry.scope_id, "priority": entry.priority, })