建可控AI Agent Web搜索：Camofox與HermesAgent集成實(shí)戰(zhàn))
1. 從“能用”到“可控”為什么我們需要一個(gè)真正的Web搜索能力最近在折騰HermesAgent想讓它幫我處理一些需要實(shí)時(shí)信息的任務(wù)比如查查最新的技術(shù)動(dòng)態(tài)、對(duì)比幾個(gè)開(kāi)源項(xiàng)目的活躍度或者看看某個(gè)API的官方文檔有沒(méi)有更新。用了一段時(shí)間我發(fā)現(xiàn)它自帶的Web Search功能或者說(shuō)很多AI Agent框架提供的類(lèi)似能力總感覺(jué)差了那么點(diǎn)意思。它們更像是“有”和“無(wú)”的區(qū)別而不是“好用”和“不好用”的區(qū)別。問(wèn)題出在哪里呢我總結(jié)了幾點(diǎn)最讓我頭疼的地方。第一是搜索質(zhì)量不可控。你讓Agent去搜“2024年最佳Python異步框架”它返回的結(jié)果可能來(lái)自某個(gè)不知名的博客內(nèi)容質(zhì)量參差不齊甚至可能是過(guò)時(shí)的信息。Agent本身缺乏對(duì)信息來(lái)源權(quán)威性和時(shí)效性的判斷能力它只是把搜索API返回的原始摘要或前幾條結(jié)果喂給了大模型。第二是操作流程不透明。你只知道它“去搜索了”但具體搜了什么關(guān)鍵詞點(diǎn)開(kāi)了哪些鏈接遇到了什么錯(cuò)誤比如網(wǎng)站反爬、超時(shí)這些過(guò)程完全是個(gè)黑盒。當(dāng)結(jié)果不盡如人意時(shí)你很難進(jìn)行調(diào)試和優(yōu)化。第三是缺乏深度交互。很多搜索是一次性的無(wú)法基于初次搜索結(jié)果進(jìn)行追問(wèn)、篩選或聚焦。比如你讓Agent“了解一下Camofox”它可能返回一個(gè)概述。但如果你想接著問(wèn)“它和另一個(gè)工具X在架構(gòu)上有什么區(qū)別”Agent往往需要發(fā)起一次全新的、與上文割裂的搜索而不是在之前的搜索上下文里進(jìn)行精煉。這讓我意識(shí)到對(duì)于希望將AI Agent投入嚴(yán)肅生產(chǎn)環(huán)境或復(fù)雜工作流的開(kāi)發(fā)者來(lái)說(shuō)一個(gè)“真正可控”的Web搜索能力不是錦上添花而是雪中送炭。可控意味著我們能定義搜索的源比如優(yōu)先使用Google、DuckDuckGo還是特定站內(nèi)搜索、能干預(yù)搜索的策略關(guān)鍵詞的生成與優(yōu)化、能審計(jì)搜索的全過(guò)程、能處理搜索中遇到的異常如驗(yàn)證碼、限流并能將多次搜索納入一個(gè)連貫的會(huì)話上下文。正是在這種需求背景下我開(kāi)始關(guān)注并嘗試集成Camofox Web Search。它不是一個(gè)簡(jiǎn)單的搜索API封裝而是一個(gè)設(shè)計(jì)給AI Agent使用的、可編程的搜索“工具箱”。它把搜索這個(gè)動(dòng)作拆解成了可觀測(cè)、可配置、可擴(kuò)展的組件。通過(guò)將其作為Python Provider集成到HermesAgent中我們就有機(jī)會(huì)親手打造一個(gè)既強(qiáng)大又聽(tīng)話的“信息偵察兵”。接下來(lái)我就詳細(xì)分享一下這次集成的完整過(guò)程、核心原理以及我踩過(guò)的那些坑。2. Camofox Web Search 核心架構(gòu)與 HermesAgent 集成點(diǎn)剖析在動(dòng)手寫(xiě)代碼之前我們必須先搞清楚Camofox Web Search到底提供了什么以及它如何與HermesAgent的架構(gòu)相匹配。盲目集成只會(huì)事倍功半。Camofox Web Search 的核心思想是“搜索即流程”。它不是一個(gè)返回字符串的search(query)函數(shù)那么簡(jiǎn)單。一個(gè)完整的、可靠的搜索流程通常包含以下步驟查詢(xún)理解與優(yōu)化將用戶的自然語(yǔ)言指令如“幫我找找最近關(guān)于Rust內(nèi)存安全的熱門(mén)文章”轉(zhuǎn)化為一個(gè)或多個(gè)有效的搜索關(guān)鍵詞如“Rust memory safety 2024 article reddit hacker news”。這里可能涉及去除停用詞、同義詞擴(kuò)展、領(lǐng)域詞加權(quán)等。搜索執(zhí)行向一個(gè)或多個(gè)搜索引擎如Google Custom Search JSON API、SerpAPI、DuckDuckGo HTML抓取等發(fā)起請(qǐng)求。結(jié)果解析與清洗從搜索引擎返回的HTML或JSON中提取出標(biāo)題、鏈接、摘要等結(jié)構(gòu)化信息并過(guò)濾掉廣告、無(wú)效鏈接或低質(zhì)量?jī)?nèi)容。內(nèi)容提取對(duì)于篩選出的結(jié)果鏈接可能需要進(jìn)一步抓取頁(yè)面主體內(nèi)容并利用Readability算法或CSS選擇器剔除導(dǎo)航欄、廣告等噪音獲取純凈的正文。結(jié)果合成與格式化將多個(gè)來(lái)源的結(jié)果進(jìn)行去重、排序、摘要并格式化成大語(yǔ)言模型LLM易于處理的文本例如Markdown格式并附上來(lái)源鏈接。Camofox 將這些步驟模塊化了。它提供了諸如QueryOptimizer、SearchEngine、ResultParser、ContentExtractor等基礎(chǔ)組件。我們的工作就是像搭積木一樣用這些組件組裝成一個(gè)符合我們需求的SearchPipeline。那么它如何融入HermesAgent呢HermesAgent的核心是工具Tool。Agent通過(guò)調(diào)用工具來(lái)與世界交互。我們的目標(biāo)就是創(chuàng)建一個(gè)名為web_search的工具。當(dāng)HermesAgent認(rèn)為需要聯(lián)網(wǎng)搜索時(shí)就會(huì)調(diào)用這個(gè)工具。這里的關(guān)鍵集成點(diǎn)是BaseTool類(lèi)。在HermesAgent中任何工具都需要繼承這個(gè)基類(lèi)并實(shí)現(xiàn)其run方法。我們的WebSearchTool的run方法內(nèi)部將封裝整個(gè)Camofox搜索流程的調(diào)用。此外我們還需要為這個(gè)工具編寫(xiě)清晰的描述Description和參數(shù)模式Parameters Schema以便HermesAgent的規(guī)劃器Planner能準(zhǔn)確理解在什么情況下、以什么參數(shù)來(lái)調(diào)用這個(gè)工具。一個(gè)高級(jí)的集成還會(huì)考慮上下文管理。例如我們的搜索工具應(yīng)該能接收會(huì)話歷史作為上下文以便優(yōu)化查詢(xún)比如避免重復(fù)搜索剛剛提過(guò)的問(wèn)題?；蛘吖ぞ邎?zhí)行后不僅返回搜索結(jié)果文本還能將關(guān)鍵的元數(shù)據(jù)如用到的搜索鏈接、耗時(shí)以結(jié)構(gòu)化的方式返回供Agent或后續(xù)流程使用。為了更直觀地理解數(shù)據(jù)流我們可以看下面這個(gè)簡(jiǎn)化的集成架構(gòu)圖用戶提問(wèn) | v HermesAgent 規(guī)劃器分析 | v 決定調(diào)用 web_search 工具 | v WebSearchTool.run(query, context) 被調(diào)用 | v 內(nèi)部創(chuàng)建或復(fù)用 Camofox SearchPipeline |-- 查詢(xún)優(yōu)化 |-- 執(zhí)行搜索 (調(diào)用 Google API) |-- 解析結(jié)果 |-- 提取關(guān)鍵頁(yè)面內(nèi)容 | v 格式化搜索結(jié)果 (Markdown 引用) | v 結(jié)果返回給 HermesAgent | v Agent 將結(jié)果整合進(jìn)回復(fù)返回給用戶理解了這些我們就有了清晰的藍(lán)圖。接下來(lái)我們從零開(kāi)始一步步搭建這個(gè)可用的工具。3. 實(shí)戰(zhàn)從零構(gòu)建 HermesAgent 的 Camofox Web Search Provider3.1 環(huán)境準(zhǔn)備與依賴(lài)安裝首先我們需要一個(gè)干凈的Python環(huán)境。我強(qiáng)烈建議使用venv或conda來(lái)管理項(xiàng)目依賴(lài)避免污染全局環(huán)境。# 創(chuàng)建并激活虛擬環(huán)境 python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 安裝 HermesAgent 核心庫(kù) pip install hermes-agent接下來(lái)是安裝Camofox Web Search。由于它可能不是一個(gè)通過(guò)PyPI直接發(fā)布的包我們通常需要從GitHub倉(cāng)庫(kù)安裝。在動(dòng)手之前最好先去它的官方倉(cāng)庫(kù)查看最新的安裝方式。# 假設(shè)Camofox可以通過(guò)pip從GitHub安裝 pip install camofox-web-search githttps://github.com/camofox-labs/camofox-web-search.git # 或者如果你克隆了倉(cāng)庫(kù) pip install -e /path/to/camofox-web-search除了核心庫(kù)我們還需要一些“引擎”依賴(lài)。Camofox本身是搜索流程的編排框架具體的搜索執(zhí)行需要底層的引擎。最常用、最穩(wěn)定的是通過(guò)官方API進(jìn)行搜索例如Google Custom Search JSON API。這需要你擁有一個(gè)Google Cloud項(xiàng)目并啟用該API以及一個(gè)可用的API密鑰和自定義搜索引擎IDCX。這種方式合法、穩(wěn)定但可能有額度限制。另一種方式是使用無(wú)頭瀏覽器如Playwright進(jìn)行模擬搜索這種方式更靈活但更復(fù)雜且容易觸發(fā)反爬機(jī)制。對(duì)于生產(chǎn)環(huán)境我強(qiáng)烈推薦使用官方API。因此我們還需要安裝對(duì)應(yīng)引擎的客戶端庫(kù)。對(duì)于Google Custom SearchCamofox可能已經(jīng)封裝好了我們只需確保requests庫(kù)已安裝。pip install requests # 如果使用Playwright引擎則需要 pip install playwright playwright install chromium注意使用非API方式抓取搜索引擎結(jié)果可能違反目標(biāo)網(wǎng)站的服務(wù)條款。請(qǐng)務(wù)必在合規(guī)的前提下進(jìn)行開(kāi)發(fā)和測(cè)試并充分尊重robots.txt。對(duì)于學(xué)習(xí)和內(nèi)部測(cè)試使用官方API是唯一推薦的方式。3.2 構(gòu)建核心的 WebSearchTool 類(lèi)現(xiàn)在我們來(lái)創(chuàng)建工具的核心類(lèi)。在你的HermesAgent項(xiàng)目目錄下創(chuàng)建一個(gè)新文件例如camofox_provider.py。import logging from typing import Dict, Any, Optional from hermes_agent.agent.tools import BaseTool from pydantic import Field, BaseModel # 配置日志方便調(diào)試 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 首先定義工具的輸入?yún)?shù)模型。這告訴HermesAgent這個(gè)工具需要什么參數(shù)。 class WebSearchInput(BaseModel): query: str Field( ..., descriptionThe search query string. Be specific and include key terms for better results. ) max_results: Optional[int] Field( 5, descriptionMaximum number of search results to return. Defaults to 5. ) search_context: Optional[str] Field( None, descriptionAdditional context from the conversation to refine the search. ) # 然后創(chuàng)建工具類(lèi)繼承自BaseTool class WebSearchTool(BaseTool): name: str web_search description: str ( A powerful and controllable web search tool. Use this when you need to find current, real-world information, verify facts, get latest news, or browse specific websites. Input should be a clear search query. ) args_schema: type[BaseModel] WebSearchInput def __init__(self, **kwargs): super().__init__(**kwargs) # 初始化Camofox搜索管道。這里采用懶加載在實(shí)際運(yùn)行時(shí)創(chuàng)建。 self._search_pipeline None # 加載配置可以從環(huán)境變量或配置文件中讀取 self.api_key kwargs.get(api_key) or os.getenv(GOOGLE_API_KEY) self.search_engine_id kwargs.get(search_engine_id) or os.getenv(GOOGLE_SEARCH_ENGINE_ID) if not self.api_key or not self.search_engine_id: logger.warning(Google API key or Search Engine ID not configured. Web search may fail.) # 你可以在這里初始化其他組件比如查詢(xún)優(yōu)化器、內(nèi)容提取器 self._init_components() def _init_components(self): 初始化Camofox的各個(gè)組件。這里是一個(gè)示例配置。 # 注意以下導(dǎo)入和類(lèi)名需要根據(jù)Camofox的實(shí)際代碼進(jìn)行調(diào)整 try: from camofox.core import SearchPipeline from camofox.engines import GoogleSearchEngine from camofox.parsers import GoogleJsonParser from camofox.extractors import ReadabilityExtractor from camofox.optimizers import SimpleQueryOptimizer # 1. 查詢(xún)優(yōu)化器可以在這里加入同義詞庫(kù)、領(lǐng)域詞等 self.query_optimizer SimpleQueryOptimizer() # 2. 搜索引擎使用Google Custom Search JSON API self.engine GoogleSearchEngine( api_keyself.api_key, engine_idself.search_engine_id, num_results10 # 向API請(qǐng)求多一些結(jié)果供后續(xù)篩選 ) # 3. 結(jié)果解析器處理Google API返回的JSON self.parser GoogleJsonParser() # 4. 內(nèi)容提取器用于深入抓取搜索結(jié)果頁(yè)面的正文 self.extractor ReadabilityExtractor( timeout10, user_agentMozilla/5.0 ... # 使用合理的User-Agent ) # 組裝管道 self._search_pipeline SearchPipeline( optimizerself.query_optimizer, engineself.engine, parserself.parser, extractorself.extractor, ) logger.info(Camofox SearchPipeline initialized successfully.) except ImportError as e: logger.error(fFailed to import Camofox modules: {e}. Please check installation.) self._search_pipeline None except Exception as e: logger.error(fFailed to initialize search pipeline: {e}) self._search_pipeline None async def _run(self, query: str, max_results: int 5, **kwargs) - str: 工具的核心運(yùn)行方法。HermesAgent會(huì)調(diào)用此方法。 if not self._search_pipeline: return Error: Web search tool is not properly initialized. Check API keys and dependencies. logger.info(fExecuting web search for query: {query}) try: # 調(diào)用Camofox管道執(zhí)行搜索 # 注意Camofox的run方法可能是異步的也可能是同步的需根據(jù)其實(shí)際設(shè)計(jì)調(diào)整。 # 這里假設(shè)它是同步的。如果是異步的需要加上await。 search_results self._search_pipeline.run( queryquery, max_resultsmax_results, # 可以傳入其他上下文參數(shù)例如之前的對(duì)話歷史 contextkwargs.get(search_context) ) # 格式化結(jié)果使其對(duì)LLM友好 formatted_results self._format_results(search_results) logger.info(fSearch completed. Found {len(search_results)} results.) return formatted_results except Exception as e: logger.exception(fWeb search failed for query {query}: {e}) # 返回一個(gè)對(duì)Agent友好的錯(cuò)誤信息而不是拋出異常 return fI encountered an error while searching for {query}: {str(e)}. Please try again or rephrase your query. def _format_results(self, results: list) - str: 將Camofox返回的結(jié)構(gòu)化結(jié)果格式化為文本。 if not results: return No relevant web search results found. formatted_lines [**Web Search Results:**, ] for i, res in enumerate(results[:5], 1): # 確保不超過(guò)max_results title res.get(title, No Title) link res.get(link, #) snippet res.get(snippet, No snippet available.) # 可能還有提取的全文內(nèi)容 res.get(extracted_content) formatted_lines.append(f{i}. **[{title}]({link})**) formatted_lines.append(f {snippet}) formatted_lines.append() # 空行分隔 formatted_lines.append(*Search powered by Camofox Web Search.*) return \n.join(formatted_lines)這個(gè)WebSearchTool類(lèi)已經(jīng)具備了基本骨架。它定義了工具的名稱(chēng)、描述和參數(shù)并在初始化時(shí)嘗試構(gòu)建Camofox的搜索管道。_run方法是執(zhí)行搜索和返回結(jié)果的核心。3.3 配置與初始化讓工具“活”起來(lái)創(chuàng)建了工具類(lèi)我們還需要將它“注冊(cè)”到HermesAgent中。通常這需要在初始化Agent時(shí)將工具實(shí)例添加到工具列表里。首先確保你的API密鑰和搜索引擎ID已經(jīng)設(shè)置好。最佳實(shí)踐是使用環(huán)境變量。export GOOGLE_API_KEYyour_google_api_key_here export GOOGLE_SEARCH_ENGINE_IDyour_search_engine_id_here然后在你的主程序文件中比如main.py這樣初始化Agentimport asyncio import os from hermes_agent.agent import HermesAgent from camofox_provider import WebSearchTool # 導(dǎo)入我們剛寫(xiě)的工具 async def main(): # 1. 實(shí)例化我們的搜索工具 web_search_tool WebSearchTool( # 可以在這里覆蓋環(huán)境變量但更推薦用環(huán)境變量 # api_keyxxx, # search_engine_idxxx ) # 2. 準(zhǔn)備工具列表 tools [web_search_tool] # 3. 初始化HermesAgent并傳入工具 # 注意這里需要你已有LLM的配置如OpenAI API Key agent HermesAgent( modelgpt-4, # 或你使用的其他模型 toolstools, # ... 其他Agent配置參數(shù) ) # 4. 與Agent交互 response await agent.run(What are the main features of the latest version of Python?) print(response) if __name__ __main__: asyncio.run(main())現(xiàn)在當(dāng)你向Agent提問(wèn)一個(gè)需要最新信息的問(wèn)題時(shí)它應(yīng)該能自主決定調(diào)用web_search工具并返回包含真實(shí)鏈接的搜索結(jié)果。4. 超越基礎(chǔ)實(shí)現(xiàn)可控性與可觀測(cè)性的高級(jí)技巧基礎(chǔ)的集成只能算“跑通”。要讓這個(gè)搜索工具真正強(qiáng)大、可靠我們必須深入細(xì)節(jié)實(shí)現(xiàn)精細(xì)化的控制和全面的可觀測(cè)性。下面是我在實(shí)戰(zhàn)中總結(jié)的幾個(gè)關(guān)鍵技巧。4.1 查詢(xún)優(yōu)化策略讓Agent“更懂”搜索默認(rèn)情況下Agent生成的搜索查詢(xún)可能直接來(lái)自用戶問(wèn)題這并不總是最優(yōu)的。例如用戶問(wèn)“怎么用Python做數(shù)據(jù)分析”這個(gè)查詢(xún)過(guò)于寬泛。我們可以通過(guò)QueryOptimizer來(lái)改進(jìn)。Camofox允許我們自定義優(yōu)化器。一個(gè)簡(jiǎn)單的策略是關(guān)鍵詞提取與重組。我們可以集成一個(gè)輕量級(jí)的NLP庫(kù)如spacy或jieba來(lái)提取名詞實(shí)體和動(dòng)詞作為核心關(guān)鍵詞并附加上下文詞。# 示例一個(gè)簡(jiǎn)單的基于規(guī)則和上下文的查詢(xún)優(yōu)化器 from camofox.optimizers import BaseOptimizer class ContextAwareQueryOptimizer(BaseOptimizer): def __init__(self, stop_wordsNone): self.stop_words stop_words or {how, to, what, is, the, a, an} def optimize(self, query: str, context: Optional[str] None) - str: # 1. 分詞并去除停用詞 (這里用簡(jiǎn)單空格分割示意) words query.lower().split() key_terms [w for w in words if w not in self.stop_words] # 2. 如果有上下文從中提取可能相關(guān)的實(shí)體這是一個(gè)簡(jiǎn)化示例 additional_terms [] if context: # 假設(shè)上下文里提到了“pandas”和“visualization” if pandas in context.lower(): additional_terms.append(pandas) if plot in context.lower() or chart in context.lower(): additional_terms.append(data visualization) # 3. 組合關(guān)鍵詞避免重復(fù) all_terms list(dict.fromkeys(key_terms additional_terms)) optimized_query .join(all_terms) # 4. 如果優(yōu)化后為空回退到原查詢(xún) return optimized_query if optimized_query.strip() else query將這個(gè)優(yōu)化器注入到我們的SearchPipeline中搜索質(zhì)量會(huì)有顯著提升。更高級(jí)的策略還可以利用小語(yǔ)言模型SLM來(lái)重寫(xiě)查詢(xún)或者維護(hù)一個(gè)領(lǐng)域特定的同義詞庫(kù)。4.2 結(jié)果過(guò)濾與評(píng)分提升信息信噪比不是所有搜索結(jié)果都是有用的。我們需要一個(gè)過(guò)濾層。Camofox的架構(gòu)允許我們?cè)诮馕鼋Y(jié)果后、提取內(nèi)容前插入一個(gè)Filter組件。一個(gè)常見(jiàn)的過(guò)濾策略是基于來(lái)源可信度和內(nèi)容新鮮度。我們可以維護(hù)一個(gè)可信域名列表如*.github.io,stackoverflow.com,python.org并給來(lái)自這些域名的結(jié)果加分。同時(shí)解析結(jié)果中的日期信息優(yōu)先選擇一年內(nèi)的內(nèi)容。from camofox.filters import BaseFilter from datetime import datetime, timedelta import re class RelevanceFilter(BaseFilter): def __init__(self, trusted_domainsNone, max_age_days365): self.trusted_domains trusted_domains or [github.com, stackoverflow.com, docs.python.org] self.max_age timedelta(daysmax_age_days) def filter(self, results: list) - list: filtered_results [] for res in results: score 0.0 link res.get(link, ) # 1. 域名可信度評(píng)分 for domain in self.trusted_domains: if domain in link: score 1.0 break # 2. 新鮮度評(píng)分如果結(jié)果里有日期 date_str res.get(date) if date_str: try: # 嘗試解析各種日期格式這里簡(jiǎn)化處理 # 實(shí)際中可能需要更復(fù)雜的日期解析庫(kù) pub_date self._parse_date(date_str) if datetime.now() - pub_date self.max_age: score 0.5 except: pass # 3. 標(biāo)題/摘要關(guān)鍵詞匹配度簡(jiǎn)單實(shí)現(xiàn) text (res.get(title, ) res.get(snippet, )).lower() if tutorial in text: score 0.3 if example in text or sample in text: score 0.2 if score 0.5: # 設(shè)置一個(gè)閾值 res[relevance_score] score filtered_results.append(res) # 按評(píng)分排序 filtered_results.sort(keylambda x: x.get(relevance_score, 0), reverseTrue) return filtered_results def _parse_date(self, date_str: str) - datetime: # 簡(jiǎn)化的日期解析實(shí)際項(xiàng)目應(yīng)使用dateutil.parser for fmt in (%Y-%m-%d, %d %b %Y, %B %d, %Y): try: return datetime.strptime(date_str[:10], fmt) except ValueError: continue return datetime.now() - timedelta(days365*2) # 默認(rèn)返回一個(gè)很舊的日期將這個(gè)過(guò)濾器加入到管道中可以自動(dòng)屏蔽垃圾站點(diǎn)和過(guò)時(shí)信息讓Agent獲取到的信息質(zhì)量更高。4.3 全面的日志與監(jiān)控讓黑盒變透明可控性的前提是可觀測(cè)性。我們需要知道每一次搜索到底發(fā)生了什么。Camofox的管道設(shè)計(jì)通常支持中間步驟的鉤子hooks或事件監(jiān)聽(tīng)。我們可以為管道注入一個(gè)日志記錄器記錄下優(yōu)化后的查詢(xún)?cè)~、請(qǐng)求的URL、返回的狀態(tài)碼、解析出的結(jié)果數(shù)量、過(guò)濾后的數(shù)量等。import json from camofox.core import SearchPipeline class LoggingPipeline(SearchPipeline): def run(self, query, **kwargs): logger.info(f[SearchPipeline] Start. Original query: {query}) optimized_query self.optimizer.optimize(query, kwargs.get(context)) logger.info(f[SearchPipeline] Optimized query: {optimized_query}) raw_results self.engine.search(optimized_query, **kwargs) logger.info(f[SearchPipeline] Engine returned {len(raw_results) if raw_results else 0} raw items.) parsed_results self.parser.parse(raw_results) logger.info(f[SearchPipeline] Parser produced {len(parsed_results)} parsed results.) filtered_results self.filter.filter(parsed_results) if hasattr(self, filter) else parsed_results logger.info(f[SearchPipeline] After filtering: {len(filtered_results)} results.) # 可以選擇性地記錄前幾個(gè)結(jié)果的標(biāo)題和鏈接用于調(diào)試 for i, res in enumerate(filtered_results[:3]): logger.debug(f Result {i1}: {res.get(title)} - {res.get(link)}) final_results [] for res in filtered_results: if self.extractor: try: content self.extractor.extract(res[link]) res[extracted_content] content[:500] ... if len(content) 500 else content # 只存摘要 except Exception as e: logger.warning(fFailed to extract content from {res[link]}: {e}) res[extracted_content] final_results.append(res) logger.info(f[SearchPipeline] Finished. Returning {len(final_results)} final results.) return final_results使用這個(gè)LoggingPipeline替代基礎(chǔ)的SearchPipeline所有的搜索過(guò)程都將被清晰記錄。你還可以將日志發(fā)送到ELK?；虮O(jiān)控系統(tǒng)繪制搜索成功率、耗時(shí)、常用查詢(xún)?cè)~等圖表實(shí)現(xiàn)運(yùn)維級(jí)別的可控。4.4 錯(cuò)誤處理與降級(jí)策略構(gòu)建魯棒的服務(wù)網(wǎng)絡(luò)搜索充滿了不確定性API額度用盡、網(wǎng)絡(luò)超時(shí)、網(wǎng)站反爬、解析失敗等等。一個(gè)健壯的工具必須有完善的錯(cuò)誤處理和降級(jí)機(jī)制。重試機(jī)制對(duì)于網(wǎng)絡(luò)超時(shí)等臨時(shí)性錯(cuò)誤應(yīng)該自動(dòng)重試?？梢允褂胻enacity庫(kù)。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions class RobustSearchEngine(GoogleSearchEngine): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def search(self, query, **kwargs): return super().search(query, **kwargs)多引擎降級(jí)配置多個(gè)搜索引擎如Google API作為主引擎DuckDuckGo作為備用。當(dāng)主引擎失敗時(shí)自動(dòng)切換到備用引擎。from camofox.engines import DuckDuckGoEngine class FallbackSearchEngine: def __init__(self, primary_engine, fallback_engine): self.primary primary_engine self.fallback fallback_engine def search(self, query, **kwargs): try: return self.primary.search(query, **kwargs) except Exception as e: logger.error(fPrimary engine failed: {e}. Switching to fallback.) return self.fallback.search(query, **kwargs)優(yōu)雅的失敗響應(yīng)在_run方法中捕獲所有異常并返回一個(gè)對(duì)用戶和Agent都友好的提示而不是讓整個(gè)Agent會(huì)話崩潰。例如“搜索服務(wù)暫時(shí)不可用可能是網(wǎng)絡(luò)問(wèn)題或API限額已滿。您可以稍后重試或嘗試詢(xún)問(wèn)一些無(wú)需實(shí)時(shí)信息的問(wèn)題?！蓖ㄟ^(guò)實(shí)施這些高級(jí)技巧你的Web Search工具將從“一個(gè)能用的功能”升級(jí)為“一個(gè)可靠、可控、可觀測(cè)的核心服務(wù)組件”。這正是在生產(chǎn)環(huán)境中集成此類(lèi)工具所必需的深度考量。 構(gòu)建可控AI Agent Web搜索：Camofox與HermesAgent集成實(shí)戰(zhàn)
1. 從“能用”到“可控”為什么我們需要一個(gè)真正的Web搜索能力最近在折騰HermesAgent想讓它幫我處理一些需要實(shí)時(shí)信息的任務(wù)比如查查最新的技術(shù)動(dòng)態(tài)、對(duì)比幾個(gè)開(kāi)源項(xiàng)目的活躍度或者看看某個(gè)API的官方文檔有沒(méi)有更新。用了一段時(shí)間我發(fā)現(xiàn)它自帶的Web Search功能或者說(shuō)很多AI Agent框架提供的類(lèi)似能力總感覺(jué)差了那么點(diǎn)意思。它們更像是“有”和“無(wú)”的區(qū)別而不是“好用”和“不好用”的區(qū)別。問(wèn)題出在哪里呢我總結(jié)了幾點(diǎn)最讓我頭疼的地方。第一是搜索質(zhì)量不可控。你讓Agent去搜“2024年最佳Python異步框架”它返回的結(jié)果可能來(lái)自某個(gè)不知名的博客內(nèi)容質(zhì)量參差不齊甚至可能是過(guò)時(shí)的信息。Agent本身缺乏對(duì)信息來(lái)源權(quán)威性和時(shí)效性的判斷能力它只是把搜索API返回的原始摘要或前幾條結(jié)果喂給了大模型。第二是操作流程不透明。你只知道它“去搜索了”但具體搜了什么關(guān)鍵詞點(diǎn)開(kāi)了哪些鏈接遇到了什么錯(cuò)誤比如網(wǎng)站反爬、超時(shí)這些過(guò)程完全是個(gè)黑盒。當(dāng)結(jié)果不盡如人意時(shí)你很難進(jìn)行調(diào)試和優(yōu)化。第三是缺乏深度交互。很多搜索是一次性的無(wú)法基于初次搜索結(jié)果進(jìn)行追問(wèn)、篩選或聚焦。比如你讓Agent“了解一下Camofox”它可能返回一個(gè)概述。但如果你想接著問(wèn)“它和另一個(gè)工具X在架構(gòu)上有什么區(qū)別”Agent往往需要發(fā)起一次全新的、與上文割裂的搜索而不是在之前的搜索上下文里進(jìn)行精煉。這讓我意識(shí)到對(duì)于希望將AI Agent投入嚴(yán)肅生產(chǎn)環(huán)境或復(fù)雜工作流的開(kāi)發(fā)者來(lái)說(shuō)一個(gè)“真正可控”的Web搜索能力不是錦上添花而是雪中送炭。可控意味著我們能定義搜索的源比如優(yōu)先使用Google、DuckDuckGo還是特定站內(nèi)搜索、能干預(yù)搜索的策略關(guān)鍵詞的生成與優(yōu)化、能審計(jì)搜索的全過(guò)程、能處理搜索中遇到的異常如驗(yàn)證碼、限流并能將多次搜索納入一個(gè)連貫的會(huì)話上下文。正是在這種需求背景下我開(kāi)始關(guān)注并嘗試集成Camofox Web Search。它不是一個(gè)簡(jiǎn)單的搜索API封裝而是一個(gè)設(shè)計(jì)給AI Agent使用的、可編程的搜索“工具箱”。它把搜索這個(gè)動(dòng)作拆解成了可觀測(cè)、可配置、可擴(kuò)展的組件。通過(guò)將其作為Python Provider集成到HermesAgent中我們就有機(jī)會(huì)親手打造一個(gè)既強(qiáng)大又聽(tīng)話的“信息偵察兵”。接下來(lái)我就詳細(xì)分享一下這次集成的完整過(guò)程、核心原理以及我踩過(guò)的那些坑。2. Camofox Web Search 核心架構(gòu)與 HermesAgent 集成點(diǎn)剖析在動(dòng)手寫(xiě)代碼之前我們必須先搞清楚Camofox Web Search到底提供了什么以及它如何與HermesAgent的架構(gòu)相匹配。盲目集成只會(huì)事倍功半。Camofox Web Search 的核心思想是“搜索即流程”。它不是一個(gè)返回字符串的search(query)函數(shù)那么簡(jiǎn)單。一個(gè)完整的、可靠的搜索流程通常包含以下步驟查詢(xún)理解與優(yōu)化將用戶的自然語(yǔ)言指令如“幫我找找最近關(guān)于Rust內(nèi)存安全的熱門(mén)文章”轉(zhuǎn)化為一個(gè)或多個(gè)有效的搜索關(guān)鍵詞如“Rust memory safety 2024 article reddit hacker news”。這里可能涉及去除停用詞、同義詞擴(kuò)展、領(lǐng)域詞加權(quán)等。搜索執(zhí)行向一個(gè)或多個(gè)搜索引擎如Google Custom Search JSON API、SerpAPI、DuckDuckGo HTML抓取等發(fā)起請(qǐng)求。結(jié)果解析與清洗從搜索引擎返回的HTML或JSON中提取出標(biāo)題、鏈接、摘要等結(jié)構(gòu)化信息并過(guò)濾掉廣告、無(wú)效鏈接或低質(zhì)量?jī)?nèi)容。內(nèi)容提取對(duì)于篩選出的結(jié)果鏈接可能需要進(jìn)一步抓取頁(yè)面主體內(nèi)容并利用Readability算法或CSS選擇器剔除導(dǎo)航欄、廣告等噪音獲取純凈的正文。結(jié)果合成與格式化將多個(gè)來(lái)源的結(jié)果進(jìn)行去重、排序、摘要并格式化成大語(yǔ)言模型LLM易于處理的文本例如Markdown格式并附上來(lái)源鏈接。Camofox 將這些步驟模塊化了。它提供了諸如QueryOptimizer、SearchEngine、ResultParser、ContentExtractor等基礎(chǔ)組件。我們的工作就是像搭積木一樣用這些組件組裝成一個(gè)符合我們需求的SearchPipeline。那么它如何融入HermesAgent呢HermesAgent的核心是工具Tool。Agent通過(guò)調(diào)用工具來(lái)與世界交互。我們的目標(biāo)就是創(chuàng)建一個(gè)名為web_search的工具。當(dāng)HermesAgent認(rèn)為需要聯(lián)網(wǎng)搜索時(shí)就會(huì)調(diào)用這個(gè)工具。這里的關(guān)鍵集成點(diǎn)是BaseTool類(lèi)。在HermesAgent中任何工具都需要繼承這個(gè)基類(lèi)并實(shí)現(xiàn)其run方法。我們的WebSearchTool的run方法內(nèi)部將封裝整個(gè)Camofox搜索流程的調(diào)用。此外我們還需要為這個(gè)工具編寫(xiě)清晰的描述Description和參數(shù)模式Parameters Schema以便HermesAgent的規(guī)劃器Planner能準(zhǔn)確理解在什么情況下、以什么參數(shù)來(lái)調(diào)用這個(gè)工具。一個(gè)高級(jí)的集成還會(huì)考慮上下文管理。例如我們的搜索工具應(yīng)該能接收會(huì)話歷史作為上下文以便優(yōu)化查詢(xún)比如避免重復(fù)搜索剛剛提過(guò)的問(wèn)題?；蛘吖ぞ邎?zhí)行后不僅返回搜索結(jié)果文本還能將關(guān)鍵的元數(shù)據(jù)如用到的搜索鏈接、耗時(shí)以結(jié)構(gòu)化的方式返回供Agent或后續(xù)流程使用。為了更直觀地理解數(shù)據(jù)流我們可以看下面這個(gè)簡(jiǎn)化的集成架構(gòu)圖用戶提問(wèn) | v HermesAgent 規(guī)劃器分析 | v 決定調(diào)用 web_search 工具 | v WebSearchTool.run(query, context) 被調(diào)用 | v 內(nèi)部創(chuàng)建或復(fù)用 Camofox SearchPipeline |-- 查詢(xún)優(yōu)化 |-- 執(zhí)行搜索 (調(diào)用 Google API) |-- 解析結(jié)果 |-- 提取關(guān)鍵頁(yè)面內(nèi)容 | v 格式化搜索結(jié)果 (Markdown 引用) | v 結(jié)果返回給 HermesAgent | v Agent 將結(jié)果整合進(jìn)回復(fù)返回給用戶理解了這些我們就有了清晰的藍(lán)圖。接下來(lái)我們從零開(kāi)始一步步搭建這個(gè)可用的工具。3. 實(shí)戰(zhàn)從零構(gòu)建 HermesAgent 的 Camofox Web Search Provider3.1 環(huán)境準(zhǔn)備與依賴(lài)安裝首先我們需要一個(gè)干凈的Python環(huán)境。我強(qiáng)烈建議使用venv或conda來(lái)管理項(xiàng)目依賴(lài)避免污染全局環(huán)境。# 創(chuàng)建并激活虛擬環(huán)境 python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 安裝 HermesAgent 核心庫(kù) pip install hermes-agent接下來(lái)是安裝Camofox Web Search。由于它可能不是一個(gè)通過(guò)PyPI直接發(fā)布的包我們通常需要從GitHub倉(cāng)庫(kù)安裝。在動(dòng)手之前最好先去它的官方倉(cāng)庫(kù)查看最新的安裝方式。# 假設(shè)Camofox可以通過(guò)pip從GitHub安裝 pip install camofox-web-search githttps://github.com/camofox-labs/camofox-web-search.git # 或者如果你克隆了倉(cāng)庫(kù) pip install -e /path/to/camofox-web-search除了核心庫(kù)我們還需要一些“引擎”依賴(lài)。Camofox本身是搜索流程的編排框架具體的搜索執(zhí)行需要底層的引擎。最常用、最穩(wěn)定的是通過(guò)官方API進(jìn)行搜索例如Google Custom Search JSON API。這需要你擁有一個(gè)Google Cloud項(xiàng)目并啟用該API以及一個(gè)可用的API密鑰和自定義搜索引擎IDCX。這種方式合法、穩(wěn)定但可能有額度限制。另一種方式是使用無(wú)頭瀏覽器如Playwright進(jìn)行模擬搜索這種方式更靈活但更復(fù)雜且容易觸發(fā)反爬機(jī)制。對(duì)于生產(chǎn)環(huán)境我強(qiáng)烈推薦使用官方API。因此我們還需要安裝對(duì)應(yīng)引擎的客戶端庫(kù)。對(duì)于Google Custom SearchCamofox可能已經(jīng)封裝好了我們只需確保requests庫(kù)已安裝。pip install requests # 如果使用Playwright引擎則需要 pip install playwright playwright install chromium注意使用非API方式抓取搜索引擎結(jié)果可能違反目標(biāo)網(wǎng)站的服務(wù)條款。請(qǐng)務(wù)必在合規(guī)的前提下進(jìn)行開(kāi)發(fā)和測(cè)試并充分尊重robots.txt。對(duì)于學(xué)習(xí)和內(nèi)部測(cè)試使用官方API是唯一推薦的方式。3.2 構(gòu)建核心的 WebSearchTool 類(lèi)現(xiàn)在我們來(lái)創(chuàng)建工具的核心類(lèi)。在你的HermesAgent項(xiàng)目目錄下創(chuàng)建一個(gè)新文件例如camofox_provider.py。import logging from typing import Dict, Any, Optional from hermes_agent.agent.tools import BaseTool from pydantic import Field, BaseModel # 配置日志方便調(diào)試 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 首先定義工具的輸入?yún)?shù)模型。這告訴HermesAgent這個(gè)工具需要什么參數(shù)。 class WebSearchInput(BaseModel): query: str Field( ..., descriptionThe search query string. Be specific and include key terms for better results. ) max_results: Optional[int] Field( 5, descriptionMaximum number of search results to return. Defaults to 5. ) search_context: Optional[str] Field( None, descriptionAdditional context from the conversation to refine the search. ) # 然后創(chuàng)建工具類(lèi)繼承自BaseTool class WebSearchTool(BaseTool): name: str web_search description: str ( A powerful and controllable web search tool. Use this when you need to find current, real-world information, verify facts, get latest news, or browse specific websites. Input should be a clear search query. ) args_schema: type[BaseModel] WebSearchInput def __init__(self, **kwargs): super().__init__(**kwargs) # 初始化Camofox搜索管道。這里采用懶加載在實(shí)際運(yùn)行時(shí)創(chuàng)建。 self._search_pipeline None # 加載配置可以從環(huán)境變量或配置文件中讀取 self.api_key kwargs.get(api_key) or os.getenv(GOOGLE_API_KEY) self.search_engine_id kwargs.get(search_engine_id) or os.getenv(GOOGLE_SEARCH_ENGINE_ID) if not self.api_key or not self.search_engine_id: logger.warning(Google API key or Search Engine ID not configured. Web search may fail.) # 你可以在這里初始化其他組件比如查詢(xún)優(yōu)化器、內(nèi)容提取器 self._init_components() def _init_components(self): 初始化Camofox的各個(gè)組件。這里是一個(gè)示例配置。 # 注意以下導(dǎo)入和類(lèi)名需要根據(jù)Camofox的實(shí)際代碼進(jìn)行調(diào)整 try: from camofox.core import SearchPipeline from camofox.engines import GoogleSearchEngine from camofox.parsers import GoogleJsonParser from camofox.extractors import ReadabilityExtractor from camofox.optimizers import SimpleQueryOptimizer # 1. 查詢(xún)優(yōu)化器可以在這里加入同義詞庫(kù)、領(lǐng)域詞等 self.query_optimizer SimpleQueryOptimizer() # 2. 搜索引擎使用Google Custom Search JSON API self.engine GoogleSearchEngine( api_keyself.api_key, engine_idself.search_engine_id, num_results10 # 向API請(qǐng)求多一些結(jié)果供后續(xù)篩選 ) # 3. 結(jié)果解析器處理Google API返回的JSON self.parser GoogleJsonParser() # 4. 內(nèi)容提取器用于深入抓取搜索結(jié)果頁(yè)面的正文 self.extractor ReadabilityExtractor( timeout10, user_agentMozilla/5.0 ... # 使用合理的User-Agent ) # 組裝管道 self._search_pipeline SearchPipeline( optimizerself.query_optimizer, engineself.engine, parserself.parser, extractorself.extractor, ) logger.info(Camofox SearchPipeline initialized successfully.) except ImportError as e: logger.error(fFailed to import Camofox modules: {e}. Please check installation.) self._search_pipeline None except Exception as e: logger.error(fFailed to initialize search pipeline: {e}) self._search_pipeline None async def _run(self, query: str, max_results: int 5, **kwargs) - str: 工具的核心運(yùn)行方法。HermesAgent會(huì)調(diào)用此方法。 if not self._search_pipeline: return Error: Web search tool is not properly initialized. Check API keys and dependencies. logger.info(fExecuting web search for query: {query}) try: # 調(diào)用Camofox管道執(zhí)行搜索 # 注意Camofox的run方法可能是異步的也可能是同步的需根據(jù)其實(shí)際設(shè)計(jì)調(diào)整。 # 這里假設(shè)它是同步的。如果是異步的需要加上await。 search_results self._search_pipeline.run( queryquery, max_resultsmax_results, # 可以傳入其他上下文參數(shù)例如之前的對(duì)話歷史 contextkwargs.get(search_context) ) # 格式化結(jié)果使其對(duì)LLM友好 formatted_results self._format_results(search_results) logger.info(fSearch completed. Found {len(search_results)} results.) return formatted_results except Exception as e: logger.exception(fWeb search failed for query {query}: {e}) # 返回一個(gè)對(duì)Agent友好的錯(cuò)誤信息而不是拋出異常 return fI encountered an error while searching for {query}: {str(e)}. Please try again or rephrase your query. def _format_results(self, results: list) - str: 將Camofox返回的結(jié)構(gòu)化結(jié)果格式化為文本。 if not results: return No relevant web search results found. formatted_lines [**Web Search Results:**, ] for i, res in enumerate(results[:5], 1): # 確保不超過(guò)max_results title res.get(title, No Title) link res.get(link, #) snippet res.get(snippet, No snippet available.) # 可能還有提取的全文內(nèi)容 res.get(extracted_content) formatted_lines.append(f{i}. **[{title}]({link})**) formatted_lines.append(f {snippet}) formatted_lines.append() # 空行分隔 formatted_lines.append(*Search powered by Camofox Web Search.*) return \n.join(formatted_lines)這個(gè)WebSearchTool類(lèi)已經(jīng)具備了基本骨架。它定義了工具的名稱(chēng)、描述和參數(shù)并在初始化時(shí)嘗試構(gòu)建Camofox的搜索管道。_run方法是執(zhí)行搜索和返回結(jié)果的核心。3.3 配置與初始化讓工具“活”起來(lái)創(chuàng)建了工具類(lèi)我們還需要將它“注冊(cè)”到HermesAgent中。通常這需要在初始化Agent時(shí)將工具實(shí)例添加到工具列表里。首先確保你的API密鑰和搜索引擎ID已經(jīng)設(shè)置好。最佳實(shí)踐是使用環(huán)境變量。export GOOGLE_API_KEYyour_google_api_key_here export GOOGLE_SEARCH_ENGINE_IDyour_search_engine_id_here然后在你的主程序文件中比如main.py這樣初始化Agentimport asyncio import os from hermes_agent.agent import HermesAgent from camofox_provider import WebSearchTool # 導(dǎo)入我們剛寫(xiě)的工具 async def main(): # 1. 實(shí)例化我們的搜索工具 web_search_tool WebSearchTool( # 可以在這里覆蓋環(huán)境變量但更推薦用環(huán)境變量 # api_keyxxx, # search_engine_idxxx ) # 2. 準(zhǔn)備工具列表 tools [web_search_tool] # 3. 初始化HermesAgent并傳入工具 # 注意這里需要你已有LLM的配置如OpenAI API Key agent HermesAgent( modelgpt-4, # 或你使用的其他模型 toolstools, # ... 其他Agent配置參數(shù) ) # 4. 與Agent交互 response await agent.run(What are the main features of the latest version of Python?) print(response) if __name__ __main__: asyncio.run(main())現(xiàn)在當(dāng)你向Agent提問(wèn)一個(gè)需要最新信息的問(wèn)題時(shí)它應(yīng)該能自主決定調(diào)用web_search工具并返回包含真實(shí)鏈接的搜索結(jié)果。4. 超越基礎(chǔ)實(shí)現(xiàn)可控性與可觀測(cè)性的高級(jí)技巧基礎(chǔ)的集成只能算“跑通”。要讓這個(gè)搜索工具真正強(qiáng)大、可靠我們必須深入細(xì)節(jié)實(shí)現(xiàn)精細(xì)化的控制和全面的可觀測(cè)性。下面是我在實(shí)戰(zhàn)中總結(jié)的幾個(gè)關(guān)鍵技巧。4.1 查詢(xún)優(yōu)化策略讓Agent“更懂”搜索默認(rèn)情況下Agent生成的搜索查詢(xún)可能直接來(lái)自用戶問(wèn)題這并不總是最優(yōu)的。例如用戶問(wèn)“怎么用Python做數(shù)據(jù)分析”這個(gè)查詢(xún)過(guò)于寬泛。我們可以通過(guò)QueryOptimizer來(lái)改進(jìn)。Camofox允許我們自定義優(yōu)化器。一個(gè)簡(jiǎn)單的策略是關(guān)鍵詞提取與重組。我們可以集成一個(gè)輕量級(jí)的NLP庫(kù)如spacy或jieba來(lái)提取名詞實(shí)體和動(dòng)詞作為核心關(guān)鍵詞并附加上下文詞。# 示例一個(gè)簡(jiǎn)單的基于規(guī)則和上下文的查詢(xún)優(yōu)化器 from camofox.optimizers import BaseOptimizer class ContextAwareQueryOptimizer(BaseOptimizer): def __init__(self, stop_wordsNone): self.stop_words stop_words or {how, to, what, is, the, a, an} def optimize(self, query: str, context: Optional[str] None) - str: # 1. 分詞并去除停用詞 (這里用簡(jiǎn)單空格分割示意) words query.lower().split() key_terms [w for w in words if w not in self.stop_words] # 2. 如果有上下文從中提取可能相關(guān)的實(shí)體這是一個(gè)簡(jiǎn)化示例 additional_terms [] if context: # 假設(shè)上下文里提到了“pandas”和“visualization” if pandas in context.lower(): additional_terms.append(pandas) if plot in context.lower() or chart in context.lower(): additional_terms.append(data visualization) # 3. 組合關(guān)鍵詞避免重復(fù) all_terms list(dict.fromkeys(key_terms additional_terms)) optimized_query .join(all_terms) # 4. 如果優(yōu)化后為空回退到原查詢(xún) return optimized_query if optimized_query.strip() else query將這個(gè)優(yōu)化器注入到我們的SearchPipeline中搜索質(zhì)量會(huì)有顯著提升。更高級(jí)的策略還可以利用小語(yǔ)言模型SLM來(lái)重寫(xiě)查詢(xún)或者維護(hù)一個(gè)領(lǐng)域特定的同義詞庫(kù)。4.2 結(jié)果過(guò)濾與評(píng)分提升信息信噪比不是所有搜索結(jié)果都是有用的。我們需要一個(gè)過(guò)濾層。Camofox的架構(gòu)允許我們?cè)诮馕鼋Y(jié)果后、提取內(nèi)容前插入一個(gè)Filter組件。一個(gè)常見(jiàn)的過(guò)濾策略是基于來(lái)源可信度和內(nèi)容新鮮度。我們可以維護(hù)一個(gè)可信域名列表如*.github.io,stackoverflow.com,python.org并給來(lái)自這些域名的結(jié)果加分。同時(shí)解析結(jié)果中的日期信息優(yōu)先選擇一年內(nèi)的內(nèi)容。from camofox.filters import BaseFilter from datetime import datetime, timedelta import re class RelevanceFilter(BaseFilter): def __init__(self, trusted_domainsNone, max_age_days365): self.trusted_domains trusted_domains or [github.com, stackoverflow.com, docs.python.org] self.max_age timedelta(daysmax_age_days) def filter(self, results: list) - list: filtered_results [] for res in results: score 0.0 link res.get(link, ) # 1. 域名可信度評(píng)分 for domain in self.trusted_domains: if domain in link: score 1.0 break # 2. 新鮮度評(píng)分如果結(jié)果里有日期 date_str res.get(date) if date_str: try: # 嘗試解析各種日期格式這里簡(jiǎn)化處理 # 實(shí)際中可能需要更復(fù)雜的日期解析庫(kù) pub_date self._parse_date(date_str) if datetime.now() - pub_date self.max_age: score 0.5 except: pass # 3. 標(biāo)題/摘要關(guān)鍵詞匹配度簡(jiǎn)單實(shí)現(xiàn) text (res.get(title, ) res.get(snippet, )).lower() if tutorial in text: score 0.3 if example in text or sample in text: score 0.2 if score 0.5: # 設(shè)置一個(gè)閾值 res[relevance_score] score filtered_results.append(res) # 按評(píng)分排序 filtered_results.sort(keylambda x: x.get(relevance_score, 0), reverseTrue) return filtered_results def _parse_date(self, date_str: str) - datetime: # 簡(jiǎn)化的日期解析實(shí)際項(xiàng)目應(yīng)使用dateutil.parser for fmt in (%Y-%m-%d, %d %b %Y, %B %d, %Y): try: return datetime.strptime(date_str[:10], fmt) except ValueError: continue return datetime.now() - timedelta(days365*2) # 默認(rèn)返回一個(gè)很舊的日期將這個(gè)過(guò)濾器加入到管道中可以自動(dòng)屏蔽垃圾站點(diǎn)和過(guò)時(shí)信息讓Agent獲取到的信息質(zhì)量更高。4.3 全面的日志與監(jiān)控讓黑盒變透明可控性的前提是可觀測(cè)性。我們需要知道每一次搜索到底發(fā)生了什么。Camofox的管道設(shè)計(jì)通常支持中間步驟的鉤子hooks或事件監(jiān)聽(tīng)。我們可以為管道注入一個(gè)日志記錄器記錄下優(yōu)化后的查詢(xún)?cè)~、請(qǐng)求的URL、返回的狀態(tài)碼、解析出的結(jié)果數(shù)量、過(guò)濾后的數(shù)量等。import json from camofox.core import SearchPipeline class LoggingPipeline(SearchPipeline): def run(self, query, **kwargs): logger.info(f[SearchPipeline] Start. Original query: {query}) optimized_query self.optimizer.optimize(query, kwargs.get(context)) logger.info(f[SearchPipeline] Optimized query: {optimized_query}) raw_results self.engine.search(optimized_query, **kwargs) logger.info(f[SearchPipeline] Engine returned {len(raw_results) if raw_results else 0} raw items.) parsed_results self.parser.parse(raw_results) logger.info(f[SearchPipeline] Parser produced {len(parsed_results)} parsed results.) filtered_results self.filter.filter(parsed_results) if hasattr(self, filter) else parsed_results logger.info(f[SearchPipeline] After filtering: {len(filtered_results)} results.) # 可以選擇性地記錄前幾個(gè)結(jié)果的標(biāo)題和鏈接用于調(diào)試 for i, res in enumerate(filtered_results[:3]): logger.debug(f Result {i1}: {res.get(title)} - {res.get(link)}) final_results [] for res in filtered_results: if self.extractor: try: content self.extractor.extract(res[link]) res[extracted_content] content[:500] ... if len(content) 500 else content # 只存摘要 except Exception as e: logger.warning(fFailed to extract content from {res[link]}: {e}) res[extracted_content] final_results.append(res) logger.info(f[SearchPipeline] Finished. Returning {len(final_results)} final results.) return final_results使用這個(gè)LoggingPipeline替代基礎(chǔ)的SearchPipeline所有的搜索過(guò)程都將被清晰記錄。你還可以將日志發(fā)送到ELK?；虮O(jiān)控系統(tǒng)繪制搜索成功率、耗時(shí)、常用查詢(xún)?cè)~等圖表實(shí)現(xiàn)運(yùn)維級(jí)別的可控。4.4 錯(cuò)誤處理與降級(jí)策略構(gòu)建魯棒的服務(wù)網(wǎng)絡(luò)搜索充滿了不確定性API額度用盡、網(wǎng)絡(luò)超時(shí)、網(wǎng)站反爬、解析失敗等等。一個(gè)健壯的工具必須有完善的錯(cuò)誤處理和降級(jí)機(jī)制。重試機(jī)制對(duì)于網(wǎng)絡(luò)超時(shí)等臨時(shí)性錯(cuò)誤應(yīng)該自動(dòng)重試?？梢允褂胻enacity庫(kù)。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions class RobustSearchEngine(GoogleSearchEngine): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def search(self, query, **kwargs): return super().search(query, **kwargs)多引擎降級(jí)配置多個(gè)搜索引擎如Google API作為主引擎DuckDuckGo作為備用。當(dāng)主引擎失敗時(shí)自動(dòng)切換到備用引擎。from camofox.engines import DuckDuckGoEngine class FallbackSearchEngine: def __init__(self, primary_engine, fallback_engine): self.primary primary_engine self.fallback fallback_engine def search(self, query, **kwargs): try: return self.primary.search(query, **kwargs) except Exception as e: logger.error(fPrimary engine failed: {e}. Switching to fallback.) return self.fallback.search(query, **kwargs)優(yōu)雅的失敗響應(yīng)在_run方法中捕獲所有異常并返回一個(gè)對(duì)用戶和Agent都友好的提示而不是讓整個(gè)Agent會(huì)話崩潰。例如“搜索服務(wù)暫時(shí)不可用可能是網(wǎng)絡(luò)問(wèn)題或API限額已滿。您可以稍后重試或嘗試詢(xún)問(wèn)一些無(wú)需實(shí)時(shí)信息的問(wèn)題?！蓖ㄟ^(guò)實(shí)施這些高級(jí)技巧你的Web Search工具將從“一個(gè)能用的功能”升級(jí)為“一個(gè)可靠、可控、可觀測(cè)的核心服務(wù)組件”。這正是在生產(chǎn)環(huán)境中集成此類(lèi)工具所必需的深度考量。