第一章：为什么需要 AI 应用商店？
1.1 当前 AI 能力交付的困境
- 数据科学家 | “我训练了 SOTA 模型，但没人会用”
- 开发者 | “每次都要重写 API 封装，重复造轮子”
- 企业用户 | “想用外部 AI 能力，但集成复杂、无保障”
1.2 MaaS 的价值主张
- 对创作者：
- 低门槛发布（无需 DevOps）
- 自动化计费与分成
- 对使用者：
- 标准化 API（OpenAPI 3.0）
- 沙箱试用 → 无缝生产切换
- 对平台：
- 构建 AI 生态网络效应
类比：App Store for AI —— 模型和智能体就是“AI Apps”。
第二章：平台架构设计
2.1 整体系统视图
[创作者]
│ (上传模型/智能体 ZIP)
↓
[Flask 后台]
├── 验证器：检查 ONNX/OpenAPI 合规性
├── 沙箱测试：Docker 中运行示例请求
├── 存储：MinIO（模型文件） + PostgreSQL（元数据）
└── 发布：生成唯一 endpoint（/api/v1/apps/{app_id}）
│
↓
[用户]
├── 浏览商店（Vue 前端）
├── 试用（免费额度）
└── 订阅（Stripe 支付）
│
↓
[执行网关]
├── 认证：API Key 验证
├── 计量：Redis 记录调用次数/时长
├── 隔离：每个 App 在独立 Docker 容器运行
└── 监控：Prometheus 采集 SLA 指标
2.2 技术栈选型
- 模型格式 | ONNX（跨框架） + Pickle（Python 专用） | 优先 ONNX
- 智能体格式 | YAML 配置 + Python 脚本 | 定义工具依赖与入口
- 容器化 | Docker SDK for Python | 动态启停容器
- 支付 | Stripe Billing | 支持按量/订阅
- 前端 | Vue 3 + Element Plus | 快速构建商城 UI
第三章：应用标准化规范
3.1 模型应用规范（ONNX 优先）
# app.yaml（模型类）
name: "sales_forecast_prophet"
version: "1.2.0"
type: "model"
entrypoint: "predict.py" # 必须包含 predict() 函数
input_schema:
type: object
properties:
history:
type: array
items: { type: number }
output_schema:
type: object
properties:
forecast:
type: array
items: { type: number }
pricing:
model: "per_call"
price_per_unit: 0.01 # $0.01 per call
3.2 智能体应用规范
# app.yaml（智能体类）
name: "customer_service_agent"
version: "0.9.0"
type: "agent"
entrypoint: "agent.py" # 必须包含 run(goal: str) -> str
tools:
– name: "crm_query"
description: "查询客户信息"
required_permissions: ["read:crm"]
pricing:
model: "per_conversation"
price_per_unit: 0.50 # $0.50 per conversation
验证流程：上传时自动校验 YAML + 文件结构。
第四章：后端核心实现
4.1 应用上传与验证
# routes/app_upload.py
@app.post('/apps/upload')
def upload_app():
zip_file = request.files['app']
with tempfile.TemporaryDirectory() as tmpdir:
extract_zip(zip_file, tmpdir)
# 1. 解析 app.yaml
with open(f"{tmpdir}/app.yaml") as f:
meta = yaml.safe_load(f)
# 2. 验证入口文件
if not os.path.exists(f"{tmpdir}/{meta['entrypoint']}"):
return jsonify({"error": "Missing entrypoint"}), 400
# 3. 沙箱测试
test_result = run_sandbox_test(tmpdir, meta)
if not test_result["success"]:
return jsonify({"error": f"Test failed: {test_result['error']}"}), 400
# 4. 保存到 MinIO
minio_client.fput_object("apps", f"{meta['name']}-{meta['version']}.zip", zip_file_path)
# 5. 注册到数据库
App(
name=meta["name"],
version=meta["version"],
owner=current_user.id,
status="published",
pricing=meta["pricing"]
).save()
return jsonify({"message": "Published successfully", "endpoint": f"/api/v1/apps/{meta['name']}"})
4.2 安全执行网关
# services/execution_gateway.py
import docker
def execute_app(app_name: str, input_data: dict, api_key: str):
# 1. 验证 API Key 与配额
user = validate_api_key(api_key)
if not has_quota(user, app_name):
raise QuotaExceededError()
# 2. 获取应用元数据
app = App.objects(name=app_name).first()
# 3. 启动 Docker 容器（首次调用缓存镜像）
client = docker.from_env()
container = client.containers.run(
image=f"ai-apps/{app_name}",
command=f"python {app.entrypoint} –input '{json.dumps(input_data)}'",
volumes={f"/apps/{app_name}": {"bind": "/app", "mode": "ro"}},
network_disabled=True, # 禁用网络（除非显式授权）
mem_limit="512m",
detach=True
)
# 4. 等待结果
result = container.wait()
logs = container.logs().decode()
container.remove()
# 5. 计量扣费
record_usage(user.id, app_name, duration=result["duration"])
return json.loads(logs)
安全加固：
- 默认禁用网络访问
- 内存/CPU 限制
- 只读挂载应用代码
第五章：前端应用商店（Vue）
5.1 应用详情页
<template>
<div class="app-detail">
<h1>{{ app.name }}</h1>
<el-tag type="success">{{ app.pricing.model }} · ${{ app.pricing.price_per_unit }}</el-tag>
<div class="demo-section">
<h3>Try it now (Free)</h3>
<el-input v-model="demoInput" placeholder="Enter input JSON" type="textarea" />
<el-button @click="runDemo" type="primary">Run Demo</el-button>
<pre v-if="demoResult">{{ demoResult }}</pre>
</div>
<div class="subscribe-section">
<el-button @click="subscribe" type="warning" size="large">
Subscribe (${{ app.pricing.price_per_unit }}/{{ app.pricing.model }})
</el-button>
</div>
</div>
</template>
<script setup>
const props = defineProps({ app: Object })
const demoInput = ref('{}')
const demoResult = ref('')
async function runDemo() {
const res = await fetch(`/api/v1/apps/${props.app.name}/demo`, {
method: 'POST',
body: demoInput.value
})
demoResult.value = await res.text()
}
async function subscribe() {
// 跳转 Stripe Checkout
window.location.href = `/api/v1/checkout?app=${props.app.name}`
}
</script>
5.2 应用市场首页
- 分类筛选：
- 模型 vs 智能体
- 场景（金融/医疗/零售）
- 价格区间
- 排行榜：
- 热门应用（调用量）
- 高评分（用户评价）
第六章：场景实战
6.1 数据科学家：销售预测模型
- 上传内容：
- prophet_model.onnx
- predict.py（加载 ONNX + 预处理）
- app.yaml（定价 $0.01/次）
- 用户调用：
curl -X POST https://store.your-ai.com/api/v1/apps/sales_forecast_prophet \\
-H "Authorization: Bearer YOUR_API_KEY" \\
-d '{"history": [100, 120, 130, …]}'
- 收益：每月 10,000 次调用 → $100 收入
6.2 AI 工程师：客服智能体
- 能力：
- 理解用户问题
- 调用 CRM/订单 API
- 生成人性化回复
- 计费：$0.50/完整对话（多轮视为一次）
- 优势：企业无需自建智能体团队
6.3 企业用户：自动补货系统
- 组合应用：
- 调用 sales_forecast_prophet 预测需求
- 调用 inventory_optimizer 计算补货量
- 调用 erp_connector 下采购单
- 自动化：每日凌晨自动运行
第七章：计量与计费
7.1 实时计量（Redis）
# 记录调用
redis.hincrby(f"usage:{user_id}", app_name, 1)
redis.hincrby(f"usage:{user_id}:bytes", app_name, len(input_data))
# 检查配额
free_tier = 100 # 免费额度
used = int(redis.hget(f"usage:{user_id}", app_name) or 0)
if used >= free_tier and not user.has_subscription(app_name):
raise QuotaExceededError()
7.2 Stripe 集成
# 创建订阅
@app.get('/checkout')
def create_checkout_session():
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price_data': {
'currency': 'usd',
'product_data': {'name': app_name},
'recurring': {'interval': 'month'},
'unit_amount': int(price * 100) # 转为美分
},
'quantity': 1
}],
mode='subscription',
success_url='https://store.your-ai.com/success',
cancel_url='https…'
)
return redirect(session.url)
计费模型支持：
- 按调用次数
- 按执行时长
- 包月套餐
第八章：安全与合规
8.1 代码扫描
- 上传时静态分析：
- 禁止 os.system, subprocess, eval
- 检查敏感 API 调用（如 requests 到外部域名）
8.2 数据隐私
- 用户数据处理：
- 默认不存储输入/输出（仅计量元数据）
- GDPR 合规：提供数据删除接口
第九章：SLA 与监控
9.1 关键指标
- P95 延迟 | <2s
- 成功率 | >99%
- 资源隔离 | 无跨应用干扰
9.2 Prometheus 集成
# 在执行网关中暴露指标
from prometheus_client import Counter, Histogram
REQUEST_COUNT = Counter('app_requests_total', 'Total app requests', ['app_name'])
REQUEST_DURATION = Histogram('app_request_duration_seconds', 'Request duration', ['app_name'])
def execute_app(…):
start = time.time()
REQUEST_COUNT.labels(app_name=app_name).inc()
try:
result = run_container(…)
finally:
REQUEST_DURATION.labels(app_name=app_name).observe(time.time() – start)
Grafana 面板：实时展示各应用性能。
第十章：生态扩展
10.1 开发者 SDK
- CLI 工具：
ai-store-cli init # 生成 app.yaml 模板
ai-store-cli test # 本地沙箱测试
ai-store-cli publish # 一键发布
10.2 应用组合（Workflow）
- 可视化编排：
- 拖拽多个应用 → 生成新复合应用
- 自动处理数据流转
- 示例：
“销售预测 → 库存优化 → 采购下单” = 自动供应链智能体
总结：开启 AI 能力经济时代
AI 的未来不是大模型垄断，而是百花齐放的创作者经济。
