如何构建多提供商搜索以提高 RAG 可靠性

import os, requests, time, json

class UnifiedSearch:
    def __init__(self):
        self.scavio_key = os.environ.get('SCAVIO_API_KEY', '')
        self.exa_key = os.environ.get('EXA_API_KEY', '')
        self.brave_key = os.environ.get('BRAVE_API_KEY', '')
        self.stats = {'scavio': 0, 'exa': 0, 'brave': 0, 'failures': 0}
    
    def _scavio(self, query, n=5):
        resp = requests.post('https://api.scavio.dev/api/v1/search',
            headers={'x-api-key': self.scavio_key, 'Content-Type': 'application/json'},
            json={'query': query, 'country_code': 'us'}, timeout=10)
        resp.raise_for_status()
        return [{'title': r.get('title', ''), 'url': r.get('link', ''),
                 'text': r.get('snippet', ''), 'source': 'scavio'}
                for r in resp.json().get('organic_results', [])[:n]]
    
    def _exa(self, query, n=5):
        if not self.exa_key: raise Exception('No Exa key')
        resp = requests.post('https://api.exa.ai/search',
            headers={'x-api-key': self.exa_key, 'Content-Type': 'application/json'},
            json={'query': query, 'numResults': n}, timeout=10)
        resp.raise_for_status()
        return [{'title': r.get('title', ''), 'url': r.get('url', ''),
                 'text': r.get('text', '')[:200], 'source': 'exa'}
                for r in resp.json().get('results', [])[:n]]
    
    def _brave(self, query, n=5):
        if not self.brave_key: raise Exception('No Brave key')
        resp = requests.get('https://api.search.brave.com/res/v1/web/search',
            headers={'X-Subscription-Token': self.brave_key},
            params={'q': query}, timeout=10)
        resp.raise_for_status()
        return [{'title': r.get('title', ''), 'url': r.get('url', ''),
                 'text': r.get('description', ''), 'source': 'brave'}
                for r in resp.json().get('web', {}).get('results', [])[:n]]
    
    def search(self, query, n=5):
        for name, fn in [('scavio', self._scavio), ('exa', self._exa), ('brave', self._brave)]:
            try:
                results = fn(query, n)
                self.stats[name] += 1
                return {'provider': name, 'results': results, 'count': len(results)}
            except Exception as e:
                self.stats['failures'] += 1
        return {'provider': 'none', 'results': [], 'count': 0}

usearch = UnifiedSearch()
result = usearch.search('rag pipeline best practices 2026')
print(f'Provider: {result["provider"]} | Results: {result["count"]}')
for r in result['results'][:3]:
    print(f'  [{r["source"]:7}] {r["title"][:50]}')

def format_for_rag(search_result, max_context_chars=3000):
    """Format search results as RAG context with source tracking."""
    if not search_result.get('results'):
        return {'context': '', 'sources': [], 'char_count': 0}
    context_parts = []
    sources = []
    char_count = 0
    for i, r in enumerate(search_result['results']):
        source_ref = f'[{i+1}]'
        text = r.get('text', '').strip()
        if not text:
            continue
        entry = f'{source_ref} {text}'
        if char_count + len(entry) > max_context_chars:
            break
        context_parts.append(entry)
        sources.append({'ref': source_ref, 'title': r['title'][:60], 'url': r['url']})
        char_count += len(entry)
    context = '\n\n'.join(context_parts)
    return {
        'context': context,
        'sources': sources,
        'char_count': char_count,
        'provider': search_result.get('provider', 'unknown'),
    }

# Build RAG context
result = usearch.search('how to implement vector search python 2026')
rag = format_for_rag(result, max_context_chars=2000)
print(f'\n=== RAG Context ===')
print(f'  Provider: {rag["provider"]}')
print(f'  Context length: {rag["char_count"]} chars')
print(f'  Sources: {len(rag["sources"])}')
for s in rag['sources']:
    print(f'    {s["ref"]} {s["title"][:50]}')
    print(f'       {s["url"][:60]}')
print(f'\n  Context preview: {rag["context"][:150]}...')

def rag_retrieve(question, max_sources=5):
    """RAG retrieval step using multi-provider search."""
    # Primary search
    result = usearch.search(question, n=max_sources)
    rag_context = format_for_rag(result)
    # If primary gives weak results, try refined query
    if len(rag_context['sources']) < 2:
        refined = usearch.search(f'{question} tutorial guide', n=max_sources)
        refined_context = format_for_rag(refined)
        if len(refined_context['sources']) > len(rag_context['sources']):
            rag_context = refined_context
    return rag_context

def rag_pipeline(question):
    """Full RAG pipeline: retrieve, format, generate."""
    print(f'\n  Question: {question}')
    # Step 1: Retrieve
    context = rag_retrieve(question)
    print(f'  Retrieved: {len(context["sources"])} sources via {context["provider"]}')
    print(f'  Context: {context["char_count"]} chars')
    # Step 2: Would pass to LLM here
    print(f'  Sources for citation:')
    for s in context['sources']:
        print(f'    {s["ref"]} {s["title"][:45]}')
    return context

print('=== Multi-Provider RAG Pipeline ===')
for q in ['best vector database 2026', 'how to optimize rag pipeline']:
    rag_pipeline(q)

print(f'\n  Provider stats: {json.dumps(usearch.stats)}')
print(f'  Primary: Scavio $0.005/query')
print(f'  Fallback: Exa $0.007/query, Brave ~$0.005/query')
print(f'  Uptime target: 99.9% with multi-provider')