1.2.1 WebResearchRetriever

首先是代码中最重要的一个封装类：WebResearchRetriever。

它的使用方式如下：

python代码解读复制代码# Initialize

web_research_retriever = WebResearchRetriever.from_llm(

vectorstore=vectorstore, llm=llm, search=search

)

接收三个主要参数：

向量数据库：用来存储网页数据llm检索引擎，这里的检索引擎 必须是 Google Search API

python代码解读复制代码class WebResearchRetriever(BaseRetriever):

"""`Google Search API` retriever."""

search: GoogleSearchAPIWrapper = Field(..., description="Google Search API Wrapper")code>

再看下其构造过程：from_llm 函数

python代码解读复制代码def from_llm(

cls,

vectorstore: VectorStore,

llm: BaseLLM,

search: GoogleSearchAPIWrapper,

prompt: Optional[BasePromptTemplate] = None,

num_search_results: int = 1,

text_splitter: RecursiveCharacterTextSplitter = RecursiveCharacterTextSplitter(

chunk_size=1500, chunk_overlap=150

),

) -> "WebResearchRetriever":

"""Initialize from llm using default template.

Args:

vectorstore: Vector store for storing web pages

llm: llm for search question generation

search: GoogleSearchAPIWrapper

prompt: prompt to generating search questions

num_search_results: Number of pages per Google search

text_splitter: Text splitter for splitting web pages into chunks

Returns:

WebResearchRetriever

"""

if not prompt:

QUESTION_PROMPT_SELECTOR = ConditionalPromptSelector(

default_prompt=DEFAULT_SEARCH_PROMPT,

conditionals=[

(lambda llm: isinstance(llm, LlamaCpp), DEFAULT_LLAMA_SEARCH_PROMPT)

],

)

prompt = QUESTION_PROMPT_SELECTOR.get_prompt(llm)

# Use chat model prompt

llm_chain = LLMChain(

llm=llm,

prompt=prompt,

output_parser=QuestionListOutputParser(),

)

return cls(

vectorstore=vectorstore,

llm_chain=llm_chain,

search=search,

num_search_results=num_search_results,

text_splitter=text_splitter,

)

这个函数用来初始化 WebResearchRetriever，除了上面说的三个主要参数外，其额外提供了默认的Prompt模板，text_splitter，QuestionListOutputParser等Retriever过程所需的工具和内容。

默认的Prompt模板内容如下：

python代码解读复制代码DEFAULT_SEARCH_PROMPT = PromptTemplate(

input_variables=["question"],

template="""You are an assistant tasked with improving Google search \code>

results. Generate THREE Google search queries that are similar to \

this question. The output should be a numbered list of questions and each \

should have a question mark at the end: {question}""",

)

从这个Prompt大致可以看出WebResearchRetriever的工作过程：

（1）根据用户的问题，利用大模型将该问题转化为3个与用户问题相近的Google搜索语句

（2）利用 Google CSE 搜索这几个问题，会得到一系列相关 URL

（3）利用上篇文章我们爬取网页内容的方法，将每个URL中的文本抓取出来

（4）对抓取出来的文本进行分块，向量存储（WebResearchRetriever的工作到这里就结束了）

（5）然后就是其它模块使用RAG的流程：用户提问 —> 查询向量数据库 —> 大模型回答问题

整体流程示意图如下：

（1）-（4）步骤的源码如下，可以对照着看一下：

<code>python代码解读复制代码def _get_relevant_documents(

self,

query: str,

*,

run_manager: CallbackManagerForRetrieverRun,

) -> List[Document]:

"""Search Google for documents related to the query input.

Args:

query: user query

Returns:

Relevant documents from all various urls.

"""

# Get search questions

logger.info("Generating questions for Google Search ...")

result = self.llm_chain({ "question": query})

logger.info(f"Questions for Google Search (raw): { result}")

questions = result["text"]

logger.info(f"Questions for Google Search: { questions}")

# Get urls

logger.info("Searching for relevant urls...")

urls_to_look = []

for query in questions:

# Google search

search_results = self.search_tool(query, self.num_search_results)

logger.info("Searching for relevant urls...")

logger.info(f"Search results: { search_results}")

for res in search_results:

if res.get("link", None):

urls_to_look.append(res["link"])

# Relevant urls

urls = set(urls_to_look)

# Check for any new urls that we have not processed

new_urls = list(urls.difference(self.url_database))

logger.info(f"New URLs to load: { new_urls}")

# Load, split, and add new urls to vectorstore

if new_urls:

loader = AsyncHtmlLoader(new_urls, ignore_load_errors=True)

html2text = Html2TextTransformer()

logger.info("Indexing new urls...")

docs = loader.load()

docs = list(html2text.transform_documents(docs))

docs = self.text_splitter.split_documents(docs)

self.vectorstore.add_documents(docs)

self.url_database.extend(new_urls)

# Search for relevant splits

# TODO: make this async

logger.info("Grabbing most relevant splits from urls...")

docs = []

for query in questions:

docs.extend(self.vectorstore.similarity_search(query))

# Get unique docs

unique_documents_dict = {

(doc.page_content, tuple(sorted(doc.metadata.items()))): doc for doc in docs

}

unique_documents = list(unique_documents_dict.values())

return unique_documents

1.2.2 GoogleSearchAPIWrapper