Daddy Makers: 구조화된 TEXT to XML · JSON · SQL · GRAPH 검색을 위한 LLM 기반 RAG AI 에이전트 개발

이 글은 최근 LLM 개발자 관심 중 하나인 LLM 기반 구조화된 형식의 데이터 생성하는 방법을 간략히 소개한다. 이를 위해, JSON 입출력이 가능하도록 RAG처리하는 방법을 개발한다.

구조화된 LLM 출력

본 글은 OpenAI ChatGPT와 같이 API를 사용하려면 구독해야 하는 상용 모델 대신 라마, 미스랄과 같은 오픈소스 모델을 사용한다.

JSON 파일을 RAG하기 위해서는 여러가지 부분을 고려해야 한다. 보통, RAG를 위해서는 랭체인, 라마 인덱스 등 라이브러리를 사용하는 데, 의존성 변화가 심한데다, 급격히 발전하고 있어 설치, 빌드에 여러 에러가 발생하고 있다. 벡터 데이터베이스를 이용한 RAG도 한계가 있어, 필요한 정보를 제대로 검색하지 못하는 이슈들이 있다. 이런 문제들을 고려하고, RAG처리해야 한다.

개발 환경 준비

다음과 같이 개발환경을 설치한다. 그리고, ollama 도구를 설치하도록 한다.

pip install llama-cpp-python

pip install 'crewai[tools]'

pip install langchain

TEXT TO JSON

라마 모델을 로딩하고, JSON 문법으로 출력하도록 GGUF 문법 정의를 이용해 JSON 출력을 생성한다. 다음 코드를 실행한다.

from llama_cpp.llama import Llama, LlamaGrammar

import httpx

grammar_text = httpx.get("https://raw.githubusercontent.com/ggerganov/llama.cpp/master/grammars/json_arr.gbnf").text

grammar = LlamaGrammar.from_string(grammar_text)

llm = Llama("llama-2-13b.Q8_0.gguf")

response = llm(

"JSON list of name strings of attractions in SF:",

grammar=grammar, max_tokens=-1

)

import json

print(json.dumps(json.loads(response['choices'][0]['text']), indent=4))

출력 결과는 다음과 같이, 샌프란시스코에 있는 놀이 시설을 보여준다.

[

{

"address": {

"country": "US",

"locality": "San Francisco",

"postal_code": 94103,

"region": "CA",

"route": "Museum Way",

"street_number": 151

"geocode": {

"latitude": 37.782569,

"longitude": -122.406605

"name": "SFMOMA",

"phone": "(415) 357-4000",

"website": "http://www.sfmoma.org/"

}

]

이와 같이, LLM 출력을 컴퓨터 처리하기 용이한 구조로 생성할 수 있다.

참고로, 여기서 사용한 JSON 문법은 다음과 같이 정형 규칙 언어로 정의된 것을 사용한 것이다.

json.gbnf

TEXT TO XML

다음은 XML에서 데이터를 검색하는 방법을 보여준다.

from langchain.output_parsers import XMLOutputParser

from langchain_community.chat_models import ChatAnthropic

from langchain_core.prompts import PromptTemplate

model = ChatAnthropic(model="claude-2", max_tokens_to_sample=512, temperature=0.1)

actor_query = "Generate the shortened filmography for Tom Hanks."

output = model.invoke(

f"""{actor_query}

Please enclose the movies in <movie></movie> tags"""

)

print(output.content)

JSON RAG 처리

이 예시는 특정 웹사이트 내용과 제품 정보가 포함된 JSON파일(예제 다운로드)를 RAG 한다. 예제파일은 input_json 폴더를 만들고, 그 아래 복사한다.

이제 다음과 같이 랭체인를 이용해 JSON 파일을 RAG 처리한다.

from langchain_community.vectorstores import Chroma

from langchain_community.chat_models import ChatOllama

from langchain_community.embeddings.fastembed import FastEmbedEmbeddings

from langchain_community.document_loaders import WebBaseLoader

from langchain.text_splitter import CharacterTextSplitter

from langchain_core.prompts import ChatPromptTemplate

from langchain.chains.combine_documents import create_stuff_documents_chain

from langchain.chains import create_retrieval_chain

from langchain_core.documents import Document

import sys, os, json

class ChatWebDoc:

vector_store = None

retriever = None

chain = None

def __init__(self):

self.model = ChatOllama(model="mistral:instruct")

#Loading embedding

self.embedding = FastEmbedEmbeddings()

self.text_splitter = CharacterTextSplitter(chunk_size=1024, chunk_overlap=100)

self.prompt = ChatPromptTemplate.from_messages(

[

("system",

"""You are an assistant for question-answering tasks. Use only the following

context to answer the question. If you don't know the answer, just say that you don't know.

CONTEXT:

{context}

"""),

("human", "{input}"),

]

)

def ingest(self, url_list):

#Load web pages

docs = WebBaseLoader(url_list).load()

chunks = self.text_splitter.split_documents(docs)

#Create vector store

vector_store = Chroma.from_documents(documents=chunks,

embedding=self.embedding, persist_directory="./chroma_db")

def ingest_json(self, input_folder):

all_chunks = []

for filename in os.listdir(input_folder):

if filename.endswith('.json') == False:

continue

file_path = os.path.join(input_folder, filename)

with open(file_path, 'r', encoding='utf-8') as file:

dataset = json.load(file)

for data in dataset:

text = json.dumps(data) if isinstance(data, dict) else str(data)

document = Document(page_content=text, metadata={"source": "local"})

all_chunks.append(document)

vector_store = Chroma.from_documents(documents=all_chunks,

embedding=self.embedding,

persist_directory="./chroma_db")

def load(self):

vector_store = Chroma(persist_directory="./chroma_db",

embedding_function=self.embedding)

self.retriever = vector_store.as_retriever(

search_type="similarity_score_threshold",

search_kwargs={

"k": 3,

"score_threshold": 0.5,

)

document_chain = create_stuff_documents_chain(self.model, self.prompt)

self.chain = create_retrieval_chain(self.retriever, document_chain)

def ask(self, query: str):

if not self.chain:

self.load()

result = self.chain.invoke({"input": query})

print(result["answer"])

for doc in result["context"]:

print("Source: ", doc.metadata["source"])

def build():

w = ChatWebDoc()

w.ingest([

"https://www.webagesolutions.com/courses/WA3446-comprehensive-angular-programming",

"https://www.webagesolutions.com/courses/AZ-1005-configuring-azure-virtual-desktop-for-the-enterprise",

"https://www.webagesolutions.com/courses/AZ-305T00-designing-microsoft-azure-infrastructure-solutions",

])

w.ingest_json("./input_json")

def chat():

w = ChatWebDoc()

w.load()

build()

while True:

query = input(">>> ")

if len(query) == 0:

continue

if query == "/exit":

break

w.ask(query)

if len(sys.argv) < 2:

chat()

elif sys.argv[1] == "--ingest":

build()

실행 결과는 다음과 같다.

답변이 잘 생성되나, 매 질문에 따라 잘못된 답을 생성하기도 한다. 이런 이유로, 답변 정확도를 높이기 위한 좀 더 다양한 LLM RAG 처리 옵션과 기법이 적용될 필요가 있다.

TEXT TO SQL 처리

TEXT에서 SQL를 생성해 DB에 적절한 질의를 통해 답변을 생성하는 방법(TEXT TO SQL)을 기술한다. 이 기술의 중요한 부분은 LLM에 정확한 컨텍스트를 제공하여 정확하고 근거가 있는 SQL 쿼리를 작성하는 데 도움이 되어야 한다. 이를 위해 임베딩 모델을 정의해야 한다.

이를 위해, 스키마를 포함하고 인덱싱 하면 모델이 동적으로 참조할 수 있는 의미 체계 메모리를 생성하여 항상 유효하고 컨텍스트를 인식하는 SQL을 생성하도록 해야 한다. 다음은 그 순서를 보여준다.

RDBMS
↓
Extract Tables + Columns
↓
Convert to Embeddings
↓
Store in Vector DB
↓
Semantic Search During Query Time

만약, 지난달 등록한 수강생 표시 란 질의가 입력될 경우, 벡터 DB에서 관련된 데이터를 저장하는 테이블의 스키마 청크를 가져와야 한다.

이후, 질문과 검색된 스키마 청크를 LLM에 컨텍스트로 전달한다.

sql_prompt = PromptTemplate.from_template(

"""

You are a SQL generator. Based on the following context, generate a SINGLE READ-ONLY SQLite SELECT query (no semicolons, no multiple statements).

Context:

{context}

Question:

{question}

Return only the SQL SELECT statement.

""")

LLM은 적절한 SQL문을 생성할 것이다. 리턴된 문장에서 SQL문만 남기고 나머지 노이즈는 제거한다.

이후, SQL 유효성 검사를 통해 DB 전체를 DELETE하는 연산자 등을 검색해 안전성을 확인한다.

그리고, 얻은 SQL문장을 간단히 해당 DB에 질의하면 결과가 리턴 될 것이다. 이를 LLM에 다시 전달해서 우리가 읽기 쉬운 자연어로 변환해달라고 요청한다.

TEXT TO GRAPH 처리

주어-동사-목적어 형식이 수십만개 이상 명사를 중심으로 연결된 데이터셋이 있다고 하자. 한쪽 끝에서 다른 한쪽 끝사에는 많은 관계가 연결되어 있을 것이다. 이런 복잡한 구조의 데이터셋 정보를 질의하고 싶을 때가 있다. 이 경우, TEXT를 그래프 DB 질의 형식으로 변환하고, AI 에이전트를 처리해야 한다.

예를 들어, IFC(Industry Foundation Classes)란 파일은 객체 그래프 구조로 이와 유사한 형태를 가진다.

IFC 구조(NEO4J)

이 예에서는 이런 IFC와 같은 그래프 데이터셋을 어떻게 RAG 방식으로 정확한 질의 답변 결과를 얻을 수 있지는 지에 대한 기술을 설명한다.

아이디어는 이렇다. 먼저, IFC를 그래프 데이터셋으로 파싱한다. 그리고, 이 데이터셋을 Neo4j 그래프 데이터베이스로 변환한다.

앞의 Text To SQL과 유사하게 그래프 스키마 구조는 LLM에 질의할 프롬프트의 컨텍스트에 삽입한다.

# Schema information prompt

schema_info = """

BIM Graph Database Schema (Based on Actual Database Structure):

Node Labels (Each IFC class has its own label):

- Element: Generic element properties

- IFCFile: IFC file metadata

- IfcBeam, IfcBuilding, IfcBuildingStorey, IfcCovering, IfcDoor, IfcFooting

- IfcFurnishingElement, IfcMember, IfcOpeningElement, IfcRailing

- IfcRoof, IfcSite, IfcSlab, IfcSpace, IfcStair, IfcStairFlight

- IfcWall, IfcWallStandardCase, IfcWindow

Common Node Properties:

- description, globalId, name, objectType, properties, sourceFileId, tag

- All properties are stored directly in each node

IFCFile Properties:

- createdDate, fileId, fileName, filePath, fileSize, importDate, modifiedDate

Relationship Types:

- AGGREGATES: Element -> Element (aggregation relationships)

- BELONGS_TO_FILE: Element -> IFCFile (links elements to their source file)

- CONTAINED_IN: Element -> Element (spatial containment)

CRITICAL Schema Rules:

1. Use specific IFC labels (IfcSpace, IfcWall, etc.) NOT generic Element label

2. Properties are stored as nested JSON in the 'properties' field

3. Do NOT try to access specific nested property paths - they vary by modeling tool

4. For property-related queries: Always return full properties JSON for LLM analysis

5. Standard approach: MATCH (s:IfcSpace {{name: 'A204'}}) RETURN s.name, s.properties

Query Examples:

- Find space properties: MATCH (s:IfcSpace {{name: 'A204'}}) RETURN s.properties

- Count walls: MATCH (w:IfcWall) RETURN count(w)

- Find all doors: MATCH (d:IfcDoor) RETURN d.name, d.properties LIMIT 10

- Find file info: MATCH (f:IFCFile) RETURN f.fileName, f.fileSize

- Get space with properties: MATCH (s:IfcSpace {{name: 'A204'}}) RETURN s.name, s.properties

- Find space by room name: MATCH (s:IfcSpace) WHERE s.properties CONTAINS 'A204' RETURN s.name, s.properties

"""

프롬프트는 다음과 같이 앞의 스키마 구조, 입력 질의와 더블어 명확한 그래프 Cypher 문법을 생성하도록 지시한다.

cypher_prompt = ChatPromptTemplate.from_messages([

("system", """You are an agent in converting natural language queries to Neo4j Cypher queries for BIM/IFC data.

""" + schema_info + """

IMPORTANT: Properties are stored as nested JSON. For property-related queries (area, volume, etc.):

- Return the full properties JSON: RETURN s.name, s.properties

- Let the response processor extract specific values from the JSON

- Don't try to access specific nested paths as they vary by modeling tool

Rules for Cypher generation:

1. Always use proper Cypher syntax

2. Use specific IFC labels (IfcSpace, IfcWall, IfcDoor, etc.) as node labels

3. Access element properties directly from the node (e.g., s.properties, s.name, s.globalId)

4. Do NOT traverse relationships for basic element properties

5. Use WHERE clauses for additional filtering by name, etc.

6. Use LIMIT to prevent large result sets (default LIMIT 100)

7. For counts, use count() function

8. For file information, match on IFCFile nodes

9. For property searches (area, volume, etc.), return the entire properties JSON and let response processing extract relevant values

10. Use simple property access: s.properties (return full JSON for analysis)

11. For specific known properties, try common patterns but always include full properties as backup

CORRECT Examples:

- MATCH (s:IfcSpace {{name: 'A204'}}) RETURN s.name, s.globalId, s.properties

- MATCH (w:IfcWall) RETURN count(w)

- MATCH (d:IfcDoor) RETURN d.name, d.globalId, d.properties LIMIT 10

- MATCH (f:IFCFile) RETURN f.fileName, f.fileSize

- For property queries: Always include s.properties in RETURN clause

WRONG Examples:

- MATCH (e:Element {{ifcClass: 'IfcSpace'}}) (Use direct IfcSpace label)

- Trying to access specific property paths like s.properties.PSet_Name.Property (paths vary by tool)

Generate ONLY the Cypher query without explanation or markdown formatting.

"""),

("user", "Convert this query to Cypher: {query}")

])

return cypher_prompt | self.cypher_generator | StrOutputParser()

다음은 그 결과를 보여준다.

관련 구현코드는 다음 링크를 참고한다.

mac999/BIM_graph_agent

결론

다양한 데이터셋에 대한 질의 목적 달성을 위해 솔류션 테크 트리를 탐색하고 시도했다. 아직 완벽하지 않고, 제약된 범위내에서 원활히 구현된다(AGI처럼 홍보하는 기술은 믿으면 안됨. 그들이 공개된 자료들도 에러가 많이 발생하였다). 앞에 예시된 내용은 그 테크트리 중 일부 성공한 것만 기술한 것이다.

사실, LLM 튜닝은 리소스 제약으로 인해 대부분 열악한 인프라 환경?인 국내에서는 RAG를 대안으로 선택하지만, 한계가 명확하다. 사실, LLM 튜닝 및 풀튜닝 할 수 있는 능력있는 국내 IT기업은 거의 없다고 봐야 한다. 참고로, 라마3 8B모델 훈련에는 H100 GPU(80GB) 하나를 사용했을 때 학습 기간은 1,388.9 개월이 걸린다(계산방법 참고 - Transformer Math 101).

Llama-3-8B-Instruct | NVIDIA NGC (How large an LLM can I train from scratch on a single A100 GPU with 80Gb memory?)

이보다 더 적은 파라메터수를 가진 라마2 7B 모델 훈련에는 114.5개월(약 10년)/A100 GPU 이 소요된다.

The Carbon Impact of Large Language Models: AI's Growing Environmental Cost

대안인 파인튜닝, 양자화, LoRA방식은 많은 학습 데이터(최소 수만건 이상 잘 정재된 데이터셋)이 필요하고, 그 결과 또한 한계가 있다(계산된 logit 확률로만 이야기하는 논문이나 리더보드 발표 결과를 믿을 수 없음. 실제 결과는 사용하기 어려운 경우가 많다). 이런 이유로 RAG를 하지만, 이 또한, 연구가 필요한 이슈들이 많다.

좀 더 상세한 내용은 아래 링크를 참고한다.

레퍼런스