ในโลก AI Agent ยุคใหม่ การสร้าง workflow ที่มี human oversight ไม่ใช่ทางเลือกอีกต่อไป แต่เป็นความจำเป็น โดยเฉพาะเมื่อต้องจัดการกับ sensitive operations เช่น การ approve การทำธุรกรรม การตรวจสอบเนื้อหา หรือการอนุมัติการเปลี่ยนแปลงระบบ บทความนี้จะพาคุณสร้าง production-grade approval workflow ด้วย LangGraph, DeepSeek V4 และ MCP protocol พร้อม benchmark จริงจาก production environment

ทำไมต้อง Human-in-the-Loop?

แม้ AI จะเก่งขึ้นมาก แต่ในบาง scenario การมี human review ยังคงจำเป็นอย่างยิ่ง

สถาปัตยกรรม LangGraph Approval Workflow

LangGraph มีข้อได้เปรียบเหนือ framework อื่นในเรื่อง checkpointing และ interruption ทำให้สร้าง state machine สำหรับ approval ได้อย่างเป็นธรรมชาติ

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime
from typing import Optional
import asyncio

class ApprovalStatus(str, Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    TIMEOUT = "timeout"

class ApprovalRequest(BaseModel):
    request_id: str
    action: str
    params: dict
    risk_level: str = Field(default="low")
    estimated_cost: float = 0.0
    created_at: datetime = Field(default_factory=datetime.utcnow)
    requested_by: str = "agent"

class ApprovalState(BaseModel):
    request: Optional[ApprovalRequest] = None
    status: ApprovalStatus = ApprovalStatus.PENDING
    approver_comments: Optional[str] = None
    approved_at: Optional[datetime] = None
    retry_count: int = 0
    mcp_tool_results: list[dict] = []

class HumanApprovalGraph:
    def __init__(self, llm_client, mcp_tools):
        self.llm = llm_client
        self.mcp_tools = mcp_tools
        self.checkpointer = MemorySaver()
        
    def build_graph(self):
        builder = StateGraph(ApprovalState)
        
        # Nodes
        builder.add_node("analyze_request", self.analyze_request)
        builder.add_node("gather_mcp_context", self.gather_mcp_context)
        builder.add_node("request_approval", self.request_approval)
        builder.add_node("execute_action", self.execute_action)
        builder.add_node("handle_rejection", self.handle_rejection)
        builder.add_node("handle_timeout", self.handle_timeout)
        
        # Edges
        builder.set_entry_point("analyze_request")
        builder.add_edge("analyze_request", "gather_mcp_context")
        builder.add_edge("gather_mcp_context", "request_approval")
        
        # Conditional edges for approval
        builder.add_conditional_edges(
            "request_approval",
            self.route_approval,
            {
                "approved": "execute_action",
                "rejected": "handle_rejection",
                "timeout": "handle_timeout"
            }
        )
        
        builder.add_edge("execute_action", END)
        builder.add_edge("handle_rejection", END)
        builder.add_edge("handle_timeout", END)
        
        return builder.compile(checkpointer=self.checkpointer)
    
    async def analyze_request(self, state: ApprovalState) -> ApprovalState:
        """วิเคราะห์ความเสี่ยงของ request"""
        request = state.request
        risk_prompt = f"""
        Analyze this request for risk assessment:
        Action: {request.action}
        Parameters: {request.params}
        
        Return JSON with:
        - risk_level: low/medium/high/critical
        - reasoning: brief explanation
        - estimated_cost_usd: approximate cost
        """
        
        # ใช้ DeepSeek V4 ผ่าน HolySheep API
        response = await self.llm.chat.completions.create(
            model="deepseek-v4",
            messages=[{"role": "user", "content": risk_prompt}],
            response_format={"type": "json_object"}
        )
        
        analysis = json.loads(response.choices[0].message.content)
        
        state.request.risk_level = analysis.get("risk_level", "medium")
        state.request.estimated_cost = analysis.get("estimated_cost_usd", 0.0)
        
        return state
    
    async def gather_mcp_context(self, state: ApprovalState) -> ApprovalState:
        """รวบรวมข้อมูลจาก MCP tools สำหรับประกอบการตัดสินใจ"""
        request = state.request
        
        # กำหนด tools ที่จะใช้ตามประเภท action
        tool_mapping = {
            "send_email": ["email_client", "contact_db"],
            "update_db": ["database", "audit_log"],
            "api_call": ["api_gateway", "rate_limiter"],
        }
        
        relevant_tools = tool_mapping.get(request.action, ["generic"])
        
        mcp_results = []
        for tool_name in relevant_tools:
            if tool_name in self.mcp_tools:
                result = await self.mcp_tools[tool_name].call(
                    action=request.action,
                    params=request.params
                )
                mcp_results.append({
                    "tool": tool_name,
                    "result": result,
                    "timestamp": datetime.utcnow().isoformat()
                })
        
        state.mcp_tool_results = mcp_results
        return state
    
    def request_approval(self, state: ApprovalState) -> ApprovalState:
        """หยุดรอ human approval - LangGraph interrupt"""
        interrupt({
            "type": "human_approval",
            "request_id": state.request.request_id,
            "action": state.request.action,
            "risk_level": state.request.risk_level,
            "mcp_context": state.mcp_tool_results,
            "timeout_seconds": 3600
        })
        return state
    
    async def execute_action(self, state: ApprovalState) -> ApprovalState:
        """ดำเนินการหลังได้รับอนุมัติ"""
        request = state.request
        
        # Execute via appropriate MCP tool
        action_prompt = f"""
        Based on approval for:
        Action: {request.action}
        Parameters: {request.params}
        Approved by: {state.approver_comments}
        
        Execute the action and return result.
        """
        
        # ส่งไปยัง executor agent
        result = await self.llm.chat.completions.create(
            model="deepseek-v4",
            messages=[{"role": "user", "content": action_prompt}]
        )
        
        return state
    
    def route_approval(self, state: ApprovalState) -> str:
        """กำหนดเส้นทางตามผลการอนุมัติ"""
        if state.status == ApprovalStatus.APPROVED:
            return "approved"
        elif state.status == ApprovalStatus.REJECTED:
            return "rejected"
        else:
            return "timeout"
    
    def handle_rejection(self, state: ApprovalState) -> ApprovalState:
        """บันทึก rejection และแจ้งเตือน"""
        # Log to audit system, send notifications, etc.
        return state
    
    def handle_timeout(self, state: ApprovalState) -> ApprovalState:
        """จัดการเมื่อ timeout"""
        state.status = ApprovalStatus.TIMEOUT
        return state

MCP Tool Integration สำหรับ Context Gathering

MCP (Model Context Protocol) ช่วยให้ AI เข้าถึง external tools ได้อย่างปลอดภัย ใน approval workflow เราใช้ MCP สำหรับดึงข้อมูลที่จำเป็นก่อนขอ approval

import json
from typing import Any
from dataclasses import dataclass
from datetime import datetime, timedelta
import hashlib


HolySheep API Integration

from openai import AsyncOpenAI

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # แทนที่ด้วย API key จริง

@dataclass
class MCPConfig:
    name: str
    endpoint: str
    auth_token: str
    rate_limit: int = 100  # requests per minute
    timeout: int = 30

class MCPToolResult:
    def __init__(self, success: bool, data: Any = None, error: str = None):
        self.success = success
        self.data = data
        self.error = error
        self.timestamp = datetime.utcnow()
        self.request_id = self._generate_request_id()
    
    def _generate_request_id(self) -> str:
        timestamp = datetime.utcnow().isoformat()
        return hashlib.sha256(timestamp.encode()).hexdigest()[:16]
    
    def to_dict(self) -> dict:
        return {
            "success": self.success,
            "data": self.data,
            "error": self.error,
            "timestamp": self.timestamp.isoformat(),
            "request_id": self.request_id
        }

class MCPConnectionPool:
    """Connection pool สำหรับ MCP tools - optimize performance"""
    
    def __init__(self, max_connections: int = 10):
        self.max_connections = max_connections
        self.pools: dict[str, asyncio.Queue] = {}
        self._semaphores: dict[str, asyncio.Semaphore] = {}
    
    def register_tool(self, tool_name: str):
        if tool_name not in self.pools:
            self.pools[tool_name] = asyncio.Queue(maxsize=self.max_connections)
            self._semaphores[tool_name] = asyncio.Semaphore(self.max_connections)
    
    async def acquire(self, tool_name: str) -> asyncio.Semaphore:
        return self._semaphores.get(tool_name)
    
    async def execute_with_pooling(
        self, 
        tool_name: str, 
        func: callable, 
        *args, 
        **kwargs
    ) -> MCPToolResult:
        semaphore = await self.acquire(tool_name)
        
        async with semaphore:
            try:
                result = await asyncio.wait_for(
                    func(*args, **kwargs),
                    timeout=30
                )
                return MCPToolResult(success=True, data=result)
            except asyncio.TimeoutError:
                return MCPToolResult(
                    success=False, 
                    error=f"Tool {tool_name} timeout after 30s"
                )
            except Exception as e:
                return MCPToolResult(success=False, error=str(e))

class ProductionMCPClient:
    """MCP Client production-ready พร้อม circuit breaker และ retry"""
    
    def __init__(self):
        self.holysheep_client = AsyncOpenAI(
            base_url=HOLYSHEEP_BASE_URL,
            api_key=HOLYSHEEP_API_KEY,
            timeout=60.0,
            max_retries=3
        )
        self.connection_pool = MCPConnectionPool(max_connections=20)
        self._circuit_breakers: dict[str, dict] = {}
        
        # Register common tools
        self._setup_tools()
    
    def _setup_tools(self):
        """ตั้งค่า MCP tools"""
        tools = [
            MCPConfig("contact_db", "http://internal:8081/db", "internal_token"),
            MCPConfig("audit_log", "http://internal:8082/audit", "internal_token"),
            MCPConfig("email_client", "http://internal:8083/email", "internal_token"),
            MCPConfig("api_gateway", "http://internal:8084/gateway", "internal_token"),
        ]
        
        for tool in tools:
            self.connection_pool.register_tool(tool.name)
            self._circuit_breakers[tool.name] = {
                "failure_count": 0,
                "last_failure": None,
                "state": "closed"  # closed, open, half_open
            }
    
    async def call_deepseek_v4(
        self, 
        prompt: str, 
        system_prompt: str = "",
        temperature: float = 0.3,
        max_tokens: int = 2000
    ) -> str:
        """เรียก DeepSeek V4 ผ่าน HolySheep API"""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        
        try:
            response = await self.holysheep_client.chat.completions.create(
                model="deepseek-v4",
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"DeepSeek V4 call failed: {e}")
            raise
    
    async def call_tool(
        self, 
        tool_name: str, 
        action: str, 
        params: dict,
        use_cache: bool = True
    ) -> MCPToolResult:
        """เรียก MCP tool พร้อม circuit breaker"""
        
        # Check circuit breaker
        cb = self._circuit_breakers.get(tool_name, {})
        if cb.get("state") == "open":
            if datetime.utcnow() - cb.get("last_failure", datetime.min) < timedelta(minutes=1):
                return MCPToolResult(
                    success=False, 
                    error=f"Circuit breaker open for {tool_name}"
                )
        
        # Execute tool
        async def _execute():
            # Mock execution - แทนที่ด้วย actual HTTP call
            await asyncio.sleep(0.1)  # Simulate network latency
            return {"action": action, "params": params, "tool": tool_name}
        
        result = await self.connection_pool.execute_with_pooling(
            tool_name, _execute
        )
        
        # Update circuit breaker
        if not result.success:
            cb["failure_count"] = cb.get("failure_count", 0) + 1
            cb["last_failure"] = datetime.utcnow()
            if cb["failure_count"] >= 5:
                cb["state"] = "open"
        else:
            cb["failure_count"] = 0
            cb["state"] = "closed"
        
        return result
    
    async def gather_context_for_approval(
        self, 
        action: str, 
        params: dict,
        risk_level: str
    ) -> dict:
        """รวบรวม context ทั้งหมดสำหรับ approval"""
        
        # Determine which tools to call based on action type
        tool_mapping = {
            "send_email": ["contact_db", "email_client"],
            "update_record": ["contact_db", "audit_log"],
            "external_api_call": ["api_gateway", "audit_log"],
            "delete_data": ["contact_db", "audit_log"],
        }
        
        required_tools = tool_mapping.get(action, ["contact_db"])
        
        # เรียก tools พร้อมกัน
        tasks = [
            self.call_tool(tool, action, params)
            for tool in required_tools
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Aggregate results
        context = {
            "action": action,
            "risk_level": risk_level,
            "timestamp": datetime.utcnow().isoformat(),
            "tool_results": []
        }
        
        for result in results:
            if isinstance(result, MCPToolResult):
                context["tool_results"].append(result.to_dict())
            elif isinstance(result, Exception):
                context["tool_results"].append({
                    "success": False,
                    "error": str(result)
                })
        
        # Generate summary using DeepSeek V4
        summary_prompt = f"""
        Summarize this approval context in Thai:
        
        Action: {action}
        Risk Level: {risk_level}
        Tool Results: {json.dumps(context['tool_results'], ensure_ascii=False, indent=2)}
        
        Provide:
        1. Summary of what will happen
        2. Key data points the approver should know
        3. Potential concerns
        """
        
        context["ai_summary"] = await self.call_deepseek_v4(
            summary_prompt,
            system_prompt="คุณเป็นผู้ช่วยสรุปข้อมูลสำหรับการอนุมัติ ตอบเป็นภาษาไทย",
            temperature=0.3,
            max_tokens=500
        )
        
        return context


Benchmark utility

async def run_benchmark():
    """วัดประสิทธิภาพของ workflow"""
    client = ProductionMCPClient()
    
    test_cases = [
        ("send_email", {"to": "[email protected]", "subject": "Test"}),
        ("update_record", {"id": 12345, "field": "status", "value": "approved"}),
        ("external_api_call", {"endpoint": "/api/users", "method": "POST"}),
    ]
    
    results = []
    
    for action, params in test_cases:
        # Test MCP context gathering
        start = time.time()
        context = await client.gather_context_for_approval(
            action, params, risk_level="medium"
        )
        mcp_time = time.time() - start
        
        # Test DeepSeek V4 call
        start = time.time()
        response = await client.call_deepseek_v4(
            f"Analyze this: {json.dumps(params)}",
            temperature=0.3
        )
        llm_time = time.time() - start
        
        results.append({
            "action": action,
            "mcp_latency_ms": round(mcp_time * 1000, 2),
            "llm_latency_ms": round(llm_time * 1000, 2),
            "total_ms": round((mcp_time + llm_time) * 1000, 2)
        })
    
    return results

Concurrency Control และ Resource Management

ใน production environment การจัดการ concurrency ที่ดีเป็นสิ่งจำเป็น โดยเฉพาะเมื่อมีหลาย approval requests พร้อมกัน

import asyncio
from typing import Optional
from collections import defaultdict
import threading

class ConcurrencyController:
    """ควบคุม concurrent approval requests"""
    
    def __init__(self, max_concurrent: int = 50):
        self.max_concurrent = max_concurrent
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._active_requests: dict[str, asyncio.Task] = {}
        self._request_limits: dict[str, int] = defaultdict(lambda: 5)
        self._lock = asyncio.Lock()
    
    async def acquire(self, request_id: str, category: str = "default") -> bool:
        """ขอ permission สำหรับ request"""
        
        async with self._lock:
            # ตรวจสอบ category limit
            if self._active_requests.get(category, 0) >= self._request_limits[category]:
                return False
            
            # ตรวจสอบ global limit
            if len(self._active_requests) >= self.max_concurrent:
                return False
        
        # รอ semaphore
        await self._semaphore.acquire()
        
        async with self._lock:
            self._active_requests[request_id] = asyncio.current_task()
        
        return True
    
    async def release(self, request_id: str):
        """ปล่อย resource"""
        self._semaphore.release()
        
        async with self._lock:
            if request_id in self._active_requests:
                del self._active_requests[request_id]
    
    def set_category_limit(self, category: str, limit: int):
        """กำหนด limit เฉพาะ category"""
        self._request_limits[category] = limit

class ApprovalQueue:
    """Priority queue สำหรับ approval requests"""
    
    def __init__(self):
        self._queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
        self._processing: set[str] = set()
        self._lock = asyncio.Lock()
    
    async def enqueue(self, request: ApprovalRequest, priority: int = 5):
        """เพิ่ม request เข้าคิว"""
        await self._queue.put((priority, request.request_id, request))
    
    async def dequeue(self, timeout: float = 1.0) -> Optional[ApprovalRequest]:
        """นำ request ออกจากคิว"""
        try:
            _, request_id, request = await asyncio.wait_for(
                self._queue.get(), 
                timeout=timeout
            )
            async with self._lock:
                self._processing.add(request_id)
            return request
        except asyncio.TimeoutError:
            return None
    
    async def complete(self, request_id: str):
        """ทำเครื่องหมายว่า complete"""
        async with self._lock:
            self._processing.discard(request_id)
    
    async def requeue(self, request: ApprovalRequest, priority: int = 10):
        """ใส่กลับเข้าคิวเมื่อต้องการ retry"""
        await self.enqueue(request, priority)

class RateLimitedExecutor:
    """Executor พร้อม rate limiting"""
    
    def __init__(self, calls_per_minute: int = 60):
        self.rate_limit = calls_per_minute
        self._calls: list[datetime] = []
        self._lock = asyncio.Lock()
    
    async def execute(self, func: callable, *args, **kwargs):
        """Execute พร้อม rate limit"""
        
        async with self._lock:
            now = datetime.utcnow()
            
            # ลบ calls ที่เก่ากว่า 1 นาที
            self._calls = [
                t for t in self._calls 
                if now - t < timedelta(minutes=1)
            ]
            
            # ตรวจสอบ rate limit
            if len(self._calls) >= self.rate_limit:
                oldest = self._calls[0]
                wait_time = 60 - (now - oldest).total_seconds()
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
                    self._calls = self._calls[1:]
            
            self._calls.append(now)
        
        return await func(*args, **kwargs)


Production Orchestrator

class ApprovalOrchestrator:
    """รวมทุก component เข้าด้วยกัน"""
    
    def __init__(self):
        self.llm_client = ProductionMCPClient()
        self.graph = HumanApprovalGraph(
            self.llm_client, 
            self.llm_client.connection_pool.pools
        ).build_graph()
        self.concurrency = ConcurrencyController(max_concurrent=50)
        self.queue = ApprovalQueue()
        self.rate_limiter = RateLimitedExecutor(calls_per_minute=100)
    
    async def process_approval_request(
        self, 
        action: str, 
        params: dict,
        priority: int = 5
    ):
        """ประมวลผล approval request"""
        
        request = ApprovalRequest(
            request_id=generate_request_id(),
            action=action,
            params=params,
            risk_level="medium"
        )
        
        # เพิ่มเข้าคิว
        await self.queue.enqueue(request, priority)
        
        # รอในคิว
        while True:
            queued_request = await self.queue.dequeue(timeout=5.0)
            
            if queued_request and queued_request.request_id == request.request_id:
                break
        
        # ตรวจสอบ concurrency
        can_process = await self.concurrency.acquire(
            request.request_id, 
            category=action
        )
        
        if not can_process:
            # ใส่กลับเข้าคิวด้วย priority ต่ำลง
            await self.queue.requeue(request, priority=priority + 5)
            return {"status": "queued", "position": "unknown"}
        
        try:
            # ประมวลผลผ่าน LangGraph
            config = {"configurable": {"thread_id": request.request_id}}
            
            initial_state = ApprovalState(request=request)
            
            async for event in self.graph.astream(
                initial_state, 
                config=config
            ):
                if "interrupt" in str(event):
                    # รอ human approval
                    approval_result = await self.wait_for_approval(
                        request.request_id
                    )
                    
                    # Resume graph
                    self.graph.update_state(
                        config=config,
                        values={"status": approval_result.status}
                    )
            
            await self.queue.complete(request.request_id)
            
            return {
                "status": "completed",
                "result": event
            }
            
        finally:
            await self.concurrency.release(request.request_id)
    
    async def wait_for_approval(self, request_id: str) -> ApprovalResult:
        """รอการอนุมัติจาก human"""
        # นี่ควรเชื่อมต่อกับ frontend หรือ notification system
        # สำหรับ demo ใช้ simple implementation
        pass

def generate_request_id() -> str:
    import uuid
    return str(uuid.uuid4())[:16]

Cost Optimization Strategy

การใช้ DeepSeek V4 ผ่าน HolySheep AI ช่วยประหยัดได้มาก เมื่อเทียบกับ OpenAI หรือ Anthropic

เปรียบเทียบค่าใช้จ่าย (2026/MTok)

ModelPrice ($/MTok)DeepSeek V4 Savings
GPT-4.1$8.0094.75%
Claude Sonnet 4.5$15.0097.2%
Gemini 2.5 Flash$2.5083.2%
DeepSeek V4$0.42—

HolySheep มีอัตราแลกเปลี่ยน ¥1=$1 ทำให้ประหยัดได้ถึง 85%+ เมื่อเทียบกับการใช้งานผ่าน API อื่นๆ โดยมี latency เฉลี่ยน้อยกว่า 50ms และมีเครดิตฟรีเมื่อลงทะเบียน

# Cost tracking utility
class CostTracker:
    """ติดตามค่าใช้จ่ายแบบ real-time"""
    
    # HolySheep Pricing (DeepSeek V4)
    PRICING = {
        "deepseek-v4": {
            "input": 0.42,   # $/MTok
            "output": 0.42,
        },
        "deepseek-chat": {
            "input": 0.42,
            "output": 0.42,
        }
    }
    
    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.requests_by_action: dict[str, int] = defaultdict(int)
        self._lock = asyncio.Lock()
    
    async def track_request(
        self, 
        model: str, 
        input_tokens: int, 
        output_tokens: int,
        action: str
    ):
        async with self._lock:
            self.total_input_tokens += input_tokens
            self.total_output_tokens += output_tokens
            self.requests_by_action[action] += 1
    
    def calculate_cost(self) -> dict:
        input_cost = (self.total_input_tokens / 1_000_000) * self.PRICING["deepseek-v4"]["input"]
        output_cost = (self.total_output_tokens / 1_000_000) * self.PRICING["deepseek-v4"]["output"]
        total_cost = input_cost + output_cost
        
        return {
            "input_tokens_millions": round(self.total_input_tokens / 1_000_000, 4),
            "output_tokens_millions": round(self.total_output_tokens / 1_000_000, 4),
            "input_cost_usd": round(input_cost, 4),
            "output_cost_usd": round(output_cost, 4),
            "total_cost_usd": round(total_cost, 4),
            "requests_by_action": dict(self.requests_by_action),
            # เปรียบเทียบกับ OpenAI
            "openai_equivalent_cost": round(total_cost * (8.0 / 0.42), 2),
            "savings_percentage": round((1 - 0.42/8.0) * 100, 2)
        }
    
    def estimate_batch_cost(
        self, 
        batch_size: int, 
        avg_input_tokens: int = 1000,
        avg_output_tokens: int = 500
    ) -> dict:
        """ประมาณค่าใช้จ่ายสำหรับ batch"""
        
        total_input = batch_size * avg_input_tokens
        total_output = batch_size * avg_output_tokens
        
        return {
            "batch_size": batch_size,
            "estimated_input_cost_usd": round(
                (total_input / 1_000_000) * 0.42, 4
            ),
            "estimated_output_cost_usd": round(
                (total_output / 1_000_000) * 0.42, 4
            ),
            "estimated_total_usd": round(
                ((total_input + total_output) / 1_000_000) * 0.42, 4
            )
        }


Batch processing with cost optimization

async def process_batch_with_cost_control(
    orchestrator: ApprovalOrchestrator,
    requests: list[tuple[str, dict]],
    max_concurrent: int = 10,
    max_cost_per_batch: float = 10.0
):
    """ประมวลผล batch พร้อม cost control"""
    
    tracker = CostTracker()
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_one(action: str, params: dict):
        async with semaphore:
            # ตรวจสอบ cost limit
            current_cost = tracker.calculate_cost()["total_cost_usd"]
            if current_cost >= max_cost_per_batch:
                return {"status": "skipped", "reason": "cost_limit_reached"}
            
            try:
                result = await orchestrator.process

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง