你有没有经历过这种情况:某天线上AI功能突然变差了,排查了半天,发现是两周前有人顺手改了一行Prompt,但没有人记录、没有人评估,就这么带进了生产环境。

这是大多数AI应用团队都踩过的坑。Prompt不是配置文件,更不是可以随便改的字符串——它是AI应用的核心逻辑。管理好Prompt,是AI应用工程化的重要一环。

为什么Prompt需要版本管理

先说清楚问题的严重性:

Prompt改动的影响面不可预测。加一个词、换一个例子,都可能对输出质量产生意料之外的影响。没有版本记录,你无法追溯到底是哪次改动引发了问题。

Prompt测试通常不完整。大家常犯的错误是只测了几个happy path,没有测边界情况。版本管理配合测试集,可以在发布前做全面的回归测试。

多人协作容易冲突。多个工程师同时在改同一个Prompt,没有协作流程就会互相覆盖。

线上监控需要版本对照。如果你想知道"v1.3版本的Prompt比v1.2版本好在哪里",就必须有版本记录和对应的评估数据。

最简单的起点:文件+Git

最低成本的方案:把Prompt存成文件,用Git管理。

目录结构

prompts/
├── system/
│   ├── code_reviewer.md
│   ├── customer_support.md
│   └── data_analyst.md
├── tasks/
│   ├── summarize.md
│   ├── extract_entities.md
│   └── classify_intent.md
└── templates/
    ├── rag_qa.md
        └── few_shot_base.md
        ```
### Prompt文件格式

每个Prompt文件包含元数据和内容:

```markdown
---
name: code_reviewer
version: 1.3.0
description: 代码审查助手,专注于Python代码质量
model: gpt-4o
temperature: 0.3
max_tokens: 2000
last_modified: 2026-05-15
author: alice
changelog:
  - "1.3.0 (2026-05-15): 增加对类型注解缺失的检测"
  -   - "1.2.1 (2026-05-10): 修复误报变量命名问题"
  -   - "1.2.0 (2026-04-28): 新增安全漏洞检测规则"
  - ---
你是一个专业的Python代码审查专家。请对提供的代码进行审查,重点关注:

1. **代码质量**:可读性、命名规范、注释完整性
2. 2. **潜在Bug**:类型错误、边界条件、异常处理
3. 3. **性能问题**:低效算法、不必要的循环、内存泄漏风险
4. 4. **安全漏洞**:SQL注入、路径遍历、硬编码密钥
5. 5. **工程规范**:函数长度、类设计、依赖管理
## 输出格式

以JSON格式输出审查结果:
```json
{
  "summary": "整体评估(一句话)",
    "severity": "low|medium|high|critical",
      "issues": [
          {
                "line": 行号,
                      "type": "问题类型",
                            "description": "问题描述",
                                  "suggestion": "修改建议",
                                        "severity": "low|medium|high"
                                            }
                                              ]
                                              }
                                              ```
如果代码质量良好,issues数组返回空列表。

加载和使用

import yaml
import re
from pathlib import Path

class PromptManager:
    def __init__(self, prompts_dir: str = "prompts"):
            self.prompts_dir = Path(prompts_dir)
                    self._cache = {}
                        
                            def load(self, name: str, version: str = "latest") -> Prompt:
                                    """加载指定名称和版本的Prompt"""
                                            cache_key = f"{name}:{version}"
                                                    if cache_key in self._cache:
                                                                return self._cache[cache_key]
                                                                        
                                                                                if version == "latest":
                                                                                            # 找最新版本(按文件名或目录结构)
                                                                                                        prompt_file = self._find_latest(name)
                                                                                                                else:
                                                                                                                            prompt_file = self.prompts_dir / f"{name}/v{version}.md"
                                                                                                                                    
                                                                                                                                            if not prompt_file.exists():
                                                                                                                                                        raise FileNotFoundError(f"Prompt not found: {name} v{version}")
                                                                                                                                                                
                                                                                                                                                                        prompt = self._parse_prompt_file(prompt_file)
                                                                                                                                                                                self._cache[cache_key] = prompt
                                                                                                                                                                                        return prompt
                                                                                                                                                                                            
                                                                                                                                                                                                def _parse_prompt_file(self, path: Path) -> "Prompt":
                                                                                                                                                                                                        content = path.read_text(encoding="utf-8")
                                                                                                                                                                                                                
                                                                                                                                                                                                                        # 解析frontmatter
                                                                                                                                                                                                                                fm_match = re.match(r'^---\n(.*?)\n---\n(.*)', content, re.DOTALL)
                                                                                                                                                                                                                                        if fm_match:
                                                                                                                                                                                                                                                    metadata = yaml.safe_load(fm_match.group(1))
                                                                                                                                                                                                                                                                body = fm_match.group(2).strip()
                                                                                                                                                                                                                                                                        else:
                                                                                                                                                                                                                                                                                    metadata = {}
                                                                                                                                                                                                                                                                                                body = content.strip()
                                                                                                                                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                return Prompt(
                                                                                                                                                                                                                                                                                                                            name=metadata.get("name", path.stem),
                                                                                                                                                                                                                                                                                                                                        version=metadata.get("version", "1.0.0"),
                                                                                                                                                                                                                                                                                                                                                    content=body,
                                                                                                                                                                                                                                                                                                                                                                model=metadata.get("model", "gpt-4o"),
                                                                                                                                                                                                                                                                                                                                                                            temperature=metadata.get("temperature", 0.7),
                                                                                                                                                                                                                                                                                                                                                                                        max_tokens=metadata.get("max_tokens", 2000),
                                                                                                                                                                                                                                                                                                                                                                                                    metadata=metadata
                                                                                                                                                                                                                                                                                                                                                                                                            )

# 使用
pm = PromptManager()
prompt = pm.load("code_reviewer")

response = await openai_client.chat.completions.create(
    model=prompt.model,
        temperature=prompt.temperature,
            max_tokens=prompt.max_tokens,
                messages=[
                        {"role": "system", "content": prompt.content},
                                {"role": "user", "content": f"请审查以下代码:\n\n```python\n{code}\n```"}
                                    ]
                                    )
                                    ```
## 进阶:Prompt评估流水线

版本管理的价值,只有配合自动化评估才能完全发挥出来。

### 评估数据集

每个Prompt都应该有一套对应的测试案例:

```python
# prompts/tests/code_reviewer_cases.json
test_cases = [
    {
            "id": "tc_001",
                    "name": "检测SQL注入漏洞",
                            "input": """
                            def get_user(username):
                                query = f"SELECT * FROM users WHERE name = '{username}'"
                                    return db.execute(query)
                                    """,
                                            "expected": {
                                                        "severity": ["high", "critical"],  # 期望严重程度
                                                                    "should_mention": ["SQL注入", "参数化查询", "sql injection"],  # 应该提到的关键词
                                                                                "should_not_mention": []  # 不应该包含的内容
                                                                                        }
                                                                                            },
                                                                                                {
                                                                                                        "id": "tc_002", 
                                                                                                                "name": "良好代码不应有误报",
                                                                                                                        "input": """
                                                                                                                        from typing import Optional
def calculate_bmi(weight_kg: float, height_m: float) -> Optional[float]:
    if height_m <= 0:
            return None
                return weight_kg / (height_m ** 2)
                """,
                        "expected": {
                                    "severity": ["low"],  # 期望低风险
                                                "issues_count_max": 1  # 最多1个小问题
                                                        }
                                                            }
                                                            ]
                                                            ```
### 自动化评估

```python
class PromptEvaluator:
    def __init__(self, llm_client, judge_llm_client=None):
            self.llm = llm_client
                    # 使用另一个LLM作为评判者(Judge模式)
                            self.judge = judge_llm_client or llm_client
                                
                                    async def evaluate(self, 
                                                            prompt: Prompt, 
                                                                                    test_cases: list[dict]) -> EvalReport:
                                                                                            results = []
                                                                                                    
                                                                                                            for case in test_cases:
                                                                                                                        # 运行Prompt
                                                                                                                                    output = await self._run_prompt(prompt, case["input"])
                                                                                                                                                
                                                                                                                                                            # 规则检查
                                                                                                                                                                        rule_score = self._check_rules(output, case["expected"])
                                                                                                                                                                                    
                                                                                                                                                                                                # LLM评判(对于主观质量)
                                                                                                                                                                                                            llm_score = await self._judge_quality(
                                                                                                                                                                                                                            prompt_name=prompt.name,
                                                                                                                                                                                                                                            input_text=case["input"],
                                                                                                                                                                                                                                                            output=output,
                                                                                                                                                                                                                                                                            criteria=case.get("quality_criteria", "整体质量和准确性")
                                                                                                                                                                                                                                                                                        )
                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                results.append(EvalResult(
                                                                                                                                                                                                                                                                                                                                case_id=case["id"],
                                                                                                                                                                                                                                                                                                                                                case_name=case["name"],
                                                                                                                                                                                                                                                                                                                                                                output=output,
                                                                                                                                                                                                                                                                                                                                                                                rule_score=rule_score,
                                                                                                                                                                                                                                                                                                                                                                                                llm_score=llm_score,
                                                                                                                                                                                                                                                                                                                                                                                                                passed=rule_score >= 0.8 and llm_score >= 7
                                                                                                                                                                                                                                                                                                                                                                                                                            ))
                                                                                                                                                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                                                                                                                                            return EvalReport(
                                                                                                                                                                                                                                                                                                                                                                                                                                                        prompt_name=prompt.name,
                                                                                                                                                                                                                                                                                                                                                                                                                                                                    prompt_version=prompt.version,
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                total_cases=len(test_cases),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            passed=sum(1 for r in results if r.passed),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        results=results
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                )
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        async def _judge_quality(self, prompt_name, input_text, output, criteria) -> int:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                """使用LLM评判输出质量,返回1-10分"""
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        judge_prompt = f"""
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        请评判以下AI输出的质量。
任务类型:{prompt_name}
输入:{input_text[:500]}
输出:{output[:1000]}
评判标准:{criteria}

请给出1-10的评分(10为满分),并说明理由。
只返回JSON:{{"score": 数字, "reason": "理由"}}
"""
        response = await self.judge.chat.completions.create(
                    model="gpt-4o",
                                messages=[{"role": "user", "content": judge_prompt}],
                                            response_format={"type": "json_object"}
                                                    )
                                                            result = json.loads(response.choices[0].message.content)
                                                                    return result["score"]
                                                                    ```
### CI/CD集成

把Prompt评估集成到你的发布流程:

```yaml
# .github/workflows/prompt-eval.yml
name: Prompt Evaluation

on:
  pull_request:
      paths:
            - 'prompts/**'
jobs:
  evaluate:
      runs-on: ubuntu-latest
          steps:
                - uses: actions/checkout@v3
                -       
                -       - name: Setup Python
                -         uses: actions/setup-python@v4
                -         with:
                -           python-version: '3.11'
                -       
                -       - name: Install dependencies
                -         run: pip install -r requirements.txt
                -       
                -       - name: Run prompt evaluation
                -         env:
                -           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
                -         run: |
                -           python scripts/eval_changed_prompts.py \
                -             --base-ref ${{ github.base_ref }} \
                -             --output-file eval_results.json
                -       
                -       - name: Check evaluation results
                -         run: |
                -           python scripts/check_eval_pass.py \
                -             --results eval_results.json \
                -             --min-pass-rate 0.9
                -       
                -       - name: Comment PR with results
                -         uses: actions/github-script@v6
                -         with:
                -           script: |
                -             const fs = require('fs');
                -             const results = JSON.parse(fs.readFileSync('eval_results.json'));
                -             // 格式化并评论到PR
                - ```
## A/B测试:数据驱动的Prompt优化

新版本Prompt上线前,通过A/B测试验证效果:

```python
class PromptABTest:
    def __init__(self, 
                     control_version: str,    # 当前生产版本
                                      treatment_version: str,  # 待测试新版本
                                                       traffic_split: float = 0.1  # 10%流量给新版本
                                                                       ):
                                                                               self.control = control_version
                                                                                       self.treatment = treatment_version
                                                                                               self.split = traffic_split
                                                                                                       self.metrics = ABMetrics()
                                                                                                           
                                                                                                               def select_version(self, request_id: str) -> str:
                                                                                                                       """基于request_id稳定分流"""
                                                                                                                               import hashlib
                                                                                                                                       hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
                                                                                                                                               if (hash_val % 100) < (self.split * 100):
                                                                                                                                                           return self.treatment
                                                                                                                                                                   return self.control
                                                                                                                                                                       
                                                                                                                                                                           def record_outcome(self, request_id: str, 
                                                                                                                                                                                                  version: str,
                                                                                                                                                                                                                         outcome: dict):
                                                                                                                                                                                                                                 """记录每次请求的结果"""
                                                                                                                                                                                                                                         self.metrics.add(
                                                                                                                                                                                                                                                     version=version,
                                                                                                                                                                                                                                                                 latency_ms=outcome["latency_ms"],
                                                                                                                                                                                                                                                                             user_feedback=outcome.get("thumbs_up"),
                                                                                                                                                                                                                                                                                         error=outcome.get("error")
                                                                                                                                                                                                                                                                                                 )
                                                                                                                                                                                                                                                                                                     
                                                                                                                                                                                                                                                                                                         def get_report(self) -> dict:
                                                                                                                                                                                                                                                                                                                 ctrl = self.metrics.summarize(self.control)
                                                                                                                                                                                                                                                                                                                         trt = self.metrics.summarize(self.treatment)
                                                                                                                                                                                                                                                                                                                                 
                                                                                                                                                                                                                                                                                                                                         return {
                                                                                                                                                                                                                                                                                                                                                     "control": {"version": self.control, **ctrl},
                                                                                                                                                                                                                                                                                                                                                                 "treatment": {"version": self.treatment, **trt},
                                                                                                                                                                                                                                                                                                                                                                             "recommendation": "promote" if trt["positive_rate"] > ctrl["positive_rate"] * 1.05 else "hold"
                                                                                                                                                                                                                                                                                                                                                                                     }
                                                                                                                                                                                                                                                                                                                                                                                     ```
## 一个实用的Prompt变更流程

最后,给一个可落地的工作流:

1. **提案**:在Prompt文件里用注释说明为什么要改,改什么,预期效果
2. 2. **本地测试**:运行对应的测试集,确认通过率不下降
3. 3. **代码评审**:像代码PR一样,让同事review Prompt的改动
4. 4. **Staging验证**:在测试环境验证集成效果
5. 5. **金丝雀发布**:先给5-10%流量用新版本
6. 6. **监控观察**:至少观察24小时的关键指标
7. 7. **全量发布或回滚**:指标好则推全量,差则立即回滚
Prompt管理不是高深的工程,但需要认真对待。一套好的流程,能让你的AI应用在迭代中保持稳定,而不是每次改动都战战兢兢。

---

*本文关键词:Prompt管理、版本控制、A/B测试、LLM工程化、评估流水线*

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐