生产就绪的 AI Agent 工作流编排器,支持 DAG 流程定义、多模型路由、可插拔代码沙箱及自动化管控,提供 Python SDK、CLI 与 MCP 协议支持。
项目概述#
Sandcastle 是一个专为生产环境设计的 AI 工作流编排框架,旨在解决 AI Agent 在生产落地时的编排、执行与管控难题。通过 YAML 定义的 DAG(有向无环图)工作流引擎,支持多步骤协作任务的自动化执行。
核心定位:"Stop babysitting your AI agents. Sandcastle runs your agent workflows so you don't have to."
作者:Tomas Pflanzer (@gizmax)
版本:v0.12.0(官网)/ v0.10.0(PyPI)
核心特性#
工作流引擎#
- DAG 编排:YAML 定义多步骤管道,支持依赖关系、并行分支、步骤间数据传递
- 并行执行:同一 DAG 层级步骤并发执行,支持
parallel_over对列表进行扇出 - 混合步骤类型:9 种步骤类型,支持 Workflow-as-Step(工作流嵌套调用)
多提供商模型路由#
- 支持模型:Claude (Opus/Sonnet/Haiku)、OpenAI (Codex/Codex-mini)、MiniMax (M2.5)、Google Gemini (via OpenRouter)
- 细粒度控制:每个步骤可指定不同模型(如 Opus 做深度研究,Haiku 做简单评分)
可插拔沙箱后端#
| 后端 | 适用场景 | 特点 |
|---|---|---|
| E2B | 云端默认 | 零基础设施设置 |
| Docker | 自托管 | 本地容器隔离 |
| Local | 开发测试 | 主机子进程,无隔离 |
| Cloudflare Workers | 边缘场景 | 全球分布式低延迟 |
生产级管控#
- 人工审批门控:暂停工作流等待人工审核,支持超时自动操作
- AutoPilot 自优化:A/B 测试不同模型/提示词,LLM-as-Judge 自动评估并部署最佳变体
- 策略引擎:声明式规则检测 PII、拦截敏感信息、注入动态审批
- 成本-延迟优化器:基于 SLO 的动态模型路由
- 预算护栏:每次运行/租户/全局成本限制
- 时间机器:从任意步骤重放或分叉
- 实时事件流:SSE 实时更新
集成与扩展#
- MCP Server:内置 Model Context Protocol 服务器,支持 Claude Desktop、Cursor、Windsurf
- 工具连接器:12 个内置集成(Slack、Jira、GitHub、HubSpot、Salesforce、Zendesk、Notion、Teams、Gmail、Google Drive、PostgreSQL、Webhooks)
- 定时调度:Cron 语法调度执行
安装与快速开始#
本地模式(推荐入门)#
# 安装
pip install sandcastle-ai
# 交互式设置向导
sandcastle init
# 启动服务(API + 仪表盘)
sandcastle serve
访问地址:仪表盘与 API 均在 http://localhost:8080
必需 API Keys:ANTHROPIC_API_KEY、E2B_API_KEY
生产部署#
git clone https://github.com/gizmax/Sandcastle.git
cd Sandcastle
# 配置环境变量
cat > .env << 'EOF'
ANTHROPIC_API_KEY=sk-ant-...
E2B_API_KEY=e2b_...
SANDBOX_BACKEND=e2b
DATABASE_URL=postgresql://...
REDIS_URL=redis://...
EOF
# 一键启动
docker compose up -d
API 与 SDK#
Python SDK#
from sandcastle import SandcastleClient
client = SandcastleClient(base_url="http://localhost:8080", api_key="sc_...")
# 运行工作流并等待完成
run = client.run("lead-enrichment",
input={"target_url": "https://example.com"},
wait=True,
)
print(run.status) # "completed"
print(run.total_cost_usd) # 0.12
# 从失败步骤分叉
new_run = client.fork(run.run_id, from_step="score", changes={"model": "opus"})
CLI 工具#
sandcastle run lead-enrichment -i target_url=https://example.com --wait
sandcastle logs <run-id> --follow
sandcastle schedule create lead-enrichment "0 9 * * *"
sandcastle ls runs --status completed
REST API#
curl -X POST http://localhost:8080/api/workflows/run \
-H "Content-Type: application/json" \
-d '{"workflow": "lead-enrichment", "input": {"target_url": "https://example.com"}}'
适用场景#
| 领域 | 典型应用 |
|---|---|
| 营销 | 博客内容转社交媒体帖子、SEO 审计、竞争对手分析 |
| 销售 | 线索丰富化与评分、外联序列生成、CRM 同步 |
| 支持 | 工单分类与优先级排序、知识库更新、情感分析 |
| HR | 简历筛选、入职清单生成 |
| 法务 | 合同审查与风险识别 |
| 运维 | 技术债务审计、事故事后分析 |
注意事项#
⚠️ 许可证不一致:GitHub 仓库显示 BSL 1.1(2030年转 Apache 2.0),PyPI 包显示 MIT License,需确认实际适用许可证。