数据工程师如何用 Git 和 LangFuse 管理 Prompt 实现可回滚、可测试的 LLM 应用

场景：用户反映近几天回答质量下降

问题排查：
  - 后端代码没变动（Git log 确认）
  - 模型版本没变（还是 gpt-4o）
  - 检索结果没变（向量库数据正常）

最后发现：
  - 产品经理上周"优化"了 System Prompt
  - 优化内容存在他的 Notion 里
  - 无法对比修改前后的差异
  - 旧版本 Prompt 已经找不到了

LangFuse 后台 → Prompts → Create Prompt
  - Name: rag-answer
  - Prompt 内容（支持 {{variable}} 变量语法）
  - 标记为 production / development 环境
  - 添加标签

from langfuse import Langfuse

langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-..."
)

def get_production_prompt(prompt_name: str, variables: dict) -> str:
    """
    从 LangFuse 获取生产版本 Prompt
    LangFuse 会自动缓存，不用担心每次调用都发请求
    """
    # 获取最新 production 版本（不指定 version 时默认取 production 标签版本）
    prompt_template = langfuse.get_prompt(prompt_name, label="production")

    # 编译：将变量填入模板
    compiled_prompt = prompt_template.compile(**variables)

    return compiled_prompt, prompt_template.config

# 使用示例
system_prompt, config = get_production_prompt(
    "rag-answer",
    variables={
        "context": "Flink 的 Checkpoint 是...",
        "user_question": "Checkpoint 和 Savepoint 的区别？"
    }
)

# config 里包含 Prompt 元数据，可以记录到 LangFuse Trace 中

# 获取特定版本号
prompt_v2 = langfuse.get_prompt("rag-answer", version=2)

# 获取 staging 环境版本（新版本先在 staging 测试）
prompt_staging = langfuse.get_prompt("rag-answer", label="staging")

import os

ENVIRONMENT = os.getenv("APP_ENV", "development")

def get_prompt_for_env(prompt_name: str, variables: dict):
    """根据运行环境自动选择 Prompt 版本"""
    label = "production" if ENVIRONMENT == "production" else "staging"

    prompt = langfuse.get_prompt(prompt_name, label=label)
    return prompt.compile(**variables)

import hashlib
from typing import Literal

def assign_prompt_variant(
    user_id: str,
    experiment_id: str,
    traffic_split: float = 0.5  # 50% 流量给新版本
) -> Literal["control", "treatment"]:
    """
    基于 user_id 的确定性分组（同一用户每次分到同一组）
    避免用 random()，否则同一用户会看到不一致的体验
    """
    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
    normalized = (hash_value % 1000) / 1000.0  # 0.0 ~ 1.0

    return "treatment" if normalized < traffic_split else "control"

def get_prompt_with_ab(user_id: str, variables: dict) -> tuple[str, str]:
    """
    返回：(编译后的 Prompt, 实验组标识)
    """
    variant = assign_prompt_variant(user_id, experiment_id="exp_rag_v2")

    if variant == "treatment":
        # 新 Prompt 版本
        prompt_template = langfuse.get_prompt("rag-answer", version=5)
    else:
        # 对照组（当前 production 版本）
        prompt_template = langfuse.get_prompt("rag-answer", label="production")

    compiled = prompt_template.compile(**variables)
    return compiled, variant

from langfuse import Langfuse

def handle_user_query(user_id: str, query: str, context: str):
    langfuse = Langfuse()
    prompt_text, variant = get_prompt_with_ab(user_id, {"context": context, "query": query})

    trace = langfuse.trace(
        name="rag-ab-test",
        user_id=user_id,
        tags=[f"experiment:exp_rag_v2", f"variant:{variant}"],
        metadata={"ab_variant": variant, "experiment_id": "exp_rag_v2"},
    )

    # ... 调用 LLM，记录结果 ...

    return answer, trace.id

import scipy.stats as stats
import numpy as np

def analyze_ab_results(
    control_scores: list[float],
    treatment_scores: list[float],
    alpha: float = 0.05
) -> dict:
    """
    对两组评估分数做 t 检验，判断差异是否统计显著
    """
    t_stat, p_value = stats.ttest_ind(control_scores, treatment_scores)

    control_mean = np.mean(control_scores)
    treatment_mean = np.mean(treatment_scores)
    relative_improvement = (treatment_mean - control_mean) / control_mean * 100

    return {
        "control_mean": control_mean,
        "treatment_mean": treatment_mean,
        "relative_improvement_pct": relative_improvement,
        "p_value": p_value,
        "statistically_significant": p_value < alpha,
        "sample_sizes": {
            "control": len(control_scores),
            "treatment": len(treatment_scores),
        },
        "recommendation": (
            "上线新版本" if p_value < alpha and treatment_mean > control_mean
            else "保持现有版本" if p_value < alpha
            else "样本量不足或差异不显著，继续收集数据"
        )
    }

# 使用示例（从 LangFuse 拉取评分数据后调用）
result = analyze_ab_results(
    control_scores=[0.72, 0.68, 0.75, 0.70, 0.73],   # 对照组 LLM-as-Judge 分数
    treatment_scores=[0.81, 0.79, 0.83, 0.80, 0.82],  # 实验组分数
)
print(result)
# {"relative_improvement_pct": 11.1, "p_value": 0.003, "recommendation": "上线新版本"}

from jinja2 import Environment, FileSystemLoader, StrictUndefined

# 加载模板目录
env = Environment(
    loader=FileSystemLoader("prompts/templates"),
    undefined=StrictUndefined,  # 未定义变量直接报错，避免静默错误
    trim_blocks=True,           # 去除块标签后的换行
    lstrip_blocks=True,
)

def render_prompt(template_name: str, **variables) -> str:
    template = env.get_template(f"{template_name}.j2")
    return template.render(**variables)

{# prompts/templates/rag-answer.j2 #}
你是一个专业的数据工程助手。

{% if language == "en" %}
Please answer the question based on the provided context.
{% else %}
请基于以下参考文档回答问题。
{% endif %}

参考文档：
{{ context }}

{% if few_shots %}
以下是一些示例，帮助你理解回答格式：
{% for example in few_shots %}
用户：{{ example.user }}
助手：{{ example.assistant }}
---
{% endfor %}
{% endif %}

用户问题：{{ user_question }}

from sentence_transformers import SentenceTransformer
import numpy as np

class DynamicFewShotSelector:
    def __init__(self, example_pool: list[dict], model_name: str = "all-MiniLM-L6-v2"):
        self.examples = example_pool
        self.model = SentenceTransformer(model_name)

        # 预计算所有示例的向量
        questions = [ex["user"] for ex in example_pool]
        self.embeddings = self.model.encode(questions, normalize_embeddings=True)

    def select(self, query: str, top_k: int = 3) -> list[dict]:
        """选择与当前问题最相关的 K 个示例"""
        query_embedding = self.model.encode([query], normalize_embeddings=True)

        # 余弦相似度（因为向量已归一化，点积即余弦相似度）
        similarities = np.dot(self.embeddings, query_embedding.T).flatten()

        top_indices = np.argsort(similarities)[::-1][:top_k]
        return [self.examples[i] for i in top_indices]

# 示例池（实际项目中从数据库或文件加载）
EXAMPLE_POOL = [
    {
        "user": "Kafka 和 RabbitMQ 有什么区别？",
        "assistant": "Kafka 是分布式日志系统，适合高吞吐量的事件流...",
    },
    {
        "user": "Flink 的水位线是什么？",
        "assistant": "水位线（Watermark）是 Flink 处理乱序事件的机制...",
    },
    # ... 更多示例
]

selector = DynamicFewShotSelector(EXAMPLE_POOL)
relevant_examples = selector.select("Spark Streaming 和 Flink 对比", top_k=2)

SYSTEM_PROMPTS = {
    "zh": """你是一个专业的数据工程助手。
请使用简洁的中文回答，避免过多技术术语堆砌。
如果需要提及技术术语，请简要解释其含义。""",

    "en": """You are a professional data engineering assistant.
Please answer concisely in English.
For technical terms, provide brief explanations when appropriate.""",
}

def get_system_prompt(language: str = "zh") -> str:
    return SYSTEM_PROMPTS.get(language, SYSTEM_PROMPTS["zh"])

from dataclasses import dataclass
from typing import Optional
import yaml
import json
from pathlib import Path
from datetime import datetime
import hashlib

@dataclass
class PromptVersion:
    name: str
    version: str
    system_prompt: str
    few_shots: list[dict]
    variables: dict
    metadata: dict

    @property
    def prompt_hash(self) -> str:
        """Prompt 内容的哈希值，用于检测内容变更"""
        content = self.system_prompt + json.dumps(self.few_shots, ensure_ascii=False)
        return hashlib.sha256(content.encode()).hexdigest()[:12]

class PromptManager:
    """
    本地 Git 管理 + LangFuse 远程同步的混合 Prompt 管理器
    """

    def __init__(self, prompts_dir: str = "prompts", langfuse_client=None):
        self.prompts_dir = Path(prompts_dir)
        self.langfuse = langfuse_client
        self._cache: dict[str, PromptVersion] = {}

    def load_from_file(self, name: str, version: Optional[str] = None) -> PromptVersion:
        """从本地文件加载 Prompt"""
        prompt_dir = self.prompts_dir / name

        if version:
            file_path = prompt_dir / f"v{version}.yaml"
        else:
            # 读取 current 软链接指向的版本
            current_link = prompt_dir / "current"
            file_path = current_link.resolve() if current_link.is_symlink() else None
            if not file_path:
                raise FileNotFoundError(f"No current version found for prompt: {name}")

        with open(file_path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f)

        return PromptVersion(
            name=data["metadata"]["name"],
            version=data["metadata"]["version"],
            system_prompt=data["system_prompt"],
            few_shots=data.get("few_shots", []),
            variables=data.get("variables", {}),
            metadata=data["metadata"],
        )

    def get(self, name: str, version: Optional[str] = None, use_cache: bool = True) -> PromptVersion:
        """获取 Prompt，优先从缓存读取"""
        cache_key = f"{name}:{version or 'current'}"

        if use_cache and cache_key in self._cache:
            return self._cache[cache_key]

        # 优先从 LangFuse 获取（如果配置了的话）
        if self.langfuse and version is None:
            try:
                prompt = self._load_from_langfuse(name)
                self._cache[cache_key] = prompt
                return prompt
            except Exception:
                pass  # LangFuse 失败时降级到本地文件

        # 降级到本地文件
        prompt = self.load_from_file(name, version)
        self._cache[cache_key] = prompt
        return prompt

    def _load_from_langfuse(self, name: str) -> PromptVersion:
        """从 LangFuse 加载 production 版本"""
        lf_prompt = self.langfuse.get_prompt(name, label="production")
        # 解析 LangFuse 返回的 Prompt 结构
        config = lf_prompt.config or {}
        return PromptVersion(
            name=name,
            version=str(lf_prompt.version),
            system_prompt=lf_prompt.prompt,
            few_shots=config.get("few_shots", []),
            variables=config.get("variables", {}),
            metadata={"source": "langfuse", "version": lf_prompt.version},
        )

    def compile(self, name: str, **variables) -> str:
        """获取并编译 Prompt（填入变量）"""
        prompt = self.get(name)
        compiled = prompt.system_prompt

        for key, value in variables.items():
            compiled = compiled.replace(f"{{{{{key}}}}}", str(value))

        return compiled

    def audit_log(self, name: str, version: str, user_id: str, action: str):
        """记录 Prompt 使用审计日志"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "prompt_name": name,
            "version": version,
            "user_id": user_id,
            "action": action,
        }
        # 写入审计日志文件
        audit_path = self.prompts_dir / "audit.jsonl"
        with open(audit_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")


# 使用示例
from langfuse import Langfuse

langfuse = Langfuse(public_key="pk-lf-...", secret_key="sk-lf-...")
pm = PromptManager(prompts_dir="prompts", langfuse_client=langfuse)

# 获取并使用 Prompt
compiled_prompt = pm.compile(
    "rag-answer",
    context="Flink 的 Checkpoint 是...",
    user_question="Checkpoint 和 Savepoint 的区别？"
)