수많은 AI 기업들이 이제 복잡한 SQL 쿼리나 수동 스프레드시트 분석 없이도 데이터를 즉시 분석하고 포괄적인 인사이트를 제공하는 에이전트 기반 비즈니스 인텔리전스 시스템을 도입하고 있습니다. 예를 들어, RAG 기반 영업 분석 시스템은 “지난 분기 최고 판매 제품은 무엇인가요?”와 같은 질문에 몇 시간이 아닌 몇 초 만에 답변할 수 있습니다.
RAG 기반 영업 분석 시스템은 AI 에이전트를 사용하여 이러한 문제를 해결합니다. AI 에이전트는 자율적으로 추론하고, 결정을 내리며, 외부 도구를 조정하여 전체 데이터 생태계에 걸쳐 자연어 질의를 가능하게 합니다. 데이터베이스 스키마나 스프레드시트 공식을 배울 필요 없이, 에이전트는 사용자의 질의를 분석하고, 최적의 데이터 소스를 결정하며, 대상 분석을 수행하고, 적절한 인용과 함께 포괄적이고 데이터 기반의 답변을 제공합니다.
예를 들어, 다른 LLM을 사용하거나, 사용자 정의 문서 형식을 사용하거나, 에이전트가 분석하고 행동하는 방식을 조정하여 특정 비즈니스 결과를 생성할 수 있습니다. 이 튜토리얼에서는 Novita AI의 통합 LLM API, LangChain의 에이전트 프레임워크, 고급 문서 처리 기능을 사용하여 자체 RAG 기반 영업 분석 시스템을 구축하는 방법을 배웁니다. 질의를 최적의 데이터 소스로 자동 라우팅하고 실행 가능한 비즈니스 인사이트를 제공하는 시스템을 만들게 됩니다.
RAG 기반 영업 분석 시스템이란?
RAG 기반 영업 분석 시스템은 대규모 언어 모델의 힘과 실제 비즈니스 데이터에 접근할 수 있는 검색 계층을 결합하여 소매 판매 질문에 정확하고 근거 있는 답변을 제공하기 위해 Retrieval-Augmented Generation을 사용합니다.
이 시스템은 일반적으로 다음을 포함합니다:
- AI 에이전트가 질의를 분석하고 구조화된 데이터 분석(SQL), 문서 분석, 또는 둘 다 필요한지 결정합니다.
- 에이전트가 질의를 가장 적절한 데이터 소스(정량적 분석을 위한 SQL 데이터베이스 또는 정성적 인사이트를 위한 문서 저장소)로 지능적으로 라우팅합니다.
- 에이전트는 CSV 파일 분석을 위한 Pandas 에이전트, 데이터베이스 질의를 위한 SQL 에이전트, 문서 검색을 위한 벡터 저장소 등 다양한 도구를 사용하여 정보를 수집합니다. 에이전트는 이러한 도구 없이는 워크플로우에서 필수적인 작업을 수행할 수 없습니다.
- 에이전트는 수집된 정보를 특수 기술(숫자 데이터를 위한 통계 분석, 텍스트 문서를 위한 의미 검색)을 사용하여 처리합니다.
- 데이터를 분석한 후 에이전트는 요약, 계산, 적절한 인용을 포함한 구조화된 응답을 생성합니다.
필요한 도구
이 문서의 구축 부분에 들어가기 전에 필요한 도구를 설정해 보겠습니다.
Novita AI
RAG 기반 분석 시스템을 구축하려면 강력한 대규모 언어 모델(LLM)과 임베딩 모델에 대한 액세스가 필요합니다. Novita AI는 단일 통합 인터페이스를 통해 최신 대규모 언어 모델(LLM), 임베딩 모델 등에 대한 액세스를 제공하는 저렴하고 고성능의 API를 제공합니다.
LangChain
LangChain은 LLM을 사용하여 애플리케이션을 구축하기 위해 설계된 오픈소스 프레임워크입니다. LangChain을 사용하면 단계별로 추론하고, 도구를 사용하며, API와 상호 작용하는 에이전트 워크플로우를 만들 수 있습니다. 영업 분석 시스템의 경우 LangChain을 활용하여 분석 프로세스를 구조화하고, SQL 에이전트 및 문서 프로세서와 같은 도구를 사용하며, 모든 데이터를 구조화된 인사이트로 종합합니다.
Streamlit
대화형 웹 인터페이스를 구축하기 위해 Streamlit을 사용합니다. 빠른 프로토타이핑에 적합하며 최소한의 코드로 전문적인 UI를 만들어 튜토리얼을 쉽게 따라할 수 있습니다.
FAISS
벡터 기반 문서 검색을 위해 FAISS는 빠른 유사성 검색 기능을 제공하여 사용자 질의를 기반으로 관련 문서 청크를 빠르게 찾을 수 있게 합니다.
SQLAlchemy & PyMySQL
SQL 데이터베이스 작업과 MySQL 연결을 처리하여 시스템이 구조화된 비즈니스 데이터를 직접 질의할 수 있게 합니다.
pandas
CSV 데이터 조작 및 분석에 필수적입니다. 시스템은 pandas 에이전트를 사용하여 테이블 형식 데이터에 대해 복잡한 계산을 수행합니다.
시스템 아키텍처 개요
RAG 시스템은 사용자 질의를 다단계 워크플로우를 통해 지능적으로 처리하며, 각 질문에 대해 최적의 데이터 소스와 처리 방법을 자동으로 결정합니다. 사용자가 데이터가 어디에 있는지 알 필요 없이 시스템이 백그라운드에서 최적의 접근 방식을 결정합니다.
질의 처리
에이전트를 사용한 작업 프로세스의 첫 번째 단계는 질의 처리입니다. 이 경우 에이전트는 사용자가 입력한 질의를 확인하고 질의의 맥락에 맞는 맞춤형 처리 계획을 수립합니다. 그런 다음 질문을 분석해야 하는 데이터 유형(구조화된 데이터, 문서 분석, 또는 둘 다)을 결정합니다.
데이터 소스 선택
사용자가 질문을 제출하면 시스템은 간단한 접근 방식을 사용합니다. 먼저 SQL 데이터베이스를 조회하여 문제를 해결하려고 시도합니다. 데이터 소스 선택은 구조화된 데이터가 정량적 질문에 가장 정확한 답변을 제공하는 경우가 많기 때문에 의미가 있습니다. SQL 에이전트는 신속하게 쿼리를 실행하고 적절한 계산과 함께 정확한 숫자를 반환할 수 있습니다.
문서 분석
SQL 에이전트가 빈 결과를 반환하거나 관련 정보를 찾을 수 없는 경우 시스템은 자동으로 문서 분석으로 전환합니다. 문서 처리 경로는 파일 유형에 따라 다릅니다. CSV 및 Excel 파일은 복잡한 데이터 분석을 위해 Pandas DataFrame에 로드되고, PDF, Word 파일 및 텍스트 문서는 의미 검색을 위해 FAISS를 사용하여 청킹 및 인덱싱됩니다.
소스 관리
제공된 정보가 신뢰할 수 있는 출처로 추적될 수 있도록 모든 출처를 적절히 인용하는 것이 중요합니다. 워크플로우의 이 단계는 청중과의 신뢰를 구축하여 최종 보고서에 제시된 주장이나 데이터를 검증할 수 있도록 합니다.
응답 합성
수집된 모든 정보의 연결점을 찾기 위해 에이전트는 분석된 모든 데이터 소스의 인사이트를 종합하여 일관된 응답을 형성합니다. 워크플로우의 이 단계가 없으면 출력이 단편적인 보고서가 될 수 있습니다.
RAG 기반 분석 워크플로우 구현
지금까지 RAG 기반 영업 분석 시스템이 무엇인지 이해했습니다. 이제 구현할 차례입니다. 시작하려면 몇 가지 주요 단계를 수행해야 합니다.
설치 및 설정
진행하기 전에 RAG 분석 시스템의 전체 파일 구조를 설정해 보겠습니다. 프로젝트 디렉터리에 다음 파일을 생성해야 합니다.
| rag-analytics-system/ ├── .env # 환경 변수 ├── requirements.txt # Python 종속성 ├── query_processor.py # QueryProcessor 클래스 ├── main.py # Streamlit 인터페이스 |
종속성 설정
먼저 requirements.txt 파일을 모든 필수 종속성으로 업데이트합니다.
| langchain==0.3.26 langchain-openai==0.3.25 python-dotenv==1.1.1 SQLAlchemy==2.0.41 pandas==2.3.0 PyPDF2==3.0.1 faiss-cpu==1.11.0 PyMySQL==1.1.1 cryptography==45.0.4 langchain-experimental==0.3.4 streamlit==1.46.0 openpyxl==3.1.5 python-docx==1.2.0 |
그런 다음 Novita AI에 로그인합니다. 로그인한 후 Manage API Keys 페이지로 이동하여 Add New Key 버튼을 클릭하고 Key Name 을 입력합니다.
Novita AI는 가입 후 다양한 모델을 시험해 볼 수 있는 무료 크레딧을 제공하므로, 구축하거나 실험하기 전에 크레딧을 구매할 걱정을 할 필요가 없습니다. 앞서 생성한 Novita API 키를 .env 파일에 추가합니다.
| NOVITA_API_KEY=”your_novita_api_key_here” |
이제 프로젝트에 필요한 모든 종속성을 설치합니다.
| # 가상 환경 생성 python -m venv rag_analytics_env source rag_analytics_env/bin/activate # Windows: rag_analytics_env\\Scripts\\activate # 종속성 설치 pip install -r requirements.txt |
Query Processor 구축
전체 RAG 분석 워크플로우는 QueryProcessor 라는 모듈식 클래스에 캡슐화됩니다. 이 클래스는 질의 라우팅, SQL 분석 수행, 문서 처리, 결과 합성, 전체 분석 루프 실행을 관리합니다.
QueryProcessor는 분석 시스템의 중앙 오케스트레이터 역할을 합니다. 다양한 데이터 유형에 대해 별도의 도구를 구축하는 대신 이 통합 접근 방식을 통해 질의를 SQL 데이터베이스, CSV 파일 또는 비정형 문서 등 가장 적절한 분석 방법으로 지능적으로 라우팅할 수 있습니다.
먼저 query_processor.py 파일에 프로젝트에 필요한 라이브러리를 임포트합니다. 이러한 임포트는 언어 모델 통합, 데이터베이스 연결, 문서 처리 및 벡터 저장소에 필요한 모든 것을 제공합니다.
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_experimental.agents.agent_toolkits import create_csv_agent, create_pandas_dataframe_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.utilities import SQLDatabase
from langchain.prompts import PromptTemplate
from langchain_community.tools import ReadFileTool
from langchain.schema import Document
import os
from dotenv import load_dotenv
from sqlalchemy.engine import Engine
import glob
import pandas as pd
load_dotenv()
다음으로, Novita API를 통해 LLM에 액세스하고 다양한 데이터 처리 도구를 사용하여 분석 에이전트를 초기화합니다. 초기화 메서드는 필요한 모든 핵심 구성 요소(질의 이해를 위한 언어 모델, 문서 유사성 검색을 위한 임베딩, 대용량 문서 처리를 위한 텍스트 분할기)를 설정합니다.
class QueryProcessor:
def __init__(self, documents_folder: str, sql_engine: Engine):
self.documents_folder = documents_folder
self.sql_engine = sql_engine
self.llm = ChatOpenAI(
model="google/gemma-3-27b-it",
temperature=0,
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="https://api.novita.ai/v3/openai",
default_headers={
"X-Model-Provider": "google"
}
)
self.embeddings = OpenAIEmbeddings(
model="baai/bge-m3",
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="https://api.novita.ai/v3/openai",
default_headers={
"X-Model-Provider": "baai"
}
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
self.sql_agent = None
self._prepare_data_sources()
생성자는 몇 가지 주요 구성 요소를 초기화합니다. LLM에 대해 temperature를 0으로 사용하여 창의적인 변형이 아닌 일관되고 사실적인 응답을 보장합니다. 텍스트 분할기는 대용량 문서를 처리할 때 맥락의 연속성을 유지하기 위해 중첩 청크로 구성됩니다. _prepare_data_sources() 메서드는 마지막에 호출되어 특수 에이전트를 설정합니다.
다중 소스 데이터 통합 생성
다음 단계는 다양한 데이터 처리 에이전트를 초기화하는 메서드를 만드는 것입니다. 각 에이전트는 마치 다양한 종류의 질문에 대해 다른 전문가를 두는 것처럼 특정 데이터 유형 및 분석 작업을 처리하는 데 특화되어 있습니다.
def _prepare_data_sources(self):
"""SQL 및 문서 데이터 소스 준비"""
# SQL 에이전트 준비
self._prepare_sql_agent()
# 문서 에이전트 준비
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""SQL 에이전트 초기화"""
# SQLAlchemy Engine을 LangChain SQLDatabase로 변환
db = SQLDatabase(self.sql_engine)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
self.sql_agent = create_sql_agent(
llm=self.llm,
toolkit=toolkit,
agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
def _prepare_document_agent(self):
"""ReadFileTool 및 벡터 저장소를 사용하여 문서 에이전트 초기화"""
# 지원되는 모든 파일 가져오기
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"\nFound {len(supported_files)} supported files in {self.documents_folder}")
if supported_files:
# ReadFileTool 생성
read_file_tool = ReadFileTool()
# 도구 목록 생성
tools = [read_file_tool]
# react 에이전트를 위한 프롬프트 템플릿 생성
prompt = PromptTemplate.from_template("""
You are a helpful assistant that can answer questions about business documents.
You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought: I should read the relevant documents to find the answer
{agent_scratchpad}
""")
agent = create_react_agent(self.llm, tools, prompt)
self.document_agent = AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
verbose=True
)
print("\nDocument agent initialized successfully")
else:
print("\nNo documents found in the specified folder.")
_prepare_sql_agent() 메서드는 데이터베이스 질의를 위한 특수 에이전트를 만듭니다. ZERO_SHOT_REACT_DESCRIPTION 에이전트 유형을 사용하므로 에이전트가 특정 예제 없이도 SQL 질의에 대해 추론할 수 있어 다양한 데이터베이스 스키마에 적응할 수 있습니다. _prepare_document_agent() 메서드는 지원되는 파일 형식을 스캔하고 다양한 문서 형식을 읽을 수 있는 유연한 에이전트를 만듭니다. ReAct(추론 및 행동) 프롬프트 템플릿은 인간 분석가가 문서 분석에 접근하는 방식과 유사하게 에이전트가 구조화된 사고 과정을 따르도록 안내합니다.
문서 처리 및 벡터 저장소
다음으로 워크플로우가 다양한 문서 유형을 처리하는 방법을 처리해야 합니다. 이 섹션에서는 pandas 에이전트를 사용하여 CSV/Excel 파일을 처리하고 벡터 저장소를 사용하여 비정형 문서를 처리하는 메서드를 만듭니다. 핵심 통찰은 데이터 유형이 다르면 다른 처리 전략이 필요하다는 것입니다.
CSV_PROMPT_PREFIX = """
IMPORTANT: You are working with a pandas DataFrame called 'df' that has been loaded with the actual data.
DO NOT create sample data or make up data. Use ONLY the actual DataFrame 'df' that is available to you.
First, explore the DataFrame by:
1. Setting pandas display options to show all columns: pd.set_option('display.max_columns', None)
2. Check the shape of the DataFrame: print(df.shape)
3. Get the column names: print(df.columns.tolist())
4. Check the data types: print(df.dtypes)
5. Look at the first few rows: print(df.head())
6. Then answer the question using the actual data in the DataFrame.
"""
CSV_PROMPT_SUFFIX = """
- **CRITICAL**: Use ONLY the actual data in the DataFrame. Do NOT create sample data or use fictional data.
- **ALWAYS** before giving the Final Answer, try another method to verify your results.
- Then reflect on the answers of the two methods you did and ask yourself if it answers correctly the original question.
- If you are not sure, try another method.
- FORMAT 4 FIGURES OR MORE WITH COMMAS.
- If the methods tried do not give the same result, reflect and try again until you have two methods that have the same result.
- If you still cannot arrive to a consistent result, say that you are not sure of the answer.
- If you are sure of the correct answer, create a beautiful and thorough response using Markdown.
- **DO NOT MAKE UP AN ANSWER OR USE PRIOR KNOWLEDGE, ONLY USE THE RESULTS OF THE CALCULATIONS YOU HAVE DONE**.
- **ALWAYS**, as part of your "Final Answer", explain how you got to the answer on a section that starts with: "\n\nExplanation:\n".
- In the explanation, mention the column names that you used to get to the final answer.
- Show your work by displaying relevant DataFrame operations and their results.
"""
def _process_document_query(self, query: str) -> str:
"""CSV/Excel 에이전트(테이블 형식 데이터)와 LLM(비정형 데이터)을 사용하여 문서 질의 처리"""
try:
print(f"\nProcessing query: {query}")
# 지원되는 모든 파일 가져오기
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"Found {len(supported_files)} supported files in {self.documents_folder}")
if not supported_files:
return "No documents found to search through."
# 먼저 CSV/Excel 파일 확인
csv_files = [f for f in supported_files if f.endswith('.csv')]
excel_files = [f for f in supported_files if f.endswith(('.xlsx', '.xls'))]
if csv_files:
csv_file = csv_files[0]
print(f"\nProcessing CSV file: {csv_file}")
try:
# 먼저 pandas DataFrame 에이전트로 시도
df = pd.read_csv(csv_file)
print(f"CSV loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# 개선된 구성으로 에이전트 생성
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 대용량 DataFrame에서 토큰 제한 방지
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# 사용자 정의 프롬프트로 질의 처리
print(f"Processing query with agent: {query}")
# 에이전트가 실제 DataFrame을 사용하도록 보장하는 개선된 프롬프트
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error with pandas DataFrame agent: {str(e)}")
print("Trying alternative CSV agent...")
# CSV 에이전트로 대체
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"Error processing CSV file: {str(e2)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing CSV file: {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\nProcessing Excel file: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# 개선된 구성으로 에이전트 생성
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 대용량 DataFrame에서 토큰 제한 방지
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# 사용자 정의 프롬프트로 질의 처리
print(f"Processing query with agent: {query}")
# 에이전트가 실제 DataFrame을 사용하도록 보장하는 개선된 프롬프트
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error processing Excel file: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing Excel file: {str(e)}"
# 비정형 텍스트 파일의 경우
print("\nProcessing unstructured text files...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Error reading file {file_path}: {str(e)}")
continue
if not all_content:
return "Could not read any documents."
# 벡터 저장소를 사용하여 비정형 텍스트 처리
print("Processing text with vector store...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\n\n".join([doc.page_content for doc in relevant_docs])
print(f"Generated context length: {len(context)}")
response = self.llm.invoke(
f"""Based on the following context, answer the question: {query}
\nContext:\n{context}\n\nAnswer:"""
)
return response.content
except Exception as e:
print(f"Error processing document query: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing document query: {str(e)}"
CSV 처리 섹션에는 정확한 분석을 보장하기 위한 상세한 프롬프트 엔지니어링이 포함되어 있습니다. 접두사 및 접미사 프롬프트는 에이전트가 데이터를 환각하거나 잘못된 결과를 제공하는 것을 방지하기 때문에 중요합니다. 에이전트가 여러 방법을 시도하는 검증 단계는 마치 주의 깊은 분석가가 계산을 다시 확인하는 것처럼 정확성을 보장하는 데 도움이 됩니다. 대용량 데이터 세트에서 더 강력한 데이터 처리와 더 나은 성능을 제공하므로 CSV 에이전트보다 pandas DataFrame 에이전트를 우선시합니다. 대체 메커니즘은 한 접근 방식이 실패할 경우 데이터를 처리할 대체 방법이 있도록 보장합니다. 이제 Excel 파일 및 비정형 문서를 처리하기 위한 _process_document_query 메서드를 완성하겠습니다.
elif excel_files:
excel_file = excel_files[0]
print(f"\nProcessing Excel file: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel loaded successfully with {len(df)} rows and columns: {df.columns.tolist()}")
# 개선된 구성으로 에이전트 생성
print("Creating pandas DataFrame agent...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 대용량 DataFrame에서 토큰 제한 방지
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# 사용자 정의 프롬프트로 질의 처리
print(f"Processing query with agent: {query}")
# 에이전트가 실제 DataFrame을 사용하도록 보장하는 개선된 프롬프트
prompt = f"""
You have access to a pandas DataFrame called 'df' with {len(df)} rows and the following columns: {df.columns.tolist()}.
Here are the first few rows of the data:
{df.head().to_string()}
Data types:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Agent response: {response}")
return response['output']
except Exception as e:
print(f"Error processing Excel file: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing Excel file: {str(e)}"
# 비정형 텍스트 파일의 경우
print("\nProcessing unstructured text files...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Error reading file {file_path}: {str(e)}")
continue
if not all_content:
return "Could not read any documents."
# 벡터 저장소를 사용하여 비정형 텍스트 처리
print("Processing text with vector store...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\n\n".join([doc.page_content for doc in relevant_docs])
print(f"Generated context length: {len(context)}")
response = self.llm.invoke(
f"""Based on the following context, answer the question: {query}
\nContext:\n{context}\n\nAnswer:"""
)
return response.content
except Exception as e:
print(f"Error processing document query: {str(e)}")
import traceback
print(f"Full traceback: {traceback.format_exc()}")
return f"Error processing document query: {str(e)}"
SQL 데이터베이스 통합
이제 문서 처리를 다루었으므로 SQL과 문서 분석 사이를 결정하는 지능형 질의 라우팅 시스템을 만들어야 합니다. 이것이 시스템의 핵심 지능입니다. 주어진 질문에 대한 답변이 포함될 가능성이 가장 높은 데이터 소스를 결정하는 것입니다.
def process_query(self, query: str) -> str:
"""
에이전트를 사용하여 SQL과 문서 사이를 지능적으로 결정하는 질의 처리
"""
# 먼저 SQL 에이전트 시도
try:
print("\nTrying SQL Agent...")
sql_result = self.sql_agent.run(query)
no_answer_phrases = [
"no results", "i don't know", "unknown", "not sure", "cannot answer", "don't have", "no data", "n/a"
]
if sql_result and not any(phrase in sql_result.lower() for phrase in no_answer_phrases) and sql_result.strip():
return f"From SQL Database: {sql_result}"
else:
print("SQL Agent could not answer, trying documents...")
except Exception as e:
print(f"SQL Agent Error: {str(e)}")
print("Falling back to documents...")
# SQL 에이전트가 실패하거나 결과를 반환하지 않으면 문서 처리 시도
try:
print("\nProcessing documents...")
doc_result = self._process_document_query(query)
if doc_result:
return f"From Documents: {doc_result}"
else:
print("Document processing returned no results")
except Exception as e:
print(f"Document Processing Error: {str(e)}")
return "Could not find relevant information in either SQL database or documents."
질의 라우팅 논리는 우선 순위 시스템을 따릅니다. SQL 데이터베이스는 일반적으로 비즈니스 질문에 빠르고 정확하게 답변할 수 있는 구조화된 정량적 데이터를 포함하고 있기 때문에 먼저 시도됩니다. SQL 에이전트가 명확하지 않거나 부정적인 응답(no_answer_phrases 목록을 통해 감지)을 반환하면 시스템은 자동으로 문서 처리로 대체됩니다. 이 접근 방식은 인간 분석가가 작업하는 방식(먼저 구조화된 데이터 소스를 확인한 다음 데이터베이스에 필요한 정보가 없을 때 문서와 보고서를 참조)을 모방합니다.
Streamlit 인터페이스 생성
이제 분석 시스템을 비즈니스 사용자가 액세스할 수 있게 만드는 사용자 인터페이스를 구축해 보겠습니다. Streamlit 인터페이스는 직관적인 채팅 기반 경험을 제공하여 기술적 복잡성을 숨기면서 강력한 분석 기능을 제공합니다. main.py 파일에 다음 코드 조각을 추가합니다.
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# 환경 변수 로드
load_dotenv()
#st.text_input
# 세션 상태 초기화
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# 페이지 설정
st.set_page_config(
page_title="Document Analysis Chatbot",
page_icon="🤖",
layout="wide"
)
# 버튼 및 제목 스타일링을 위한 사용자 정의 CSS
st.markdown("""
<style>
.stButton > button {
background-color: #23D57C;
color: white;
border: none;
border-radius: 8px;
padding: 0.5rem 1rem;
font-weight: 600;
transition: all 0.3s ease;
}
.stButton > button:hover {
background-color: #1fb36b;
box-shadow: 0 4px 8px rgba(35, 213, 124, 0.3);
transform: translateY(-2px);
}
.stButton > button:active {
background-color: #1a9960;
transform: translateY(0px);
}
h1 {
color: #23D57C !important;
font-weight: 700;
}
</style>
""", unsafe_allow_html=True)
# 제목 및 설명
st.title("Document Analysis Chatbot using Novita")
이 초기 설정은 사용자 정의 스타일링이 적용된 전문적인 인터페이스를 만듭니다. 세션 상태 관리는 사용자가 인터페이스와 상호 작용할 때 대화 기록을 잃거나 데이터 소스를 다시 초기화할 필요가 없도록 보장합니다. 사용자 정의 CSS는 시각적 피드백을 제공하고 일관된 브랜드 경험을 유지합니다.
최종 시스템 통합
데이터 소스 초기화 및 채팅 기능을 사용하여 Streamlit 인터페이스를 완성해 보겠습니다. 이 마지막 섹션에서는 모든 구성 요소를 비즈니스 분석가가 기술 전문 지식 없이도 사용할 수 있는 사용자 친화적인 애플리케이션으로 통합합니다.
# 데이터 소스가 초기화되었는지 확인
if st.session_state.processor is None:
# 데이터 소스 구성을 중앙에 배치
st.markdown("<br><br>", unsafe_allow_html=True)
# 중앙 정렬 컬럼 생성
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 시작하기")
st.write("문서와 대화를 시작하려면 데이터 소스를 초기화하세요.")
# SQL 데이터베이스 구성 (숨김)
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# 문서 폴더 구성
st.write("**문서 폴더 경로:**")
documents_folder = st.text_input(
"Documents Folder Path",
placeholder="문서 폴더 경로 입력 (예: docs)",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# 버튼 중앙 정렬
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("데이터 소스 초기화", use_container_width=True):
try:
# 문서 폴더 경로 검증 및 해석
if not documents_folder:
st.error("문서 폴더 경로를 입력하세요.")
else:
# 절대 경로로 변환
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"문서 폴더를 찾을 수 없음: {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"경로가 디렉터리가 아님: {abs_documents_folder}")
else:
# 더미 SQL 엔진으로 쿼리 프로세서 초기화
# SQL 엔진 생성
connection_string = f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}"
sql_engine = create_engine(connection_string)
st.session_state.processor = QueryProcessor(abs_documents_folder, sql_engine)
st.success("데이터 소스가 성공적으로 초기화되었습니다!")
st.rerun()
except Exception as e:
st.error(f"데이터 소스 초기화 오류: {str(e)}")
else:
# 사이드바에 데이터 소스 상태 표시
with st.sidebar:
st.header("📊 데이터 소스")
st.success("✅ 데이터 소스 초기화됨")
if st.button("데이터 소스 재설정"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# 메인 채팅 인터페이스
st.header("채팅 인터페이스")
# 채팅 메시지 표시
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# 채팅 입력
if prompt := st.chat_input("문서에 대해 질문하세요."):
# 사용자 메시지를 채팅 기록에 추가
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# 질의 처리
try:
response = st.session_state.processor.process_query(prompt)
# 어시스턴트 응답을 채팅 기록에 추가
st.session_state.messages.append({"role": "assistant", "content": response})
with st.chat_message("assistant"):
st.write(response)
except Exception as e:
st.error(f"질의 처리 오류: {str(e)}")
# 메시지가 있을 때만 채팅 지우기 버튼 추가
if st.session_state.messages:
if st.button("채팅 지우기"):
st.session_state.messages = []
st.rerun()
초기화 프로세스에는 일반적인 사용자 오류를 방지하기 위한 강력한 오류 처리 및 경로 검증이 포함됩니다. 사이드바는 명확한 상태 정보를 제공하며, 채팅 인터페이스는 사용자가 최신 AI 어시스턴트에서 기대하는 친숙한 패턴을 따릅니다. 시스템은 대화 기록을 유지하여 사용자가 이전 질문을 기반으로 자연스러운 분석 워크플로우를 구축할 수 있게 합니다. 채팅 지우기 기능은 필요할 때 새로 시작할 수 있게 합니다.
분석 시스템 실행
이제 RAG 기반 영업 분석 시스템을 구축했으므로 실제 비즈니스 질의에서 얼마나 잘 작동하는지 테스트해 보겠습니다.
먼저 data_generator.py 파일을 만들고 여기의 Python 스크립트 코드를 복사하여 시스템에 제공할 샘플 데이터를 생성하고 다음 명령으로 스크립트를 실행합니다.
| python data_generator.py |
그러면 sample_documents 폴더가 생성됩니다.
- large_sales_dataset.csv - 10,000개의 판매 기록
- business_strategy_2024.txt - 전략적 비즈니스 문서
- sales_meeting_notes.txt - 회의록 및 실행 항목
그런 다음 Streamlit 애플리케이션을 시작합니다.
| streamlit run main.py |
애플리케이션이 브라우저에서 http://localhost:8501에서 열립니다.
초기화되면 문서 경로를 sample_documents 로 입력하고 다양한 비즈니스 질문을 테스트할 수 있습니다.
| # 시스템을 테스트할 샘플 질의 topic_1 = “2024년 전자제품 총 판매 금액은 얼마인가요?” topic_2 = “고객 경험을 위한 주요 전략적 이니셔티브는 무엇인가요?” topic_3 = “가장 높은 총 판매액을 기록한 영업 담당자는 누구이며, 금액은 얼마인가요?” |
결론
이 튜토리얼을 통해 우리는 현대 AI가 비즈니스 인텔리전스 워크플로우를 어떻게 혁신할 수 있는지 보여주는 RAG 기반 분석 시스템을 구축했습니다. 단일 지능형 인터페이스 아래 여러 데이터 처리 접근 방식을 결합함으로써 분석가가 매일 직면하는 비즈니스 질문의 전체 스펙트럼을 처리할 수 있는 도구를 만들었습니다. 시스템의 주요 강점은 다음과 같습니다.
- 각 질문에 가장 적합한 데이터 소스를 자동으로 결정하는 지능형 질의 라우팅
- SQL 데이터베이스, CSV 파일, Excel 스프레드시트 및 텍스트 문서를 처리하는 다중 형식 지원
- 대체 메커니즘과 명확한 오류 메시지를 제공하는 강력한 오류 처리
- 기술 전문 지식이 필요 없는 채팅 기반 상호 작용
- 안정적인 결과를 보장하기 위한 내장 검증 및 확인
이 애플리케이션을 확장하여 추가 소스, 고급 분석 기술 및 사용자 정의 비즈니스 로직을 통합할 수 있습니다. 이제 이 지식을 바탕으로 Novita AI로 자체 에이전트를 구축하거나 기존 프로젝트에 통합해 보세요.
Novita AI는 AI 야망을 실현하는 올인원 클라우드 플랫폼입니다. 통합 API, 서버리스, GPU 인스턴스 — 필요한 비용 효율적인 도구. 인프라를 제거하고, 무료로 시작하며, AI 비전을 현실로 만드세요.
