如何构建代理搜索重试链

import requests, os, time
from enum import Enum

API_KEY = os.environ['SCAVIO_API_KEY']

class SearchError(Enum):
    SUCCESS = 'success'
    EMPTY = 'empty_results'
    TIMEOUT = 'timeout'
    RATE_LIMIT = 'rate_limit'
    AUTH_ERROR = 'auth_error'
    SERVER_ERROR = 'server_error'
    UNKNOWN = 'unknown'

def search_with_status(query: str, timeout: int = 15) -> tuple:
    """Returns (results, error_type)."""
    try:
        resp = requests.post('https://api.scavio.dev/api/v1/search',
            headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
            json={'query': query, 'country_code': 'us'},
            timeout=timeout)
        if resp.status_code == 429:
            return [], SearchError.RATE_LIMIT
        if resp.status_code == 401:
            return [], SearchError.AUTH_ERROR
        if resp.status_code >= 500:
            return [], SearchError.SERVER_ERROR
        results = resp.json().get('organic_results', [])
        if not results:
            return [], SearchError.EMPTY
        return results, SearchError.SUCCESS
    except requests.exceptions.Timeout:
        return [], SearchError.TIMEOUT
    except Exception:
        return [], SearchError.UNKNOWN

def simplify_query(query: str) -> list:
    """Generate progressively simpler versions of a query."""
    words = query.split()
    variants = [query]  # original first
    # Remove quotes and special chars
    cleaned = query.replace('"', '').replace("'", '')
    if cleaned != query:
        variants.append(cleaned)
    # Remove year qualifiers
    import re
    no_year = re.sub(r'\b202[0-9]\b', '', query).strip()
    if no_year != query:
        variants.append(no_year)
    # First N words (progressively shorter)
    if len(words) > 5:
        variants.append(' '.join(words[:5]))
    if len(words) > 3:
        variants.append(' '.join(words[:3]))
    # Remove filler words
    filler = {'the', 'a', 'an', 'is', 'are', 'was', 'were', 'for', 'and', 'or', 'but'}
    no_filler = ' '.join(w for w in words if w.lower() not in filler)
    if no_filler != query and len(no_filler) > 5:
        variants.append(no_filler)
    return list(dict.fromkeys(variants))  # dedupe preserving order

# Test:
for v in simplify_query('"best enterprise CRM software" for startups 2026'):
    print(f'  {v}')

import logging

log = logging.getLogger('search_retry')

def search_retry_chain(query: str, max_retries: int = 3) -> dict:
    """Retry chain: try original, then simplified queries, with backoff."""
    queries = simplify_query(query)
    all_attempts = []
    for attempt, q in enumerate(queries[:max_retries]):
        results, error = search_with_status(q)
        all_attempts.append({'query': q, 'error': error.value, 'result_count': len(results)})
        if error == SearchError.SUCCESS:
            return {
                'results': results,
                'query_used': q,
                'attempts': len(all_attempts),
                'all_attempts': all_attempts
            }
        if error == SearchError.AUTH_ERROR:
            log.error('Authentication failed. Check your API key.')
            break
        if error in (SearchError.RATE_LIMIT, SearchError.SERVER_ERROR, SearchError.TIMEOUT):
            wait = 2 ** attempt
            log.warning(f'Transient error ({error.value}), retrying in {wait}s...')
            time.sleep(wait)
            # Retry same query for transient errors
            results, error = search_with_status(q)
            if error == SearchError.SUCCESS:
                all_attempts.append({'query': q, 'error': 'retry_success', 'result_count': len(results)})
                return {'results': results, 'query_used': q,
                        'attempts': len(all_attempts), 'all_attempts': all_attempts}
        # For EMPTY, move to next simplified query
        log.info(f'Attempt {attempt + 1}: "{q}" -> {error.value}')
    return {'results': [], 'query_used': query,
            'attempts': len(all_attempts), 'all_attempts': all_attempts}

import hashlib

_cache = {}
CACHE_TTL = 3600  # 1 hour

def cached_search(query: str) -> dict:
    key = hashlib.md5(query.lower().strip().encode()).hexdigest()
    now = time.time()
    # Check cache first
    if key in _cache:
        cached, timestamp = _cache[key]
        if now - timestamp < CACHE_TTL:
            return {**cached, 'from_cache': True, 'cache_age_s': int(now - timestamp)}
    # Run retry chain
    result = search_retry_chain(query)
    # Cache successful results
    if result['results']:
        _cache[key] = (result, now)
    elif key in _cache:
        # All retries failed, return stale cache with warning
        cached, timestamp = _cache[key]
        return {**cached, 'from_cache': True,
                'cache_age_s': int(now - timestamp),
                'warning': 'Serving stale cached results (all retries failed)'}
    return {**result, 'from_cache': False}

# Test:
result = cached_search('best crm software 2026')
print(f'Results: {len(result["results"])}, cache: {result["from_cache"]}, '
      f'attempts: {result["attempts"]}')

def agent_search_tool(query: str) -> str:
    """Production-ready search tool for agents.
    Handles retries, query simplification, and caching automatically."""
    result = cached_search(query)
    if not result['results']:
        return f'Search returned no results after {result["attempts"]} attempts. Please rephrase your query.'
    formatted = []
    for r in result['results'][:5]:
        formatted.append(f'Title: {r["title"]}\nURL: {r["link"]}\nSnippet: {r.get("snippet", "")}')
    header = f'Found {len(result["results"])} results'
    if result.get('from_cache'):
        header += f' (cached, {result["cache_age_s"]}s old)'
    if result['query_used'] != query:
        header += f' (simplified query: "{result["query_used"]}")'     
    return f'{header}\n\n' + '\n\n'.join(formatted)

# Use in any agent framework:
# @tool
# def web_search(query: str) -> str:
#     return agent_search_tool(query)

print(agent_search_tool('best enterprise CRM software pricing comparison 2026'))