Python Async Generator: การเรียก AI API แบบ Streaming พร้อมการตรวจสอบความคืบหน้า

ในยุคที่ AI กลายเป็นหัวใจหลักของการพัฒนาซอฟต์แวร์ การเรียกใช้ Large Language Model (LLM) อย่างมีประสิทธิภาพถือเป็นทักษะที่จำเป็น โดยเฉพาะเมื่อต้องการประสบการณ์ผู้ใช้ที่ราบรื่นผ่านการแสดงผลแบบ Streaming ในบทความนี้เราจะมาเรียนรู้การสร้าง Async Generator สำหรับการเรียก AI API แบบ Streaming พร้อมระบบ Progress Monitor ที่ครบวงจร

ทำไมต้อง Streaming?

การเรียก API แบบปกติต้องรอจนได้คำตอบครบ 100% ก่อนจึงแสดงผล ซึ่งอาจใช้เวลา 10-30 วินาทีสำหรับคำตอบที่ยาว แต่การใช้ Streaming ช่วยให้เราแสดงผลทีละ Token ได้ทันที ลด perceived latency ได้อย่างมาก และยังช่วยให้ผู้ใช้เห็นว่า AI กำลังทำงานอยู่

ตารางเปรียบเทียบบริการ AI API

เกณฑ์	HolySheep AI	API อย่างเป็นทางการ	บริการรีเลย์อื่นๆ
อัตราแลกเปลี่ยน	¥1 = $1 (ประหยัด 85%+)	ราคา USD ตามปกติ	มักมี Premium 10-50%
วิธีชำระเงิน	WeChat / Alipay / USDT	บัตรเครดิตระหว่างประเทศ	แตกต่างกันไป
Latency	< 50ms	50-200ms	100-500ms
เครดิตฟรี	มีเมื่อลงทะเบียน	มี $5 ฟรี	น้อยครั้ง
GPT-4.1	$8 / 1M Tokens	$8 / 1M Tokens	$10-15 / 1M Tokens
Claude Sonnet 4.5	$15 / 1M Tokens	$15 / 1M Tokens	$18-22 / 1M Tokens
Gemini 2.5 Flash	$2.50 / 1M Tokens	$2.50 / 1M Tokens	$3-5 / 1M Tokens
DeepSeek V3.2	$0.42 / 1M Tokens	ไม่มี	$0.50-1 / 1M Tokens

การตั้งค่า Environment และ Dependencies

ก่อนเริ่มต้น ให้ติดตั้งแพ็กเกจที่จำเป็น:

pip install openai httpx aiofiles

โครงสร้าง Async Generator พื้นฐาน

Async Generator คือฟังก์ชันที่สามารถหยุดพักและทำงานต่อได้ (Suspend/Resume) โดยไม่บล็อก Event Loop ทำให้เหมาะสำหรับการประมวลผล Streaming Data

import asyncio
import httpx
import json
from typing import AsyncIterator, Dict, Any
from dataclasses import dataclass
from datetime import datetime

@dataclass
class StreamChunk:
    """โครงสร้างข้อมูลสำหรับ Streaming Response"""
    content: str
    is_complete: bool
    token_count: int
    timestamp: datetime
    usage: Dict[str, int] = None

class HolySheepStreamClient:
    """Client สำหรับเรียก HolySheep AI API แบบ Streaming"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.total_tokens = 0
        self.start_time = None
    
    async def stream_chat(
        self,
        model: str,
        messages: list,
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> AsyncIterator[StreamChunk]:
        """
        เรียก API แบบ Streaming และ Yield ข้อมูลทีละส่วน
        
        Args:
            model: ชื่อโมเดล เช่น "gpt-4.1", "claude-sonnet-4-20250514"
            messages: รายการข้อความในรูปแบบ OpenAI format
            temperature: ค่าความสุ่ม (0-2)
            max_tokens: จำนวน Token สูงสุด
        
        Yields:
            StreamChunk: ข้อมูลแต่ละส่วนของ Response
        """
        self.start_time = datetime.now()
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": True
        }
        
        async with httpx.AsyncClient(timeout=60.0) as client:
            async with client.stream(
                "POST",
                f"{self.BASE_URL}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                response.raise_for_status()
                buffer = ""
                
                async for line in response.aiter_lines():
                    if not line.startswith("data: "):
                        continue
                    
                    data = line[6:]  # ตัด "data: " ออก
                    
                    if data == "[DONE]":
                        yield StreamChunk(
                            content="",
                            is_complete=True,
                            token_count=self.total_tokens,
                            timestamp=datetime.now()
                        )
                        break
                    
                    try:
                        chunk_data = json.loads(data)
                        delta = chunk_data.get("choices", [{}])[0].get("delta", {})
                        content = delta.get("content", "")
                        
                        if content:
                            buffer += content
                            self.total_tokens += 1
                            
                            yield StreamChunk(
                                content=content,
                                is_complete=False,
                                token_count=self.total_tokens,
                                timestamp=datetime.now()
                            )
                        
                        # ตรวจสอบ usage เมื่อจบ
                        if chunk_data.get("usage"):
                            final_usage = chunk_data["usage"]
                            yield StreamChunk(
                                content="",
                                is_complete=True,
                                token_count=final_usage.get("total_tokens", self.total_tokens),
                                timestamp=datetime.now(),
                                usage=final_usage
                            )
                            
                    except json.JSONDecodeError:
                        continue

ระบบ Progress Monitor

การมี Progress Monitor ช่วยให้ผู้ใช้เห็นสถานะการทำงาน ความเร็วในการประมวลผล และเวลาที่เหลืออยู่

import asyncio
from typing import Optional
import sys

class ProgressMonitor:
    """ระบบติดตามความคืบหน้าการ Streaming"""
    
    def __init__(self, total_tokens: Optional[int] = None):
        self.total_tokens = total_tokens
        self.processed_tokens = 0
        self.start_time = None
        self.chars_per_second = 0.0
        self.tokens_per_second = 0.0
    
    def start(self):
        self.start_time = asyncio.get_event_loop().time()
    
    def update(self, chunk: StreamChunk):
        """อัปเดตสถานะจาก chunk ล่าสุด"""
        self.processed_tokens = chunk.token_count
        
        if self.start_time:
            elapsed = asyncio.get_event_loop().time() - self.start_time
            if elapsed > 0:
                self.tokens_per_second = self.processed_tokens / elapsed
    
    def get_progress_bar(self, bar_length: int = 30) -> str:
        """สร้าง Progress Bar สำหรับ Terminal"""
        if not self.start_time:
            return "[" + " " * bar_length + "] Initializing..."
        
        if self.total_tokens:
            progress = min(self.processed_tokens / self.total_tokens, 1.0)
            filled = int(bar_length * progress)
            bar = "█" * filled + "░" * (bar_length - filled)
            percentage = progress * 100
            return f"[{bar}] {percentage:.1f}% | {self.processed_tokens}/{self.total_tokens} tokens"
        else:
            elapsed = asyncio.get_event_loop().time() - self.start_time
            return (
                f"[Streaming] {self.processed_tokens} tokens | "
                f"{self.tokens_per_second:.1f} tok/s | "
                f"{elapsed:.1f}s elapsed"
            )
    
    def get_stats(self) -> dict:
        """สถิติภาพรวม"""
        if not self.start_time:
            return {}
        
        elapsed = asyncio.get_event_loop().time() - self.start_time
        return {
            "elapsed_seconds": round(elapsed, 2),
            "tokens_processed": self.processed_tokens,
            "tokens_per_second": round(self.tokens_per_second, 2),
            "estimated_total_time": (
                elapsed * (self.total_tokens / self.processed_tokens)
                if self.processed_tokens > 0 and self.total_tokens
                else None
            )
        }

async def stream_with_progress(
    client: HolySheepStreamClient,
    model: str,
    messages: list,
    show_progress: bool = True
):
    """
    Stream พร้อมแสดง Progress แบบ Real-time
    รวมข้อความทั้งหมดในกระบวนการ
    """
    monitor = ProgressMonitor()
    monitor.start()
    
    full_response = []
    
    async for chunk in client.stream_chat(model, messages):
        monitor.update(chunk)
        full_response.append(chunk.content)
        
        if show_progress and chunk.content:
            # ลบบรรทัดเดิมและพิมพ์ใหม่
            progress = monitor.get_progress_bar()
            print(f"\r{progress}", end="", flush=True)
    
    if show_progress:
        print()  # ขึ้นบรรทัดใหม่
        stats = monitor.get_stats()
        print(f"✨ เสร็จสิ้น! ประมวลผล {stats['tokens_processed']} tokens "
              f"ใน {stats['elapsed_seconds']} วินาที "
              f"({stats['tokens_per_second']:.1f} tokens/s)")
    
    return "".join(full_response)

ตัวอย่างการใช้งาน
async def main():
    client = HolySheepStreamClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    messages = [
        {"role": "system", "content": "คุณเป็นผู้ช่วย AI ที่เป็นมิตร"},
        {"role": "user", "content": "อธิบายเกี่ยวกับ Quantum Computing แบบง่ายๆ"}
    ]
    
    response = await stream_with_progress(
        client,
        model="gpt-4.1",
        messages=messages
    )
    
    print(f"\n📝 คำตอบ:\n{response}")

if __name__ == "__main__":
    asyncio.run(main())

การรวม Progress เข้ากับ Web Interface

สำหรับการใช้งานจริงกับ Web Application เราสามารถส่ง Event Stream ไปยัง Frontend ได้

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(request: Request):
    """
    API Endpoint สำหรับ Streaming Chat
    ส่ง Server-Sent Events (SSE) ไปยัง Frontend
    """
    body = await request.json()
    api_key = body.get("api_key", "YOUR_HOLYSHEEP_API_KEY")
    model = body.get("model", "gpt-4.1")
    messages = body.get("messages", [])
    
    client = HolySheepStreamClient(api_key)
    monitor = ProgressMonitor()
    monitor.start()
    
    async def event_generator():
        async for chunk in client.stream_chat(model, messages):
            monitor.update(chunk)
            
            # ส่งข้อมูลเป็น SSE format
            event_data = {
                "content": chunk.content,
                "is_complete": chunk.is_complete,
                "token_count": chunk.token_count,
                "tokens_per_second": monitor.tokens_per_second,
                "elapsed_seconds": monitor.get_stats().get("elapsed_seconds", 0)
            }
            
            yield f"data: {json.dumps(event_data, ensure_ascii=False)}\n\n"
            
            if chunk.is_complete and chunk.usage:
                usage_event = {
                    "type": "usage",
                    "prompt_tokens": chunk.usage.get("prompt_tokens", 0),
                    "completion_tokens": chunk.usage.get("completion_tokens", 0),
                    "total_tokens": chunk.usage.get("total_tokens", 0)
                }
                yield f"data: {json.dumps(usage_event, ensure_ascii=False)}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

Frontend JavaScript ตัวอย่าง
FRONTEND_JS = '''
async function streamChat() {
    const response = await fetch('/chat/stream', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
            model: 'gpt-4.1',
            messages: [
                {"role": "user", "content": "Hello!"}
            ]
        })
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let fullResponse = '';
    
    while (true) {
        const {done, value} = await reader.read();
        if (done) break;
        
        const lines = decoder.decode(value).split('\\n');
        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = JSON.parse(line.slice(6));
                
                if (data.content) {
                    document.getElementById('output').textContent += data.content;
                    document.getElementById('progress').textContent = 
                        ${data.token_count} tokens | ${data.tokens_per_second.toFixed(1)} tok/s;
                }
                
                if (data.type === 'usage') {
                    console.log('Total tokens used:', data.total_tokens);
                }
            }
        }
    }
}
'''

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: Response ไม่เป็น Streaming Format

อาการ: API คืนค่ากลับมาเป็น JSON ทั้งหมดแทนที่จะเป็น Streaming เมื่อตรวจสอบพบว่าส่ง Parameter ผิด

# ❌ วิธีที่ผิด - ใช้ streaming ใน payload อย่างเดียว
payload = {
    "model": "gpt-4.1",
    "messages": messages,
    "stream": True  # ไม่เพียงพอ
}

✅ วิธีที่ถูกต้อง - ต้องใช้ stream() method ด้วย
async with httpx.AsyncClient() as client:
    async with client.stream("POST", url, json=payload) as response:
        # ต้องใช้ .stream() ไม่ใช่ .post()
        async for line in response.aiter_lines():
            ...

✅ หรือใช้ Official Client
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

stream = await client.chat.completions.create(
    model="gpt-4.1",
    messages=messages,
    stream=True  # ต้องมี flag นี้
)

กรณีที่ 2: Token Rate Limit หรือ Quota Exceeded

อาการ: ได้รับ Error 429 หรือ 400 พร้อมข้อความ "rate limit" หรือ "quota exceeded"

# ❌ วิธีที่ผิด - ไม่มีการจัดการ Rate Limit
async def call_api_once():
    return await client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
        stream=True
    )

✅ วิธีที่ถูกต้อง - Implement Retry with Exponential Backoff
import asyncio
from httpx import HTTPStatusError

class RateLimitHandler:
    def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def call_with_retry(self, func, *args, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return await func(*args, **kwargs)
            except HTTPStatusError as e:
                if e.response.status_code == 429:
                    # Rate limit - รอแล้วลองใหม่
                    delay = self.base_delay * (2 ** attempt)
                    print(f"⏳ Rate limited. Retrying in {delay:.1f}s...")
                    await asyncio.sleep(delay)
                elif e.response.status_code == 400:
                    # อาจเป็น Quota exceeded
                    error_data = e.response.json()
                    if "quota" in error_data.get("error", {}).get("message", "").lower():
                        raise Exception(
                            "Quota exceeded. Please check your HolySheep AI credits."
                        )
                    raise
                else:
                    raise
        raise Exception(f"Failed after {self.max_retries} retries")

การใช้งาน
handler = RateLimitHandler(max_retries=5)

async def safe_stream():
    return handler.call_with_retry(
        client.chat.completions.create,
        model="gpt-4.1",
        messages=messages,
        stream=True
    )

กรรีที่ 3: Connection Timeout และ SSL Error

อาการ: ได้รับข้อผิดพลาด "Connection timeout" หรือ "SSL verification failed" โดยเฉพาะเมื่อเรียกใช้จาก Server ใน China

# ❌ วิธีที่ผิด - ใช้ Timeout สั้นเกินไป
async with httpx.AsyncClient(timeout=10.0) as client:  # 10 วินาทีน้อยเกินไป
    ...

✅ วิธีที่ถูกต้อง - ตั้ง Timeout ที่เหมาะสมและจัดการ SSL
import ssl

async def create_optimized_client():
    """สร้าง Client ที่ปรับแต่งสำหรับ HolySheep API"""
    
    # กรณีต้องการ Bypass SSL (ไม่แนะนำใน Production)
    # ssl_context = ssl.create_default_context()
    # ssl_context.check_hostname = False
    # ssl_context.verify_mode = ssl.CERT_NONE
    
    limits = httpx.Limits(
        max_keepalive_connections=20,
        max_connections=100,
        keepalive_expiry=30.0
    )
    
    return httpx.AsyncClient(
        timeout=httpx.Timeout(
            connect=10.0,    # เชื่อมต่อ
            read=120.0,      # อ่านข้อมูล (Streaming ต้องให้เวลามาก)
            write=10.0,      # เขียนข้อมูล
            pool=30.0        # รอใน Pool
        ),
        limits=limits,
        http2=True  # เปิด HTTP/2 สำหรับความเร็ว
    )

✅ สำหรับ China Server - ใช้ Proxy
async def create_china_client(proxy_url: str = None):
    return httpx.AsyncClient(
        timeout=httpx.Timeout(120.0),
        proxy=proxy_url,  # เช่น "http://127.0.0.1:7890"
        limits=httpx.Limits(max_connections=50)
    )

สรุป

การใช้ Async Generator สำหรับ AI API Streaming เป็นเทคนิคที่ช่วยเพิ่มประสบการณ์ผู้ใช้อย่างมาก โดยผสมผสานความสามารถของ Asyncio เข้ากับการ Stream ข้อมูลแบบ Real-time ทำให้สามารถแสดงผลลัพธ์ทีละส่วนได้ทันที พร้อมระบบ Progress Monitor ที่ช่วยให้ผู้ใช้ติดตามสถานะได้

ด้วย HolySheep AI คุณจะได้รับประโยชน์จากอัตราแลกเปลี่ยนที่คุ้มค่า (¥1 = $1), Latency ต่ำกว่า 50ms, และการรองรับการชำระเงินผ่าน WeChat และ Alipay ทำให้เหมาะอย่างยิ่งสำหรับนักพัฒนาที่ต้องการบริการ AI API คุณภาพสูงในราคาที่เข้าถึงได้

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน