Literal AI Testing Guide: Thread Tracking, Datasets, and Scoring

pip install literalai

from literalai import LiteralClient

lai = LiteralClient(api_key="your-api-key")

from literalai import LiteralClient
import openai

lai = LiteralClient(api_key="your-api-key")
oai = openai.OpenAI()

def run_support_chat(conversation_history: list[dict], user_message: str) -> str:
    """Run one turn of the support chatbot."""
    
    # Add the new user message to history
    conversation_history.append({
        "role": "user",
        "content": user_message
    })
    
    response = oai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a helpful customer support agent."},
            *conversation_history
        ]
    )
    
    assistant_message = response.choices[0].message.content
    conversation_history.append({
        "role": "assistant",
        "content": assistant_message
    })
    
    return assistant_message


def run_support_session(user_id: str, initial_message: str):
    """Run a full support session, tracked as a Literal AI thread."""
    
    with lai.thread(name=f"Support Session - {user_id}", 
                    metadata={"user_id": user_id, "channel": "web"}) as thread:
        
        conversation_history = []
        current_message = initial_message
        
        # Track the first user message as a step
        with lai.step(name="user-message", type="user_message") as step:
            step.input = {"message": current_message}
        
        # Run the conversation loop (in a real app, this would be event-driven)
        for turn in range(5):  # Max 5 turns
            with lai.step(name=f"turn-{turn}", type="llm") as step:
                step.input = {
                    "history": conversation_history,
                    "user_message": current_message
                }
                
                response = run_support_chat(conversation_history, current_message)
                
                step.output = {"response": response}
                step.metadata = {
                    "turn": turn,
                    "history_length": len(conversation_history)
                }
            
            print(f"Bot: {response}")
            
            # Simulate user follow-up (in real app, this is the next event)
            if turn == 0:
                current_message = "It's too expensive"
            elif turn == 1:
                current_message = "Sure, check my eligibility"
            else:
                break
        
        return thread.id

from literalai import LiteralClient

lai = LiteralClient(api_key="your-api-key")
oai = lai.instrumented_openai(openai.OpenAI())

# All calls through oai are automatically tracked as steps
# within whatever thread is currently active
with lai.thread(name="automated-support") as thread:
    response = oai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello, I need help"}]
    )

from literalai import LiteralClient

lai = LiteralClient(api_key="your-api-key")

# Create a dataset for conversational test cases
dataset = lai.api.create_dataset(
    name="support-chatbot-v1",
    description="Regression suite for the customer support chatbot",
    type="key_value"
)

# Add items to the dataset
items = [
    {
        "input": {
            "conversation": [
                {"role": "user", "content": "I want to cancel"},
                {"role": "assistant", "content": "I'm sorry to hear that. What's the reason?"},
                {"role": "user", "content": "Too expensive"},
                {"role": "assistant", "content": "Would you like to explore our more affordable plan?"},
                {"role": "user", "content": "Yes"}
            ]
        },
        "expected_output": {
            "contains": ["promotional", "plan", "discount"],
            "should_not_contain": ["cancel", "sorry to see you go"],
            "action": "check_promotion_eligibility"
        },
        "metadata": {
            "category": "retention",
            "source": "production-failure",
            "ticket": "TKT-8823"
        }
    },
    {
        "input": {
            "conversation": [
                {"role": "user", "content": "What's your refund policy?"}
            ]
        },
        "expected_output": {
            "contains": ["30 days", "full refund"],
            "action": "provide_policy_info"
        },
        "metadata": {
            "category": "policy",
            "source": "manual",
            "difficulty": "easy"
        }
    }
]

for item in items:
    lai.api.create_dataset_item(
        dataset_id=dataset.id,
        input=item["input"],
        expected_output=item["expected_output"],
        metadata=item["metadata"]
    )

print(f"Dataset created: {dataset.id}")

def add_human_score(step_id: str, score_value: float, comment: str):
    """Add a human evaluation score to a specific step."""
    lai.api.create_score(
        step_id=step_id,
        name="human-quality",
        type="HUMAN",
        value=score_value,  # 0 to 1
        comment=comment
    )

# After reviewing a conversation thread:
add_human_score(
    step_id="step-abc123",
    score_value=0.9,
    comment="Response was helpful and on-topic, minor phrasing issue"
)

import openai
from literalai import LiteralClient

lai = LiteralClient(api_key="your-api-key")
oai = openai.OpenAI()

def score_retention_handling(
    conversation: list[dict], 
    response: str,
    step_id: str
) -> float:
    """Use GPT-4 to score how well the bot handled a retention scenario."""
    
    conversation_text = "\n".join([
        f"{msg['role'].upper()}: {msg['content']}" 
        for msg in conversation
    ])
    
    judge_prompt = f"""You are evaluating a customer support chatbot's retention handling.

Conversation history:
{conversation_text}

Bot's latest response: {response}

Rate the bot's retention handling on a scale of 0 to 1:
- 1.0: Excellently handled retention — showed empathy, offered alternatives, didn't push cancellation
- 0.7: Good handling — addressed the concern but missed an opportunity
- 0.4: Mediocre — acknowledged but didn't actively try to retain
- 0.1: Poor — facilitated cancellation without attempting retention

Return only a decimal number between 0 and 1."""
    
    judge_response = oai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": judge_prompt}]
    )
    
    score_value = float(judge_response.choices[0].message.content.strip())
    
    # Record the score in Literal AI
    lai.api.create_score(
        step_id=step_id,
        name="retention-handling",
        type="AI",
        value=score_value,
        comment=f"AI judge scored retention handling at {score_value:.2f}"
    )
    
    return score_value

def run_experiment(dataset_id: str, experiment_name: str):
    """Run the chatbot against the dataset and score all outputs."""
    
    items = lai.api.get_dataset_items(dataset_id=dataset_id)
    scores = []
    
    for item in items:
        conversation = item.input["conversation"]
        expected = item.expected_output
        
        # Run the chatbot on the conversation
        with lai.thread(name=f"eval-{experiment_name}") as thread:
            with lai.step(name="eval-turn", type="llm") as step:
                step.input = {"conversation": conversation}
                
                # Get the bot's response to the conversation
                response = run_support_chat(conversation, conversation[-1]["content"])
                step.output = {"response": response}
                
                # Score the response
                score = evaluate_response(response, expected, step_id=step.id)
                scores.append(score)
    
    avg_score = sum(scores) / len(scores) if scores else 0
    print(f"Experiment '{experiment_name}': avg score = {avg_score:.3f}")
    return avg_score

def evaluate_response(response: str, expected: dict, step_id: str) -> float:
    """Multi-factor scoring of a chatbot response."""
    
    scores = []
    
    # Check required content
    if "contains" in expected:
        hits = sum(1 for kw in expected["contains"] if kw.lower() in response.lower())
        content_score = hits / len(expected["contains"])
        scores.append(content_score)
        
        lai.api.create_score(
            step_id=step_id,
            name="required-content",
            type="AI",
            value=content_score,
            comment=f"Found {hits}/{len(expected['contains'])} required keywords"
        )
    
    # Check prohibited content
    if "should_not_contain" in expected:
        violations = [
            kw for kw in expected["should_not_contain"] 
            if kw.lower() in response.lower()
        ]
        safety_score = 1.0 if not violations else 0.0
        scores.append(safety_score)
        
        lai.api.create_score(
            step_id=step_id,
            name="content-safety",
            type="AI",
            value=safety_score,
            comment=f"Violations: {violations}" if violations else "No violations"
        )
    
    return sum(scores) / len(scores) if scores else 0.0

import pytest
from literalai import LiteralClient

lai = LiteralClient(api_key="your-api-key")

def test_context_retention_across_turns():
    """Verify the bot correctly uses context from earlier in the conversation."""
    
    # Conversation where turn 3 requires knowing what was said in turn 1
    conversation_progression = [
        "My account number is 12345",                          # Turn 1: provides account
        "What's the weather like?",                            # Turn 2: off-topic
        "What account number did I just tell you?",            # Turn 3: tests memory
    ]
    
    conversation_history = []
    responses = []
    
    with lai.thread(name="context-retention-test") as thread:
        for message in conversation_progression:
            with lai.step(name="turn") as step:
                step.input = {"message": message, "history": conversation_history}
                response = run_support_chat(conversation_history, message)
                step.output = {"response": response}
                responses.append(response)
    
    # The bot should recall "12345" in response to the third message
    final_response = responses[-1]
    assert "12345" in final_response, (
        f"Bot failed to recall account number from turn 1. "
        f"Got: '{final_response}'. "
        f"Thread: https://cloud.literal.ai/threads/{thread.id}"
    )

def test_intent_continuity():
    """Verify the bot maintains the intent context across topic switches."""
    
    with lai.thread(name="intent-continuity-test") as thread:
        # Establish a refund request
        with lai.step(name="turn-1") as step:
            h1 = []
            r1 = run_support_chat(h1, "I need a refund for my purchase from last week")
            step.output = {"response": r1}
        
        # Slightly off-topic
        with lai.step(name="turn-2") as step:
            r2 = run_support_chat(h1, "Actually, what IS your refund policy exactly?")
            step.output = {"response": r2}
        
        # Return to the original request
        with lai.step(name="turn-3") as step:
            r3 = run_support_chat(h1, "OK, so can I get my refund?")
            step.output = {"response": r3}
        
        # Turn 3 should still be about the refund, not asking for details again
        refund_keywords = ["refund", "process", "initiate", "submit"]
        has_refund_content = any(kw in r3.lower() for kw in refund_keywords)
        
        assert has_refund_content, (
            f"Bot lost intent context. Turn 3 response: '{r3}'. "
            f"Thread: https://cloud.literal.ai/threads/{thread.id}"
        )

# .github/workflows/conversational-ai-tests.yml
name: Conversational AI Tests

on:
  pull_request:
    paths:
      - 'src/chatbot/**'
      - 'prompts/**'

jobs:
  dataset-eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Install dependencies
        run: pip install literalai openai pytest
      
      - name: Run conversation tests
        run: pytest tests/conversation/ -v
        env:
          LITERAL_API_KEY: ${{ secrets.LITERAL_API_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      
      - name: Run dataset evaluation
        run: |
          python scripts/run_literal_eval.py \
            --dataset support-chatbot-v1 \
            --experiment ${{ github.sha }} \
            --min-score 0.80
        env:
          LITERAL_API_KEY: ${{ secrets.LITERAL_API_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}