《大模型应用开发极简入门》学习笔记

2.4 开始使用OpenAI Python库

调用ChatCompletion端点示例

from dotenv import load_dotenv

load_dotenv()#To load environmental viriation and add it in openai
import os
from openai import OpenAI

client = OpenAI()

# Make sure the environment variable OPENAI_API_KEY is set.

# Call the openai ChatCompletion endpoint, with th ChatGPT model
response = client.chat.completions.create(model="gpt-3.5-turbo",
messages=[
      {"role": "user", "content": "Hello World!"}
  ])

# Extract the response
print(response.choices[0].message.content)

将有以下输出：

1	Hello there! How may I assist you today?

ChatCompletion端点的输入选项

接下来更详细的了解一下如何使用ChatCompletion端点及其create方法。

create方法让用户能够调用OpenAI的模型。

1.必须输入的参数

model（字符串）：所选模型的ID
messages(数组)：表示对话的消息对象数组。消息对象有两个属性：role（可能的值有：system,user,assitant）和conten（包含对话消息的字符串）

2.一些可选参数

functions（数组）：由可用函数组成的数组
function_call（字符串或对象）：控制模型的相应方式
- none表示模型必须以标准方式响应用户
- {"name":"my_function"}表示模型必须给出使用指定函数的回答
- auto表示模型可以在标准方式响应用户和functions数组定义的函数之间进行选择
temperature（数值，默认值为1；可接受介于0和2之间的值）：温度值越高，结果的随机性就越强。当温度被设置为0时，LLM始终选择概率最高的标记。较高的温度可以产生更多样化、更具创造力的输出
n（整型，默认值为1）：通过这个参数，可以为给定的输入消息生成多个回答。
sream（布尔型，默认为False）：这个参数将允许回答以流的格式呈现，这意味着并非一次性发送整条消息。
max_tokens（整型）：这个参数指定在聊天中生成的最大标记数。

LLM驱动型应用程序的漏洞

即容易受到提示词注入攻击。

如果计划开发和部署一个面向用户的应用程序，那么建议结合以下两种方法。

添加分析层来过滤用户输入和模型输出
意识到提示词注入不可避免，并采取一定的预防措施

分析输入和输出

使用特定规则控制用户输入
控制输入长度
控制输出
监控和审计
意图分析

示例项目

1.构建新闻稿生成器

from typing import List
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI()

def ask_chatgpt(messages):
    response = client.chat.completions.create(model="gpt-3.5-turbo",
                                             messages=messages)
    return (response.choices[0].message.content)

prompt_role='''You are an assitant for journalists.
Your task is to write articles, based on the FACTS that given to you.
You should respect the instructions: the TONE, the LENGTH, and the STYLE'''

def assist_journalists(
		facts:List[str],)
		tone: str, length_words: int, style: str):
     facts = ",".join(facts)
    prompt = f'{prompt_role}\nFACTS: {facts}\nTONE: {tone}\nLENGTH: 			{length_words} words\nSTYLE: {style}'

print(
	assist_journalist(
    	['The sky is blue', 'The grass is green'],
    	'informal','100','blogpost'))

3.打造《塞尔达传说：旷野之息》专家

由于任天堂指南的PDF太大，不可能通过提示词发送给OpenAI。这里使用一种以软件为导向的方法。基本思想是使用GPT-4或ChatGPT进行信息还原，我们要求它根据我们认为可能与问题匹配的文本摘录来生成答案。

需要以下三个组件。

意图服务

用于检测用户的意图。

from openai import OpenAI

client = OpenAI()

class ResponseService():
     def __init__(self):
        pass
     
     def generate_response(self, facts, user_question):
         response = client.chat.completions.create(model="gpt-3.5-turbo",
         messages=[
               {"role": "user", "content": 'Based on the FACTS, give an answer to the QUESTION.'+ 
                f'QUESTION: {user_question}. FACTS: {facts}'}
            ])

         # extract the response
         return (response.choices[0].message.content)

信息检索服务

该服务将获取意图服务的输出并检索正确的信息。

import numpy as np
from openai import OpenAI

client = OpenAI()
from pypdf import PdfReader
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query

import redis

INDEX_NAME = "embeddings-index"           # name of the search index
PREFIX = "doc"                            # prefix for the document keys
# distance metric for the vectors (ex. COSINE, IP, L2)
DISTANCE_METRIC = "COSINE"

REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_PASSWORD = "" 

class DataService():

    def __init__(self):
        # Connect to Redis
        self.redis_client = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            password=REDIS_PASSWORD
        )

    def drop_redis_data(self, index_name: str = INDEX_NAME):
        try:
            self.redis_client.ft(index_name).dropindex()
            print('Index dropped')
        except:
            # Index doees not exist
            print('Index does not exist')

    def load_data_to_redis(self, embeddings):
        # Constants
        vector_dim = len(embeddings[0]['vector'])  # length of the vectors
        
		# Initial number of vectors
        vector_number = len(embeddings)

        # Define RediSearch fields
        text = TextField(name="text")
        text_embedding = VectorField("vector",
                                     "FLAT", {
                                         "TYPE": "FLOAT32",
                                         "DIM": vector_dim,
                                         "DISTANCE_METRIC": "COSINE",
                                         "INITIAL_CAP": vector_number,
                                     }
                                     )
        fields = [text, text_embedding]

        # Check if index exists
        try:
            self.redis_client.ft(INDEX_NAME).info()
            print("Index already exists")
        except:
            # Create RediSearch Index
            self.redis_client.ft(INDEX_NAME).create_index(
                fields=fields,
                definition=IndexDefinition(
                    prefix=[PREFIX], index_type=IndexType.HASH)
            )

        for embedding in embeddings:
            key = f"{PREFIX}:{str(embedding['id'])}"
            embedding["vector"] = np.array(
                embedding["vector"], dtype=np.float32).tobytes()
            self.redis_client.hset(key, mapping=embedding)
        print(
            f"Loaded {self.redis_client.info()['db0']['keys']} documents in Redis search index with name: {INDEX_NAME}")

    def pdf_to_embeddings(self, pdf_path: str, chunk_length: int = 1000):
        # Read data from pdf file and split it into chunks
        reader = PdfReader(pdf_path)
        chunks = []
        for page in reader.pages:
            text_page = page.extract_text()
            chunks.extend([text_page[i:i+chunk_length].replace('\n', '')
                          for i in range(0, len(text_page), chunk_length)])

        # Create embeddings
        response = client.embeddings.create(model='text-embedding-ada-002', input=chunks)
        return [{'id': value.index, 'vector':value.embedding, 'text':chunks[value.index]} for value in response.data]

    def search_redis(self,
                     user_query: str,
                     index_name: str = "embeddings-index",
                     vector_field: str = "vector",
                     return_fields: list = ["text", "vector_score"],
                     hybrid_fields="*",
                     k: int = 5,
                     print_results: bool = False,
                     ):
        # Creates embedding vector from user query
        embedded_query = client.embeddings.create(input=user_query,
                                                 model="text-embedding-ada-002").data[0].embedding
        # Prepare the Query
        base_query = f'{hybrid_fields}=>[KNN {k} @{vector_field} $vector AS vector_score]'
        query = (
            Query(base_query)
            .return_fields(*return_fields)
            .sort_by("vector_score")
            .paging(0, k)
            .dialect(2)
        )
        params_dict = {"vector": np.array(
            embedded_query).astype(dtype=np.float32).tobytes()}
        # perform vector search
        results = self.redis_client.ft(index_name).search(query, params_dict)
        if print_results:
            for i, doc in enumerate(results.docs):
                score = 1 - float(doc.vector_score)
                print(f"{i}. {doc.text} (Score: {round(score ,3) })")
        return [doc['text'] for doc in results.docs]

响应服务

该服务将使用信息检索服务的输出，并从中生成用户所提问题的答案。

from openai import OpenAI

client = OpenAI()

class IntentService():
     def __init__(self):
        pass
     
     def get_intent(self, user_question: str):
         # call the openai ChatCompletion endpoint
         response = client.chat.completions.create(model="gpt-3.5-turbo",
         messages=[
               {"role": "user", "content": f'Extract the keywords from the following question: {user_question}'+
                 'Do not answer anything else, only the keywords.'}
            ])

         # extract the response
         return (response.choices[0].message.content)

整合所有内容

from dotenv import load_dotenv

load_dotenv()
from intentservice import IntentService
from responseservice import ResponseService
from dataservice import DataService

# Example pdf
pdf = 'files/ExplorersGuide.pdf'

data_service = DataService()

# Drop all data from redis if needed
data_service.drop_redis_data()

# Load data from pdf to redis
data = data_service.pdf_to_embeddings(pdf)

data_service.load_data_to_redis(data)

intent_service = IntentService()
response_service = ResponseService()

# Question 
question = 'Where to find treasure chests?'
# Get the intent
intents = intent_service.get_intent(question)
# Get the facts
facts = data_service.search_redis(intents)
# Get the answer
answer = response_service.generate_response(facts, question)
print(answer)

4.1 提示工程

4.1.1 设计有效的提示词

通过提示词，可以完成摘要、文本分类、情感分析和问题回答等任务。在这些任务中，通常需要在提示词中定义三大要素：角色，上下文和任务。

上下文

上下文通过引导GPT模型进行思考来使其输出更有价值的回答。构建良好上下文的提示词是一个迭代过程，通常需要试错。为了找到可改进之处，可以请GPT提些意见。

任务

任务定义了对GPT模型的用法，并且应该明确且具体。应该提供足够的任务信息，并在提示词中使用合适的短语来引导模型给出你所期望的结果。

角色

在撰写提示词时，影响模型的最后一种方式是赋予其一个角色。角色和上下文可以独立使用，但同时使用可以加强对模型输出的控制。

4.1.2 逐步思考

在提示词的末尾添加“让我们逐步思考”这样的话，已被证明可以使模型解决更复杂的推理问题。这种技术称为零样本思维链策略。论文“Large Language Models are Zero-Shot Reasoners”作者发现，这个技巧对于多步算术问题、涉及符号推理的问题、涉及策略的问题和其他涉及推理的问题十分有效。

4.1.3 实现少样本学习

少样本学习是由Tom B. Brown等人在论文“Language Models Are Few-Shot Learners”中提出的，它指的是LLM仅通过提示词中的几个示例就能进行概括并给出有价值的结果

4.1.4 改善提示效果

指示模型提出更多的问题

在提示词的末尾，询问模型是否理解问题并指示模型提出更多问题。

格式化输出

重复指示

使用负面提示

添加长度限制

4.2 微调

微调就是使用特定领域的数据集来更新现有GPT模型的内部权重。微调的目标是使新模型能够在特定领域中做出比原始GPT模型更好的预测。即针对特定任务在一组数据上重新训练现有模型，以提高模型的性能并使其回答更准确。在微调过程中，模型的内部参数得到更新。

准备好训练数据后，需要将其上传到OpenAI服务器。OpenAI 提供了不同的函数来操作文件。

上传文件

openai.File.crate(

file=open("out_openai_completion_prepared.jsonl", "rb"),

purpose='file-tune'

)
删除文件

openai.File.delete("file-")
列出所有已上传的文件

openai.File.list()
列出微调作业

openai.FineTun.list()
取消微调作业

openai.FineTune.cancel()

openai.FineTune.create的一些输入参数（具体可查询OpenAI网站）

training_file
modle
validation_file
suffix