Manage prompt template versions with Langfuse

Loading diagram...

• Line 1: Declares a Mermaid flowchart with top-down (TD) layout direction.
• Line 2: Defines the entry node A labeled "Incoming Request" and connects it via an arrow to node B labeled "PromptVersionManager".
• Line 3: Connects B to a decision diamond C labeled "A/B Rollout Selection", indicating a branching point based on traffic splitting.
• Lines 4-5: Define the two weighted branches from the A/B decision: 70% of traffic routes to node D (the stable production prompt v3) and 30% routes to node E (the candidate prompt v4 being tested).
• Lines 6-7: Merge both branches by connecting nodes D and E into a single node F labeled "Fetch Template from Langfuse API", meaning regardless of which version was selected, the prompt template is retrieved from Langfuse.
• Line 8: Connects F to node G labeled "Render Template with Variables", where the fetched prompt template is hydrated with runtime variable values.
• Line 9: Connects G to node H labeled "LiteLLM Completion Call", where the rendered prompt is sent to an LLM via the LiteLLM abstraction layer.
• Line 10: Connects H to node I labeled "Langfuse Trace with version metadata", recording the LLM call as an observability trace in Langfuse that includes which prompt version was used.
• Line 11: Connects I to the terminal node J labeled "Quality Dashboard - Segmented by Version", where traced results are visualized on a dashboard that segments quality metrics by prompt version for A/B comparison analysis.

1import random
2from dataclasses import dataclass
3from typing import Optional
4
5from langfuse import Langfuse
6
7@dataclass
8class ABRolloutConfig:
9    version: int
10    weight: float  # 0.0 to 1.0
11
12class PromptVersionManager:
13    """Manages prompt template versions via Langfuse API."""
14
15    def __init__(self, public_key: str, secret_key: str, host: str):
16        self.client = Langfuse(
17            public_key=public_key,
18            secret_key=secret_key,
19            host=host,
20        )
21
22    def register_prompt(
23        self,
24        name: str,
25        template: str,
26        labels: Optional[list[str]] = None,
27        config: Optional[dict] = None,
28    ) -> int:
29        """Create a new prompt version; returns the version number."""
30        prompt = self.client.create_prompt(
31            name=name,
32            prompt=template,
33            labels=labels or [],
34            config=config or {},
35            type="text",
36        )
37        return prompt.version
38
39    def get_prompt(
40        self,
41        name: str,
42        version: Optional[int] = None,
43        label: Optional[str] = None,
44    ) -> tuple[str, int]:
45        """Fetch prompt template and version number."""
46        kwargs = {"name": name}
47        if version is not None:
48            kwargs["version"] = version
49        elif label is not None:
50            kwargs["label"] = label
51        prompt = self.client.get_prompt(**kwargs)
52        return prompt.prompt, prompt.version
53
54    def select_version_ab(
55        self,
56        name: str,
57        variants: list[ABRolloutConfig],
58        request_variables: dict,
59    ) -> tuple[str, dict]:
60        """Select a prompt version by A/B weight and render it."""
61        total = sum(v.weight for v in variants)
62        if abs(total - 1.0) > 1e-6:
63            raise ValueError(
64                f"Weights must sum to 1.0, got {total:.4f}"
65            )
66
67        roll = random.random()
68        cumulative = 0.0
69        selected_version = variants[-1].version
70
71        for variant in variants:
72            cumulative += variant.weight
73            if roll < cumulative:
74                selected_version = variant.version
75                break
76
77        template, version = self.get_prompt(
78            name=name, version=selected_version
79        )
80
81        for var_name, value in request_variables.items():
82            placeholder = "{" + "{" + var_name + "}" + "}"
83            template = template.replace(placeholder, str(value))
84
85        metadata = {
86            "prompt_name": name,
87            "prompt_version": version,
88            "ab_roll": round(roll, 6),
89            "ab_variants": {v.version: v.weight for v in variants},
90        }
91        return template, metadata

• Lines 1-5: Import random for probabilistic selection, dataclass for the lightweight container, Optional for parameter annotations, and the Langfuse SDK client.
• Lines 8-11: Define ABRolloutConfig with two fields: version (integer prompt version in Langfuse) and weight (a float controlling traffic proportion).
• Lines 14-22: The __init__ instantiates a Langfuse client using the public key, secret key, and host URL — these three values uniquely identify your Langfuse project.
• Lines 24-38: The register_prompt method calls self.client.create_prompt with the template string, an optional list of labels (e.g., ["production"]), and an optional config dictionary for model parameters. The method returns the integer version number assigned by Langfuse, which auto-increments with each call for the same prompt name.
• Lines 40-53: The get_prompt method builds a keyword arguments dictionary dynamically. When version is not None, it fetches that exact version. When label is not None, it fetches the version carrying that label. If both are None, Langfuse returns the latest production-labeled version by default. The method returns a tuple of the template string and version number — both needed for downstream trace tagging.
• Lines 55-69: The select_version_ab method first validates that all weights sum to 1.0 within a tolerance of 1e-6. If the check fails, it raises a ValueError with the actual total — preventing silent traffic misallocation that would invalidate experiment results.
• Lines 71-78: A single call to random.random() generates a uniform float in [0.0, 1.0). The loop accumulates weights until the cumulative sum exceeds the roll value, selecting the corresponding version. If no variant triggers the break (a floating-point edge case), the last variant serves as fallback.
• Lines 80-86: The selected version number is passed to get_prompt, which fetches the exact template text from Langfuse. A simple loop then substitutes each Mustache-style double-brace placeholder in the template with the corresponding request value, converting each to a string via str().
• Lines 88-93: The metadata dictionary captures everything needed for trace correlation: prompt name, selected version, the raw roll value (useful for debugging selection logic), and the full variant configuration. This dictionary gets passed directly to Langfuse's trace or generation span.

1from langfuse.decorators import observe, langfuse_context
2import litellm
3
4manager = PromptVersionManager(
5    public_key="pk-lf-...",
6    secret_key="sk-lf-...",
7    host="https://cloud.langfuse.com",
8)
9
10v1 = manager.register_prompt(
11    name="rag_answer",
12    template="Context: {{context}}\nQuestion: {{question}}\nAnswer concisely.",
13    labels=["production"],
14)
15v2 = manager.register_prompt(
16    name="rag_answer",
17    template="Context: {{context}}\nQuestion: {{question}}\nProvide a detailed answer with reasoning.",
18)
19
20rollout = [
21    ABRolloutConfig(version=v1, weight=0.7),
22    ABRolloutConfig(version=v2, weight=0.3),
23]
24
25@observe(name="rag_pipeline")
26def answer_question(context: str, question: str) -> str:
27    rendered, meta = manager.select_version_ab(
28        name="rag_answer",
29        variants=rollout,
30        request_variables={"context": context, "question": question},
31    )
32
33    langfuse_context.update_current_observation(
34        metadata=meta,
35    )
36
37    response = litellm.completion(
38        model="gpt-4o-mini",
39        messages=[{"role": "user", "content": rendered}],
40        metadata={"prompt_version": meta["prompt_version"]},
41    )
42    return response.choices[0].message.content

• Lines 1-2: Import the @observe decorator and langfuse_context for trace manipulation, plus litellm for the model call.
• Lines 5-9: Instantiate the PromptVersionManager with Langfuse project credentials. In production, these values come from environment variables, never hardcoded.
• Lines 11-19: Register two versions of the "rag_answer" prompt. Version v1 receives the "production" label and uses concise instruction phrasing. Version v2 omits the label (it becomes a candidate) and instructs the model to provide detailed reasoning. Each register_prompt call returns the auto-assigned version integer.
• Lines 21-24: Build the rollout configuration: 70% of traffic uses v1 (production), 30% uses v2 (candidate). Adjusting these weights requires only changing the float values—no code redeployment needed if you externalize the config to a YAML file or environment variable.
• Lines 27-28: The @observe decorator wraps answer_question in a Langfuse trace named "rag_pipeline". Every call to this function automatically starts a new trace with timing, input/output capture, and a unique trace ID.
• Lines 29-33: Inside the function, select_version_ab picks a version, fetches and renders the template, and returns the metadata dictionary.
• Lines 35-37: langfuse_context.update_current_observation attaches the version metadata to the active trace span. This is the critical correlation step—without it, the Langfuse dashboard cannot filter or group traces by prompt version.
• Lines 39-43: The litellm.completion call sends the rendered prompt to the model. The metadata parameter on the LiteLLM call passes the prompt version through to Langfuse's generation span via LiteLLM's built-in Langfuse callback integration, providing a second correlation point at the generation level in addition to the trace level.