feat: Apply WEB_SEARCH_CONCURRENT_REQUESTS to all search engines using semaphore (#20070)
* sequential * zero default * fix
This commit is contained in:
@@ -2994,7 +2994,7 @@ WEB_SEARCH_DOMAIN_FILTER_LIST = PersistentConfig(
|
||||
WEB_SEARCH_CONCURRENT_REQUESTS = PersistentConfig(
|
||||
"WEB_SEARCH_CONCURRENT_REQUESTS",
|
||||
"rag.web.search.concurrent_requests",
|
||||
int(os.getenv("WEB_SEARCH_CONCURRENT_REQUESTS", "10")),
|
||||
int(os.getenv("WEB_SEARCH_CONCURRENT_REQUESTS", "0")),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -2103,16 +2103,38 @@ async def process_web_search(
|
||||
f"trying to web search with {request.app.state.config.WEB_SEARCH_ENGINE, form_data.queries}"
|
||||
)
|
||||
|
||||
search_tasks = [
|
||||
run_in_threadpool(
|
||||
search_web,
|
||||
request,
|
||||
request.app.state.config.WEB_SEARCH_ENGINE,
|
||||
query,
|
||||
user,
|
||||
)
|
||||
for query in form_data.queries
|
||||
]
|
||||
# Use semaphore to limit concurrent requests based on WEB_SEARCH_CONCURRENT_REQUESTS
|
||||
# 0 or None = unlimited (previous behavior), positive number = limited concurrency
|
||||
# Set to 1 for sequential execution (rate-limited APIs like Brave free tier)
|
||||
concurrent_limit = request.app.state.config.WEB_SEARCH_CONCURRENT_REQUESTS
|
||||
|
||||
if concurrent_limit:
|
||||
# Limited concurrency with semaphore
|
||||
semaphore = asyncio.Semaphore(concurrent_limit)
|
||||
|
||||
async def search_with_limit(query):
|
||||
async with semaphore:
|
||||
return await run_in_threadpool(
|
||||
search_web,
|
||||
request,
|
||||
request.app.state.config.WEB_SEARCH_ENGINE,
|
||||
query,
|
||||
user,
|
||||
)
|
||||
|
||||
search_tasks = [search_with_limit(query) for query in form_data.queries]
|
||||
else:
|
||||
# Unlimited parallel execution (previous behavior)
|
||||
search_tasks = [
|
||||
run_in_threadpool(
|
||||
search_web,
|
||||
request,
|
||||
request.app.state.config.WEB_SEARCH_ENGINE,
|
||||
query,
|
||||
user,
|
||||
)
|
||||
for query in form_data.queries
|
||||
]
|
||||
|
||||
search_results = await asyncio.gather(*search_tasks)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user