74 lines
2.7 KiB
Python
74 lines
2.7 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 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}
|
||
|
||
|
||
# ── 全局注册表 ──
|
||
|
||
_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)
|