DSPy와 W&B Weave로 RAG 시스템 구축 및 평가

class GenerateAnswer(dspy.Signature):
    """Signature for generating answers based on retrieved context."""
    context = dspy.InputField(desc="Relevant facts")  # Input field for facts
    question = dspy.InputField()  # Input field for user questions
    answer = dspy.OutputField(desc="Short fact-based answer")  # Output field for the generated answer

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)  # Retrieve relevant content
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)  # Use the Chain-of-Thought method

    def forward(self, question: str):
        context = self.retrieve(question).passages  # Fetch top-k relevant passages
        prediction = self.generate_answer(context=context, question=question)  # Generate answer
        return dspy.Prediction(context=context, answer=prediction.answer)

import os
import re
import glob
import weaviate
import weaviate.classes.config as wvcc

# Function to download the Weave documentation .md files
def download_weave_docs():
    repo_dir = "weave_docs"
    if not os.path.exists(repo_dir):
        print("Downloading Weave documentation...")
        os.system(f"git clone --depth 1 --filter=blob:none --sparse https://github.com/wandb/weave.git {repo_dir}")
        os.chdir(repo_dir)
        os.system("git sparse-checkout set docs/docs/guides/tracking")
        os.chdir("..")
        print("Weave documentation downloaded successfully.")
    else:
        print(f"{repo_dir} already exists, skipping download.")

# Function to read and process the .md files
def process_md_files():
    md_files = glob.glob('weave_docs/docs/docs/guides/tracking/*.md')
    all_chunks = []

    for md_file in md_files:
        with open(md_file, 'r', encoding='utf-8') as f:
            content = f.read()
            # Clean the content if necessary
            content = clean_markdown(content)
            # Split content into sentences or paragraphs
            chunks = split_into_chunks(content, chunk_size=200)
            all_chunks.extend(chunks)
    return all_chunks

# Function to clean markdown content
def clean_markdown(text):
    """Remove unnecessary elements but retain code, symbols, and structure."""
    text = re.sub(r'!\[.*?\]\(.*?\)', '', text)  # Remove images
    text = re.sub(r'\[([^\]]+)\]\((.*?)\)', r'\1', text)  # Keep link text but remove URLs
    text = text.strip()
    return text

# Function to split text into chunks of specified character length
def split_into_chunks(text, chunk_size=200):
    words = text.split()
    chunks = []
    current_chunk = []
    current_length = 0

    for word in words:
        current_chunk.append(word)
        current_length += len(word) + 1  # +1 for space
        if current_length >= chunk_size:
            chunks.append(' '.join(current_chunk))
            current_chunk = []
            current_length = 0

    if current_chunk:
        chunks.append(' '.join(current_chunk))

    return chunks

# Main execution
if __name__ == "__main__":
    # Step 1: Download Weave documentation
    download_weave_docs()

    # Step 2: Process the .md files
    print("Processing .md files...")
    content_chunks = process_md_files()
    print(f"Total content chunks: {len(content_chunks)}")


    # Step 3: Connect to Weaviate instance
    client = weaviate.connect_to_wcs(
        cluster_url="your weaviate cluster url ",  # Replace with your WCS URL
        auth_credentials=weaviate.auth.AuthApiKey("your weaviate auth key"),  # Replace with your WCS key
        headers={
            'X-Cohere-Api-Key': "your cohere api key"  # Replace with your Cohere API key
        }
    )

    try:
        # CAUTION: Running this will delete the collection along with the objects
        # Delete the collection if it exists
        collection_name = "WeaveDocsChunk"
        if client.collections.exists(collection_name):
            client.collections.delete(collection_name)
            print(f"Existing {collection_name} collection deleted.")

        # Create the collection with the specified vectorizer configuration and properties
        collection = client.collections.create(
            name=collection_name,
            vectorizer_config=wvcc.Configure.Vectorizer.text2vec_cohere(
                model="embed-multilingual-v3.0"
            ),
            properties=[
                wvcc.Property(name="content", data_type=wvcc.DataType.TEXT),
                wvcc.Property(name="source", data_type=wvcc.DataType.TEXT),
            ]
        )
        print(f"{collection_name} collection created successfully.")

        # Import Objects
        collection_ref = client.collections.get(collection_name)

        for idx, content_chunk in enumerate(content_chunks):
            collection_ref.data.insert(
                properties={
                    "content": content_chunk,
                    "source": "WeaveDocs"
                }
            )
            print(f"Uploaded chunk {idx + 1}/{len(content_chunks)}")

        print("Content chunks successfully imported into Weaviate!")

    finally:
        # Close the client connection
        client.close()
        print("Client connection closed.")

import weaviate
import weave; weave.init("dspy-inference")

# Connect to Weaviate cloud instance
client = weaviate.connect_to_wcs(
    cluster_url="your-cluster-url",  # Replace with your WCS URL
    auth_credentials=weaviate.auth.AuthApiKey("your-api-key"),  # Replace with your WCS API key
    headers={'X-Cohere-Api-Key': "your-api-key"}  # Replace with your Cohere API key
)

# Create a WeaviateRM client for retrieval
retriever_model = WeaviateRM(
    weaviate_collection_name="WeaveDocsChunk",
    weaviate_client=client,
    k=5
)

# Configure Language Model
llm = dspy.OpenAI(
    model='gpt-4o-mini',
    api_key="your-api-key"
)

# Configure dspy settings
dspy.settings.configure(lm=llm, rm=retriever_model)

class GenerateAnswer(dspy.Signature):
    """Answer questions with short factoid answers."""
    context = dspy.InputField(desc="Relevant facts")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="Short answer")

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question: str):
        # Retrieve relevant passages
        context = self.retrieve(question).passages
        # Generate an answer using the retrieved context
        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(context=context, answer=prediction.answer)

# Step 1: Import necessary libraries and modules
import dspy
from dspy.teleprompt import BootstrapFewShot
from dspy.evaluate.evaluate import Evaluate
from sklearn.model_selection import train_test_split
import pandas as pd
import random
import weave
import weaviate
from dspy.retrieve.weaviate_rm import WeaviateRM
from dspy.primitives.example import Example

# Import Ragas metric for factual correctness
from ragas.llms import LangchainLLMWrapper
from ragas.dataset_schema import SingleTurnSample
from ragas.metrics._factual_correctness import FactualCorrectness
from langchain_openai import ChatOpenAI
import asyncio
import nest_asyncio

# Apply nest_asyncio to avoid event loop conflicts in Weave
nest_asyncio.apply()


weave.init(project_name="dspy")
SEED = 42
random.seed(SEED)

dataset_path = './generated_testset.csv'

# Step 2: Configure the Language Model (LM) and Retrieval Model (RM)
llm = dspy.OpenAI(model='gpt-4o-mini', api_key="sk-")

client = weaviate.connect_to_wcs(
    cluster_url="",  # Replace with your WCS URL
    auth_credentials=weaviate.auth.AuthApiKey(""),  # Replace with your WCS key
    headers={
        'X-Cohere-Api-Key': ""  # Replace with your Cohere API key
    }
)

retriever_model = WeaviateRM(
    weaviate_collection_name="WeaviateBlogChunk",  # Use 'class_name' instead of 'collection_name'
    weaviate_client=client,
    k=5  # Number of top results to retrieve
)

dspy.settings.configure(lm=llm, rm=retriever_model)

# Initialize the evaluator LLM using Langchain and OpenAI GPT-4 model
evaluator_llm = LangchainLLMWrapper(
    ChatOpenAI(model="gpt-4o-mini", openai_api_key="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")
)

# Step 3: Load and Format Your Custom Dataset
df = pd.read_csv("/Users/brettyoung/Desktop/dev_24/tutorials/ds_py/v2/generated_testset.csv")

def format_dataset(df):
    """Format the dataset into DSPy-compatible examples with initialized inputs."""
    examples = []
    for _, row in df.iterrows():
        example = Example({
            'question': row['user_input'],
            'answer': row['reference']
        }).with_inputs('question')  # Input key as 'question'
        examples.append(example)
    return examples

# Split the dataset into training and evaluation sets
train_df, eval_df = train_test_split(df, train_size=20, test_size=10, random_state=SEED)

trainset = format_dataset(train_df)
devset = format_dataset(eval_df)

print(f"Trainset Size: {len(trainset)}, Devset Size: {len(devset)}")
print(f"First Trainset Example: {trainset[0]}")

# Step 4: Define the Signatures for the RAG Pipeline
class GenerateAnswer(dspy.Signature):
    """Answer questions with 1-3 sentence answers."""
    context = dspy.InputField(desc="may contain relevant facts")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="Answer in 1-3 sentences")

# Step 5: Build the RAG Pipeline
class RAG(dspy.Module):
    def __init__(self, num_passages=5):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        context = self.retrieve(question).passages
        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(context=context, answer=prediction.answer)

# Step 6: Define the Factual Correctness Metric
def factual_correctness_metric(example, pred, trace=None):
    """Use Ragas factual correctness metric."""
    
    # Extract response and reference correctly
    response = pred.answer  # The model-generated answer
    reference = example.answer  # The ground truth reference

    # Create the sample in the correct format
    sample = SingleTurnSample(
        response=response,
        reference=reference
    )

    # Initialize and calculate the factual correctness score
    factual_correctness_metric = FactualCorrectness(llm=evaluator_llm)

    loop = asyncio.get_event_loop()
    score = loop.run_until_complete(factual_correctness_metric.single_turn_ascore(sample=sample))

    score = factual_correctness_metric.single_turn_score(sample=sample)
    print("#"*20); print(score); print("#"*20)    
    # Return numerical score (0-1 range) for optimization
    return score 

# Step 7: Compile the RAG Program with the Custom Metric and Teleprompter
teleprompter = BootstrapFewShot(
    metric=factual_correctness_metric,  # Use the factual correctness metric
    metric_threshold=0.7  # Accept scores >= 0.5
)

compiled_rag = teleprompter.compile(RAG(), trainset=trainset)

# Optional: Save the compiled RAG program
save_path = './compiled_rag_program_v1.json'
compiled_rag.save(save_path)

print(f"RAG Program compiled and saved to {save_path}.")

import os
import asyncio
import pandas as pd
import random
import weave
from weave import Evaluation, Model

# Import your RAG pipeline components
from dspy.primitives.example import Example
import dspy
from dspy.datasets import HotPotQA
from dspy.teleprompt import BootstrapFewShot
from dspy.evaluate.evaluate import Evaluate
from sklearn.model_selection import train_test_split
import weaviate
from dspy.retrieve.weaviate_rm import WeaviateRM

# Import RAGAS metrics and related classes
from ragas.llms import LangchainLLMWrapper
from langchain.chat_models import ChatOpenAI
from ragas.dataset_schema import SingleTurnSample
from ragas.metrics import FactualCorrectness
from weave.trace.box import unbox

import nest_asyncio
import math


import weave; weave.init("ragas_eval")
# Seed for reproducibility
SEED = 42
random.seed(SEED)

# Apply nest_asyncio to avoid event loop conflicts in Weave
nest_asyncio.apply()

# Configure Language Model (LM) and Retriever Model (RM)
turbo = dspy.OpenAI(model='gpt-4o-mini', api_key="your api key")

client = weaviate.connect_to_wcs(
    cluster_url="your cluser url",
    auth_credentials=weaviate.auth.AuthApiKey("your weaviate api key"),
    headers={'X-Cohere-Api-Key': "your cohere api key"}
)

retriever_model = WeaviateRM(
    weaviate_collection_name="WeaveDocsChunk",
    weaviate_client=client,
    k=5
)

# Configure DSPy settings
dspy.settings.configure(lm=turbo, rm=retriever_model)


class GenerateAnswer(dspy.Signature):
    """Answer questions with short factoid answers."""
    context = dspy.InputField(desc="may contain relevant facts")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="Answer in 1-3 sentences")


class RAG(dspy.Module):
    def __init__(self, num_passages=5):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question: str):
        context = self.retrieve(question).passages
        if isinstance(question, weave.trace.box.BoxedStr):
            question = unbox(question)

        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(context=context, answer=prediction.answer)


class RAGModel(Model):
    model_name: str
    rag_pipeline: RAG

    @weave.op
    def predict(self, question: str):
        print(rag_pipeline)
        pred = self.rag_pipeline(question)
        return {
            "output": pred.answer,
            "input": question,
            "retrieved_contexts": pred.context
        }


def safe_score(score):
    """Helper function to return 0 if the score is NaN or None."""
    if score is None or (isinstance(score, float) and math.isnan(score)):
        return 0
    return score

@weave.op
def ragas_factual_correctness_score(expected: str, question: str, model_output: dict) -> dict:
    
    if not model_output:
        print("Model output is None.")
        return {'factual_correctness_score': 0}

    sample = SingleTurnSample(
        response=model_output.get('output', "No output."),
        reference=expected
    )

    evaluator_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini"))
    factual_correctness_metric = FactualCorrectness(llm=evaluator_llm)

    loop = asyncio.get_event_loop()
    score = loop.run_until_complete(factual_correctness_metric.single_turn_ascore(sample=sample))

    return {'factual_correctness_score': safe_score(score)}


def format_dataset(df):
    """Format the dataset into a list of dictionaries for Weave evaluation."""
    evaluation_dataset = []
    for _, row in df.iterrows():
        entry = {
            "question": row['user_input'],
            "expected": row['reference']
        }
        evaluation_dataset.append(entry)
    return evaluation_dataset

if __name__ == "__main__":
    dataset_path = './generated_testset.csv'
    df = pd.read_csv(dataset_path)
    eval_data = format_dataset(df)

    save_path = './compiled_rag_program_v1.json'
    rag_pipeline = RAG()
    rag_pipeline.load(path=save_path)
    rag_model = RAGModel(model_name='RAG Model', rag_pipeline=rag_pipeline)

    evaluation = Evaluation(dataset=eval_data, scorers=[ragas_factual_correctness_score])

    print("Starting Weave evaluation...")
    asyncio.run(evaluation.evaluate(rag_model))