Khôi phục lỗi cho Agent
Cách LLM agent nên xử lý và khôi phục từ lỗi — thử lại, lưu, học.
Khôi phục lỗi cho Agent
Lỗi xảy ra. Mạng thất bại, API trả về 500, Mind Key hết hạn. Hướng dẫn này cho thấy cách LLM agent nên xử lý lỗi một cách duyên dáng và học từ chúng.
Nguyên tắc xử lý lỗi
- Thử lại với backoff — lỗi tạm thời thường được giải quyết
- Lưu lỗi — học từ mẫu
- Giảm cấp duyên dáng — không làm sập toàn bộ phiên
- Thông báo con người — cho lỗi bạn không thể giải quyết
Xử lý lỗi HTTP
Thử lại với exponential backoff
import time
import requests
def call_with_retry(url, max_retries=3, backoff_base=2):
"""Call URL with exponential backoff."""
for attempt in range(max_retries):
try:
r = requests.get(url, headers={"Authorization": f"Bearer {KEY}"})
# Success
if r.status_code < 400:
return r
# Rate limited — wait and retry
if r.status_code == 429:
wait = int(r.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {wait}s...")
time.sleep(wait)
continue
# Server error — retry with backoff
if r.status_code >= 500:
wait = backoff_base ** attempt
print(f"Server error {r.status_code}. Retrying in {wait}s...")
time.sleep(wait)
continue
# Client error — don't retry
if 400 <= r.status_code < 500:
raise ClientError(f"{r.status_code}: {r.text}")
except requests.RequestException as e:
# Network error — retry
wait = backoff_base ** attempt
print(f"Network error: {e}. Retrying in {wait}s...")
time.sleep(wait)
raise MaxRetriesError(f"Failed after {max_retries} retries")Xử lý lỗi xác thực
def safe_call(url):
"""Call with auth error handling."""
try:
return call_with_retry(url)
except ClientError as e:
if "401" in str(e):
# Mind Key invalid — critical, can't recover
store_error("auth_invalid", str(e), "Check Mind Key")
notify_human("My Mind Key is invalid. Please update.")
raise AuthError("Cannot continue without valid Mind Key")
elif "403" in str(e):
store_error("forbidden", str(e), "Wrong token type?")
raise ForbiddenError(str(e))
elif "404" in str(e):
# Path doesn't exist — don't retry
store_error("not_found", str(e), "Check endpoint path")
raise NotFoundError(str(e))
else:
raiseLưu lỗi dưới dạng bộ nhớ
Khi lỗi xảy ra, lưu chúng để các phiên tương lai có thể học:
def store_error(error_type, error_message, recovery_hint=""):
"""Store an error as a memory for future reference."""
requests.post(f"{URL}/memory",
headers={"Authorization": f"Bearer {KEY}",
"Content-Type": "application/json"},
json={
"category": "mistake",
"key": f"error_{error_type}_{int(time.time())}",
"content": f"Error: {error_type}\nMessage: {error_message}\nRecovery: {recovery_hint}",
"tags": ["error", error_type],
"priority": "high"
})
# Example
try:
deploy()
except DeployError as e:
store_error("deploy_failed", str(e),
"Check CI logs, verify Docker image exists")
raiseKịch bản lỗi phổ biến
Kịch bản 1: Mind Key không hợp lệ
# Detection: 401 on every call
# Recovery: Cannot recover — need human intervention
def handle_invalid_mind_key():
store_error("mind_key_invalid",
"All API calls returning 401",
"Mind Key may be revoked. Need new key.")
# Notify human via chat (if possible)
try:
reply("⚠️ My Mind Key is invalid. I cannot access memories. "
"Please check and update SYNAPSE_MIND_KEY.")
except:
pass # Chat may also fail with bad key
# Exit gracefully
raise CriticalError("Cannot continue without valid Mind Key")Kịch bản 2: Lỗi mạng
# Detection: ConnectionError, Timeout
# Recovery: Retry with backoff, then degrade
def handle_network_error(url, retry=3):
for attempt in range(retry):
try:
return requests.get(url, timeout=10)
except (requests.ConnectionError, requests.Timeout) as e:
wait = 2 ** attempt
print(f"Network error, retrying in {wait}s: {e}")
time.sleep(wait)
# All retries failed — degrade
store_error("network_failure",
f"Cannot reach {url}",
"Check internet connection, Synapse may be down")
# Work offline if possible
return work_offline()Kịch bản 3: Bị giới hạn tốc độ
# Detection: 429 with Retry-After header
# Recovery: Wait and retry, or switch to header auth
def handle_rate_limit(response):
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {retry_after}s...")
time.sleep(retry_after)
# If this keeps happening, suggest switching to header auth
if has_query_param_auth(url):
store_error("rate_limited",
"Frequent 429s with ?key= auth",
"Switch to Authorization: Bearer header (no rate limit)")Kịch bản 4: Lỗi server (5xx)
# Detection: 500, 502, 503
# Recovery: Retry with backoff, check /health
def handle_server_error(url):
# Check if server is up
health = requests.get(f"{URL}/health")
if health.status_code != 200:
store_error("server_down",
"Synapse health check failing",
"Wait for server recovery")
raise ServerDownError()
# Retry with backoff
return call_with_retry(url, max_retries=5)Kịch bản 5: Lệnh gọi công cụ thất bại
# Detection: Tool returns error content
# Recovery: Try alternative approach, store failure
def call_tool_safely(tool_name, args, alternatives=None):
try:
result = call_tool(tool_name, args)
if result.get("isError"):
raise ToolError(result["content"])
return result
except ToolError as e:
store_error(f"tool_{tool_name}_failed",
f"Args: {args}\nError: {e}",
f"Try: {alternatives or 'no alternatives'}")
# Try alternatives
if alternatives:
for alt in alternatives:
try:
return call_tool(alt, args)
except:
continue
raiseMẫu: Circuit Breaker
Cho thất bại lặp lại, tạm thời ngừng thử:
class CircuitBreaker:
def __init__(self, threshold=5, reset_time=300):
self.failures = 0
self.threshold = threshold
self.reset_time = reset_time
self.last_failure = 0
def call(self, fn, *args, **kwargs):
if self.failures >= self.threshold:
if time.time() - self.last_failure < self.reset_time:
raise CircuitOpenError("Circuit breaker open")
else:
# Reset
self.failures = 0
try:
result = fn(*args, **kwargs)
self.failures = 0 # Reset on success
return result
except:
self.failures += 1
self.last_failure = time.time()
raise
# Usage
breaker = CircuitBreaker(threshold=5)
try:
result = breaker.call(api_call, url)
except CircuitOpenError:
print("Too many failures, waiting before retry")