편집자 주: 이 글은 Dhruv Ateja가 작성한 게스트 블로그 포스트입니다. 에이전트를 사용하여 데이터를 쿼리(Query)하고 해당 데이터를 표시하는 방법까지 선택하는 풀스택 애플리케이션 구축에 대해 다룹니다. 이 프로젝트는 LangGraph와 LangGraph Cloud를 활용합니다.

주요 링크:

LangGraph Cloud의 스트리밍 API (Streaming API)를 활용하여 데이터 시각화 에이전트를 만드는 흥미로운 프로젝트를 살펴보겠습니다. SQLite 데이터베이스나 CSV 파일을 업로드하고 데이터에 대해 질문하면, 에이전트가 적절한 시각화 결과물을 생성합니다. 이 블로그는 에이전트의 워크플로우 (Workflow)와 주요 기능에 대해 간략히 살펴봅니다.

0:00 /0:281×

전체 워크플로우는 LangGraph Cloud를 사용하여 오케스트레이션 (Orchestration)됩니다. LangGraph Cloud는 복잡한 AI 에이전트를 쉽게 구축할 수 있는 프레임워크, 실시간 업데이트를 위한 스트리밍 API (Streaming API), 그리고 에이전트의 동작을 모니터링하고 실험할 수 있는 비주얼 스튜디오 (Visual Studio)를 제공합니다.

먼저, 현재의 SOTA (State-of-the-Art) Text-to-SQL 워크플로우를 살펴보겠습니다:

스키마 (Schema) 및 메타데이터 (Metadata) 추출:

• 시스템은 제공된 데이터베이스(예: SQLite 또는 CSV)를 처리하여 테이블 구조 및 컬럼 (Column) 상세 정보와 같은 중요한 정보를 추출합니다.
• 이 초기 단계는 데이터베이스의 구성에 대한 포괄적인 이해를 제공합니다.

임베딩 (Embedding) 생성:

• 더 큰 데이터셋의 경우, 스키마 요소(테이블, 컬럼) 및 샘플 데이터에 대한 임베딩 (Embeddings)이 생성됩니다. 이러한 임베딩은 이후 검색 및 매칭 작업 시 효율성을 높여줍니다.

엔티티 (Entity) 및 컨텍스트 (Context) 검색:

• 사용자의 질의를 분석하여 엔티티와 전반적인 컨텍스트를 식별합니다.
• 데이터베이스 값의 경우, LSH (Locality Sensitive Hashing) 인덱스를 활용한 구문 검색 (Syntactic search)을 구현할 수 있습니다.

검색 증강 생성 (RAG, Retrieval-Augmented Generation)을 사용한 관련 테이블 추출:

• 이 단계에서는 RAG (Retrieval-Augmented Generation)를 활용하여 사용자가 찾는 정보를 담고 있는 관련 테이블을 정확히 찾아냅니다.

실험적 접근 방식 (Experimental Approaches):

• 스키마 (Schema)가 컨텍스트 윈도우 (Context Window) 내에서 관리 가능한 수준이라면, 이 단계를 건너뛸 수 있습니다.
• 멀티 홉 (Multi-hop) 기능을 위한 지식 그래프 (Knowledge Graph) 기반의 RAG 탐색은 향후 개발을 위한 잠재적인 방안입니다.
• 관련 컬럼 (Column)을 추출하여 RAG에 입력하면 더욱 정밀한 테이블 추출이 가능합니다.

대규모 스키마 처리 (Large Schema Handling):

• 방대한 테이블 스키마를 다룰 때는 이를 효과적으로 관리하기 위해 다음과 같은 기술을 사용할 수 있습니다:
• 스키마 세부 정보를 필수 정보로 축소하여 효율적인 처리를 보장합니다.
• 메타데이터 태그 (Metadata Tag)를 기반으로 컬럼을 가지치기 (Pruning) 하여 분석을 간소화합니다.
• 가지치기 된 스키마를 언어 모델 (LLM, Large Language Model)에 제공하여 테이블 관련성을 평가합니다.

테이블 및 관련성 검증 (Table and Relevance Validation):

• 추출된 테이블들이 사용자의 질의와 실제로 관련이 있는지 세심하게 검증합니다.

SQL 쿼리 생성 (SQL Query Generation):

• 관련 테이블, 해당 스키마, 그리고 샘플 데이터 행을 LLM에 입력하여 SQL 쿼리를 생성합니다.

실험 (Experimentation): 필터링된 테이블 내 각 컬럼의 필요성을 평가하도록 LLM에 프롬프팅 (Prompting) 하고, 이를 사고의 사슬 (CoT, Chain-of-Thought) 설명과 결합하면 생성된 쿼리의 추론 근거에 대한 귀중한 통찰을 얻을 수 있습니다.

쿼리 구조 검증 (Query Structure Validation):

• 워크플로 (Workflow)가 생성된 SQL 쿼리의 구조를 검증하고 수정하여, 실행 전 정확성을 보장합니다.

저희 프로젝트에서는 더 작은 데이터셋에 집중했기 때문에 RAG나 LSH 기술의 필요성이 없었습니다. 하지만 핵심 워크플로는 동일하게 유지됩니다. 더 큰 데이터셋에 대한 Text-to-SQL 구현을 탐구하려면 Pinterest Engineering의 이 통찰력 있는 기사를 확인해 보세요.

다음은 Text-to-SQL 워크플로 구현의 개요입니다:

그래프 설정 (Setting up the graph)

def create_workflow(self) -> StateGraph:

"""워크플로 그래프를 생성하고 구성합니다."""

workflow = StateGraph(State)

그래프에 노드 추가

workflow.add_node("parse_question", self.sql_agent.parse_question)

workflow.add_node("get_unique_nouns", self.sql_agent.get_unique_nouns)

workflow.add_node("generate_sql", self.sql_agent.generate_sql)

workflow.add_node("validate_and_fix_sql", self.sql_agent.validate_and_fix_sql)

workflow.add_node("execute_sql", self.sql_agent.execute_sql)

workflow.add_node("format_results", self.sql_agent.format_results)

workflow.add_node("choose_visualization", self.sql_agent.choose_visualization)

workflow.add_node("format_data_for_visualization", self.data_formatter.format_data_for_visualization)

Define edges

workflow.add_edge("parse_question", "get_unique_nouns")

workflow.add_edge("get_unique_nouns", "generate_sql")

workflow.add_edge("generate_sql", "validate_and_fix_sql")

workflow.add_edge("validate_and_fix_sql", "execute_sql")

workflow.add_edge("execute_sql", "format_results")

workflow.add_edge("execute_sql", "choose_visualization")

workflow.add_edge("choose_visualization", "format_data_for_visualization")

workflow.set_entry_point("parse_question")

return workflow

1. 스키마 및 메타데이터 추출 (Schema and Metadata Extraction):

• 이 프로젝트를 위해 SQLite 파일을 저장하고 쿼리할 서버를 개발했습니다: https://github.com/DhruvAtreja/sqllite-server
• 이 서버는 데이터베이스 쿼리와 스키마 검색이라는 두 가지 주요 기능을 가지고 있습니다.
• 우리는 모든 테이블의 스키마와 컨텍스트를 위해 각 테이블의 처음 세 행을 추출합니다.

스키마 추출 (Extracting schema)

const db = new sqlite3.Database(dbPath);

db.all(

"SELECT name, sql FROM sqlite_master WHERE type='table';",

[],

(err, tables) => {

if (err) {

db.close();

return res.status(500).json({ error: err.message });
}

const schema = [];

const processTable = (index) => {

if (index >= tables.length) {

db.close();

return res.json({ schema: schema.join("\n") });
}

const { name: tableName, sql: createStatement } = tables[index];

schema.push(Table: ${tableName});

schema.push(CREATE statement: ${createStatement}\n);

db.all(SELECT * FROM '${tableName}' LIMIT 3;, [], (err, rows) => {

if (err) {

console.error(Error fetching rows for table ${tableName}:, err);

} else if (rows.length > 0) {

schema.push("Example rows:");

rows.forEach((row) => schema.push(JSON.stringify(row)));

}

schema.push(""); // 테이블 사이에 빈 줄 추가

processTable(index + 1);

});

};

processTable(0);

}

);

2. 사용자 질문 파싱 (Parsing the user's question):

• 데이터베이스의 스키마 (Schema)와 함께 사용자의 질문을 SQLAgent에 전달합니다. 이 데이터를 사용하여 관련 테이블과 컬럼 (Column)을 추출합니다.
• 또한 명사 (Noun)를 포함하는 컬럼을 식별합니다. 이것이 왜 중요한지는 다음 단계에서 확인하게 됩니다.
• 질문이 데이터베이스와 관련이 없거나 질문에 답하기 위한 정보가 충분하지 않은 경우, is_relevant를 false로 설정하고 워크플로 (Workflow)를 종료합니다.

프롬프트 (Prompt):

당신은 SQL 테이블을 요약하고 데이터베이스에 관한 사용자 질문을 파싱할 수 있는 데이터 분석가입니다.

질문과 데이터베이스 스키마가 주어지면, 관련 있는 테이블과 컬럼을 식별하세요.

질문이 데이터베이스와 관련이 없거나 질문에 답하기 위한 정보가 충분하지 않다면, is_relevant를 false로 설정하세요.

"noun_columns" 필드에는 질문과 관련이 있으면서 명사나 이름을 포함하는 컬럼만 포함해야 합니다. 예를 들어, "Artist name" 컬럼은 "What are the top selling artists?"라는 질문과 관련된 명사를 포함하지만, "Artist ID" 컬럼은 명사를 포함하지 않으므로 관련이 없습니다. 숫자를 포함하는 컬럼은 포함하지 마세요.

응답 형식 (Response Format):

{ "is_relevant": boolean, "relevant_tables": [ { "table_name": string, "columns": [string], "noun_columns": [string] } ] }

3. 고유 명사 가져오기 (Getting the unique nouns):

• 만약 사용자가 "가장 많이 팔린 아티스트는 누구인가요?" 또는 "각 카테고리의 시장 점유율은 어떻게 되나요?"라고 질문한다면, 올바른 SQL 쿼리 (SQL query)를 생성하기 위해 어떤 아티스트나 카테고리를 지칭하는지 알아야 합니다.
• 만약 사용자가 "Top selling ac dc songs?"라고 질문한다면 (우리 모두 그 노래가 Thunderstruck라는 것을 알고 있지만), 테이블에는 "ac dc" 대신 "AC/DC"라는 이름이 들어있다면 어떻게 될까요? 올바른 SQL 쿼리를 생성하기 위해서는 아티스트 이름의 정확한 철자를 알아내야 합니다.
• 여기서 고유 명사 (unique nouns)가 필요합니다. 우리는 질문과 스키마 (schema)에서 고유 명사를 추출하여 서로 매칭합니다. 고유 명사를 사용하면 아티스트 이름의 정확한 철자와 엔티티 (entities) 목록을 얻을 수 있습니다.

함수 (Function)

def get_unique_nouns(self, state: dict) -> dict:

"""관련 테이블 및 컬럼에서 고유 명사를 찾습니다."""

parsed_question = state['parsed_question']

if not parsed_question['is_relevant']:

return {"unique_nouns": []}

unique_nouns = set()

for table_info in parsed_question['relevant_tables']:

table_name = table_info['table_name']

noun_columns = table_info['noun_columns']

if noun_columns:

column_names = ', '.join(f"{col}" for col in noun_columns)

query = f"SELECT DISTINCT {column_names} FROM {table_name}"

results = self.db_manager.execute_query(state['uuid'], query)

for row in results:

unique_nouns.update(str(value) for value in row if value)

return {"unique_nouns": list(unique_nouns)}

4. SQL 쿼리 생성 (Generating the SQL query):

우리는 스키마, 사용자의 질문, 파싱된 질문 (parsed question), 그리고 고유 명사를 SQLAgent에 전달합니다. 컬럼 값이 null, "N/A" 또는 ""인 행은 건너뜁니다.

프롬프트 (Prompt)

당신은 사용자의 질문, 데이터베이스 스키마, 그리고 관련 테이블에서 발견된 고유 명사를 기반으로 SQL 쿼리를 생성하는 AI 어시스턴트입니다. 사용자의 질문에 답하기 위한 유효한 SQL 쿼리를 생성하세요.

SQL 쿼리를 작성하기 위한 정보가 충분하지 않다면, "NOT_ENOUGH_INFO"라고 응답하세요.

다음은 몇 가지 예시입니다:

1. 가장 많이 팔린 제품은 무엇인가요?

2. 각 제품별 총 매출액은 얼마인가요?

Answer: SELECT product_name, SUM(quantity * price) as total_revenue FROM sales WHERE product_name IS NOT NULL AND quantity IS NOT NULL AND price IS NOT NULL AND product_name != "" AND quantity != "" AND price != "" AND product_name != "N/A" AND quantity != "N/A" AND price != "N/A" GROUP BY product_name ORDER BY total_revenue DESC

3. 각 제품의 시장 점유율은 얼마인가요?

Answer: SELECT product_name, SUM(quantity) * 100.0 / (SELECT SUM(quantity) FROM sales) as market_share FROM sales WHERE product_name IS NOT NULL AND quantity IS NOT NULL AND product_name != "" AND quantity != "" AND product_name != "N/A" AND quantity != "N/A" GROUP BY product_name ORDER BY market_share DESC

4. 시간에 따른 소득 분포를 플롯하세요.

Answer: SELECT income, COUNT(*) as count FROM users WHERE income IS NOT NULL AND income != "" AND income != "N/A" GROUP BY income

결과는 다음 형식으로만 제공되어야 합니다:

[[x, y]]

또는

[[label, x, y]]

'남성과 여성의 요금 분포를 플롯하세요'와 같은 질문의 경우, 각 요금의 빈도를 계산하여 플롯해야 합니다. x축은 요금이 되어야 하고, y축은 해당 요금을 지불한 사람의 수(count)가 되어야 합니다.

어떤 열이라도 NULL이거나 "N/A" 또는 ""인 행은 건너뛰세요.

쿼리 문자열만 제공하세요. 형식 지정하지 마세요. 고유 명사는 제공된 고유 명사 목록의 정확한 철자를 사용해야 합니다.

데이터 전달

===Database schema:

{schema}

===User question:

{question}

===Relevant tables and columns:

{parsed_question}

===Unique nouns in relevant tables:

{unique_nouns}

5. SQL 쿼리 유효성 검사 및 수정:

• 우리는 SQL 쿼리를 SQLAgent에 전달합니다. SQLAgent는 쿼리가 유효한지, 쿼리에 사용된 모든 테이블과 컬럼이 관련이 있는지 확인하며, 유효하다면 SQL 쿼리를 반환합니다.
• 예를 들어, 데이터를 문자열(string)에서 날짜(date)나 정수(integer)로 변환해야 하는 경우가 있는데, 이 단계에서 이를 수정합니다.

Prompt

당신은 SQL 쿼리를 검증하고 수정하는 AI 어시스턴트입니다. 당신의 작업은 다음과 같습니다:

1. SQL 쿼리가 유효한지 확인합니다.

2. 모든 테이블 및 컬럼 이름의 철자가 정확한지, 그리고 스키마(schema)에 존재하는지 확인합니다.

3. 문제가 있다면 이를 수정하고 교정된 SQL 쿼리를 제공합니다.

4. 문제가 발견되지 않으면 원래의 쿼리를 반환합니다.

다음 구조를 가진 JSON 형식으로 응답하세요. JSON으로만 응답해야 합니다:

{{
"valid": boolean,
"issues": string 또는 null,
"corrected_query": string
}}

'''),

("human", '''===Database schema:

{schema}

===Generated SQL query:

{sql_query}

다음 구조를 가진 JSON 형식으로 응답하세요. JSON으로만 응답해야 합니다:

{{
"valid": boolean,
"issues": string 또는 null,
"corrected_query": string
}}

예시:

1. {{
   "valid": true,
   "issues": null,
   "corrected_query": "None"
   }}

2. {{
   "valid": false,
   "issues": "Column USERS does not exist",