内容
# Aegis 数据流（集成数据工程代理）
**将自然语言目标转化为已验证、监控和人工批准的管道 DAG。**
Aegis 数据流是一种自主数据工程 AI 代理，它通过 [模型上下文协议（MCP）](https://modelcontextprotocol.io) 将 9 个开放工具（[OpenIngest](https://github.com/shubhmartin107-web/openingest)、[OpenCDC](https://github.com/shubhmartin107-web/opencdc)、[OpenLake](https://github.com/shubhmartin107-web/openlake)、[OpenStream](https://github.com/shubhmartin107-web/openstream)、[OpenPipe](https://github.com/shubhmartin107-web/openpipe)、[OpenQuality](https://github.com/shubhmartin107-web/openquality)、[OpenCatalog](https://github.com/shubhmartin107-web/opencatalog)、[OpenVectorDB](https://github.com/shubhmartin107-web/openvectordb)、[OpenFeatureStore](https://github.com/shubhmartin107-web/openfeaturestore)）集成到一个统一的调度系统中。
## 功能
- **🤖 AI 驱动的管道生成** — 用自然语言描述目标；LLM 计划器生成多步骤管道 DAG
- **⚡ DAG 调度** — 并行波执行，拓扑排序，可配置重试，指数退避和每阶段超时
- **👁️ 人工参与** — 在关键阶段，审批门暂停执行，进行人工确认
- **🔌 MCP 工具总线** — 通过 stdio、HTTP 和自定义协议适配器，为所有工具提供统一接口
- **⏰ 调度执行** — 基于 Cron 的触发系统，用于循环管道
- **🔐 API 认证** — 基于密钥的认证，SHA-256 哈希加密
- **📜 管道版本控制** — 完整的版本历史，支持恢复
- **🔗 Webhook 通知** — 管道完成/失败的 HMAC 签名 Webhook
- **🔁 幂等运行** — 幂等键，防止重复管道执行
- **🔄 并发锁定** — 自动防止重复管道运行
- **📦 导入/导出** — 基于 JSON 的管道可移植性
- **📊 实时 SSE 事件** — 将执行事件实时流式传输到 Web UI
- **🛡️ OpenTelemetry** — 分布式跟踪和指标
## 架构
```
┌─────────────────────────────────────────────────────────┐
│ Aegis 数据流 │
├─────────────────────────────────────────────────────────┤
│ 用户输入（目标 / API / 调度） │
│ │ │
│ ┌──────▼──────┐ ┌──────────────────┐ │
│ │ LLM 计划器 │─▶│ DAG 执行器 │ │
│ │ （Python） │ │ （Go） │ │
│ └──────────────┘ └───────┬──────────┘ │
│ │ │
│ ┌─────────────────────────▼──────────────────────────┐ │
│ │ MCP 工具总线 │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ stdio │ │ HTTP │ │ 自定义 │ │ │
│ │ │ 适配器 │ │ 适配器 │ │ 适配器 │ │ │
│ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │
│ └───────┼───────────┼───────────┼──────────────────────┘ │
│ │ │ │ │
│ ┌───────▼───┐ ┌────▼────┐ ┌───▼────────┐ │
│ │ OpenIngest│ │ OpenLake│ │ OpenCDC │ ... 9 个工具 │
│ └───────────┘ └─────────┘ └────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 状态存储（SQLite） │ │
│ │ 管道 │ 版本 │ 运行 │ 审批门 │ │
│ │ Webhook │ 调度 │ API 密钥 │ 工具注册 │ │
│ └────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ Web UI（React + TypeScript） │ │
│ │ 仪表盘 │ 管道图 │ 目标输入 │ 日志 │ │
│ └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
```
## 快速入门
### 前置条件
- Go 1.24+ ([安装](https://go.dev/dl/))
- Python 3.10+，用于 LLM 计划器
- [Ollama](https://ollama.ai)，附带模型（默认：`qwen2.5:7b`）
- Node.js 20+，用于 Web UI（可选）
### 1. 克隆和构建
```bash
git clone https://github.com/shubhmartin107-web/aegis-dataflow.git
cd aegis-dataflow
make build
```
### 2. 配置
```bash
cp config.example.json config.json
# 编辑 config.json 以指向您的 MCP 工具服务器
# 或设置环境变量：
export AEGIS_OLLAMA_ENDPOINT=http://localhost:11434
export AEGIS_PLANNER_MODEL=qwen2.5:7b
```
### 3. 运行
```bash
# 启动服务器
./bin/aegis
# 使用配置：
./bin/aegis -config config.json
```
服务器启动在 `http://0.0.0.0:9876`。
## 配置
| 环境变量 | 默认值 | 描述 |
|---|---|---|
| `AEGIS_HOST` | `0.0.0.0` | 服务器绑定地址 |
| `AEGIS_PORT` | `9876` | 服务器端口 |
| `AEGIS_OLLAMA_ENDPOINT` | `http://localhost:11434` | Ollama API 端点 |
| `AEGIS_PLANNER_MODEL` | `qwen2.5:7b` | 用于管道规划的 LLM 模型 |
| `AEGIS_AUTH_ENABLED` | `false` | 启用 API 密钥认证 |
| `AEGIS_AI_API_KEY` | — | 外部 AI 提供商的 API 密钥 |
## API 参考
### 管道
| 方法 | 路径 | 描述 |
|---|---|---|
| `GET` | `/api/pipelines` | 列出管道（`?status=draft\|active`） |
| `POST` | `/api/pipelines` | 从 JSON 创建管道 |
| `GET` | `/api/pipelines/{id}` | 获取管道详情 |
| `PUT` | `/api/pipelines/{id}` | 更新管道 |
| `DELETE` | `/api/pipelines/{id}` | 删除管道 |
| `POST` | `/api/pipelines/{id}/run` | 执行管道（支持 `idempotency_key`） |
| `GET` | `/api/pipelines/{id}/runs` | 列出管道运行 |
| `GET` | `/api/pipelines/{id}/versions` | 列出版本历史 |
| `POST` | `/api/pipelines/{id}/restore/{v}` | 恢复以前的版本 |
| `GET` | `/api/pipelines/{id}/export` | 以 JSON 导出管道 |
### 目标
| 方法 | 路径 | 描述 |
|---|---|---|
| `POST` | `/api/goals` | 提交自然语言目标进行规划 |
### 审批
| 方法 | 路径 | 描述 |
|---|---|---|
| `GET` | `/api/approvals/{runID}` | 获取待处理审批门 |
| `POST` | `/api/approvals/{gateID}/approve` | 批准某个阶段 |
| `POST` | `/api/approvals/{gateID}/reject` | 拒绝某个阶段 |
### Webhook 和调度
| 方法 | 路径 | 描述 |
|---|---|---|
| `POST/GET/DELETE` | `/api/pipelines/{id}/webhooks` | Webhook CRUD |
| `POST/GET/DELETE` | `/api/pipelines/{id}/schedules` | Cron 调度 CRUD |
### 管理
| 方法 | 路径 | 描述 |
|---|---|---|
| `POST/GET/DELETE` | `/api/api-keys` | API 密钥管理 |
| `POST` | `/api/import` | 从 JSON 导入管道 |
| `GET` | `/api/events` | 执行事件的 SSE 流 |
### MCP 接口
| 方法 | 路径 | 描述 |
|---|---|---|
| `GET` | `/mcp` | MCP SSE 端点 |
| `POST` | `/mcp` | MCP JSON-RPC 端点 |
## 开发
```bash
# 构建
make build
# 测试
make test
# 运行所有检查
make lint
# 构建 Web UI
make ui-build
```
## 技术栈
| 层 | 语言 | 库 |
|---|---|---|
| 调度 | Go | chi, go-sqlite3, dominikbraun/graph |
| AI 计划器 | Python | ollama, pydantic |
| Web UI | TypeScript/React | React Flow, Vite |
| 存储 | SQLite | WAL 模式，外键 |
| 可观察性 | Go | OpenTelemetry |
| 协议 | — | MCP 2025-11-25, JSON-RPC 2.0 |
## 许可证
Apache 2.0 — 请参阅 [LICENSE](LICENSE)。
## 项目状态
v0.1.0 — 核心规范完成。所有 9 个开放工具均通过 MCP 适配器支持。准备好进行本地部署。
