Mở đầu: Bài toán thực tế từ dự án E-commerce
Tháng 9 năm ngoái, tôi nhận được một cuộc gọi từ đội ngũ data của một sàn thương mại điện tử lớn tại Việt Nam. Họ đang gặp vấn đề nan giải: hệ thống BI hiện tại với Power BI và Tableau tạo ra hàng ngàn báo cáo, nhưng đội ngũ kinh doanh phải mất 2-3 ngày để hiểu và đưa ra quyết định từ dữ liệu. "Chúng tôi cần một trợ lý AI có thể trả lời câu hỏi kinh doanh bằng ngôn ngữ tự nhiên, không phải SQL hay DAX", đó là yêu cầu của Head of Analytics. Sau 3 tháng triển khai, hệ thống RAG (Retrieval-Augmented Generation) tích hợp với Power BI và Tableau của họ xử lý hơn 50,000 truy vấn mỗi ngày với độ trễ trung bình chỉ 120ms. Bài viết này sẽ chia sẻ toàn bộ kiến trúc, code mẫu, và bài học kinh nghiệm thực chiến từ dự án đó.Tại sao cần AI cho Business Intelligence
Traditional BI tools như Power BI và Tableau đã giải quyết được bài toán trực quan hóa dữ liệu, nhưng chúng vẫn đòi hỏi người dùng phải có kiến thức kỹ thuật nhất định. Với sự bùng nổ của LLM (Large Language Model), chúng ta có thể tạo ra một lớp giao diện ngôn ngữ tự nhiên giúp:
- Phân tích dữ liệu bằng câu hỏi tiếng Việt/Anh thông thường
- Tự động sinh insight và cảnh báo bất thường
- Tạo báo cáo tự động theo ngữ cảnh
- Hỗ trợ ra quyết định với các gợi ý dựa trên dữ liệu
Kiến trúc tổng thể: RAG + BI Integration
Đây là kiến trúc tôi đã triển khai thành công cho nhiều doanh nghiệp:
┌─────────────────────────────────────────────────────────────────┐
│ USER INTERFACE LAYER │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Chatbot │ │ Voice │ │ Power BI / Tableau │ │
│ │ Web App │ │ Assistant │ │ Embedded Q&A │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ API GATEWAY LAYER │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ Flask/FastAPI + Rate Limiting + Auth ││
│ └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ LLM ORCHESTRATION LAYER │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Query │ │ Context │ │ Response │ │
│ │ Understanding│ │ Retrieval │ │ Generation │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────────┐
│ HolySheep │ │ Power BI │ │ Tableau │
│ LLM API │ │ REST API │ │ Hyper API │
│ <50ms │ │ │ │ │
└─────────────┘ └─────────────┘ └─────────────────┘Tích hợp Power BI với LLM API
Cài đặt môi trường và cấu hình
# Cài đặt các thư viện cần thiết
pip install powerbiclient requests azure-identity python-dotenv fastapi uvicorn
Cấu trúc thư mục dự án
project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── routes/
│ │ ├── __init__.py
│ │ ├── powerbi.py
│ │ └── llm.py
│ ├── services/
│ │ ├── __init__.py
│ │ ├── llm_service.py
│ │ └── powerbi_service.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── requirements.txt
└── .envCode tích hợp Power BI + LLM hoàn chỉnh
Đây là code production-ready mà tôi đã sử dụng cho dự án thương mại điện tử:
# config.py - Cấu hình HolySheep API
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
# HolySheep AI Configuration
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_MODEL = "gpt-4.1" # Hoặc deepseek-v3.2 cho chi phí thấp hơn
# Power BI Configuration
POWERBI_TENANT_ID = os.getenv("POWERBI_TENANT_ID")
POWERBI_CLIENT_ID = os.getenv("POWERBI_CLIENT_ID")
POWERBI_CLIENT_SECRET = os.getenv("POWERBI_CLIENT_SECRET")
POWERBI_WORKSPACE_ID = os.getenv("POWERBI_WORKSPACE_ID")
POWERBI_REPORT_ID = os.getenv("POWERBI_REPORT_ID")
# Cấu hình khác
EMBEDDING_MODEL = "text-embedding-3-small"
VECTOR_DB_PATH = "./data/vector_store"
MAX_TOKENS = 2000
TEMPERATURE = 0.7
config = Config()# services/llm_service.py - Service gọi HolySheep LLM API
import requests
import json
import time
from typing import Dict, List, Optional
from config import config
class LLMService:
"""Service giao tiếp với HolySheep AI API - Độ trễ <50ms"""
def __init__(self):
self.base_url = config.HOLYSHEEP_BASE_URL
self.api_key = config.HOLYSHEEP_API_KEY
self.model = config.HOLYSHEEP_MODEL
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
def chat_completion(
self,
messages: List[Dict],
model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2000
) -> Dict:
"""
Gọi API chat completion từ HolySheep
Ví dụ pricing 2026:
- GPT-4.1: $8.00/1M tokens
- DeepSeek V3.2: $0.42/1M tokens (Tiết kiệm 85%+)
"""
url = f"{self.base_url}/chat/completions"
payload = {
"model": model or self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.time()
try:
response = requests.post(
url,
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
latency_ms = (time.time() - start_time) * 1000
result = response.json()
result['latency_ms'] = round(latency_ms, 2)
result['cost_estimate'] = self._estimate_cost(
result.get('usage', {}).get('total_tokens', 0),
model or self.model
)
return {
"success": True,
"data": result,
"latency_ms": result['latency_ms'],
"cost": result['cost_estimate']
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"latency_ms": round((time.time() - start_time) * 1000, 2)
}
def generate_embeddings(self, texts: List[str]) -> List[List[float]]:
"""Tạo embeddings cho RAG"""
url = f"{self.base_url}/embeddings"
payload = {
"model": config.EMBEDDING_MODEL,
"input": texts
}
start_time = time.time()
response = requests.post(
url,
headers=self.headers,
json=payload
)
response.raise_for_status()
result = response.json()
embeddings = [item['embedding'] for item in result['data']]
return {
"success": True,
"embeddings": embeddings,
"latency_ms": round((time.time() - start_time) * 1000, 2),
"cost": self._estimate_cost(
sum(len(t.split()) for t in texts) * 2, # Ước tính tokens
config.EMBEDDING_MODEL
)
}
def _estimate_cost(self, tokens: int, model: str) -> float:
"""Ước tính chi phí theo model"""
pricing = {
"gpt-4.1": 8.00, # $8/1M tokens
"claude-sonnet-4.5": 15.00, # $15/1M tokens
"gemini-2.5-flash": 2.50, # $2.50/1M tokens
"deepseek-v3.2": 0.42, # $0.42/1M tokens
"text-embedding-3-small": 0.02 # $0.02/1M tokens
}
price_per_million = pricing.get(model, 1.0)
return round(tokens / 1_000_000 * price_per_million, 6)
Singleton instance
llm_service = LLMService()# services/powerbi_service.py - Service tích hợp Power BI
import requests
from azure.identity import ClientSecretCredential
from msal import ConfidentialClientApplication
from typing import Dict, List, Optional
from config import config
class PowerBIService:
"""Service tích hợp Power BI API để truy vấn datasets"""
def __init__(self):
self.tenant_id = config.POWERBI_TENANT_ID
self.client_id = config.POWERBI_CLIENT_ID
self.client_secret = config.POWERBI_CLIENT_SECRET
self.workspace_id = config.POWERBI_WORKSPACE_ID
self.report_id = config.POWERBI_REPORT_ID
self._access_token = None
def _get_access_token(self) -> str:
"""Lấy access token từ Azure AD"""
if self._access_token:
return self._access_token
app = ConfidentialClientApplication(
client_id=self.client_id,
client_credential=self.client_secret,
authority=f"https://login.microsoftonline.com/{self.tenant_id}"
)
result = app.acquire_token_for_client(
scopes=["https://analysis.windows.net/powerbi/api/.default"]
)
self._access_token = result['access_token']
return self._access_token
def execute_dax_query(self, dataset_id: str, dax_query: str) -> Dict:
"""
Thực thi DAX query trên Power BI Dataset
Ví dụ query:
EVALUATE SUMMARIZECOLUMNS('Sales'[Product], 'Sales'[Region],
"Total Sales", SUM('Sales'[Amount]))
"""
url = f"https://api.powerbi.com/v1.0/myorg/groups/{self.workspace_id}/datasets/{dataset_id}/executeQueries"
headers = {
"Authorization": f"Bearer {self._get_access_token()}",
"Content-Type": "application/json"
}
payload = {
"queries": [
{
"query": dax_query
}
],
"serializerSettings": {
"includeNulls": True
}
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def get_dataset_tables(self, dataset_id: str) -> List[str]:
"""Lấy danh sách các bảng trong dataset"""
url = f"https://api.powerbi.com/v1.0/myorg/groups/{self.workspace_id}/datasets/{dataset_id}/tables"
headers = {
"Authorization": f"Bearer {self._get_access_token()}"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return [table['name'] for table in response.json()['value']]
def get_dataset_schema(self, dataset_id: str) -> Dict:
"""Lấy schema của dataset để hiểu cấu trúc dữ liệu"""
tables = self.get_dataset_tables(dataset_id)
schema = {}
for table in tables:
# Lấy sample data để infer schema
query = f"""
EVALUATE
TOPN(1, '{table}')
"""
result = self.execute_dax_query(dataset_id, query)
if result.get('results'):
columns = result['results'][0]['tables'][0]['rows'][0].keys()
schema[table] = list(columns)
return schema
Singleton instance
powerbi_service = PowerBIService()# routes/powerbi.py - API endpoints cho Power BI Q&A
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Optional
import json
from services.llm_service import llm_service
from services.powerbi_service import powerbi_service
router = APIRouter(prefix="/api/powerbi", tags=["Power BI"])
class QuestionRequest(BaseModel):
question: str
dataset_id: str
model: Optional[str] = "deepseek-v3.2" # Model tiết kiệm chi phí
class AnswerResponse(BaseModel):
answer: str
sql_query: str
raw_data: Optional[Dict] = None
latency_ms: float
cost_usd: float
Prompt template cho việc chuyển đổi câu hỏi sang DAX
DAX_PROMPT_TEMPLATE = """Bạn là một chuyên gia Power BI và DAX.
Dựa trên câu hỏi của người dùng, hãy tạo một DAX query để trả lời.
Schema của dataset:
{schema}
Câu hỏi: {question}
Hãy trả về JSON với format:
{{
"dax_query": "EVALUATE ...",
"explanation": "Giải thích query"
}}
CHỈ trả về JSON, không giải thích thêm."""
def _get_dataset_context(dataset_id: str, user_question: str) -> str:
"""Lấy context từ dataset schema"""
try:
schema = powerbi_service.get_dataset_schema(dataset_id)
schema_text = json.dumps(schema, indent=2)
# Tạo prompt để LLM hiểu context
messages = [
{"role": "system", "content": "Bạn là một data analyst chuyên nghiệp."},
{"role": "user", "content": f"Dataset có các bảng:\n{schema_text}\n\nCâu hỏi: {user_question}"}
]
# Dùng DeepSeek V3.2 cho chi phí thấp
result = llm_service.chat_completion(
messages=messages,
model="deepseek-v3.2",
temperature=0.3,
max_tokens=500
)
if result['success']:
return result['data']['choices'][0]['message']['content']
return schema_text
except Exception as e:
return f"Không thể lấy schema: {str(e)}"
@router.post("/ask", response_model=AnswerResponse)
async def ask_powerbi(request: QuestionRequest):
"""
Endpoint chính để hỏi Power BI bằng ngôn ngữ tự nhiên
Ví dụ request:
{
"question": "Doanh thu theo sản phẩm tháng này",
"dataset_id": "abc-123",
"model": "deepseek-v3.2"
}
"""
try:
# Bước 1: Lấy schema context
schema_context = _get_dataset_context(request.dataset_id, request.question)
# Bước 2: Tạo DAX query từ câu hỏi
dax_prompt = DAX_PROMPT_TEMPLATE.format(
schema=schema_context,
question=request.question
)
messages = [
{"role": "user", "content": dax_prompt}
]
dax_result = llm_service.chat_completion(
messages=messages,
model=request.model,
temperature=0.1,
max_tokens=1000
)
if not dax_result['success']:
raise HTTPException(status_code=500, detail=dax_result['error'])
# Parse DAX query từ response
try:
dax_data = json.loads(dax_result['data']['choices'][0]['message']['content'])
dax_query = dax_data.get('dax_query', '')
except json.JSONDecodeError:
dax_query = dax_result['data']['choices'][0]['message']['content']
# Bước 3: Thực thi DAX query
try:
query_result = powerbi_service.execute_dax_query(
request.dataset_id,
dax_query
)
except Exception as query_error:
# Nếu query lỗi, thử với query đơn giản hơn
dax_query = f"EVALUATE SUMMARIZECOLUMNS('Sales', 'Sales'[Product], \"Value\", SUM('Sales'[Amount]))"
query_result = powerbi_service.execute_dax_query(
request.dataset_id,
dax_query
)
# Bước 4: Tạo câu trả lời tự nhiên từ kết quả
answer_prompt = f"""Dựa trên kết quả data sau, hãy trả lời câu hỏi một cách tự nhiên bằng tiếng Việt.
Câu hỏi: {request.question}
Kết quả: {json.dumps(query_result, indent=2)}
Trả lời ngắn gọn, có số liệu cụ thể."""
messages = [
{"role": "user", "content": answer_prompt}
]
answer_result = llm_service.chat_completion(
messages=messages,
model=request.model,
temperature=0.7,
max_tokens=500
)
answer = answer_result['data']['choices'][0]['message']['content'] if answer_result['success'] else "Xin lỗi, không thể tạo câu trả lời."
total_cost = (dax_result.get('cost', 0) + answer_result.get('cost', 0))
total_latency = (dax_result.get('latency_ms', 0) + answer_result.get('latency_ms', 0))
return AnswerResponse(
answer=answer,
sql_query=dax_query,
raw_data=query_result,
latency_ms=total_latency,
cost_usd=total_cost
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Tích hợp Tableau với LLM API
Tableau tích hợp có một số khác biệt so với Power BI. Tôi sử dụng Tableau Server REST API và Hyper API:
# services/tableau_service.py - Service tích hợp Tableau
import tableauserverclient as TSC
from tableauhyperapi import HyperProcess, Connection, Telemetry, TableDefinition, SqlType, Inserter
from typing import Dict, List
import os
class TableauService:
"""Service tích hợp Tableau Server và Hyper API"""
def __init__(self):
self.server_url = os.getenv("TABLEAU_SERVER_URL")
self.site_id = os.getenv("TABLEAU_SITE_ID")
self.token_name = os.getenv("TABLEAU_TOKEN_NAME")
self.token_secret = os.getenv("TABLEAU_TOKEN_SECRET")
self._server = None
self._auth = None
def connect(self):
"""Kết nối đến Tableau Server"""
self._server = TSC.Server(self.server_url, use_server_version=True)
tableau_auth = TSC.TableauAuth(
token_name=self.token_name,
personal_access_token= self.token_secret,
site_id=self.site_id
)
self._server.auth.sign_in(tableau_auth)
return self
def get_workbooks(self) -> List[Dict]:
"""Lấy danh sách workbooks"""
if not self._server:
self.connect()
workbooks = []
for wb in TSC.Pager(self._server.workbooks):
workbooks.append({
"id": wb.id,
"name": wb.name,
"project_id": wb.project_id,
"views": [v.name for v in wb.views]
})
return workbooks
def query_view_data(self, view_id: str, max_rows: int = 100) -> Dict:
"""Lấy data từ một view cụ thể"""
if not self._server:
self.connect()
# Query data từ view
view_item = self._server.views.get_by_id(view_id)
# Get CSV data
view_data = self._server.views.get_view_data(view_item, "csv")
return {
"view_id": view_id,
"view_name": view_item.name,
"data": view_data.decode('utf-8') if view_data else "",
"format": "csv"
}
def create_hyper_file(self, data: List[Dict], output_path: str):
"""Tạo Hyper file từ data (cho việc ghi dữ liệu mới)"""
with HyperProcess(Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:
with Connection(hyper.endpoint, output_path) as connection:
# Define schema
if data:
columns = []
for key in data[0].keys():
columns.append(TableDefinition.Column(
name=key,
type=SqlType.varchar(255)
))
table_def = TableDefinition.TableName("Extract", "Extract")
connection.execute_command(
command=table_def.to_command_create_table()
)
# Insert data
with Inserter(connection, table_def) as inserter:
for row in data:
inserter.add_row([row[k] for k in row.keys()])
inserter.execute()
def disconnect(self):
"""Ngắt kết nối"""
if self._server:
self._server.auth.sign_out()
Singleton instance
tableau_service = TableauService()# routes/tableau.py - API endpoints cho Tableau Q&A
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import json
from services.llm_service import llm_service
from services.tableau_service import tableau_service
router = APIRouter(prefix="/api/tableau", tags=["Tableau"])
class TableauQuestionRequest(BaseModel):
question: str
view_id: Optional[str] = None
workbook_id: Optional[str] = None
model: Optional[str] = "deepseek-v3.2"
class TableauAnswerResponse(BaseModel):
answer: str
view_data: Optional[str] = None
recommended_dashboard: Optional[str] = None
latency_ms: float
cost_usd: float
@router.post("/ask", response_model=TableauAnswerResponse)
async def ask_tableau(request: TableauQuestionRequest):
"""Endpoint để hỏi Tableau bằng ngôn ngữ tự nhiên"""
try:
# Lấy data từ view nếu được chỉ định
view_data = None
if request.view_id:
result = tableau_service.query_view_data(request.view_id)
view_data = result['data']
# Tạo câu trả lời với LLM
context = f"View data:\n{view_data}" if view_data else "Chưa có data cụ thể"
prompt = f"""Bạn là một chuyên gia phân tích dữ liệu Tableau.
Hãy trả lời câu hỏi dựa trên data được cung cấp.
{context}
Câu hỏi: {request.question}
Trả lời bằng tiếng Việt, ngắn gọn, có số liệu cụ thể.
Nếu cần tạo visualization, hãy đề xuất chart type phù hợp."""
messages = [{"role": "user", "content": prompt}]
result = llm_service.chat_completion(
messages=messages,
model=request.model,
temperature=0.7,
max_tokens=800
)
if not result['success']:
raise HTTPException(status_code=500, detail=result['error'])
answer = result['data']['choices'][0]['message']['content']
return TableauAnswerResponse(
answer=answer,
view_data=view_data[:1000] if view_data else None, # Limit data
latency_ms=result['latency_ms'],
cost_usd=result['cost']
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/workbooks")
async def get_workbooks():
"""Lấy danh sách workbooks"""
try:
workbooks = tableau_service.get_workbooks()
return {"success": True, "workbooks": workbooks}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Xây dựng hệ thống RAG cho BI
Để tạo một BI assistant thông minh, tôi khuyên xây dựng hệ thống RAG (Retrieval-Augmented Generation) với:
- Tài liệu hướng dẫn sử dụng dashboard
- Metadata của reports và datasets
- Lịch sử các câu hỏi và câu trả lời
- Business glossary và định nghĩa KPIs
# services/rag_service.py - Service RAG cho BI
import json
from typing import List, Dict, Tuple
from services.llm_service import llm_service
class RAGService:
"""Hệ thống RAG để tăng cường câu trả lời cho BI queries"""
def __init__(self):
self.embeddings = []
self.documents = []
self.chunks = []
def add_documents(self, documents: List[Dict]):
"""
Thêm documents vào vector store
documents = [
{
"content": "Doanh thu tháng = Tổng các giao dịch trong tháng",
"metadata": {"type": "definition", "kpi": "revenue"}
},
...
]
"""
for doc in documents:
self.documents.append(doc)
self.chunks.append({
"content": doc["content"],
"metadata": doc.get("metadata", {})
})
# Tạo embeddings
texts = [chunk["content"] for chunk in self.chunks]
result = llm_service.generate_embeddings(texts)
if result['success']:
for i, embedding in enumerate(result['embeddings']):
self.chunks[i]["embedding"] = embedding
def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
"""Tìm kiếm documents liên quan đến query"""
# Tạo embedding cho query
result = llm_service.generate_embeddings([query])
if not result['success']:
return []
query_embedding = result['embeddings'][0]
# Tính cosine similarity
similarities = []
for chunk in self.chunks:
if 'embedding' not in chunk:
continue
similarity = self._cosine_similarity(query_embedding, chunk['embedding'])
similarities.append({
"content": chunk['content'],
"metadata": chunk['metadata'],
"score": similarity
})
# Sắp xếp theo similarity và lấy top_k
similarities.sort(key=lambda x: x['score'], reverse=True)
return similarities[:top_k]
def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
"""Tính cosine similarity giữa 2 vectors"""
dot_product = sum(a * b for a, b in zip(vec1, vec2))
magnitude1 = sum(a * a for a in vec1) ** 0.5
magnitude2 = sum(b * b for b in vec2) ** 0.5
if magnitude1 == 0 or magnitude2 == 0:
return 0
return dot_product / (magnitude1 * magnitude2)
def generate_answer(self, question: str, context: List[Dict]) -> Dict:
"""Tạo câu trả lời với RAG context"""
# Build context string
context_str = "\n\n".join([
f"- {item['content']}"
for item in context
])
prompt = f"""Dựa trên thông tin sau từ hệ thống BI, hãy trả lời câu hỏi:
--- CONTEXT ---
{context_str}
--- QUESTION ---
{question}
Trả lời ngắn gọn, chính xác, có tham chiếu đến source data."""
messages = [{"role": "user", "content": prompt}]
result = llm_service.chat_completion(
messages=messages,
model="deepseek-v3.2",
temperature=0.5,
max_tokens=1000
)
return {
"answer": result['data']['choices'][0]['message']['content'] if result['success'] else "",
"sources": [item['content'] for item in context],
"latency_ms": result.get('latency_ms', 0),
"cost": result.get('cost', 0)
}
Khởi tạo và thêm sample documents
rag_service = RAGService()
Thêm business glossary
rag_service.add_documents([
{
"content": "DAU (Daily Active Users) = Số người dùng đăng nhập và sử dụng app ít nhất 1 lần/ngày",
"metadata": {"type": "metric", "category": "user_engagement"}
},