跳至主要內容

Prompt 工程化管理——不要再用记事本存你的 Prompt 了

老张2026/4/30大约 9 分钟

Prompt 工程化管理——不要再用记事本存你的 Prompt 了

适读人群：有 AI 应用开发经验的工程师、技术负责人 | 阅读时长：约 12 分钟 | 核心价值：建立 Prompt 版本控制和工程化管理体系，防止团队协作中的 Prompt 失控

去年有一次事故，到现在想起来还有点后怕。

那是个周五下午，我们的客服机器人突然开始给用户输出奇怪的回复——语气变了，边界也变了，有几个用户截图发出来，说我们的 AI 态度很差。我一查，发现是我们组的小李在当天下午「优化」了一下 Prompt，觉得原来的太啰嗦，改简洁了。

结果简洁是简洁了，但那个 Prompt 里有一段关于「不得推荐竞品」和「不得承诺不存在的功能」的约束，被他当成冗余删掉了。

问题不是小李不负责任。问题是我们根本没有 Prompt 的版本控制。那个 Prompt 存在一个共享文档里，谁都可以改，改了没有记录，出了问题也不知道哪个版本是好的。我们花了三个小时才从聊天记录里翻出一个「差不多可用」的旧版本——注意，是差不多可用，不是完全正确的那个。

那次之后，我花了两周时间建了一套 Prompt 工程化管理体系。踩了很多坑，也想明白了不少东西，今天全写出来。

大多数团队的 Prompt 管理有多混乱

我调研过几个不同规模的技术团队，发现 Prompt 管理的现状基本是这样的：

  • 初级水平：Prompt 硬编码在代码里，散落在各个文件的字符串里，靠代码注释区分版本
  • 中级水平：单独抽出来放在一个 prompts/ 目录下，或者放到数据库里，但没有版本控制，改了就改了
  • "高级"水平：放到共享文档（比如 Notion、语雀），团队都能看到，但依然是覆盖式修改

这三种方式有个共同问题：出了事回不去。

更深层的问题是：Prompt 不是代码，但它的行为影响和代码一样大。一行代码改错了，单测会告诉你；一个 Prompt 改坏了，你可能要等用户投诉才知道。

Prompt 工程化管理需要解决的四个问题

在开始之前，先把问题定义清楚：

  1. 存储问题：Prompt 存在哪里，怎么组织
  2. 版本控制：每次修改有记录，可以对比，可以回滚
  3. A/B 测试：新 Prompt 上线前怎么验证效果
  4. 回滚机制：出了问题怎么快速切换

逐个来说。

一、Prompt 存储方案

Prompt 不适合直接存在代码里，原因是：

  • 修改 Prompt 需要走代码发布流程，效率低
  • 多个 Prompt 散落在代码各处，管理混乱
  • 非技术人员无法参与 Prompt 优化

我现在用的方案是数据库 + Git 双轨制：

Git 负责归档和审计：所有 Prompt 以文本文件形式存在一个独立的 Git 仓库里，每次修改都有提交记录，可以 diff，可以回滚。

数据库负责运行时读取：线上系统从数据库读取 Prompt，支持热更新，不需要重新发布服务。

数据库表结构：

CREATE TABLE prompts (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    prompt_key VARCHAR(100) NOT NULL COMMENT 'Prompt的业务标识，如customer_service_system',
    version VARCHAR(20) NOT NULL COMMENT '版本号，如v1.2.3',
    content TEXT NOT NULL COMMENT 'Prompt内容',
    description VARCHAR(500) COMMENT '这个版本做了什么改动',
    author VARCHAR(50) NOT NULL,
    status TINYINT NOT NULL DEFAULT 0 COMMENT '0=草稿, 1=测试中, 2=线上, 3=已废弃',
    is_active TINYINT NOT NULL DEFAULT 0 COMMENT '1=当前使用的版本',
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_key_status (prompt_key, status),
    INDEX idx_key_active (prompt_key, is_active)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

几个设计细节：

  • prompt_key 是业务标识，不是数字 ID，这样代码里写的是 customer_service_system 而不是 42，可读性好
  • version 用语义化版本号，主版本号变动意味着 Prompt 结构性重写，小版本是优化调整
  • is_active 独立字段控制当前激活的版本，切换时只改这个字段，不删除历史记录

二、版本控制的实现

光有数据库还不够，还需要一个管理层来做版本控制。

import hashlib
from datetime import datetime
from typing import Optional
from dataclasses import dataclass
import json

@dataclass
class PromptVersion:
    prompt_key: str
    version: str
    content: str
    description: str
    author: str
    status: int  # 0=draft, 1=testing, 2=production, 3=deprecated
    checksum: str  # 内容的MD5，用来快速检测变化
    
    def to_dict(self):
        return {
            'prompt_key': self.prompt_key,
            'version': self.version,
            'content': self.content,
            'description': self.description,
            'author': self.author,
            'status': self.status,
            'checksum': self.checksum
        }

class PromptManager:
    def __init__(self, db_client, cache_client=None):
        self.db = db_client
        self.cache = cache_client
        self._local_cache = {}  # 进程内缓存
    
    def get_active_prompt(self, prompt_key: str) -> Optional[str]:
        """获取当前激活的Prompt内容，带缓存"""
        cache_key = f"prompt:active:{prompt_key}"
        
        # 1. 先查进程内缓存（1分钟TTL）
        if cache_key in self._local_cache:
            cached = self._local_cache[cache_key]
            if cached['expire_at'] > datetime.now().timestamp():
                return cached['content']
        
        # 2. 查Redis缓存
        if self.cache:
            cached_content = self.cache.get(cache_key)
            if cached_content:
                self._local_cache[cache_key] = {
                    'content': cached_content,
                    'expire_at': datetime.now().timestamp() + 60
                }
                return cached_content
        
        # 3. 查数据库
        result = self.db.query_one(
            "SELECT content FROM prompts WHERE prompt_key = %s AND is_active = 1",
            (prompt_key,)
        )
        
        if result:
            content = result['content']
            # 写入缓存，TTL 5分钟
            if self.cache:
                self.cache.setex(cache_key, 300, content)
            self._local_cache[cache_key] = {
                'content': content,
                'expire_at': datetime.now().timestamp() + 60
            }
            return content
        
        return None
    
    def create_version(self, prompt_key: str, content: str, 
                       description: str, author: str, 
                       base_version: Optional[str] = None) -> PromptVersion:
        """创建新版本"""
        # 计算新版本号
        if base_version:
            new_version = self._increment_version(base_version)
        else:
            # 查最新版本
            latest = self.db.query_one(
                "SELECT version FROM prompts WHERE prompt_key = %s ORDER BY created_at DESC LIMIT 1",
                (prompt_key,)
            )
            if latest:
                new_version = self._increment_version(latest['version'])
            else:
                new_version = "v1.0.0"
        
        checksum = hashlib.md5(content.encode()).hexdigest()
        
        pv = PromptVersion(
            prompt_key=prompt_key,
            version=new_version,
            content=content,
            description=description,
            author=author,
            status=0,  # 默认草稿
            checksum=checksum
        )
        
        self.db.execute(
            """INSERT INTO prompts (prompt_key, version, content, description, author, status, is_active)
               VALUES (%s, %s, %s, %s, %s, %s, 0)""",
            (pv.prompt_key, pv.version, pv.content, 
             pv.description, pv.author, pv.status)
        )
        
        return pv
    
    def activate_version(self, prompt_key: str, version: str):
        """激活指定版本（自动停用当前激活的版本）"""
        # 事务操作
        with self.db.transaction():
            # 停用当前激活版本
            self.db.execute(
                "UPDATE prompts SET is_active = 0 WHERE prompt_key = %s AND is_active = 1",
                (prompt_key,)
            )
            # 激活新版本
            self.db.execute(
                "UPDATE prompts SET is_active = 1, status = 2 WHERE prompt_key = %s AND version = %s",
                (prompt_key, version)
            )
        
        # 清除缓存
        cache_key = f"prompt:active:{prompt_key}"
        if self.cache:
            self.cache.delete(cache_key)
        if cache_key in self._local_cache:
            del self._local_cache[cache_key]
    
    def rollback(self, prompt_key: str, target_version: str):
        """回滚到指定版本"""
        # 验证目标版本存在
        target = self.db.query_one(
            "SELECT id FROM prompts WHERE prompt_key = %s AND version = %s",
            (prompt_key, target_version)
        )
        if not target:
            raise ValueError(f"版本 {target_version} 不存在")
        
        self.activate_version(prompt_key, target_version)
        
        # 记录回滚日志
        self.db.execute(
            """INSERT INTO prompt_rollback_log (prompt_key, from_version, to_version, reason, created_at)
               SELECT %s, version, %s, 'manual_rollback', NOW()
               FROM prompts WHERE prompt_key = %s AND is_active = 1""",
            (prompt_key, target_version, prompt_key)
        )
    
    def get_version_history(self, prompt_key: str, limit: int = 20):
        """获取版本历史"""
        return self.db.query(
            """SELECT version, description, author, status, is_active, created_at, checksum
               FROM prompts 
               WHERE prompt_key = %s 
               ORDER BY created_at DESC 
               LIMIT %s""",
            (prompt_key, limit)
        )
    
    def diff_versions(self, prompt_key: str, version_a: str, version_b: str):
        """对比两个版本的内容差异"""
        import difflib
        
        va = self.db.query_one(
            "SELECT content FROM prompts WHERE prompt_key = %s AND version = %s",
            (prompt_key, version_a)
        )
        vb = self.db.query_one(
            "SELECT content FROM prompts WHERE prompt_key = %s AND version = %s",
            (prompt_key, version_b)
        )
        
        if not va or not vb:
            raise ValueError("版本不存在")
        
        diff = list(difflib.unified_diff(
            va['content'].splitlines(keepends=True),
            vb['content'].splitlines(keepends=True),
            fromfile=f'{prompt_key}@{version_a}',
            tofile=f'{prompt_key}@{version_b}'
        ))
        
        return ''.join(diff)
    
    def _increment_version(self, version: str) -> str:
        """版本号自增（patch版本）"""
        # v1.2.3 -> v1.2.4
        parts = version.lstrip('v').split('.')
        parts[-1] = str(int(parts[-1]) + 1)
        return 'v' + '.'.join(parts)

三、A/B 测试方案

版本控制做好了，新 Prompt 上线前还需要 A/B 测试。我的做法是在数据库里加一个实验配置表：

CREATE TABLE prompt_experiments (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    prompt_key VARCHAR(100) NOT NULL,
    experiment_name VARCHAR(100) NOT NULL,
    version_a VARCHAR(20) NOT NULL COMMENT '对照组版本',
    version_b VARCHAR(20) NOT NULL COMMENT '实验组版本',
    traffic_ratio DECIMAL(3,2) NOT NULL DEFAULT 0.10 COMMENT '实验组流量比例，0.10表示10%',
    status TINYINT NOT NULL DEFAULT 1 COMMENT '1=运行中, 0=已结束',
    started_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    ended_at DATETIME,
    INDEX idx_key_status (prompt_key, status)
);

然后在 get_active_prompt 里加入实验逻辑：

def get_prompt_for_request(self, prompt_key: str, user_id: str) -> tuple[str, str]:
    """
    返回 (prompt_content, version)
    根据用户ID做流量分配，保证同一用户始终在同一组
    """
    # 查看是否有进行中的实验
    experiment = self.db.query_one(
        """SELECT * FROM prompt_experiments 
           WHERE prompt_key = %s AND status = 1 
           ORDER BY started_at DESC LIMIT 1""",
        (prompt_key,)
    )
    
    if experiment:
        # 用用户ID做哈希，确保同一用户始终分到同一组
        hash_value = int(hashlib.md5(f"{user_id}:{prompt_key}".encode()).hexdigest(), 16)
        in_experiment = (hash_value % 100) < int(experiment['traffic_ratio'] * 100)
        
        version = experiment['version_b'] if in_experiment else experiment['version_a']
        
        # 记录实验分配（用于后续效果统计）
        self._log_experiment_assignment(experiment['id'], user_id, version)
        
        content = self.db.query_one(
            "SELECT content FROM prompts WHERE prompt_key = %s AND version = %s",
            (prompt_key, version)
        )
        return content['content'], version
    
    # 没有实验，返回激活版本
    content = self.get_active_prompt(prompt_key)
    return content, 'active'

def _log_experiment_assignment(self, experiment_id: int, user_id: str, version: str):
    """记录实验分配，每个用户每天只记录一次"""
    self.db.execute(
        """INSERT IGNORE INTO experiment_logs (experiment_id, user_id, version, date)
           VALUES (%s, %s, %s, CURDATE())""",
        (experiment_id, user_id, version)
    )

四、Git 同步（双轨制的另一条轨）

数据库里的 Prompt 要定期同步到 Git 仓库做归档。我写了一个简单的同步脚本：

import os
import git
from pathlib import Path

def sync_prompts_to_git(db_client, repo_path: str):
    """将数据库中所有激活的Prompt同步到Git仓库"""
    repo = git.Repo(repo_path)
    prompts_dir = Path(repo_path) / 'prompts'
    prompts_dir.mkdir(exist_ok=True)
    
    # 获取所有prompt_key
    keys = db_client.query(
        "SELECT DISTINCT prompt_key FROM prompts WHERE status != 3"
    )
    
    changed_files = []
    
    for key_row in keys:
        key = key_row['prompt_key']
        
        # 获取该key的所有非废弃版本
        versions = db_client.query(
            """SELECT version, content, description, author, status, is_active, created_at
               FROM prompts WHERE prompt_key = %s AND status != 3
               ORDER BY created_at""",
            (key,)
        )
        
        # 为每个key创建目录
        key_dir = prompts_dir / key
        key_dir.mkdir(exist_ok=True)
        
        for v in versions:
            version_file = key_dir / f"{v['version']}.txt"
            meta_file = key_dir / f"{v['version']}.meta.json"
            
            # 写Prompt内容
            with open(version_file, 'w', encoding='utf-8') as f:
                f.write(v['content'])
            
            # 写元数据
            meta = {
                'version': v['version'],
                'description': v['description'],
                'author': v['author'],
                'status': v['status'],
                'is_active': v['is_active'],
                'created_at': str(v['created_at'])
            }
            with open(meta_file, 'w', encoding='utf-8') as f:
                import json
                json.dump(meta, f, ensure_ascii=False, indent=2)
            
            changed_files.extend([str(version_file), str(meta_file)])
    
    # Git提交
    repo.index.add(changed_files)
    if repo.index.diff('HEAD') or repo.untracked_files:
        repo.index.commit(f"sync prompts from db - {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        print(f"已同步 {len(keys)} 个Prompt key 到Git")
    else:
        print("没有变化，跳过提交")

五、紧急回滚怎么做

出了事，最重要的是快。我在运维脚本里加了一个快速回滚命令：

#!/bin/bash
# rollback_prompt.sh
# 用法: ./rollback_prompt.sh customer_service_system v1.2.1

PROMPT_KEY=$1
TARGET_VERSION=$2

if [ -z "$PROMPT_KEY" ] || [ -z "$TARGET_VERSION" ]; then
    echo "用法: $0 <prompt_key> <target_version>"
    exit 1
fi

echo "准备回滚 $PROMPT_KEY 到版本 $TARGET_VERSION ..."

# 调用管理API
curl -X POST "http://your-admin-api/prompts/rollback" \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"prompt_key\": \"$PROMPT_KEY\", \"target_version\": \"$TARGET_VERSION\"}"

echo "回滚完成，请验证系统行为"

加上一个查看版本历史的命令：

# list_prompt_versions.sh
PROMPT_KEY=$1
curl -s "http://your-admin-api/prompts/$PROMPT_KEY/history" | jq '.[] | {version, description, author, status, is_active, created_at}'

几点经验

Prompt 的修改要有审批流程。不是每个人改完都能直接上生产。我们现在的做法是：修改 -> 创建新版本（status=draft）-> Reviewer 审核 -> 进入 A/B 测试（status=testing）-> 数据达标后激活（status=production）。

Prompt 里的变量要标准化。我们用 {{variable_name}} 的格式标记变量，有专门的变量替换逻辑，不允许在 Prompt 里直接做字符串拼接。

定期清理废弃版本。status=3 的版本保留半年后可以归档，不然数据库会越来越大。但要确保 Git 里有备份。

关键 Prompt 要有自动化测试。写一批标准测试用例，每次 Prompt 修改后跑一遍，看输出是否符合预期。这个投入是值得的。

那次事故之后，我们再也没有出现过类似的「Prompt 被改坏回不去」的问题。工程化不是多此一举，是在说：Prompt 是一等公民，和代码一样需要被认真对待。

编辑此页
最近更新:
贡献者: xiaozhangge