生产级实战

提示词版本管理

为什么需要版本管理？

在生产环境中，提示词不是一次性写好就不变的——它需要持续优化、修复问题、适应模型升级。没有版本管理，你无法知道是哪次修改导致了输出质量下降，也无法在出问题时快速回滚。

提示词版本管理系统设计

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
import hashlib

@dataclass
class PromptVersion:
    prompt_id: str             # e.g. "customer_support_v2"
    content: str               # 提示词内容
    version: str               # 语义版本号 "2.1.0"
    model: str                 # 目标模型
    created_at: datetime
    created_by: str
    description: str           # 此版本的变更说明
    metrics: Dict[str, float]  # 性能指标
    is_production: bool = False
    tags: List[str] = field(default_factory=list)

    @property
    def content_hash(self) -> str:
        """内容哈希，用于检测重复"""
        return hashlib.sha256(self.content.encode()).hexdigest()[:8]

class PromptRegistry:
    """提示词注册中心"""

    def get_production_prompt(self, prompt_id: str) -> Optional[PromptVersion]:
        """获取生产版本提示词"""
        versions = self.store.get(prompt_id, [])
        prod = [v for v in versions if v.is_production]
        return prod[0] if prod else None

    def rollback(self, prompt_id: str, version: str) -> None:
        """回滚到指定版本"""
        versions = self.store.get(prompt_id, [])
        for v in versions:
            v.is_production = (v.version == version)
        print(f"已回滚 {prompt_id} 到版本 {version}")

A/B 测试框架

科学评估提示词效果

A/B 测试是量化评估提示词改进效果的标准方法。关键是要有清晰的成功指标和足够的样本量。

import random
from typing import Callable

class PromptABTest:
    """提示词 A/B 测试框架"""

    def __init__(self, prompt_a: str, prompt_b: str,
                 evaluator: Callable, traffic_split: float = 0.5):
        self.prompt_a = prompt_a
        self.prompt_b = prompt_b
        self.evaluator = evaluator  # 评分函数：(output, expected) -> float
        self.traffic_split = traffic_split
        self.results = {"a": [], "b": []}

    def run(self, test_cases: List, llm_fn: Callable) -> Dict:
        for case in test_cases:
            variant = "a" if random.random() < self.traffic_split else "b"
            prompt = self.prompt_a if variant == "a" else self.prompt_b

            output = llm_fn(prompt.format(**case["input"]))
            score = self.evaluator(output, case["expected"])
            self.results[variant].append(score)

        return self.report()

    def report(self) -> Dict:
        a_scores = self.results["a"]
        b_scores = self.results["b"]

        return {
            "a_mean": sum(a_scores) / len(a_scores),
            "b_mean": sum(b_scores) / len(b_scores),
            "a_count": len(a_scores),
            "b_count": len(b_scores),
            "winner": "b" if sum(b_scores)/len(b_scores) > sum(a_scores)/len(a_scores) else "a"
        }

成本优化策略

Token 成本的五种优化方式

团队协作规范

提示词工程的 Git 工作流

# prompts/customer_support/v2.1.0.yaml
id: customer_support
version: "2.1.0"
model: gpt-4o-2024-08-06
temperature: 0.3
max_tokens: 500
created_by: alice@company.com
reviewed_by: bob@company.com
status: production  # draft | staging | production | archived

changelog:
  - version: "2.1.0"
    date: "2025-01-15"
    changes: "增加了对退款政策的详细处理，修复了多语言切换问题"
    metrics:
      resolution_rate: 0.87
      user_satisfaction: 4.3

system_prompt: |
  你是 {company_name} 的客户服务专员...
  [提示词内容]

test_cases:
  - input: "我的订单什么时候到？"
    expected_contains: ["订单号", "物流"]
    expected_not_contains: ["不知道"]

可观测性：提示词监控

关键监控指标

from dataclasses import dataclass
import time

@dataclass
class LLMCallMetrics:
    prompt_id: str
    version: str
    latency_ms: int
    input_tokens: int
    output_tokens: int
    cost_usd: float
    success: bool
    error_type: str = None
    user_feedback: float = None  # 用户评分 1-5

def tracked_llm_call(prompt_id: str, version: str, prompt: str) -> str:
    start = time.time()
    try:
        response = call_llm(prompt)
        metrics = LLMCallMetrics(
            prompt_id=prompt_id,
            version=version,
            latency_ms=int((time.time() - start) * 1000),
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            cost_usd=calculate_cost(response.usage),
            success=True
        )
        send_to_monitoring(metrics)  # 发送到 DataDog/Prometheus 等
        return response.choices[0].message.content
    except Exception as e:
        metrics = LLMCallMetrics(..., success=False, error_type=type(e).__name__)
        send_to_monitoring(metrics)
        raise

生产环境的故障排查与回滚