Prompt Management:版本化与灰度

为什么要把 prompt 从代码里拎出来

很多项目早期把 prompt 硬编码在源码里——改 prompt 要发版,发版要排期,排期要过 code review。结果是:产品同学改 prompt 的成本比写代码还高。

把 prompt 挪到 Langfuse 之后:

• 非工程同学可以在 UI 里直接改 prompt,改完点"发布"就上线
• 每次改都自动产生一个新版本,出问题随时切回上一版
• 通过 production / staging label 做灰度:先灰到 staging,看 trace 和评分没问题再 promote 到 production
• A/B 同一个 name 下的两个 label(如 variant-a / variant-b),按用户分桶

数据模型:name / version / label

创建与更新:SDK + UI 双通道

SDK 侧创建

from langfuse import Langfuse

lf = Langfuse()

lf.create_prompt(
    name="customer-bot-system",
    type="chat",
    prompt=[
        {"role": "system",
         "content": "你是客服小古,只回答和订单、物流、退款相关的问题。\n"
                     "用户姓名: {{user_name}}\n等级: {{tier}}"},
    ],
    labels=["production", "latest"],
    config={"model": "gpt-4o-mini", "temperature": 0.2},
    tags=["customer-bot", "zh"],
)

几个要点:

• {{user_name}} 是 Mustache 风格占位符,运行时用 .compile(user_name="...") 渲染
• labels=["production"] 会把这版直接打到生产;想先灰度,改成 ["staging"] 即可
• config 放和 prompt 一起绑定的运行参数(model、temperature、top_p),SDK 运行时能直接读
• tags 便于 UI 过滤,按业务/语言/模型划分

UI 侧改 prompt

实际工作流通常是:

1. 工程首次把 prompt create 进 Langfuse(上面 SDK 那段,放在迁移脚本里)
2. 之后产品/运营在 UI 里 Prompts → customer-bot-system → Edit,改完保存
3. 新版本默认只挂 latest,不自动进 production
4. 在 staging 环境跑 dataset 回归(下一章)、LLM-as-Judge(第 7 章)
5. UI 里把 production label 从旧版本挪到新版本,上线

运行时:拉 + 编译 + 绑 trace

from langfuse import Langfuse
import openai

lf = Langfuse()

# 1) 拉 prompt(默认本地缓存 60s)
prompt = lf.get_prompt("customer-bot-system", label="production")

# 2) 编译出实际 messages
messages = prompt.compile(user_name="小王", tier="VIP")

# 3) 调模型, 顺带把 prompt 绑到 trace 上
rsp = openai.chat.completions.create(
    model=prompt.config["model"],
    temperature=prompt.config["temperature"],
    messages=messages + [{"role": "user", "content": user_query}],
    # langfuse.openai 包装器识别这个参数, 自动关联
    langfuse_prompt=prompt,
)

关联后,在 Trace 详情页的 generation span 上,你会看到一个 "Linked Prompt: customer-bot-system v7" 跳转,点进去能看到那一版的完整内容。一条 trace 挂一个 prompt 版本——A/B / 回归分析的基础。

@observe 方式

from langfuse.decorators import observe, langfuse_context

@observe(as_type="generation")
def answer(query: str, user_name: str):
    prompt = lf.get_prompt("customer-bot-system", label="production")
    messages = prompt.compile(user_name=user_name, tier="VIP")
    langfuse_context.update_current_observation(prompt=prompt)  # 关联
    return openai.chat.completions.create(
        model=prompt.config["model"],
        messages=messages + [{"role": "user", "content": query}],
    )

缓存:别把 Langfuse 拉爆

默认 SDK 本地缓存 prompt 60 秒。没有缓存,每次 LLM 调用都往 Langfuse Web 打一个 HTTP 请求——QPS 一上来立刻打爆。

# 默认 60s 缓存
prompt = lf.get_prompt("customer-bot-system", label="production")

# 改缓存 TTL(单位秒), 0 = 禁用
prompt = lf.get_prompt("customer-bot-system", label="production", cache_ttl_seconds=300)

# 禁用缓存(开发调试时有用)
prompt = lf.get_prompt("customer-bot-system", label="production", cache_ttl_seconds=0)

prompt = lf.get_prompt(
    "customer-bot-system",
    label="production",
    fallback=[{"role": "system", "content": "你是客服小古..."}],
)

灰度发布:label 挪动就是流量切换

典型的三段式灰度:

1. 开发阶段:新版本挂 dev label,本地环境拉 label=dev
2. 灰度阶段:挂 staging,staging 环境代码 label=staging,跑 CI 回归 + 少量真实流量
3. 上线:把 production label 从旧版本挪到新版本——瞬间全量切换。出事了直接挪回旧版本,秒级回滚

# 用 SDK 改 label(也能在 UI 点)
lf.update_prompt(
    name="customer-bot-system",
    version=8,
    new_labels=["production"],  # 从旧版本把 production 抢过来
)

A/B 分桶:两个 label 按用户切

Langfuse 本身不做流量分桶,但它不拦着你。最常见的做法是在应用层按 user_id 哈希,决定去拉哪个 label:

import hashlib

def pick_variant(user_id: str) -> str:
    # 50/50 稳定分桶
    h = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
    return "variant-a" if h % 2 == 0 else "variant-b"

label = pick_variant(user_id)
prompt = lf.get_prompt("customer-bot-system", label=label)

# 把分桶信息写入 trace,后续按 label 分组分析
langfuse_context.update_current_trace(tags=[f"ab:{label}"])

之后在 Traces 面板按 ab:variant-a / ab:variant-b 过滤,对比 cost / latency / score 三个维度的差异。分桶足够大几天就能看出统计显著性。

和 LangChain 绑定

Langfuse 的 ChatPromptClient 可以直接转 LangChain 的 ChatPromptTemplate:

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langfuse.callback import CallbackHandler

handler = CallbackHandler()

lf_prompt = lf.get_prompt("customer-bot-system", label="production")

# 关键: 转 LangChain, 同时保留 langfuse_prompt 关联
lc_prompt = ChatPromptTemplate.from_messages(lf_prompt.get_langchain_prompt())
lc_prompt.metadata = {"langfuse_prompt": lf_prompt}

llm = ChatOpenAI(model=lf_prompt.config["model"])
chain = lc_prompt | llm

rsp = chain.invoke(
    {"user_name": "小王", "tier": "VIP"},
    config={"callbacks": [handler]},
)

LangChain 在渲染占位符时会自动识别 metadata["langfuse_prompt"],把那条 generation 关联到对应 prompt 版本。

和 LlamaIndex 绑定

from llama_index.core import PromptTemplate
from langfuse.llama_index import LlamaIndexInstrumentor

LlamaIndexInstrumentor().start()

lf_prompt = lf.get_prompt("rag-answer", label="production", type="text")

# LlamaIndex 的 PromptTemplate 用 {var} 占位, 注意转换
template_str = lf_prompt.get_langchain_prompt()  # 返回 {var} 风格
qa_prompt = PromptTemplate(template_str)

query_engine.update_prompts({"response_synthesizer:text_qa_template": qa_prompt})

回滚策略:三条腿

本章小结