Большие языковые модели обладают глубокими знаниями о мире, но они не всеведущий. Из-за продолжительности процесса обучения информация, использованная во время последнего обучения, может быть превышает. Хотя эти модели знакомы с общедоступной информацией в Интернете, они у меня нет знаний о специфический данные, необходимые для реализации бизнес-логики приложений на основе искусственного интеллекта. До появления LLM эти модели обновлялись посредством дополнительного обучения. Однако по мере увеличения масштаба и объема данных, используемых для обучения, прогрессивное обучение стало эффективным только для ограниченного числа случаев использования. ТКАНЬ это метод интеграции пользовательских данных в LLM, основанный на использовании инструкции.

Статья посвящена процессу создания системы формирования ответов службы технической поддержки. Для этого используется методика с Расширенный поиск, известный как генерация расширенного восстановления (RAG). Этот процесс основан на шаблонах поиска и реальных вопросах и ответах службы технической поддержки. Основными используемыми инструментами являются ЯндексGPT И ChromaDB.

Архитектура системы

РАГ с векторным поиском

Для ввода — первое сообщение или продолжающийся диалог от клиента,
Интеграция – получить интеграции с API YandexGPT,
Исследовательская служба – ChromaDB будет хранить вставки вопросов и ответов, искать похожие вставки,
Документы — вопросы и ответы внутренней службы технической поддержки,
Магистр права — YandexGPT, TextGenerationAsync.completion,
Выйти – ответ, требующий модерации.

Создание обертки для YandexGPT

[Исходный код]

Все вызовы API YandexGPT имеют одинаковые заголовки и требуют шаблонного URI.

class YandexGpt:
    def __init__(self, api_key: str, model_uri: str):
        self.api_str = api_key
        self.model_uri = model_uri

    def get_headers(self):
        return {
            "Content-Type": "application/json",
            "Authorization": f"Api-Key {self.api_str}",
            "x-data-logging-enabled": "false"
        }

Во время закрытого тестирования по неизвестным причинам возникало множество ошибок. Чтобы не нарушать логику приложения, нужно было отправлять один и тот же запрос до тех пор, пока не будет получен правильный ответ. Сейчас, когда идет открытое тестирование, я не видел подобных ошибок, но ограничение скорости было ограничено (ошибка 429). Для обработки ошибок написан декоратор, который пытается отправить запрос повторно имеет status_code != 200:

def retry_yandex_gpt_factory(reties=2):
    def retry_yandex_gpt(func):
        def wrapper_retry_yandex_gpt(*args, **kwargs):
            for retry in range(reties):
                res = func(*args, **kwargs)
                if (res.status_code) == 200:
                    return res.json()
                else:
                    print(f"Request failed {res.status_code}: {res.json()}, retry number: {retry + 1}")
                    if res.status_code == 429:
                        sleep(5)

        return wrapper_retry_yandex_gpt

    return retry_yandex_gpt

API-обертка для создания интеграций (цифровых векторов):

class Embeddings(YandexGpt):
    @retry_yandex_gpt_factory(5)
    def text_embedding(self, text: str):
        url = "
        data = {
            "modelUri": self.model_uri,
            "text": text
        }

        return requests.post(url, json=data, headers=self.get_headers())

API YandexGPT имеет синхронный И асинхронный вызовы методов completion. Согласно документации асинхронный вызов медленнее синхронного (8-15 секунд), точнее и в 2 раза дешевле. Для техподдержки точность ответа важнее, чем ожидание в 15 секунд. Метод sync_completion делать completion из асинхронного в синхронный.

class MessageRole(Enum):
    SYSTEM = 'system'
    ASSISTANT = 'assistant'
    USER = 'user'


class Message:
    def __init__(self, role: MessageRole, text: str):
        self.role = role
        self.text = text


class TextGenerationAsync(YandexGpt):
    @retry_yandex_gpt_factory()
    def completion(self, messages: list[Message], stream: bool, temperature: int, max_tokens: int):
        url = "
        data = {
            "modelUri": self.model_uri,
            "completionOptions": {
                "stream": stream,
                "temperature": temperature,
                "maxTokens": max_tokens
            },
            "messages": [{"role": str(msg.role.value), "text": msg.text} for msg in messages]
        }
        return requests.post(url, json=data, headers=self.get_headers())

    def get_operation(self, operation_id: str):
        url = " + operation_id
        return requests.get(url, headers=self.get_headers()).json()

    def sync_completion(self, messages: list[Message], stream: bool, temperature: float, max_tokens: int, max_wait_secs: int):
        operation_id = self.completion(messages, stream, temperature, max_tokens)['id']

        for i in range(max_wait_secs):
            res = self.get_operation(operation_id)
            if res["done"]:
                return res
            sleep(1)

Извлечение данных

[Исходный код]

ЧИТАТЬ   Совет директоров утверждает бизнес-план и бюджет на 2024 год.

Предположим, есть 2 источника информации для расширенного поиска: intents И messages.

Код извлечения намерения
def get_intents_df():
    intents_url = "
    total_count = 3815
    limit = 500
    headers = {
        "Cookie": os.getenv("COOKIE")
    }
    
    df = pd.DataFrame(columns=['id', 'text', 'pattern', 'intentId', 'groupId', 'answer'])
    
    for page in tqdm(range(math.ceil(total_count / limit))):
        res = requests.get(
            intents_url, 
            params={"limit": limit, "count": True, "page": page+1}, 
            headers=headers
        )
        rows = res.json()["rows"]
        df = pd.concat([df, pd.DataFrame(rows)], ignore_index=True)
    return df

intents_df = get_intents_df()
intents_df.to_csv("data/intents.csv", index=False)

Структура исходных данных intents:

intents_df.info()
RangeIndex: 3815 entries, 0 to 3814
Data columns (total 6 columns):
 #   Column    Non-Null Count  Dtype 
---  ------    --------------  ----- 
 0   id        3815 non-null   object
 1   text      3815 non-null   object
 2   pattern   0 non-null      object
 3   intentId  3815 non-null   object
 4   groupId   0 non-null      object
 5   answer    3815 non-null   object
код извлечения сообщения
def remove_now_answered_column(df):
    return df.drop(['nowAnswered'], axis=1)

def get_messages_df():
    messages_url = "
    total_count = 35574
    limit = 500
    page = 1
    headers = {
        "Cookie": os.getenv("COOKIE")
    }
    
    df = pd.DataFrame(columns=['answer', 'answered', 'chatId', 
                               'clientId', 'messageId', "success", 
                               "text"])
    
    for page in tqdm(range(math.ceil(total_count / limit))):
        res = requests.get(
            messages_url, 
            params={"limit": limit, "count": True, "page": page + 1, 
                    "sortBy": "TIMESTAMP", "ascending": False}, 
            headers=headers
        )
        rows = res.json()["rows"]
        df = pd.concat([df, pd.DataFrame(rows)], ignore_index=True) 
    return df.pipe(remove_now_answered_column)

messages_df = get_messages_df()
messages_df.to_csv("data/messages.csv", index=False)

Структура исходных данных messages. В базу данных будут занесены только строки с ненулевым полем. answerпотому что вложения этого поля будут ключом поля text.

messages_df.info()
RangeIndex: 35576 entries, 0 to 35575
Data columns (total 7 columns):
 #   Column     Non-Null Count  Dtype 
---  ------     --------------  ----- 
 0   answer     8414 non-null   object
 1   answered   35576 non-null  object
 2   chatId     35576 non-null  object
 3   clientId   35576 non-null  object
 4   messageId  35576 non-null  object
 5   success    35576 non-null  object
 6   text       35576 non-null  object

Получите интеграции и создайте векторную базу данных в ChromaDB.

[Исходный код]

Создание и сохранение интеграций в формате CSV
def get_emdeddings(text):
    sleep(0.5)
    return embeddings.text_embedding(text)["embedding"]

embeddings = yandexgpt.Embeddings(os.getenv("YANDEX_GPT_KEY"), 
                                  os.getenv("YANDEX_GPT_EMBEDDINGS_URI")
                              )

# Получение эмбеддиногов для intents
df = pd.read_csv("data/intents.csv")
text_embeddings = []
for txt in tqdm(df["text"]):
    text_embeddings.append(get_emdeddings(txt))
df["text_embeddings"] = text_embeddings
df.to_csv("data/intents_with_embeddings.csv", index=False)

# Получение эмбеддиногов для messages
df = pd.read_csv("data/messages.csv").dropna(subset=['answer'])
text_embeddings = []
for txt in tqdm(df["text"]):
    text_embeddings.append(get_emdeddings(txt))
df["text_embeddings"] = text_embeddings
df.to_csv("data/messages_with_embeddings.csv", index=False)
Запуск ChromaDB
docker pull chromadb/chroma
docker run -p 8000:8000 chromadb/chroma

Подключение СУБД к коллекции

chroma_client = chromadb.HttpClient(host="localhost", 
                                    port="8000", 
                                    settings=Settings(anonymized_telemetry=False))

collection = chroma_client.get_or_create_collection("intents")

Загрузка интеграций вопросов, источников и вопросов (метаданных), ответов (документов) в базу данных:

df = pd.read_csv('data/intents_with_embeddings.csv')
texts = df["text"].tolist()
text_embeddings = list(map(
    lambda str_arr: ast.literal_eval(str_arr), 
    df["text_embeddings"].tolist()))
ids = df["id"].astype(str).tolist()
answers = df["answer"].tolist()
collection.upsert(
    ids=ids,
    embeddings=text_embeddings,
    metadatas=[{"source": "intents", "text": txt} for txt in texts],
    documents=answers
)


df = pd.read_csv('data/messages_with_embeddings.csv')
texts = df["text"].tolist()
text_embeddings = list(map(
    lambda str_arr: ast.literal_eval(str_arr), 
    df["text_embeddings"].tolist()))
ids = df["messageId"].astype(str).tolist()
answers = df["answer"].tolist()
collection.upsert(
    ids=ids,
    embeddings=text_embeddings,
    metadatas=[{"source": "messages", "text": txt} for txt in texts],
    documents=answers
)

Генерировать ответы

Исходный код с примерами сборки

ЧИТАТЬ   Владелец KFC в России завершил продажу. Что теперь будет с крупнейшей сетью быстрого питания?

Связь в ChromaDB и Создание Обертки YandexGPT для интеграций и генерации текста:

chroma_client = chromadb.HttpClient(host="localhost", 
                                    port="8000", 
                                    settings=Settings(anonymized_telemetry=False))
embeddings = yandexgpt.Embeddings(os.getenv("YANDEX_GPT_KEY"), 
                                  os.getenv("YANDEX_GPT_EMBEDDINGS_URI"))
textGenerationAsync = yandexgpt.TextGenerationAsync(os.getenv("YANDEX_GPT_KEY"), 
                                  os.getenv("YANDEX_GPT_URI"))
  1. получать интеграции вопрос,

  2. исследовать соответствующий интеграции с документами и метаданными (6 шт.),

  3. мусор тех, кто расстояние более 1,

  4. упражнение быстрыйСоздание диалог на основе 6 актуальных вопросов и ответов, генерация описание при температуре 0, макс. количество сгенерированных токенов: 250.

def format_qa(func):
    def wrapper_format_qa(*args, **kwargs):
        print("Вопрос:", args[0])
        print("Ответ:", func(*args, **kwargs))
    return wrapper_format_qa

@format_qa
def generate_answer(question: str):
    result = collection.query(
        query_embeddings=[embeddings.text_embedding(question)["embedding"]],
        n_results=6,
    )
    
    messages = [yandexgpt.Message(yandexgpt.MessageRole.SYSTEM, "Ты специалист технической поддежки. На основе сообщений, написанных тобой выше, сгенерируй сообщение")]
    for distance, metadata, document in zip(
            result["distances"][0], result["metadatas"][0], result["documents"][0]
    ):
        if distance < 1:
            messages.append(yandexgpt.Message(yandexgpt.MessageRole.USER, metadata["text"]))
            messages.append(yandexgpt.Message(yandexgpt.MessageRole.ASSISTANT, document))
    
    messages.append(yandexgpt.Message(yandexgpt.MessageRole.USER, question))
    return textGenerationAsync.sync_completion(messages, False, 0, 250, 20)["response"]["alternatives"][0]["message"]["text"]

Заключение

Сообщения, сгенерированные RAG, могут ускорить работу ТП, но сообщения, сгенерированные таким способом, нельзя допускать к отправке клиентам без модерации. «Дополнительное обучение» может проходить в режиме реального времени, добавляя новый или удаляя старый вектор вопросов с ответом в базу данных.

Source

От admin