Function process_single_sequence

scripts/run_naive_rag_report.py:163–268

Process a single question through RAG pipeline

(
    question: str,
    client: AsyncOpenAI,
    semaphore: asyncio.Semaphore,
    args: argparse.Namespace,
    search_cache: Dict,
    url_cache: Dict,
)

async def process_single_sequence(
    question: str,
    client: AsyncOpenAI,
    semaphore: asyncio.Semaphore,
    args: argparse.Namespace,
    search_cache: Dict,
    url_cache: Dict,
) -> Dict:
    """Process a single question through RAG pipeline"""
    
    results = {}
    # Search for relevant documents
    try:
        if question in search_cache:
            results = search_cache[question]
        else:
            if args.search_engine == "bing":
                if not args.bing_subscription_key:
                    print(f"Error: Bing search engine is selected, but BING_SUBSCRIPTION_KEY is not provided for question: {question}")
                    results = {}
                else:
                    results = await bing_web_search_async(question, args.bing_subscription_key, args.bing_endpoint)
            elif args.search_engine == "serper":
                if not args.serper_api_key:
                    print(f"Error: Serper search engine is selected, but SERPER_API_KEY is not provided for question: {question}")
                    results = {}
                else:
                    results = await google_serper_search_async(question, args.serper_api_key)
            else:
                print(f"Error: Unknown search engine: {args.search_engine}")
                results = {}
            
            if results: # Only cache if results are not empty (i.e., search was successful)
                search_cache[question] = results

    except Exception as e:
        print(f"Error during search for '{question}' using {args.search_engine}: {e}")
        results = {}

    # Extract and process relevant documents
    relevant_info = []
    if results:
        if args.search_engine == "bing":
            relevant_info = extract_relevant_info(results)[:args.top_k]
        elif args.search_engine == "serper":
            relevant_info = extract_relevant_info_serper(results)[:args.top_k]
    
    # Fetch page content for each result
    documents = []
    for idx, doc_info in enumerate(relevant_info):
        url = doc_info['url']
        if url not in url_cache:
            try:
                contents = await fetch_page_content_async(
                    [url], 
                    use_jina=args.use_jina, 
                    jina_api_key=args.jina_api_key, 
                    keep_links=args.keep_links