Авторы: Бруно Пистоне, Арис Цакпинис
В области обработки естественного языка (NLP) стремление к созданию высококачественного, контекстуально релевантного текста было давней проблемой.
С появлением генеративных моделей аббревиатура RAG представляет собой инновационный подход, известный как Поисково-дополненная генерация. Будучи представленным исследованиями в 2021 году, RAG сочетает в себе мощь методов, основанных на поиске, с креативностью генеративных моделей, революционизируя область и расширяя границы генерации текста.
Включая внешние знания и методы поиска информации, RAG позволяет языковые модели для создания текстов, которые не только беглы и связны, но и основаны на реальных фактах и контекстах.
В этой статье мы углубимся в тонкости RAG, изучая лежащие в его основе механизмы, инструменты и сервисы, которые могут помочь нам в использовании базовых моделей в качестве сервиса для создания надежных приложений генеративного ИИ с использованием подхода RAG для нескольких вариантов использования, таких как поиск. Вопросы и ответы, приложения для чат-ботов и расширенный чат с поиском контента.
Обзор пошагового руководства
В этом блоге мы проведем вас через пошаговый процесс создания приложения генеративного ИИ с использованием подхода SaaS, используя документы вашей базы знаний. Следующие шаги описывают процесс:
- Автоматизация извлечения текста из файлов формата документа: узнайте, как автоматизировать извлечение текста из файлов формата документа, уделяя особое внимание документам PDF.
- Введение в методы извлечения контента. Мы представим обзор различных методов извлечения релевантного контента из базы знаний.
- Создание структуры данных базы знаний с помощью базы данных векторного поиска. Узнайте, как создать структуру данных базы знаний с помощью базы данных векторного поиска, используя Amazon OpenSearch с Elasticsearch.
- Проектирование и разработка цепочки действий для борьбы с галлюцинациями и повышения фактической правильности ответов на основе LLM: изучите дизайн и разработку цепочки действий, в которой используются базовые модели и извлеченные документы для получения точных и отслеживаемых ответов для трех различных вариантов использования: Чат-бот , и вопросы и ответы в чате.
- Разработка внешнего приложения и адаптация поведения к вариантам использования: узнайте, как разработать внешнее приложение и адаптировать его поведение к различным вариантам использования, реализованным на предыдущем шаге.
Кроме того, мы обсудим, как сопоставить архитектуру решения высокого уровня, выделив используемые сервисы и модели Foundation:
- Amazon OpenSearch с Elasticsearch в качестве векторной базы данных: используйте Amazon OpenSearch с Elasticsearch для хранения и извлечения векторных представлений документов в базе знаний.
- Amazon Textract для извлечения текста из PDF-документов: используйте Amazon Textract для автоматического извлечения текста из PDF-документов, делая его доступным для дальнейшей обработки.
- Пошаговые функции AWS для оркестровки: используйте пошаговые функции AWS для оркестровки шагов, связанных с извлечением и индексированием текста.
- Amazon SageMaker Jumpstart для развертывания базовых моделей: разверните базовые модели, в том числе Falcon 40B (большая языковая модель для генерации текста) и HuggingFace GPT-J 6B (для создания вложений из извлеченного текста), с помощью Amazon SageMaker Jumpstart.
В дополнение к блогу мы предоставляем репозиторий GitHub с блокнотами и примерами кода.
Поиск контента: краткая предыстория
Благодаря последним достижениям в области генеративного ИИ концепция извлечения информации из документов достигает новых высот в эпоху генеративного ИИ. Но это не совсем новая концепция.
Начиная с начала 2000-х, благодаря Apache Lucene, мы могли получать доступ к информации, полученной из полуструктурированных и неструктурированных данных, с помощью подхода, основанного на API. Эта концепция, называемая семантическим поиском, развивалась с помощью таких сервисов, как Apache Solr, Elasticsearch, и была усовершенствована за счет внедрения методов обработки естественного языка (NLP), таких как Word2vec, которые дали возможность генерировать вложения из данный текст.
В 2017 году был представлен новый тип архитектуры модели под названием трансформеры, который значительно превзошел все существующие эталоны в области NLP и за ее пределами. Благодаря достижениям этого нового класса моделей NLP способность преобразовывать текст во вложения быстро значительно улучшилась благодаря расширенной способности этих моделей фиксировать контекстуализированную семантику. Это также произошло, когда решения для поиска контента претерпели значительные дальнейшие улучшения.
Использование Фундаментальных моделей, которые обучаются на огромных объемах данных в масштабе, вводит дополнительный уровень интеллекта в архитектуру извлечения контента, тем самым расширяя концепцию Retributed Augmented Generation (RAG).
Есть три важных аспекта, на которые следует обратить внимание при использовании RAG:
- Как обрабатывать документы для создания релевантной базы знаний. На основе характеристик выбранной модели внедрения очень важно правильно обрабатывать информацию, извлеченную из документов, и генерировать меньшую ее часть, называемую фрагменты, создавать вложения векторов и сохранять их в масштабируемой базе данных векторов.
- Синергетическая интеграция поиска информации и генерации текста: базовая модель использует полученную информацию для управления и обогащения процесса генерации, в результате чего результаты являются релевантными, информативными и фактически правильными в данном контексте. Кроме того, поскольку мы знаем, откуда мы взяли информацию, мы также можем отнести конкретный ответ к исходному источнику информации.
- Качество модели встраивания и базовой модели. В RAG важно качество векторных вложений, сгенерированных с помощью модели встраивания, а также качество базовой модели, используемой для понимания и создания последовательных и информативных выходы.
В следующих абзацах мы подробно расскажем, как правильно определить сквозную архитектуру для вышеупомянутых трех основных аспектов RAG, выделив сервисы AWS, принятые для решения.
Как обрабатывать документы для создания соответствующей базы знаний
В области машинного обучения одна из ключевых проблем заключается в обработке различных форм полуструктурированных и неструктурированных данных. К ним относятся текстовые файлы, содержащие длинные абзацы, или многостраничные PDF-файлы, в которых часто отсутствует формат с возможностью поиска, что создает трудности при извлечении содержимого в текстовом формате.
Чтобы извлечь ценную информацию из таких данных, требуются специальные методы и инструменты для преодоления сложностей. Обработка естественного языка (NLP) играет жизненно важную роль в решении этой проблемы, используя алгоритмы и модели, которые могут анализировать и интерпретировать неструктурированные текстовые данные.
Благодаря применению методов НЛП исследователи и специалисты по данным могут разблокировать информацию, содержащуюся в этих сложных форматах данных, что позволяет извлекать значимый текст и расширяет возможности дальнейшего анализа и процессов принятия решений.
AWS предоставляет несколько сервисов, которые могут помочь в извлечении необходимой информации из PDF-документов, например Amazon Textract, облачный сервис, упрощающий извлечение текста и данных из различных типов документов.
Он использует технологию оптического распознавания символов (OCR) вместе с алгоритмами машинного обучения для автоматического анализа и извлечения информации из отсканированных документов, файлов PDF, изображений и других источников неструктурированных данных.
Оркестрацию заданий Amazon Textract и извлечение контента можно выполнять с помощью сервисов AWS Lambda и AWS Step Functions.
AWS Lambda — это сервис безсерверных вычислений, который позволяет запускать кода без предоставления серверов или управления ими, сосредоточившись только на коде приложения, в то время как AWS позаботится о базовой инфраструктуре и масштабировании.
AWS Step Functions — это бессерверный рабочий процесс, который позволяет координировать и организовывать несколько сервисов и задач AWS для создания масштабируемых и отказоустойчивых приложений со сложными рабочими процессами.
При работе с многостраничными документами Amazon Textract предлагает возможность выполнять асинхронные задания для извлечения соответствующего текста. Время извлечения напрямую зависит от количества страниц в документе. Чтобы эффективно управлять выполнением задания и отслеживать его ход, мы можем использовать удобные SDK и модули Python, такие как amazon-text-caller. Этот модуль Python, доступный на PyPi, создан на основе AWS SDK boto3 и обеспечивает упрощенный подход к эффективному управлению заданиями Textract.
...
from textractcaller.t_call import call_textract, Textract_Features
def start_textract_async(bucket_name, object_key):
try:
logger.info("Start textract job")
if len(object_key.split("/")) == 5:
output_path = "/".join(object_key.split("/")[:len(object_key.split("/")) - 3]) + "/output"
else:
output_path = "/".join(object_key.split("/")[:len(object_key.split("/")) - 2]) + "/output"
logger.info("Textract output: {}".format(output_path))
response = call_textract(
input_document="s3://{}/{}".format(bucket_name, object_key),
features = [Textract_Features.TABLES],
force_async_api=True,
return_job_id=True)
return response
except Exception as e:
stacktrace = traceback.format_exc()
logger.error("{}".format(stacktrace))
raise e
Для эффективной обработки больших документов рекомендуется инкапсулировать логику извлечения текста в специальную функцию AWS Lambda. Кроме того, мы можем использовать модуль Python amazon-texttract-textractor, чтобы упростить создание текстовых объектов с использованием вывода, сгенерированного Textract. Этот модуль упрощает процесс преобразования результатов Textract в легкодоступные объекты txt, повышая общую эффективность рабочего процесса извлечения текста.
...
from textractcaller.t_call import get_full_json, Textract_API
from trp import Document
def write_blocks(textract_resp, file_path):
try:
doc = Document(textract_resp)
page_number = 1
for page in doc.pages:
print("Page ", page_number)
text = ""
for line in page.lines:
text = text + " " + line.text
# Print tables
for table in page.tables:
text = text + "\n\n"
for r, row in enumerate(table.rows):
for c, cell in enumerate(row.cells):
print("Table[{}][{}] = {}".format(r, c, cell.text))
f = open("{}/output_{}.txt".format(file_path, page_number), "a")
f.write(text)
page_number += 1
except Exception as e:
stacktrace = traceback.format_exc()
print(stacktrace)
raise e
def extract_blocks(job_id, job_status, file_path):
if job_status == "SUCCEEDED":
blocks = get_full_json(
job_id,
textract_api=Textract_API.ANALYZE,
boto3_textract_client=textract_client
)
write_blocks(blocks, file_path)
Успешно преобразовав содержимое многостраничного PDF-документа в файлы txt, наша следующая цель — сгенерировать текстовые отрывки или сегменты из этих файлов на основе желаемого размера ввода, заданного выбранной моделью встраивания.
Для этой цели мы решили использовать модель GPT-J 6B, к которой легко получить доступ через Amazon SageMaker Jumpstart и HuggingFace.
Вы можете развернуть GPT-J 6B непосредственно из Amazon SageMaker Studio через SageMaker JumpStart или использовать блокнот, предоставленный в репозитории GitHub.
Процесс извлечения фрагментов из текстовых страниц можно автоматизировать с помощью фреймворков с открытым исходным кодом, таких как Langchain. Langchain предлагает такие модули, как RecursiveCharacterTextSplitter, которые очень помогают в создании сегментов. Эти модули также включают такие методы, как перекрытие, для повышения качества результирующих отрывков. Наши сегменты будут иметь длину 768 токенов, что является максимальным вводом, принимаемым GPT-J 6B. Вводя перекрытие, мы обеспечиваем непрерывность и уменьшаем потери информации между созданными проходами, тем самым улучшая общее качество вывода.
...
from langchain.text_splitter import RecursiveCharacterTextSplitter
def get_chunks(file_name, object_key, file_path):
try:
print("Get file chunks")
chunks = []
total_passages = 0
## Create image path for S3
#
if len(object_key.split("/")) == 5:
index_name = object_key.split("/")[3]
s3_path_pdf_images = f"{object_key.split('/')[0]}/{object_key.split('/')[1]}/images/{index_name}/{object_key.split('/')[-1]}"
else:
s3_path_pdf_images = f"{object_key.split('/')[0]}/{object_key.split('/')[1]}/images/{object_key.split('/')[-1]}"
for doc_name, page, doc in tqdm(doc_iterator(file_path)):
n_passages = 0
doc = re.sub(r"(\w)-\n(\w)", r"\1\2", doc)
doc = re.sub(r"(?<!\n)\n(?!\n)", " ", doc)
doc = re.sub(r"\n{2,}", "\n", doc)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""],
chunk_overlap=200,
)
tmp_chunks = text_splitter.split_text(doc)
for i, chunk in enumerate(tmp_chunks):
chunks.append({
"file_name": file_name,
"page": page,
"passage": chunk
})
n_passages += 1
total_passages += 1
logger.info(f'Document segmented into {n_passages} passages')
logger.info(f'Total passages to index: {total_passages}')
return chunks
except Exception as e:
stacktrace = traceback.format_exc()
logger.error("{}".format(stacktrace))
raise e
Теперь мы подходим к решающему этапу индексации этого объекта в подходящей векторной базе данных. В настоящее время существует несколько вариантов решений в памяти, таких как FAISS, который предлагает эффективную возможность поиска сходства для плотных векторов. Однако внедрение масштабируемых решений, включающих несколько документов, может быть затруднено. Кроме того, при наличии различных вариантов размещения вычислительного аспекта решений RAG, которые могут включать в себя службы без сохранения состояния, крайне важно поддерживать четкое разделение между вычислениями и хранилищем. Чтобы решить эту проблему, нам нужна выделенная база данных векторов, которая позволяет нам беспрепятственно работать с многочисленными документами, предоставляя необходимую инфраструктуру и возможности для эффективного индексирования и поиска.
В этом сценарии мы решили внедрить Amazon OpenSearch, распределенный поисковый и аналитический пакет с открытым исходным кодом, управляемый сообществом, который используется для широкого набора вариантов использования, таких как мониторинг приложений в реальном времени, анализ журналов и поиск по веб-сайтам. обеспечивая быстрый доступ и реакцию на большие объемы данных.
Чтобы настроить домен в OpenSearch, мы начинаем с доступа к консоли AWS и поиска OpenSearch. В консоли выберите «Разработка/тестирование» в качестве типа развертывания, «Домен без ожидания» в параметрах развертывания и выберите версию 7.10 в параметрах ядра. Выбор типа экземпляра будет зависеть от объема индексируемых документов. В нашем демонстрационном приложении мы выбрали тип экземпляра r6g.large.search с тремя узлами (количество главных узлов). Для этого конкретного типа экземпляра требуется хранилище EBS, и мы соответственно выделили том размером 10 ГБ.
Обязательно включите функцию «Автонастройка» для кластера, которая автоматически настраивает параметры на уровне узла без простоев. Это включает в себя настройку очередей и размеров кэша, оптимизацию производительности вашего кластера OpenSearch.
После того, как домен создан, мы можем приступить к настройке индекса. Прежде всего, первым шагом является определение подходящей схемы, соответствующей информации о документе, которую мы собираемся хранить.
mapping = {
'settings': {
'index': {
'knn': True # Enable KNN search for this index
}
},
'mappings': {
'properties': {
'embedding': { # KNN vector field
'type': 'knn_vector',
'dimension': 4096,
'similarity': 'cosine'
},
'file_name': {
'type': 'text'
},
'image': {
'type': 'text'
},
'page': {
'type': 'text'
},
'passage': {
'type': 'text'
}
}
}
}
Чтобы улучшить функциональность поиска и определить приоритетность наиболее релевантных документов, мы включили функцию поиска KNN (K-ближайших соседей). Используя поиск KNN, мы можем эффективно извлекать документы, которые похожи или тесно связаны с заданным запросом, улучшая общий исследовательский опыт и релевантность результатов.
...
import boto3
import requests
sagemaker_runtime_client = boto3.client('sagemaker-runtime')
def index_documents(url, chunks):
try:
logger.info("Indexing documents")
i = 1
for chunk in chunks:
payload = {'text_inputs': [chunk["passage"]]}
payload = json.dumps(payload).encode('utf-8')
response = sagemaker_runtime_client.invoke_endpoint(EndpointName=sagemaker_endpoint,
ContentType='application/json',
Body=payload)
model_predictions = json.loads(response['Body'].read())
embedding = model_predictions['embedding'][0]
document = {
'embedding': embedding,
'file_name': chunk["file_name"],
'image': chunk["image"],
'page': chunk["page"],
"passage": chunk["passage"]
}
response = requests.post(f'{url}/_doc/{i}', auth=HTTPBasicAuth(es_username, es_password), json=document)
i += 1
logger.info(response.text)
if response.status_code not in [200, 201]:
logger.info(response.status_code)
logger.info(response.text)
break
except Exception as e:
stacktrace = traceback.format_exc()
logger.error("{}".format(stacktrace))
raise e
Этап индексации состоит из следующих шагов:
- Генерировать embeddigs для каждого созданного чанка
- Определение документа JSON для Amazon OpenSearch
- Поместите документ в созданный индекс
Чтобы еще больше улучшить взаимодействие с пользователем, мы можем извлекать изображения с каждой страницы PDF-файла и сохранять их в корзине Amazon S3 для удобного поиска во внешнем приложении. Этого можно добиться, используя другую функцию AWS Lambda, оснащенную модулем Python pdf2image. Эта функция обеспечивает плавное преобразование страниц PDF в формат изображения, обеспечивая эффективное хранение и извлечение изображений во внешнем приложении.
...
from pdf2image import convert_from_path
def get_images(bucket_name, object_key):
try:
local_pdf_path = "/tmp/pdf_file"
output_jpeg_path = "/tmp/images"
if len(object_key.split("/")) == 5:
index_name = object_key.split("/")[3]
s3_path_pdf_images = f"{object_key.split('/')[0]}/{object_key.split('/')[1]}/images/{index_name}/{object_key.split('/')[-1]}"
else:
s3_path_pdf_images = f"{object_key.split('/')[0]}/{object_key.split('/')[1]}/images/{object_key.split('/')[-1]}"
if not os.path.exists(local_pdf_path):
os.makedirs(local_pdf_path)
if not os.path.exists(output_jpeg_path):
os.makedirs(output_jpeg_path)
local_path = os.path.join(local_pdf_path, 'temp.pdf')
s3_client.download_file(bucket_name, object_key, local_path)
images = convert_from_path(local_path)
results = {}
for i, image in enumerate(images):
image.save(os.path.join(output_jpeg_path, f"page_{i + 1}.jpeg"), format='JPEG')
s3_key = f"{s3_path_pdf_images}/page_{i + 1}.jpeg"
with open(os.path.join(output_jpeg_path, f"page_{i + 1}.jpeg"), "rb") as f:
s3_client.upload_fileobj(f, bucket_name, s3_key)
os.remove(os.path.join(output_jpeg_path, f"page_{i + 1}.jpeg"))
results[str(i + 1)] = s3_key
print(results)
return results
except Exception as e:
stacktrace = traceback.format_exc()
logger.error("{}".format(stacktrace))
raise e
В качестве дополнительного улучшения варианта использования мы также можем напрямую обрабатывать документы txt наряду с PDF-файлами. Этот процесс относительно проще по сравнению с общей обработкой PDF. Он включает в себя специальную функцию AWS Lambda, которая выполняет те же действия, что и обработка PDF, например создание фрагментов и индексирование отрывков в индексе Amazon OpenSearch. Расширяя эту функциональность на текстовые документы, мы можем еще больше улучшить общее взаимодействие с пользователем и расширить диапазон поддерживаемых типов документов.
Мы можем получить доступ к полному коду описанных функций AWS Lambda в репозитории GitHub.
Как упоминалось ранее, оркестровка и выполнение различных лямбда-функций AWS могут выполняться с помощью AWS Step Functions.
Мы можем определить структуру рабочего процесса в формате JSON через AWS CloudFormation следующим образом:
StepFunctionOrchestrator:
Type: AWS::StepFunctions::StateMachine
Properties:
StateMachineName: PDF-Parser-Orchestrator
RoleArn: !GetAtt StateMachineRole.Arn
DefinitionString:
!Sub |
{
"StartAt": "Lambda Detect Documents",
"States": {
"Lambda Detect Documents": {
"Type": "Task",
"Resource": "${LambdaDetectDocuments.Arn}",
"OutputPath": "$",
"Next": "Choice document type"
},
"Choice document type": {
"Type": "Choice",
"Choices": [
{
"Variable": "$.is_pdf",
"StringEquals": "true",
"Next": "Lambda Textract Job"
},
{
"Variable": "$.is_pdf",
"StringEquals": "false",
"Next": "Lambda Index Txt"
}
],
"Default": "DefaultState"
},
"Lambda Textract Job": {
"Type": "Task",
"Resource": "${LambdaTextractJob.Arn}",
"InputPath": "$",
"OutputPath": "$",
"Next": "Lambda Index Documents"
},
"Lambda Index Documents": {
"Type": "Task",
"Resource": "${LambdaIndexDocuments.Arn}",
"InputPath": "$",
"OutputPath": "$",
"Next": "Lambda Extract PDF Images"
},
"Lambda Extract PDF Images": {
"Type": "Task",
"Resource": "${LambdaExtractImages.Arn}",
"InputPath": "$",
"End": true
},
"Lambda Index Txt": {
"Type": "Task",
"Resource": "${LambdaIndexTxt.Arn}",
"InputPath": "$",
"End": true
},
"DefaultState": {
"Type": "Fail",
"Cause": "No Document Matches!"
}
}
}
Этот ресурс CloudFormation будет развертывать следующий рабочий процесс:
Качество модели вложений и генеративной модели
Модели Falcon являются одними из самых эффективных LLM в среде с открытым исходным кодом. Эти модели бывают двух размеров: 7B, обученные на токенах 1,5T, и 40B, обученные на токенах 1T.
Для этого примера мы решили использовать самую большую доступную модель, Falcon 40B, и использовать оптимизированный контейнер, доступный на Amazon SageMaker и недавно объявленный HuggingFace, для использования относительно небольшого экземпляра, доступного для хостинга SageMaker, ml.g5. экземпляр 12xlarge.
Все этапы развертывания действительно хорошо описаны в этом блоге.
В нашем примере мы предоставляем пример блокнота в репозитории GitHub, который может помочь нам в развертывании модели. Кроме того, мы предоставляем шаблон AWS CloudFormation для автоматизации развертывания всей инфраструктуры, включая модель встраивания GPT-J, а также функции AWS Lambda и AWS Step Functions, описанные в предыдущем абзаце.
Мы можем протестировать поведение развернутого LLM, вызвав конечную точку Amazon SageMaker непосредственно из ноутбука.
endpoint_name = "falcon-40b-endpoint"
payload = """
Hello, how are you?
"""
parameters = {
"inputs": payload,
"parameters": {
"max_new_tokens": 512,
"temperature": 0.2,
"top_p": 0.9,
}
}
print(json.dumps(parameters).encode("utf-8"))
results = sagemaker_runtime.invoke_endpoint(
EndpointName=endpoint_name,
ContentType='application/json',
Body=json.dumps(parameters).encode("utf-8"))
response = json.loads(results["Body"].read())
print(response)
Полная оценка модели Falcon 40B в различных областях доступна на HuggingFace.
Синергетическая интеграция поиска информации и генерации текста
Теперь пришло время соединить все элементы вместе. Мы успешно создали всеобъемлющий конвейер обработки данных, который позволяет нам индексировать различные типы документов. С помощью Falcon 40b-instruct мы также развернули мощную генеративную базовую модель, способную генерировать текст различными способами в соответствии с инструкциями. Последний шаг включает в себя создание масштабируемого серверного решения, которое может адаптироваться к входящим запросам, а также удобный интерфейс для беспрепятственного взаимодействия с нашими решениями.
Для серверной части мы собираемся использовать Langchain и принять одну из цепочек, доступных для реализации решения Q&A RAG, ConversationalRetrievalChain.
Интеграция с LLM и моделью Embedding осуществляется через коннекторы Langchain SageMakerEndpoint и SageMakerEndpointEmbeddings.
Первый шаг — правильно определить классы Python для обработки как ввода, так и вывода, требуемых двумя моделями ML.
class FalconHandler(LLMContentHandler):
content_type = "application/json"
accepts = "application/json"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def transform_input(self, prompt: str, model_kwargs: dict) -> bytes:
self.len_prompt = len(prompt)
input_str = json.dumps({"inputs": prompt,
"parameters": model_kwargs})
return input_str.encode('utf-8')
def transform_output(self, output: bytes) -> str:
response_json = output.read()
res = json.loads(response_json)
ans = res[0]['generated_text'][self.len_prompt:]
ans = ans[:ans.rfind("Human")].strip()
return ans
class GPTJHandler(EmbeddingsContentHandler):
content_type = "application/json"
accepts = "application/json"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def transform_input(self, prompt: str, model_kwargs: dict) -> bytes:
input_str = {'text_inputs': prompt, **model_kwargs}
return json.dumps(input_str).encode('utf-8')
def transform_output(self, output: bytes) -> str:
results = output.read().decode("utf-8")
response = json.loads(results)
return response["embedding"]
Эти два класса будут предоставлены в качестве аргумента для SageMakerEndpoint и SageMakerEndpointEmbeddings в качестве обработчиков содержимого.
self.embeddings = SagemakerEndpointEmbeddings(
endpoint_name=embedding_endpoint_name,
region_name=region,
content_handler=GPTJHandler()
)
self.llm = SagemakerEndpoint(
endpoint_name=llm_endpoint_name,
region_name=region,
model_kwargs=llm_model_kwargs,
content_handler=FalconHandler()
)
Ссылаясь на нашу эталонную архитектуру, начальный шаг нашего рабочего процесса включает преобразование пользовательского ввода в вектор внедрения. Затем мы приступаем к поиску наиболее релевантных документов в индексе Amazon OpenSearch, что позволяет нам извлекать отрывки, которые служат дополнительным контекстом для нашего генеративного FM при ответе на вопрос. Этот процесс обогащает входные данные и повышает качество выходных данных LLM.
Для этого мы используем OpenSearchVectorSearch и расширенную версию BaseRetriever, предоставленную Langchain, которую мы назвали DocumentRetrieverExtended.
class DocumentRetrieverExtended(BaseRetriever):
def __init__(self, retriever, vector_field, text_field, k=3, return_source_documents=False, score_threshold=None, **kwargs):
self.k = k
self.vector_field = vector_field
self.text_field = text_field
self.return_source_documents = return_source_documents
self.retriever = retriever
self.filter = filter
self.score_threshold = score_threshold
self.kwargs = kwargs
def get_relevant_documents(self, query: str) -> List[Document]:
results = []
docs = self.retriever.similarity_search_with_score(query, k=self.k, vector_field=self.vector_field, text_field=self.text_field, **self.kwargs)
if docs:
for doc in docs:
metadata = doc[0].metadata
metadata["score"] = doc[1]
if self.score_threshold is None or \
(self.score_threshold is not None and metadata["score"] >= self.score_threshold):
results.append(Document(
page_content=doc[0].page_content,
metadata=metadata
))
return results
async def aget_relevant_documents(self, query: str) -> List[Document]:
return await super().aget_relevant_documents(query)
BaseRetriever расширен для использования метода similarity_search_with_score, изначально предоставленного OpenSearchVectorSearch. Мы выполняем этот шаг, чтобы вернуть оценку пользователю через интерфейс как указание на актуальность исходного документа в контексте ответа.
self.vector_search = OpenSearchVectorSearch(
opensearch_url=config["es_credentials"]["endpoint"],
index_name=config["es_credentials"]["index"],
embedding_function=self.embeddings,
http_auth=(config["es_credentials"]["username"], config["es_credentials"]["password"])
)
self.retriever = DocumentRetrieverExtended(
self.vector_search,
"embedding",
"passage",
k=config["llms"][self.llm_endpoint]["query_results"],
)
Последним шагом перед созданием цепочки является определение правильного шаблона приглашения для обработки дополнительного контекста и создание правильного ответа с помощью LLM.
В этом блоге мы не будем углубляться в то, как эффективно определить шаблон приглашения для LLM, но вы можете легко следовать руководствам и учебным пособиям, доступным в Интернете, например, представленным на странице Langchain.
falcon_template = """
Use the following pieces of context to answer the question at the end. You must not answer a question not related to the documents.
If you don't know the answer, just say "Unfortunately, I can't help you with that", don't try to make up an answer.
{chat_history}
{context}
Question: {question}
Detailed Answer:
"""
PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question", "chat_history"]
)
В этом шаблоне мы определяем инструкцию для LLM, относящуюся к ее начальной части, предоставляя контекст в виде chat_history и контекста. При этом мы обеспечиваем нашу модель контекстуализированной информацией из базы знаний, а также возможностью запоминать предыдущие ходы разговора в форме чатистории, наряду с вопросами или вопросами пользователей. Все это упаковано в подсказку с инструкциями, адаптированную к модели, которую мы используем.
Сообщения, обрабатываемые во время сеанса, могут быть предоставлены в качестве входных данных для LLM через ConversationBufferWindowMemory, предоставленную Langchain.
self.memory = ConversationBufferWindowMemory(
k=config["llms"][self.llm_endpoint]["memory_window"],
chat_memory=history,
memory_key="chat_history",
return_messages=True)
На данный момент у нас есть все элементы для создания цепочки.
qa = ConversationalRetrievalChain.from_llm(
llm=self.llm,
retriever=self.retriever,
combine_docs_chain_kwargs={"prompt": PROMPT},
return_source_documents=True,
verbose=True,
memory=self.memory
)
Последний шаг — определить обработчик AWS Lambda для создания цепочки и сгенерировать ответ на основе пользовательского ввода.
def lambda_handler(event, context):
try:
config = read_configs(bucket, config_file)
logger.info(event)
user = event["user"]
question = event["question"]
chat_memory = event["chat_memory"]
llm_endpoint = event["llm_endpoint"]
embeddings_endpoint = event["embeddings_endpoint"]
selected_type = event["selected_type"]
history = ChatMessageHistory()
for message in chat_memory:
history.add_user_message(message[0])
history.add_ai_message(message[1])
if user != "":
config["es_credentials"]["index"] = config["es_credentials"]["index"] + "-" + user
if selected_type == "Chat Q&A":
chain = ChatQAChain(embeddings_endpoint, llm_endpoint)
elif selected_type == "Chatbot":
chain = ChatbotChain(embeddings_endpoint, llm_endpoint)
else:
chain = SearchQAChain(embeddings_endpoint, llm_endpoint)
qa = chain.build(config, history)
sources = []
answer = qa({"question": question, "chat_history": chat_memory})
if len(answer.get("source_documents", [])) > 0:
for el in answer.get("source_documents"):
sources.append({
"image": el.metadata["image"] if "image" in el.metadata else "",
"details": f'Document = {el.metadata["file_name"]} | Page = {el.metadata["page"]} | Score = {el.metadata["score"]}',
"passage": (el.page_content[:300] + '..') if len(el.page_content) > 300 else el.page_content
})
if len(sources) == 3:
break
if "answer" not in answer and "text" in answer:
answer["answer"] = answer["text"]
return {
'statusCode': 200,
'body': json.dumps(
{
"answer": answer.get("answer").strip(),
"sources": sources
}
)
}
except Exception as e:
stacktrace = traceback.format_exc()
logger.error("{}".format(stacktrace))
raise e
Входные параметры, необходимые для функции AWS Lambda:
- пользователь: пользователь, который в данный момент взаимодействует с решением
- вопрос: пользовательский ввод, предоставленный в разговоре
- chat_memory: исторические сообщения
- llm_endpoint: конечная точка, используемая для LLM.
- embedding_endpoint: конечная точка, используемая для модели внедрения
- selected_type: тип варианта использования, который вы хотите протестировать во внешнем приложении.
Это решение дает нам возможность протестировать несколько вариантов использования, таких как чат-бот и чат вопросов и ответов. По этой причине полный код функции AWS Lambda предоставляет несколько вариантов цепочки, которые можно использовать благодаря параметру selected_type.
Кроме того, это решение было создано, чтобы дать нам возможность выбирать различные конечные точки, размещенные на Amazon SageMaker, которые можно включить в цепочку с помощью параметров llm_endpoint и embedding_endpoint после определения правильный обработчик содержимого, необходимый для обработки ввода и вывода моделей ML.
Мы можем легко выбрать другую генеративную модель из Amazon SageMaker JumpStart и развернуть ее через конечную точку SageMaker.
Чтобы использовать эту конечную точку в нашем решении, мы должны внести два изменения:
- приложение: мы можем отредактировать файл configs.yaml для внешнего приложения, добавить конечную точку в раздел llms, чтобы сделать ее доступной для выбора из пользовательского интерфейса.
- бэкенд:
- мы можем отредактировать файл configs.yaml для внутреннего приложения, добавить конечную точку в раздел llms, добавив все необходимые параметры, которые должны быть предоставлены модели во время вызова
- мы можем отредактировать файл lambda_function.py, чтобы добавить новый обработчик контента для обработки как ввода, так и вывода, требуемых новой моделью ML.
Вы можете получить доступ к полному коду лямбда-функции в репозитории GitHub.
Пользовательский интерфейс
Взаимодействие с нашим бэкенд-решением может осуществляться через удобный интерфейс, реализованный через Streamlit, фреймворк Python с открытым исходным кодом, упрощающий процесс создания интерактивных веб-приложений для задач обработки данных и машинного обучения.
Наше приложение Streamlit предоставляет два разных интерфейса: расширенный чат вопросов и ответов, который предоставляет дополнительную информацию вместе с ответом, сгенерированным LLM, и классический чат-бот.
Мы можем легко использовать различные документы в качестве базы знаний для нашего приложения GenAI, загружая их в корзину Amazon S3 через пользовательский интерфейс. Эти документы будут проиндексированы и сохранены в Amazon OpenSearch посредством выполнения AWS Step Functions, описанного в предыдущих абзацах.
Вы можете получить доступ к полному коду фронтенд-приложения в репозитории GitHub.
Выводы
В этом блоге представлено пошаговое руководство по созданию масштабируемого приложения GenAI для вопросов и ответов с помощью RAG с использованием В качестве бонуса мы включаем код для развертывания внешнего приложения на AWS с помощью AWS Fargate.
Если вы хотите протестировать блокноты и полный код всех компонентов, описанных в этом блоге, загляните в репозиторий GitHub.
Это всего лишь отправная точка для создания более сложных и специализированных приложений. с этими продвинутыми языковыми моделями.
Одним из способов значительно улучшить это приложение было бы сохранение разговоров для каждого пользовательского сеанса в соответствующей базе данных и использование одного из модулей Langchain для динамической загрузки разговоров из истории и предоставления их пользователю. цепь.
