2024年您產品必備的10大AI API推薦
你可能也喜歡這些API文章!
（2）根據 MindSearch 里的 SearcherAgent 提示詞中的 few shot 例子，可以看出其實現的 搜索 API Action需要有個可以進一步檢索網站內容的函數（網頁內容的抓取），也就是 select 函數 。這是因為多數 搜索 API 返回的內容都是網站內容里的 片段，并不會包含太多有用的信息。
### select
為了找到王者榮耀s36賽季最強射手，我需要尋找提及王者榮耀s36射手的網頁。初步瀏覽網頁后，
發現網頁0提到王者榮耀s36賽季的信息，但沒有具體提及射手的相關信息。網頁3提到“s36最強射手出現？”，
有可能包含最強射手信息。網頁13提到“四大T0英雄崛起，射手榮耀降臨”，可能包含最強射手的信息。
因此，我選擇了網頁3和網頁13進行進一步閱讀。
<|action_start|><|plugin|>{{"name": "FastWebBrowser.select",
"parameters": {{"index": [3, 13]}}}}<|action_end|>
"""（3）搜索 API Action 中的 search 函數 返回的最終內容最好符合以下所期望的格式，否則在 MindSearch 內部代碼解析內容時有可能會報錯，MindSearch 里的 SeacherAgent 會根據此結果來調用 select 函數：
[{'type': 'text', 'content': '{"0": {"url": "https:", "summ": "...", "title": "..."},]為了在 MindSearch 中支持新的搜索 API，我們可以采用兩種方法：一是在 lagent/actions/ 文件夾下新建文件從零開始實現，二是直接在現有的 bing_browser.py 中進行功能擴展。
本文將著重介紹第二種方法，即在現有代碼基礎上引入新的搜索 API，這樣不僅避免了重復開發，還能確保代碼的一致性和可維護性。鑒于 bing_browser.py 已內置了 search 函數和 select 函數等功能，并已妥善處理了前文提到的關鍵注意事項，我們將聚焦于通過這種方法來擴展現有功能。接下來，我們將詳細解析 bing_browser.py 文件的內容。
SearcherAgent 調用 BingBrowser 類的代碼流程如下：def search() -> self.searcher.search() -> self.searcher._call_serper_api() -> self.searcher._parse_response() -> self.searcher._filter_results() -> def select() -> self.fetcher.fetch()
在上述流程中，標記為黃色以及藍色的函數是 SearcherAgent 觸發 search() 函數時會執行到的關鍵函數。其中標記為藍色的函數則是我們在支持新的 搜索 API 時，需要在新的 searcher 類中實現的函數。而標記為綠色的函數，則是在 SearcherAgent 觸發 select() 函數時會執行到的函數。
在 bing_browser.py 文件中，定義了三個關鍵的類，它們分別是：
此類是被設計為 SearcherAgent 中的 Action 組件，負責處理搜索相關的核心邏輯。此類含有兩個重要的函數，分別是 search() 和 select()，分別對應前置內容中的 第一點 和 第二點。
def search() 函數
當接收到 SearcherAgent 生成的多個 query（以列表形式表示）后，單獨給每個在 queries 列表中的 query 開啟一個線程，并且調用對應的 searcher.serach() 函數來執行相應的 搜索 API 調用。
@tool_api
def search(self, query: Union[str, List[str]]) -> dict:
"""BING search API
Args:
query (List[str]): list of search query strings
"""
queries = query if isinstance(query, list) else [query]
search_results = {}
with ThreadPoolExecutor() as executor:
future_to_query = {
executor.submit(self.searcher.search, q): q
for q in queries
}def select() 函數
在 SearcherAgent 接收到 search() 函數返回的搜索 API 結果后，它會判斷哪些網站的內容需要進一步深入查詢，并調用 select() 函數來處理這些需求。select()函數會為每個需要深入查詢的網頁（通過索引值標識）單獨開啟一個線程，并利用 ContentFetcher 類（即 fetcher）來抓取這些網站的詳細內容。值得注意的是，所有的 searcher 都共享同一個 ContentFetcher 實例。
@tool_api
def select(self, select_ids: List[int]) -> dict:
"""get the detailed content on the selected pages.
Args:
select_ids (List[int]): list of index to select. Max number of index to be selected is no more than 4.
"""
if not self.search_results:
raise ValueError('No search results to select from.')
new_search_results = {}
with ThreadPoolExecutor() as executor:
future_to_id = {
executor.submit(self.fetcher.fetch,
self.search_results[select_id]['url']):
select_id
for select_id in select_ids if select_id in self.search_results
}ContentFetcher 類中的 fetch 函數負責使用 Python 的 requests 模塊從網站抓取內容，并通過 BeautifulSoup 庫將獲取的 HTML 文檔結構化。
注意，需要 cookie 授權的網站會訪問失敗。
class ContentFetcher:
@cached(cache=TTLCache(maxsize=100, ttl=600))
def fetch(self, url: str) -> Tuple[bool, str]:
try:
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
html = response.content
except requests.RequestException as e:
return False, str(e)
text = BeautifulSoup(html, 'html.parser').get_text()
cleaned_text = re.sub(r'\n+', '\n', text)
return True, cleaned_text這是實現新的 Searcher 類時需要繼承的一個基類，其主要目的是調用內部的 _filter_results 函數。該函數的作用是確保從 searcher 返回的內容不包含黑名單中的 URL ，并且確保返回的內容數量不超過 topk。同時對內容進行統一格式化，這對應于前置內容中的 第三點 要求。
class BaseSearch:
def _filter_results(self, results: List[tuple]) -> dict:
filtered_results = {}
count = 0
for url, snippet, title in results:
if all(domain not in url
for domain in self.black_list) and not url.endswith('.pdf'):
filtered_results[count] = {
'url': url,
'summ': json.dumps(snippet, ensure_ascii=False)[1:-1],
'title': title
}
count += 1
if count >= self.topk:
break
return filtered_results綜上所述，bing_broswer.py 里已經提供了核心的相關類以及函數，現在只需要實現一個新的 Searcher 類（對應 bing_broswer.py 里的 def search() 函數中的 self.searcher）
當前，SearcherAgent 中的提示詞設計緊密圍繞 BingBrowser 類展開。因此，為了最便捷地支持新的搜索 API，我們只需在現有基礎上新增一個 Searcher 類即可實現（方法二），這樣的改動既直接又高效。否則，如果基于方法一實現且未遵循前置內容中的注意事項，則可能需要對 SearcherAgent 中的提示詞進行調整。在開始實現新的 searcher 類之前，需要在 conda 環境中對 lagent 進行源碼安裝，以便 lagent 文件夾中的代碼改動能夠即時生效。
以 GoogleSearch Seacher 為例（Google Serper API），需要實現的函數有：def search()，def _call_serper_api() 和 def _parse_response()，其中 def search() 是 Searcher 的主函數。
首先定義一個 GoogleSearch 類，繼承 BaseSearch 類，并且將參數賦值為對象的屬性（參數由 BingBrowser 類傳入）。black_list 參數由 BaseSearch 類中的 _filter_results 函數調用。api_key ，search_type，kwargs 參數都是和 Google Serper API 相關的參數，使用于對 搜索 API 發送請求。topk 參數在向 搜索 API 發送請求時使用，并在 _filter_results 函數中再次被調用，以進一步確保最終返回的內容數量不超過 topk 。
class GoogleSearch(BaseSearch):
def __init__(self,
api_key: str,
topk: int = 3,
black_list: List[str] = [
'enoN',
'youtube.com',
'bilibili.com',
'researchgate.net',
],
**kwargs):
self.api_key = api_key
self.proxy = kwargs.get('proxy')
self.search_type = kwargs.get('search_type', 'search')
self.kwargs = kwargs
super().__init__(topk, black_list)調用內部的 _call_serper_api 函數進行搜索，并隨后調用內部 _parse_response 函數對返回的結果進行結構化處理。在調用過程中，如果發生異常，該函數會實施重試機制，即在短暫等待后重新嘗試，直至達到預設的最大重試次數。
對于有每秒訪問限制的搜索 API，由于用的多線程調用，此函數在嘗試最大重試次數之后仍可能報錯。
@cached(cache=TTLCache(maxsize=100, ttl=600))
def search(self, query: str, max_retry: int = 3) -> dict:
for attempt in range(max_retry):
try:
response = self._call_serper_api(query)
return self._parse_response(response)
except Exception as e:
logging.exception(str(e))
warnings.warn(
f'Retry {attempt + 1}/{max_retry} due to error: {e}')
time.sleep(random.randint(2, 5))
raise Exception(
'Failed to get search results from Google Serper Search after retries.'
)對相對應的 搜索 API 發送請求，并且獲得對應結果，其參數以及請求時的格式請參考對應的搜索 API 文檔。
def _call_serper_api(self, query: str) -> dict:
endpoint = f'https://google.serper.dev/{self.search_type}'
params = {
'q': query,
'num': self.topk,
**{
key: value
for key, value in self.kwargs.items() if value is not None
},
}
headers = {
'X-API-KEY': self.api_key or '',
'Content-Type': 'application/json'
}
response = requests.get(
endpoint, headers=headers, params=params, proxies=self.proxy)
response.raise_for_status()
return response.json()對于 搜索API 返回的每一個結果，將其提取并包裝成 (url,snippest,title) 格式的元組，將這些元組添加到一個名為 raw_results 的列表中，隨后將 raw_results 列表作為參數傳遞給 BaseSearch 類中的 _filter_results 函數。
def _parse_response(self, response: dict) -> dict:
raw_results = []
for result in response[self.result_key_for_type[
self.search_type]][:self.topk]:
description = result.get('snippet', '')
attributes = '. '.join(
f'{attribute}: {value}'
for attribute, value in result.get('attributes', {}).items())
raw_results.append(
(result.get('link', ''),
f'{description}. {attributes}' if attributes else description,
result.get('title', '')))
return self._filter_results(raw_results)本文深入探討了在 MindSearch 中實現新的 搜索 API 所需注意的關鍵事項，并詳細介紹了 SearcherAgent 的調用流程，包括涉及的類和函數。特別地，我們重點介紹了如何在 bing_browser.py 中支持新的搜索 API，具體包括實現新的 Searcher 類，以及定義 def search()、def _call_serper_api()和def _parse_response()函數，以確保新的搜索 API 能夠無縫集成并擴展現有功能。
文章轉自微信公眾號@OpenMMLab
