Spaces:
Sleeping
Sleeping
File size: 5,621 Bytes
79899c0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
"""
Web search service for retrieving and processing web content.
This module provides functionality to search the web using Serper API
and extract content from web pages using crawl4ai.
"""
import asyncio
import os
from typing import List, Optional
from bio_requests.rag_request import RagRequest
from dto.bio_document import BaseBioDocument, create_bio_document
from search_service.base_search import BaseSearchService
from service.web_search import SerperClient, scrape_urls, url_to_fit_contents
from utils.bio_logger import bio_logger as logger
class WebSearchService(BaseSearchService):
"""
Web search service that retrieves content from web pages.
This service uses Serper API for web search and crawl4ai for content extraction.
"""
def __init__(self):
"""Initialize the web search service."""
self.data_source = "web"
self._serper_client: Optional[SerperClient] = None
self._max_results = 5
self._content_length_limit = 40000 # ~10k tokens
@property
def serper_client(self) -> SerperClient:
"""Lazy initialization of SerperClient."""
if self._serper_client is None:
# 从环境变量获取API密钥
api_key = os.getenv("SERPER_API_KEY")
if not api_key:
logger.warning("SERPER_API_KEY environment variable not set, using default key")
self._serper_client = SerperClient(api_key=api_key)
return self._serper_client
async def search(self, rag_request: RagRequest) -> List[BaseBioDocument]:
"""
Perform web search and extract content from search results.
Args:
rag_request: The RAG request containing the search query
Returns:
List of BaseBioDocument objects with extracted web content
"""
try:
query = rag_request.query
logger.info(f"Starting web search for query: {query}")
# Search for URLs using Serper
url_results = await self.search_serper(query, rag_request.top_k)
if not url_results:
logger.info(f"No search results found for query: {query}")
return []
# Extract content from URLs
search_results = await self.enrich_url_results_with_contents(url_results)
logger.info(f"Web search completed. Found {len(search_results)} documents")
return search_results
except Exception as e:
logger.error(f"Error during web search: {str(e)}", exc_info=e)
return []
async def enrich_url_results_with_contents(
self, results: List
) -> List[BaseBioDocument]:
"""
Extract content from URLs and create BaseBioDocument objects.
Args:
results: List of search results with URLs
Returns:
List of BaseBioDocument objects with extracted content
"""
try:
# Create tasks for concurrent content extraction
tasks = [self._extract_content_from_url(res) for res in results]
contents = await asyncio.gather(*tasks, return_exceptions=True)
enriched_results = []
for res, content in zip(results, contents):
# Handle exceptions from content extraction
if isinstance(content, Exception):
logger.error(f"Failed to extract content from {res.url}: {content}")
continue
bio_doc = create_bio_document(
title=res.title,
url=res.url,
text=str(content)[: self._content_length_limit],
source=self.data_source,
)
enriched_results.append(bio_doc)
return enriched_results
except Exception as e:
logger.error(f"Error enriching URL results: {str(e)}", exc_info=e)
return []
async def _extract_content_from_url(self, res) -> str:
"""
Extract content from a single URL with error handling.
Args:
res: Search result object containing URL information
Returns:
Extracted content as string
"""
try:
return await url_to_fit_contents(res)
except Exception as e:
logger.error(f"Error extracting content from {res.url}: {str(e)}")
return f"Error extracting content: {str(e)}"
async def search_serper(
self, query: str, max_results: Optional[int] = None
) -> List:
"""
Perform web search using Serper API.
Args:
query: Search query string
max_results: Maximum number of results to return
Returns:
List of search results with URLs
"""
try:
max_results = max_results or self._max_results
logger.info(f"Searching Serper for: {query} (max_results: {max_results})")
search_results = await self.serper_client.search(
query, filter_for_relevance=True, max_results=max_results
)
if not search_results:
logger.info(f"No search results from Serper for query: {query}")
return []
# Scrape content from URLs
results = await scrape_urls(search_results)
logger.info(f"Serper search completed. Found {len(results)} results")
return results
except Exception as e:
logger.error(f"Error in Serper search: {str(e)}", exc_info=e)
return []
|