Khi xây dựng hệ thống sinh ảnh AI quy mô lớn, tôi đã thử nghiệm qua nhiều giải pháp và nhận ra rằng việc kết hợp GPT-4o cho xử lý ngôn ngữ với DALL-E 3 cho sinh ảnh là combo mạnh mẽ nhất hiện nay. Bài viết này chia sẻ kinh nghiệm thực chiến khi triển khai kiến trúc này với HolySheep AI — nền tảng API trung gian với tỷ giá quy đổi cực kỳ hấp dẫn.
Tại Sao Chọn HolySheep AI Thay Vì Direct API?
Trước khi đi vào code, tôi muốn giải thích lý do tôi chọn HolySheep làm lớp trung gian. Với chi phí chỉ từ $0.42/MTok cho DeepSeek V3.2 và thanh toán qua WeChat/Alipay, đây là giải pháp tối ưu cho các dự án hướng thị trường châu Á. Độ trễ trung bình dưới 50ms giúp đảm bảo trải nghiệm người dùng mượt mà.
Kiến Trúc Tổng Quan
┌─────────────────────────────────────────────────────────────┐
│ Client Application │
├─────────────────────────────────────────────────────────────┤
│ API Gateway / Load Balancer │
├─────────────────────────────────────────────────────────────┤
│ Application Layer (Your Backend) │
│ ┌─────────────────┐ ┌─────────────────┐ ┌──────────────┐ │
│ │ GPT-4o Client │ │ DALL-E 3 Client │ │ Task Queue │ │
│ │ (Text/Context) │ │ (Image Gen) │ │ (Redis/Kafka)│ │
│ └────────┬────────┘ └────────┬────────┘ └──────┬───────┘ │
├───────────┴────────────────────┴────────────────────┴─────────┤
│ HolySheep AI Gateway │
│ base_url: https://api.holysheep.ai/v1 │
└─────────────────────────────────────────────────────────────┘
Setup Client Với Retry Logic và Timeout
Đây là production-ready client mà tôi sử dụng trong dự án thực tế, bao gồm retry mechanism với exponential backoff:
import openai
import httpx
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass
import logging
@dataclass
class HolySheepConfig:
"""Cấu hình cho HolySheep AI API"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
max_retries: int = 3
timeout: float = 30.0
max_concurrent_requests: int = 10
class HolySheepAIClient:
"""
Production client cho GPT-4o và DALL-E 3 integration.
Hỗ trợ concurrency control, retry logic, và rate limiting.
"""
def __init__(self, config: HolySheepConfig):
self.config = config
self.logger = logging.getLogger(__name__)
# HTTP client với connection pooling
self._http_client = httpx.AsyncClient(
base_url=config.base_url,
timeout=httpx.Timeout(config.timeout),
limits=httpx.Limits(
max_connections=config.max_concurrent_requests * 2,
max_keepalive_connections=config.max_concurrent_requests
)
)
# Semaphore để kiểm soát đồng thời
self._semaphore = asyncio.Semaphore(config.max_concurrent_requests)
# OpenAI-compatible client
self.client = openai.AsyncOpenAI(
api_key=config.api_key,
base_url=config.base_url,
http_client=self._http_client
)
async def generate_with_retry(
self,
prompt: str,
model: str = "gpt-4o",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""GPT-4o với automatic retry và backoff"""
async with self._semaphore:
last_error = None
for attempt in range(self.config.max_retries):
try:
response = await self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Bạn là trợ lý AI chuyên nghiệp."},
{"role": "user", "content": prompt}
],
temperature=temperature,
max_tokens=max_tokens
)
return {
"content": response.choices[0].message.content,
"model": response.model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
},
"latency_ms": response.response_headers.get("x-response-time", 0)
}
except openai.RateLimitError as e:
last_error = e
wait_time = 2 ** attempt # Exponential backoff
self.logger.warning(f"Rate limit hit, retrying in {wait_time}s...")
await asyncio.sleep(wait_time)
except httpx.TimeoutException as e:
last_error = e
self.logger.error(f"Timeout on attempt {attempt + 1}: {e}")
except Exception as e:
last_error = e
self.logger.error(f"Unexpected error: {e}")
break
raise RuntimeError(f"Failed after {self.config.max_retries} retries: {last_error}")
Khởi tạo client
config = HolySheepConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_retries=3,
timeout=30.0,
max_concurrent_requests=10
)
client = HolySheepAIClient(config)
Tích Hợp DALL-E 3 Với Batch Processing
Sinh ảnh với DALL-E 3 đòi hỏi chiến lược khác so với text generation. Tôi áp dụng batch processing để tối ưu chi phí:
import asyncio
from typing import List, Dict, Optional
from enum import Enum
import json
import hashlib
class ImageSize(str, Enum):
"""Kích thước ảnh DALL-E 3"""
SQUARE_1024 = "1024x1024"
PORTRAIT_1024x1792 = "1024x1792"
LANDSCAPE_1792x1024 = "1792x1024"
class ImageQuality(str, Enum):
STANDARD = "standard"
HD = "hd" # Chi phí cao hơn 2x
class DALL_E3Service:
"""
Service cho DALL-E 3 integration với HolySheep AI.
Hỗ trợ batch processing và image variation generation.
"""
def __init__(self, client: HolySheepAIClient):
self.client = client
self._image_cache: Dict[str, str] = {}
async def generate_image(
self,
prompt: str,
size: ImageSize = ImageSize.SQUARE_1024,
quality: ImageQuality = ImageQuality.STANDARD,
style: str = "vivid"
) -> Dict[str, Any]:
"""Sinh ảnh đơn với DALL-E 3"""
# Check cache trước
cache_key = hashlib.md5(
f"{prompt}{size}{quality}".encode()
).hexdigest()
if cache_key in self._image_cache:
self.logger.info("Using cached image result")
return {"url": self._image_cache[cache_key], "cached": True}
try:
response = await self.client.client.images.generate(
model="dall-e-3",
prompt=prompt,
size=size.value,
quality=quality.value,
style=style,
n=1
)
result = {
"url": response.data[0].url,
"revised_prompt": response.data[0].revised_prompt,
"cached": False
}
# Cache kết quả
self._image_cache[cache_key] = result["url"]
return result
except Exception as e:
self.logger.error(f"DALL-E 3 generation failed: {e}")
raise
async def generate_batch(
self,
prompts: List[str],
size: ImageSize = ImageSize.SQUARE_1024,
max_concurrent: int = 3
) -> List[Dict[str, Any]]:
"""
Batch processing cho nhiều prompts.
Giới hạn concurrent requests để tránh rate limit.
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def generate_with_limit(prompt: str, index: int) -> Dict[str, Any]:
async with semaphore:
try:
result = await self.generate_image(prompt, size)
return {"index": index, "success": True, **result}
except Exception as e:
return {"index": index, "success": False, "error": str(e)}
# Execute all tasks
tasks = [
generate_with_limit(prompt, idx)
for idx, prompt in enumerate(prompts)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Sort theo index gốc
return sorted(
[r if isinstance(r, dict) else {"success": False, "error": str(r)}
for r in results],
key=lambda x: x["index"]
)
Ví dụ sử dụng batch generation
async def main():
dalle_service = DALL_E3Service(client)
# Prompt list cho marketing campaign
product_prompts = [
"A sleek smartphone with holographic display, studio lighting, white background",
"Professional laptop setup with dual monitors, minimalist desk, morning light",
"Wireless earbuds with charging case, soft shadows, pastel gradient background",
"Smartwatch fitness tracker on wrist, active lifestyle, outdoor setting"
]
results = await dalle_service.generate_batch(
prompts=product_prompts,
size=ImageSize.PORTRAIT_1024x1792,
max_concurrent=2
)
for result in results:
if result["success"]:
print(f"Image {result['index']}: {result['url']}")
else:
print(f"Image {result['index']} failed: {result['error']}")
Chạy với asyncio
asyncio.run(main())
Tối Ưu Chi Phí Với Token Management
Đây là phần quan trọng nhất — tối ưu chi phí khi chạy production. HolySheep cung cấp bảng giá cạnh tranh:
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime
import asyncio
@dataclass
class ModelPricing:
"""Bảng giá HolySheep AI 2026 (USD per 1M tokens)"""
GPT_4_1: float = 8.0
CLAUDE_SONNET_4_5: float = 15.0
GEMINI_2_5_FLASH: float = 2.50
DEEPSEEK_V3_2: float = 0.42
DALL_E_3_1024: float = 0.04 # Per image
DALL_E_3_HD: float = 0.08 # Per HD image
class CostOptimizer:
"""
Tối ưu chi phí bằng cách chọn model phù hợp cho từng task.
Sử dụng caching và batching để giảm tổng chi phí.
"""
def __init__(self):
self.pricing = ModelPricing()
self.usage_stats: Dict[str, int] = {}
self.cost_limit_daily = 100.0 # Giới hạn chi phí ngày
def calculate_text_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""Tính chi phí text generation"""
rate = getattr(self.pricing, model.upper().replace("-", "_").replace(".", "_"), None)
if rate is None:
raise ValueError(f"Unknown model: {model}")
total_tokens = prompt_tokens + completion_tokens
cost = (total_tokens / 1_000_000) * rate
# Track usage
self.usage_stats[model] = self.usage_stats.get(model, 0) + total_tokens
return cost
def calculate_image_cost(self, size: str, quality: str = "standard") -> float:
"""Tính chi phí image generation"""
if quality == "hd":
return self.pricing.DALL_E_3_HD
return self.pricing.DALL_E_3_1024
async def smart_routing(
self,
task_type: str,
complexity: str,
max_budget: Optional[float] = None
) -> str:
"""
Chọn model tối ưu chi phí dựa trên task complexity.
Args:
task_type: "chat", "coding", "analysis", "creative"
complexity: "low", "medium", "high"
max_budget: Ngân sách tối đa cho request
Returns:
Model name được chọn
"""
routing_rules = {
("chat", "low"): "deepseek-v3.2",
("chat", "medium"): "gemini-2.5-flash",
("chat", "high"): "gpt-4.1",
("coding", "low"): "gemini-2.5-flash",
("coding", "medium"): "deepseek-v3.2",
("coding", "high"): "gpt-4.1",
("analysis", "low"): "deepseek-v3.2",
("analysis", "medium"): "gemini-2.5-flash",
("analysis", "high"): "claude-sonnet-4.5",
}
model = routing_rules.get((task_type, complexity))
if model is None:
model = "gpt-4.1" # Default fallback
return model
def get_cost_report(self) -> Dict[str, any]:
"""Tạo báo cáo chi phí chi tiết"""
total_cost = 0
model_costs = {}
for model, tokens in self.usage_stats.items():
rate = getattr(self.pricing, model.upper().replace("-", "_").replace(".", "_"), 0)
cost = (tokens / 1_000_000) * rate
model_costs[model] = {
"total_tokens": tokens,
"cost_usd": round(cost, 4)
}
total_cost += cost
return {
"total_cost_usd": round(total_cost, 4),
"models": model_costs,
"estimated_savings_vs_openai": round(total_cost * 0.85, 4), # ~85% cheaper
"generated_at": datetime.now().isoformat()
}
Benchmark thực tế
async def benchmark_models():
"""So sánh performance và cost giữa các model"""
optimizer = CostOptimizer()
test_prompts = [
"Giải thích quantum computing trong 50 từ",
"Viết code Python để sort array",
"Phân tích pros/cons của microservices"
]
models_to_test = [
("gpt-4.1", "high"),
("gemini-2.5-flash", "medium"),
("deepseek-v3.2", "low")
]
results = []
for model, complexity in models_to_test:
times = []
costs = []
for prompt in test_prompts:
start = asyncio.get_event_loop().time()
try:
response = await client.generate_with_retry(
prompt=prompt,
model=model,
max_tokens=500
)
elapsed = (asyncio.get_event_loop().time() - start) * 1000
cost = optimizer.calculate_text_cost(
model,
response["usage"]["prompt_tokens"],
response["usage"]["completion_tokens"]
)
times.append(elapsed)
costs.append(cost)
except Exception as e:
print(f"Error with {model}: {e}")
if times:
results.append({
"model": model,
"avg_latency_ms": round(sum(times) / len(times), 2),
"total_cost_usd": round(sum(costs), 4),
"complexity": complexity
})
# In benchmark results
print("\n=== BENCHMARK RESULTS ===")
for r in sorted(results, key=lambda x: x["total_cost_usd"]):
print(f"{r['model']:20} | Latency: {r['avg_latency_ms']:6.2f}ms | Cost: ${r['total_cost_usd']:.4f}")
return results
Chạy benchmark
asyncio.run(benchmark_models())
Concurrency Control Với Rate Limiter
Để đảm bảo hệ thống ổn định khi có nhiều requests đồng thời, tôi implement custom rate limiter:
import time
import asyncio
from typing import Dict, Optional
from collections import deque
from dataclasses import dataclass, field
@dataclass
class RateLimitConfig:
"""Cấu hình rate limiting"""
requests_per_minute: int = 60
requests_per_second: int = 10
tokens_per_minute: int = 100_000
burst_size: int = 20
class TokenBucketRateLimiter:
"""
Token bucket algorithm cho rate limiting chính xác.
Hỗ trợ burst traffic nhưng vẫn giữ nguyên average rate.
"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.tokens = config.burst_size
self.last_update = time.time()
self.rate = config.requests_per_second
self._lock = asyncio.Lock()
async def acquire(self, tokens_needed: int = 1) -> float:
"""
Acquire tokens từ bucket.
Returns: Số giây phải đợi trước khi có thể proceed
"""
async with self._lock:
now = time.time()
# Refill tokens dựa trên thời gian trôi qua
elapsed = now - self.last_update
self.tokens = min(
self.config.burst_size,
self.tokens + elapsed * self.rate
)
self.last_update = now
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return 0.0 # Không cần đợi
# Tính thời gian đợi để có đủ tokens
wait_time = (tokens_needed - self.tokens) / self.rate
return wait_time
class SlidingWindowCounter:
"""
Sliding window counter cho tracking requests theo thời gian.
Chính xác hơn fixed window counter.
"""
def __init__(self, window_size_seconds: int = 60, max_requests: int = 60):
self.window_size = window_size_seconds
self.max_requests = max_requests
self.requests: deque = deque()
self._lock = asyncio.Lock()
async def is_allowed(self) -> bool:
"""Kiểm tra xem request có được phép không"""
async with self._lock:
now = time.time()
window_start = now - self.window_size
# Remove requests cũ
while self.requests and self.requests[0] < window_start:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
async def time_until_next_slot(self) -> float:
"""Trả về số giây cho đến khi có slot trống"""
async with self._lock:
if len(self.requests) < self.max_requests:
return 0.0
oldest = self.requests[0]
return max(0.0, self.window_size - (time.time() - oldest))
class HolySheepRateLimiter:
"""
Rate limiter tổng hợp cho HolySheep API.
Kết hợp Token Bucket và Sliding Window.
"""
def __init__(self):
self.bucket = TokenBucketRateLimiter(RateLimitConfig())
self.window = SlidingWindowCounter(window_size_seconds=60, max_requests=60)
self._total_requests = 0
self._total_wait_time = 0.0
async def wait_if_needed(self) -> Dict[str, any]:
"""Wait nếu cần thiết, trả về thông tin về wait time"""
start_wait = time.time()
# Check sliding window
if not await self.window.is_allowed():
wait_time = await self.window.time_until_next_slot()
await asyncio.sleep(wait_time)
# Check token bucket
bucket_wait = await self.bucket.acquire()
if bucket_wait > 0:
await asyncio.sleep(bucket_wait)
total_wait = time.time() - start_wait
self._total_requests += 1
self._total_wait_time += total_wait
return {
"allowed": True,
"wait_time_ms": round(total_wait * 1000, 2),
"total_requests": self._total_requests,
"avg_wait_time_ms": round((self._total_wait_time / self._total_requests) * 1000, 2)
}
def get_stats(self) -> Dict[str, any]:
"""Lấy statistics của rate limiter"""
return {
"total_requests": self._total_requests,
"avg_wait_time_ms": round((self._total_wait_time / max(1, self._total_requests)) * 1000, 2),
"current_bucket_tokens": round(self.bucket.tokens, 2)
}
Sử dụng rate limiter
rate_limiter = HolySheepRateLimiter()
async def rate_limited_request(prompt: str):
"""Wrapper để thực hiện request với rate limiting"""
wait_info = await rate_limiter.wait_if_needed()
if wait_info["wait_time_ms"] > 100:
print(f"Waited {wait_info['wait_time_ms']}ms for rate limit")
response = await client.generate_with_retry(prompt)
return {
**response,
"rate_limit_info": wait_info
}
Monitoring Và Observability
Production system cần monitoring đầy đủ. Đây là setup logging và metrics:
import logging
from typing import Dict, Any
from datetime import datetime
import json
Structured logging setup
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
)
class APIMetrics:
"""
Metrics collector cho HolySheep API calls.
Theo dõi latency, success rate, và costs.
"""
def __init__(self):
self.metrics: Dict[str, list] = {
"latencies": [],
"errors": [],
"costs": [],
"models": {}
}
self._lock = asyncio.Lock()
async def record_request(
self,
model: str,
latency_ms: float,
success: bool,
cost_usd: float,
error: Optional[str] = None
):
"""Record một API request"""
async with self._lock:
self.metrics["latencies"].append({
"timestamp": datetime.now().isoformat(),
"model": model,
"latency_ms": latency_ms,
"success": success
})
self.metrics["costs"].append(cost_usd)
if model not in self.metrics["models"]:
self.metrics["models"][model] = {
"requests": 0,
"errors": 0,
"total_latency_ms": 0
}
self.metrics["models"][model]["requests"] += 1
self.metrics["models"][model]["total_latency_ms"] += latency_ms
if not success:
self.metrics["errors"].append({
"model": model,
"error": error,
"timestamp": datetime.now().isoformat()
})
self.metrics["models"][model]["errors"] += 1
def get_summary(self) -> Dict[str, Any]:
"""Lấy summary metrics"""
latencies = [m["latency_ms"] for m in self.metrics["latencies"]]
return {
"total_requests": len(self.metrics["latencies"]),
"success_rate": round(
(len(self.metrics["latencies"]) - len(self.metrics["errors"]))
/ max(1, len(self.metrics["latencies"])) * 100,
2
),
"latency_p50_ms": round(sorted(latencies)[len(latencies)//2] if latencies else 0, 2),
"latency_p95_ms": round(
sorted(latencies)[int(len(latencies)*0.95)] if latencies else 0,
2
),
"latency_p99_ms": round(
sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0,
2
),
"total_cost_usd": round(sum(self.metrics["costs"]), 4),
"models": {
model: {
"requests": data["requests"],
"errors": data["errors"],
"avg_latency_ms": round(
data["total_latency_ms"] / max(1, data["requests"]),
2
),
"error_rate": round(
data["errors"] / max(1, data["requests"]) * 100,
2
)
}
for model, data in self.metrics["models"].items()
}
}
def export_json(self, filepath: str):
"""Export metrics ra JSON file"""
with open(filepath, 'w') as f:
json.dump({
"summary": self.get_summary(),
"raw_latencies": self.metrics["latencies"][-100:], # Last 100
"raw_errors": self.metrics["errors"][-100:]
}, f, indent=2)
Metrics instance
metrics = APIMetrics()
Decorator cho automatic metrics collection
def track_metrics(metrics: APIMetrics):
"""Decorator để tự động track metrics cho async functions"""
def decorator(func):
async def wrapper(*args, **kwargs):
start = asyncio.get_event_loop().time()
success = False
error = None
try:
result = await func(*args, **kwargs)
success = True
return result
except Exception as e:
error = str(e)
raise
finally:
latency_ms = (asyncio.get_event_loop().time() - start) * 1000
model = kwargs.get('model', 'unknown')
await metrics.record_request(
model=model,
latency_ms=latency_ms,
success=success,
cost_usd=0, # Calculate separately
error=error
)
return wrapper
return decorator
Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi Rate Limit (429 Too Many Requests)
Triệu chứng: Request bị rejected với HTTP 429, Response body chứa "Rate limit exceeded"
# ❌ Code sai - không handle rate limit
async def bad_example():
response = await client.generate_with_retry(prompt="Hello")
return response
✅ Code đúng - implement retry với backoff
async def good_example():
max_retries = 3
for attempt in range(max_retries):
try:
response = await client.generate_with_retry(prompt="Hello")
return response
except openai.RateLimitError as e:
if attempt == max_retries - 1:
raise
# Parse retry-after header nếu có
retry_after = getattr(e.response, 'headers', {}).get('retry-after', '1')
wait_time = int(retry_after) if retry_after.isdigit() else 2 ** attempt
logging.warning(f"Rate limited. Waiting {wait_time}s before retry...")
await asyncio.sleep(wait_time)
except Exception as e:
logging.error(f"Unexpected error: {e}")
raise
2. Timeout Khi Sinh Ảnh DALL-E 3
Triệu chứng: DALL-E 3 requests timeout sau 30s, đặc biệt với ảnh HD
# ❌ Timeout quá ngắn cho DALL-E 3
async def bad_dalle_call():
response = await client.client.images.generate(
model="dall-e-3",
prompt="Complex detailed illustration...",
timeout=10.0 # Too short!
)
return response
✅ Tăng timeout và sử dụng polling cho long-running tasks
async def good_dalle_call():
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def generate_with_long_timeout():
try:
response = await client.client.images.generate(
model="dall-e-3",
prompt="Complex detailed illustration...",
quality="hd",
timeout=120.0 # 2 phút cho HD images
)
return response
except httpx.TimeoutException:
logging.warning("DALL-E 3 timeout, retrying...")
# Clear cache để tránh stale state
client._image_cache.clear()
raise
return await generate_with_long_timeout()
3. Chi Phí Vượt Ngân Sách Do Caching Không Hoạt Động
Triệu chứng: Billing tăng đột biến dù traffic không đổi, nhiều requests trùng lặp
# ❌ Không có cache hoặc cache không persistent
class BadCache:
def __init__(self):
self.cache = {} # Reset mỗi lần restart
async def get(self, key):
return None # Cache luôn miss
✅ Persistent cache với TTL và deduplication
import hashlib
import asyncio
from typing import Optional
class PersistentImageCache:
"""Cache với Redis backend cho production"""
def __init__(self, redis_client, ttl_seconds: int = 86400):
self.redis = redis_client
self.ttl = ttl_seconds
self._pending: Dict[str, asyncio.Task] = {}
self._lock = asyncio.Lock()
def _hash_prompt(self, prompt: str, **params) -> str:
"""Tạo unique key cho cache"""
content = json.dumps({"prompt": prompt, **params}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
async def get_or_generate(
self,
prompt: str,
generator_func,
**params
) -> str:
"""
Get từ cache hoặc generate mới.
Sử dụng distributed lock để tránh duplicate generation.
"""
cache_key = self._hash_prompt(prompt, **params)
# Try get from cache
cached = await self.redis.get(f"dalle_cache:{cache_key}")
if cached:
logging.info(f"Cache HIT for key: {cache_key[:16]}...")
return json.loads(cached)["url"]
# Check nếu đang có pending request cho same key
async with self._lock:
if cache_key in self._pending:
logging.info("Waiting for existing generation...")
return await self._pending[cache_key]
# Create new task
self._pending[cache_key] = asyncio.create_task(
generator_func(prompt, **params)
)
try:
# Wait for result
result = await self._pending[cache_key