为什么需要 Prompt 管理平台 当团队有 3 个以上 LLM 应用时，Prompt 管理就会失控：Prompt 散落在代码里、改一个字要重新部署、无法 A/B 测试、没有版本回滚。Prompt 管理平台把 Prompt 当作独立资产来管理。
核心功能需求 功能模块 说明 优先级 Prompt CRUD 创建/读取/更新/删除 Prompt P0 版本管理 每次修改生成新版本，支持回滚 P0 变量模板 支持 {{variable}} 变量插值 P0 A/B 测试 流量分割对比不同 Prompt 效果 P1 权限控制 团队成员角色与审批流 P1 效果监控 Prompt 调用的成功率/延迟/成本 P1 批量测试 用测试集自动评估 Prompt P2 架构设计 +----------------------------------------------+ | Prompt 管理平台架构 | | | | +------------+ +------------------+ | | | Web UI | | API Gateway | | | | (React) | | (FastAPI) | | | +-----+------+ +-------+----------+ | | | | | | +-----+-----------------+----------------+ | | | Prompt Service Layer | | | | CRUD | Version | AB Test | RBAC | | | +-------------------+--------------------+ | | | | | +---------+---------+---------+----------+ | | |PostgreSQL| Redis缓存 |S3存储 | | | |(元数据) | (热Prompt) |(备份) | | | +---------+--------------+----------+ | +----------------------------------------------+ 数据库设计 -- Prompt 主表 CREATE TABLE prompts ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(128) NOT NULL UNIQUE, description TEXT, category VARCHAR(64), current_version_id UUID REFERENCES prompt_versions(id), created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- Prompt 版本表 CREATE TABLE prompt_versions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), prompt_id UUID NOT NULL REFERENCES prompts(id) ON DELETE CASCADE, version INT NOT NULL, content TEXT NOT NULL, variables JSONB DEFAULT '[]', model_config JSONB DEFAULT '{}', change_log TEXT, created_by UUID NOT NULL REFERENCES users(id), created_at TIMESTAMP DEFAULT NOW(), UNIQUE(prompt_id, version) ); -- A/B 测试表 CREATE TABLE ab_tests ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(128) NOT NULL, prompt_id UUID NOT NULL REFERENCES prompts(id), variant_a_version_id UUID NOT NULL REFERENCES prompt_versions(id), variant_b_version_id UUID NOT NULL REFERENCES prompt_versions(id), traffic_split JSONB DEFAULT '{"a": 50, "b": 50}', status VARCHAR(16) DEFAULT 'running', metrics JSONB DEFAULT '{}', created_at TIMESTAMP DEFAULT NOW(), ended_at TIMESTAMP ); 核心 API 实现 from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel import asyncpg, json app = FastAPI() class PromptCreate(BaseModel): name: str description: str = "" content: str variables: list[dict] = [] model_config: dict = {} class PromptUpdate(BaseModel): content: str change_log: str = "" variables: list[dict] = [] @app.post("/api/prompts") async def create_prompt(data: PromptCreate, db=Depends(get_db)): async with db.transaction(): prompt = await db.fetchrow( "INSERT INTO prompts (name, description) VALUES ($1, $2) RETURNING *", data.name, data.description ) version = await db.fetchrow( "INSERT INTO prompt_versions (prompt_id, version, content, variables, model_config, created_by) VALUES ($1, 1, $2, $3, $4, $5) RETURNING *", prompt["id"], data.content, json.dumps(data.variables), json.dumps(data.model_config), current_user_id ) await db.execute( "UPDATE prompts SET current_version_id = $1 WHERE id = $2", version["id"], prompt["id"] ) return {"prompt": dict(prompt), "version": dict(version)} @app.put("/api/prompts/{prompt_id}") async def update_prompt(prompt_id: str, data: PromptUpdate, db=Depends(get_db)): async with db.transaction(): current = await db.fetchrow( "SELECT version FROM prompt_versions WHERE prompt_id = $1 ORDER BY version DESC LIMIT 1", prompt_id ) new_version = (current["version"] + 1) if current else 1 version = await db.fetchrow( "INSERT INTO prompt_versions (prompt_id, version, content, variables, change_log, created_by) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *", prompt_id, new_version, data.content, json.dumps(data.variables), data.change_log, current_user_id ) await db.execute( "UPDATE prompts SET current_version_id = $1, updated_at = NOW() WHERE id = $2", version["id"], prompt_id ) return {"version": dict(version)} @app.post("/api/prompts/{prompt_id}/rollback/{version}") async def rollback_prompt(prompt_id: str, version: int, db=Depends(get_db)): target = await db.fetchrow( "SELECT id FROM prompt_versions WHERE prompt_id = $1 AND version = $2", prompt_id, version ) if not target: raise HTTPException(404, "Version not found") await db.execute( "UPDATE prompts SET current_version_id = $1 WHERE id = $2", target["id"], prompt_id ) return {"rolled_back_to": version} A/B 测试实现 import random class ABTestRouter: def __init__(self, db, redis): self.db = db self.redis = redis async def get_variant(self, test_id: str, user_id: str) -> str: cache_key = f"ab:{test_id}:{user_id}" cached = await self.redis.get(cache_key) if cached: return cached test = await self.db.fetchrow( "SELECT * FROM ab_tests WHERE id = $1 AND status = 'running'", test_id ) if not test: return "a" split = json.loads(test["traffic_split"]) variant = "a" if random.random() * 100 < split["a"] else "b" await self.redis.setex(cache_key, 86400, variant) return variant async def record_metric(self, test_id, variant, metric, value): await self.db.execute( "INSERT INTO ab_metrics (test_id, variant, metric, value) VALUES ($1, $2, $3, $4)", test_id, variant, metric, value ) 与现有工具对比 工具 优点 缺点 适用场景 LangSmith LangChain 生态集成好 绑定 LangChain，贵 已用 LangChain 的团队 PromptHub UI 好用，有协作功能 自定义能力弱 非技术团队 自建平台 完全可控，可定制 开发成本高 大团队/特殊需求 建议：团队 < 5 人且用 LangChain，直接上 LangSmith。团队 > 10 人或有特殊需求，自建平台 ROI 更高。
...
