通过强化学习学习提示变量注入¶

# 安装必要的包
# ! pip install langchain langchain-experimental matplotlib vowpal_wabbit_next sentence-transformers pandas

# 定义了四个餐点，其中一些是素食，一些不是

meals = [
    "Beef Enchiladas with Feta cheese. Mexican-Greek fusion",  # 牛肉恩奇拉达配菲塔奶酪。墨西哥-希腊融合
    "Chicken Flatbreads with red sauce. Italian-Mexican fusion",  # 鸡肉薄饼配红酱。意大利-墨西哥融合
    "Veggie sweet potato quesadillas with vegan cheese",  # 素食甜薯饼配纯素奶酪
    "One-Pan Tortelonni bake with peppers and onions",  # 一锅烤托特洛尼意面配辣椒和洋葱
]

# 选择并配置您选择的LLM（语言模型）

from langchain_openai import OpenAI

# 创建一个OpenAI对象，并指定使用的模型为"gpt-3.5-turbo-instruct"
llm = OpenAI(model="gpt-3.5-turbo-instruct")

from langchain.prompts import PromptTemplate

# 在这里，我使用变量meal，它将被上面的一餐替换
# 还有一些变量，如user、preference和text_to_personalize，我将在链运行时提供

PROMPT_TEMPLATE = """这是一餐的描述: "{meal}".

将这餐嵌入到给定的文本中: "{text_to_personalize}".

在开头添加一个包含用户姓名"{user}"和他们的偏好"{preference}"的个性化消息。

让它听起来不错。
"""

PROMPT = PromptTemplate(
    input_variables=["meal", "text_to_personalize", "user", "preference"],
    template=PROMPT_TEMPLATE,
)

# 导入langchain_experimental包中的rl_chain模块
import langchain_experimental.rl_chain as rl_chain

# 使用rl_chain模块中的PickBest类的from_llm方法，传入llm和prompt参数，赋值给chain变量
chain = rl_chain.PickBest.from_llm(llm=llm, prompt=PROMPT)

# 定义一个response变量，使用chain.run方法运行一系列操作
response = chain.run(
    # 选择餐点
    meal=rl_chain.ToSelectFrom(meals),
    # 基于用户Tom
    user=rl_chain.BasedOn("Tom"),
    # 基于用户的偏好，包括素食和普通乳制品
    preference=rl_chain.BasedOn(["Vegetarian", "regular dairy is ok"]),
    # 个性化文本信息
    text_to_personalize="This is the weeks specialty dish, our master chefs \
        believe you will love it!",
)

# 定义评分标准模板
scoring_criteria_template = (
    "Given {preference} rank how good or bad this selection is {meal}"
)

# 创建一个PickBest对象，使用llm作为参数
chain = rl_chain.PickBest.from_llm(
    llm=llm,
    prompt=PROMPT,
    selection_scorer=rl_chain.AutoSelectionScorer(
        llm=llm, scoring_criteria_template_str=scoring_criteria_template
    ),
)

class CustomSelectionScorer(rl_chain.SelectionScorer):
    def score_response(
        self, inputs, llm_response: str, event: rl_chain.PickBestEvent
    ) -> float:
        print(event.based_on)  # 打印基于的信息
        print(event.to_select_from)  # 打印可供选择的信息

        # 在这里可以构建一个复杂的评分函数
        # 最好的情况是评分在0到1之间，但不强制要求

        selected_meal = event.to_select_from["meal"][event.selected.index]  # 获取选择的餐点
        print(f"selected meal: {selected_meal}")  # 打印选择的餐点

        if "Tom" in event.based_on["user"]:
            if "Vegetarian" in event.based_on["preference"]:
                if "Chicken" in selected_meal or "Beef" in selected_meal:
                    return 0.0  # 如果用户是Tom且偏好素食，且选择的餐点中包含鸡肉或牛肉，则返回0.0
                else:
                    return 1.0  # 如果用户是Tom且偏好素食，且选择的餐点中不包含鸡肉或牛肉，则返回1.0
            else:
                if "Chicken" in selected_meal or "Beef" in selected_meal:
                    return 1.0  # 如果用户是Tom但不偏好素食，且选择的餐点中包含鸡肉或牛肉，则返回1.0
                else:
                    return 0.0  # 如果用户是Tom但不偏好素食，且选择的餐点中不包含鸡肉或牛肉，则返回0.0
        else:
            raise NotImplementedError("I don't know how to score this user")  # 如果用户不是Tom，则抛出NotImplementedError异常，表示无法对该用户进行评分

# 从llm中创建一个PickBest对象的链
chain = rl_chain.PickBest.from_llm(
    llm=llm,  # 使用给定的llm模型
    prompt=PROMPT,  # 使用给定的PROMPT作为提示
    selection_scorer=CustomSelectionScorer(),  # 使用自定义的选择评分器
)

# 自定义选择评分器类，继承自SelectionScorer类
class CustomSelectionScorer(rl_chain.SelectionScorer):
    def score_preference(self, preference, selected_meal):
        # 如果用户偏好中包含"Vegetarian"
        if "Vegetarian" in preference:
            # 如果选中的餐点中包含"Chicken"或"Beef"
            if "Chicken" in selected_meal or "Beef" in selected_meal:
                return 0.0
            else:
                return 1.0
        else:
            # 如果选中的餐点中包含"Chicken"或"Beef"
            if "Chicken" in selected_meal or "Beef" in selected_meal:
                return 1.0
            else:
                return 0.0

    def score_response(
        self, inputs, llm_response: str, event: rl_chain.PickBestEvent
    ) -> float:
        # 获取选中的餐点
        selected_meal = event.to_select_from["meal"][event.selected.index]

        # 如果基于用户的信息中包含"Tom"
        if "Tom" in event.based_on["user"]:
            # 根据用户偏好评分选中的餐点
            return self.score_preference(event.based_on["preference"], selected_meal)
        # 如果基于用户的信息中包含"Anna"
        elif "Anna" in event.based_on["user"]:
            # 根据用户偏好评分选中的餐点
            return self.score_preference(event.based_on["preference"], selected_meal)
        else:
            # 抛出未实现错误
            raise NotImplementedError("I don't know how to score this user")

# 创建一个链式模型，使用PickBest策略
chain = rl_chain.PickBest.from_llm(
    llm=llm,  # 使用llm作为模型
    prompt=PROMPT,  # 设置提示文本
    selection_scorer=CustomSelectionScorer(),  # 使用自定义的选择评分器
    metrics_step=5,  # 每5个步骤计算一次指标
    metrics_window_size=5,  # 指标计算的滚动窗口大小为5
)

# 创建一个随机链式模型，使用PickBest策略
random_chain = rl_chain.PickBest.from_llm(
    llm=llm,  # 使用llm作为模型
    prompt=PROMPT,  # 设置提示文本
    selection_scorer=CustomSelectionScorer(),  # 使用自定义的选择评分器
    metrics_step=5,  # 每5个步骤计算一次指标
    metrics_window_size=5,  # 指标计算的滚动窗口大小为5
    policy=rl_chain.PickBestRandomPolicy,  # 设置随机策略，而不是默认策略
)

# 循环20次
for _ in range(20):
    try:
        # 使用chain运行指定参数的函数
        chain.run(
            meal=rl_chain.ToSelectFrom(meals),  # 选择餐点
            user=rl_chain.BasedOn("Tom"),  # 用户为Tom
            preference=rl_chain.BasedOn(["Vegetarian", "regular dairy is ok"]),  # 偏好为素食和普通乳制品
            text_to_personalize="This is the weeks specialty dish, our master chefs believe you will love it!",  # 个性化文本
        )
        # 使用random_chain运行指定参数的函数
        random_chain.run(
            meal=rl_chain.ToSelectFrom(meals),  # 选择餐点
            user=rl_chain.BasedOn("Tom"),  # 用户为Tom
            preference=rl_chain.BasedOn(["Vegetarian", "regular dairy is ok"]),  # 偏好为素食和普通乳制品
            text_to_personalize="This is the weeks specialty dish, our master chefs believe you will love it!",  # 个性化文本
        )

        # 使用chain运行指定参数的函数
        chain.run(
            meal=rl_chain.ToSelectFrom(meals),  # 选择餐点
            user=rl_chain.BasedOn("Anna"),  # 用户为Anna
            preference=rl_chain.BasedOn(["Loves meat", "especially beef"]),  # 偏好为喜欢肉类，尤其是牛肉
            text_to_personalize="This is the weeks specialty dish, our master chefs believe you will love it!",  # 个性化文本
        )
        # 使用random_chain运行指定参数的函数
        random_chain.run(
            meal=rl_chain.ToSelectFrom(meals),  # 选择餐点
            user=rl_chain.BasedOn("Anna"),  # 用户为Anna
            preference=rl_chain.BasedOn(["Loves meat", "especially beef"]),  # 偏好为喜欢肉类，尤其是牛肉
            text_to_personalize="This is the weeks specialty dish, our master chefs believe you will love it!",  # 个性化文本
        )
    except Exception as e:
        print(e)  # 捕获异常并打印