Cách tôi xây dựng Trợ lý phân tích dữ liệu với BigQuery và Langchain

Vì các doanh nghiệp tạo ra lượng dữ liệu khổng lồ hàng ngày nên việc rút ra những hiểu biết hữu ích từ tất cả thông tin này có thể khó khăn, đặc biệt là với các bộ dữ liệu phức tạp và khối lượng dữ liệu khổng lồ. Nhưng với AI tổng quát, chúng ta có thể hợp lý hóa và tự động hóa việc phân tích dữ liệu, giúp dữ liệu trở nên hiệu quả và dễ tiếp cận. Trong bài viết này, tôi sẽ hướng dẫn bạn cách thiết lập và sử dụng trợ lý phân tích dữ liệu AI bằng Google Langchain , OpenAI, BigQuery và Ngăn chặn mất dữ liệu (DLP).

Ca sử dụng: Tự động hóa phân tích dữ liệu với BigQuery

Thiết kế giải pháp

Giải pháp này bao gồm việc thiết lập ứng dụng Streamlit bằng Langchain và OpenAI tương tác với bộ dữ liệu BigQuery để tự động hóa phân tích dữ liệu. Nhân viên hỗ trợ này sẽ sử dụng các công cụ tùy chỉnh cho các nhiệm vụ cụ thể như che giấu các thuộc tính PII của khách hàng và trực quan hóa dữ liệu. Ngoài ra, tổng đài viên sẽ được định cấu hình để lưu giữ lịch sử trò chuyện, đảm bảo phản hồi chính xác theo ngữ cảnh.

Đây là sơ đồ của kiến trúc giải pháp:

Hãy xem xét một tình huống trong đó chúng ta có tập dữ liệu BigQuery chứa các bảng sau:

• Bảng khách hàng : Chứa dữ liệu khách hàng.
• Bảng liên hệ : Chứa thông tin liên hệ của khách hàng.
• Bảng địa chỉ khách hàng : Liên kết khách hàng với các địa chỉ.
• Bảng địa chỉ : Chứa thông tin địa chỉ.
• Bảng thống kê công việc : Ghi nhật ký tóm tắt công việc hàng loạt ETL cắt ngắn và tải dữ liệu vào bảng hồ sơ khách hàng

Thiết lập Langchain

Langchain là gì?

LangChain cung cấp cho các nhà phát triển AI các công cụ để kết nối các mô hình ngôn ngữ với các nguồn dữ liệu bên ngoài. Nó là nguồn mở và được hỗ trợ bởi một cộng đồng tích cực. Các tổ chức có thể sử dụng LangChain miễn phí và nhận được hỗ trợ từ các nhà phát triển khác thành thạo trong khuôn khổ này.

Để thực hiện phân tích dữ liệu bằng Langchain, trước tiên chúng ta cần cài đặt thư viện Langchain và OpenAI. Điều này có thể được thực hiện bằng cách tải xuống các thư viện cần thiết và sau đó nhập chúng vào dự án của bạn.

Cài đặt Langchain:

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

Xác định Mô hình Langchain và Thiết lập kết nối 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)

Thiết lập công cụ tùy chỉnh

Để nâng cao khả năng của tổng đài viên, chúng tôi có thể thiết lập các công cụ tùy chỉnh cho các tác vụ cụ thể, chẳng hạn như che giấu dữ liệu PII và trực quan hóa dữ liệu.

1. Che giấu PII bằng Google Cloud DLP

   Quyền riêng tư dữ liệu là rất quan trọng. Để bảo vệ PII ở đầu ra, chúng tôi có thể sử dụng tính năng Ngăn chặn mất dữ liệu trên đám mây của Google (DLP). Chúng tôi sẽ xây dựng một công cụ tùy chỉnh gọi API DLP để che giấu mọi dữ liệu PII có trong phản hồi.

 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. Python REPL

   Tiếp theo, để cho phép LLM thực hiện trực quan hóa dữ liệu bằng Python, chúng tôi sẽ tận dụng REPL của Python và xác định công cụ tùy chỉnh cho tác nhân của mình.

 python_repl = PythonREPL()

Bây giờ, hãy tạo các công cụ tác nhân, bao gồm mask_pii_data và 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

Sử dụng các ví dụ ít ảnh

Việc cung cấp cho mô hình một vài ví dụ ngắn gọn sẽ giúp định hướng phản hồi của mô hình và cải thiện hiệu suất.

Xác định các truy vấn SQL mẫu,

 # 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; """, }, ... ]

Tiếp theo, thêm ví dụ vào mẫu lời nhắc vài cảnh quay.

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

Tiếp theo, xác định tiền tố và hậu tố, sau đó chuyển few_shot_prompt trực tiếp vào phương thức xuất xưởng 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)

Giải thích các biến

• đầu vào : Đầu vào hoặc truy vấn của người dùng.

• Agent_scratchpad : Vùng lưu trữ tạm thời cho các bước hoặc suy nghĩ trung gian.

• chat_history : Theo dõi các tương tác trước đó để duy trì ngữ cảnh.

• hand_parsing_errors : Đảm bảo tác nhân có thể xử lý và khôi phục một cách linh hoạt sau các lỗi phân tích cú pháp.

• bộ nhớ : Mô-đun dùng để lưu trữ và truy xuất lịch sử trò chuyện.

  
  

Đã đến lúc thực hiện bước cuối cùng. Hãy xây dựng ứng dụng!

Xây dựng ứng dụng LLM với Streamlit

Để tạo giao diện tương tác nhằm kiểm tra tác nhân Langchain mà chúng ta vừa xây dựng, chúng ta có thể sử dụng 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"])

Chúng tôi đã thiết lập mọi thứ. Hãy chạy ứng dụng Streamlit

 streamlit run app.py

và kiểm tra nó bằng cách đặt một vài câu hỏi phân tích.

Phần kết luận

Bằng cách tận dụng Langchain và OpenAI, chúng tôi có thể tự động hóa các tác vụ phân tích dữ liệu phức tạp, giúp việc thu thập thông tin chi tiết từ các bộ dữ liệu lớn trở nên dễ dàng hơn nhiều. Cách tiếp cận này không chỉ tiết kiệm thời gian mà còn đảm bảo phân tích chính xác và nhất quán. Cho dù bạn đang làm việc với hồ sơ khách hàng, thông tin liên hệ hay thống kê công việc, trợ lý phân tích dữ liệu được hỗ trợ bởi AI có thể cải thiện đáng kể khả năng xử lý dữ liệu của bạn. Để có mã nguồn đầy đủ, hãy truy cập Kho lưu trữ GitHub .