diff --git a/README.md b/README.md index ef9b567..cbf03fe 100644 --- a/README.md +++ b/README.md @@ -1,202 +1,41 @@ -# pipeline-service v3.0 — 通用产线执行引擎 +# pipeline_service -## 定位 +产线执行引擎 —— 任务调度、步骤执行、人工任务交互、LLM 桥接。 -通用产线执行引擎模块。把 Hermes Agent 验证过的业务流程固化为可重复、可并发的产线业务环境。 +## 功能 -**v3.0 新增:** 人工交互步骤(human_task/approval_gate)、step_type 可装卸注册、多角色协作。 - -## 核心价值 - -- Hermes Agent 中用 cron/delegate/terminal 跑通的流程 → 固化为产线步骤定义 → pipeline-service 自动调度执行 -- 一次验证,无限次自动执行 -- 多租户并发:同一产线,不同租户同时使用,数据完全隔离 -- **人机协作:** 自动步骤 + 人工步骤混合执行,支持审批驳回回退 - -## 架构 - -``` -宿主应用 (pipeline-app / 其他) - │ - ├── load_pipeline_service() ← 注册函数到 ServerEnv - │ - ├── 任务生命周期 - │ ├── pipeline_submit(tenant_id, pipeline_id, owner_id, title, params) - │ ├── pipeline_list(tenant_id, pipeline_id?) - │ ├── pipeline_detail(tenant_id, task_id) - │ ├── pipeline_node(tenant_id, task_id, step_name, version?) - │ ├── pipeline_modify(tenant_id, task_id, updates, rerun_from) - │ ├── pipeline_pause(tenant_id, task_id) - │ ├── pipeline_resume(tenant_id, task_id) - │ └── pipeline_cancel(tenant_id, task_id) - │ - ├── 步骤类型注册(可装卸) - │ ├── pipeline_step_types() ← 列出所有类型 - │ ├── pipeline_register_step_type(type, meta) ← 注册新类型 - │ └── pipeline_unregister_step_type(type) ← 卸载类型 - │ - ├── 人工交互 - │ ├── human_task_complete(tenant_id, task_id, step_name, data, operator) - │ ├── approval_approve(tenant_id, task_id, step_name, reviewer, comments) - │ ├── approval_reject(tenant_id, task_id, step_name, reviewer, comments, rollback_to) - │ └── human_task_list(tenant_id?, role?, user?, status?) - │ - └── Handler管理 - └── pipeline_register_handler(step_type, fn) -``` - -## 引擎工作原理 - -1. **提交任务** → 读取 pipeline_steps 表的步骤定义 → 创建 pipeline_task_steps 记录 → 启动执行 -2. **执行循环** → 解析 DAG 依赖图 → 找到可执行步骤 → 判断步骤类型: - - **自动步骤:** 调用 handler → 存 artifact → 继续下一步 - - **交互步骤(human_task/approval_gate):** 创建 human_tasks 记录 → 步骤进入 waiting → 任务进入 waiting -3. **人工完成** → 调用 human_task_complete/approval_approve → 步骤标记完成 → 恢复执行 -4. **审批驳回** → 调用 approval_reject(rollback_to=步骤名) → 回退指定步骤 → 级联重跑 -5. **多租户** → 所有查询按 tenant_id 隔离 - -## 状态机 - -### 任务状态 -``` -submitted → running → completed - → failed - → paused → running (resume) - → waiting → running (human complete) - → cancelled -``` - -### 步骤状态 -``` -pending → running → completed - → failed - → skipped - → waiting → completed (human complete) - → rejected (approval reject) -``` - -## step_type 可装卸 - -每条产线可以注册自己的 step_type,引擎按类型匹配 handler 和交互协议。 - -```python -# 注册一个 SDLC 产线的步骤类型 -from pipeline_service import register_step_type, register_handler - -# 注册自动步骤 -register_step_type("code_review_auto", { - "display_name": "自动代码审查", - "category": "devops", - "is_interactive": False, -}) - -# 注册人工步骤 -register_step_type("code_review_manual", { - "display_name": "人工代码审查", - "category": "interactive", - "is_interactive": True, - "form_schema": { - "type": "object", - "properties": { - "approved": {"type": "boolean", "title": "是否通过"}, - "comments": {"type": "string", "title": "审查意见"} - } - }, - "timeout_hours": 48, - "on_timeout": "escalate", -}) - -# 注册 handler(自动步骤需要,交互步骤不需要) -register_handler("code_review_auto", auto_review_handler) -``` - -### 内置交互类型 - -| step_type | 用途 | 行为 | -|-----------|------|------| -| human_task | 人工填写表单/执行操作 | 步骤 waiting → 人提交 → 继续 | -| approval_gate | 审批关卡 | 步骤 waiting → 通过继续 / 驳回回退 | - -### 步骤定义中的配置 - -在 pipeline_steps 表的 step_config JSON 中指定交互参数: - -```json -{ - "deps": ["develop"], - "assignee_role": "reviewer", - "assignee_id": "user123", - "form_schema": {"type": "object", "properties": {...}}, - "timeout_hours": 48, - "on_timeout": "escalate" -} -``` +- **任务执行**:DAG 步骤调度与状态机 +- **人工任务**:审批/输入等待与交互 +- **LLM 桥接**:统一 LLM 调用接口 +- **Agent Loop**:AI Agent 多轮任务执行 +- **意图分类**:自然语言意图识别 +- **产物管理**:步骤输入输出存储 ## 数据表 -| 表名 | 用途 | -|------|------| -| pipeline_tasks | 任务主表(tenant_id 隔离) | -| pipeline_task_steps | 任务步骤执行记录 | -| pipeline_artifacts | 步骤产物(input/output,支持版本) | -| **pipeline_human_tasks** | **人工任务记录(v3新增)** | -| **pipeline_step_types** | **步骤类型注册表(v3新增)** | -| pipeline_steps | 产线步骤定义(由 pipeline_core 模块管理) | -| pipelines | 产线定义(由 pipeline_core 模块管理) | +| 表 | 说明 | +|---|------| +| pipeline_tasks | 任务实例 | +| pipeline_task_steps | 步骤执行记录 | +| pipeline_artifacts | 步骤产物 | +| pipeline_human_tasks | 人工任务 | +| pipeline_step_types | 步骤类型注册 | -## 宿主集成 - -任何应用只需一行代码即可使用: - -```python -from pipeline_service.init import load_pipeline_service -load_pipeline_service() -``` - -宿主负责: -- HTTP 路由(ahserver 管) -- 用户认证(RBAC 管) -- 前端交互(bricks 管) -- 产线定义和定价(pipeline_core/ops/dist 管) - -pipeline-service 只做:调度 + 执行 + 存储 + 人工交互。 - -## 目录结构 - -``` -pipeline-service/ -├── pipeline_service/ -│ ├── __init__.py # 包导出 -│ ├── init.py # load_pipeline_service() + ServerEnv 注册 -│ ├── state.py # DAG 解析、步骤状态机 -│ ├── handler.py # 步骤处理器注册表 -│ ├── step_registry.py # 步骤类型注册表(v3新增) -│ ├── human.py # 人工任务操作(v3新增) -│ ├── storage.py # MySQL 存储层(sqlor) -│ ├── executor.py # 执行循环 -│ └── handlers_ktv.py # KTV产线专用 handlers -├── models/ -│ ├── pipeline_tasks.json -│ ├── pipeline_task_steps.json -│ ├── pipeline_artifacts.json -│ ├── pipeline_human_tasks.json (v3新增) -│ └── pipeline_step_types.json (v3新增) -├── init/ -│ └── data.json # appcodes 初始化数据 -├── pyproject.toml -└── README.md -``` - -## 构建与部署 +## 安装 ```bash -cd ~/repos/pipeline-service -pip install . - -# 建表 -json2ddl mysql models/ > mysql.ddl.sql -mysql -u root -p pipeline < mysql.ddl.sql - -# 加载 appcodes -# (通过宿主应用的 build.sh 自动加载 init/data.json) +cd pkgs/pipeline-service && pip install . ``` + +## 核心模块 + +| 文件 | 职责 | +|------|------| +| `executor.py` | 任务/步骤调度引擎 | +| `storage.py` | 数据库读写 | +| `llm_bridge.py` | LLM API 调用 | +| `agent_loop.py` | AI Agent 多轮执行 | +| `human.py` | 人工任务处理 | +| `intent_classifier.py` | 意图识别 | +| `state.py` | 状态机 | +| `step_registry.py` | 步骤类型注册表 | diff --git a/skill/SKILL.md b/skill/SKILL.md new file mode 100644 index 0000000..d1ce4f0 --- /dev/null +++ b/skill/SKILL.md @@ -0,0 +1,53 @@ +--- +name: pipeline-service +description: "Pipeline execution engine — task scheduling, step DAG execution, human task interaction, and LLM bridging." +--- + +# pipeline_service + +Core execution engine that drives pipeline task lifecycle: scheduling, DAG step execution, human-in-the-loop interaction, and LLM integration. + +## Architecture + +``` +pipeline_service/ +├── executor.py # Task/step scheduling engine +├── storage.py # Database CRUD (pipeline_tasks, task_steps, artifacts) +├── llm_bridge.py # Unified LLM API call interface +├── agent_loop.py # AI Agent multi-turn execution loop +├── human.py # Human task interaction (approval, input) +├── intent_classifier.py # Natural language intent recognition +├── state.py # Task/step state machine +├── step_registry.py # Step type handler registry +└── init.py # ServerEnv registration +``` + +## Data Model + +| Table | Purpose | +|-------|---------| +| `pipeline_tasks` | Task instances (id, pipeline_id, status, version, params) | +| `pipeline_task_steps` | Step execution records (task_id, step_name, state, input, output) | +| `pipeline_artifacts` | Step input/output artifacts | +| `pipeline_human_tasks` | Human-in-the-loop tasks (approval, input forms) | +| `pipeline_step_types` | Registered step type handlers | + +## Key Functions (registered via ServerEnv) + +- `submit_task()` — Create and start a new task +- `get_task_detail()` — Task state + all steps +- `get_task_steps()` — Steps for a task +- `control_task()` — Pause/resume/cancel +- `restart_task()` — Restart completed/failed task +- `list_tasks()` — Task list with filters +- `llm_call()` — LLM API call +- `call_llm()` — SDLC handler interface (delegates to llm_call) + +## Pitfalls + +- **DBPools init**: Must check `db.databases` and load from `config.databases` if empty. DBPools() is NOT a singleton. +- **sor.R() → sqlExe**: Never use `sor.R('table', where, sort)` 3-arg. Use `sqlExe("SELECT ... WHERE ... ORDER BY", params)`. +- **dir() → vars()**: Use `vars(rec)` not `dir(rec)` for attribute iteration — `dir()` includes methods. +- **dict access → getattr**: sqlor returns DictObject, not dict. Use `getattr(rec, 'field')` not `rec['field']`. +- **pipeline_service must be pip installed**: After git pull on server, run `pip install --upgrade`. +SKILLEOF