你有沒有和一位從不睡覺、無需休息的資料分析師共事過？或者一位處理數字的速度比你說“資料透視表”還快的資料分析師？如果沒有，那就坐穩了，因為我們即將打造這樣的人！今天，我們將建立一個資料分析師人工智慧代理，用於閃電般的資料分析。使用 OpenAI 的函式呼叫，這個人工智慧自動化系統可以用簡單的英語解釋你提出的問題，並在幾秒鐘內給出所需的輸出。
如果一切都按照我們設想的方式設定好，你可以向代理提問，例如“上個季度我們某個部門排名最高的產品是什麼？”或“向我展示營銷支出和銷售額之間的相關性”。作為回報，你將得到即時準確的答案和精美的圖表。這就是 OpenAI 函式呼叫與 OpenAI 資料分析功能相結合所能為你做到的。
為什麼如此激動人心？
過去資料思維存在的問題是必須瞭解 SQL。需要高階思維來理解所分析資料的複雜性。否則，光是瀏覽各種儀表盤介面就得花上幾個小時。現在，函式呼叫使我們能夠建立 AI 代理，使其成為人類語言和資料指令之間的翻譯媒介。想象一下，一位能夠流利地使用“人類”和“資料庫”進行交流的翻譯員！
當 OpenAI 語言模型根據您的自然語言查詢選擇需要呼叫的函式時，奇蹟就發生了。如果您詢問趨勢，它會呼叫時間序列分析函式。如果您請求比較，它會呼叫統計比較函式。AI 就像您的助手，它知道針對任何問題的正確工具。
架構：所有元件如何協同工作
我們的資料分析師 AI 是由多個主要元件組成的，它們彼此同步工作。以下是所有協同工作的元件：
- 大腦（OpenAI 的 GPT 模型）：處理自然語言查詢並決定呼叫哪些函式。您可以將其視為一位經驗豐富的資料分析師，瞭解業務問題和技術實現難題。
- 工具箱（函式庫）：我們將為從統計到圖形的每個不同分析建立一個獨立的函式。每個函式都旨在高效地執行給定的資料操作。
- 資料層：負責載入、清理和準備所有資料集。我們將處理多種型別的資料，並確保我們的代理可以處理所有可用的雜亂資料。
- 通訊介面：這將確保使用者、AI 模型和功能任務之間的來回互動有效併產生有意義的結果。
這種架構的美妙之處在於它的簡潔性。只需編寫幾個新函式並將它們註冊到 AI 中即可。需要新的資料來源？只需插入新的資料聯結器即可。無需人工資料分析師，即可實現無限的擴充套件！
設定您的開發環境
首先，我們需要為我們尋求的 AI 驅動的資料科學搭建一個工作區。操作方法如下。
- 必要的依賴項：您需要 OpenAI 的 Python 包用於 API 呼叫。您還需要 pandas 用於資料處理（因為 pandas 就像資料科學中的瑞士軍刀），matplotlib 和 seaborn 用於繪圖，以及 numpy 用於數字運算。
- API 配置：從 OpenAI 獲取您的 API 金鑰。此外，我們將新增一些帶有速率限制的錯誤處理功能，以確保工作順利進行。
- 資料準備工具：安裝用於 CSV、JSON、Excel 檔案的庫，如果您有雄心壯志，甚至可以安裝資料庫連線！
核心功能：AI分析師的核心
我們希望開發一套基本功能，賦予我們的 AI 代理強大的分析能力：
- 載入和檢查：從各種格式/來源載入資料，並初步瞭解其結構、資料型別和基本統計資料。這些功能是 AI 熟悉資料的階段。
- 統計分析：這些功能提供資料的數學解釋，涵蓋從基本的描述性統計到更復雜的相關性分析。它們旨在以適合 AI 解釋和使用者元件描述的格式呈現結果。
- 視覺化：這些功能將在 AI 確定分析結果時生成圖表、圖形和曲線圖。它們必須足夠靈活，能夠處理各種資料型別，並生成人類可讀的輸出。
過濾和資料轉換：透過這些功能，AI 可以根據使用者查詢對資料進行剪下、切塊和重塑。
函式呼叫的魔力
事情變得非常有趣。當你問“我們月銷售額的趨勢是什麼？”這樣的問題時，AI 不會給出一個泛泛而談的答案。相反，它會執行以下操作：
- 首先，它會分析問題，準確理解你的需求。它會識別“趨勢”和“每月”等詞語。然後，它會將這些詞語與一些合適的分析方法聯絡起來。
- 基於這種理解，它會決定呼叫哪些函式以及呼叫順序。它可能會決定首先呼叫載入資料函式，然後應用基於時間的過濾、趨勢分析，最後建立視覺化效果。
- AI 會按順序執行這些函式，並在執行過程中穿插一些資料傳遞。每個函式都會提供結構化的輸出，供 AI 進行處理和構建。
- 總而言之，AI 會將多個分析階段的所有輸出整合成一個連貫的解釋。然後，它會將這些資訊返回給終端使用者，並提供洞察、視覺化效果和行動建議。
實踐專案：構建你的資料分析師AI代理
讓我們更進一步，構建一個完整的資料分析師 AI 代理，它能夠處理真實的業務資料並提供切實可行的洞察。為此，我們將設計一個 AI 代理來分析電商銷售資料。該代理將能夠回答有關產品效能、客戶行為、季節性趨勢以及提升收入的領域等問題。
1. 安裝所需軟體包
!pip install openai pandas matplotlib seaborn numpy plotly
2. 匯入庫並設定
import openai
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
import json
import warnings
warnings.filterwarnings('ignore')
# Set your OpenAI API key here
openai.api_key = "your-openai-api-key-here" # Replace with your actual API key
print(" All libraries imported successfully!")
3. 生成示例電子商務資料
def generate_sample_data():
"""Generate realistic e-commerce sales data for demonstration"""
np.random.seed(42)
# Product categories and names
categories = ['Electronics', 'Clothing', 'Books', 'Home & Garden', 'Sports']
products = {
'Electronics': ['Smartphone', 'Laptop', 'Headphones', 'Tablet', 'Smart Watch'],
'Clothing': ['T-Shirt', 'Jeans', 'Sneakers', 'Jacket', 'Dress'],
'Books': ['Fiction Novel', 'Science Book', 'Cookbook', 'Biography', 'Self-Help'],
'Home & Garden': ['Coffee Maker', 'Plant Pot', 'Lamp', 'Pillow', 'Rug'],
'Sports': ['Running Shoes', 'Yoga Mat', 'Dumbbell', 'Basketball', 'Tennis Racket']
}
# Generate data for the last 12 months
start_date = datetime.now() - timedelta(days=365)
dates = pd.date_range(start=start_date, end=datetime.now(), freq='D')
data = []
customer_id = 1000
for date in dates:
# Simulate seasonal patterns
month = date.month
seasonal_multiplier = 1.2 if month in [11, 12] else (1.1 if month in [6, 7] else 1.0)
# Generate 10-50 orders per day
daily_orders = np.random.poisson(25 * seasonal_multiplier)
for _ in range(daily_orders):
category = np.random.choice(categories, p=[0.3, 0.25, 0.15, 0.15, 0.15])
product = np.random.choice(products[category])
# Price based on category
price_ranges = {
'Electronics': (50, 1000),
'Clothing': (15, 200),
'Books': (10, 50),
'Home & Garden': (20, 300),
'Sports': (25, 250)
}
price = np.random.uniform(*price_ranges[category])
quantity = np.random.choice([1, 2, 3], p=[0.7, 0.2, 0.1])
data.append({
'date': date,
'customer_id': customer_id,
'product_name': product,
'category': category,
'quantity': quantity,
'unit_price': round(price, 2),
'total_amount': round(price * quantity, 2)
})
customer_id += 1
return pd.DataFrame(data)
# Generate and display sample data
df = generate_sample_data()
print(f" Generated {len(df)} sales records")
print("\n Sample Data Preview:")
print(df.head())
print(f"\n Date Range: {df['date'].min()} to {df['date'].max()}")
print(f" Total Revenue: ${df['total_amount'].sum():,.2f}")
4. 定義分析函式
class DataAnalyzer:
def __init__(self, dataframe):
self.df = dataframe.copy()
self.df['date'] = pd.to_datetime(self.df['date'])
def get_revenue_summary(self, period='monthly'):
"""Calculate revenue summary by time period"""
try:
if period == 'daily':
grouped = self.df.groupby(self.df['date'].dt.date)
elif period == 'weekly':
grouped = self.df.groupby(self.df['date'].dt.isocalendar().week)
elif period == 'monthly':
grouped = self.df.groupby(self.df['date'].dt.to_period('M'))
else:
return {"error": "Invalid period. Use 'daily', 'weekly', or 'monthly'"}
revenue_data = grouped['total_amount'].sum().reset_index()
revenue_data.columns = ['period', 'revenue']
return {
"success": True,
"data": revenue_data.to_dict('records'),
"total_revenue": float(self.df['total_amount'].sum()),
"average_revenue": float(revenue_data['revenue'].mean()),
"period": period
}
except Exception as e:
return {"error": str(e)}
def get_top_products(self, limit=10, metric='revenue'):
"""Get top performing products"""
try:
if metric == 'revenue':
top_products = self.df.groupby('product_name')['total_amount'].sum().sort_values(ascending=False).head(limit)
elif metric == 'quantity':
top_products = self.df.groupby('product_name')['quantity'].sum().sort_values(ascending=False).head(limit)
else:
return {"error": "Invalid metric. Use 'revenue' or 'quantity'"}
return {
"success": True,
"data": [{"product": prod, "value": float(val)} for prod, val in top_products.items()],
"metric": metric,
"limit": limit
}
except Exception as e:
return {"error": str(e)}
def get_category_performance(self):
"""Analyze performance by product category"""
try:
category_stats = self.df.groupby('category').agg({
'total_amount': ['sum', 'mean'],
'quantity': 'sum',
'customer_id': 'nunique'
}).round(2)
category_stats.columns = ['total_revenue', 'avg_order_value', 'total_quantity', 'unique_customers']
category_stats = category_stats.reset_index()
return {
"success": True,
"data": category_stats.to_dict('records')
}
except Exception as e:
return {"error": str(e)}
def get_customer_insights(self):
"""Analyze customer behavior patterns"""
try:
customer_stats = self.df.groupby('customer_id').agg({
'total_amount': 'sum',
'date': ['min', 'max', 'nunique']
}).round(2)
customer_stats.columns = ['total_spent', 'first_purchase', 'last_purchase', 'purchase_frequency']
insights = {
"total_customers": len(customer_stats),
"avg_customer_value": float(customer_stats['total_spent'].mean()),
"avg_purchase_frequency": float(customer_stats['purchase_frequency'].mean()),
"top_spenders": customer_stats.nlargest(5, 'total_spent')['total_spent'].to_dict()
}
return {"success": True, "data": insights}
except Exception as e:
return {"error": str(e)}
def create_visualization(self, chart_type, data_params):
"""Create various types of visualizations"""
try:
plt.figure(figsize=(12, 6))
if chart_type == 'revenue_trend':
# Monthly revenue trend
monthly_data = self.df.groupby(self.df['date'].dt.to_period('M'))['total_amount'].sum()
plt.plot(range(len(monthly_data)), monthly_data.values, marker='o', linewidth=2)
plt.title('Monthly Revenue Trend', fontsize=16, fontweight='bold')
plt.xlabel('Month')
plt.ylabel('Revenue ($)')
plt.xticks(range(len(monthly_data)), [str(x) for x in monthly_data.index], rotation=45)
plt.grid(True, alpha=0.3)
elif chart_type == 'category_pie':
# Category revenue distribution
category_revenue = self.df.groupby('category')['total_amount'].sum()
plt.pie(category_revenue.values, labels=category_revenue.index, autopct='%1.1f%%', startangle=90)
plt.title('Revenue Distribution by Category', fontsize=16, fontweight='bold')
elif chart_type == 'top_products_bar':
# Top products bar chart
top_products = self.df.groupby('product_name')['total_amount'].sum().sort_values(ascending=False).head(10)
plt.barh(range(len(top_products)), top_products.values)
plt.yticks(range(len(top_products)), top_products.index)
plt.title('Top 10 Products by Revenue', fontsize=16, fontweight='bold')
plt.xlabel('Revenue ($)')
plt.tight_layout()
plt.show()
return {"success": True, "message": f"Created {chart_type} visualization"}
except Exception as e:
return {"error": str(e)}
# Initialize analyzer
analyzer = DataAnalyzer(df)
print(" Data Analyzer initialized successfully!")
5. OpenAI的函式定義
def get_revenue_summary(period='monthly'):
"""Get revenue summary by time period (daily, weekly, monthly)"""
return analyzer.get_revenue_summary(period)
def get_top_products(limit=10, metric='revenue'):
"""Get top performing products by revenue or quantity"""
return analyzer.get_top_products(limit, metric)
def get_category_performance():
"""Analyze performance metrics by product category"""
return analyzer.get_category_performance()
def get_customer_insights():
"""Get insights about customer behavior and patterns"""
return analyzer.get_customer_insights()
def create_visualization(chart_type, data_params=None):
"""Create visualizations (revenue_trend, category_pie, top_products_bar)"""
return analyzer.create_visualization(chart_type, data_params or {})
def get_basic_stats():
"""Get basic statistics about the dataset"""
return {
"success": True,
"data": {
"total_records": len(analyzer.df),
"date_range": {
"start": str(analyzer.df['date'].min().date()),
"end": str(analyzer.df['date'].max().date())
},
"total_revenue": float(analyzer.df['total_amount'].sum()),
"unique_products": analyzer.df['product_name'].nunique(),
"unique_customers": analyzer.df['customer_id'].nunique(),
"categories": analyzer.df['category'].unique().tolist()
}
}
6. OpenAI函式模式
functions = [
{
"name": "get_revenue_summary",
"description": "Get revenue summary grouped by time period",
"parameters": {
"type": "object",
"properties": {
"period": {
"type": "string",
"enum": ["daily", "weekly", "monthly"],
"description": "Time period for grouping revenue data"
}
},
"required": ["period"]
}
},
{
"name": "get_top_products",
"description": "Get top performing products by revenue or quantity",
"parameters": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Number of top products to return (default: 10)"
},
"metric": {
"type": "string",
"enum": ["revenue", "quantity"],
"description": "Metric to rank products by"
}
},
"required": ["metric"]
}
},
{
"name": "get_category_performance",
"description": "Analyze performance metrics by product category including revenue, quantity, and customers",
"parameters": {
"type": "object",
"properties": {}
}
},
{
"name": "get_customer_insights",
"description": "Get insights about customer behavior, spending patterns, and purchase frequency",
"parameters": {
"type": "object",
"properties": {}
}
},
{
"name": "create_visualization",
"description": "Create data visualizations like charts and graphs",
"parameters": {
"type": "object",
"properties": {
"chart_type": {
"type": "string",
"enum": ["revenue_trend", "category_pie", "top_products_bar"],
"description": "Type of chart to create"
},
"data_params": {
"type": "object",
"description": "Additional parameters for the chart"
}
},
"required": ["chart_type"]
}
},
{
"name": "get_basic_stats",
"description": "Get basic statistics and overview of the dataset",
"parameters": {
"type": "object",
"properties": {}
}
}
]
print(" Function schemas defined successfully!")
7. 主要的AI代理類
class DataAnalystAI:
def __init__(self, api_key):
self.client = openai.OpenAI(api_key=api_key)
self.functions = {
"get_revenue_summary": get_revenue_summary,
"get_top_products": get_top_products,
"get_category_performance": get_category_performance,
"get_customer_insights": get_customer_insights,
"create_visualization": create_visualization,
"get_basic_stats": get_basic_stats
}
self.conversation_history = []
def process_query(self, user_query):
"""Process user query and return AI response with function calls"""
try:
# Add user message to conversation
messages = [
{
"role": "system",
"content": """You are a helpful data analyst AI assistant. You can analyze e-commerce sales data and create visualizations.
Always provide clear, actionable insights. When showing numerical data, format it nicely with commas for large numbers.
If you create visualizations, mention that the chart has been displayed.
Be conversational and explain your findings in business terms."""
},
{"role": "user", "content": user_query}
]
# Add conversation history
messages = messages[:-1] + self.conversation_history + messages[-1:]
# Call OpenAI API with function calling
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
functions=functions,
function_call="auto",
temperature=0.7
)
message = response.choices[0].message
# Handle function calls
if message.function_call:
function_name = message.function_call.name
function_args = json.loads(message.function_call.arguments)
print(f" Calling function: {function_name} with args: {function_args}")
# Execute the function
function_result = self.functions[function_name](**function_args)
# Get AI's interpretation of the results
messages.append({
"role": "assistant",
"content": None,
"function_call": {
"name": function_name,
"arguments": message.function_call.arguments
}
})
messages.append({
"role": "function",
"name": function_name,
"content": json.dumps(function_result)
})
# Get final response from AI
final_response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=0.7
)
ai_response = final_response.choices[0].message.content
# Update conversation history
self.conversation_history.append({"role": "user", "content": user_query})
self.conversation_history.append({"role": "assistant", "content": ai_response})
return ai_response
else:
# No function call needed
ai_response = message.content
self.conversation_history.append({"role": "user", "content": user_query})
self.conversation_history.append({"role": "assistant", "content": ai_response})
return ai_response
except Exception as e:
return f" Error processing query: {str(e)}"
# Initialize the AI agent
ai_agent = DataAnalystAI("your-openai-api-key-here") # Replace with your API key
print(" AI Data Analyst Agent initialized successfully!")
8.互動式查詢介面
def ask_ai(query):
"""Simple interface to ask questions to the AI agent"""
print(f" Question: {query}")
print(" AI Response:")
response = ai_agent.process_query(query)
print(response)
print("\n" + "="*80 + "\n")
return response
# Cell 9: Example Queries - Run these to test your agent!
print(" Let's test our AI Data Analyst Agent with some example queries:\n")
# Test basic stats
ask_ai("Give me an overview of our sales data")
# Test revenue analysis
ask_ai("Show me the monthly revenue trend")
# Test product analysis
ask_ai("What are our top 5 products by revenue?")
# Test category performance
ask_ai("How are different product categories performing?")
# Test customer insights
ask_ai("Tell me about our customer behavior patterns")
# Test visualization
ask_ai("Create a pie chart showing revenue distribution by category")
# Test comparative analysis
ask_ai("Which product category generates the highest average order value?")
print(" All tests completed! Your AI Data Analyst Agent is ready to use!")
輸出
高階技術與最佳化
在基礎代理到位後，隨著時間的推移，我們將進行多項增強：
- 函式鏈：這些多步驟分析步驟被連結在一起，並由人工智慧輔助執行。許多多步驟分析工作流程原本需要人工協調。
- 上下文感知：為代理實現一些上下文管理，以便其跟蹤已完成的分析並在此基礎上進行構建。這使得對話類似於接聽電話。
- 效能最佳化：快取高開銷的計算，並行執行任何可以獨立完成的分析。這通常可以使函式實現更快、更少的記憶體佔用。
- 錯誤處理：整合全面的錯誤捕獲功能，以便優雅地處理問題。在出現資料問題、API 故障或使用者意外輸入時尤其有用。也有助於為使用者提供合理的反饋。
實際應用和用例
您的資料分析師 AI 代理擁有無限可能：
- 商業智慧：提供定期報告，讓普通人能夠自助分析，併為決策者提供即時洞察。
- 營銷分析：使用自然語言查詢，檢視廣告系列效果指標、客戶細分和投資回報率 (ROI) 計算。
- 財務分析：監控關鍵績效指標 (KPI) 和差異，並使用通俗易懂的語言提問提交財務報告。
- 運營最佳化：監控效能資料和瓶頸，並基於資料驅動的洞察最佳化流程。
小結
構建資料分析師 AI 代理不僅僅是一項技術練習——它關乎資料分析的普及，併為所有人提供洞察。您構建的工具或許有助於改變人與資料之間的互動方式，消除障礙，從而基於資料做出決策。您所學到的技術為許多其他 AI 應用奠定了基礎。
函式呼叫是一個用途廣泛的概念，可用於從客戶服務自動化到複雜的工作流程編排等各種應用。請記住，最好的人工智慧並非取代人類智慧，而是對其進行補充。你的資料分析師人工智慧應該鼓勵使用者針對資料提出更深入的問題，激勵他們更深入地挖掘和分析資料，並幫助他們做出更明智的決策。因此，關鍵不在於掌握所有答案，而在於掌握一些答案，以便找到所有其他答案。
評論留言