IndustrialKnowledgeAgent: 스마트 산업 설비 지식 에이전트
IndustrialKnowledgeAgent: 스마트 산업 설비 지식 에이전트
산업 설비 정보(RAG 기반 문서 + 데이터베이스 질의)를 통합해 실시간·문맥 인식 응답을 제공하는 다중 에이전트 워크플로를 구축하는 쿡북이에요. 비정형 문서(RAG)와 정형 데이터(SQLite + 함수 호출)를 모두 다룹니다.
출처: 문서
본문
문제 정의 (Problem Statement)
산업 환경에서 엔지니어와 기술자는 다양한 설비에 대한 포괄적인 정보를 관리하고 검색하는 데 어려움을 겪어요. 이 정보는 기술 매뉴얼, 유지보수 로그, 안전 프로토콜, 트러블슈팅 가이드, 부품 인벤토리에 흩어져 있어요. 데이터의 분산 특성 때문에 접근·활용이 어려워 비효율과 잠재적 안전 위험이 생겨요. 실시간·문맥 인식 응답을 제공하는 지능적이고 적응적인 솔루션이 필요해요.
제안 솔루션 (Proposed Solution)
이 문제를 해결하기 위해 RAG 시스템과 데이터베이스 질의 시스템(FunctionCalling)을 통합한 에이전틱 워크플로를 제안해요. 이 솔루션은 LLM(구조화 출력 메커니즘 포함), 임베딩 모델, 구조화 데이터 검색을 활용해 문맥적으로 관련된 정확한 정보를 제공해요. 워크플로는 각각 특정 역할을 가진 여러 에이전트로 오케스트레이션돼요:
- RAGAgent: LLM과 임베딩 모델로 기술 문서에서 문맥 관련 정보를 검색·생성.
- DatabaseQueryAgent: 유지보수 로그, 기술 사양, 부품 인벤토리, 규정 준수 기록이 담긴 데이터베이스에서 정확하고 구조화된 데이터를 검색.
- WorkflowOrchestrator: RAGSearchAgent와 DatabaseAgent 간 상호작용을 조율해 매끄러운 질의 해결 보장.
데이터셋 상세 (Dataset Details)
PDF 문서 — 다양한 산업 설비에 대한 상세 정보를 담고 있으며, 다음으로 분류돼요:
- Technical Manuals (기술 매뉴얼): 운영·유지보수 가이드
- Maintenance Guides (유지보수 가이드): 정기·예방 유지보수 작업
- Troubleshooting Guides (트러블슈팅 가이드): 흔한 문제의 해결책
- Safety Protocols (안전 프로토콜): 안전 절차와 지침
데이터베이스 — PDF 문서를 보완하는 구조화 정보를 담고 있어요:
- Compliance Database (
compliance_db): 안전 인증과 규정 준수 상태 - Maintenance Database (
maintenance_db): 유지보수 활동 로그 - Technical Specifications Database (
technical_specifications_db): 상세 기술 사양 - Parts Inventory and Compatibility Database (
parts_inventory_compatibility_db): 부품, 호환성, 인벤토리 상태 정보
참고: 이 데모에 사용된 모든 데이터는 인공적으로 생성된(synthetically generated) 것이에요.
설치 (Installation)
IndusAgent 시스템에 필요한 Python 패키지를 설치해요:
!pip install mistralai==1.5.1 # Mistral AI client
!pip install qdrant-client==1.13.2 # Vector database client
!pip install gdown==5.2.0 # Google Drive download
임포트 (Imports)
LLM 연산, 데이터 처리, 벡터 데이터베이스 관리, 유틸리티 함수에 필요한 라이브러리를 가져와요:
# Core libraries
import os
import json
import functools
import warnings
from typing import List, Dict, Any, Tuple
# LLM and Data Processing
from mistralai.client import Mistral
from pydantic import BaseModel
import pandas as pd
import sqlite3
from tqdm import tqdm
# Vector Database
from qdrant_client import QdrantClient
from qdrant_client.models import (
PointStruct, VectorParams, Distance,
Filter, FieldCondition, MatchValue
)
# Data Download
import gdown
import zipfile
# Suppress warnings
warnings.filterwarnings('ignore', category=DeprecationWarning)
데이터 다운로드
Google Drive에서 데이터셋을 다운로드해 data 디렉터리에 푼 뒤 작업 환경을 설정해요. 데이터셋에는 데이터베이스용 CSV 파일과 문서 처리용 PDF가 있어요.
file_id = "1lwYSN6ry3JOA7pw3WAx72a_IXGqqmR8y"
output_file = "data.zip" # Change this if your file is not a ZIP file
# Google Drive direct download URL
gdrive_url = f"https://drive.google.com/uc?id={file_id}"
# Download the file
gdown.download(gdrive_url, output_file, quiet=False)
print(f"✅ File downloaded: {output_file}")
# Unzip the file into the current directory
with zipfile.ZipFile(output_file, 'r') as zip_ref:
zip_ref.extractall(".") # Extracts directly to the current directory
print(f"✅ Files extracted to: {os.getcwd()}") # Confirm extraction path
output_dir = "data"
# Change working directory to the extracted folder
os.chdir(output_dir)
# Verify the new working directory
print(f"📂 Current directory: {os.getcwd()}")
환경 변수 설정
인증을 위해 Mistral API 키를 환경 변수로 설정해요:
os.environ["MISTRAL_API_KEY"] = "<YOUR MISTRAL API KEY>" # Get your Mistral API key from https://console.mistral.ai/api-keys/
Mistral LLM과 Qdrant 벡터 데이터베이스 초기화
텍스트 생성용 Mistral LLM 클라이언트와 유사도 검색용 Qdrant 벡터 데이터베이스 클라이언트를 초기화해요.
- 데모에는 최신 모델 Mistral Small 3를 사용해요.
- 진행 전에 Qdrant Cloud 또는 Docker 설정이 필요해요. 설정 방법은 문서를 참고하세요.
model = "mistral-small-latest"
mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
qdrant_client = QdrantClient(
url= "<URL>",
api_key= "<API KEY>",
) # Replace with your Qdrant API key and URL if you are using Qdrant Cloud - https://cloud.qdrant.io/
시스템 프롬프트 (System Prompts)
시스템은 응답 생성을 안내하는 세 가지 유형의 프롬프트를 사용해요:
- PDF 요약 프롬프트:
summarization_prompt는 PDF 문서의 간결한 요약을 만드는 데 사용. - 응답 생성 프롬프트:
response_generation_prompt는 검색된 컨텍스트를 바탕으로 응답을 생성. - 최종 응답 통합 프롬프트:
final_response_generation_prompt는 PDF와 여러 데이터베이스 등 여러 출처의 응답을 요약.
# Define the prompt for generating a response
response_generation_prompt = '''Based on the following context answer the query:\n\n Context: {context}\n\n Query: {query}'''
# Prompt for summarizing the PDF text
summarization_prompt = '''Your task is to summarize the following text focusing on the core essence of the text in maximum of 2-3 sentences.'''
# Prompt for final response summarization
final_response_summarization_prompt = """You are an expert technical assistant. Your task is to create a comprehensive,
coherent response by combining information from multiple sources: database records and documentation.
Consider the following guidelines:
1. Integrate information from both sources seamlessly
2. Resolve any conflicts between sources, if they exist
3. Present information in a logical, step-by-step manner when applicable
4. Include specific technical details, measurements, and procedures when available
5. Prioritize safety-related information when present
6. Add relevant maintenance intervals or schedules if mentioned
7. Reference specific part numbers or specifications when provided
The user's query is: {query}
Based on the following responses from different sources, create a unified, clear answer:
{responses}
Remember to:
- Focus on accuracy and completeness
- Maintain technical precision
- Use clear, professional language
- Address all aspects of the query
- Highlight any important warnings or precautions"""
DataProcessor
DataProcessor 클래스는 시스템의 모든 데이터 처리 연산을 관리하는 포괄적인 컴포넌트예요. 비정형(PDF)과 구조화(CSV) 데이터, 임베딩 생성·저장을 모두 처리해요.
- Mistral OCR을 사용한 PDF 문서 처리와 텍스트 추출
- CSV를 데이터베이스로 수집(ingestion)
- 임베딩 생성과 벡터 저장
- 문서·데이터 일괄(batch) 처리
주요 컴포넌트 (Main Components)
1. 문서 처리 (Document Processing)
get_categorized_filepaths: 디렉터리 구조를 순회해 분류된 PDF 파일 경로를 얻기parse_pdf: Mistral OCR로 PDF의 모든 페이지에서 텍스트 추출process_single_pdf: 개별 PDF를 파이프라인 전체로 처리process_documents: 여러 문서의 순차 처리 담당
2. 요약과 임베딩 (Summarization and Embeddings)
summarize: Mistral 모델로 텍스트의 간결한 요약 생성get_text_embedding: Mistral 임베딩 모델로 텍스트 임베딩 생성qdrant_insert_embeddings: 메타데이터와 함께 임베딩을 Qdrant에 저장process_and_store_embeddings: 임베딩 일괄 처리 담당
3. 데이터베이스 연산 (Database Operations)
insert_csv_to_table: 단일 CSV 파일을 특정 데이터베이스 테이블로 로드insert_data_database: 여러 CSV 파일을 각각의 테이블에 삽입
class DataProcessor:
"""
Handles all data processing operations including:
- PDF parsing and text extraction
- CSV to database ingestion
- Embedding generation and storage
- Batch processing of documents and data
"""
def __init__(self, mistral_client: Mistral, qdrant_client: QdrantClient):
self.mistral_client = mistral_client
self.qdrant_client = qdrant_client
def get_categorized_filepaths(self, root_dir: str) -> List[Dict[str, str]]:
"""
Walk through the directory structure and get file paths with their categories.
"""
categorized_files = []
for category in os.listdir(root_dir):
category_path = os.path.join(root_dir, category)
if not os.path.isdir(category_path):
continue
for root, _, files in os.walk(category_path):
for file in files:
if file.lower().endswith('.pdf'):
filepath = os.path.join(root, file)
categorized_files.append({
'filepath': filepath,
'category': category
})
return categorized_files
def parse_pdf(self, file_path: str) -> str:
"""Parse a PDF file and extract text from all pages using Mistral OCR."""
# Upload a file
uploaded_pdf = self.mistral_client.files.upload(
file={
"file_name": file_path,
"content": open(file_path, "rb"),
},
purpose="ocr"
)
# Get a signed URL for the uploaded file
signed_url = self.mistral_client.files.get_signed_url(file_id=uploaded_pdf.id)
# Get OCR results
ocr_response = self.mistral_client.ocr.process(
model="mistral-ocr-latest",
document={
"type": "document_url",
"document_url": signed_url.url,
}
)
# Extract text from the OCR response
text = "\n".join([x.markdown for x in (ocr_response.pages)])
return text
def summarize(self, text: str, summarization_prompt: str = summarization_prompt) -> str:
"""Summarize the given text using the Mistral model."""
chat_response = self.mistral_client.chat.complete(
model=model,
messages=[
{
"role": "system",
"content": summarization_prompt
},
{
"role": "user",
"content": text
},
],
temperature=0
)
# ...
parse_pdf는 파일을 purpose="ocr"로 업로드하고, 서명된 URL을 얻어 mistral-ocr-latest로 OCR을 수행한 뒤 모든 페이지의 텍스트를 이어 붙여요. summarize는 요약 프롬프트로 텍스트를 요약해요.
RAGAgent
RAGAgent 클래스는 지능적인 검색과 응답 생성을 제공하기 위해 RAG(검색 증강 생성)를 구현해요. 벡터 검색 기능과 LLM을 결합해 문맥 관련 답변을 제공해요.
- 쿼리 분류(categorization)와 분류
- Qdrant 내 벡터 유사도 검색
- 문맥 인식 응답 생성
- 문서 인용(citation) 처리
주요 컴포넌트
1. 쿼리 처리 (Query Processing)
query_categorization: 쿼리를 사전 정의된 범주(기술 매뉴얼, 안전 프로토콜 등)로 분류query: 쿼리에서 최종 응답까지 전체 RAG 파이프라인 조율
2. 검색 (Search and Retrieval)
qdrant_search: 쿼리 임베딩으로 의미 검색을 수행하고, 문서 범주로 결과를 필터링한 뒤 top-k 관련 문서 반환
3. 응답 생성 (Response Generation)
generate_response: 검색된 컨텍스트로 자연어 응답 생성, 특화 프롬프트로 LLM 사용, 출처 문서에 대한 인용 제공
Query Category Model
쿼리 분류의 구조를 정의하는 Pydantic 모델로, RAGAgent가 쿼리를 관련 범주(technical_manual, safety_protocol 등)로 분류하는 데 사용돼요.
# Define category model for query classification
class Category(BaseModel):
category: str
class RAGAgent:
"""
Agent responsible for Retrieval-Augmented Generation (RAG) operations.
"""
def __init__(self, mistral_client: Mistral, qdrant_client: QdrantClient):
self.mistral_client = mistral_client
self.qdrant_client = qdrant_client
def generate_response(self, context: str, query: str) -> str:
"""Generate a response based on the given context and query."""
chat_response = self.mistral_client.chat.complete(
model=model,
messages=[
{
"role": "user",
"content": response_generation_prompt.format(context=context, query=query)
},
]
)
return chat_response.choices[0].message.content
def query_categorization(self, query: str) -> str:
"""Categorize the query into predefined categories."""
chat_response = self.mistral_client.chat.parse(
model=model,
messages=[
{
"role": "system",
"content": "Classify the query into one or more categories of the following list: ['technical_manual', 'safety_protocol', 'maintenance_guide', 'troubleshooting_guide']"
},
{
"role": "user",
"content": query,
},
],
response_format=Category,
max_tokens=256,
temperature=0
)
return json.loads(chat_response.choices[0].message.content)
def qdrant_search(self, query: str, category: str = None, top_k: int = 5) -> List[Dict[str, Any]]:
"""Search for similar texts in Qdrant based on the query and category."""
query_vector = DataProcessor(self.mistral_client, self.qdrant_client).get_text_embedding([query])
retrieval_results = self.qdrant_client.search(
collection_name="embeddings",
query_vector=query_vector,
query_filter=Filter(
must=[
FieldCondition(
key='category',
match=MatchValue(value=category)
)
]
),
limit=top_k
)
return retrieval_results
def query(self, query_text: str, top_k: int = 3) -> Tuple[str, str]:
"""Process a natural language query using RAG."""
category = self.query_categorization(query_text)["category"]
results = self.qdrant_search(query_text, category, top_k=top_k)
file_paths = [result.payload["filepath"] for result in results]
retrieved_text = "\n".join([result.payload["text"] for result in results])
citations = ",".join([result.payload["filepath"] for result in results])
return self.generate_response(retrieved_text, query_text), citations
query_categorization은 chat.parse와 response_format=Category(Pydantic)로 구조화 출력을 사용해 쿼리를 범주로 분류해요. query는 범주를 얻어 해당 범주로 Qdrant를 검색하고, 검색된 텍스트로 응답을 생성하며 인용을 함께 반환해요.
데이터베이스 쿼리 도구 (Database Query Tools)
DatabaseQueryAgent를 위한 함수 호출 인터페이스를 정의하는 데이터베이스 쿼리 함수 도구들이에요. 서로 다른 데이터베이스 테이블의 구조화 쿼리를 가능하게 해요:
tools = [
{
"type": "function",
"function": {
"name": "query_compliance",
"description": '''Query compliance records with filters. \n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,ComplianceType,Certification,IssueDate,ExpiryDate,ComplianceStatus,ResponsiblePerson
1,CNC Machine,ABC Corp,Model X,Safety,ISO 9001,2020-01-15,2025-01-15,Active,John Doe''',
"parameters": {
"type": "object",
"properties": {
"filters": {
"type": "object",
"description": '''Dictionary of column names and values to filter by.''',
"additionalProperties": {
"type": "string"
}
}
},
"required": ["filters"],
},
},
},
{
"type": "function",
"function": {
"name": "query_maintenance",
"description": '''Query maintenance records with filters. \n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,InstallationDate,LastMaintenanceDate,NextMaintenanceDate,MaintenanceType,MaintenanceDetails,MaintenanceStatus,ResponsibleTechnician
1,CNC Machine,ABC Corp,Model X,2020-01-15,2023-09-01,2023-12-01,Preventive,Oil change,Completed,John Doe''',
"parameters": {
"type": "object",
"properties": {
"filters": {
"type": "object",
"description": "Dictionary of column names and values to filter by",
"additionalProperties": {
"type": "string"
}
}
},
"required": ["filters"],
},
},
},
{
"type": "function",
"function": {
"name": "query_technical_specs",
"description": '''Query technical specifications with filters.\n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,SpecificationType,SpecificationDetail,Unit,Value,DateMeasured,MeasuredBy
1,CNC Machine,ABC Corp,Model X,Power,Motor Power,kW,15,2023-01-15,John Doe''',
"parameters": {
"type": "object",
"properties": {
"filters": {
"type": "object",
"description": "Dictionary of column names and values to filter by",
"additionalProperties": {
"type": "string"
}
}
},
"required": ["filters"],
},
},
},
{
"type": "function",
"function": {
"name": "query_parts_inventory_compatibility",
"description": '''Query parts, inventory and compatibility with filters.\n\n A sample example of columns and corresponding values from db are:\n\n PartID,PartName,EquipmentID,EquipmentName,Manufacturer,Model,PartType,Quantity,Compatibility,Supplier,LastOrderDate,NextOrderDate,PartStatus
1,Oil Filter,1,CNC Machine,ABC Corp,Model X,Filter,50,Compatible,Supplier A,2023-01-15,2023-12-01,In Stock''',
"parameters": {
"type": "object",
"properties": {
"filters": {
"type": "object",
"description": "Dictionary of column names and values to filter by",
"additionalProperties": {
"type": "string"
}
}
},
"required": ["filters"],
},
},
}
]
각 도구는 filters 딕셔너리(열 이름과 값)를 받아 해당 테이블을 조건부로 조회하는 함수 호출 인터페이스예요. 도구 설명에 샘플 열과 값이 포함되어 있어 모델이 올바른 쿼리를 만들 수 있어요.
DatabaseQueryAgent
DatabaseQueryAgent 클래스는 SQLite 데이터베이스와의 상호작용을 관리해요. 유지보수 로그, 기술 사양, 부품 인벤토리, 규정 준수 기록을 담은 여러 데이터베이스에 대해 함수 호출로 구조화 데이터 쿼리를 처리해요. 테이블별 특화 쿼리 기능을 제공해요.
- 자연어 쿼리 처리
- 구조화 데이터베이스 쿼리
- 쿼리 실행을 위한 함수 호출
- JSON 응답 포맷팅
주요 컴포넌트
1. 테이블별 쿼리 (Table-Specific Queries)
query_compliance: 필터링된 규정 준수 기록 검색query_maintenance: 유지보수 관련 정보 접근query_technical_specs: 기술 사양 가져오기query_parts_inventory_compatibility: 부품·호환성 데이터 검색
2. 쿼리 처리 (Query Processing)
query: 함수 호출로 자연어 쿼리 처리, 적절한 데이터베이스 연산을 위한 툴 콜 처리, 데이터베이스 툴 인용 추적, 쿼리 결과로 응답 포맷팅
class DatabaseQueryAgent:
"""
Agent responsible for interacting with the SQLite database.
"""
def __init__(self, db_path: str, mistral_client: Mistral):
self.db_path = db_path
self.tools = tools
self.names_to_functions = {
'query_compliance': functools.partial(self.query_compliance),
'query_maintenance': functools.partial(self.query_maintenance),
'query_technical_specs': functools.partial(self.query_technical_specs),
'query_parts_inventory_compatibility': functools.partial(self.query_parts_inventory_compatibility)
}
self.mistral_client = mistral_client
def query_compliance(self, filters: Dict[str, str]) -> str:
"""
Query compliance table with filters.
Args:
filters (Dict[str, str]): Dictionary of column names and values to filter by.
Returns:
str: The query result in JSON format.
"""
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
where_conditions = []
params = []
for column, value in filters.items():
where_conditions.append(f"{column} = ?")
params.append(value)
where_clause = " AND ".join(where_conditions)
query = f"SELECT * FROM compliance WHERE {where_clause}"
cursor.execute(query, params)
columns = [description[0] for description in cursor.description]
result = cursor.fetchone()
if result:
record = dict(zip(columns, result))
return json.dumps({'result': record})
return json.dumps({'error': 'No matching records found'})
except Exception as e:
return json.dumps({'error': str(e)})
finally:
conn.close()
def query_maintenance(self, filters: Dict[str, str]) -> str:
"""
Query maintenance table with filters.
...
"""
# maintenance 테이블 SELECT 처리 (query_compliance와 동일한 패턴)
...
def query_technical_specs(self, filters: Dict[str, str]) -> str:
"""Query technical specifications table with filters."""
# technical_specifications 테이블 SELECT 처리
...
def query_parts_inventory_compatibility(self, filters: Dict[str, str]) -> str:
"""Query parts inventory and compatibility table with filters."""
# parts_inventory_compatibility 테이블 SELECT 처리
...
def query(self, query_text: str) -> str:
"""
Process a natural language query using the database tools.
Args:
query_text (str): Natural language query
Returns:
str: Response from the database query
"""
messages = [{"role": "user", "content": query_text}]
# Get initial response with potential tool calls
response = self.mistral_client.chat.complete(
model=model,
messages=messages,
tools=self.tools,
tool_choice="any",
)
messages.append(response.choices[0].message)
citations = set()
# Handle any tool calls
if hasattr(response.choices[0].message, 'tool_calls') and response.choices[0].message.tool_calls:
for tool_call in response.choices[0].message.tool_calls:
function_name = tool_call.function.name
print(f"Tool call: {function_name}")
citations.add(function_name)
function_params = json.loads(tool_call.function.arguments)
print(f"Tool call parameters: {function_params}")
function_result = self.names_to_functions[function_name](**function_params)
messages.append({
"role": "tool",
"name": function_name,
"content": function_result,
"tool_call_id": tool_call.id
})
# ...
각 테이블 쿼리 함수는 filters를 받아 WHERE 절을 만들고 SQLite에서 조회한 뒤 결과를 JSON으로 반환해요. query는 tool_choice="any"로 함수 호출을 유도하고, 툴 콜을 실행해 결과를 대화에 붙인 뒤 최종 응답을 구성해요.
WorkflowOrchestrator
WorkflowOrchestrator 클래스는 RAGAgent와 DatabaseQueryAgent 간의 상호작용을 오케스트레이션해, 구조화·비구조화 데이터 소스의 정보를 결합해 포괄적인 응답을 제공해요.
- 워크플로 오케스트레이션과 조정
- 응답 결합과 통합
- 최종 응답 요약
- 출처 인용 관리
주요 컴포넌트
1. 워크플로 실행 (Workflow Execution)
workflow: 전체 쿼리 처리 파이프라인 관리, 두 에이전트의 응답 조율, 최종 통합 응답 생성, 인용으로 추적성 유지
2. 응답 요약 (Response summarization)
combine_and_summarize_responses: 두 에이전트의 응답 병합·요약, 결합 응답에 구조화 포맷 적용, 요약 프롬프트로 일관된 출력 생성
class WorkflowOrchestrator:
"""
WorkflowOrchestrator is responsible for orchestrating the workflow between RAGSearchAgent and DatabaseQueryAgent.
Handles query processing, response combination, and final summarization.
"""
def __init__(self,
rag_agent: RAGAgent,
db_query_agent: DatabaseQueryAgent,
client: Mistral):
"""..."""
self.rag_agent = rag_agent
self.db_query_agent = db_query_agent
self.client = client
def combine_and_summarize_responses(self,
responses: Dict[str, str],
query: str,
summarization_prompt: str = summarization_prompt) -> str:
"""Combine and summarize multiple responses into a coherent final response."""
# Format responses into a structured text
combined_text = "\n\n".join([
f"{source}: {content}"
for source, content in responses.items()
])
# Generate summarized response
chat_response = self.client.chat.complete(
model=model,
messages=[
{
"role": "system",
"content": summarization_prompt
},
{
"role": "user",
"content": f"Query: {query}\n\nResponses:\n{combined_text}"
},
],
temperature=0
)
return chat_response.choices[0].message.content
def workflow(self, query: str) -> str:
"""Execute the workflow for processing a query."""
# Get responses from both agents
db_response, tools_citations = self.db_query_agent.query(query)
rag_response, rag_citations = self.rag_agent.query(query)
# Combine responses into a dictionary
responses = {
"Database Response": db_response,
"RAG Response": rag_response
}
# Generate final summarized response
final_response = self.combine_and_summarize_responses(
responses=responses,
query=query,
summarization_prompt=final_response_summarization_prompt
)
# Add citations
citations = (
f"\n\nSources:\n"
f"- Database Tools: {((tools_citations))}\n"
f"- PDF Sources: {rag_citations}"
)
return final_response + citations
workflow는 두 에이전트의 응답(데이터베이스 + RAG)을 받아 final_response_summarization_prompt로 통합·요약한 뒤, 데이터베이스 도구와 PDF 출처 인용을 붙여 최종 답변을 반환해요.
문서 초기화와 처리 (Initialize and Process Documents)
DataProcessor를 초기화하고 파일 수집에서 임베딩 저장까지의 전체 파이프라인으로 PDF 문서를 처리해요:
# Initialize the processor
doc_processor = DataProcessor(mistral_client, qdrant_client)
# Process documents
file_list = doc_processor.get_categorized_filepaths(root_dir='./pdf_data')
processed_docs = doc_processor.process_documents(file_list)
doc_processor.process_and_store_embeddings(processed_docs)
데이터베이스 테이블에 데이터 삽입
여러 CSV 파일을 SQLite의 대응 데이터베이스 테이블로 로드해요:
# Insert data into tables
db_path = "./database.db"
file_mappings = {
"compliance": "./csv_data/compliance_db.csv",
"maintenance": "./csv_data/maintenance_db.csv",
"technical_specifications": "./csv_data/technical_specifications_db.csv",
"parts_inventory_compatibility": "./csv_data/parts_inventory_compatibility_db.csv"
}
doc_processor.insert_data_database(db_path, file_mappings)
에이전트 초기화 (Initialise the Agents)
세 개의 핵심 에이전트를 초기화해요:
- 문서 검색·응답용 RAGAgent
- 구조화 데이터 쿼리용 DatabaseQueryAgent
- 응답 오케스트레이션용 WorkflowAgent
rag_agent = RAGAgent(mistral_client, qdrant_client)
db_query_agent = DatabaseQueryAgent(db_path, mistral_client)
workflow_orchestrator = WorkflowOrchestrator(rag_agent, db_query_agent, mistral_client)
예시 쿼리 (Example Queries)
이렇게 워크플로를 호출해 질문할 수 있어요:
query = "What are the troubleshooting steps for inaccurate machining in CNC Machine (Model X) and when was its last maintenance performed?"
print(f"Query: {query}")
print("----------------------")
answer = workflow_orchestrator.workflow(query)
print("------------Answer----------")
print(answer)
query = "What are the safety protocols for the Cooling System (Model Y), and when is its next scheduled maintenance?"
print(f"Query: {query}")
print("----------------------")
answer = workflow_orchestrator.workflow(query)
print("------------Answer----------")
print(answer)
각 쿼리는 워크플로가 데이터베이스 응답과 RAG(문서) 응답을 결합해 최종 답변과 출처 인용을 생성하는 식으로 처리돼요.