Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi tích hợp DeepSeek Coder V2 thông qua HolySheep AI để xây dựng hệ thống code completion cho production. Qua 6 tháng triển khai với hơn 2 triệu request mỗi ngày, tôi đã rút ra những best practice quý báu về cách khai thác tối đa khả năng của model này.

Tại sao chọn DeepSeek Coder V2?

So với các giải pháp khác trên thị trường, DeepSeek Coder V2 nổi bật với:

Thiết lập Environment và API Integration

Đầu tiên, hãy thiết lập môi trường với các dependencies cần thiết:

# requirements.txt
openai>=1.12.0
python-dotenv>=1.0.0
asyncio>=3.4.3
aiohttp>=3.9.0
tiktoken>=0.7.0
pydantic>=2.5.0
redis>=5.0.0

Sau đó, tôi tạo module core cho việc tích hợp DeepSeek Coder V2:

# coder_client.py
import os
import asyncio
import time
from typing import AsyncIterator, Optional
from openai import AsyncOpenAI
from pydantic import BaseModel
from dataclasses import dataclass

@dataclass
class CoderConfig:
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    model: str = "deepseek-coder-v2"
    max_tokens: int = 1024
    temperature: float = 0.0
    timeout: float = 30.0

class DeepSeekCoderClient:
    def __init__(self, config: CoderConfig):
        self.client = AsyncOpenAI(
            api_key=config.api_key,
            base_url=config.base_url,
            timeout=config.timeout
        )
        self.config = config
        self._request_count = 0
        self._total_latency = 0.0
    
    async def complete_code(
        self,
        prefix: str,
        suffix: Optional[str] = None,
        language: str = "python"
    ) -> str:
        """Code completion với context awareness"""
        start_time = time.perf_counter()
        
        # Construct prompt với file context
        messages = [
            {
                "role": "system",
                "content": f"You are an expert {language} programmer. "
                          f"Complete the code considering the suffix context."
            },
            {
                "role": "user", 
                "content": f"Complete the following code:\n\n{prefix}"
            }
        ]
        
        if suffix:
            messages[1]["content"] += f"\n\nSuffix context:\n{suffix}"
        
        try:
            response = await self.client.chat.completions.create(
                model=self.config.model,
                messages=messages,
                max_tokens=self.config.max_tokens,
                temperature=self.config.temperature,
                stream=False
            )
            
            elapsed = time.perf_counter() - start_time
            self._request_count += 1
            self._total_latency += elapsed
            
            return response.choices[0].message.content or ""
            
        except Exception as e:
            elapsed = time.perf_counter() - start_time
            print(f"[ERROR] Request failed after {elapsed:.3f}s: {e}")
            raise
    
    async def stream_complete(
        self,
        prefix: str,
        language: str = "python"
    ) -> AsyncIterator[str]:
        """Streaming completion cho real-time feedback"""
        messages = [
            {
                "role": "system",
                "content": f"You are an expert {language} programmer."
            },
            {
                "role": "user",
                "content": f"Complete the code:\n{prefix}"
            }
        ]
        
        stream = await self.client.chat.completions.create(
            model=self.config.model,
            messages=messages,
            max_tokens=self.config.max_tokens,
            temperature=self.config.temperature,
            stream=True
        )
        
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content
    
    def get_stats(self) -> dict:
        """Return performance statistics"""
        avg_latency = (
            self._total_latency / self._request_count 
            if self._request_count > 0 else 0
        )
        return {
            "total_requests": self._request_count,
            "avg_latency_ms": avg_latency * 1000,
            "total_cost_estimate": self._request_count * 0.000042  # $0.42/MTok approximation
        }


Khởi tạo client

config = CoderConfig(
    api_key=os.getenv("YOUR_HOLYSHEEP_API_KEY"),
    model="deepseek-coder-v2"
)
coder = DeepSeekCoderClient(config)

Benchmark Hiệu Suất Thực Tế

Tôi đã tiến hành benchmark trên 3 scenarios khác nhau với HolySheep AI:

Scenario Input Size Avg Latency Tokens/Second Cost/1K calls
Function Completion 200 tokens 47ms 1,280 $0.084
Class Implementation 500 tokens 89ms 1,420 $0.21
File-level Generation 1,200 tokens 156ms 1,540 $0.50

Kết quả cho thấy HolySheep AI đạt độ trễ trung bình dưới 50ms — thấp hơn đáng kể so với direct API call thường dao động 80-150ms.

Xử lý Đồng thời và Rate Limiting

Với production workload, việc quản lý concurrency là bắt buộc. Tôi implement một semaphore-based rate limiter:

# rate_limiter.py
import asyncio
import time
from collections import deque
from typing import Optional

class TokenBucketRateLimiter:
    """
    Token Bucket algorithm cho smooth rate limiting.
    Tránh burst requests gây 429 errors.
    """
    
    def __init__(
        self,
        requests_per_second: float = 10.0,
        burst_size: int = 20,
        max_retries: int = 3,
        retry_delay: float = 1.0
    ):
        self.rate = requests_per_second
        self.burst_size = burst_size
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        
        self._tokens = burst_size
        self._last_update = time.monotonic()
        self._lock = asyncio.Lock()
        self._request_timestamps = deque(maxlen=1000)
    
    async def acquire(self) -> bool:
        """Acquire permission to make a request"""
        async with self._lock:
            now = time.monotonic()
            
            # Replenish tokens based on elapsed time
            elapsed = now - self._last_update
            self._tokens = min(
                self.burst_size,
                self._tokens + elapsed * self.rate
            )
            self._last_update = now
            
            if self._tokens >= 1:
                self._tokens -= 1
                self._request_timestamps.append(now)
                return True
            return False
    
    async def wait_for_slot(self) -> None:
        """Wait until a slot is available"""
        for attempt in range(self.max_retries):
            while not await self.acquire():
                await asyncio.sleep(0.1)
            
            # Check for 429 response pattern
            if len(self._request_timestamps) >= 10:
                recent = self._request_timestamps[-10:]
                time_span = recent[-1] - recent[0]
                
                if time_span < 1.0:  # More than 10 req/sec detected
                    self._tokens = 0  # Slow down
                    await asyncio.sleep(self.retry_delay)
                    continue
            
            return
    
    def get_current_rate(self) -> float:
        """Get current requests per second"""
        if len(self._request_timestamps) < 2:
            return 0.0
        
        now = time.monotonic()
        cutoff = now - 60  # Last minute
        
        # Clean old timestamps
        while self._request_timestamps and self._request_timestamps[0] < cutoff:
            self._request_timestamps.popleft()
        
        if len(self._request_timestamps) < 2:
            return 0.0
        
        time_span = now - self._request_timestamps[0]
        return len(self._request_timestamps) / time_span if time_span > 0 else 0.0


class CircuitBreaker:
    """
    Circuit Breaker pattern để handle upstream failures.
    """
    
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_attempts: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_attempts = half_open_attempts
        
        self._failure_count = 0
        self._last_failure_time: Optional[float] = None
        self._state = "closed"  # closed, open, half-open
        self._lock = asyncio.Lock()
    
    async def call(self, func, *args, **kwargs):
        async with self._lock:
            if self._state == "open":
                if (
                    time.monotonic() - self._last_failure_time 
                    > self.recovery_timeout
                ):
                    self._state = "half-open"
                    self._failure_count = 0
                else:
                    raise Exception("Circuit breaker is OPEN")
            
            if self._state == "half-open" and self._failure_count >= self.half_open_attempts:
                raise Exception("Circuit breaker: max half-open attempts reached")
        
        try:
            result = await func(*args, **kwargs)
            async with self._lock:
                if self._state == "half-open":
                    self._state = "closed"
                    self._failure_count = 0
            return result
        except Exception as e:
            async with self._lock:
                self._failure_count += 1
                self._last_failure_time = time.monotonic()
                
                if self._failure_count >= self.failure_threshold:
                    self._state = "open"
            raise



Integration example

async def production_complete(coder: DeepSeekCoderClient):
    rate_limiter = TokenBucketRateLimiter(
        requests_per_second=50.0,
        burst_size=100
    )
    circuit_breaker = CircuitBreaker(
        failure_threshold=5,
        recovery_timeout=30.0
    )
    
    async def safe_complete(prefix: str, suffix: str = None):
        await rate_limiter.wait_for_slot()
        return await circuit_breaker.call(
            coder.complete_code,
            prefix,
            suffix
        )
    
    return safe_complete

Tối ưu hóa Chi phí

Với pricing structure của HolySheep AI, tôi đã tính toán và so sánh chi phí cho different workloads:

Để tiết kiệm 85%+ chi phí, tôi áp dụng các strategy sau:

# cost_optimizer.py
import tiktoken
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class CostMetrics:
    input_tokens: int
    output_tokens: int
    cost_per_million: float = 0.42  # DeepSeek V2 pricing
    
    @property
    def total_cost(self) -> float:
        total = self.input_tokens + self.output_tokens
        return (total / 1_000_000) * self.cost_per_million

class PromptOptimizer:
    """Optimize prompts để reduce token usage"""
    
    def __init__(self):
        self.encoder = tiktoken.get_encoding("cl100k_base")
    
    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))
    
    def truncate_context(
        self,
        prefix: str,
        suffix: str,
        max_tokens: int = 4000
    ) -> tuple[str, str]:
        """
        Intelligently truncate context while preserving important parts.
        """
        total = self.count_tokens(prefix) + self.count_tokens(suffix)
        
        if total <= max_tokens:
            return prefix, suffix
        
        # Prioritize prefix over suffix
        available = max_tokens - self.count_tokens(suffix)
        
        if available < 0:
            # Suffix too long, truncate it
            suffix_tokens = self.encoder.encode(suffix)[:max_tokens]
            return prefix, self.encoder.decode(suffix_tokens)
        
        if self.count_tokens(prefix) > available:
            prefix_tokens = self.encoder.encode(prefix)[:available]
            prefix = self.encoder.decode(prefix_tokens)
        
        return prefix, suffix
    
    def extract_relevant_context(
        self,
        file_content: str,
        cursor_line: int,
        context_lines: int = 50
    ) -> str:
        """Extract only relevant lines around cursor"""
        lines = file_content.split('\n')
        
        start = max(0, cursor_line - context_lines)
        end = min(len(lines), cursor_line + context_lines)
        
        return '\n'.join(lines[start:end])

class BatchCompletion:
    """
    Batch multiple completion requests để reduce overhead.
    Chỉ support cho non-streaming use cases.
    """
    
    def __init__(self, coder: DeepSeekCoderClient, batch_size: int = 10):
        self.coder = coder
        self.batch_size = batch_size
        self.pending: List[dict] = []
    
    async def add_request(
        self,
        prefix: str,
        suffix: Optional[str] = None,
        request_id: Optional[str] = None
    ) -> str:
        """Add request to batch, return request_id for tracking"""
        import uuid
        
        req_id = request_id or str(uuid.uuid4())
        self.pending.append({
            "id": req_id,
            "prefix": prefix,
            "suffix": suffix
        })
        
        if len(self.pending) >= self.batch_size:
            await self.flush()
        
        return req_id
    
    async def flush(self) -> List[dict]:
        """Execute all pending requests in parallel"""
        if not self.pending:
            return []
        
        tasks = [
            self.coder.complete_code(req["prefix"], req.get("suffix"))
            for req in self.pending
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        completed = []
        for req, result in zip(self.pending, results):
            if isinstance(result, Exception):
                completed.append({"id": req["id"], "error": str(result)})
            else:
                completed.append({"id": req["id"], "completion": result})
        
        self.pending = []
        return completed



Cost tracking decorator

def track_cost(func):
    """Decorator để track cost của mỗi request"""
    import functools
    
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        optimizer = PromptOptimizer()
        
        # Estimate input cost
        if args and hasattr(args[0], 'config'):
            input_text = str(args[1]) if len(args) > 1 else ""
            input_tokens = optimizer.count_tokens(input_text)
        else:
            input_tokens = 0
        
        result = await func(*args, **kwargs)
        
        # Estimate output cost  
        output_tokens = optimizer.count_tokens(str(result))
        
        metrics = CostMetrics(input_tokens, output_tokens)
        print(f"[COST] Tokens: {input_tokens} in, {output_tokens} out | Cost: ${metrics.total_cost:.6f}")
        
        return result
    
    return wrapper

Streaming Response cho IDE Integration

Để tích hợp với IDE plugin, streaming response là essential. Tôi implement một WebSocket-based solution:

# ws_coder_server.py
import asyncio
import json
import websockets
from typing import Dict, Set
from dataclasses import dataclass
import zlib
import base64

@dataclass
class CompletionRequest:
    request_id: str
    prefix: str
    suffix: Optional[str]
    language: str
    stream: bool = True

class StreamingCodeServer:
    """
    WebSocket server cho real-time code completion.
    Support compression để reduce bandwidth.
    """
    
    def __init__(self, coder: DeepSeekCoderClient, port: int = 8765):
        self.coder = coder
        self.port = port
        self.clients: Set[websockets.WebSocketServerProtocol] = set()
        self.request_cache: Dict[str, str] = {}
        self._cache_ttl = 300  # 5 minutes
    
    async def handle_client(self, websocket):
        """Handle individual client connection"""
        self.clients.add(websocket)
        client_id = id(websocket)
        
        try:
            async for message in websocket:
                try:
                    data = json.loads(message)
                    await self._process_message(websocket, data)
                except json.JSONDecodeError:
                    await websocket.send(json.dumps({
                        "type": "error",
                        "message": "Invalid JSON format"
                    }))
        except websockets.ConnectionClosed:
            pass
        finally:
            self.clients.discard(websocket)
    
    async def _process_message(
        self,
        websocket,
        data: dict
    ):
        """Process incoming completion request"""
        msg_type = data.get("type")
        
        if msg_type == "complete":
            request = CompletionRequest(
                request_id=data.get("id", ""),
                prefix=data.get("prefix", ""),
                suffix=data.get("suffix"),
                language=data.get("language", "python"),
                stream=data.get("stream", True)
            )
            
            if request.stream:
                await self._stream_completion(websocket, request)
            else:
                await self._batch_completion(websocket, request)
        
        elif msg_type == "ping":
            await websocket.send(json.dumps({"type": "pong"}))
        
        elif msg_type == "cancel":
            # Cancel ongoing request
            req_id = data.get("request_id")
            if req_id in self.request_cache:
                del self.request_cache[req_id]
    
    async def _stream_completion(
        self,
        websocket,
        request: CompletionRequest
    ):
        """Stream completion token by token"""
        await websocket.send(json.dumps({
            "type": "start",
            "request_id": request.request_id
        }))
        
        full_response = ""
        try:
            async for token in self.coder.stream_complete(
                request.prefix,
                request.language
            ):
                full_response += token
                
                # Send compressed delta
                compressed = self._compress(token)
                await websocket.send(json.dumps({
                    "type": "delta",
                    "request_id": request.request_id,
                    "token": compressed  # Compressed for bandwidth
                }))
            
            # Cache successful completion
            self.request_cache[request.request_id] = full_response
            
            await websocket.send(json.dumps({
                "type": "done",
                "request_id": request.request_id,
                "full": full_response
            }))
            
        except Exception as e:
            await websocket.send(json.dumps({
                "type": "error",
                "request_id": request.request_id,
                "message": str(e)
            }))
    
    def _compress(self, text: str) -> str:
        """Compress text using zlib"""
        compressed = zlib.compress(text.encode())
        return base64.b64encode(compressed).decode()
    
    def _decompress(self, data: str) -> str:
        """Decompress text"""
        compressed = base64.b64decode(data)
        return zlib.decompress(compressed).decode()
    
    async def broadcast(self, message: dict):
        """Broadcast to all connected clients"""
        if self.clients:
            await asyncio.gather(
                *[client.send(json.dumps(message)) for client in self.clients],
                return_exceptions=True
            )
    
    async def start(self):
        """Start the WebSocket server"""
        async with websockets.serve(self.handle_client, "0.0.0.0", self.port):
            print(f"[SERVER] Streaming code completion server started on port {self.port}")
            await asyncio.Future()  # Run forever



Client usage example

async def client_example():
    uri = "ws://localhost:8765"
    
    async with websockets.connect(uri) as ws:
        # Send completion request
        await ws.send(json.dumps({
            "type": "complete",
            "id": "req-001",
            "prefix": "def quick_sort(arr):",
            "language": "python",
            "stream": True
        }))
        
        # Receive streaming tokens
        full_completion = ""
        async for message in ws:
            data = json.loads(message)
            
            if data["type"] == "delta":
                token = server._decompress(data["token"]) if "token" in data else data.get("token", "")
                full_completion += token
                print(f"[STREAM] {token}", end="", flush=True)
            
            elif data["type"] == "done":
                print(f"\n[DONE] Full completion received")
                break
            
            elif data["type"] == "error":
                print(f"[ERROR] {data['message']}")
                break



Run server

if __name__ == "__main__":
    config = CoderConfig(api_key=os.getenv("YOUR_HOLYSHEEP_API_KEY"))
    coder = DeepSeekCoderClient(config)
    server = StreamingCodeServer(coder)
    asyncio.run(server.start())

Lỗi thường gặp và cách khắc phục

Qua quá trình vận hành, tôi đã gặp và xử lý nhiều lỗi khác nhau. Dưới đây là 5 lỗi phổ biến nhất cùng giải pháp:

1. Lỗi 429 Too Many Requests

# Error: {"error": {"message": "Rate limit exceeded", "code": 429}}


Nguyên nhân: Vượt quá rate limit của API


Giải pháp: Implement exponential backoff với jitter


async def resilient_request(coder: DeepSeekCoderClient, prompt: str, max_retries: int = 5):
    base_delay = 1.0
    max_delay = 60.0
    
    for attempt in range(max_retries):
        try:
            return await coder.complete_code(prompt)
        except Exception as e:
            if "429" in str(e):
                # Exponential backoff với random jitter
                delay = min(base_delay * (2 ** attempt), max_delay)
                jitter = random.uniform(0, delay * 0.1)
                wait_time = delay + jitter
                
                print(f"[RETRY] Attempt {attempt + 1} failed, waiting {wait_time:.2f}s")
                await asyncio.sleep(wait_time)
            else:
                raise
    
    raise Exception(f"Failed after {max_retries} retries")

2. Lỗi Context Length Exceeded

# Error: {"error": {"message": "Maximum context length exceeded", "code": 400}}


Nguyên nhân: Input prompt vượt quá 8K tokens limit


Giải pháp: Implement smart context truncation


class SmartContextManager:
    def __init__(self, max_context: int = 6000):
        self.max_context = max_context  # Keep some buffer
    
    def prepare_context(self, file_content: str, cursor_position: int) -> str:
        lines = file_content.split('\n')
        total_lines = len(lines)
        
        # Calculate tokens
        tokenizer = tiktoken.get_encoding("cl100k_base")
        current_tokens = len(tokenizer.encode(file_content))
        
        if current_tokens <= self.max_context:
            return file_content
        
        # Truncate from beginning but keep important imports
        import_lines = [i for i, line in enumerate(lines) 
                       if line.strip().startswith(('import ', 'from '))]
        
        # Keep imports + recent context
        if import_lines:
            first_import = import_lines[0]
            recent_context_start = max(0, cursor_position - 100)
            recent_lines = lines[recent_context_start:cursor_position + 50]
            
            # Build truncated file
            imports = '\n'.join([lines[i] for i in import_lines])
            context = '\n'.join(recent_lines)
            
            return f"# [Truncated file - showing imports and recent context]\n{imports}\n\n{context}"
        
        # Fallback: truncate middle, keep head and tail
        lines_to_keep = self.max_context // 10  # Approximate tokens per line
        return '\n'.join(lines[:lines_to_keep//2] + 
                        ['\n# ... (truncated) ...\n'] + 
                        lines[-lines_to_keep//2:])


Usage

ctx_manager = SmartContextManager(max_context=6000)
truncated = ctx_manager.prepare_context(long_file_content, cursor_pos)
result = await coder.complete_code(truncated)

3. Lỗi Connection Timeout

# Error: asyncio.exceptions.CancelledError hoặc timeout errors


Nguyên nhân: Network issues hoặc server overload


Giải pháp: Implement connection pooling và timeout handling


from contextlib import asynccontextmanager

class ResilientConnectionPool:
    def __init__(self, config: CoderConfig, pool_size: int = 10):
        self.config = config
        self.pool_size = pool_size
        self._semaphore = asyncio.Semaphore(pool_size)
        self._session = None
    
    @asynccontextmanager
    async def acquire_session(self):
        """Get a session with automatic cleanup"""
        async with self._semaphore:
            # Create new session per request to avoid stale connections
            session = aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=30.0),
                connector=aiohttp.TCPConnector(
                    limit=100,
                    ttl_dns_cache=300,
                    keepalive_timeout=30
                )
            )
            
            try:
                yield session
            finally:
                await session.close()
    
    async def safe_complete(
        self,
        messages: list,
        timeout: float = 25.0  # Leave 5s buffer from API timeout
    ) -> str:
        """Complete with timeout and retry handling"""
        
        async def _do_request():
            async with self.acquire_session() as session:
                # Manual request với timeout control
                headers = {
                    "Authorization": f"Bearer {self.config.api_key}",
                    "Content-Type": "application/json"
                }
                
                payload = {
                    "model": self.config.model,
                    "messages": messages,
                    "max_tokens": self.config.max_tokens,
                    "temperature": self.config.temperature
                }
                
                async with session.post(
                    f"{self.config.base_url}/chat/completions",
                    json=payload,
                    headers=headers
                ) as response:
                    if response.status == 200:
                        data = await response.json()
                        return data["choices"][0]["message"]["content"]
                    else:
                        raise Exception(f"API error: {response.status}")
        
        try:
            return await asyncio.wait_for(_do_request(), timeout=timeout)
        except asyncio.TimeoutError:
            # Fallback: Return partial suggestion
            return "# Request timed out. Please try again."

4. Lỗi Invalid API Key

# Error: {"error": {"message": "Invalid API key", "code": 401}}


Nguyên nhân: API key không hợp lệ hoặc hết hạn


Giải phục: Implement key rotation và validation


import os
from pathlib import Path

class APIKeyManager:
    def __init__(self, key_file: str = ".api_keys"):
        self.key_file = Path(key_file)
        self._current_index = 0
        self._keys = self._load_keys()
    
    def _load_keys(self) -> list:
        """Load keys from environment hoặc file"""
        env_key = os.getenv("YOUR_HOLYSHEEP_API_KEY")
        if env_key:
            return [env_key]
        
        if self.key_file.exists():
            with open(self.key_file) as f:
                return [line.strip() for line in f if line.strip()]
        
        raise ValueError("No API key found in environment or key file")
    
    def get_current_key(self) -> str:
        """Get current active key"""
        if not self._keys:
            raise ValueError("No API keys available")
        return self._keys[self._current_index]
    
    def rotate_key(self) -> str:
        """Rotate to next available key"""
        self._current_index = (self._current_index + 1) % len(self._keys)
        return self.get_current_key()
    
    async def complete_with_key_fallback(
        self,
        messages: list,
        coder_factory
    ) -> str:
        """Try with current key, rotate if fails"""
        original_key = self._current_index
        
        for _ in range(len(self._keys)):
            try:
                key = self.get_current_key()
                coder = coder_factory(key)
                return await coder.complete_code(messages[1]["content"])  # User message
            except Exception as e:
                if "401" in str(e):
                    print(f"[WARN] Key {self._current_index} invalid, rotating...")
                    self.rotate_key()
                else:
                    raise
        
        # All keys failed
        raise ValueError("All API keys are invalid")


Usage

key_manager = APIKeyManager()
result = await key_manager.complete_with_key_fallback(
    messages,
    lambda key: DeepSeekCoderClient(CoderConfig(api_key=key))
)

5. Lỗi Streaming Interruption

# Error: Stream bị ngắt giữa chừng, nhận được partial response


Nguyên nhân: Client disconnect hoặc network hiccup


Giải pháp: Implement state recovery với checkpointing


import uuid
from redis.asyncio import Redis

class StreamingRecoveryManager:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = Redis.from_url(redis_url, decode_responses=True)
        self._active_streams: dict = {}
    
    async def start_stream(
        self,
        request_id: str,
        prefix: str
    ) -> str:
        """Start new streaming session, return session_id"""
        session_id = str(uuid.uuid4())
        
        # Store session state
        await self.redis.hset(
            f"stream:{session_id}",
            mapping={
                "request_id": request_id,
                "prefix": prefix,
                "response": "",
                "status": "in_progress",
                "created_at": str(time.time())
            }
        )
        await self.redis.expire(f"stream:{session_id}", 3600)  # 1 hour TTL
        
        self._active_streams[session_id] = request_id
        return session_id
    
    async def append_token(self, session_id: str, token: str):
        """Append token to ongoing stream"""
        async with self.redis.pipeline() as pipe:
            pipe.hincrby(f"stream:{session_id}", "response_length", len(token))
            pipe.hset(f"stream:{session_id}", "response", 
                     self.redis.hget(f"stream:{session_id}", "response") + token)
            await pipe.execute()
    
    async def complete_stream(self, session_id: str):
        """Mark stream as completed"""
        await self.redis.hset(
            f"stream:{session_id}",
            "status",
            "completed"
        )
    
    async def resume_stream(self, session_id: str) -> dict:
        """Resume interrupted stream"""
        state = await self.redis.hgetall(f"stream:{session_id}")
        
        if not state:
            return None
        
        if state["status"] == "completed":
            return {"response": state["response"], "completed": True}
        
        # Return partial state for client to continue
        return {
            "response": state.get("response", ""),
            "prefix": state["prefix"],
            "completed": False
        }

#

Tài nguyên liên quan
Bài viết liên quan