문서 인텔리전스 활용하기: 벡터 기반 검색이 포함된 E2E Azure 기반 챗봇(1부 - 임베딩)
위에 제공된 정보는 2024년 2월 29일 기준으로 업데이트되었으며 편집팀에 의해 최종 수정되었습니다.
저자: Shravankumar Hiregoudar
원래 이 사이트에 게시되었습니다.
현대의 조직 환경에서는 빠른 업무 속도로 인해 포괄적인 정보에 신속하게 액세스하는 것이 무엇보다 중요합니다. 상당수의 기업은 텍스트 콘텐츠와 시각적 미디어 등 다양한 형식을 포괄하는 정리되지 않은 방대한 양의 데이터에 직면하고 있으며, 이로부터 관련 세부 정보를 추출할 수 있는 능력이 필요합니다. 또한 자연어 인터페이스를 통합하면 사용자가 대화 방식으로 해당 정보를 쉽게 조사할 수 있어 효율성과 전반적인 만족도를 모두 높일 수 있습니다.
사진 Paul Melki on Unsplash
목차:
서론 문제 진술 솔루션 개요
3.1 아키텍처
3.2 임베딩 파이프라인 구현 하이라이트 임베딩 구현
5.1 프로젝트 레이아웃
5.2 전제 조건
5.3 프로젝트 파일 다음 파트
서론
기술 개발의 이 단계에서는 신속하고 정확한 검색 작업을 촉진하는 초석이 되는 벡터 형태의 문서 표현 및 보존의 핵심 요소에 초점을 맞춥니다. 그 후, 언어적 탐구를 통해 이 벡터 저장소를 조사하는 것과 관련된 복잡성을 탐구하여 궁극적으로 문서 처리 프로세스의 완전한 역량을 실현할 것입니다.
파트 (출처: 저자)
문제 설명
PDF, Microsoft Word 파일, PowerPoint 프레젠테이션 및 그래픽 이미지를 포함한 광범위한 문서 유형의 관리와 관련된 기업 활동을 감독하는 시나리오를 상상해 보십시오. 목표는 다음을 수행할 수 있는 포괄적인 솔루션을 개발하는 것입니다:
표, 키-값 쌍과 같은 다양한 문서 형식에서 관련 정보를 추출하는 프로세스를 자동화하는 것은 데이터를 효율적으로 정리하는 데 매우 중요합니다. 이러한 목적을 위해 특별히 설계된 고급 알고리즘을 활용하면 이를 달성할 수 있습니다.이렇게 추출된 세부 정보는 쉽게 액세스할 수 있는 형식으로 저장되어 필요할 때 빠르게 검색할 수 있습니다. 또한 대화형 인터페이스를 통합하여 사용자와 시스템 간의 상호 작용을 용이하게 함으로써 사용자가 질문을 하고 정확한 답변을 즉시 얻을 수 있습니다.
솔루션 개요
아키텍처
아키텍처(출처: 작성자)
임베딩 파이프라인
임베딩 파이프라인은 Azure Form Recognizer를 활용하여 처리하고 최종적으로 추출된 데이터를 Azure Search Index 내에서 검색에 최적화된 형식으로 저장하여 Azure Blob Storage에 저장된 정보를 검색하도록 설계되어 있습니다. 다양한 단계와 프로세스를 이해하기 위해
(출처: Microsoft) 문서 검색을 개별적으로 살펴보겠습니다: 이 코드는 먼저 문서가 저장되어 있는 Azure Storage 컨테이너에 연결합니다. 컨테이너에 있는 문서를 반복하여 특정 파일 형식(이 사용 사례에서는 PDF 파일만 추출합니다. .docx, .pptx 등을 구현하려는 경우 양식 인식기가 다른 Microsoft 데이터 파일을 지원하지 않으므로 파일을 .pdf로 변환하는 것이 좋습니다) 및 문서가 이미 처리되었는지 여부를 확인합니다. ( 블롭서비스클라이언트 클래스 ) 텍스트 추출: 문서가 아직 임베드되지 않은 경우(임베드 메타데이터를 추적하기 위해 데이터베이스를 유지함), 코드에서는 Azure 양식 인식기 을 사용하여 문서에서 텍스트, 표 및 키 값 정보를 추출합니다. 텍스트 청킹: 추출된 텍스트는 효율적인 처리를 위해 더 작은 청크로 분할됩니다. ( 랭체인 텍스트 스플리터 ) 임베딩: 이 코드는 OpenAI의 “text-embedding-ada-002” 모델을 사용하여 텍스트 청크에 대한 임베딩을 생성합니다. 이러한 임베딩은 텍스트 콘텐츠를 나타내며 나중에 검색할 때 사용됩니다. (NLP의 맥락에서 임베딩은 비슷한 의미를 가진 단어나 구가 비슷한 표현을 가질 수 있도록 하는 텍스트의 숫자 표현입니다. Azure OpenAI 임베딩 모델은 이러한 임베딩을 만듭니다. 텍스트에 대한 임베딩 벡터를 얻으려면 임베딩 엔드포인트를 요청합니다.) 벡터 저장소: 임베딩은 Azure Cognitive Search 에 저장되어 문서를 빠르고 확장 가능하게 쿼리할 수 있습니다. (임베딩을 만든 후에는 벡터 저장소(예: Pinecone, Chroma 및 Faiss)에 저장할 수 있습니다.우리가 사용하는 벡터 저장소는 대량의 고차원 벡터를 효율적으로 검색하고 비교할 수 있는 메모리 효율적인 방법인 Azure 검색의 인덱스입니다. 이러한 벡터 표현은 언어에 구애받지 않고 다양한 데이터 유형을 처리할 수 있으므로 여러 애플리케이션에 다양하게 활용할 수 있습니다.)
임베딩 파이프라인은 텍스트 데이터와 다큐멘터리 데이터의 처리, 임베딩, 접근성을 간소화된 방식으로 효과적으로 처리하도록 설계되었습니다.
구현 하이라이트
이 코드는 상당한 양의 Python 프로그래밍 언어를 사용하며 다음과 같은 여러 라이브러리 및 서비스를 활용하여 기능을 원활하게 수행합니다:
이 시스템의 구현에는 문서 액세스 및 저장을 위한 Azure Blob Storage, 텍스트 및 문서 데이터 추출을 위한 Azure Form Recognizer, 문서 임베딩을 저장하고 효율적인 검색을 위한 Azure Cognitive Search, OpenAI의 GPT-3 등 다양한 소스에서 제공하는 몇 가지 주요 구성 요소가 포함되어 있습니다.5 사용자 문의에 대한 응답을 생성하는 Turbo, 임베딩 생성을 위한 OpenAI의 텍스트 임베딩-ada-002, 텍스트 구문 분석, 텍스트 임베딩 생성, Azure Cognitive Search와의 협업, OpenAI 채팅 시스템과의 상호 작용 등 자연어 처리 및 문서 검색을 위한 기본 기능을 제공하는 Langchain, 마지막으로 직관적인 채팅 인터페이스 구축을 위한 Streamlit이 사용되었습니다.
임베딩 구현
처음부터 끝까지 완전한 문서 처리 프로세스 파이프라인을 보여주는 Python 프로그램이 제공됩니다.
상세 임베딩 파이프라인(출처: 작성자)
프로젝트 레이아웃
# Project Layout
.
├── README.md
├── requirements.txt
├── app.py
├── .gitignore
├── .env
├── db/
│ ├── metadata.db
└── src/
├── database_manager.py
전제 조건
코드를 실행하기 전에 필요한 전제 조건이 제대로 설정 및 구성되었는지 확인하여 필요한 전제 조건이 갖추어져 있는지 확인합니다. 여기에는 필요한 소프트웨어 패키지 또는 라이브러리 설치, 환경 변수 설정, 프로그램 내에서 필요한 데이터를 사용할 수 있는지 여부 등이 포함될 수 있습니다. 이러한 단계를 수행하면 코드가 원활하고 성공적으로 실행되도록 도울 수 있습니다.
Azure 서비스: Azure Blob 저장소, Azure 양식 인식기 및 Azure 인지 검색 서비스를 만듭니다. OpenAI API: 채팅 모델 및 임베딩을 사용하려면 OpenAI API에 액세스해야 합니다. Python 종속성: pip install -r requirements.txt 을 실행하여 필요한 Python 패키지를 설치합니다.환경 변수: 다음 환경 변수를 사용하여 스크립트와 동일한 디렉터리에 .env 파일을 만듭니다:
# Azure Cognitive Search configurations
AZURE_COGNITIVE_SEARCH_SERVICE_URL=<your_search_service_url>
AZURE_COGNITIVE_SEARCH_SERVICE_NAME=<your_search_service_name>
AZURE_COGNITIVE_SEARCH_API_KEY=<your_search_service_api_key># Azure Storage configurations
AZURE_STORAGE_CONNECTION_STRING=<your_storage_connection_string>
AZURE_CONTAINER_NAME=<your_container_name># Azure AI Document Intelligence (Form Recognizer) configurations
FORM_RECOGNIZER_ENDPOINT=<your_form_recognizer_endpoint>
FORM_RECOGNIZER_KEY=<your_form_recognizer_key># OpenAI Chat API configurations
OPENAI_API_TYPE=<your_openai_api_type>
OPENAI_API_BASE=<your_openai_api_base_url>
OPENAI_API_KEY=<your_openai_api_key>
OPENAI_API_VERSION=<your_openai_api_version># OpenAI Embedding API configurations
EMBED_API_BASE=<your_openai_embedding_api_base_url>
EMBED_API_KEY=<your_openai_embedding_api_key>
EMBED_API_VERSION=<your_openai_embedding_api_version>
Azure 서비스 연결 확인은 다양한 Azure 서비스와 관련된 API의 연결을 확인하는 프로세스입니다. 이를 통해 이러한 서비스와 다른 구성 요소 간의 통신 문제 또는 중단을 즉시 식별하고 해결하여 시스템 전체에서 원활한 작동을 유지할 수 있습니다.
프로젝트 파일
코드의 아키텍처 및 기능을 자세히 살펴볼 수 있도록 허용해 주세요.
requirements.txt
# Project Layout
.
├── README.md
├── requirements.txt
├── app.py
├── .gitignore
├── .env
├── db/
│ ├── metadata.db
└── src/
├── database_manager.py# requiremnt.txt
streamlit==1.0.0
python-dotenv==0.19.1
langchain==0.0.276
azure-ai-formrecognizer==3.3.0
azure-common==1.1.28
azure-core==1.29.4
azure-identity==1.14.0
azure-search-documents==11.4.0b8
azure-storage-blob==12.18.2
database_manager.py
# Project Layout
.
├── README.md
├── requirements.txt
├── app.py
├── .gitignore
├── .env
├── db/
│ ├── metadata.db
└── src/
├── database_manager.py
database_manager.py 는 SQLite 데이터베이스 (메타데이터) 을 관리하기 위한 것입니다. db) 임베딩 파이프라인에서 처리된 문서와 같은 임베딩된 블롭에 대한 메타데이터를 저장하는 ((임베딩메타데이터) 테이블)
임베딩 파이프라인 내에서 처리된 문서의 현재 상태와 배경 정보를 모니터링하여 문서의 존재 증거와 가장 최근에 임베딩된 시간을 보존하는 역할을 수행합니다.
import sqlite3
from datetime import datetime
class DatabaseManager:def __init__(self, db_file):
"""
Initialize the DatabaseManager.
:param db_file: The name of the SQLite database file.
"""
self.db_file = db_file
self.conn = sqlite3.connect(self.db_file)
self.cursor = self.conn.cursor()
self.create_database_table()def create_database_table(self):
"""
Create the 'EmbeddingMetadata' table in the database if it doesn't exist.
This table stores metadata about embedded blobs.
:return: None
"""
self.cursor.execute('''CREATE TABLE IF NOT EXISTS EmbeddingMetadata
(BlobName TEXT PRIMARY KEY, ExistsInBlob TEXT, ExistsInSearch TEXT, LatestEmbeddedOn DATE)''')
self.conn.commit()def record_exists_in_database(self, blob_name):
"""
Check if a record with the given `blob_name` exists in the database.
:param blob_name: The name of the blob to check.
:return: True if the record exists, False otherwise.
"""
self.cursor.execute("SELECT COUNT(*) FROM EmbeddingMetadata WHERE BlobName=?", (blob_name,))
count = self.cursor.fetchone()[0]
return count > 0def insert_record_to_database(self, blob_name, exists_in_blob, exists_in_search):
"""
Insert a new record into the 'EmbeddingMetadata' table.
:param blob_name: The name of the blob to insert.
:param exists_in_blob: A flag indicating if the blob exists.
:param exists_in_search: A flag indicating if the blob exists in search.
:return: None
"""
latest_embedded_on = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if not self.record_exists_in_database(blob_name):
self.cursor.execute("INSERT INTO EmbeddingMetadata (BlobName, ExistsInBlob, ExistsInSearch, LatestEmbeddedOn) VALUES (?, ?, ?, ?)",
(blob_name, exists_in_blob, exists_in_search, latest_embedded_on))
self.conn.commit()# Project Layout
.
├── README.md
├── requirements.txt
├── app.py
├── .gitignore
├── .env
├── db/
│ ├── metadata.db
└── src/
├── database_manager.py
필요한 라이브러리를 가져오고 .env 파일에서 모든 환경 변수를 검색하세요.
# app.py
import os
import logging
import streamlit as st
import openai
import json
from dotenv import load_dotenvfrom azure.ai.formrecognizer import FormRecognizerClient
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores.azuresearch import AzureSearch
from langchain.chat_models import AzureChatOpenAI
from langchain.vectorstores.base import Document
from langchain.chains import RetrievalQA, RetrievalQAWithSourcesChain
from langchain.retrievers import AzureCognitiveSearchRetrieverfrom src.database_manager import DatabaseManagerload_dotenv()# Configure the logger
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Disable logging for azure.core.pipeline.policies.http_logging_policy
logging.getLogger("azure.core.pipeline.policies.http_logging_policy").setLevel(
logging.WARNING)# Azure Search configurations
AZURE_SEARCH_URL = os.getenv("AZURE_COGNITIVE_SEARCH_SERVICE_URL")
AZURE_SEARCH_NAME = os.getenv("AZURE_COGNITIVE_SEARCH_SERVICE_NAME")
AZURE_SEARCH_KEY = os.getenv("AZURE_COGNITIVE_SEARCH_API_KEY")# Azure Storage configurations
STORAGE_CONTAINER_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
STORAGE_CONTAINER_NAME = os.getenv("AZURE_CONTAINER_NAME")# Form Recognizer configurations
FORM_RECOGNIZER_ENDPOINT = os.getenv("FORM_RECOGNIZER_ENDPOINT")
FORM_RECOGNIZER_KEY = os.getenv("FORM_RECOGNIZER_KEY")# Azure OpenAI Chat Completion configurations
OPENAI_API_TYPE = os.getenv('OPENAI_API_TYPE')
CHAT_API_BASE = os.getenv('OPENAI_API_BASE')
CHAT_API_KEY = os.getenv('OPENAI_API_KEY')
CHAT_API_VERSION = os.getenv('OPENAI_API_VERSION')# Azure OpenAI Embedding configurations
EMBED_API_BASE = os.getenv('EMBED_API_BASE')
EMBED_API_KEY = os.getenv('EMBED_API_KEY')
EMBED_API_VERSION = os.getenv('EMBED_API_VERSION')# Default items
DEFAULT_EMBEDDING_MODEL = "text-embedding-ada-002"
DEFAULT_CHAT_MODEL = "gpt-35-turbo"
DEFAULT_SEARCH_INDEX = "<your-index>"
DEFAULT_SEARCH_FILE_EXTENSION = ".pdf"
LOAD_VECTORS = True
임베딩 파이프라인 클래스는 app.py 에 존재하며 문서에서 텍스트를 추출하고, 텍스트를 청크로 분할하고, 검색을 위해 텍스트 임베딩을 저장하는 등 문서 임베딩 프로세스를 담당합니다.
# app.py
class EmbeddingPipeline:
def __init__(self):
load_dotenv()
self.form_recognizer_client = self.get_form_recognizer_client()
self.db_manager = DatabaseManager("db/metadata.db")
self.embedder = OpenAIEmbeddings(model=DEFAULT_EMBEDDING_MODEL,
openai_api_base=EMBED_API_BASE,
openai_api_key=EMBED_API_KEY,
openai_api_version=EMBED_API_VERSION,
openai_api_type=OPENAI_API_TYPE)def get_form_recognizer_client(self):
"""
Get an instance of the Form Recognizer Client.Returns:
FormRecognizerClient: An instance of the FormRecognizerClient class.
"""
credential = AzureKeyCredential(FORM_RECOGNIZER_KEY)
return FormRecognizerClient(endpoint=FORM_RECOGNIZER_ENDPOINT, credential=credential)def form_recognizer_data_extract(self, blob_content):
"""
Azure Form Recogniser extracts text from the PDF files loaded from the container.
NOTE: You can process other data files (.docx, .pptx etc) by manually converting them to .pdf as form recognizer doesnt support all the microsoft file types.Args:
blob_content (bytes): The content of the blob to extract data from.Returns:
tuple: A tuple containing:
- list of dictionaries: Extracted table data.
- list of dictionaries: Extracted line data.
- str: Extracted text data.
"""
table_data = []
line_data = []
text = ""try:
form_recognizer_result = self.form_recognizer_client.begin_recognize_content(
blob_content).result()for page in form_recognizer_result:
for table in page.tables:
table_info = {"table_cells": []}
for cell in table.cells:
cell_info = {
"text": cell.text,
"bounding_box": cell.bounding_box,
"column_index": cell.column_index,
"row_index": cell.row_index
}
table_info["table_cells"].append(cell_info)
table_data.append(table_info)
for line in page.lines:
text \\+= " ".join([word.text for word in line.words]) \\+ "\n"
line_info = {
"text": line.text,
"bounding_box": line.bounding_box
}
line_data.append(line_info)logger.info(
"\t\tStep 3: Azure Form Recognizer - Extracted text from the file/s loaded from the container")return table_data, line_data, textexcept Exception as e:
logger.warning(
f"\t\tStep 3 (ERROR): Azure Form Recognizer - An error occurred while extracting form data: {e}")
return [], [], []def get_text_chunks(self, text, blob_name):
"""
Split a large text into smaller chunks for further processing.Args:
text (str): The text to be split into chunks.Returns:
list of Document: List of Document objects representing text chunks.
"""
text_splitter = CharacterTextSplitter(
separator="\n",
chunk_size=1000,
chunk_overlap=200,
length_function=len,
is_separator_regex=False
)
chunks = text_splitter.split_text(text)
docs = [Document(page_content=chunk, metadata = {"source":blob_name}) for chunk in chunks]The process of pre-embedding involves dividing a file into numerous smaller segments for enhancing its security by reducing the risk of unauthorized access to sensitive information, as confirmed by the human expert in the field.return docsdef load_vectorstore(self, documents):
"""
Azure OpenAI "text-embedding-ada-002" model prepare embeddings to the chunked files and upload vectors into Azure Cognitive Search Index.Args:
documents (list of dict): List of documents to be added to Azure Cognitive Search.Returns:
AzureSearch: An instance of AzureSearch containing the loaded vectors.
"""
try:
vectorstore = AzureSearch(
azure_search_endpoint=AZURE_SEARCH_URL,
azure_search_key=AZURE_SEARCH_KEY,
index_name=DEFAULT_SEARCH_INDEX,
embedding_function=self.embedder.embed_query,
)
vectorstore.add_documents(documents=documents)
logger.info(
f"\t\tStep 5: Azure Cognitive Search - Embeddings are created and vectors are stored in Azure Search index: '{DEFAULT_SEARCH_INDEX}'")except openai.error.APIError as api_error:
logger.error(
f"\t\tStep 5 (ERROR): Azure Cognitive Search - Error: {api_error}")def perform_embedding_pipeline(self):
"""
Process documents in an Azure Storage container and perform an embedding pipeline on them.This function retrieves documents stored in an Azure Storage container specified by 'container_name'
and processes each document. It checks if the document's content type matches a predefined extension
(e.g., '.pdf') and, if so, extracts data using Form Recognizer, processes the extracted data,
and loads it into a vector store.Parameters:
storage_connection_string (str): The connection string for the Azure Storage account where the
container is located.
container_name (str): The name of the Azure Storage container containing the documents to process.
"""
logger.info( f"__NOTE__ Processing only {DEFAULT_SEARCH_FILE_EXTENSION} types")try:
blob_service_client = BlobServiceClient.from_connection_string(STORAGE_CONTAINER_STRING)
blob_container_client = blob_service_client.get_container_client(STORAGE_CONTAINER_NAME)
exists, inserted = 0, 0
for blob in blob_container_client.list_blobs():
if blob.name.endswith(DEFAULT_SEARCH_FILE_EXTENSION) and not self.db_manager.record_exists_in_database(blob.name):
blob_client = blob_service_client.get_blob_client(container = STORAGE_CONTAINER_NAME, blob = blob.name)
blob_content = blob_client.download_blob().readall()
logger.info( f"\tProcessing Document '{blob.name}' : ")
logger.info( f"\t\tStep 2: Azure Storage Container - Blob content fetched successfully")# only using 'raw_text' as of now
table_data, line_data, raw_text = self.form_recognizer_data_extract(blob_content)
documents = self.get_text_chunks(raw_text, blob.name)
self.load_vectorstore(documents)
self.db_manager.insert_record_to_database(blob.name, 'Y', 'Y')
inserted \\+=1elif blob.name.endswith(DEFAULT_SEARCH_FILE_EXTENSION) and self.db_manager.record_exists_in_database(blob.name):
exists \\+=1logger.info(f"Embedding Summary : Processed {inserted} new file(s) out of {exists \\+ inserted} total file(s) in the container. Stored vectors for {inserted} new file(s); as {exists} file(s) already had vectors.")
except Exception as e:
print(e)class ChatPipeline:
# ........ (refer to part 2)def main():
# ........ (refer to part 2)if __name__ == "__main__":
main()
이번 기술 탐사 단계에서는 문서 임베딩의 핵심 요소인 벡터 형식의 보존과 이를 통해 신속하고 능숙한 검색 작업을 촉진하는 강력한 기반이 되는 문서 임베딩에 대해 살펴봅니다. 다음 단계에서는 이 벡터 저장소에 액세스하기 위해 자연어 쿼리를 생성하는 것과 관련된 복잡성을 밝혀내고, 이를 통해 문서 처리 시스템의 완전한 기능을 실현할 것입니다.
다양한 영역에서 인공지능의 최첨단 발전 동향을 파악하는 80,000명 이상의 구독자로 구성된 독점 AI 뉴스레터 커뮤니티에 오신 것을 환영합니다. AI 스타트업의 일원이든, AI 기반 제품을 개발하든, 이 분야와 관련된 서비스를 제공하든, 스폰서는 우리의 사명을 지원하는 데 중요한 역할을 합니다. 여러분도 함께해 주시면 기쁩니다.
이 사이트를 통해 게시됨