利用OpenAI、LangChain和Streamlit进行智能数据分析和可视化

如何从海量数据中提取有价值的信息，并将其转化为直观、易于理解的视觉形式，是当下亟待解决的一个问题。幸运的是，随着人工智能技术的飞速发展，特别是自然语言处理（NLP）和机器学习（ML）领域的突破，我们有了更为强大的工具来应对这一挑战。今天我们来聊一聊如何利用OpenAI、LangChain和Streamlit构建一个智能数据分析和可视化系统。

一、技术背景

（一）OpenAI

OpenAI 以其强大的语言模型而闻名，如 GPT-4o-mini。这些模型经过大量数据的训练，具备出色的自然语言理解和生成能力(OpenAI o1与GPT4o的对比分析)。在数据分析场景中，它们可以理解用户的自然语言查询，将其转化为可执行的操作，如生成 SQL 查询语句，从而实现与数据库的交互。此外，还能对数据分析结果进行解读，以自然语言的形式为用户提供深入的见解，帮助用户更好地理解数据背后的含义。

（二）LangChain

LangChain 是一个专门为开发基于语言模型的应用程序而设计的框架。它提供了一系列工具和接口，简化了与语言模型的集成过程。在数据分析与可视化项目中，LangChain 能够连接语言模型与外部数据源（如 BigQuery 数据库），实现数据的获取、转换和处理。它还支持构建复杂的工作流程，将多个数据处理步骤串联起来，形成一个完整的数据处理管道，从用户查询到最终的可视化呈现，确保整个过程的顺畅运行。

（三）Streamlit

Streamlit 是一个用于创建交互式 Web 应用程序的 Python 库。其主要优势在于能够快速将数据脚本转化为可共享的 Web 应用，无需过多关注前端开发细节。在数据分析领域，Streamlit 提供了丰富的可视化组件，如折线图、柱状图、地图等，可直接用于展示数据分析结果。它还支持实时交互，用户可以通过界面输入参数、调整查询条件，实时更新可视化内容，极大地提升了数据分析的灵活性和效率。

二、系统架构设计

（一）整体架构概述

用户通过 Streamlit 构建的用户界面输入自然语言查询，该查询首先被传递到 LangChain 进行处理。LangChain 利用 OpenAI 的语言模型将查询转换为 SQL 语句，并执行该语句从 BigQuery 数据库中获取数据。获取的数据经过一系列处理步骤，包括坐标提取、可视化推荐和数据转换，最终以可视化的形式在 Streamlit 界面上呈现给用户。

（二）核心组件功能

1. 用户界面（Streamlit）

   • 作为用户与系统交互的入口，提供简洁直观的查询输入框。用户在此输入自然语言问题，如 “查询客户数量最多的前 5 个来源系统”。

   • 负责展示数据分析结果，包括可视化图表（如柱状图、折线图等）和文本形式的解释说明。例如，以柱状图展示不同来源系统的客户数量对比，并在下方给出相应的文字解读。

2. 文本到 SQL 转换（LangChain + OpenAI）

   • 使用 GPT - 4o-mini 模型将用户输入的自然语言问题转换为语法正确的 SQL 查询语句。例如，将上述用户问题转换为 “SELECT source_system_name, COUNT (*) as count FROM customer GROUP BY source_system_name ORDER BY count DESC LIMIT 5”。

   • 对生成的 SQL 语句进行检查和优化，避免常见错误，如数据类型不匹配、使用错误的函数等。

3. 查询执行（LangChain + BigQuery）

   • 连接到 BigQuery 数据库，执行转换后的 SQL 查询语句，获取相关数据。例如，从包含客户信息的数据库表中检索出符合条件的数据记录。

   • 处理查询执行过程中的错误，如数据库连接问题、查询语法错误等，并向用户提供相应的错误提示。

4. 坐标提取（自定义函数）

   • 当查询结果包含地理数据（如地址信息）时，从结果中提取纬度和经度信息。例如，从包含客户地址的数据库表中提取出地址对应的经纬度坐标。

   • 对提取的坐标数据进行整理和验证，确保其准确性和可用性，以便后续用于地图可视化。

5. 可视化推荐（LangChain + OpenAI）

   • 根据用户问题、SQL 查询语句和查询结果，利用 GPT - 4o-mini 模型推荐(大模型哪个好用？)最合适的可视化类型。例如，对于比较不同类别数据分布的问题，推荐使用柱状图；对于展示时间序列数据趋势的问题，推荐使用折线图。

   • 提供可视化推荐的理由，帮助用户理解为什么选择特定的可视化类型，增强系统的透明度和可解释性。

6. 数据转换（LangChain + OpenAI）

   • 根据推荐的可视化类型，使用 GPT - 4o-mini 模型将 SQL 查询结果转换为适合该可视化类型的格式。例如，对于柱状图，将数据转换为包含标签（类别）和值（数量）的 JSON 格式。

   • 确保转换后的数据结构与所选可视化工具（如 Streamlit 或 Plotly）兼容，以便能够正确地进行可视化渲染。

7. 可视化渲染（Streamlit + Plotly）

   • 使用 Streamlit 和 Plotly 库将转换后的数据渲染成可视化（Multi-Agent实战：构建复杂的数据处理与可视化系统）图表。例如，利用 Plotly 的柱状图绘制函数，根据数据生成直观的柱状图展示在 Streamlit 界面上。

   • 支持多种可视化类型的渲染，包括地图可视化（当有地理数据时），以提供丰富多样的数据分析视角。

三、核心代码

文本到 SQL 转换

from langchain.chains import create_sql_query_chain​sql_prompt = PromptTemplate.from_template("""            You are a SQL expert with access to a BigQuery dataset containing customers and customer addresses.            Given an input question, generate a syntactically correct SQL query to answer it. Unless explicitly requested otherwise, limit the results to {top_k} rows.​            Relevant Table Information:            {table_info}​            Question: {input}​            Guidelines:            1. Ensure that all attribute searches are case-insensitive.            2. ALWAYS add 'LIMIT {top_k}' at the end of the query unless:              - The question explicitly asks for all records              - The query uses GROUP BY and needs to show all groups              - The query is counting records (using COUNT)              - The query calculates aggregates that need all data​            Address and Location Queries:            1. For questions about addresses, locations, or properties, always include latitude and longitude columns in the SELECT clause.​            Double check the user's sql query for common mistakes, including:            - Using NOT IN with NULL values            - Using UNION when UNION ALL should have been used            - Using BETWEEN for exclusive ranges            - Data type mismatch in predicates            - Properly quoting identifiers            - Using the correct number of arguments for functions            - Casting to the correct data type            - Using the proper columns for joins            - Missing LIMIT clause when returning raw records​            If there are any of the above mistakes, rewrite the query.            If there are no mistakes, just reproduce the original query with no further commentary.​            Provide only the final SQL query as plain text without any formatting.            If the question is not about customers or addresses, respond with "I don't know"            """)​text_to_sql = create_sql_query_chain(            model, db, sql_prompt)

坐标提取

def extract_coordinates(self, result: dict):    try:        if isinstance(result, dict):            if "result" in result:                result = result["result"]                if isinstance(result, dict) and "result" in result:                    result_str = result["result"]                else:                    result_str = str(result)            else:                return None        else:            return None​        try:            if isinstance(result_str, str):                result_data = eval(result_str)            else:                result_data = result_str        except Exception as e:            print(f"Error evaluating result string: {e}")            return None​        if not isinstance(result_data, list):            return None​        unique_lat_values = set()        unique_long_values = set()​        for row in result_data:            if isinstance(row, dict):                if 'latitude' in row and row['latitude'] is not None:                    try:                        unique_lat_values.add(float(row['latitude']))                    except (ValueError, TypeError):                        pass​                if 'longitude' in row and row['longitude'] is not None:                    try:                        unique_long_values.add(float(row['longitude']))                    except (ValueError, TypeError):                        pass​        if unique_lat_values and unique_long_values:            return {                "latitude": list(unique_lat_values),                "longitude": list(unique_long_values)            }        return None​    except Exception as e:        print(f"Error extracting coordinates: {e}")        return None

可视化推荐

viz_prompt = PromptTemplate.from_template("""You are an AI assistant that recommends appropriate data visualizations for customer and address analytics. Based on the user's question, SQL query, and query results, suggest the most suitable type of graph or chart to visualize the data.​Available chart types and their best use cases:​- Bar Graphs (for 3+ categories):   * Comparing distributions across multiple categories  * Customer counts by source system  * Customer demographics across regions/states  * Age group distributions  * Monthly/yearly registration counts​- Horizontal Bar Graphs (for 2-3 categories or large value disparities):  * Binary comparisons (e.g., gender distribution)  * Limited category comparisons (2-3 items)  * Cases with large value differences between categories​- Line Graphs (for time series only):  * Customer registration trends over time  * Growth patterns by source system  * Any metric tracked over time periods  Note: X-axis MUST represent time (create_timestamp or similar)​- Pie Charts (for proportions, 3-7 categories max):  * Distribution percentages  * Market share analysis  * Proportional comparisons  Note: Total should sum to 100%​- Scatter Plots (for numeric relationships):  * Age vs other numeric metrics  * Timestamp patterns  * Distribution analysis  Note: Both axes must be numeric, non-categorical​Special Cases:1. Geographic Data:  * If result contains latitude and longitude → No chart (will display map)  * For address/location questions → No chart (will display map)​2. Raw Data:  * Individual customer records → No chart (tabular display)  * Non-aggregated data → No chart (tabular display)​Tables in scope:- customer: customer_key, first_name, last_name, source_system_name, dob, gender, create_timestamp- customer_address: customer_key, address_key- address: address_key, full_address, state, country, latitude, longitude​Question: {question}SQL Query: {query}SQL Result: {result}​Provide your response in the following format:Recommended Visualization: [Chart type or "none"]. ONLY use the following names: bar, horizontal_bar, line, pie, scatter, noneReason: [Brief explanation for your recommendation]""")​@RunnableLambdadef transform_data_for_visualization_chain(args, config):    try:        dispatch_custom_event(            "process.transform_data_for_visualization_chain",            {                "status": ""            },            config=config        )        chart_type = args.get("visualization").get("type")        result = args.get("result")​        if not chart_type or not result:            return {"chart_data": None}​        if chart_type == 'bar':            transform_prompt = bar_prompt        elif chart_type == 'horizontal_bar':            transform_prompt = horizontal_bar_prompt        elif chart_type == 'pie':            transform_prompt = pie_prompt        elif chart_type == 'scatter':            transform_prompt = scatter_prompt        elif chart_type == 'line':            transform_prompt = line_prompt        else:            transform_prompt = None​        assign_chart_type_and_result = RunnableLambda(            lambda args: {**args, "chart_type": args.get("visualization", {}).get(                "type"), "result": args.get("result")}        )​        if transform_prompt:            transform_chain = (                assign_chart_type_and_result                | transform_prompt                | model            )            return transform_chain​        return {"chart_data": None}​    except Exception as e:        print(e)        print(f"Error in transform_data_for_visualization: {e}")        return {"chart_data": None}​​viz_prompt | model| parse_visualization_response

其它代码获取地址：https://github.com/yai333/AI-DATA-VISUALIZATION-ASSISTANT

通过 OpenAI、LangChain 和 Streamlit 的有机结合，我们成功构建了一个智能数据分析与可视化系统。该系统在降低数据分析门槛、提高数据处理效率和可视化效果方面展现出显著优势，为企业和研究人员提供了一种便捷、高效的数据探索和决策支持工具。