LLM Playgrounds & Prompt Hubs

import chainlit as cl
import openai
 
client = openai.AsyncOpenAI()
 
@cl.on_chat_start
async def start():
    cl.user_session.set("messages", [{"role": "system", "content": "You are a helpful assistant."}])
 
@cl.on_message
async def main(message: cl.Message):
    messages = cl.user_session.get("messages")
    messages.append({"role": "user", "content": message.content})
 
    # Step 1: Simulate dynamic RAG retrieval log
    async with cl.Step(name="RAG Context Retrieval") as step:
        step.input = message.content
        # In production, query Qdrant here
        context = "Retrieved document chunk: Caching saves token cost."
        step.output = f"Injecting Context: {context}"
        messages.append({"role": "system", "content": f"Context: {context}"})
 
    # Step 2: Stream final model completion
    msg = cl.Message(content="")
    await msg.stream_token("Thinking...")
 
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        stream=True
    )
 
    msg.content = ""
    async for chunk in stream:
        token = chunk.choices[0].delta.content or ""
        await msg.stream_token(token)
 
    messages.append({"role": "assistant", "content": msg.content})
    await msg.send()

import streamlit as st
import openai
import anthropic
 
st.set_page_config(layout="wide")
st.title("Side-by-Side Model Playground")
 
# Sidebar Configuration
st.sidebar.header("Parameters")
temp = st.sidebar.slider("Temperature", 0.0, 1.0, 0.7)
prompt = st.text_area("User Prompt", "Write a short summary explaining prompt version control.")
 
col1, col2 = st.columns(2)
 
with col1:
    st.header("OpenAI GPT-4o-mini")
    if st.button("Run OpenAI"):
        client = openai.OpenAI()
        res = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=temp
        )
        st.write(res.choices[0].message.content)
 
with col2:
    st.header("Anthropic Claude 3.5 Haiku")
    if st.button("Run Anthropic"):
        client = anthropic.Anthropic()
        res = client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
            temperature=temp
        )
        st.write(res.content[0].text)

import gradio as gr
from pydantic import BaseModel, Field
import openai
import os
 
client = openai.OpenAI()
 
class ServerReport(BaseModel):
    server_name: str = Field(description="Unique node identifier")
    cpu_load: float = Field(description="CPU load percentage")
    status: str = Field(description="Healthy or Unhealthy status")
 
def extract_metrics(syslog_text: str) -> str:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract server status metrics from the raw log."},
            {"role": "user", "content": syslog_text}
        ],
        response_format=ServerReport
    )
    # Returns formatted JSON string
    return response.choices[0].message.content
 
# Interface Definition
demo = gr.Interface(
    fn=extract_metrics,
    inputs=gr.Textbox(lines=5, placeholder="Paste syslog text here..."),
    outputs=gr.JSON(label="Parsed Pydantic JSON Output")
)
 
if __name__ == "__main__":
    demo.launch()

metadata:
  name: doc_summarization_prompt
  version: 1.2.0
  created_at: "2026-06-21"
  author: "Prompt Engineering Team"
parameters:
  max_bullets: 5
templates:
  system: |
    You are an expert technical editor. Summarize the text inside the <document> tags.
    Format your output strictly as a markdown list containing at most {{ max_bullets }} bullet points.
  user: |
    <document>
    {{ document_text }}
    </document>

import yaml
from jinja2 import Template
 
class PromptRegistry:
    def __init__(self, directory_path: str):
        self.dir = directory_path
 
    def load_prompt(self, filename: str, variables: dict) -> dict:
        file_path = f"{self.dir}/{filename}"
        with open(file_path, "r") as file:
            data = yaml.safe_load(file)
        
        # Compile system and user templates using Jinja2
        system_template = Template(data["templates"]["system"])
        user_template = Template(data["templates"]["user"])
        
        # Merge default parameters with user variables
        merged_vars = {**data.get("parameters", {}), **variables}
        
        return {
            "system": system_template.render(merged_vars),
            "user": user_template.render(merged_vars),
            "version": data["metadata"]["version"]
        }
 
# Usage:
# registry = PromptRegistry(directory_path="./prompts")
# compiled = registry.load_prompt("summarize.yaml", {"document_text": "Sample raw text...", "max_bullets": 3})
# print(compiled["system"])

# OpenTelemetry Span Enrichment
from opentelemetry import trace
 
tracer = trace.get_tracer(__name__)
 
with tracer.start_as_current_span("llm_generation") as span:
    # Inject prompt tracking keys
    span.set_attribute("llm.prompt_name", "doc_summarization_prompt")
    span.set_attribute("llm.prompt_version", "1.2.0")
    span.set_attribute("llm.temperature", 0.0)