BigQuery 및 Langchain을 사용하여 데이터 분석 도우미를 구축한 방법

기업은 매일 엄청난 양의 데이터를 생성하므로 이 모든 정보에서 유용한 통찰력을 끌어내는 것은 어려울 수 있으며, 특히 복잡한 데이터 세트와 엄청난 양의 데이터의 경우 더욱 그렇습니다. 하지만 생성적 AI를 사용하면 데이터 분석을 간소화하고 자동화하여 효율성과 접근성을 높일 수 있습니다. 이 글에서는 Google Langchain , OpenAI, BigQuery 및 DLP( Data Loss Prevention )를 사용하여 AI 데이터 분석 도우미를 설정하고 사용하는 방법을 보여 드리겠습니다.

사용 사례: BigQuery로 데이터 분석 자동화

솔루션 설계

이 솔루션에는 BigQuery 데이터세트와 상호작용하여 데이터 분석을 자동화하는 Langchain 및 OpenAI를 사용하여 Streamlit 앱을 설정하는 작업이 포함됩니다. 이 에이전트는 PII 고객 속성 마스킹 및 데이터 시각화와 같은 특정 작업을 위해 사용자 정의 도구를 사용합니다. 또한 상담원은 채팅 기록을 유지하여 상황에 맞는 정확한 응답을 보장하도록 구성됩니다.

솔루션 아키텍처의 다이어그램은 다음과 같습니다.

다음 테이블이 포함된 BigQuery 데이터 세트가 있는 시나리오를 고려해 보겠습니다.

• 고객 테이블 : 고객 데이터가 포함되어 있습니다.
• 연락처 테이블 : 고객 연락처 정보가 포함되어 있습니다.
• 고객 주소 테이블 : 고객을 주소로 연결합니다.
• 주소 테이블 : 주소 정보를 담고 있습니다.
• 작업 통계 테이블 : 데이터를 잘라 고객 프로필 테이블에 로드하는 ETL 일괄 작업 요약을 기록합니다.

랭체인 설정

랭체인이란 무엇인가요?

LangChain은 AI 개발자에게 언어 모델을 외부 데이터 소스와 연결하는 도구를 제공합니다. 오픈 소스이며 활발한 커뮤니티에서 지원됩니다. 조직은 LangChain을 무료로 사용할 수 있으며 프레임워크에 능숙한 다른 개발자로부터 지원을 받을 수 있습니다.

Langchain을 사용하여 데이터 분석을 수행하려면 먼저 Langchain 및 OpenAI 라이브러리를 설치해야 합니다. 필요한 라이브러리를 다운로드한 다음 프로젝트로 가져오면 됩니다.

Langchain을 설치하십시오:

 pip install langchain matplotlib pandas streamlit pip install -qU langchain-openai langchain-community

Langchain 모델을 정의하고 BigQuery 연결을 설정합니다.

 import os import re import streamlit as st from google.cloud import dlp_v2 from google.cloud.dlp_v2 import types from langchain.agents import create_sql_agent from langchain_community.vectorstores import FAISS from langchain_core.example_selectors import SemanticSimilarityExampleSelector from langchain_core.messages import AIMessage from langchain_core.prompts import ( SystemMessagePromptTemplate, PromptTemplate, FewShotPromptTemplate, ) from langchain_core.prompts.chat import ( ChatPromptTemplate, HumanMessagePromptTemplate, MessagesPlaceholder, ) from langchain.memory import ConversationBufferMemory from langchain_experimental.utilities import PythonREPL from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.sql_database import SQLDatabase from langchain.tools import Tool service_account_file = f"{os.getcwd()}/service-account-key.json" os.environ["OPENAI_API_KEY"] = ( "xxxxxx" ) os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = service_account_file model = ChatOpenAI(model="gpt-4o", temperature=0) project = "lively-metrics-295911" dataset = "customer_profiles" sqlalchemy_url = ( f"bigquery://{project}/{dataset}?credentials_path={service_account_file}" ) db = SQLDatabase.from_uri(sqlalchemy_url)

사용자 정의 도구 설정

에이전트의 기능을 향상시키기 위해 PII 데이터 마스킹 및 데이터 시각화와 같은 특정 작업을 위한 사용자 정의 도구를 설정할 수 있습니다.

1. Google Cloud DLP로 PII 마스킹

   데이터 개인 정보 보호는 매우 중요합니다. 출력에서 PII를 보호하기 위해 Google Cloud 데이터 손실 방지(DLP)를 활용할 수 있습니다. 우리는 DLP API를 호출하여 응답에 있는 모든 PII 데이터를 마스킹하는 사용자 정의 도구를 구축할 것입니다.

 def mask_pii_data(text): dlp = dlp_v2.DlpServiceClient() project_id = project parent = f"projects/{project_id}" info_types = [ {"name": "EMAIL_ADDRESS"}, {"name": "PHONE_NUMBER"}, {"name": "DATE_OF_BIRTH"}, {"name": "LAST_NAME"}, {"name": "STREET_ADDRESS"}, {"name": "LOCATION"}, ] deidentify_config = types.DeidentifyConfig( info_type_transformations=types.InfoTypeTransformations( transformations=[ types.InfoTypeTransformations.InfoTypeTransformation( primitive_transformation=types.PrimitiveTransformation( character_mask_config=types.CharacterMaskConfig( masking_character="*", number_to_mask=0, reverse_order=False ) ) ) ] ) ) item = {"value": text} inspect_config = {"info_types": info_types} request = { "parent": parent, "inspect_config": inspect_config, "deidentify_config": deidentify_config, "item": item, } response = dlp.deidentify_content(request=request) return response.item.value

2. 파이썬 REPL

   다음으로, LLM이 Python을 사용하여 데이터 시각화를 수행할 수 있도록 Python REPL을 활용하고 에이전트를 위한 사용자 정의 도구를 정의하겠습니다.

 python_repl = PythonREPL()

이제 mask_pii_data 및 python_repl:

 def sql_agent_tools(): tools = [ Tool.from_function( func=mask_pii_data, name="mask_pii_data", description="Masks PII data in the input text using Google Cloud DLP.", ), Tool( name="python_repl", description=f"A Python shell. Use this to execute python commands. \ Input should be a valid python command. \ If you want to see the output of a value, \ you should print it out with `print(...)`.", func=python_repl.run, ), ] return tools

Few-Shot 예제 사용

모델에 몇 가지 예시를 제공하면 응답을 안내하고 성능을 향상하는 데 도움이 됩니다.

샘플 SQL 쿼리를 정의합니다.

 # Example Queries sql_examples = [ { "input": "Count of Customers by Source System", "query": f""" SELECT source_system_name, COUNT(*) AS customer_count FROM `{project}.{dataset}.customer` GROUP BY source_system_name ORDER BY customer_count DESC; """, }, { "input": "Average Age of Customers by Gender", "query": f""" SELECT gender, AVG(EXTRACT(YEAR FROM CURRENT_DATE()) - EXTRACT(YEAR FROM dob)) AS average_age FROM `{project}.{dataset}.customer` GROUP BY gender; """, }, ... ]

다음으로 몇 번의 프롬프트 템플릿에 예제를 추가합니다.

 example_selector = SemanticSimilarityExampleSelector.from_examples( sql_examples, OpenAIEmbeddings(), FAISS, k=2, input_keys=["input"], )

다음으로, 접두사와 접미사를 정의한 다음, few_shot_prompt from_messages 팩토리 메소드에 직접 전달하세요.

 PREFIX = """ You are a SQL expert. You have access to a BigQuery database. Identify which tables can be used to answer the user's question and write and execute a SQL query accordingly. Given an input question, create a syntactically correct SQL query to run against the dataset customer_profiles, then look at the results of the query and return the answer. Unless the user specifies a specific number of examples they wish to obtain, always limit your query to at most {top_k} results. You can order the results by a relevant column to return the most interesting examples in the database. Never query for all the columns from a specific table; only ask for the relevant columns given the question. You have access to tools for interacting with the database. Only use the information returned by these tools to construct your final answer. You MUST double check your query before executing it. If you get an error while executing a query, rewrite the query and try again.DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database. If the question does not seem related to the database, just return "I don't know" as the answer. If the user asks for a visualization of the results, use the python_agent tool to create and display the visualization. After obtaining the results, you must use the mask_pii_data tool to mask the results before providing the final answer. """ SUFFIX = """Begin! {chat_history} Question: {input} Thought: I should look at the tables in the database to see what I can query. Then I should query the schema of the most relevant tables. {agent_scratchpad}""" few_shot_prompt = FewShotPromptTemplate( example_selector=example_selector, example_prompt=PromptTemplate.from_template( "User input: {input}\nSQL query: {query}" ), prefix=PREFIX, suffix="", input_variables=["input", "top_k"], example_separator="\n\n", ) messages = [ SystemMessagePromptTemplate(prompt=few_shot_prompt), MessagesPlaceholder(variable_name="chat_history"), HumanMessagePromptTemplate.from_template("{input}"), AIMessage(content=SUFFIX), MessagesPlaceholder(variable_name="agent_scratchpad"), ] prompt = ChatPromptTemplate.from_messages(messages)

변수 설명

• input : 사용자의 입력 또는 쿼리입니다.

• Agent_scratchpad : 중간 단계나 생각을 위한 임시 저장 영역입니다.

• chat_history : 컨텍스트를 유지하기 위해 이전 상호 작용을 추적합니다.

• handler_parsing_errors : 에이전트가 구문 분석 오류를 정상적으로 처리하고 복구할 수 있도록 합니다.

• memory : 채팅 기록을 저장하고 검색하는 데 사용되는 모듈입니다.

  
  

이제 마지막 단계를 시작할 시간입니다. 앱을 만들어보자!

Streamlit을 사용하여 LLM 앱 구축

방금 구축한 Langchain 에이전트를 테스트하기 위한 대화형 인터페이스를 생성하려면 Streamlit을 사용할 수 있습니다.

 st.title("Data Analysis Assistant") if "history" not in st.session_state: st.session_state.history = [] user_input = st.text_input("Ask your question:") if st.button("Run Query"): if user_input: with st.spinner("Processing..."): st.session_state.history.append(f"User: {user_input}") response = agent_executor.run(input=user_input) if "sandbox:" in response: response = response.replace(f"sandbox:", "") match = re.search(r"\((.+\.png)\)", response) if match: image_file_path = match.group(1) if os.path.isfile(image_file_path): st.session_state.history.append({"image": image_file_path}) else: st.error("The specified image file does not exist.") else: st.session_state.history.append(f"Agent: {response}") st.experimental_rerun() else: st.error("Please enter a question.") for message in st.session_state.history: if isinstance(message, str): st.write(message) elif isinstance(message, dict) and "image" in message: st.image(message["image"])

우리는 모든 것을 설정했습니다. Streamlit 앱을 실행해보자

 streamlit run app.py

몇 가지 분석 질문을 통해 테스트해 보세요.

결론

Langchain과 OpenAI를 활용하면 복잡한 데이터 분석 작업을 자동화하여 대규모 데이터 세트에서 통찰력을 훨씬 쉽게 얻을 수 있습니다. 이 접근 방식은 시간을 절약할 뿐만 아니라 정확하고 일관된 분석을 보장합니다. 고객 프로필, 연락처 정보, 작업 통계 등 어떤 작업을 하든 AI 기반 데이터 분석 도우미는 데이터 처리 기능을 크게 향상시킬 수 있습니다. 전체 소스 코드를 보려면 다음을 방문하세요. GitHub 저장소 .