Python CLI สรุปไฟล์เยอะๆ ด้วย Gemini ไม่ต้องกลัวหลุด ไม่ต้องรอนาน

สวัสดีครับเพื่อนๆ โปรแกรมเมอร์

วันนี้ผมจะพาเพื่อนๆ มาสร้างเครื่องมือ CLI แบบง่ายๆ ด้วย Python กันนะครับ เอาไว้สรุปไฟล์ข้อความเยอะๆ ที่มีอยู่ในเครื่องของเรานี่ละครับ ใช้ Gemini API มาช่วยให้งานของเรามันจบได้ไวขึ้น แถมยังทำแบบมี Resillience กับ Optimization ด้วยนะ เพื่อให้มันไม่หลุดกลางทาง แล้วก็ทำงานได้แบบไหลลื่นเลยละครับ

ปัญหาที่เราเจอบ่อยๆ คืออะไร?

เราอาจจะมีไฟล์ text หลายสิบ หลายร้อยไฟล์เลยนะครับ ที่ต้องส่งไปให้ Gemini สรุปเนื้อหาให้

• ถ้าส่งไปทีเดียวหมด มันก็อาจจะโดน Rate Limit หรือ API มีปัญหาชั่วคราวได้
• ส่งไปทีละไฟล์ก็ช้าอีก

แบบนี้มันไม่เวิร์คเลยนะครับ เรามาทำให้มันดีขึ้นกันดีกว่าครับ

1. เตรียม Gemini API กันก่อน

อันดับแรกเพื่อนๆ ต้องมี API Key ของ Gemini ก่อนนะครับ เข้าไปเอาได้ที่ Google AI Studio นะครับ จากนั้นก็ Install Library นี้เลยครับ

pip install google-generativeai python-dotenv

แล้วก็สร้างไฟล์ .env ไว้เก็บ API Key นะครับ ปลอดภัยกว่าเอาไป hardcode ในโค๊ดตรงๆ นะ

GOOGLE_API_KEY=\"YOUR_API_KEY_ตรงนี้\"

โค๊ดเริ่มต้นของเราก็จะประมาณนี้นะครับ

import google.generativeai as genai
import os
from dotenv import load_dotenv

load_dotenv()
genai.configure(api_key=os.getenv(\"GOOGLE_API_KEY\"))

model = genai.GenerativeModel('gemini-pro')

def summarize_text(text: str) -> str:
    try:
        response = model.generate_content(f\"กรุณาสรุปข้อความต่อไปนี้:\
{text}\")
        return response.text
    except Exception as e:
        print(f\"เกิดข้อผิดพลาดในการสรุป: {e}\")
        return \"\"

# ลองทดสอบดู
# content = \"เนื้อหาบทความยาวๆ ตรงนี้...\"
# summary = summarize_text(content)
# print(summary)

2. จัดการไฟล์เยอะๆ ด้วย File-Handling และ Context Manager

ทีนี้เรามาทำให้ CLI ของเราอ่านไฟล์ได้ทีละหลายๆ ไฟล์กันครับ แล้วก็เขียนผลลัพธ์ลงไฟล์ใหม่ด้วยนะ เพื่อให้โค๊ดมันดูดีขึ้น เราจะใช้ context-manager หรือ with open(...) นี่ละครับ มันจะช่วยให้เราไม่ต้องมานั่ง close() ไฟล์เอง ปลอดภัยกว่าเยอะเลย

import pathlib

def process_files(input_dir: str, output_dir: str):
    input_path = pathlib.Path(input_dir)
    output_path = pathlib.Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    for file_path in input_path.glob('*.txt'): # หาไฟล์ .txt ทั้งหมด
        print(f\"กำลังประมวลผลไฟล์: {file_path.name}\")
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        summary = summarize_text(content)

        output_file = output_path / f\"summary_{file_path.name}\"
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(summary)
        print(f\"บันทึกสรุปแล้วที่: {output_file.name}\")

# วิธีเรียกใช้งาน
# process_files('./input_texts', './output_summaries')

3. เพิ่ม Resilience ให้ CLI ของเรา ไม่ให้ API ล่มง่ายๆ

API มันก็มีหลุดบ้างนะครับ หรือโดน Rate Limit บ้าง เราเลยต้องทำใจเย็นๆ ลองส่งใหม่ดูหน่อยครับ ถ้ามันล่ม เราจะรอซักพักแล้วลองใหม่ แบบนี้เรียกว่า Resilience นะครับ ผมจะใช้ time.sleep แบบง่ายๆ ก่อนนะ เพื่อนๆ อาจจะใช้ Library อย่าง tenacity ก็ได้นะครับ แต่เพื่อความกระชับเอาแบบนี้ไปก่อนครับ

import time

MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 5

def summarize_text_with_resilience(text: str) -> str:
    for attempt in range(MAX_RETRIES):
        try:
            response = model.generate_content(f\"กรุณาสรุปข้อความต่อไปนี้:\
{text}\")
            return response.text
        except Exception as e:
            print(f\"ครั้งที่ {attempt+1}/{MAX_RETRIES} เกิดข้อผิดพลาด: {e}\")
            if attempt < MAX_RETRIES - 1:
                print(f\"รอ {RETRY_DELAY_SECONDS} วินาที แล้วลองใหม่นะครับ...\")
                time.sleep(RETRY_DELAY_SECONDS)
            else:
                print(\"ลองใหม่ครบจำนวนแล้วครับ ไม่สามารถสรุปข้อความนี้ได้\")
    return \"\"

# ทีนี้ก็เอาไปใส่ใน process_files ของเราได้เลย
# summary = summarize_text_with_resilience(content)

4. Optimization: ประมวลผลแบบเป็นชุดๆ ไม่ต้องรอนาน

เพื่อไม่ให้เราโดน Rate Limit หรือทำให้ API ทำงานหนักเกินไปนะครับ เราอาจจะเพิ่มดีเลย์เล็กน้อยระหว่างการเรียก API แต่ละครั้งก็ได้ครับ หรือจะ Batch (รวม) ไฟล์เล็กๆ หลายๆ ไฟล์แล้วส่งไปทีเดียวก็ได้เหมือนกันนะครับ แต่สำหรับตัวอย่างนี้ ผมจะเน้นที่การเพิ่มดีเลย์นิดหน่อยนะครับ เพื่อให้ API มีเวลาพักบ้าง

# เพิ่มในส่วนของ process_files นะครับ

# ... โค๊ดเดิม ...

    for i, file_path in enumerate(input_path.glob('*.txt')):
        print(f\"กำลังประมวลผลไฟล์: {file_path.name}\")
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        summary = summarize_text_with_resilience(content) # ใช้ฟังก์ชันที่มี Resilience

        output_file = output_path / f\"summary_{file_path.name}\"
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(summary)
        print(f\"บันทึกสรุปแล้วที่: {output_file.name}\")

        if i < len(list(input_path.glob('*.txt'))) - 1: # ไม่ต้องรอถ้าเป็นไฟล์สุดท้าย
            API_CALL_DELAY = 1 # วินาที
            print(f\"รอ {API_CALL_DELAY} วินาที ก่อนเรียก API ครั้งต่อไปนะครับ...\")
            time.sleep(API_CALL_DELAY)

# ... โค๊ดเดิม ...

สุดท้ายนี้ครับ

เพื่อนๆ ลองเอาไปปรับใช้กับงานของตัวเองดูได้นะครับ จะเห็นว่าเราสามารถสร้าง CLI ง่ายๆ ที่มีความสามารถในการจัดการไฟล์ การเชื่อมต่อกับ API และยังเพิ่มความทนทาน (Resilience) กับการเพิ่มประสิทธิภาพ (Optimization) ได้ด้วยนะ โดยไม่ต้องเขียนโค๊ดอะไรที่ซับซ้อนเลยครับ ลองดูนะครับ

อ้างอิง: * Google AI Studio Documentation * Python Pathlib Module

บทความนี้ละ "