diff --git a/pipeline_core/__init__.py b/pipeline_core/__init__.py index da825ef..84033e8 100644 --- a/pipeline_core/__init__.py +++ b/pipeline_core/__init__.py @@ -20,11 +20,22 @@ from .agent_config import ( MemoryConfig, SkillConfig, ToolDefinition, + GENERAL_TOOLS, + DEFAULT_AGENT_CONFIG, + DEFAULT_ABILITY_ID, SDLC_DEFAULT_CONFIG, SDLC_DEFAULT_TOOLS, load_agent_config, save_agent_config, ) +from .ability import ( + PipelineAbility, + register_ability, + get_ability, + list_abilities, + get_ability_tools, + get_ability_handler, +) from .tool_registry import ( ToolRegistry, get_tool_registry, diff --git a/pipeline_core/ability.py b/pipeline_core/ability.py new file mode 100644 index 0000000..81a8e56 --- /dev/null +++ b/pipeline_core/ability.py @@ -0,0 +1,73 @@ +""" +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) diff --git a/pipeline_core/agent_config.py b/pipeline_core/agent_config.py index 92d517a..7b3727c 100644 --- a/pipeline_core/agent_config.py +++ b/pipeline_core/agent_config.py @@ -175,10 +175,15 @@ class AgentConfig: # ═══════════════════════════════════════════════════════════ -# SDLC 默认配置 +# 通用 agent 内核（产线无关） +# +# GENERAL_TOOLS：任何产线 agent 都具备的基础能力 +# （项目管理/终端/文件/搜索/会话/规划/澄清/委派）。 +# 产线专属能力（如 SDLC 的 create_task/diagnose_project）以「能力包」形式 +# 独立注册，见 pipeline_service.sdlc_ability，通过 PipelineAbility 挂载。 # ═══════════════════════════════════════════════════════════ -SDLC_DEFAULT_TOOLS = [ +GENERAL_TOOLS = [ ToolDefinition( name="switch_project", description="切换到指定项目。用户说「切换到XXX」时调用。", @@ -191,82 +196,6 @@ SDLC_DEFAULT_TOOLS = [ parameters={"name": "项目名称", "description": "项目描述"}, category="project", ), - ToolDefinition( - name="create_task", - description="创建开发任务，分配给角色agent执行", - parameters={ - "title": "任务标题", - "role": "目标角色(requirement/design/develop/test/deploy)", - "description": "任务详细描述", - }, - category="task", - ), - ToolDefinition( - name="list_tasks", - description="查看当前项目的任务列表", - parameters={"role": "按角色筛选(可选)", "state": "按状态筛选(可选)"}, - category="task", - ), - ToolDefinition( - name="task_detail", - description="查看任务详情", - parameters={"task_id": "任务ID"}, - category="task", - ), - ToolDefinition( - name="start_agents", - description="启动角色agent执行已提交的任务", - parameters={}, - category="agent", - ), - ToolDefinition( - name="diagnose_project", - description="诊断项目：汇总卡点/失败/问题/运行中任务", - parameters={}, - category="agent", - ), - ToolDefinition( - name="list_deliverables", - description="查看交付件列表", - parameters={"role": "按角色筛选(可选)"}, - category="task", - ), - ToolDefinition( - name="view_deliverable", - description="查看交付件详细内容", - parameters={"deliverable_id": "交付件ID"}, - category="task", - ), - ToolDefinition( - name="list_questions", - description="查看待回答的问题", - parameters={}, - category="agent", - ), - ToolDefinition( - name="answer_question", - description="回答agent提出的问题", - parameters={"question_id": "问题ID", "answer": "回答内容"}, - category="agent", - ), - ToolDefinition( - name="add_repo", - description="添加项目关联的Git仓库", - parameters={"repo_url": "仓库URL", "repo_name": "仓库名称", "default_branch": "默认分支(默认main)"}, - category="repo", - ), - ToolDefinition( - name="list_repos", - description="查看项目关联的仓库列表", - parameters={}, - category="repo", - ), - ToolDefinition( - name="clone_repo", - description="克隆项目关联的仓库到工作空间repos/下（不传url则clone全部，传则clone指定）", - parameters={"repo_url": "仓库URL(可选)"}, - category="repo", - ), ToolDefinition( name="run_command", description="在工作空间中执行shell命令", @@ -274,18 +203,6 @@ SDLC_DEFAULT_TOOLS = [ category="shell", requires_confirmation=True, ), - ToolDefinition( - name="check_progress", - description="查看项目整体进展", - parameters={}, - category="agent", - ), - ToolDefinition( - name="add_bug", - description="提交Bug", - parameters={"title": "Bug标题", "description": "描述", "severity": "严重程度"}, - category="task", - ), # ── v2 新增 ── ToolDefinition( name="ask_user", @@ -339,36 +256,26 @@ SDLC_DEFAULT_TOOLS = [ ] -SDLC_DEFAULT_CONFIG = AgentConfig( +DEFAULT_AGENT_CONFIG = AgentConfig( model_name="deepseek-v4-pro", temperature=0.4, max_turns=30, - system_prompt="""你是 SDLC 开发产线的驾驶舱 agent。你的职责：在自动化项目实施过程中，主动发现问题、定位根因、用工具解决问题，推动项目前进。 - -## 你的身份 -你是一个具备通用推理与判断能力的 agent，在此之上额外配备了项目实施工具。像一名有经验的工程负责人那样思考：先理解意图，再拆解问题，判断自己能否解决，必要时澄清或诚实说明。 + system_prompt="""你是一个通用 agent，具备通用推理与判断能力，并根据当前产线配备了对应的工具。像一名有经验的负责人那样思考：先理解意图，再拆解问题，判断自己能否解决，必要时澄清或诚实说明。 ## 工作原则 1. 先理解用户意图，再决定行动。意图模糊时用 ask_user 澄清，不要臆测。 2. 用工具是手段，不是目的。工具能解决就用工具，用工具只是为了把事做成。 -3. 能力自省与诚实降级：如果用户的请求超出你的工具能力（例如"重做已完成任务"，但你没有重置任务状态的工具），不要硬套一个不相关的工具。你必须： +3. 能力自省与诚实降级：如果用户的请求超出你的工具能力，不要硬套一个不相关的工具。你必须： a) 明确告诉用户：你做不到、缺什么能力、为什么； - b) 给出你能做到的替代方案（如用 create_task 新建重做任务）； + b) 给出你能做到的替代方案； c) 必要时 ask_user 让用户拍板。 -4. 发现项目卡点（任务卡死、审核超时、失败、僵尸 claimed_by、等待无响应）时，主动定位根因并推动修复，而不是只列一张诊断清单。 +4. 发现异常/卡点时，主动定位根因并推动解决，而不是只列清单。 5. 每轮输出 tool_call / reply / ask_user 三者之一，根据实际情况选择，没有任何强制。 -## 典型场景（帮助判断何时用哪个工具，非强制） -- 用户提出新的开发需求/功能（"实现XX""XX系统要做XX"）→ 通常用 create_task 创建任务（可按 requirement→design→develop 拆分），再 start_agents 启动。 -- 用户问进展/状态 → list_tasks / diagnose_project / check_progress。 -- 用户问待回答问题 → list_questions / answer_question。 -- 用户报告异常/故障 → 先用 run_command / diagnose_project 定位根因，再决定修复动作。 -- 用户要求做你工具能力之外的事 → 按"能力自省"原则诚实说明 + 给替代方案 + 必要时 ask_user。 - ## 工具集 {tools_description} """, - tools=SDLC_DEFAULT_TOOLS, + tools=GENERAL_TOOLS, session_isolation="project", history_limit=50, compression=CompressionConfig(enabled=True, threshold=0.50, target_ratio=0.20, keep_recent=6), @@ -377,18 +284,56 @@ SDLC_DEFAULT_CONFIG = AgentConfig( ) +# 兼容别名（过渡期）：SDLC 能力已迁移到 pipeline_service.sdlc_ability 能力包， +# 此处保留别名以免破坏既有引用，后续可删除。 +SDLC_DEFAULT_TOOLS = GENERAL_TOOLS +SDLC_DEFAULT_CONFIG = DEFAULT_AGENT_CONFIG + +# 默认产线：pipeline_id 为空时回退的能力包（对应 pipelines 表的「通用软件开发产线」） +DEFAULT_ABILITY_ID = "sdlc_general" + + # ═══════════════════════════════════════════════════════════ # 配置加载器 # ═══════════════════════════════════════════════════════════ async def load_agent_config(pipeline_id: str = None, project_id: str = None) -> AgentConfig: - """加载产线的 Agent 配置。 + """加载产线的 Agent 配置（通用内核 + 可插拔产线能力）。 - 优先级： - 1. 项目级 sd_org_settings.agent_config（sd_org_settings 以 org_id 为键，由项目解析 org_id） - 2. 产线级 pipelines.agent_config - 3. SDLC_DEFAULT_CONFIG + 组装顺序： + 1. 基础 = DEFAULT_AGENT_CONFIG（通用心智 + GENERAL_TOOLS） + 2. 产线能力 = get_ability(pipeline_id) 动态挂载（工具 + prompt 片段） + —— pipeline_id 为空时回退 DEFAULT_ABILITY_ID（默认产线） + 3. DB 覆盖：sd_org_settings.agent_config / pipelines.agent_config """ + from .ability import get_ability + + # 1. 基础通用配置 + base = DEFAULT_AGENT_CONFIG + + # 2. 产线能力挂载 + ability = get_ability(pipeline_id) if pipeline_id else get_ability(DEFAULT_ABILITY_ID) + tools = list(GENERAL_TOOLS) + prompt = base.system_prompt + if ability: + tools = _merge_tools(ability.tools, tools) + if ability.system_prompt: + prompt += "\n\n## 产线专属能力\n" + ability.system_prompt + + cfg = AgentConfig( + model_name=base.model_name, + temperature=base.temperature, + max_turns=base.max_turns, + system_prompt=prompt, + tools=tools, + session_isolation=base.session_isolation, + history_limit=base.history_limit, + compression=base.compression, + memory=base.memory, + skills=base.skills, + ) + + # 3. DB 覆盖（项目级 / 产线级） try: from sqlor.dbpools import DBPools @@ -411,7 +356,7 @@ async def load_agent_config(pipeline_id: str = None, project_id: str = None) -> if raw: try: data = json.loads(raw) if isinstance(raw, str) else raw - data["tools"] = _merge_tools(data.get("tools", []), SDLC_DEFAULT_TOOLS) + data["tools"] = _merge_tools(data.get("tools", []), tools) return AgentConfig.from_dict(data) except (json.JSONDecodeError, TypeError): pass @@ -428,22 +373,18 @@ async def load_agent_config(pipeline_id: str = None, project_id: str = None) -> if raw: try: data = json.loads(raw) if isinstance(raw, str) else raw - data["tools"] = _merge_tools(data.get("tools", []), SDLC_DEFAULT_TOOLS) - # 产线缺省模型：agent_config 未显式设 model_name 时用 default_model 兜底 + data["tools"] = _merge_tools(data.get("tools", []), tools) if not data.get("model_name") and default_model: data["model_name"] = default_model return AgentConfig.from_dict(data) except (json.JSONDecodeError, TypeError): pass elif default_model: - # agent_config 为空，仅产线设了缺省模型 - cfg = AgentConfig.from_dict(SDLC_DEFAULT_CONFIG.to_dict()) cfg.model_name = default_model - return cfg except Exception: pass - return SDLC_DEFAULT_CONFIG + return cfg def _merge_tools(custom_tools: list, default_tools: list) -> list: