CFC使用教程

# -*- coding: utf-8 -*-

from typing import List
import os
import re
import json

def reward_func(queries: List[str], prompts: List[str], labels: List[str]) -> List[float]:
    """
    RLHF reward function based on set (Jaccard) similarity.
    
    Args:
        queries (List[str]): List of concatenated prompt and content strings.
        prompts (List[str]): List of prompt strings.
        labels (List[str]): List of desired response strings.
    
    Returns:
        List[float]: List of reward scores (0 to 1) corresponding to each query.
    """
    rewards = []
    max_prompt_len = 160000
    
    for query, prompt, label in zip(queries, prompts, labels):
        # Extract content by removing the prompt from the query
        content = query[min(int(max_prompt_len), len(prompt)):]
        
        set_content = set(tokenize(content))
        set_label = set(tokenize(label))
        if not set_content and not set_label:
            similarity = 1.0
        else:
            intersection = set_content.intersection(set_label)
            union = set_content.union(set_label)
            similarity = len(intersection) / len(union)
        # Ensure similarity is between 0 and 1
        similarity = max(0.0, min(similarity, 1.0))
        rewards.append(similarity)
        # 注意，返回的reward不用再转换成torch里的tensor了，这一步在训练代码中做
    return rewards

def tokenize(text: str) -> List[str]:
    """
    Tokenize the input text into words.
    
    Args:
        text (str): Input text.
    
    Returns:
        List[str]: List of word tokens.
    """
    return re.findall(r'\b\w+\b', text.lower())

def handler(event, context): 
    """
    服务端收到 RFT训练工程请求的event示例：
    {
        'resource': '/get_rewards',
        'path': '/get_rewards',
        'httpMethod': 'POST',
        'headers': {
            'Accept': '*/*',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'close',
            'Content-Length': '392',
            'Content-Type': 'application/json',
            'User-Agent': 'python-requests/2.31.0',
            'X-Bce-Request-Id': '44dbabc1-8a24-4b10-aa15-8f0933c905a4'
        },
        'queryStringParameters': {},
        'pathParameters': {},
        'requestContext': {
            'stage': 'cfc',
            'requestId': '44dbabc1-8a24-4b10-aa15-8f0933c905a4',
            'resourcePath': '/get_rewards',
            'httpMethod': 'POST',
            'apiId': '7eada4qvyq9vk',
            'sourceIp': '220.181.3.189'
        },
        'body': '{"query": ["Prompt: How are you?\\nI\'m doing well, thank you!", "Prompt: Tell me a joke.\\nWhy did the chicken cross the road?", "Prompt: What\'s the weather today?\\nIt\'s sunny and warm."], "prompts": ["Prompt: How are you?", "Prompt: Tell me a joke.", "Prompt: What\'s the weather today?"], "labels": ["I\'m doing well, thank you!", "Why did the chicken cross the road?", "It\'s sunny and warm."]}',
        'isBase64Encoded': False
    }
    这里的event是一个dict形式，POST请求传入的信息在event["body"]中
    """
    print(event)
    data = event.get("body")
    try:
        data = json.loads(data)
        queries = data.get("query")
        prompts = data.get("prompts")
        labels = data.get("labels")
        rewards = reward_func(queries, prompts, labels)
        results = {
            "rewards": rewards
        }
    except Exception as e:
        results = {
            "error": str(e),
            "input_events": str(event),
        }
    return results

import time
import requests


def request_api_wrapper(url, data, score_key="rewards", try_max_times=5):
    """Synchronous request API wrapper"""
    headers = {
        "Content-Type": "application/json",
    }
    for _ in range(try_max_times):
        try:
            response = requests.post(url=url, json=data, headers=headers, timeout=180)
            response.raise_for_status()
            response = response.json()
            assert score_key in response, f"{score_key} not in {response}"
            return response.get(score_key)
        except requests.RequestException as e:
            print(f"Request error, please check: {e}")
        except Exception as e:
            print(f"Unexpected error, please check: {e}")
        time.sleep(1)

    raise Exception(f"Request error for {try_max_times} times, returning None. Please check the API server.")


def remote_rm_fn(api_url, queries, prompts, labels, score_key="rewards"):
    """remote reward model API
    api_url: RM API, We assume that the API supports two modes: merging query + response and not merging
    queries: query+response with the template
    design is made optional.
    score_key: RM score key
    """
    scores = request_api_wrapper(api_url, {"query": queries, "prompts": prompts, "labels": labels}, score_key)
    return scores



if __name__ == "__main__":
    # test utils
    url = "https://XXXX/get_rewards"
    queries = [
        "Prompt: How are you?\nI'm doing well, thank you!",
        "Prompt: Tell me a joke.\nWhy did the chicken cross the road?",
        "Prompt: What's the weather today?\nIt's sunny and warm."
    ]
    
    prompts = [
        "Prompt: How are you?",
        "Prompt: Tell me a joke.",
        "Prompt: What's the weather today?"
    ]
    
    labels = [
        "I'm doing well, thank you!",
        "Why did the chicken cross the road?",
        "It's sunny and warm."
    ]
    score = remote_rm_fn(url, queries, prompts, labels)
    print(score)