一、问题：你的 Prompt 还在代码里写死吗？

看一段真实的生产代码：

# 散落在各处的 prompt —— 噩梦的开始
def summarize(text):
    return llm.chat(f"Summarize the following text in 3 bullet points:\n{text}")

def analyze(text):
    return llm.chat(f"You are a data analyst. Analyze:\n{text}")

def translate(text, lang):
    return llm.chat(f"Translate to {lang}: {text}")

三个月后，你需要：

• 给所有 prompt 加一个 system message 前缀
• 把 summarize 的 bullet points 从 3 改成 5
• 上线一个新版本的 prompt，同时保留旧版本做 A/B 对比
• 搞清楚哪个版本的 prompt 效果好

你会发现：prompt 散落在代码里 = 技术债务。

本文给你一套可落地的 Prompt 工程化方案：模板管理、Git 版本控制、注册中心和 A/B 测试框架。

---

二、整体架构

┌─────────────────────────────────────────────────────┐
│                   PromptRegistry                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ v1.0.0   │  │ v1.1.0   │  │ v2.0.0   │  ...     │
│  │ (active) │  │ (A test) │  │ (draft)  │          │
│  └──────────┘  └──────────┘  └──────────┘          │
│                                                     │
│  ┌──────────────────────────────────────┐           │
│  │       Jinja2 Template Engine         │           │
│  │  system_prompt + user_prompt + vars  │           │
│  └──────────────────────────────────────┘           │
│                                                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ Git Repo │  │ A/B Test │  │ Metrics  │          │
│  │ Version  │  │ Router   │  │ Collector│          │
│  └──────────┘  └──────────┘  └──────────┘          │
└─────────────────────────────────────────────────────┘

核心设计原则：

1. Prompt 是资产，不是字符串 —— 有版本、有元数据、可追溯
2. 模板与逻辑分离 —— Jinja2 模板在独立文件/目录，代码只负责填入变量
3. 变更有据可查 —— Git 管理 prompt 文件，每次变更都有 commit message
4. 效果可量化 —— 内置 A/B 测试，数据驱动 prompt 优化

---

三、核心实现

3.1 Prompt 模板文件结构

prompts/
├── manifest.yaml          # 注册表：记录每个 prompt 的所有版本
├── summarization/
│   ├── v1.0.0/
│   │   ├── system.j2      # System prompt 模板
│   │   ├── user.j2        # User prompt 模板
│   │   └── metadata.yaml  # 元数据
│   └── v1.1.0/
│       ├── system.j2
│       ├── user.j2
│       └── metadata.yaml
├── code_review/
│   └── v1.0.0/
└── customer_support/
    ├── v1.0.0/
    └── v2.0.0-beta/

3.2 元数据定义 (metadata.yaml)

# prompts/summarization/v1.0.0/metadata.yaml
name: summarization
version: 1.0.0
status: active          # active | a_test | b_test | deprecated
author: zhangsan
created: "2026-01-15"
description: "通用文本摘要，输出 3 条 bullet points"
model_requirements:
  min_tokens: 4096
  recommended_model: "gpt-4o-mini"
input_schema:
  text: "string (required) - 待摘要文本"
  max_bullets: "int (optional, default=3)"
output_schema:
  format: "markdown bullet list"
tags: ["summarization", "general"]

3.3 Jinja2 模板

{# prompts/summarization/v1.0.0/system.j2 #}
You are a professional content summarizer. Your task is to distill the key
points from the provided text into clear, concise bullet points.

Rules:
- Output exactly {{ max_bullets | default(3) }} bullet points
- Each bullet should be one sentence, no more than 25 words
- Use the same language as the input text
- Do NOT add information that is not in the original text
- Format output as a markdown bullet list

{% if style == "executive" %}
Tone: Executive summary style — direct and decision-oriented.
{% elif style == "academic" %}
Tone: Academic style — precise and neutral.
{% else %}
Tone: General audience — accessible and jargon-free.
{% endif %}

{# prompts/summarization/v1.0.0/user.j2 #}
Please summarize the following text:

---
{{ text }}
---

3.4 Prompt 注册中心（完整可运行代码）

# prompt_registry.py - Prompt 注册中心
# 支持：模板加载、版本管理、A/B 路由、元数据查询
import os
import yaml
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, List
from jinja2 import Environment, FileSystemLoader


@dataclass
class PromptVersion:
    """单个 prompt 版本"""
    name: str
    version: str
    status: str                    # active | a_test | b_test | deprecated
    author: str
    created: str
    description: str
    model_requirements: Dict[str, Any]
    input_schema: Dict[str, str]
    output_schema: Dict[str, str]
    tags: List[str]
    template_path: str

    def to_dict(self) -> dict:
        return {
            "name": self.name, "version": self.version,
            "status": self.status, "author": self.author,
            "created": self.created, "description": self.description,
            "model_requirements": self.model_requirements,
            "input_schema": self.input_schema,
            "output_schema": self.output_schema,
            "tags": self.tags, "template_path": self.template_path
        }


class PromptRegistry:
    """Prompt 注册中心

    用法:
        registry = PromptRegistry("prompts/")

        # 获取生产版本
        prompt = registry.get("summarization")
        messages = prompt.render(text="...", max_bullets=5)

        # 获取 A/B 测试版本
        prompt_a = registry.get("summarization", channel="a")
        prompt_b = registry.get("summarization", channel="b")
    """

    def __init__(self, prompts_dir: str):
        self.prompts_dir = Path(prompts_dir)
        self._cache: Dict[str, Dict[str, "PromptTemplate"]] = {}
        self._jinja_env = Environment(
            loader=FileSystemLoader("/"),
            trim_blocks=True,
            lstrip_blocks=True,
        )
        self._load_manifest()

    def _load_manifest(self):
        manifest_path = self.prompts_dir / "manifest.yaml"
        if not manifest_path.exists():
            raise FileNotFoundError(f"manifest.yaml not found at {manifest_path}")
        with open(manifest_path, "r", encoding="utf-8") as f:
            self.manifest = yaml.safe_load(f) or {}

    def get(self, name: str, version: Optional[str] = None,
            channel: Optional[str] = None) -> "PromptTemplate":
        if name not in self.manifest:
            raise KeyError(f"Prompt '{name}' not found in manifest")
        versions_info = self.manifest[name]

        # 1. 指定版本
        if version:
            target = next((v for v in versions_info if v["version"] == version), None)
            if not target:
                raise ValueError(f"Version '{version}' not found for prompt '{name}'")
            return self._load_template(name, target)

        # 2. A/B 测试通道
        if channel:
            status_map = {"a": "a_test", "b": "b_test"}
            target_status = status_map.get(channel, "active")
            target = next((v for v in versions_info if v["status"] == target_status), None)
            if target:
                return self._load_template(name, target)

        # 3. 默认生产版本
        target = next((v for v in versions_info if v["status"] == "active"), None)
        if not target:
            raise ValueError(f"No active version found for prompt '{name}'")
        return self._load_template(name, target)

    def _load_template(self, name: str, version_info: dict) -> "PromptTemplate":
        version_key = version_info["version"]
        if name in self._cache and version_key in self._cache[name]:
            return self._cache[name][version_key]

        template_dir = self.prompts_dir / name / version_key
        metadata_path = template_dir / "metadata.yaml"
        with open(metadata_path, "r", encoding="utf-8") as f:
            metadata = yaml.safe_load(f)

        system_tpl = (template_dir / "system.j2").read_text(encoding="utf-8")
        user_tpl = (template_dir / "user.j2").read_text(encoding="utf-8")

        prompt_version = PromptVersion(
            name=metadata["name"], version=metadata["version"],
            status=metadata["status"], author=metadata["author"],
            created=metadata["created"], description=metadata["description"],
            model_requirements=metadata["model_requirements"],
            input_schema=metadata["input_schema"],
            output_schema=metadata["output_schema"],
            tags=metadata["tags"], template_path=str(template_dir),
        )

        template = PromptTemplate(
            system_template=system_tpl, user_template=user_tpl,
            version=prompt_version, jinja_env=self._jinja_env,
        )

        if name not in self._cache:
            self._cache[name] = {}
        self._cache[name][version_key] = template
        return template

    def list_versions(self, name: str) -> List[Dict]:
        if name not in self.manifest:
            return []
        return [
            {"version": v["version"], "status": v["status"],
             "description": v.get("description", "")}
            for v in self.manifest[name]
        ]

    def list_all(self) -> Dict[str, List[Dict]]:
        return {name: self.list_versions(name) for name in self.manifest}

    def add_version(self, name: str, version: str, metadata: dict,
                    system_template: str, user_template: str):
        template_dir = self.prompts_dir / name / version
        template_dir.mkdir(parents=True, exist_ok=True)
        with open(template_dir / "metadata.yaml", "w", encoding="utf-8") as f:
            yaml.dump(metadata, f, allow_unicode=True)
        (template_dir / "system.j2").write_text(system_template, encoding="utf-8")
        (template_dir / "user.j2").write_text(user_template, encoding="utf-8")
        if name not in self.manifest:
            self.manifest[name] = []
        self.manifest[name].append({
            "version": version,
            "status": metadata.get("status", "active"),
            "description": metadata.get("description", ""),
        })
        with open(self.prompts_dir / "manifest.yaml", "w", encoding="utf-8") as f:
            yaml.dump(self.manifest, f, allow_unicode=True)
        self._cache.pop(name, None)


class PromptTemplate:
    """一个具体的 prompt 模板实例"""

    def __init__(self, system_template: str, user_template: str,
                 version: PromptVersion, jinja_env: Environment):
        self._system_tpl = jinja_env.from_string(system_template)
        self._user_tpl = jinja_env.from_string(user_template)
        self.version = version

    def render(self, **kwargs) -> Dict[str, str]:
        system = self._system_tpl.render(**kwargs)
        user = self._user_tpl.render(**kwargs)
        return {"system": system, "user": user}

    def render_openai(self, **kwargs) -> List[Dict[str, str]]:
        rendered = self.render(**kwargs)
        messages = []
        if rendered["system"].strip():
            messages.append({"role": "system", "content": rendered["system"]})
        messages.append({"role": "user", "content": rendered["user"]})
        return messages

    def __repr__(self):
        return f"<PromptTemplate name={self.version.name} v={self.version.version}>"


# ============================================================
# A/B 测试路由器
# ============================================================

import hashlib
from collections import defaultdict


class ABTestRouter:
    """A/B 测试路由器 - 按用户 ID 哈希分流"""

    def __init__(self, registry: PromptRegistry):
        self.registry = registry
        self.experiments: Dict[str, dict] = {}

    def create_experiment(self, name: str, prompt_name: str,
                          channel_a: str, channel_b: str,
                          traffic_split: float = 0.5):
        self.experiments[name] = {
            "prompt_name": prompt_name,
            "channel_a": channel_a,
            "channel_b": channel_b,
            "traffic_split": traffic_split,
            "created_at": datetime.now().isoformat(),
            "metrics": {"a_count": 0, "b_count": 0},
        }

    def route(self, prompt_name: str, user_id: str) -> PromptTemplate:
        active_experiments = [
            (name, exp) for name, exp in self.experiments.items()
            if exp["prompt_name"] == prompt_name
        ]
        if not active_experiments:
            return self.registry.get(prompt_name)

        exp_name, experiment = active_experiments[0]
        hash_val = int(hashlib.md5(f"{exp_name}:{user_id}".encode()).hexdigest()[:8], 16)
        bucket = (hash_val % 10000) / 10000.0

        if bucket < experiment["traffic_split"]:
            if bucket < experiment["traffic_split"] / 2:
                experiment["metrics"]["a_count"] += 1
                return self.registry.get(prompt_name, version=experiment["channel_a"])
            else:
                experiment["metrics"]["b_count"] += 1
                return self.registry.get(prompt_name, version=experiment["channel_b"])
        else:
            return self.registry.get(prompt_name)

    def get_metrics(self, experiment_name: str) -> dict:
        return self.experiments[experiment_name]["metrics"]

    def stop_experiment(self, experiment_name: str):
        self.experiments.pop(experiment_name, None)


# ============================================================
# 指标收集器
# ============================================================

@dataclass
class PromptExecution:
    prompt_name: str
    version: str
    user_id: str
    input_vars: Dict[str, Any]
    rendered_messages: List[Dict]
    output: str
    latency_ms: float
    token_count: int
    cost_usd: float
    feedback: Optional[str] = None
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


class MetricsCollector:
    def __init__(self):
        self.executions: List[PromptExecution] = []

    def record(self, execution: PromptExecution):
        self.executions.append(execution)

    def get_summary(self, prompt_name: str, version: Optional[str] = None) -> dict:
        filtered = [
            e for e in self.executions
            if e.prompt_name == prompt_name
            and (version is None or e.version == version)
        ]
        if not filtered:
            return {"count": 0}

        latencies = [e.latency_ms for e in filtered]
        likes = sum(1 for e in filtered if e.feedback == "thumbs_up")
        dislikes = sum(1 for e in filtered if e.feedback == "thumbs_down")
        total_with_feedback = likes + dislikes

        return {
            "count": len(filtered),
            "latency": {
                "avg_ms": sum(latencies) / len(latencies),
                "p50_ms": sorted(latencies)[len(latencies) // 2],
                "p95_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            },
            "feedback": {
                "rate": total_with_feedback / len(filtered) if filtered else 0,
                "satisfaction": likes / total_with_feedback if total_with_feedback else None,
            },
        }


# ============================================================
# 使用示例
# ============================================================

if __name__ == "__main__":
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        prompts_dir = Path(tmpdir) / "prompts"
        prompts_dir.mkdir()

        manifest = {
            "summarization": [
                {"version": "1.0.0", "status": "active",
                 "description": "摘要 v1 - 3 bullet points"},
                {"version": "2.0.0", "status": "draft",
                 "description": "摘要 v2 - 5 bullet points + 标签"}
            ]
        }
        with open(prompts_dir / "manifest.yaml", "w") as f:
            yaml.dump(manifest, f)

        # 创建 v1.0.0
        v1_dir = prompts_dir / "summarization" / "1.0.0"
        v1_dir.mkdir(parents=True)
        metadata_v1 = {
            "name": "summarization", "version": "1.0.0",
            "status": "active", "author": "dev-team",
            "created": "2026-01-15",
            "description": "通用文本摘要 v1",
            "model_requirements": {"min_tokens": 4096},
            "input_schema": {"text": "string"},
            "output_schema": {"format": "markdown bullet list"},
            "tags": ["summarization"]
        }
        with open(v1_dir / "metadata.yaml", "w") as f:
            yaml.dump(metadata_v1, f)
        (v1_dir / "system.j2").write_text(
            "You are a professional summarizer. "
            "Output {{ max_bullets | default(3) }} bullet points.\n"
            "Style: {{ style | default('general') }}."
        )
        (v1_dir / "user.j2").write_text("Text to summarize:\n{{ text }}")

        # 使用注册中心
        registry = PromptRegistry(str(prompts_dir))
        print("=" * 60)
        print("所有 Prompt 版本：")
        for name, versions in registry.list_all().items():
            print(f"\n  {name}:")
            for v in versions:
                print(f"    v{v['version']} [{v['status']}]")

        prompt = registry.get("summarization")
        rendered = prompt.render(text="Python 3.13 引入无 GIL 支持...",
                                 max_bullets=3, style="executive")
        print(f"\n渲染结果: {rendered['system'][:60]}...")

        # A/B 测试
        router = ABTestRouter(registry)
        router.create_experiment(
            name="summarize_v2_test", prompt_name="summarization",
            channel_a="1.0.0", channel_b="2.0.0", traffic_split=0.5,
        )
        print("\nA/B 路由测试：")
        for uid in ["user_001", "user_002", "user_003", "user_004"]:
            p = router.route("summarization", uid)
            print(f"  {uid} -> v{p.version.version}")

        print("\nPrompt 工程化框架运行成功！")

---

四、接入真实 LLM：工厂函数

# llm_client.py - PromptTemplate 对接 OpenAI API
from openai import OpenAI
import time


class LLMService:
    """LLM 服务层：渲染 + 调用 + 记录"""

    def __init__(self, registry, router, collector, openai_client):
        self.registry = registry
        self.router = router
        self.collector = collector
        self.client = openai_client

    def invoke(self, prompt_name: str, user_id: str,
               model: str = "gpt-4o-mini", **template_vars) -> str:
        # 1. A/B 路由
        template = self.router.route(prompt_name, user_id)
        # 2. 渲染
        messages = template.render_openai(**template_vars)
        # 3. 调用 LLM
        start = time.time()
        response = self.client.chat.completions.create(
            model=model, messages=messages, temperature=0.7,
        )
        latency_ms = (time.time() - start) * 1000
        output = response.choices[0].message.content
        # 4. 记录指标
        self.collector.record(PromptExecution(
            prompt_name=prompt_name,
            version=template.version.version,
            user_id=user_id,
            input_vars=template_vars,
            rendered_messages=messages,
            output=output,
            latency_ms=latency_ms,
            token_count=response.usage.total_tokens if response.usage else 0,
            cost_usd=self._estimate_cost(model, response.usage),
        ))
        return output

    def _estimate_cost(self, model: str, usage) -> float:
        if not usage:
            return 0.0
        rates = {"gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006)}
        input_rate, output_rate = rates.get(model, (0.001, 0.002))
        cost = (usage.prompt_tokens / 1000) * input_rate
        cost += (usage.completion_tokens / 1000) * output_rate
        return round(cost, 6)

---

五、对比：有无 Prompt 工程化的差距

维度	散落字符串	Prompts as Code
修改 prompt	改代码 → 重新部署整个应用	改 YAML/模板 → 热加载
版本管理	git log 里翻找	manifest.yaml + 语义化版本
A/B 测试	改代码 if/else	声明式实验配置
效果评估	“感觉变好了”	延迟/成本/满意度 量化对比
团队协作	“你改了我 prompt？”	Pull Request + Code Review
回滚	git revert + 重新部署	改 manifest status 一行

---

六、进阶方向

1. 自动 Prompt 优化 —— 接入 DSPy 或 promptfoo，自动搜索最优 prompt
2. 多语言 prompt —— 模板中通过 {% if lang == 'zh' %} 分支支持多语言
3. 动态 few-shot —— 从向量库检索最相似示例插入模板
4. Prompt 安全审计 —— 自动扫描模板中的注入风险、敏感词

---

七、总结

Prompt 工程化的核心就是把 prompt 当代码管：

1. 模板化：Jinja2 分离逻辑与文本
2. 版本化：Git + manifest.yaml = 可追溯
3. 可测试：渲染后本地验证，不需调 LLM
4. 可度量：A/B 测试 + 指标收集 = 数据驱动决策