AI Guardrails: Coherence scorers

import asyncio
import weave; weave.init("coherence-scorer")
from weave.scorers import CoherenceScorer

async def main():
    # Initialize the CoherenceScorer
    coherence_scorer = CoherenceScorer(
        model_name_or_path="wandb/coherence_scorer",  # Replace with your model path if local
        device="auto"  # Uses CUDA if available
    )
    
    # Input and output examples
    input_text = "a query testing the model?"
    output_text = "a response from the model"

    # Evaluate coherence
    result = await coherence_scorer.score(input=input_text, output=output_text)

    # Print the results
    print("Coherence Scoring Result:")
    print(f"Flagged as incoherent: {result['flagged']}")
    print(f"Coherence Label: {result['extras']['coherence_label']}")
    print(f"Coherence Score: {result['extras']['coherence_score']}")
    print(f"Coherence ID: {result['extras']['coherence_id']}")

# Run the async main function
if __name__ == "__main__":
    asyncio.run(main())


import time
import asyncio
from litellm import acompletion
from pydantic import BaseModel, Field
from typing import Literal, Any

import nest_asyncio

import weave; weave.init("coherence-scorer")

# Define prompts
COHERENCE_SYSTEM_PROMPT = """Given some <prompt> from a user and an <response> generated by an AI system, I am running a few minutes late; my previous meeting is running over.
determine if the <response> is coherent or not.

Coherence of the <response> is defined as:
- The <response> is self consistent in terms of content, style of writing, and does not contradict itself.
- The <response> can be logically followed and understood by a human. 
- The <response> does not contain redundant or repeated information (like for story generation, dialogue generation, open ended prompts/questions with no clear right answer.)

# Steps
1. Carefully read and understand the <prompt>.
2. Examine the model <response>.
3. Compare the <response> to the <prompt>, identifying any inconsistencies or additions.
4. Measure how lucid, cogent, and self-consistent the model’s <response> is. 

# Guidelines
- Focus on coherence and clarity of the <response>
- Consider both explicit and implicit information in the <prompt>
- Identify degree to which the <response> is clear, easy to understand and maintains a proper logical flow.

# Scoring
Score the coherence of the <response> on a likert scale of 0 to 4:
- 4 (Perfectly Coherent and Clear): The response is perfectly clear and self-consistent throughout. There are no contradictory assertions or statements, the writing flows logically and following the train of thought/story is not challenging.
- 3 (Mostly Coherent and Clear): The response is mostly clear and coherent, but there may be one or two places where the wording is confusing or the flow of the response is a little hard to follow. Over all, the response can mostly be followed with a little room for improvement.
- 2 (A Little Unclear and/or Incoherent): The response is a little unclear. There are some inconsistencies or contradictions, run on sentences, confusing statements, or hard to follow sections of the response.
- 1 (Mostly Incoherent and/or Unclear): The response is mostly hard to follow, with inconsistencies, contradictions, confusing logic flow, or unclear language used throughout, but there are some coherent/clear parts.
- 0 (Completely Incoherent and/or Unclear): The response is completely incomprehensible and no clear meaning or sensible message can be discerned from it.
"""

COHERENCE_USER_PROMPT = """Analyze the following <prompt> and <response> and determine if the <response> is coherent or not.
<prompt>
{input}
</prompt>
<response>
{output}
</response>
"""


# Define the CoherenceClassification model
class CoherenceClassification(BaseModel):
    chain_of_thought: str = Field(..., description="The chain of thought that led to the prediction")
    coherence_score: int = Field(..., description="Score the coherence of the <response> on a likert scale of 0 to 4")
    coherence: Literal[
        "Perfectly Coherent", "Mostly Coherent", "A Little Incoherent", "Mostly Incoherent", "Completely Incoherent"
    ] = Field(..., description="The level of coherence of the <response>")
    coherent: bool = Field(..., description="Whether the <response> is coherent or not, anything above 2 is coherent")
    confidence: float = Field(..., description="The confidence of the prediction", ge=0.0, le=1.0)


# Define the scorer class using LiteLLM
class CoherenceScorer:
    def __init__(self, model_name="gpt-4o-2024-08-06", api_key="your api key", temperature=0.99, max_tokens=2048, top_p=1.0):
        self.model_name = model_name
        self.api_key = api_key
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.top_p = top_p

    async def score(self, input_text: str, output_text: str) -> dict[str, Any]:
        formatted_user_prompt = COHERENCE_USER_PROMPT.format(input=input_text, output=output_text)
        response = await acompletion(
            model=self.model_name,
            api_key=self.api_key,
            messages=[
                {"role": "system", "content": COHERENCE_SYSTEM_PROMPT},
                {"role": "user", "content": formatted_user_prompt},
            ],
            temperature=self.temperature,
            max_tokens=self.max_tokens,
            top_p=self.top_p,
        )
        chain_of_thought = response["choices"][0]["message"]["content"]
        # Parse coherence score from the response text, assuming it outputs coherence score as a structured result
        # Replace this with actual parsing logic if required
        coherence_score = int(chain_of_thought.split("Score:")[1].strip()[0])
        coherence_label = ["Completely Incoherent", "Mostly Incoherent", "A Little Incoherent", "Mostly Coherent", "Perfectly Coherent"][coherence_score]
        confidence = 0.9  # Placeholder confidence, refine based on your model's output

        return CoherenceClassification(
            chain_of_thought=chain_of_thought,
            coherence_score=coherence_score,
            coherence=coherence_label,
            coherent=coherence_score >= 2,
            confidence=confidence,
        ).dict()


# Example usage
async def main():
    scorer = CoherenceScorer(api_key="your api key", model_name="gpt-4o")
    input_text = "What is the capital of France?"
    output_text = "The capital of France is Paris."
    result = await scorer.score(input_text, output_text)
    print(result)


# Run the example
nest_asyncio.apply()  # Allows nested event loops for environments like Jupyter
asyncio.run(main())

import weave
from weave.scorers import CoherenceScorer
import pandas as pd
from datasets import load_dataset
import asyncio
from weave.trace.box import unbox



import time
import asyncio
from litellm import acompletion
from pydantic import BaseModel, Field
from typing import Literal, Any

import nest_asyncio



# Define prompts
COHERENCE_SYSTEM_PROMPT = """Given some <prompt> from a user and an <response> generated by an AI system, \
determine if the <response> is coherent or not.

Coherence of the <response> is defined as:
- The <response> is self consistent in terms of content, style of writing, and does not contradict itself.
- The <response> can be logically followed and understood by a human. 
- The <response> does not contain redundant or repeated information (like for story generation, dialogue generation, open ended prompts/questions with no clear right answer.)

# Steps
1. Carefully read and understand the <prompt>.
2. Examine the model <response>.
3. Compare the <response> to the <prompt>, identifying any inconsistencies or additions.
4. Measure how lucid, cogent, and self-consistent the model’s <response> is. 

# Guidelines
- Focus on coherence and clarity of the <response>
- Consider both explicit and implicit information in the <prompt>
- Identify degree to which the <response> is clear, easy to understand and maintains a proper logical flow.

# Scoring
Score the coherence of the <response> on a likert scale of 0 to 4:
- 4 (Perfectly Coherent and Clear): The response is perfectly clear and self-consistent throughout. There are no contradictory assertions or statements, the writing flows logically and following the train of thought/story is not challenging.
- 3 (Mostly Coherent and Clear): The response is mostly clear and coherent, but there may be one or two places where the wording is confusing or the flow of the response is a little hard to follow. Over all, the response can mostly be followed with a little room for improvement.
- 2 (A Little Unclear and/or Incoherent): The response is a little unclear. There are some inconsistencies or contradictions, run on sentences, confusing statements, or hard to follow sections of the response.
- 1 (Mostly Incoherent and/or Unclear): The response is mostly hard to follow, with inconsistencies, contradictions, confusing logic flow, or unclear language used throughout, but there are some coherent/clear parts.
- 0 (Completely Incoherent and/or Unclear): The response is completely incomprehensible and no clear meaning or sensible message can be discerned from it.
"""

COHERENCE_USER_PROMPT = """Analyze the following <prompt> and <response> and determine if the <response> is coherent or not.
<prompt>
{input}
</prompt>
<response>
{output}
</response>
"""


# Define the CoherenceClassification model
class CoherenceClassification(BaseModel):
    chain_of_thought: str = Field(..., description="The chain of thought that led to the prediction")
    coherence_score: int = Field(..., description="Score the coherence of the <response> on a likert scale of 0 to 4")
    coherence: Literal[
        "Perfectly Coherent", "Mostly Coherent", "A Little Incoherent", "Mostly Incoherent", "Completely Incoherent"
    ] = Field(..., description="The level of coherence of the <response>")
    coherent: bool = Field(..., description="Whether the <response> is coherent or not, anything above 2 is coherent")
    confidence: float = Field(..., description="The confidence of the prediction", ge=0.0, le=1.0)


# Define the scorer class using LiteLLM
class GPTCoherenceScorer:
    def __init__(self, model_name="gpt-4o-2024-08-06", api_key="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", temperature=0.0, max_tokens=2048, top_p=1.0):
        self.model_name = model_name
        self.api_key = api_key
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.top_p = top_p

    async def score(self, input_text: str, output_text: str) -> dict[str, Any]:
        formatted_user_prompt = COHERENCE_USER_PROMPT.format(input=input_text, output=output_text)
        response = await acompletion(
            model=self.model_name,
            api_key=self.api_key,
            messages=[
                {"role": "system", "content": COHERENCE_SYSTEM_PROMPT},
                {"role": "user", "content": formatted_user_prompt},
            ],
            temperature=self.temperature,
            max_tokens=self.max_tokens,
            top_p=self.top_p,
        )
        chain_of_thought = response["choices"][0]["message"]["content"]
        # Parse coherence score from the response text, assuming it outputs coherence score as a structured result
        # Replace this with actual parsing logic if required
        # coherence_score = int(chain_of_thought.split("Score:")[1].strip()[0])
        try:
            if "Score:" in chain_of_thought:
                coherence_score = int(chain_of_thought.split("Score:")[1].strip().split()[0])
            else:
                coherence_score = 0  # 
        except (IndexError, ValueError) as e:
            print(f"Error parsing coherence score: {e}")
            coherence_score = 0  # Default to 0 or any fallback score you prefer

        coherence_label = ["Completely Incoherent", "Mostly Incoherent", "A Little Incoherent", "Mostly Coherent", "Perfectly Coherent"][coherence_score]
        confidence = 0.9  # Placeholder confidence, refine based on your model's output

        return CoherenceClassification(
            chain_of_thought=chain_of_thought,
            coherence_score=coherence_score,
            coherence=coherence_label,
            coherent=coherence_score >= 2,
            confidence=confidence,
        ).dict()["coherence_score"]



gpt4oscorer = GPTCoherenceScorer(model_name="gpt-4o-2024-08-06", api_key="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")
# Initialize Weave
gpt4ominiscorer = GPTCoherenceScorer(model_name="gpt-4o-mini", api_key="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")


weave.init("coherence_eval")
scorer = CoherenceScorer()


# Load and prepare the dataset
def load_coherence_dataset():
    # Load the HelpSteer2 dataset
    dataset = load_dataset("nvidia/HelpSteer2", split="validation")
    
    # Convert to a Pandas DataFrame for easier manipulation
    df = pd.DataFrame(dataset)
    
    balanced_samples = df.groupby('coherence').apply(lambda x: x.sample(min(len(x), 20)))
    
    # Reset the index after grouping and sampling
    balanced_samples = balanced_samples.reset_index(drop=True)

    # Prepare dataset for evaluation
    dataset_prepared = [
        {"output": row["response"], "label": row["coherence"], "prompt": row["prompt"]}
        for _, row in balanced_samples.iterrows()
    ]
    
    return dataset_prepared


# Define Weave Coherence Scorer Model
class WeaveCoherenceScorerModel(weave.Model):

    @weave.op
    async def predict(self, prompt: str, output: str) -> int:
        """Predict coherence scores."""
        time.sleep(2)
        
        result = await scorer.score(input=prompt, output=output)
        coherence_id = result.get("extras", {}).get("coherence_id", 0)
        return coherence_id

@weave.op
def coherence_scorer_close_match(label: int, model_output: int) -> dict:
    """
    Score for close matches: considers the prediction correct if it is within 1 class of the true label.
    Returns 1 if the prediction is considered close enough, otherwise 0.
    """
    is_close_match = abs(label - model_output) <= 1
    return {"close_match": int(is_close_match)}

# Define the Weave model for 4o coherence scoring
class GPT4oCoherenceModel(weave.Model):
    
    @weave.op
    async def predict(self, prompt: str, output: str) -> dict:
        """
        Use the 4o coherence scorer to predict coherence and return results.
        """
        
        result = await gpt4oscorer.score(prompt, output)
        return result

class GPT4oMiniCoherenceModel(weave.Model):
    
    @weave.op
    async def predict(self, prompt: str, output: str) -> dict:
        """
        Use the 4o coherence scorer to predict coherence and return results.
        """
        
        result = await gpt4ominiscorer.score(prompt, output)
        return result


# Define the evaluation scorer for coherence
@weave.op
def coherence_scorer_exact_match(label: int, model_output: int) -> dict:
    """Score the coherence prediction."""
    return {"coherence_accuracy": int(model_output == label)}


# Define the evaluation scorer for coherence
@weave.op
def coherence_scorer_error(label: int, model_output: int) -> dict:
    if isinstance(model_output, weave.trace.box.BoxedStr):
        model_output = int(unbox(model_output))

    """Score the coherence prediction."""
    return {"coherence_error": abs(int(model_output - label))}

@weave.op
def coherence_scorer_false_positive(label: int, model_output: int) -> dict:
    """
    Score for false positives: model predicts coherent (3 or 4) when the true label is incoherent (0 or 1).
    Returns 1 if it is a false positive, otherwise 0.
    """
    is_false_positive = (label in [0, 1] and model_output in [3, 4])
    return {"false_positive": int(is_false_positive)}

@weave.op
def coherence_scorer_false_negative(label: int, model_output: int) -> dict:
    """
    Score for false negatives: model predicts incoherent (0 or 1) when the true label is coherent (3 or 4).
    Returns 1 if it is a false negative, otherwise 0.
    """
    is_false_negative = (label in [3, 4] and model_output in [0, 1])
    return {"false_negative": int(is_false_negative)}


# Run the evaluations
async def run_evaluations():
    """Run evaluations for the coherence scorers."""
    # Load dataset
    dataset = load_coherence_dataset()
    print("Dataset loaded...")

    # Initialize models
    models = {
        "GPT4oCoherenceScorer": GPT4oCoherenceModel(),
        "WeaveCoherenceScorer": WeaveCoherenceScorerModel(),
        "GPT4oMiniCoherenceScorer": GPT4oMiniCoherenceModel(),
    }

    # Define evaluation scorers
    scorers = [
        coherence_scorer_exact_match,
        coherence_scorer_error,
        coherence_scorer_false_positive,
        coherence_scorer_false_negative,
        coherence_scorer_close_match
    ]

    # Run evaluations
    results = {}
    for model_name, model in models.items():
        print(f"\nEvaluating {model_name}...")
        evaluation = weave.Evaluation(
            dataset=dataset,
            scorers=scorers,
            name=model_name + " Eval"
        )
        results[model_name] = await evaluation.evaluate(model)

    # Print results
    for model_name, result in results.items():
        print(f"\nResults for {model_name}:")
        print(result)


if __name__ == "__main__":
    asyncio.run(run_evaluations())