W&B Weave로 Azure AI Foundry에서 GPT 모델 비교하기

pip install openai==1.54.5 arxiv==2.1.3 PyMuPDF==1.24.9 weave==0.51.18 fitz==0.0.1.dev2 bert-score==0.3.13 rouge-score==0.1.2

import weave
import os
from openai import AzureOpenAI
import json

# Initialize Weave for logging
weave.init('azure-api')

# Initialize the AzureOpenAI client
client = AzureOpenAI(
    azure_endpoint="your enpoint url",
    api_key="your key",
    api_version="2024-09-01-preview"
)

@weave.op
def run_inference(prompt, client):
    """
    Function to perform inference using the provided client and prompt.
    """
    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        # Parse the response
        response_json = json.loads(response.model_dump_json(indent=2))
        choices = response_json.get("choices", [])
        
        if choices:
            content = choices[0].get("message", {}).get("content", "")
            print("Generated Content:")
            print(content)
            return content
        else:
            print("No content found in response")
            return None
            
    except Exception as e:
        print(f"Failed to get response: {e}")
        return None

# Define the prompt and perform inference
PROMPT = "What steps should I think about when writing my first Python API?"
run_inference(PROMPT, client)

import os
import arxiv
import fitz  # PyMuPDF
import json
from openai import AzureOpenAI
import weave
import re
from time import sleep

# Initialize Weave for logging
weave.init('azure_paper_abstract_gen')

# Set up Azure OpenAI
os.environ["AZURE_OPENAI_ENDPOINT"] = "your endpoint url"
os.environ["AZURE_OPENAI_API_KEY"] = "your api key"

client = AzureOpenAI(
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_version="2024-09-01-preview"
)

# Directory to save downloaded papers
download_dir = "arxiv_papers"
os.makedirs(download_dir, exist_ok=True)

# Define AI-specific search queries
search_queries = [
    "Large Language Models for vision tasks AND cat:cs.AI",
    "Multimodal AI techniques AND cat:cs.CV",
    "Applications of Transformers in healthcare AI AND cat:cs.LG",
    "Few-shot learning in AI and ML AND cat:cs.LG",
    "Vision and language models integration AND cat:cs.CV",
    "Domain-specific fine-tuning for ML models AND cat:cs.LG",
    "Foundational models in AI and CV applications AND cat:cs.AI",
    "NLP in robotics and vision systems AND cat:cs.AI",
    "Bias and fairness in AI for CV AND cat:cs.CV",
    "Evaluation metrics for multimodal AI AND cat:cs.LG"
]

def download_papers(max_pages=15, max_attempts_per_query=20):
    """Download one suitable paper for each query, retrying if papers exceed page limit."""
    papers = []
    downloaded_titles = set()
    client = arxiv.Client()

    for query in search_queries:
        paper_found = False
        attempt = 0
        
        while not paper_found and attempt < max_attempts_per_query:
            search = arxiv.Search(
                query=query,
                max_results=100,
                sort_by=arxiv.SortCriterion.SubmittedDate
            )
            
            try:
                results = list(client.results(search))
                start_idx = attempt * 5
                end_idx = start_idx + 5
                current_batch = results[start_idx:end_idx]
                
                for result in current_batch:
                    if result.title not in downloaded_titles:
                        print(f"Downloading: {result.title}")
                        paper_id = result.entry_id.split('/')[-1]
                        pdf_filename = f"{paper_id}.pdf"
                        pdf_path = os.path.join(download_dir, pdf_filename)
                        
                        result.download_pdf(dirpath=download_dir, filename=pdf_filename)
                        
                        try:
                            with fitz.open(pdf_path) as pdf:
                                if pdf.page_count <= max_pages:
                                    papers.append({
                                        "title": result.title,
                                        "file_path": pdf_path,
                                        "arxiv_id": paper_id
                                    })
                                    downloaded_titles.add(result.title)
                                    print(f"Accepted: {result.title}")
                                    paper_found = True
                                    break
                                else:
                                    os.remove(pdf_path)
                                    print(f"Skipped (too many pages: {pdf.page_count}): {result.title}")
                        except Exception as e:
                            print(f"Error checking PDF {pdf_path}: {e}")
                            if os.path.exists(pdf_path):
                                os.remove(pdf_path)
            
                attempt += 1
                if not paper_found:
                    print(f"Attempt {attempt}/{max_attempts_per_query} for query: {query}")
                    sleep(3)
                    
            except Exception as e:
                print(f"Error during download: {e}")
                sleep(3)
                attempt += 1
                continue
        
        if not paper_found:
            print(f"Failed to find suitable paper for query after {max_attempts_per_query} attempts: {query}")

    print(f"\nSuccessfully downloaded {len(papers)} papers")
    return papers

def extract_first_page_text(pdf_path):
    """Extract text from only the first page of the PDF."""
    with fitz.open(pdf_path) as pdf:
        if pdf.page_count > 0:
            page = pdf[0]
            return page.get_text()
    return ""

@weave.op
def extract_abstract_with_azure(text, title):
    """Extract abstract using Azure GPT-4o with JSON output."""
    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a research paper analysis assistant. Extract ONLY the abstract section from the paper content provided. Return the result in JSON format with 'abstract' as the key. If you cannot find the abstract, return an empty string as the value."},
                {"role": "user", "content": f"Paper title: {title}\n\nPaper content:\n\n{text}"}
            ],
            response_format={"type": "json_object"}
        )
        
        content = response.choices[0].message.content
        return json.loads(content)
    except Exception as e:
        print(f"Error extracting abstract for {title}: {e}")
        sleep(3)
        return {"abstract": ""}

def count_words(text):
    """Count words excluding punctuation and special characters."""
    cleaned_text = re.sub(r'[^\w\s]', ' ', text.lower())
    words = [word for word in cleaned_text.split() if word.strip()]
    return len(words)

def main():
    # Download papers
    papers = download_papers()
    print(f"\nDownloaded {len(papers)} papers. Processing abstracts...\n")
    
    # Process papers and extract abstracts
    paper_data = []
    for paper in papers:
        title = paper["title"]
        pdf_path = paper["file_path"]
        
        print(f"Processing: {title}")
        first_page_text = extract_first_page_text(pdf_path)
        abstract_json = extract_abstract_with_azure(first_page_text, title)
        abstract_text = abstract_json.get('abstract', '')
        word_count = count_words(abstract_text)
        
        paper_data.append({
            "title": title,
            "file_path": pdf_path,
            "abstract": abstract_text,
            "word_count": word_count,
            "arxiv_id": paper["arxiv_id"]
        })
        
        sleep(2)

    # Save to JSONL file
    output_file = "paper_abstracts.jsonl"
    with open(output_file, "w") as f:
        for entry in paper_data:
            json.dump(entry, f)
            f.write("\n")

    print(f"\nProcessed {len(paper_data)} papers. Results saved to {output_file}")

if __name__ == "__main__":
    main()

import weave
from weave import Model
import json
from time import sleep
import asyncio
from rouge_score.rouge_scorer import RougeScorer
from typing import Dict, Any
import bert_score
import fitz
from weave.trace.box import unbox
import time
from openai import AzureOpenAI
import time
import json

from weave import Scorer
import json


# Initialize Weave
weave.init('azure_abstract_eval')


gpt4o_client = AzureOpenAI(
    azure_endpoint="https://<your-resource-name>.openai.azure.com/openai/deployments/<gpt-4o-deployment-name>/chat/completions?api-version=2024-08-01-preview",
    api_key="<your-api-key>",
    api_version="2024-08-01-preview"
)

gpt4o_mini_client = AzureOpenAI(
    azure_endpoint="https://<your-resource-name>.openai.azure.com/openai/deployments/<gpt-4o-mini-deployment-name>/chat/completions?api-version=2024-08-01-preview",
    api_key="<your-api-key>",
    api_version="2024-08-01-preview"
)

def create_prediction_prompt(text, title, target_length):
    """Create a prompt for abstract generation with target length."""
    return (
        f"You are tasked with generating an abstract for a research paper titled '{title}'. "
        f"The abstract should be approximately {target_length} words long.\n\n"
        f"Generate an abstract that summarizes the key points of the paper, including the "
        f"research objective, methodology, and main findings. The abstract should be "
        f"self-contained and clearly communicate the paper's contribution. Respond only with the ABSTRACT, and NOT the title\n\n"
        f"Paper content:\n\n{text}"
        f"Respond only with the ABSTRACT!"
    )

def create_evaluation_prompt(gt_abstract: str, generated_abstract: str) -> str:
    """Create standardized evaluation prompt."""
    return f'''You are evaluating how well a generated abstract captures the information from a ground truth abstract.

Ground Truth Abstract:
{gt_abstract}

Generated Abstract:
{generated_abstract}

Rate the generated abstract on a scale from 1-5, where:
1: Poor - Missing most key information or seriously misrepresenting the research
2: Fair - Captures some information but misses crucial elements
3: Good - Captures most key points but has some gaps or inaccuracies
4: Very Good - Accurately captures nearly all key information with minor omissions
5: Excellent - Perfectly captures all key information and maintains accuracy

Respond ONLY with a JSON object containing a single "score" field with an integer value from 1-5.
Example response format:
{{"score": 4}}'''


def extract_text_after_page_one(pdf_path: str) -> str:
    """Extract text from page 2 onwards."""
    if isinstance(pdf_path, weave.trace.box.BoxedStr):
        pdf_path = unbox(pdf_path)
        
    text = ""
    try:
        with fitz.open(pdf_path) as pdf:
            if pdf.page_count > 1:
                for page_num in range(1, pdf.page_count):  # Start from page 2
                    page = pdf[page_num]
                    text += page.get_text()
    except Exception as e:
        print(f"Error extracting text from PDF {pdf_path}: {e}")
    return text


def run_inference(client: AzureOpenAI, model: str, prompt: str, max_retries: int = 10, base_wait: int = 10) -> str:
    """
    Function to perform inference using specified Azure model with exponential backoff retry.
    
    Args:
        client: AzureOpenAI client
        model: Model name/id
        prompt: Input prompt
        max_retries: Maximum number of retry attempts (default: 10)
        base_wait: Initial wait time in seconds (default: 10)
    """
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": "You are a research paper abstract writer. Write clear, concise, and informative abstracts."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.0
            )
            response_json = json.loads(response.model_dump_json(indent=2))
            choices = response_json.get("choices", [])
            if choices:
                content = choices[0].get("message", {}).get("content", "")
                return content
            else:
                print("No content found in response")
                
        except Exception as e:
            wait_time = base_wait * (2 ** attempt)  # Exponential backoff
            print(f"Attempt {attempt + 1}/{max_retries} failed: {e}")
            print(f"Waiting {wait_time} seconds before retrying...")
            time.sleep(wait_time)
            continue
            
    print(f"Failed to get response after {max_retries} retries")
    return None




class GPT4o(Model):
    @weave.op
    def predict(self, title: str, pdf_path: str, word_count: int) -> dict:
        """Predict abstract using GPT-4O."""
        paper_text = extract_text_after_page_one(pdf_path)
        prompt = create_prediction_prompt(title, paper_text, word_count)
        prediction = run_inference("gpt-4o-2024-08-06", prompt)
        return {"model_output": prediction}

class GPT4oMini(Model):
    @weave.op
    def predict(self, title: str, pdf_path: str, word_count: int) -> dict:
        """Predict abstract using GPT-4O-mini."""
        paper_text = extract_text_after_page_one(pdf_path)
        prompt = create_prediction_prompt(title, paper_text, word_count)
        prediction = run_inference("gpt-4o-mini", prompt)
        return {"model_output": prediction}

@weave.op
def bert_scorer(gt_abstract: str, model_output: dict) -> Dict[str, float]:
    """Calculate BERTScore for the abstract."""
    if not model_output or 'model_output' not in model_output:
        return {'bert_score': 0.0}
    
    try:
        P, R, F1 = bert_score.score(
            [model_output['model_output']],
            [gt_abstract],
            lang='en',
            model_type='microsoft/deberta-xlarge-mnli'
        )
        return {'bert_score': float(F1.mean())}
    except Exception as e:
        print(f"Error calculating BERTScore: {e}")
        return {'bert_score': 0.0}


# you can also use the "function" format for your scorer 
# @weave.op
# def GPT4oScorer(gt_abstract: str, model_output: dict) -> dict:
#     """Evaluate abstract using GPT-4o."""
#     if not model_output or 'model_output' not in model_output:
#         return {'gpt4o_score': 0.0}
    
#     try:
#         prompt = create_evaluation_prompt(gt_abstract, model_output["model_output"])
#         response = run_inference(gpt4o_client, "gpt-4o", prompt)
#         if response:
#             # Clean the response text
#             response_text = response.strip()
#             # Remove any additional text before or after the JSON
#             response_text = response_text.split('{')[1].split('}')[0]
#             response_text = '{' + response_text + '}'
            
#             try:
#                 result = json.loads(response_text)
#                 if 'score' in result and isinstance(result['score'], (int, float)):
#                     score = float(result['score'])
#                     if 1 <= score <= 5:
#                         return {'gpt4o_score': score}
#             except json.JSONDecodeError:
#                 print(f"Invalid JSON response: {response_text}")
        
#         print("Using default score due to invalid response")
#         return {'gpt4o_score': 0.0}
        
#     except Exception as e:
#         print(f"Error in GPT-4o evaluation: {e}")
#         return {'gpt4o_score': 0.0}
    


class GPT4oScorer(Scorer):
    model_id: str = "gpt-4o"
    system_prompt: str = "You are evaluating how well a generated abstract captures the information from a ground truth abstract."
    
    @weave.op
    def create_evaluation_prompt(self, gt_abstract: str, generated_abstract: str) -> str:
        """Create standardized evaluation prompt."""
        return f'''Ground Truth Abstract:
{gt_abstract}

Generated Abstract:
{generated_abstract}

Rate the generated abstract on a scale from 1-5, where:
1: Poor - Missing most key information or seriously misrepresenting the research
2: Fair - Captures some information but misses crucial elements
3: Good - Captures most key points but has some gaps or inaccuracies
4: Very Good - Accurately captures nearly all key information with minor omissions
5: Excellent - Perfectly captures all key information and maintains accuracy

Respond ONLY with a JSON object containing a single "score" field with an integer value from 1-5.
Example response format:
{{"score": 4}}'''

    @weave.op
    def call_llm(self, gt_abstract: str, model_output: str) -> dict:
        """Call GPT-4o for evaluation."""
        try:
            prompt = self.create_evaluation_prompt(gt_abstract, model_output)
            response = run_inference(gpt4o_client, self.model_id, prompt)
            
            if response:
                # Clean the response text
                response_text = response.strip()
                # Remove any additional text before or after the JSON
                response_text = response_text.split('{')[1].split('}')[0]
                response_text = '{' + response_text + '}'
                
                try:
                    result = json.loads(response_text)
                    if 'score' in result and isinstance(result['score'], (int, float)):
                        score = float(result['score'])
                        if 1 <= score <= 5:
                            return {'gpt4o_score': score}
                except json.JSONDecodeError:
                    print(f"Invalid JSON response: {response_text}")
            
            print("Using default score due to invalid response")
            return {'gpt4o_score': 0.0}
            
        except Exception as e:
            print(f"Error in GPT-4o evaluation: {e}")
            return {'gpt4o_score': 0.0}

    @weave.op
    def score(self, model_output: dict, gt_abstract: str) -> dict:
        """Score the generated abstract against the ground truth.
        
        Args:
            model_output: Dictionary containing the generated abstract under 'model_output' key
            gt_abstract: The ground truth abstract to compare against
        """
        if not model_output or 'model_output' not in model_output:
            return {'gpt4o_score': 0.0}
        
        return self.call_llm(gt_abstract, model_output['model_output'])
    

@weave.op
def rouge_scorer(gt_abstract: str, model_output: dict) -> Dict[str, float]:
    """Calculate ROUGE scores for the abstract."""
    if not model_output or 'model_output' not in model_output:
        return {
            'rouge1_f': 0.0,
            'rouge2_f': 0.0,
            'rougeL_f': 0.0
        }
    
    try:
        scorer = RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
        scores = scorer.score(gt_abstract, model_output['model_output'])
        
        return {
            'rouge1_f': float(scores['rouge1'].fmeasure),
            'rouge2_f': float(scores['rouge2'].fmeasure),
            'rougeL_f': float(scores['rougeL'].fmeasure)
        }
    except Exception as e:
        print(f"Error calculating ROUGE scores: {e}")
        return {
            'rouge1_f': 0.0,
            'rouge2_f': 0.0,
            'rougeL_f': 0.0
        }

@weave.op
def compression_scorer(gt_abstract: str, model_output: dict) -> Dict[str, float]:
    """Calculate compression ratio of the abstract."""
    if not model_output or 'model_output' not in model_output:
        return {'compression_ratio': 0.0}
    
    try:
        gt_words = len(gt_abstract.split())
        generated_words = len(model_output['model_output'].split())
        
        compression_ratio = min(gt_words, generated_words) / max(gt_words, generated_words)
        
        return {'compression_ratio': float(compression_ratio)}
    except Exception as e:
        print(f"Error calculating compression ratio: {e}")
        return {'compression_ratio': 0.0}

@weave.op
def coverage_scorer(gt_abstract: str, model_output: dict) -> Dict[str, float]:
    """Calculate content coverage using word overlap."""
    if not model_output or 'model_output' not in model_output:
        return {'coverage_score': 0.0}
    
    try:
        gt_words = set(gt_abstract.lower().split())
        generated_words = set(model_output['model_output'].lower().split())
        
        intersection = len(gt_words.intersection(generated_words))
        union = len(gt_words.union(generated_words))
        
        coverage_score = intersection / union if union > 0 else 0.0
        
        return {'coverage_score': float(coverage_score)}
    except Exception as e:
        print(f"Error calculating coverage score: {e}")
        return {'coverage_score': 0.0}


def create_evaluation_dataset(gt_file: str):
    """Create dataset from ground truth file."""
    dataset = []
    with open(gt_file, 'r') as f:
        for line in f:
            entry = json.loads(line)
            dataset.append({
                "title": entry["title"],
                "gt_abstract": entry["abstract"],
                "pdf_path": entry["file_path"],
                "word_count": entry["word_count"]                
            })
    return dataset


async def run_evaluations(gt_file: str):
    """Run evaluations for each model."""
    eval_dataset = create_evaluation_dataset(gt_file)
    
    # Initialize models
    models = {
        "gpt4o": GPT4o(),
        "gpt4o_mini": GPT4oMini()
    }

    gpt_scorer = GPT4oScorer()
    # Setup scorers
    scorers = [
        gpt_scorer, # class based scorer 
        rouge_scorer,
        compression_scorer,
        coverage_scorer,
        bert_scorer
    ]
    
    # Run evaluations
    results = {}
    for model_name, model in models.items():
        print(f"\nEvaluating {model_name}...")
        evaluation = weave.Evaluation(
            dataset=eval_dataset,
            scorers=scorers,
            name=model_name + " Eval"
        )
        results[model_name] = await evaluation.evaluate(model)
    
    # Print results
    print("\nEvaluation Results:")
    for model_name, result in results.items():
        print(f"\n{model_name} Results:")
        print(json.dumps(result, indent=2))
    
    # Save results to file
    output_file = "gpt4o_evaluation_results.json"
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)
    print(f"\nResults saved to {output_file}")
    
    return results

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