Trong bài viết này, tôi sẽ chia sẻ cách xây dựng một RAG (Retrieval-Augmented Generation) knowledge base hoàn chỉnh sử dụng dữ liệu từ Amberdata — nền tảng cung cấp dữ liệu blockchain real-time — kết hợp với LangChain và HolySheep AI làm LLM backend.
Tại sao chọn Amberdata + LangChain + HolySheep AI?
Amberdata cung cấp API truy cập dữ liệu on-chain (gas prices, token transfers, smart contract events, DeFi metrics) với độ trễ thấp. Kết hợp LangChain framework, chúng ta có thể xây dựng chatbot phân tích blockchain chính xác.
Lý do chọn HolySheep AI:
- Tỷ giá ¥1 = $1 — tiết kiệm 85%+ so với OpenAI
- Hỗ trợ WeChat/Alipay — thuận tiện cho developers châu Á
- Độ trễ trung bình <50ms cho inference
- Đăng ký tại đây để nhận tín dụng miễn phí
Kiến trúc hệ thống
┌─────────────────────────────────────────────────────────────────┐
│ RAG Architecture │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Amberdata │───▶│ Extract │───▶│ Text Splitter │ │
│ │ API │ │ Module │ │ (Recursive) │ │
│ └──────────────┘ └──────────────┘ └────────┬─────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Vector │◀───│ Embedding │◀───│ Document Store │ │
│ │ Store │ │ (HolySheep)│ │ (Chroma/FAISS) │ │
│ └──────┬───────┘ └──────────────┘ └──────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Retrieval │───▶│ LLM │───▶│ Response │ │
│ │ (Similar) │ │ (HolySheep) │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘Cài đặt môi trường
# requirements.txt
langchain==0.3.7
langchain-community==0.3.5
langchain-huggingface==0.1.2
chromadb==0.5.5
openai==1.54.5
amberdata-sdk==2.0.1
pydantic==2.9.2
httpx==0.27.2
Cài đặt
pip install -r requirements.txtModule 1: Amberdata Data Extractor
# amberdata_extractor.py
import httpx
from typing import List, Dict, Any
from datetime import datetime, timedelta
import asyncio
class AmberdataExtractor:
"""Trích xuất dữ liệu từ Amberdata API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://web3api.io/api/v2"
self.headers = {
"x-api-key": api_key,
"Content-Type": "application/json"
}
async def fetch_gas_prices(self, hours: int = 24) -> List[Dict]:
"""Lấy dữ liệu gas prices"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/market/eth gas",
headers=self.headers,
timeout=30.0
)
data = response.json()
return [{
"type": "gas_price",
"blockNumber": record.get("blockNumber"),
"gasPrice": record.get("gasPrice"),
"slow": record.get("slow", {}).get("gasPrice"),
"standard": record.get("standard", {}).get("gasPrice"),
"fast": record.get("fast", {}).get("gasPrice"),
"instant": record.get("instant", {}).get("gasPrice"),
"timestamp": record.get("timestamp"),
"source": "amberdata"
} for record in data.get("payload", {}).get("records", [])]
async def fetch_token_transfers(self, address: str, limit: int = 100) -> List[Dict]:
"""Lấy token transfers cho một địa chỉ"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/transactions/address/{address}/transfer",
headers=self.headers,
params={"limit": limit}
)
data = response.json()
return [{
"type": "token_transfer",
"hash": tx.get("hash"),
"from": tx.get("from"),
"to": tx.get("to"),
"value": tx.get("value"),
"token": tx.get("token", {}).get("symbol"),
"timestamp": tx.get("timestamp"),
"source": "amberdata"
} for tx in data.get("payload", {}).get("transfers", [])]
async def fetch_defi_metrics(self, protocol: str) -> List[Dict]:
"""Lấy DeFi metrics (TVL, volume, etc.)"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/defi/protocol/{protocol}/metrics",
headers=self.headers
)
data = response.json()
return [{
"type": "defi_metrics",
"protocol": protocol,
"tvl": data.get("tvl"),
"volume24h": data.get("volume24h"),
"fees24h": data.get("fees24h"),
"timestamp": datetime.now().isoformat(),
"source": "amberdata"
}]
async def fetch_all_data(self, addresses: List[str], protocol: str) -> List[Dict]:
"""Fetch tất cả dữ liệu song song"""
tasks = []
# Gas prices
tasks.append(self.fetch_gas_prices(hours=48))
# Token transfers cho mỗi address
for addr in addresses:
tasks.append(self.fetch_token_transfers(addr, limit=50))
# DeFi metrics
tasks.append(self.fetch_defi_metrics(protocol))
# Execute song song
results = await asyncio.gather(*tasks, return_exceptions=True)
# Flatten results
all_data = []
for result in results:
if isinstance(result, list):
all_data.extend(result)
elif isinstance(result, dict):
all_data.append(result)
return all_data
Sử dụng
if __name__ == "__main__":
extractor = AmberdataExtractor(api_key="YOUR_AMBERDATA_KEY")
addresses = [
"0xA0b86a33E6441C8C99D4d9b86e8C3B1e6fD5c7a5",
"0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D" # Uniswap V2
]
data = asyncio.run(extractor.fetch_all_data(addresses, "uniswap-v2"))
print(f"Fetched {len(data)} records")Module 2: Document Processing với Chunking Strategy
# document_processor.py
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing import List, Dict, Any
import hashlib
class BlockchainDocumentProcessor:
"""Xử lý và chunk dữ liệu blockchain cho RAG"""
def __init__(
self,
chunk_size: int = 1000,
chunk_overlap: int = 200,
separators: List[str] = None
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.separators = separators or [
"\n\n",
"\n",
" | ",
" -> ",
": ",
". ",
", "
]
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=self.separators
)
def gas_price_to_text(self, records: List[Dict]) -> str:
"""Convert gas prices sang text format"""
sections = ["# ETH Gas Price Analysis\n"]
for record in records:
section = f"""
Block {record.get('blockNumber')}
- Slow Gas: {record.get('slow', 'N/A')} wei
- Standard Gas: {record.get('standard', 'N/A')} wei
- Fast Gas: {record.get('fast', 'N/A')} wei
- Instant Gas: {record.get('instant', 'N/A')} wei
- Timestamp: {record.get('timestamp', 'N/A')}
"""
sections.append(section)
return "\n".join(sections)
def transfers_to_text(self, records: List[Dict]) -> str:
"""Convert transfers sang text format"""
sections = ["# Token Transfer History\n"]
grouped = {}
for tx in records:
token = tx.get('token', 'ETH')
if token not in grouped:
grouped[token] = []
grouped[token].append(tx)
for token, txs in grouped.items():
sections.append(f"\n## {token} Transfers\n")
for tx in txs[:20]: # Giới hạn 20 tx per token
section = f"""
- Hash: {tx.get('hash', 'N/A')}
From: {tx.get('from', 'N/A')}
To: {tx.get('to', 'N/A')}
Value: {tx.get('value', 'N/A')} {token}
Time: {tx.get('timestamp', 'N/A')}
"""
sections.append(section)
return "\n".join(sections)
def defi_metrics_to_text(self, records: List[Dict]) -> str:
"""Convert DeFi metrics sang text format"""
sections = ["# DeFi Protocol Metrics\n"]
for metrics in records:
section = f"""
{metrics.get('protocol', 'Unknown Protocol')}
- Total Value Locked (TVL): ${metrics.get('tvl', 0):,.2f}
- 24h Volume: ${metrics.get('volume24h', 0):,.2f}
- 24h Fees: ${metrics.get('fees24h', 0):,.2f}
- Last Updated: {metrics.get('timestamp', 'N/A')}
"""
sections.append(section)
return "\n".join(sections)
def create_documents(self, data: List[Dict]) -> List[Document]:
"""Tạo LangChain Documents từ raw data"""
documents = []
# Group by type
by_type = {}
for item in data:
doc_type = item.get('type', 'unknown')
if doc_type not in by_type:
by_type[doc_type] = []
by_type[doc_type].append(item)
# Convert each type to document
if 'gas_price' in by_type:
text = self.gas_price_to_text(by_type['gas_price'])
doc = Document(
page_content=text,
metadata={
"source": "amberdata",
"doc_type": "gas_price",
"record_count": len(by_type['gas_price']),
"doc_id": hashlib.md5(f"gas_{by_type['gas_price'][0].get('blockNumber')}".encode()).hexdigest()[:8]
}
)
documents.append(doc)
if 'token_transfer' in by_type:
text = self.transfers_to_text(by_type['token_transfer'])
doc = Document(
page_content=text,
metadata={
"source": "amberdata",
"doc_type": "token_transfer",
"record_count": len(by_type['token_transfer']),
"doc_id": hashlib.md5("transfers".encode()).hexdigest()[:8]
}
)
documents.append(doc)
if 'defi_metrics' in by_type:
text = self.defi_metrics_to_text(by_type['defi_metrics'])
doc = Document(
page_content=text,
metadata={
"source": "amberdata",
"doc_type": "defi_metrics",
"record_count": len(by_type['defi_metrics']),
"doc_id": hashlib.md5("defi".encode()).hexdigest()[:8]
}
)
documents.append(doc)
return documents
def chunk_documents(self, documents: List[Document]) -> List[Document]:
"""Split documents thành chunks nhỏ hơn"""
chunks = self.splitter.split_documents(documents)
# Thêm metadata cho chunks
for i, chunk in enumerate(chunks):
chunk.metadata.update({
"chunk_id": i,
"chunk_size": len(chunk.page_content)
})
return chunks
Sử dụng
if __name__ == "__main__":
processor = BlockchainDocumentProcessor(chunk_size=800, chunk_overlap=150)
# processor.create_documents(data) và processor.chunk_documents()Module 3: Vector Store với HolySheep AI Embeddings
# vector_store.py
import chromadb
from chromadb.config import Settings
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from typing import List, Optional
import time
class HolySheepEmbeddings(HuggingFaceEmbeddings):
"""Custom embeddings sử dụng HolySheep API thay vì HuggingFace"""
def __init__(self, api_key: str, model: str = "text-embedding-3-small"):
super().__init__(model_name=model)
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = model
def embed_query(self, text: str) -> List[float]:
"""Embed một query"""
import requests
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": text,
"model": self.model
},
timeout=30.0
)
if response.status_code != 200:
raise ValueError(f"Embedding failed: {response.text}")
return response.json()["data"][0]["embedding"]
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""Embed nhiều documents"""
import requests
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": texts,
"model": self.model
},
timeout=60.0
)
if response.status_code != 200:
raise ValueError(f"Embedding failed: {response.text}")
return [item["embedding"] for item in response.json()["data"]]
class VectorStoreManager:
"""Quản lý Vector Store với Chroma"""
def __init__(
self,
persist_directory: str = "./chroma_db",
collection_name: str = "blockchain_knowledge"
):
self.persist_directory = persist_directory
self.collection_name = collection_name
# Khởi tạo Chroma client
self.client = chromadb.PersistentClient(
path=persist_directory,
settings=Settings(anonymized_telemetry=False)
)
self.embeddings = None
self.vectorstore = None
def initialize(self, api_key: str):
"""Khởi tạo embeddings với HolySheep"""
self.embeddings = HolySheepEmbeddings(
api_key=api_key,
model="text-embedding-3-small" # 1536 dimensions
)
self.vectorstore = Chroma(
client=self.client,
collection_name=self.collection_name,
embedding_function=self.embeddings
)
print(f"✅ Vector store initialized at {self.persist_directory}")
def add_documents(
self,
documents: List[Document],
batch_size: int = 100
) -> int:
"""Thêm documents vào vector store"""
total_added = 0
start_time = time.time()
# Batch processing để tránh rate limit
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
self.vectorstore.add_documents(batch)
total_added += len(batch)
print(f" Added {len(batch)} documents ({total_added}/{len(documents)})")
# Small delay để tránh overload
if i + batch_size < len(documents):
time.sleep(0.1)
elapsed = time.time() - start_time
print(f"✅ Added {total_added} documents in {elapsed:.2f}s")
return total_added
def similarity_search(
self,
query: str,
k: int = 5,
filter_dict: Optional[dict] = None
) -> List[Document]:
"""Tìm kiếm similar documents"""
return self.vectorstore.similarity_search(
query=query,
k=k,
filter=filter_dict
)
def similarity_search_with_score(
self,
query: str,
k: int = 5,
threshold: float = 0.7
) -> List[tuple]:
"""Tìm kiếm với similarity score"""
results = self.vectorstore.similarity_search_with_score(
query=query,
k=k
)
# Filter by threshold
filtered = [(doc, score) for doc, score in results if score >= threshold]
return filtered
def get_stats(self) -> dict:
"""Lấy statistics của collection"""
collection = self.client.get_collection(self.collection_name)
return {
"name": collection.name,
"count": collection.count(),
"embedding_dimension": self.embeddings.model == "text-embedding-3-small" and 1536 or 0
}
Sử dụng
if __name__ == "__main__":
manager = VectorStoreManager(
persist_directory="./data/chroma_db",
collection_name="amberdata_knowledge"
)
manager.initialize(api_key="YOUR_HOLYSHEEP_API_KEY")
# Stats
stats = manager.get_stats()
print(f"Collection stats: {stats}")
# Search example
results = manager.similarity_search(
query="What is the current gas price for fast transactions?",
k=5
)
for doc in results:
print(f"\n--- Document ---")
print(doc.page_content[:500])Module 4: RAG Chain với HolySheep AI LLM
# rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from typing import List, Optional
import time
class BlockchainRAGChain:
"""RAG Chain cho phân tích blockchain data"""
def __init__(
self,
vectorstore,
api_key: str,
model: str = "gpt-4.1",
temperature: float = 0.3,
max_tokens: int = 2000
):
self.vectorstore = vectorstore
# Khởi tạo LLM với HolySheep AI
# ⚠️ QUAN TRỌNG: Sử dụng HolySheep thay vì OpenAI
self.llm = ChatOpenAI(
base_url="https://api.holysheep.ai/v1",
api_key=api_key,
model=model,
temperature=temperature,
max_tokens=max_tokens,
streaming=True
)
# Prompt templates
self._setup_prompts()
# Build chains
self._build_chains()
def _setup_prompts(self):
"""Setup prompt templates"""
# Context-aware prompt để handle chat history
self.history_aware_prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
("user", "Given the above conversation, generate a search query to look up "
"information relevant to the current question.")
])
# System prompt cho qa chain
self.qa_prompt = ChatPromptTemplate.from_messages([
("system", """Bạn là chuyên gia phân tích dữ liệu blockchain và DeFi.
Sử dụng thông tin được cung cấp từ Amberdata để trả lời câu hỏi một cách chính xác.
QUY TẮC QUAN TRỌNG:
1. Chỉ sử dụng dữ liệu từ context được cung cấp
2. Nếu không có đủ thông tin, hãy nói rõ "Tôi không có đủ dữ liệu để trả lời câu hỏi này"
3. Trích dẫn nguồn dữ liệu khi có thể (block number, timestamp)
4. Format số liệu rõ ràng (VD: Gas price: 25 Gwei)
5. Cung cấp insights và recommendations nếu được yêu cầu
Context: {context}"""),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}")
])
def _build_chains(self):
"""Build retrieval and qa chains"""
# History-aware retriever chain
self.history_aware_retriever = create_history_aware_retriever(
llm=self.llm,
retriever=self.vectorstore.as_retriever(
search_kwargs={"k": 5}
),
prompt=self.history_aware_prompt
)
# QA chain (stuff documents)
self.qa_chain = create_stuff_documents_chain(
llm=self.llm,
prompt=self.qa_prompt
)
# Full retrieval chain
self.rag_chain = create_retrieval_chain(
retriever=self.history_aware_retriever,
combine_docs_chain=self.qa_chain
)
def query(
self,
question: str,
chat_history: Optional[List] = None,
return_sources: bool = True
) -> dict:
"""Thực hiện query với RAG"""
start_time = time.time()
inputs = {
"input": question,
"chat_history": chat_history or []
}
response = self.rag_chain.invoke(inputs)
elapsed = time.time() - start_time
result = {
"answer": response["answer"],
"latency_ms": round(elapsed * 1000, 2),
"question": question
}
if return_sources:
result["sources"] = [
{
"content": doc.page_content[:200] + "...",
"metadata": doc.metadata
}
for doc in response.get("context", [])
]
return result
def batch_query(self, questions: List[str]) -> List[dict]:
"""Batch query để benchmark"""
results = []
for q in questions:
result = self.query(q)
results.append(result)
print(f"Q: {q[:50]}... | Latency: {result['latency_ms']}ms")
return results
Benchmark với HolySheep AI
if __name__ == "__main__":
from vector_store import VectorStoreManager
from document_processor import BlockchainDocumentProcessor
from amberdata_extractor import AmberdataExtractor
import asyncio
# Khởi tạo components
vector_manager = VectorStoreManager()
vector_manager.initialize(api_key="YOUR_HOLYSHEEP_API_KEY")
# Tạo sample data (thay thế bằng real Amberdata call)
sample_data = [
{"type": "gas_price", "blockNumber": 19500000, "slow": "20000000000", "standard": "25000000000", "fast": "30000000000", "instant": "35000000000"},
{"type": "token_transfer", "hash": "0x123...abc", "from": "0xA...", "to": "0xB...", "value": "1000000000000000000", "token": "ETH"},
{"type": "defi_metrics", "protocol": "uniswap-v2", "tvl": 150000000, "volume24h": 50000000, "fees24h": 250000}
]
# Setup RAG chain
rag = BlockchainRAGChain(
vectorstore=vector_manager.vectorstore,
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1" # $8/MTok với HolySheep
)
# Test query
result = rag.query("What is the current ETH gas price?")
print(f"\nAnswer: {result['answer']}")
print(f"Latency: {result['latency_ms']}ms")Tối ưu hóa hiệu suất: Batch Processing và Concurrency
# performance_optimization.py
import asyncio
import httpx
import time
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import threading
@dataclass
class PerformanceMetrics:
"""Metrics cho performance monitoring"""
total_requests: int
successful_requests: int
failed_requests: int
total_latency_ms: float
avg_latency_ms: float
p50_latency_ms: float
p95_latency_ms: float
p99_latency_ms: float
throughput_rps: float
cost_usd: float
class ConcurrencyController:
"""Kiểm soát concurrency để tránh rate limit và optimize throughput"""
def __init__(
self,
max_concurrent: int = 10,
requests_per_minute: int = 60,
retry_attempts: int = 3,
retry_delay: float = 1.0
):
self.max_concurrent = max_concurrent
self.requests_per_minute = requests_per_minute
self.retry_attempts = retry_attempts
self.retry_delay = retry_delay
self._semaphore = asyncio.Semaphore(max_concurrent)
self._rate_limiter = asyncio.Semaphore(requests_per_minute // 60)
self._lock = threading.Lock()
self._request_count = 0
self._start_time = time.time()
async def execute_with_retry(
self,
func: Callable,
*args,
**kwargs
) -> Any:
"""Execute function với retry và concurrency control"""
async with self._semaphore:
async with self._rate_limiter:
for attempt in range(self.retry_attempts):
try:
with self._lock:
self._request_count += 1
result = await func(*args, **kwargs)
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Rate limited - wait và retry
wait_time = self.retry_delay * (2 ** attempt)
print(f"Rate limited, waiting {wait_time}s...")
await asyncio.sleep(wait_time)
elif e.response.status_code >= 500:
# Server error - retry
await asyncio.sleep(self.retry_delay)
else:
raise
except Exception as e:
if attempt == self.retry_attempts - 1:
raise
await asyncio.sleep(self.retry_delay)
def get_rate_limited_client(self) -> httpx.AsyncClient:
"""Tạo HTTP client với rate limiting"""
return httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(
max_connections=self.max_concurrent,
max_keepalive_connections=5
)
)
class BatchProcessor:
"""Batch processor với optimization strategies"""
def __init__(
self,
concurrency_controller: ConcurrencyController,
batch_size: int = 50,
embedding_batch_size: int = 100
):
self.controller = concurrency_controller
self.batch_size = batch_size
self.embedding_batch_size = embedding_batch_size
async def process_amberdata_batches(
self,
addresses: List[str],
fetch_func: Callable
) -> List[Dict]:
"""Process Amberdata fetches trong batches"""
all_results = []
# Semaphore để giới hạn concurrent requests
semaphore = asyncio.Semaphore(self.controller.max_concurrent)
async def fetch_with_limit(addr: str) -> Dict:
async with semaphore:
return await self.controller.execute_with_retry(
fetch_func, addr
)
# Fetch all addresses concurrently
tasks = [fetch_with_limit(addr) for addr in addresses]
# Process in chunks để tránh memory issues
for i in range(0, len(tasks), self.batch_size):
chunk = tasks[i:i + self.batch_size]
results = await asyncio.gather(*chunk, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
print(f"Error fetching: {result}")
else:
all_results.append(result)
return all_results
async def process_embeddings(
self,
texts: List[str],
embed_func: Callable
) -> List[List[float]]:
"""Process embeddings với batching để optimize cost"""
all_embeddings = []
for i in range(0, len(texts), self.embedding_batch_size):
batch = texts[i:i + self.embedding_batch_size]
try:
embeddings = await self.controller.execute_with_retry(
embed_func, batch
)
all_embeddings.extend(embeddings)
print(f"Processed {len(all_embeddings)}/{len(texts)} embeddings")
except Exception as e:
print(f"Batch embedding failed: {e}")
# Retry với smaller batch
for text in batch:
try:
emb = await self.controller.execute_with_retry(
embed_func, [text]
)
all_embeddings.append(emb[0])
except Exception as e2:
print(f"Single embedding failed: {e2}")
return all_embeddings
class CostOptimizer:
"""Tối ưu chi phí khi sử dụng HolySheep AI"""
# HolySheep Pricing 2026
PRICING = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $2/$8 per M tokens
"gpt-4o": {"input": 2.5, "output": 10.0},
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 0.35, "output": 2.50},
"deepseek-v3.2": {"input": 0.08, "output": 0.42}
}
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
self.costs = {}
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Tính chi phí cho một request"""
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
def estimate_batch_cost(
self,
model: str,
num_requests: int,
avg_input_tokens: int,
avg_output_tokens: int
) -> Dict[str, float]:
"""Ước tính chi phí cho batch processing"""
single_cost = self.calculate_cost(
model, avg_input_tokens, avg_output_tokens
)
return {
"per_request": single_cost,
"total_cost": single_cost * num_requests,
"avg_cost_per_1k": (single_cost / avg_output_tokens) * 1000
}
def select_optimal_model(
self,
task_type: str,
quality_requirement: float
) -> str:
"""Chọn model tối ưu cost-quality tradeoff"""
if task_type == "embedding":
return "text-embedding-3-small" # Cheap embedding
if quality_requirement >= 0.95:
return "claude-sonnet-4.5" # Best quality
elif quality_requirement >= 0.85:
return "gpt-4.1" # Good quality, reasonable cost