Kimi API กับการประมวลผลแบบ Batch: พัฒนาเครื่องมือสร้างบทสรุปวรรณกรรมอัตโนมัติ

บทนำ: ทำไมต้อง Batch Processing สำหรับงานวิจัย

ในฐานะนักพัฒนาซอฟต์แวร์ที่ทำงานด้าน AI Application มาหลายปี ผมเคยเจอปัญหาเดิมซ้ำ ๆ กับลูกค้าหลายราย โดยเฉพาะสถาบันวิจัยและมหาวิทยาลัยที่ต้องการสร้างบทสรุปวรรณกรรม (Literature Review) จากเอกสารจำนวนมาก การส่งทีละเอกสารผ่าน API นั้นช้า แพง และไม่มีประสิทธิภาพ วันนี้ผมจะมาแชร์วิธีการใช้ Kimi API (ผ่าน สมัครที่นี่) เพื่อสร้างระบบ Batch Processing ที่ประหยัดเวลาและค่าใช้จ่ายได้ถึง 85% เมื่อเทียบกับการใช้ OpenAI โดยตรง บทความนี้จะอธิบายการพัฒนาเครื่องมือที่รวมเอกสาร PDF หลายร้อยฉบับเข้ามาประมวลผลพร้อมกัน สร้างบทสรุปวรรณกรรมที่มีคุณภาพในระยะเวลาอันสั้น พร้อมโค้ดตัวอย่างที่รันได้จริง

กรณีศึกษา: ระบบ Literature Review อัตโนมัติสำหรับสถาบันวิจัย

สมมติว่าคุณเป็นหัวหน้าโครงการวิจัยที่ต้องสรุปวรรณกรรม 500 บทความภายใน 1 สัปดาห์ วิธีดั้งเดิมใช้เวลา 2-3 ชั่วโมงต่อบทความ รวม 1,000-1,500 ชั่วโมง แต่ด้วยระบบ Batch Processing ที่เราจะสร้าง คุณจะประมวลผลเอกสารทั้ง 500 ฉบับในครั้งเดียว ใช้เวลาประมาณ 30-60 นาที และค่าใช้จ่ายเพียง $0.50-2.00 ผ่าน HolySheep AI

หลักการทำงานของ Batch Processing

Batch Processing คือการรวบรวมงานหลาย ๆ ชิ้นมาประมวลผลพร้อมกันแทนที่จะทำทีละอัน ข้อดีคือ:

• ประหยัดเวลาในการรอ API Response
• ลดจำนวน HTTP Request
• เพิ่ม Throughput ของระบบ
• ประหยัดค่าใช้จ่ายเมื่อใช้กับ Provider ที่มีโครงสร้างราคาต่อ Token ที่ต่ำ

การตั้งค่า Environment และการเชื่อมต่อ API

ก่อนเริ่มต้น คุณต้องติดตั้ง Python dependencies และตั้งค่า API Key:

pip install openai python-dotenv pypdf2 tiktoken aiofiles asyncio

จากนั้นสร้างไฟล์ config.py สำหรับเชื่อมต่อกับ HolySheep AI:

import os
from openai import OpenAI

ตั้งค่า HolySheep AI API
สมัครได้ที่ https://www.holysheep.ai/register
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # ต้องใช้ URL นี้เท่านั้น
)

ตรวจสอบการเชื่อมต่อ
def test_connection():
    try:
        response = client.chat.completions.create(
            model="kimi-pro",
            messages=[{"role": "user", "content": "ทดสอบการเชื่อมต่อ"}],
            max_tokens=10
        )
        print("✅ เชื่อมต่อสำเร็จ!")
        return True
    except Exception as e:
        print(f"❌ เกิดข้อผิดพลาด: {e}")
        return False

if __name__ == "__main__":
    test_connection()

สิ่งสำคัญคือต้องใช้ base_url เป็น https://api.holysheep.ai/v1 เท่านั้น เพราะ HolySheep AI รองรับ OpenAI-compatible API ทำให้คุณสามารถใช้งานได้ทันทีโดยไม่ต้องแก้ไขโค้ดมาก

การสร้าง Batch Processor หลัก

นี่คือหัวใจของระบบ คลาส BatchLiteratureReviewer ที่จัดการกับเอกสารทั้งหมดอย่างมีประสิทธิภาพ:

import json
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional
import PyPDF2

class BatchLiteratureReviewer:
    def __init__(self, client, batch_size: int = 50):
        self.client = client
        self.batch_size = batch_size
        self.results = []
        
    def extract_text_from_pdf(self, pdf_path: str) -> str:
        """ดึงข้อความจากไฟล์ PDF"""
        try:
            with open(pdf_path, 'rb') as file:
                reader = PyPDF2.PdfReader(file)
                text = ""
                for page in reader.pages[:10]:  # จำกัด 10 หน้าแรก
                    text += page.extract_text() + "\n"
                return text
        except Exception as e:
            print(f"ไม่สามารถอ่าน {pdf_path}: {e}")
            return ""
    
    def create_review_prompt(self, documents: List[Dict]) -> str:
        """สร้าง Prompt สำหรับการสร้างบทสรุปวรรณกรรม"""
        combined_content = ""
        for i, doc in enumerate(documents, 1):
            combined_content += f"""
=== เอกสารที่ {i} ===
ชื่อ: {doc.get('title', 'N/A')}
ผู้เขียน: {doc.get('author', 'N/A')}
ปี: {doc.get('year', 'N/A')}
บทคัดย่อ: {doc.get('abstract', doc.get('content', ''))[:500]}
---
"""
        
        prompt = f"""คุณคือนักวิจัยผู้เชี่ยวชาญด้านการสร้างบทสรุปวรรณกรรม
จงวิเคราะห์เอกสารต่อไปนี้และสร้างบทสรุปวรรณกรรมที่ครอบคลุม:

1. ประเด็นหลักที่พบในแต่ละเอกสาร
2. ความเชื่อมโยงระหว่างเอกสาร
3. ช่องว่างของความรู้ที่ยังไม่ได้รับการศึกษา
4. แนวทางการวิจัยในอนาคต

{combined_content}

กรุณาตอบเป็นภาษาไทย พร้อมจัดรูปแบบที่อ่านง่าย"""
        return prompt
    
    def process_batch(self, documents: List[Dict]) -> Dict:
        """ประมวลผลชุดเอกสารหนึ่งชุด"""
        prompt = self.create_review_prompt(documents)
        
        try:
            response = self.client.chat.completions.create(
                model="kimi-pro",
                messages=[
                    {"role": "system", "content": "คุณคือผู้ช่วยวิจัย AI ที่เชี่ยวชาญ"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=4000
            )
            
            return {
                "status": "success",
                "content": response.choices[0].message.content,
                "documents_count": len(documents)
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e),
                "documents_count": len(documents)
            }
    
    def process_all_documents(self, documents: List[Dict]) -> List[Dict]:
        """ประมวลผลเอกสารทั้งหมดเป็นชุด ๆ"""
        all_results = []
        total_batches = (len(documents) + self.batch_size - 1) // self.batch_size
        
        print(f"📚 เอกสารทั้งหมด: {len(documents)} ฉบับ")
        print(f"📦 จำนวน Batch: {total_batches} ชุด")
        print(f"⏳ เริ่มประมวลผล...")
        
        for i in range(0, len(documents), self.batch_size):
            batch_num = i // self.batch_size + 1
            batch = documents[i:i + self.batch_size]
            
            print(f"\n🔄 กำลังประมวลผล Batch {batch_num}/{total_batches}")
            start_time = time.time()
            
            result = self.process_batch(batch)
            result['batch_number'] = batch_num
            result['processing_time'] = time.time() - start_time
            
            all_results.append(result)
            
            print(f"✅ Batch {batch_num} เสร็จสิ้น (ใช้เวลา {result['processing_time']:.2f}s)")
            
            # หน่วงเวลาเพื่อหลีกเลี่ยง Rate Limit
            if batch_num < total_batches:
                time.sleep(1)
        
        return all_results
    
    def generate_final_report(self, results: List[Dict], output_path: str):
        """สร้างรายงานฉบับสมบูรณ์"""
        report = "# 📋 บทสรุปวรรณกรรม (Literature Review)\n\n"
        report += f"**วันที่สร้าง:** {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
        
        success_count = sum(1 for r in results if r['status'] == 'success')
        report += f"## 📊 สรุปผลการประมวลผล\n\n"
        report += f"- จำนวน Batch ที่ประมวลผล: {len(results)}\n"
        report += f"- สำเร็จ: {success_count}\n"
        report += f"- ล้มเหลว: {len(results) - success_count}\n\n"
        
        report += "---\n\n"
        
        for result in results:
            if result['status'] == 'success':
                report += f"## 📑 ส่วนที่ {result['batch_number']}\n\n"
                report += result['content'] + "\n\n"
                report += "---\n\n"
        
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(report)
        
        print(f"💾 รายงานถูกบันทึกที่: {output_path}")
        return report

การใช้งาน Batch Processor ในโปรเจกต์จริง

นี่คือตัวอย่างการใช้งาน Batch Processor กับเอกสารจริง:

import os
from pathlib import Path
from config import client
from batch_literature_reviewer import BatchLiteratureReviewer

def load_documents_from_folder(folder_path: str) -> list:
    """โหลดเอกสารทั้งหมดจากโฟลเดอร์"""
    documents = []
    folder = Path(folder_path)
    
    for pdf_file in folder.glob("*.pdf"):
        reviewer = BatchLiteratureReviewer(client)
        content = reviewer.extract_text_from_pdf(str(pdf_file))
        
        documents.append({
            'path': str(pdf_file),
            'filename': pdf_file.name,
            'title': pdf_file.stem,
            'content': content[:2000]  # จำกัดความยาว
        })
    
    return documents

def main():
    # กำหนดค่าต่าง ๆ
    DOCUMENTS_FOLDER = "./research_papers"
    OUTPUT_FILE = "./literature_review_report.md"
    
    print("🚀 เริ่มระบบสร้างบทสรุปวรรณกรรมอัตโนมัติ\n")
    
    # โหลดเอกสาร
    print("📂 กำลังโหลดเอกสาร...")
    documents = load_documents_from_folder(DOCUMENTS_FOLDER)
    print(f"✅ โหลดเอกสารสำเร็จ: {len(documents)} ฉบับ\n")
    
    # สร้าง Batch Processor
    # batch_size = 10 หมายถึงประมวลผล 10 เอกสารต่อ Batch
    processor = BatchLiteratureReviewer(client, batch_size=10)
    
    # ประมวลผลเอกสารทั้งหมด
    results = processor.process_all_documents(documents)
    
    # สร้างรายงานฉบับสมบูรณ์
    report = processor.generate_final_report(results, OUTPUT_FILE)
    
    print("\n" + "="*50)
    print("🎉 การประมวลผลเสร็จสิ้น!")
    print(f"📄 รายงาน: {OUTPUT_FILE}")

if __name__ == "__main__":
    main()

การคำนวณค่าใช้จ่ายและการเปรียบเทียบราคา

สมมติว่าคุณมีเอกสาร 500