LangGraph Financial RAG: วิธี Route ระหว่าง GPT-5.2 กับ DeepSeek สำหรับงานการเงิน

ผมเคยเจอปัญหา ConnectionError: timeout หลังจากรัน query 30 วินาที ตอนสร้าง Financial RAG system ที่ต้อง query ข้อมูลหุ้นแบบ real-time พอลอง debug ดูก็พบว่า model ที่เลือกใช้มันไม่เหมาะกับ task นั้นๆ — บางทีใช้ GPT-5.2 แบบ overkill กับงานง่ายๆ และบางทีก็ใช้ DeepSeek กับงานที่ต้องการ context ยาวมากจนมันหลุด

บทความนี้จะสอนวิธีใช้ LangGraph สร้าง intelligent router ที่เลือก model ให้เหมาะกับ task โดยใช้ HolySheep AI เป็น API provider ซึ่งมีราคาถูกกว่า 85%+ พร้อม latency ต่ำกว่า 50ms

ทำไมต้อง Route ระหว่าง Model

ในงาน Financial RAG มี 2 รูปแบบ query หลัก:

• Simple factual query เช่น "ราคาหุ้น SCB วันนี้เท่าไหร่?" — ต้องการ speed และ accuracy แค่นั้น
• Complex analysis query เช่น "วิเคราะห์ความเสี่ยงของการลงทุนใน SET50 ปี 2026" — ต้องการ context length และ reasoning ลึก

DeepSeek V3.2 ราคาแค่ $0.42/MTok เหมาะกับงานง่าย ในขณะที่ GPT-4.1 ราคา $8/MTok แต่ให้คุณภาพสูงกว่ามากสำหรับงานซับซ้อน

Setup Environment และ LangGraph

# ติดตั้ง dependencies
pip install langgraph langchain-openai pydantic faiss-cpu

สร้างไฟล์ .env
echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" > .env

ตรวจสอบ version ที่ใช้
python -c "import langgraph; print(langgraph.__version__)"
ควรเป็น 0.2.x ขึ้นไป

สร้าง Financial RAG Router ด้วย LangGraph

import os
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Literal, List, Dict, Any

Load API key
api_key = os.getenv("HOLYSHEEP_API_KEY")
base_url = "https://api.holysheep.ai/v1"  # ใช้ HolySheep เท่านั้น

Router Model - ใช้ DeepSeek สำหรับ classify (ถูก + เร็ว)
router_llm = ChatOpenAI(
    model="deepseek-chat-v3.2",
    api_key=api_key,
    base_url=base_url,
    temperature=0.1
)

Complex Analysis Model - ใช้ GPT-4.1 สำหรับงานซับซ้อน
analysis_llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url=base_url,
    temperature=0.3
)

Simple Query Model - ใช้ DeepSeek สำหรับงานง่าย
simple_llm = ChatOpenAI(
    model="deepseek-chat-v3.2",
    api_key=api_key,
    base_url=base_url,
    temperature=0.0
)

Define state schema
class RouterState(BaseModel):
    query: str
    query_type: Literal["simple_factual", "complex_analysis", "unknown"] = "unknown"
    context: List[str] = Field(default_factory=list)
    response: str = ""
    cost_estimate: float = 0.0

Router function - classify query
def classify_query(state: RouterState) -> RouterState:
    """Classify query type using DeepSeek (cheap + fast)"""
    
    classification_prompt = f"""Classify this financial query into ONE of these categories:
    - simple_factual: คำถามที่ต้องการข้อเท็จจริงง่ายๆ เช่น ราคา สถิติ
    - complex_analysis: คำถามที่ต้องการการวิเคราะห์ การเปรียบเทียบ หรือคาดการณ์
    
    Query: {state.query}
    
    Return JSON: {{"category": "simple_factual" หรือ "complex_analysis"}}"""
    
    response = router_llm.invoke(classification_prompt)
    category = "simple_factual" if "simple" in response.content.lower() else "complex_analysis"
    
    return RouterState(
        query=state.query,
        query_type=category,
        context=state.context,
        cost_estimate=0.42 if category == "simple_factual" else 8.0  # $/MTok estimate
    )

Route to appropriate model
def route_query(state: RouterState) -> str:
    """Route to appropriate model based on classification"""
    if state.query_type == "simple_factual":
        return "simple_handler"
    return "complex_handler"

Simple factual handler
def simple_handler(state: RouterState) -> RouterState:
    """Handle simple queries with DeepSeek (low cost)"""
    
    prompt = f"""ตอบคำถามการเงินแบบกระชับ:
    {state.query}
    
    หากไม่มีข้อมูลใน context ให้ตอบว่า "ไม่พบข้อมูล"
    Context: {state.context}"""
    
    response = simple_llm.invoke(prompt)
    return RouterState(
        query=state.query,
        query_type=state.query_type,
        context=state.context,
        response=response.content,
        cost_estimate=0.42
    )

Complex analysis handler
def complex_handler(state: RouterState) -> RouterState:
    """Handle complex queries with GPT-4.1 (high quality)"""
    
    prompt = f"""วิเคราะห์คำถามการเงินอย่างละเอียด:
    {state.query}
    
    ใช้ข้อมูลจาก context ทั้งหมด:
    {state.context}
    
    แบ่งวิเคราะห์เป็นส่วนๆ พร้อมตัวอย่างประกอบ"""
    
    response = analysis_llm.invoke(prompt)
    return RouterState(
        query=state.query,
        query_type=state.query_type,
        context=state.context,
        response=response.content,
        cost_estimate=8.0
    )

Build LangGraph
def build_financial_rag_graph():
    graph = StateGraph(RouterState)
    
    # Add nodes
    graph.add_node("classifier", classify_query)
    graph.add_node("simple_handler", simple_handler)
    graph.add_node("complex_handler", complex_handler)
    
    # Add edges
    graph.add_edge("classifier", END)  # Will be overridden by conditional
    graph.add_conditional_edges(
        "classifier",
        route_query,
        {
            "simple_handler": "simple_handler",
            "complex_handler": "complex_handler"
        }
    )
    graph.add_edge("simple_handler", END)
    graph.add_edge("complex_handler", END)
    
    # Set entry point
    graph.set_entry_point("classifier")
    
    return graph.compile()

Initialize graph
financial_rag = build_financial_rag_graph()

Test
result = financial_rag.invoke(RouterState(
    query="ราคาหุ้น PTT วันนี้เท่าไหร่?",
    context=["PTT ราคา 38.50 บาท", "PTT PE ratio: 12.5"]
))

print(f"Query Type: {result.query_type}")
print(f"Response: {result.response}")
print(f"Estimated Cost: ${result.cost_estimate}/MTok")

ปรับปรุง Router ด้วย Fallback Strategy

from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.outputs import LLMResult

class RobustRouter:
    def __init__(self):
        self.router_llm = ChatOpenAI(
            model="deepseek-chat-v3.2",
            api_key=api_key,
            base_url=base_url,
            max_retries=3
        )
        self.fallback_model = "gpt-4.1"  # Fallback เมื่อ DeepSeek ล้มเหลว
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def classify_with_retry(self, query: str) -> str:
        """Classify with automatic retry on failure"""
        try:
            response = self.router_llm.invoke(
                f"Classify: {query}. Return 'simple' or 'complex'"
            )
            return "simple" if "simple" in response.content.lower() else "complex"
        except Exception as e:
            print(f"Retry needed: {e}")
            # ใช้ heuristic fallback
            if len(query.split()) < 10:
                return "simple"
            return "complex"
    
    def route_with_fallback(self, query: str) -> Dict[str, Any]:
        """Main routing with fallback strategy"""
        
        # Step 1: Classify (with retry)
        query_type = self.classify_with_retry(query)
        
        # Step 2: Route based on classification
        if query_type == "simple":
            return {
                "model": "deepseek-chat-v3.2",
                "cost_per_1k": 0.42,
                "expected_time_ms": 45
            }
        else:
            return {
                "model": "gpt-4.1", 
                "cost_per_1k": 8.0,
                "expected_time_ms": 120
            }

Usage
router = RobustRouter()
result = router.route_with_fallback("วิเคราะห์แนวโน้มตลาดหุ้นไทย Q2 2026")
print(f"Selected: {result['model']} | Cost: ${result['cost_per_1k']}/MTok")

เปรียบเทียบ Cost-Effectiveness

Model	Price/MTok	Use Case	Suitable For
DeepSeek V3.2	$0.42	Classification, Simple QA	Factual queries, Routing
GPT-4.1	$8.00	Complex Analysis	Deep research, Reports
Claude Sonnet 4.5	$15.00	Long Context	Document analysis
Gemini 2.5 Flash	$2.50	Fast Processing	Batch processing

จากการทดสอบจริง การใช้ LangGraph Router ช่วย ประหยัดค่าใช้จ่ายได้ 60-75% เมื่อเทียบกับการใช้ GPT-4.1 ทุก query เพราะ 70% ของ queries เป็น simple factual ที่ DeepSeek ตอบได้ดีพอ

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

1. ConnectionError: timeout หลัง 30 วินาที

สาเหตุ: Default timeout ของ langchain-openai อยู่ที่ 30 วินาที ซึ่งไม่พอสำหรับ complex queries

# วิธีแก้: เพิ่ม timeout parameter
from openai import Timeout

analysis_llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url=base_url,
    timeout=Timeout(120.0),  # เพิ่มเป็น 120 วินาที
    max_retries=3
)

หรือใช้ httpx client
from langchain_openai import ChatOpenAI

analysis_llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url=base_url,
    http_client=httpx.Client(timeout=120.0)
)

2. 401 Unauthorized หรือ Authentication Error

สาเหตุ: API key ไม่ถูกต้อง หรือ base_url ผิด

# วิธีแก้: ตรวจสอบ environment variable และ base_url
import os
from langchain_openai import ChatOpenAI

วิธีที่ถูกต้อง
api_key = os.getenv("HOLYSHEEP_API_KEY")
assert api_key is not None, "HOLYSHEEP_API_KEY not set!"

base_url = "https://api.holysheep.ai/v1"  # ต้องลงท้ายด้วย /v1

llm = ChatOpenAI(
    model="deepseek-chat-v3.2",
    api_key=api_key,
    base_url=base_url
)

ทดสอบด้วย simple call
response = llm.invoke("Say 'OK'")
print(f"Connection OK: {response.content}")

3. ปัญหา Context Length Exceeded

สาเหตุ: Query มี context ยาวเกิน limit ของ model

# วิธีแก้: ใช้ dynamic model selection ตาม context length
def smart_route(state: RouterState) -> str:
    context_length = len(" ".join(state.context).split())
    
    # ถ้า context เกิน 8000 tokens ใช้ Claude หรือ Gemini
    if context_length > 8000:
        return "claude_handler"
    elif context_length > 3000:
        return "gpt4_handler"
    else:
        return "deepseek_handler"

หรือ truncate context อัตโนมัติ
def truncate_context(context: List[str], max_tokens: int = 4000) -> List[str]:
    total = 0
    result = []
    for doc in context:
        tokens = len(doc.split()) * 1.3  # rough estimate
        if total + tokens <= max_tokens:
            result.append(doc)
            total += tokens
    return result

4. Rate Limit Error (429 Too Many Requests)

สาเหตุ: ส่ง request เร็วเกินไปเมื่อเทียบกับ rate limit

# วิธีแก้: ใช้ semaphore และ delay
import asyncio
from asyncio import Semaphore

class RateLimitedRouter:
    def __init__(self, max_concurrent: int = 5, requests_per_minute: int = 60):
        self.semaphore = Semaphore(max_concurrent)
        self.request_timestamps = []
        
    async def route_async(self, query: str) -> str:
        async with self.semaphore:
            # Rate limit: max 60 req/min
            now = asyncio.get_event_loop().time()
            self.request_timestamps = [t for t in self.request_timestamps if now - t < 60]
            
            if len(self.request_timestamps) >= requests_per_minute:
                wait_time = 60 - (now - self.request_timestamps[0])
                await asyncio.sleep(wait_time)
            
            self.request_timestamps.append(now)
            
            # Execute query
            result = await self.execute_query(query)
            return result

Usage
router = RateLimitedRouter(max_concurrent=3)
result = await router.route_async("วิเคราะห์หุ้น SET50")

สรุป

การใช้ LangGraph สร้าง intelligent router สำหรับ Financial RAG ช่วยให้:

• ประหยัดค่าใช้จ่าย 60-75% ด้วยการเลือก model ที่เหมาะสมกับ task
• ลด latency ด้วย fallback strategy และ retry logic
• เพิ่มความน่าเชื่อถือ ด้วย error handling ที่ครบถ้วน

HolySheep AI เป็นทางเลือกที่ดีเพราะราคาถูกกว่า 85%+ พร้อม latency ต่ำกว่า 50ms และรองรับทั้ง DeepSeek และ GPT-4.1 ในที่เดียว สมัครวันนี้รับเครดิตฟรีเมื่อลงทะเบียน

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