主题
12 - 控制面设计：开源实现与模块拆解
一句话：控制面只做编排与记账、不含任何 Agent 逻辑，API 层收管理请求，引擎层跑生命周期、分发、配额与闲置检测，状态落在 PostgreSQL 的 desired/observed 与乐观锁 revision 上；这让实例侧保持无感，代价是它成为唯一有权写实例配置的角色，自己宕机时用户对话仍能继续但暂存、分发、审批回传全部停摆，且每个接口都与 QwenPaw 原生 API 的演进强耦合。
一、设计目标
产品控制面是整个系统的"大脑"，负责：
- 实例全生命周期管理（创建、暂存、恢复、回收）
- 密钥治理（RT 生成/轮换/吊销）
- Skill/MCP 分发编排
- 配额管理与计量
- 产物取回与交付
- 审计日志
控制面代码随本方案一起开源，采用 Python FastAPI 框架，与 QwenPaw 同生态。
二、模块架构
三、核心数据模型
3.1 实例注册表
sql
CREATE TABLE instance_registry (
id BIGSERIAL PRIMARY KEY,
instance_id VARCHAR(64) UNIQUE NOT NULL,
user_id VARCHAR(64) UNIQUE NOT NULL,
tenant_id VARCHAR(64) NOT NULL,
-- 状态机
desired_state VARCHAR(20) NOT NULL DEFAULT 'active',
observed_state VARCHAR(20) NOT NULL DEFAULT 'pending',
revision INTEGER NOT NULL DEFAULT 1,
-- 资源引用
keys_ref VARCHAR(256),
snapshot_ref VARCHAR(512),
image_ref VARCHAR(256) NOT NULL,
-- 时间戳
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_active_at TIMESTAMPTZ,
retired_at TIMESTAMPTZ,
-- 元数据
metadata JSONB DEFAULT '{}',
last_error TEXT,
CONSTRAINT chk_state CHECK (desired_state IN ('active', 'staged', 'retired')),
CONSTRAINT chk_observed CHECK (observed_state IN ('pending', 'active', 'starting', 'staged', 'staging', 'restoring', 'retired', 'failed'))
);
CREATE INDEX idx_registry_user ON instance_registry(user_id);
CREATE INDEX idx_registry_state ON instance_registry(observed_state);
CREATE INDEX idx_registry_tenant ON instance_registry(tenant_id, observed_state);3.2 凭据 Vault
sql
CREATE TABLE credentials (
id BIGSERIAL PRIMARY KEY,
instance_id VARCHAR(64) NOT NULL REFERENCES instance_registry(instance_id),
credential_type VARCHAR(50) NOT NULL, -- runtime_token / model_key / mcp_credential
encrypted_value BYTEA NOT NULL, -- Fernet 加密
revision INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
expires_at TIMESTAMPTZ,
UNIQUE(instance_id, credential_type)
);
CREATE TABLE credential_grace_period (
id BIGSERIAL PRIMARY KEY,
instance_id VARCHAR(64) NOT NULL,
credential_type VARCHAR(50) NOT NULL,
old_value_hash VARCHAR(128) NOT NULL, -- 旧值哈希（不存明文）
expires_at TIMESTAMPTZ NOT NULL -- 24h 后过期
);
CREATE INDEX idx_cred_instance ON credentials(instance_id);3.3 任务注册表
sql
CREATE TABLE task_registry (
task_id VARCHAR(64) PRIMARY KEY,
user_id VARCHAR(64) NOT NULL,
instance_id VARCHAR(64) NOT NULL,
session_id VARCHAR(128) NOT NULL,
input JSONB NOT NULL,
request_context JSONB DEFAULT '{}',
status VARCHAR(20) NOT NULL DEFAULT 'submitted',
artifacts JSONB DEFAULT '[]',
artifact_count INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
submitted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
accepted_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
trace_id VARCHAR(64),
source VARCHAR(20),
error_code VARCHAR(50),
error_message TEXT
);
CREATE INDEX idx_task_user ON task_registry(user_id, submitted_at DESC);
CREATE INDEX idx_task_status ON task_registry(status) WHERE status NOT IN ('completed', 'failed', 'timeout', 'rejected');3.4 审计日志
sql
CREATE TABLE audit_log (
id BIGSERIAL PRIMARY KEY,
trace_id VARCHAR(64),
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
event_type VARCHAR(50) NOT NULL,
source VARCHAR(20) NOT NULL, -- gateway / control / instance
user_id VARCHAR(64),
instance_id VARCHAR(64),
session_id VARCHAR(128),
task_id VARCHAR(64),
action VARCHAR(100),
target VARCHAR(256),
result VARCHAR(20),
details JSONB DEFAULT '{}',
source_ip INET,
user_agent TEXT
);
CREATE INDEX idx_audit_time ON audit_log(timestamp DESC);
CREATE INDEX idx_audit_user ON audit_log(user_id, timestamp DESC);
CREATE INDEX idx_audit_trace ON audit_log(trace_id);
CREATE INDEX idx_audit_event ON audit_log(event_type, timestamp DESC);
-- 分区表（按月分区，便于归档清理）
-- CREATE TABLE audit_log_2026_10 PARTITION OF audit_log
-- FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');四、核心引擎实现
4.1 生命周期引擎
python
class LifecycleEngine:
"""实例生命周期管理"""
def __init__(self, k8s_client, registry, vault, nacos_client, minio_client):
self.k8s = k8s_client
self.registry = registry
self.vault = vault
self.nacos = nacos_client
self.minio = minio_client
async def create_instance(self, user_id: str, tenant_id: str, tier: str) -> InstanceRecord:
"""创建实例"""
instance_id = f"inst_{uuid4().hex[:12]}"
# 1. 生成密钥
rt = self.vault.get_or_create_runtime_token(instance_id)
# 2. 创建 K8s 资源
await self.k8s.create_stateful_set(
name=f"qwenpaw-{user_id}",
namespace="qwenpaw-instances",
image=IMAGE_REF,
user_id=user_id,
instance_id=instance_id,
runtime_token=rt,
tier=tier,
)
# 3. 等待就绪
await self._wait_for_ready(instance_id, timeout=120)
# 4. 注册
await self.nacos.register(instance_id, user_id)
await self.registry.create(instance_id, user_id, tenant_id, state="active")
return await self.registry.get(instance_id)
async def stage_instance(self, instance_id: str, reason: str):
"""暂存实例"""
inst = await self.registry.get(instance_id)
# 1. 备份
backup_id = await self._create_backup(inst)
snapshot_ref = f"backups/{inst.user_id}/{backup_id}.tar.gz"
# 2. 停止 Pod
await self.k8s.scale_stateful_set(inst.instance_id, replicas=0)
# 3. Nacos 注销
await self.nacos.deregister(instance_id)
# 4. 更新状态
await self.registry.update(instance_id, {
"observed_state": "staged",
"snapshot_ref": snapshot_ref,
})
async def restore_instance(self, instance_id: str):
"""恢复实例"""
inst = await self.registry.get(instance_id)
# 1. 重建 Pod
await self.k8s.scale_stateful_set(inst.instance_id, replicas=1)
# 2. 等待就绪
await self._wait_for_ready(instance_id, timeout=120)
# 3. 恢复数据
if inst.snapshot_ref:
await self._restore_backup(inst, inst.snapshot_ref)
# 4. Nacos 注册
await self.nacos.register(instance_id, inst.user_id)
# 5. 更新状态
await self.registry.update(instance_id, {
"observed_state": "active",
})
async def retire_instance(self, instance_id: str):
"""回收实例"""
inst = await self.registry.get(instance_id)
# 1. 交接归档
handover_ref = await self._export_handover(inst)
# 2. 吊销密钥
await self.vault.revoke(instance_id)
# 3. 删除 K8s 资源
await self.k8s.delete_stateful_set(inst.instance_id)
await self.k8s.delete_pvc(f"qwenpaw-{inst.user_id}-data")
await self.k8s.delete_pvc(f"qwenpaw-{inst.user_id}-secrets")
await self.k8s.delete_pvc(f"qwenpaw-{inst.user_id}-backups")
# 4. Nacos 注销
await self.nacos.deregister(instance_id)
# 5. 更新状态
await self.registry.update(instance_id, {
"observed_state": "retired",
"retired_at": datetime.utcnow(),
"snapshot_ref": handover_ref,
})4.2 分发编排引擎
python
class DistributionEngine:
"""Skill/MCP 分发编排"""
async def distribute_skills(self, user_id: str, skill_slugs: list[str]):
"""向实例分发 Skill"""
inst = await self.registry.get_by_user(user_id)
rt = await self.vault.get_runtime_token(inst.instance_id)
results = []
for slug in skill_slugs:
try:
# 1. 从市场获取 Skill 包
skill_info = await self.market.get_skill(slug)
# 2. 安装到实例
resp = await httpx.post(
f"http://{inst.endpoint}/api/skills/pool/import",
headers={"x-qwenpaw-runtime-token": rt},
json={"bundle_url": skill_info["bundle_url"], "provider_key": "enterprise"},
timeout=30.0,
)
if resp.status_code == 200:
# 3. 启用
await httpx.post(
f"http://{inst.endpoint}/api/skills/{slug}/enable",
headers={"x-qwenpaw-runtime-token": rt},
timeout=10.0,
)
results.append({"slug": slug, "status": "installed"})
else:
results.append({"slug": slug, "status": "failed", "error": resp.text})
except Exception as e:
results.append({"slug": slug, "status": "error", "error": str(e)})
# 4. 记录审计
await self.audit.record("skill_distribution", {
"user_id": user_id,
"skills": skill_slugs,
"results": results,
})
return results
async def distribute_mcp_clients(self, user_id: str, mcp_configs: list[dict]):
"""向实例分发 MCP Client"""
inst = await self.registry.get_by_user(user_id)
rt = await self.vault.get_runtime_token(inst.instance_id)
results = []
for config in mcp_configs:
try:
# 1. 从 Vault 获取 MCP 凭证
credentials = await self.vault.get_mcp_credentials(config["credential_ref"])
# 2. 创建 MCP Client
headers = config.get("headers", {})
if credentials:
headers["Authorization"] = f"Bearer {credentials['token']}"
resp = await httpx.post(
f"http://{inst.endpoint}/api/mcp",
headers={"x-qwenpaw-runtime-token": rt},
json={
"transport": config["transport"],
"url": config.get("url"),
"headers": headers,
"command": config.get("command"),
},
timeout=30.0,
)
if resp.status_code == 200:
client_id = resp.json()["client_id"]
# 3. 设置工具白名单
if config.get("tools_whitelist"):
await httpx.put(
f"http://{inst.endpoint}/api/mcp/tools/{client_id}",
headers={"x-qwenpaw-runtime-token": rt},
json={"tools": config["tools_whitelist"]},
timeout=10.0,
)
results.append({"name": config["name"], "status": "created", "client_id": client_id})
else:
results.append({"name": config["name"], "status": "failed", "error": resp.text})
except Exception as e:
results.append({"name": config["name"], "status": "error", "error": str(e)})
return results4.3 配额引擎
python
class QuotaEngine:
"""配额管理与计量"""
TIER_LIMITS = {
"basic": {"qps": 10, "daily_tokens": 100_000, "monthly_cost": 100},
"standard": {"qps": 30, "daily_tokens": 500_000, "monthly_cost": 500},
"enhanced": {"qps": 100, "daily_tokens": 2_000_000, "monthly_cost": 2000},
}
async def check_quota(self, user_id: str, token_count: int) -> tuple[bool, str]:
"""检查配额"""
user = await self.user_store.get(user_id)
tier = user.get("tier", "basic")
limits = self.TIER_LIMITS[tier]
# 日配额
daily_key = f"quota:daily:{user_id}:{date.today()}"
used = await self.redis.get(daily_key)
if used and int(used) + token_count > limits["daily_tokens"]:
return False, f"daily_quota_exceeded ({used}/{limits['daily_tokens']})"
# 月配额
monthly_key = f"quota:monthly:{user_id}:{date.today().month}"
used = await self.redis.get(monthly_key)
if used and int(used) + token_count > limits["monthly_cost"]:
return False, f"monthly_quota_exceeded ({used}/{limits['monthly_cost']})"
return True, "ok"
async def record_usage(self, user_id: str, token_count: int, cost: float):
"""记录用量"""
today = date.today()
month = today.month
pipe = self.redis.pipeline()
pipe.incrby(f"quota:daily:{user_id}:{today}", token_count)
pipe.incrby(f"quota:monthly:{user_id}:{month}", int(cost * 100))
pipe.expire(f"quota:daily:{user_id}:{today}", 86400 * 2)
await pipe.execute()
# 告警检查
user = await self.user_store.get(user_id)
tier = user.get("tier", "basic")
limits = self.TIER_LIMITS[tier]
daily_used = await self.redis.get(f"quota:daily:{user_id}:{today}")
if daily_used and int(daily_used) > limits["daily_tokens"] * 0.8:
await self.alert.send("quota_warning", {
"user_id": user_id,
"tier": tier,
"used": int(daily_used),
"limit": limits["daily_tokens"],
})五、API 端点设计
5.1 实例管理 API
| 端点 | 方法 | 说明 | 权限 |
|---|---|---|---|
/api/instances | GET | 列出实例（支持分页/过滤） | admin |
/api/instances | POST | 创建实例 | admin |
/api/instances/{id} | GET | 获取实例详情 | admin / owner |
/api/instances/{id}/stage | POST | 暂存实例 | admin |
/api/instances/{id}/restore | POST | 恢复实例 | admin |
/api/instances/{id}/retire | POST | 回收实例 | admin |
/api/instances/{id}/restart | POST | 重启实例 | admin / owner |
/api/instances/{id}/upgrade | POST | 升级镜像 | admin |
5.2 用户管理 API
| 端点 | 方法 | 说明 | 权限 |
|---|---|---|---|
/api/users | GET | 列出用户 | admin |
/api/users/{id} | GET | 获取用户详情 | admin / owner |
/api/users/{id}/tier | PUT | 更新订阅档位 | admin |
/api/users/{id}/quota | GET | 查询配额使用情况 | admin / owner |
5.3 分发管理 API
| 端点 | 方法 | 说明 | 权限 |
|---|---|---|---|
/api/distribution/skills | POST | 批量分发 Skill | admin |
/api/distribution/mcp | POST | 批量分发 MCP Client | admin |
/api/distribution/status | GET | 查询分发状态 | admin |
六、开源仓库结构
qwenpaw-enterprise/
├── README.md
├── pyproject.toml
├── Dockerfile
├── docker-compose.yaml # 本地开发环境
├── k8s/ # K8s 部署清单
│ ├── namespace.yaml
│ ├── control-plane/
│ │ ├── deployment.yaml
│ │ ├── service.yaml
│ │ └── configmap.yaml
│ ├── gateway/
│ │ ├── higress-config.yaml
│ │ └── ext-authz-deployment.yaml
│ ├── instances/
│ │ ├── statefulset-template.yaml
│ │ ├── pvc-template.yaml
│ │ └── network-policy.yaml
│ └── infra/
│ ├── postgres.yaml
│ ├── redis.yaml
│ ├── nacos.yaml
│ └── minio.yaml
├── src/
│ └── qwenpaw_enterprise/
│ ├── __init__.py
│ ├── main.py # FastAPI 入口
│ ├── api/ # API 路由
│ │ ├── instances.py
│ │ ├── users.py
│ │ ├── distribution.py
│ │ ├── quotas.py
│ │ └── audit.py
│ ├── engine/ # 核心引擎
│ │ ├── lifecycle.py
│ │ ├── distribution.py
│ │ ├── quota.py
│ │ ├── image.py
│ │ └── idle_detector.py
│ ├── store/ # 数据存储
│ │ ├── registry.py
│ │ ├── vault.py
│ │ ├── task_store.py
│ │ └── audit_store.py
│ ├── infra/ # 基础设施客户端
│ │ ├── k8s_client.py
│ │ ├── nacos_client.py
│ │ ├── minio_client.py
│ │ └── redis_client.py
│ └── models/ # 数据模型
│ ├── instance.py
│ ├── user.py
│ ├── task.py
│ └── audit.py
├── tests/
│ ├── test_lifecycle.py
│ ├── test_distribution.py
│ └── test_quota.py
└── docs/
├── deployment.md
├── api-reference.md
└── architecture.md