หมดเวลาและตัวตัดวงจร

จำกัดการเรียกใช้บริการภายนอกทุกครั้ง

HTTP Timeouts

import requests
response = requests.get(url, timeout=(3, 10))   # 3s connect, 10s read

LLM Timeouts

from openai import OpenAI
client = OpenAI(timeout=30.0)

# Or per call:
response = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=messages,
    timeout=30.0
)

เลือกการกำหนดหมดเวลาที่เหมาะสม

การกำหนดหมดเวลารวมของเอเจนต์

import asyncio

async def run_agent_with_deadline(query, deadline_s=60):
    try:
        return await asyncio.wait_for(run_agent(query), timeout=deadline_s)
    except asyncio.TimeoutError:
        return 'Sorry, this is taking too long. Please try again.'

ตัวตัดวงจรคืออะไร

Implementing with pybreaker

# pip install pybreaker
import pybreaker

breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

@breaker
def fragile_call():
    return requests.get(url, timeout=5).json()

หลังจากล้มเหลว 5 ครั้ง

ส่งคืนทางเลือกสำรองที่เหมาะสม

try:
    result = fragile_call()
except pybreaker.CircuitBreakerError:
    result = {'error': 'service-unavailable', 'fallback': 'cached or stale data'}

ตัวตัดวงจรแยกตามเครื่องมือ

การแบ่งส่วนทรัพยากร

from asyncio import Semaphore
search_sem = Semaphore(10)

async def search(query):
    async with search_sem:
        return await tavily_search(query)

การตรวจสอบสถานะ

async def health_check():
    for tool in [tavily, openai, vector_db]:
        try:
            await tool.ping()
        except Exception as e:
            alert(f'{tool.name} unhealthy: {e}')

แสดงสถานะให้ผู้ใช้ทราบ

if tavily_breaker.current_state == 'open':
    return 'Web search is temporarily unavailable.'

เหตุใดจึงต้องใช้ตัวตัดวงจร

สรุป