132 lines
5.0 KiB
Python
132 lines
5.0 KiB
Python
"""
|
||
pipeline-core: PipelineAbility — 产线能力包注册表
|
||
|
||
可插拔产线能力：core 只提供通用 agent 内核（通用工具 + 通用心智），
|
||
每个产线（SDLC 开发、未来 A/B 产线）以「能力包」形式独立注册，
|
||
包含该产线专属的工具定义、prompt 片段、工具 handler。
|
||
|
||
AgentExecutor 按 pipeline_id 从注册表动态挂载能力 → 多实例能力各异。
|
||
|
||
依赖方向：
|
||
pipeline-core （本文件，纯定义 + 注册表，不依赖任何产线）
|
||
pipeline-service （sdlc_ability.py 注册 SDLC 能力，依赖 core + 底层函数）
|
||
"""
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
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 + 一组角色。
|
||
|
||
handlers 签名统一为：async def handler(sor, params, ctx) -> str
|
||
- sor: 已打开的 pipeline DB context（由 AgentExecutor 统一管理）
|
||
- params: 工具参数 dict
|
||
- ctx: {"project_id", "user_id", "workspace_dir", "model_name", "config"}
|
||
"""
|
||
pipeline_id: str # 产线标识（对应 pipelines 表 id）
|
||
name: str = "" # 产线名称
|
||
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) # 产线角色集
|
||
menus: List[Dict] = field(default_factory=list) # 产线功能菜单（AgentIO 上方）[{"label","icon","url","type"}]
|
||
|
||
|
||
# ── 全局注册表 ──
|
||
|
||
_abilities: Dict[str, PipelineAbility] = {}
|
||
|
||
|
||
def register_ability(ability: PipelineAbility):
|
||
"""注册一个产线能力包（幂等：重复注册覆盖）。"""
|
||
_abilities[ability.pipeline_id] = ability
|
||
logger.info(f"PipelineAbility registered: {ability.pipeline_id} "
|
||
f"({len(ability.tools)} tools, {len(ability.handlers)} handlers)")
|
||
|
||
|
||
def get_ability(pipeline_id: str) -> Optional[PipelineAbility]:
|
||
"""按 pipeline_id 取能力包。"""
|
||
if not pipeline_id:
|
||
return None
|
||
return _abilities.get(pipeline_id)
|
||
|
||
|
||
def list_abilities() -> List[PipelineAbility]:
|
||
"""列出所有已注册能力包。"""
|
||
return list(_abilities.values())
|
||
|
||
|
||
def get_ability_tools(pipeline_id: str) -> List:
|
||
"""取某产线的专属工具定义列表。"""
|
||
a = get_ability(pipeline_id)
|
||
return a.tools if a else []
|
||
|
||
|
||
def get_ability_handler(pipeline_id: str, tool_name: str) -> Optional[Callable]:
|
||
"""取某产线某工具的 handler。"""
|
||
a = get_ability(pipeline_id)
|
||
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 []
|
||
|
||
|
||
def get_ability_menus(pipeline_id: str) -> List[Dict]:
|
||
"""取某产线的功能菜单（AgentIO 上方，菜单/卡片/按钮）。"""
|
||
a = get_ability(pipeline_id)
|
||
return a.menus if a else []
|