Comment j'ai créé un assistant d'analyse de données avec BigQuery et Langchain

Étant donné que les entreprises génèrent quotidiennement d’énormes quantités de données, il peut être difficile d’extraire des informations utiles de toutes ces informations, en particulier avec des ensembles de données complexes et d’énormes volumes de données. Mais grâce à l’IA générative, nous pouvons rationaliser et automatiser l’analyse des données, la rendant efficace et accessible. Dans cet article, je vais vous montrer comment configurer et utiliser un assistant d'analyse de données IA à l'aide de Google Langchain , OpenAI, BigQuery et Data Loss Prevention (DLP).

Cas d'utilisation : Automatiser l'analyse des données avec BigQuery

Conception de solutions

La solution consiste à configurer une application Streamlit utilisant Langchain et OpenAI qui interagit avec l'ensemble de données BigQuery pour automatiser l'analyse des données. Cet agent utilisera des outils personnalisés pour des tâches spécifiques telles que le masquage des attributs client PII et la visualisation des données. De plus, l'agent sera configuré pour conserver l'historique des discussions, garantissant ainsi des réponses contextuellement précises.

Voici un schéma de l'architecture de la solution :

Prenons un scénario dans lequel nous disposons d'un ensemble de données BigQuery contenant les tables suivantes :

• Table Client : Contient les données client.
• Table de contact : contient les coordonnées du client.
• Table d'adresses client : relie les clients aux adresses.
• Table d'adresses : contient des informations sur l'adresse.
• Tableau des statistiques des tâches : enregistre les résumés des tâches par lots ETL qui tronquent et chargent les données dans les tables de profil client.

Configuration de Langchain

Qu’est-ce que Langchain ?

LangChain fournit aux développeurs d'IA des outils pour connecter des modèles de langage à des sources de données externes. Il est open source et soutenu par une communauté active. Les organisations peuvent utiliser LangChain gratuitement et bénéficier du soutien d'autres développeurs maîtrisant le framework.

Pour effectuer une analyse de données à l'aide de Langchain, nous devons d'abord installer les bibliothèques Langchain et OpenAI. Cela peut être fait en téléchargeant les bibliothèques nécessaires, puis en les important dans votre projet.

Installez Langchain :

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

Définissez le modèle Langchain et configurez la connexion 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)

Configuration d'outils personnalisés

Pour améliorer les capacités de notre agent, nous pouvons configurer des outils personnalisés pour des tâches spécifiques, telles que le masquage des données PII et la visualisation des données.

1. Masquage des informations personnelles avec Google Cloud DLP

   La confidentialité des données est cruciale. Pour protéger les informations personnelles dans les sorties, nous pouvons utiliser Google Cloud Data Loss Prevention (DLP). Nous allons créer un outil personnalisé qui appelle l'API DLP pour masquer toutes les données PII présentes dans la réponse.

 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 Python

   Ensuite, pour permettre au LLM d'effectuer la visualisation des données à l'aide de Python, nous exploiterons Python REPL et définirons un outil personnalisé pour notre agent.

 python_repl = PythonREPL()

Créons maintenant les outils d'agent, qui incluront mask_pii_data et 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

Utiliser des exemples de quelques plans

Fournir au modèle quelques exemples concrets permet de guider ses réponses et d’améliorer ses performances.

Définir des exemples de requêtes 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; """, }, ... ]

Ensuite, ajoutez des exemples au modèle d'invite en quelques étapes.

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

Ensuite, définissez le préfixe et le suffixe, puis transmettez few_shot_prompt directement dans la méthode d'usine 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)

Explication des variables

• input : L’entrée ou la requête de l’utilisateur.

• agent_scratchpad : Une zone de stockage temporaire pour les étapes ou pensées intermédiaires.

• chat_history : garde une trace des interactions précédentes pour maintenir le contexte.

• handle_parsing_errors : garantit que l'agent peut gérer et récupérer correctement les erreurs d'analyse.

• mémoire : Le module utilisé pour stocker et récupérer l'historique des discussions.

  
  

Il est temps de passer à la dernière étape. Créons l'application !

Créer une application LLM avec Streamlit

Pour créer une interface interactive pour tester l'agent Langchain que nous venons de construire, nous pouvons utiliser 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"])

Nous avons tout mis en place. Lançons l'application Streamlit

 streamlit run app.py

et testez-le en posant quelques questions d’analyse.

Conclusion

En tirant parti de Langchain et d'OpenAI, nous pouvons automatiser des tâches complexes d'analyse de données, ce qui facilite grandement l'obtention d'informations à partir de grands ensembles de données. Cette approche permet non seulement de gagner du temps, mais garantit également une analyse précise et cohérente. Que vous travailliez avec des profils clients, des informations de contact ou des statistiques d'emploi, un assistant d'analyse de données alimenté par l'IA peut considérablement améliorer vos capacités de traitement des données. Pour le code source complet, visitez le Dépôt GitHub .