Build, Test, and Optimize a Trip-Planner Prompt

pip install "arize>=8.0.0" openai pandas

export ARIZE_SPACE_ID="<your-space-id>"
export ARIZE_API_KEY="<your-api-key>"
export OPENAI_API_KEY="<your-openai-api-key>"

import os
from uuid import uuid4

from arize.client import ArizeClient

ARIZE_API_KEY = os.environ["ARIZE_API_KEY"]
ARIZE_SPACE_ID = os.environ["ARIZE_SPACE_ID"]

client = ArizeClient(api_key=ARIZE_API_KEY)

# Every artifact below is tagged with a unique RUN_ID, so you can rerun this end to end
# without colliding with prompts or datasets from a previous pass.
RUN_ID = uuid4().hex[:8]
print(f"Run ID: {RUN_ID}")

from arize.prompts.types import InputVariableFormat, InvocationParams, LlmProvider, LLMMessage, MessageRole

PROMPT_NAME = f"trip-planner-{RUN_ID}"

# v1 system message: deliberately vague — no format constraints, no examples, no rules
# about what to include. This is the "first attempt" most teams ship.
system_message_v1 = (
    "You are a helpful travel planner. Given a destination, duration, and travel style, "
    "produce a trip plan the user can follow."
)

# The user template is the part of the prompt that gets filled in per request. The
# {placeholders} match column names in the dataset we'll build in Section 2 — at run time
# the experiment substitutes the row's value for each placeholder.
user_message = (
    "Plan a {duration} trip to {destination}. Travel style: {travel_style}.\n\n"
    "Research: {research}\n"
    "Budget: {budget_info}\n"
    "Local notes: {local_info}"
)

# client.prompts.create writes a new Prompt Object to Prompt Hub. Every save produces an
# immutable, hashed version that lives in history forever — you can always load this exact
# version later, even after edits.
prompt_v1 = client.prompts.create(
    space=ARIZE_SPACE_ID,
    name=PROMPT_NAME,
    description="Day-by-day itinerary generator for a given destination, duration, and travel style.",
    commit_message="v1: baseline trip planner",  # like a git commit message — shows up in the UI's version history
    input_variable_format=InputVariableFormat.F_STRING,  # {name} placeholders (Python f-string style); MUSTACHE is the {{name}} alternative
    provider=LlmProvider.OPEN_AI,
    model="gpt-5.4-mini",  # The model is part of the Prompt Object — changing it is a new version
    messages=[
        # System message sets the assistant's role; user message carries the per-request payload.
        LLMMessage(role=MessageRole.SYSTEM, content=system_message_v1),
        LLMMessage(role=MessageRole.USER, content=user_message),
    ],
    invocation_params=InvocationParams(temperature=0.7, max_completion_tokens=600),
)

# The returned PromptWithVersion exposes the prompt's metadata (name, id, description) at
# the top level, plus a nested .version with the immutable snapshot (messages, model,
# invocation_params, labels, etc.).
print(f"Created prompt {prompt_v1.name} (id={prompt_v1.id})")
print(f"Initial version: {prompt_v1.version.id}")

import pandas as pd

DATASET_NAME = f"trip-planner-test-set-{RUN_ID}"

# Each dict in this list is one dataset row. The keys must match the {placeholders} in the
# prompt's user_message — the experiment substitutes row[col] for {col} when it renders
# the prompt per row.
#
# Five rows is enough to demonstrate the comparison while keeping the API spend negligible.
# In production you'd build a much larger and more carefully-balanced golden dataset; here we keep it small and varied (different durations, regions, travel styles)
# so the prompts have to handle range.
examples = [
    {
        "destination": "Istanbul, Turkey",
        "duration": "3 days",
        "travel_style": "standard",
        "research": "Best time to visit is April-May. Top attractions: Hagia Sophia, Blue Mosque, Grand Bazaar.",
        "budget_info": "Mid-range hotels $80-120/night, meals $15-30 each, transport $5-10/day.",
        "local_info": "Try Karaköy for breakfast spots; avoid Sultanahmet for dinner due to tourist markup.",
    },
    {
        "destination": "Bangkok, Thailand",
        "duration": "4 days",
        "travel_style": "family-friendly",
        "research": "Cool season Nov-Feb best. Top kid-friendly: Safari World, Dream World, river cruise.",
        "budget_info": "Family rooms $100-180/night, street food $2-5, taxis $5-15.",
        "local_info": "Skytrain reliable; visit temples early morning to beat heat.",
    },
    {
        "destination": "Barcelona, Spain",
        "duration": "5 days",
        "travel_style": "romantic",
        "research": "Spring/fall ideal. Sagrada Familia, Park Güell, tapas tours, beach access from city center.",
        "budget_info": "Boutique hotels $150-250/night, tapas dinner $30-50, metro $12 day pass.",
        "local_info": "Dinner starts late (9pm+); reserve Sagrada Familia tickets weeks ahead.",
    },
    {
        "destination": "Reykjavik, Iceland",
        "duration": "4 days",
        "travel_style": "adventure",
        "research": "Sept-Mar for northern lights; June-Aug for midnight sun. Blue Lagoon, Golden Circle, glacier tours.",
        "budget_info": "Hotels $200-300/night, dinner $40-70, day tours $80-150.",
        "local_info": "Rent a car for flexibility; weather changes fast — pack layers.",
    },
    {
        "destination": "New York City, USA",
        "duration": "2 days",
        "travel_style": "standard",
        "research": "Times Square, Central Park, museums (Met, MoMA), Brooklyn Bridge walk.",
        "budget_info": "Hotels $250-400/night, meals $20-50, subway $33 7-day pass.",
        "local_info": "Walk where possible; subway faster than taxis in midtown.",
    },
]

# client.datasets.create accepts a list of dicts or a pandas DataFrame. Each column in the
# DataFrame becomes a dataset column in Arize. The dataset is versioned — appending
# examples later creates a new version of the same dataset.
examples_df = pd.DataFrame(examples)
dataset = client.datasets.create(
    name=DATASET_NAME,
    space=ARIZE_SPACE_ID,
    examples=examples_df,
)
print(f"Created dataset {dataset.name} with {len(examples)} rows (id={dataset.id})")

from openai import OpenAI

# A single OpenAI client is reused across all task invocations.
oai = OpenAI()


def make_task(system_message: str, user_template: str, model: str = "gpt-5.4-mini"):
    """Build an experiment task that runs the trip-planner prompt against each row.

    Using a factory lets v1 and v2 share rendering + LLM-call logic. The only thing that
    differs between the two runs is the system message and (optionally) the model — both
    are captured by closure.
    """

    def task(dataset_row) -> str:
        # The experiment runner passes one dataset row per call. The row is dict-like,
        # so .format(**row) substitutes each {placeholder} with the row's column value.
        rendered_user = user_template.format(**dataset_row)

        # Standard chat-completions call. The Prompt Object's model and invocation params
        # would normally come from Prompt Hub; we inline them here so the
        # task stands alone.
        resp = oai.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system_message},
                {"role": "user", "content": rendered_user},
            ],
            temperature=0.7,  # Matches the Prompt Object's invocation_params for consistency
        )
        return resp.choices[0].message.content

    return task


# Bind v1's system message into a ready-to-run task. We'll do the same for v2 in Section 3.
task_v1 = make_task(system_message_v1, user_message)

import re
from arize.experiments.evaluators.types import EvaluationResult


def itinerary_structure_eval(output, dataset_row) -> EvaluationResult:
    """Score: how closely the output follows the strict 'Day N: HH:MM - Activity - $cost' shape.

    Three signals, each worth a third:
      1. Every day mentioned (Day 1, Day 2, ...).
      2. At least one HH:MM time stamp per day.
      3. At least one $cost figure per day.

    Returns an EvaluationResult — the canonical evaluator return type. The SDK's experiment
    runner will turn this into eval.<name>.{label,score,explanation} columns on the result
    dataframe and on the experiment record in the AX UI.
    """
    try:
        # Defensive: a failing task can return None or a non-string. Score it 0 and move on
        # rather than crashing the whole experiment.
        if not output or not isinstance(output, str):
            return EvaluationResult(label="empty", score=0.0, explanation="Empty or non-string output.")

        # The experiment runner passes a ReadOnly view rather than a plain dict, so normalize
        # before doing dict-style access.
        row = dataset_row if isinstance(dataset_row, dict) else dict(dataset_row)

        # Parse the duration string ("3 days", "5 days") to the number of days we expect to see.
        duration = str(row.get("duration", ""))
        match = re.search(r"\d+", duration)
        n_days = int(match.group()) if match else 1

        # ---------- Signal 1: every day mentioned ----------
        # Look for "Day 1", "Day 2", ... up to n_days. \b ensures word boundaries so
        # "Day 12" doesn't accidentally satisfy n_days=1.
        days_present = sum(
            1 for d in range(1, n_days + 1)
            if re.search(rf"\bDay\s*{d}\b", output, re.IGNORECASE)
        )
        days_coverage = days_present / n_days  # 0.0 to 1.0

        # ---------- Signal 2: HH:MM time stamps ----------
        # Matches "9:00", "09:30", "14:45", etc. Requiring at least one per day on average
        # filters out outputs that only use "Morning/Afternoon" section headers. This is
        # the signal that reliably separates v1 (prose with vague time-of-day labels) from
        # v2 (which is forced into "Day N: HH:MM - ..." rows).
        time_markers = len(re.findall(r"\b\d{1,2}:\d{2}\b", output))
        has_times = time_markers >= n_days

        # ---------- Signal 3: dollar cost figures ----------
        # \$\d matches '$' followed immediately by a digit ($30, $80, $250, etc.).
        # This catches both per-activity costs ('Hagia Sophia - $30') AND prices in budget
        # summaries ('$80-120/night'). For this tutorial that's fine — verbose v1 outputs
        # tend to include budget tables that pass this check, so signal 3 alone isn't the
        # discriminator across our v1 vs v2 comparison; signal 2 above usually is.
        cost_markers = len(re.findall(r"\$\d", output))
        has_costs = cost_markers >= n_days

        # Combined score: average the three signals so each contributes equally. A perfect
        # output scores 1.0; missing either time stamps or cost figures caps the score at
        # 0.67 (since days_coverage of 1.0 + one zero + one one = 2.0/3).
        score = (days_coverage + (1.0 if has_times else 0.0) + (1.0 if has_costs else 0.0)) / 3

        if score >= 0.9:
            label = "well_structured"
        elif score >= 0.5:
            label = "partial"
        else:
            label = "poor"

        # The explanation is what shows up next to the score in the AX UI — write it so
        # someone reading a low score can immediately see which signal failed and why.
        return EvaluationResult(
            label=label,
            score=round(float(score), 2),
            explanation=(
                f"days:{days_present}/{n_days} | "
                f"times:{time_markers} ({'pass' if has_times else 'fail'}) | "
                f"costs:{cost_markers} ({'pass' if has_costs else 'fail'})"
            ),
        )
    except Exception as e:
        # Catch-all so a buggy evaluator doesn't sink the whole experiment. The
        # experiment record will show label=error with the exception text.
        return EvaluationResult(label="error", score=0.0, explanation=f"Eval crashed: {e}")

experiment_v1, results_v1 = client.experiments.run(
    name=f"trip-planner-v1-{RUN_ID}",  # unique experiment name; shows up in the UI under the dataset
    dataset=dataset.name,
    space=ARIZE_SPACE_ID,
    task=task_v1,
    evaluators={"itinerary_structure": itinerary_structure_eval},  # dict key becomes the eval column prefix
    concurrency=3,                                                  # how many task calls to run in parallel
    exit_on_error=True,                                             # surface task/evaluator errors loudly instead of silently scoring None
)

print(f"Columns: {list(results_v1.columns)}")
print()
print("Per-row scores (v1):")

# Find the eval columns by suffix — actual column name format may vary by SDK version,
# but they always end in .score / .label.
score_col = next((c for c in results_v1.columns if c.endswith("score") and "itinerary" in c), None)
label_col = next((c for c in results_v1.columns if c.endswith("label") and "itinerary" in c), None)
if score_col and label_col:
    print(results_v1[[score_col, label_col]].to_string())
    print(f"\nMean score: {results_v1[score_col].mean():.2f}")
else:
    # Defensive fallback if the column names shift — print enough to debug.
    print(results_v1.head().to_string())

# v2 system message: tightened to enforce the exact format our evaluator scores against.
# Key changes vs v1:
#   - Names the exact output shape: "Day N: HH:MM - Activity - $cost"
#   - Uses "MUST" / "ALWAYS" — concrete, unambiguous instructions outperform softer wording
#   - Tells the model what NOT to do ("no preamble or epilogue") to suppress the chatty intros v1 produced
system_message_v2 = (
    "You are a disciplined trip planner. For every day of the trip you MUST output:\n"
    "  Day N: HH:MM - Activity - $cost\n"
    "Use one bullet per time slot. ALWAYS include a $cost figure (estimate if exact "
    "price unknown). Cover the full duration. Be concise; no preamble or epilogue."
)

# create_version vs create: we pass prompt=prompt_v1.id so this becomes a new version of
# the SAME prompt object — same name, same history, just a new immutable snapshot.
prompt_v2 = client.prompts.create_version(
    prompt=prompt_v1.id,
    space=ARIZE_SPACE_ID,
    commit_message="v2: enforce strict day/time/cost format, drop preamble",
    input_variable_format=InputVariableFormat.F_STRING,
    provider=LlmProvider.OPEN_AI,
    model="gpt-5.4-mini",
    messages=[
        LLMMessage(role=MessageRole.SYSTEM, content=system_message_v2),
        LLMMessage(role=MessageRole.USER, content=user_message),  # user template is unchanged
    ],
    # Lower temperature for v2: when a prompt has strict format rules, you want less
    # creative drift. 0.5 keeps outputs varied enough to feel natural but reduces the
    # chance of the model going off-format.
    invocation_params=InvocationParams(temperature=0.5, max_completion_tokens=600),
)

print(f"Saved v2 (version id: {prompt_v2.id})")

# Tag v2 as production. Tags are mutable pointers at versions — when an application calls
# client.prompts.get(prompt=PROMPT_NAME, label="production"), it gets whatever version
# this tag currently points at. Moving the tag is the deployment.
#
# v1 is still in Prompt Hub and recoverable — it just no longer carries the production
# label. A future v3 can take it next.
client.prompts.set_labels(version_id=prompt_v2.id, labels=["production"])
print("v2 tagged as production.")

# Build the v2 task from the new system message. Everything else (user template, model,
# evaluator, dataset) is identical — that's exactly what makes the experiment comparable.
task_v2 = make_task(system_message_v2, user_message)

experiment_v2, results_v2 = client.experiments.run(
    name=f"trip-planner-v2-{RUN_ID}",
    dataset=dataset.name,       # SAME dataset as v1 — required for apples-to-apples comparison
    space=ARIZE_SPACE_ID,
    task=task_v2,
    evaluators={"itinerary_structure": itinerary_structure_eval},  # SAME evaluator as v1
    concurrency=3,
    exit_on_error=True,
)

print("Per-row scores (v2):")
score_col_v2 = next((c for c in results_v2.columns if c.endswith("score") and "itinerary" in c), None)
label_col_v2 = next((c for c in results_v2.columns if c.endswith("label") and "itinerary" in c), None)
if score_col_v2 and label_col_v2:
    print(results_v2[[score_col_v2, label_col_v2]].to_string())
    print(f"\nMean score: {results_v2[score_col_v2].mean():.2f}")
else:
    print(results_v2.head().to_string())

# Build a side-by-side comparison frame. Because v1 and v2 ran on the SAME dataset rows
# in the SAME order, we can compare element-wise — row 0 of v1 corresponds to row 0 of v2.
score_col = next(c for c in results_v1.columns if c.endswith("score") and "itinerary" in c)
delta = pd.DataFrame({
    "v1_score": results_v1[score_col].values,
    "v2_score": results_v2[score_col].values,
})

# Positive delta = v2 improved on that row. Negative delta = v2 regressed on that row.
# In a real iteration cycle you want most rows positive AND no row strongly negative —
# regressing on a single edge case is how prompt edits ship silent bugs.
delta["delta"] = delta["v2_score"] - delta["v1_score"]
print(delta.to_string())
print(f"\nMean v1: {delta['v1_score'].mean():.2f}")
print(f"Mean v2: {delta['v2_score'].mean():.2f}")