Prompt 管理平台搭建指南

为什么需要 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 更高。

实战建议

Prompt 即代码：把 Prompt 纳入 CI/CD 流程，修改 Prompt 需要代码审查
变量校验：在 API 层校验变量类型和必填项，防止运行时出错
缓存策略：热 Prompt 缓存在 Redis，TTL 5 分钟，版本更新时主动刷新
评估自动化：每次创建新版本时自动跑测试集，评分低于阈值阻止发布—

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

为什么需要 Prompt 管理平台#

核心功能需求#

架构设计#

数据库设计#

核心 API 实现#

A/B 测试实现#

与现有工具对比#

实战建议#

加入讨论#

📌 相关推荐

为什么需要 Prompt 管理平台

核心功能需求

架构设计

数据库设计

核心 API 实现

A/B 测试实现

与现有工具对比

实战建议

加入讨论