54 lines
2.4 KiB
Markdown
54 lines
2.4 KiB
Markdown
---
|
|
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
|