LangChain กับ Custom Tool: คู่มือสร้าง API Integration และการประมวลผลข้อมูลแบบเต็มรูปแบบ

สรุปก่อนอ่าน: คุณจะได้อะไรจากบทความนี้

บทความนี้จะพาคุณสร้าง Custom Tool ใน LangChain ที่เชื่อมต่อกับ AI API หลากหลายรูปแบบ ตั้งแต่การดึงข้อมูลจากฐานข้อมูล การค้นหาข้อมูลภายนอก จนถึงการประมวลผลข้อมูลแบบเรียลไทม์ พร้อมตารางเปรียบเทียบความคุ้มค่าระหว่าง HolySheep AI กับผู้ให้บริการ API อื่น ๆ โดยเน้นจุดเด่นด้านราคาถูกกว่า 85% และความเร็วตอบสนองน้อยกว่า 50 มิลลิวินาที

ทำไมต้องสร้าง Custom Tool ใน LangChain

LangChain ถูกออกแบบมาให้คุณสามารถขยายความสามารถของ AI Agent ได้อย่างไม่มีขีดจำกัดผ่านระบบ Tool Calling เมื่อคุณสร้าง Custom Tool คุณจะสามารถ:

• ดึงข้อมูลจาก API ภายนอกได้ทุกรูปแบบ
• ประมวลผลข้อมูลเฉพาะโดเมนที่คุณต้องการ
• สร้าง Workflow อัตโนมัติที่ซับซ้อน
• เชื่อมต่อกับฐานข้อมูล SQL/NoSQL ได้โดยตรง

การสร้าง Custom Tool พื้นฐาน

ก่อนอื่น คุณต้องติดตั้งแพ็กเกจที่จำเป็น:

pip install langchain langchain-core langchain-community requests

ตัวอย่างที่ 1: Tool สำหรับดึงข้อมูลสภาพอากาศ

from langchain.tools import tool
from typing import Optional
import requests

@tool
def get_weather(location: str, unit: str = "celsius") -> str:
    """
    ดึงข้อมูลสภาพอากาศปัจจุบันของเมืองที่ระบุ
    
    Args:
        location: ชื่อเมืองที่ต้องการทราบสภาพอากาศ
        unit: หน่วยอุณหภูมิ (celsius หรือ fahrenheit)
    
    Returns:
        ข้อมูลสภาพอากาศในรูปแบบข้อความ
    """
    # เรียกใช้ API ภายนอก
    response = requests.get(
        f"https://api.weather.example.com/current",
        params={"city": location, "unit": unit}
    )
    
    if response.status_code == 200:
        data = response.json()
        return f"อุณหภูมิที่ {location}: {data['temp']}°{unit[0].upper()}"
    else:
        return f"ไม่สามารถดึงข้อมูลสภาพอากาศของ {location} ได้"

ทดสอบการทำงาน
result = get_weather.invoke({"location": "กรุงเทพมหานคร", "unit": "celsius"})
print(result)

ตัวอย่างที่ 2: Tool สำหรับค้นหาข้อมูลจาก API ด้วย LangChain Expression Language (LCEL)

from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
import requests

กำหนด configuration สำหรับ HolySheep AI
HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY"
}

@tool
def search_wikipedia(query: str, language: str = "th") -> str:
    """
    ค้นหาข้อมูลจาก Wikipedia
    
    Args:
        query: คำค้นหา
        language: ภาษาที่ต้องการ (th, en, zh)
    
    Returns:
        ข้อมูลบทความจาก Wikipedia
    """
    base_url = f"https://{language}.wikipedia.org/api/rest_v1/page/summary"
    
    try:
        response = requests.get(base_url, params={"title": query}, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            return f"หัวข้อ: {data.get('title', 'ไม่ระบุ')}\n\n{data.get('extract', 'ไม่พบข้อมูล')}"
        elif response.status_code == 404:
            return f"ไม่พบบทความเกี่ยวกับ '{query}'"
        else:
            return f"เกิดข้อผิดพลาด: HTTP {response.status_code}"
    except requests.exceptions.Timeout:
        return "การค้นหาใช้เวลานานเกินไป กรุณาลองใหม่"
    except Exception as e:
        return f"เกิดข้อผิดพลาด: {str(e)}"

สร้าง Chain ที่ใช้ Tool ร่วมกับ LLM
llm = ChatOpenAI(
    base_url=HOLYSHEEP_CONFIG["base_url"],
    api_key=HOLYSHEEP_CONFIG["api_key"],
    model="gpt-4.1"
)

ผูก Tool เข้ากับ LLM
tools = [search_wikipedia]
llm_with_tools = llm.bind_tools(tools)

ทดสอบการใช้งาน
prompt = ChatPromptTemplate.from_messages([
    ("system", "คุณเป็นผู้ช่วยที่สามารถค้นหาข้อมูลจาก Wikipedia ได้"),
    ("user", "หาข้อมูลเกี่ยวกับ {topic}")
])

chain = prompt | llm_with_tools | StrOutputParser()
result = chain.invoke({"topic": "ปัญญาประดิษฐ์"})
print(result)

การสร้าง Tool สำหรับประมวลผลข้อมูลแบบ Complex

from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
import json

Define input schema
class DataAnalysisInput(BaseModel):
    data: List[dict] = Field(description="รายการข้อมูลที่ต้องการวิเคราะห์")
    analysis_type: str = Field(description="ประเภทการวิเคราะห์: summary, comparison, trend")
    group_by: Optional[str] = Field(default=None, description="ฟิลด์ที่ต้องการจัดกลุ่ม")

@tool(args_schema=DataAnalysisInput)
def analyze_sales_data(
    data: List[dict], 
    analysis_type: str = "summary",
    group_by: Optional[str] = None
) -> str:
    """
    วิเคราะห์ข้อมูลยอดขายแบบหลากหลายรูปแบบ
    
    Args:
        data: ข้อมูลยอดขายในรูปแบบ list of dict
        analysis_type: ประเภทการวิเคราะห์
        group_by: ฟิลด์สำหรับจัดกลุ่ม
    
    Returns:
        ผลการวิเคราะห์ในรูปแบบข้อความ
    """
    
    if not data:
        return "ไม่มีข้อมูลสำหรับวิเคราะห์"
    
    if analysis_type == "summary":
        total = sum(item.get("amount", 0) for item in data)
        count = len(data)
        average = total / count if count > 0 else 0
        
        result = f"สรุปภาพรวม:\n"
        result += f"- ยอดขายรวม: {total:,.2f} บาท\n"
        result += f"- จำนวนรายการ: {count} รายการ\n"
        result += f"- ค่าเฉลี่ยต่อรายการ: {average:,.2f} บาท\n"
        
    elif analysis_type == "comparison" and group_by:
        groups = {}
        for item in data:
            key = item.get(group_by, "unknown")
            if key not in groups:
                groups[key] = []
            groups[key].append(item.get("amount", 0))
        
        result = f"เปรียบเทียบตาม {group_by}:\n"
        for key, amounts in groups.items():
            result += f"- {key}: {sum(amounts):,.2f} บาท ({len(amounts)} รายการ)\n"
    
    elif analysis_type == "trend":
        sorted_data = sorted(data, key=lambda x: x.get("date", ""))
        result = "แนวโน้มยอดขาย:\n"
        for item in sorted_data[:10]:  # แสดง 10 รายการล่าสุด
            result += f"- {item.get('date', 'N/A')}: {item.get('amount', 0):,.2f} บาท\n"
    
    else:
        result = "ประเภทการวิเคราะห์ไม่ถูกต้อง"
    
    return result

ทดสอบการทำงาน
test_data = [
    {"date": "2026-01-01", "amount": 15000, "product": "สินค้า A", "region": "กรุงเทพฯ"},
    {"date": "2026-01-02", "amount": 23000, "product": "สินค้า B", "region": "ภูเก็ต"},
    {"date": "2026-01-03", "amount": 18500, "product": "สินค้า A", "region": "กรุงเทพฯ"},
]

result = analyze_sales_data.invoke({
    "data": test_data,
    "analysis_type": "comparison",
    "group_by": "region"
})
print(result)

ตารางเปรียบเทียบราคาและบริการ AI API

บริการ	ราคา/ล้าน Token	ความหน่วง (Latency)	วิธีชำระเงิน	โมเดลที่รองรับ	ทีมที่เหมาะสม
HolySheep AI	GPT-4.1: $8 Claude Sonnet 4.5: $15 Gemini 2.5 Flash: $2.50 DeepSeek V3.2: $0.42	<50ms	WeChat, Alipay, บัตรเครดิต	GPT-4, Claude, Gemini, DeepSeek	ทีม Startup, นักพัฒนาไทย, ผู้ที่ต้องการประหยัด 85%
OpenAI (API ทางการ)	GPT-4o: $15 GPT-4o-mini: $0.60	100-300ms	บัตรเครดิตระหว่างประเทศ	GPT-4, GPT-3.5	องค์กรใหญ่, บริษัทต่างประเทศ
Anthropic (API ทางการ)	Claude 3.5 Sonnet: $15 Claude 3 Haiku: $1.25	150-400ms	บัตรเครดิตระหว่างประเทศ	Claude 3, Claude 2	ทีมที่เน้นการวิเคราะห์เชิงลึก
Google Gemini API	Gemini 1.5 Pro: $7 Gemini 1.5 Flash: $0.70	80-200ms	บัตรเครดิตระหว่างประเทศ	Gemini 1.5, Gemini 1.0	ทีมที่ใช้ Google Cloud
สรุป: HolySheep AI	💰 ถูกที่สุด 85%+	⚡ เร็วที่สุด	✅ รองรับ WeChat/Alipay	🎯 ครอบคลุมทุกโมเดล	🇹🇭 เหมาะกับนักพัฒนาไทย

การใช้งาน HolySheep AI กับ LangChain Agent

from langchain.agents import AgentType, initialize_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool

กำหนดค่าการเชื่อมต่อ HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

@tool
def calculate_bmi(weight_kg: float, height_m: float) -> str:
    """คำนวณค่า BMI จากน้ำหนักและส่วนสูง"""
    if height_m <= 0:
        return "ส่วนสูงต้องมากกว่า 0"
    
    bmi = weight_kg / (height_m ** 2)
    
    if bmi < 18.5:
        category = "น้ำหนักต่ำกว่าเกณฑ์"
    elif bmi < 25:
        category = "น้ำหนักปกติ"
    elif bmi < 30:
        category = "น้ำหนักเกิน"
    else:
        category = "โรคอ้วน"
    
    return f"ค่า BMI: {bmi:.2f} ({category})"

@tool
def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
    """แปลงสกุลเงิน (สมมติอัตราแลกเปลี่ยน)"""
    rates = {
        ("THB", "USD"): 0.029,
        ("USD", "THB"): 34.5,
        ("CNY", "THB"): 4.8,
        ("THB", "CNY"): 0.21,
    }
    
    rate = rates.get((from_currency, to_currency))
    if rate is None:
        return f"ไม่รองรับการแปลงจาก {from_currency} เป็น {to_currency}"
    
    result = amount * rate
    return f"{amount:,.2f} {from_currency} = {result:,.2f} {to_currency}"

สร้าง LLM instance ด้วย HolySheep
llm = ChatOpenAI(
    base_url=HOLYSHEEP_BASE_URL,
    api_key="YOUR_HOLYSHEEP_API_KEY",
    model="gpt-4.1",
    temperature=0.7
)

รวบรวม tools
tools = [calculate_bmi, convert_currency]

สร้าง Agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

ทดสอบ Agent
response = agent.run("น้ำหนักฉัน 70 กิโลกรัม สูง 1.75 เมตร แปลงเป็น USD ให้หน่อย (ถ้าน้ำหนัก 70 กิโลกรัม)")
print(response)

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

1. ข้อผิดพลาด: API Key ไม่ถูกต้อง (Authentication Error)

# ❌ วิธีที่ผิด - key ไม่ตรงกับ base_url
llm = ChatOpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key="sk-openai-xxxxx"  # ใช้ key ของ OpenAI
)

✅ วิธีที่ถูก - สร้าง key ใหม่จาก HolySheep
llm = ChatOpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY"  # Key จาก HolySheep Dashboard
)

หากยังไม่มี key สามารถสมัครได้ที่: https://www.holysheep.ai/register

2. ข้อผิดพลาด: Rate Limit Exceeded

# ❌ วิธีที่ผิด - เรียกใช้ API มากเกินไปโดยไม่มีการควบคุม
for item in large_dataset:
    result = agent.run(item)  # จะถูก rate limit ทันที

✅ วิธีที่ถูก - ใช้ rate limiter และ retry logic
import time
from functools import wraps

def rate_limit(max_calls=100, period=60):
    def decorator(func):
        calls = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [t for t in calls if now - t < period]
            
            if len(calls) >= max_calls:
                sleep_time = period - (now - calls[0])
                if sleep_time > 0:
                    time.sleep(sleep_time)
            
            calls.append(time.time())
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(max_calls=50, period=60)  # 50 ครั้งต่อนาที
def safe_agent_run(prompt):
    return agent.run(prompt)

3. ข้อผิดพลาด: Tool Schema ไม่ตรงกับ Input

# ❌ วิธีที่ผิด - Schema ไม่ตรงกับ function signature
from pydantic import BaseModel

class WrongSchema(BaseModel):
    query: str
    limit: int  # ฟิลด์นี้ไม่ตรงกับ function

@tool(args_schema=WrongSchema)
def search_data(query: str, category: str = "all"):  # ได้รับ category ไม่ใช่ limit
    pass

✅ วิธีที่ถูก - กำหนด schema ให้ตรงกับ arguments
from pydantic import BaseModel, Field

class CorrectSchema(BaseModel):
    query: str = Field(description="คำค้นหา")
    category: str = Field(default="all", description="หมวดหมู่ที่ต้องการค้นหา")

@tool(args_schema=CorrectSchema)
def search_data(query: str, category: str = "all"):
    """ค้นหาข้อมูลตามหมวดหมู่"""
    # Logic การค้นหา
    return f"ผลการค้นหา '{query}' ในหมวด '{category}'"

4. ข้อผิดพลาด: Timeout ขณะเรียกใช้ API ภายนอก

# ❌ วิธีที่ผิด - ไม่กำหนด timeout
import requests

@tool
def fetch_external_data(url: str):
    response = requests.get(url)  # อาจค้างได้
    return response.json()

✅ วิธีที่ถูก - กำหนด timeout และ handle error
@tool
def fetch_external_data(url: str, timeout: int = 10) -> str:
    """
    ดึงข้อมูลจาก URL ภายนอก
    
    Args:
        url: URL ที่ต้องการดึงข้อมูล
        timeout: ระยะเวลาสูงสุดในการรอ (วินาที)
    """
    try:
        response = requests.get(url, timeout=timeout)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        return f"การเชื่อมต่อใช้เวลานานเกิน {timeout} วินาที"
    except requests.exceptions.ConnectionError:
        return f"ไม่สามารถเชื่อมต่อกับ {url}"
    except requests.exceptions.HTTPError as e:
        return f"HTTP Error: {e}"
    except Exception as e:
        return f"ข้อผิดพลาดที่ไม่ทราบสาเหตุ: {str(e)}"

สรุปและแนะนำ

การสร้าง Custom Tool ใน LangChain เป็นทักษะที่จำเป็นสำหรับนักพัฒนา AI ยุคใหม่ เพราะช่วยให้คุณสามารถขยายความสามารถของ AI Agent ได้อย่างไม่มีขีดจำกัด ไม่ว่าจะเป็นการดึงข้อมูลจากฐานข้อมูล การคำนวณทางคณิตศาสตร์ หรือการเชื่อมต่อกับ API ภายนอก หากคุณกำลังมองหาบริการ AI API ที่ ราคาถูกกว่า 85% และ รองรับ WeChat/Alipay พร้อมความเร็วในการตอบสนองน้อยกว่า 50 มิลลิวินาที HolySheep AI คือตัวเลือกที่เหมาะสมที่สุดสำหรับนักพัฒนาไทย 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน