20.2 serapi.py源码逐行剖析

我们可以看一下Google查询的例子，在LangChain中有多种实现的方式。

如图20-5所示，在utilities的serpapi.py代码文件中实现了SerpAPIWrapper。

图20- 5 utilities的serpapi.py的SerpAPIWrapper

在langchain目录的serpapi.py代码中，从langchain.utilities.serpapi中导入SerpAPIWrapper。

1. """用于向后兼容性"""
2. from langchain.utilities.serpapi import SerpAPIWrapper
3. __all__ = ["SerpAPIWrapper"]

utilities目录的serpapi.py的SerpAPIWrapper代码实现：

Gavin大咖微信：NLP_Matrix_Space

1. class SerpAPIWrapper(BaseModel):
2. """ SerpAPI封装。
3. 要使用，应该安装google-search-results的python包，并使用API键设置环境变量SERPAPI_API_KEY，或者将SERPAPI_API_KEY作为命名参数传递给构造函数。
4. 示例:
5. .. code-block:: python
6. from langchain.utilities import SerpAPIWrapper
7. serpapi = SerpAPIWrapper()
8. """
9. ...
10. def run(self, query: str, **kwargs: Any) -> str:
11. """通过SerpAPI运行查询并解析结果."""
12. return self._process_response(self.results(query))
13. ...

以下是SerpAPIWrapper的一个使用示例，展示了使用LangChain 库中的代理和工具来实现自问自答，并进行搜索的示例。其中，定义了一个工具列表 tools，包含一个“Intermediate Answer”（“中间答案”）的工具。该工具的 func 属性指向了 search.run 方法，用于执行搜索操作，而search即是SerpAPIWrapper类的一个实例。

1. from langchain import OpenAI, SerpAPIWrapper
2. from langchain.agents import initialize_agent, Tool
3. from langchain.agents import AgentType
4. llm = OpenAI(temperature=0)
5. search = SerpAPIWrapper()
6. tools = [
7. Tool(
8. name="Intermediate Answer",
9. func=search.run,
10. description="useful for when you need to ask with search",
11. )
12. ]
13. self_ask_with_search = initialize_agent(
14. tools, llm, agent=AgentType.SELF_ASK_WITH_SEARCH, verbose=True
15. )
16. self_ask_with_search.run(
17. "What is the hometown of the reigning men's U.S. Open champion?"
18. )

这是utilities目录中serpapi.py部分的代码。接下来，让我们转到tools目录里面的google_search目录，看一个使用GoogleSearchRun、GoogleSearchResults工具进行Goolge查询的方式，如图20-6所示。Gavin大咖微信：NLP_Matrix_Space

图20- 6 tools代码目录

google_search的__init__.py的代码实现：

1. """谷歌搜索API工具包"""
2. from langchain.tools.google_search.tool import GoogleSearchResults, GoogleSearchRun
3. __all__ = ["GoogleSearchRun", "GoogleSearchResults"]

google_search的tool.py的GoogleSearchResults代码实现：

1. class GoogleSearchResults(BaseTool):
2. """查询Google Search API并获取json的工具."""
3. name = "Google Search Results JSON"
4. description = (
5. "A wrapper around Google Search. "
6. "Useful for when you need to answer questions about current events. "
7. "Input should be a search query. Output is a JSON array of the query results"
8. )
9. num_results: int = 4
10. api_wrapper: GoogleSearchAPIWrapper
11. def _run(
12. self,
13. query: str,
14. run_manager: Optional[CallbackManagerForToolRun] = None,
15. ) -> str:
16. """使用工具."""
17. return str(self.api_wrapper.results(query, self.num_results))
18. async def _arun(
19. self,
20. query: str,
21. run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
22. ) -> str:
23. """异步使用该工具."""
24. raise NotImplementedError("GoogleSearchRun does not support async")

GoogleSearchResults 是一个用于与 Google 搜索 API 进行交互的工具类。通过使用该工具类，可以方便地进行搜索查询并获取返回的 JSON 格式的结果。

GoogleSearchResults类的属性：

• name：工具的名称，设定为 "Google Search Results JSON"。这个名称描述了工具的主要功能，即获取 Google 搜索结果的 JSON 数据。
• description：工具的描述提供了更详细的说明。该工具在处理当前事件相关的问题时非常有用。用户需要提供一个搜索查询作为输入，并且工具会返回一个包含查询结果的 JSON 数组。
• num_results：定义了需要获取的搜索结果数量，默认为 4。
• api_wrapper：是一个 GoogleSearchAPIWrapper 类的实例，用于封装与 Google 搜索 API 的交互。

GoogleSearchResults类的方法：

• _arun 方法用于异步使用该工具。尽管该方法在代码中被定义，但它抛出了 NotImplementedError，表示该方法尚未实现。这意味着当前版本的GoogleSearchResults工具类不支持异步运行。
• _run 方法用于同步使用该工具。用户可以传入一个查询字符串 query 作为参数，以及一个可选的 run_manager 参数，用于管理工具运行过程的回调函数。在方法内部，工具类调用 api_wrapper 对象的 results 方法，传递查询字符串和结果数量，并将结果转换为字符串类型返回。Gavin大咖微信：NLP_Matrix_Space

上述代码中第19行，调用了api_wrapper.results方法，api_wrapper是一个GoogleSearchAPIWrapper类，使用Google进行查询的时候，它们的原理机制都是一样的。

utilities的google_search.py的GoogleSearchAPIWrapper代码实现：

1. class GoogleSearchAPIWrapper(BaseModel):
2. """谷歌搜索API封装器
3. ...
4. """
5. 
   
6. def results(self, query: str, num_results: int) -> List[Dict]:
7. """通过GoogleSearch运行查询并返回元数据
8. 参数:
9. query: 要搜索的查询
10. num_results:要返回的结果数
11. 返回:
12. 具有以下关键字的词典列表：
13. snippet-结果的描述。
14. title-结果的标题。
15. link-指向结果的链接。
16. """
17. metadata_results = []
18. results = self._google_search_results(query, num=num_results)
19. if len(results) == 0:
20. return [{"Result": "No good Google Search Result was found"}]
21. for result in results:
22. metadata_result = {
23. "title": result["title"],
24. "link": result["link"],
25. }
26. if "snippet" in result:
27. metadata_result["snippet"] = result["snippet"]
28. metadata_results.append(metadata_result)
29. return metadata_results

GoogleSearchAPIWrapper 是一个封装了谷歌搜索 API 的类，包含一个 results方法，用于执行搜索查询并返回结果的元数据，这是Google API的调用。results方法接受两个参数：query是要搜索的查询字符串。num_results是要返回的结果数量。首先，创建一个空的 metadata_results 列表，用于存储结果的元数据。然后，调用 _google_search_results方法执行查询，并将结果存储在results变量中。如果结果数量为零，返回一个包含文本“No good Google Search Result was found”的字典。接着，遍历结果列表，并为每个结果创建一个 metadata_result 字典。该字典包含以下关键字，“title”是结果的标题，“link”是指向结果的链接，“snippet”是结果的描述。然后，将每个metadata_result添加到metadata_results列表中，返回metadata_results列表。Gavin大咖微信：NLP_Matrix_Space

接下来，我们看另外一种使用Google进行查询的方式，使用GoogleSerperRun、GoogleSerperResults进行查询。如图20-7所示。

图20- 7 Google Serper的查询方式

google_serper的__init__.py的代码实现：

1. from langchain.tools.google_serper.tool import GoogleSerperResults, GoogleSerperRun
2. """Google Serper API工具包."""
3. """Serer.dev谷歌搜索API工具."""
4. __all__ = ["GoogleSerperRun", "GoogleSerperResults"]

google_serper的tool.py的代码实现：

1. class GoogleSerperResults(BaseTool):
2. """ 查询Serper.dev谷歌搜索API并获取json的工具."""
3. name = "google_serrper_results_json"
4. description = (
5. "A low-cost Google Search API."
6. "Useful for when you need to answer questions about current events."
7. "Input should be a search query. Output is a JSON object of the query results"
8. )
9. api_wrapper: GoogleSerperAPIWrapper = Field(default_factory=GoogleSerperAPIWrapper)
10. def _run(
11. self,
12. query: str,
13. run_manager: Optional[CallbackManagerForToolRun] = None,
14. ) -> str:
15. """使用工具."""
16. return str(self.api_wrapper.results(query))
17. 
    
18. async def _arun(
19. self,
20. query: str,
21. run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
22. ) -> str:
23. """异步使用该工具."""
24. return (await self.api_wrapper.aresults(query)).__str__()

GoogleSerperResults类继承自 BaseTool 类，用于通过调用 Serper.dev 谷歌搜索 API 并获取 JSON 格式的查询结果。

GoogleSerperResults类的属性：

• name：工具的名称为 "google_serrper_results_json"。
• description：工具的描述表明它是一个低成本的谷歌搜索 API，适用于回答与当前事件相关的问题。输入是一个搜索查询，输出是查询结果的 JSON 对象。
• api_wrapper：是一个 GoogleSerperAPIWrapper 类的实例。

GoogleSerperResults类的方法：Gavin大咖微信：NLP_Matrix_Space

• _run 方法：该方法用于执行工具的功能。它接受一个查询字符串作为参数，并返回一个字符串，表示查询结果的 JSON。这跟Google Search的方式是一致的。上述代码第18行，调用了api_wrapper.results方法。
• _arun方法：异步使用该工具。

utilities的google_serper.py的results代码实现：

1. class GoogleSerperAPIWrapper(BaseModel):
2. """Serper.dev谷歌搜索API的封装
3. ....
4. """
5. def results(self, query: str, **kwargs: Any) -> Dict:
6. """通过GoogleSearch运行查询"""
7. return self._google_serper_api_results(
8. query,
9. gl=self.gl,
10. hl=self.hl,
11. num=self.k,
12. tbs=self.tbs,
13. search_type=self.type,
14. **kwargs,
15. )

以上代码中第7行，调用的是内部的方法_google_serper_api_results。

utilities的google_serper.py的_google_serper_api_results代码实现：

1. class GoogleSerperAPIWrapper(BaseModel):
2. """Serper.dev谷歌搜索API的封装
3. ...
4. """
5. ...
6. def _google_serper_api_results(
7. self, search_term: str, search_type: str = "search", **kwargs: Any
8. ) -> dict:
9. headers = {
10. "X-API-KEY": self.serper_api_key or "",
11. "Content-Type": "application/json",
12. }
13. params = {
14. "q": search_term,
15. **{key: value for key, value in kwargs.items() if value is not None},
16. }
17. response = requests.post(
18. f"https://google.serper.dev/{search_type}", headers=headers, params=params
19. )
20. response.raise_for_status()
21. search_results = response.json()
22. return search_results
23. 
    
24. async def _async_google_serper_search_results(
25. self, search_term: str, search_type: str = "search", **kwargs: Any
26. ) -> dict:
27. headers = {
28. "X-API-KEY": self.serper_api_key or "",
29. "Content-Type": "application/json",
30. }
31. url = f"https://google.serper.dev/{search_type}"
32. params = {
33. "q": search_term,
34. **{key: value for key, value in kwargs.items() if value is not None},
35. }
36. if not self.aiosession:
37. async with aiohttp.ClientSession() as session:
38. async with session.post(
39. url, params=params, headers=headers, raise_for_status=False
40. ) as response:
41. search_results = await response.json()
42. else:
43. async with self.aiosession.post(
44. url, params=params, headers=headers, raise_for_status=True
45. ) as response:
46. search_results = await response.json()
47. return search_results

_google_serper_api_results函数，通过调用 Serper.dev 的谷歌搜索 API ，获取搜索结果。在进行Google网络搜索的时候，涉及到一些传输协议的内容。对于有基本开发经验的人来说，这应该是常识。从底层实现的角度，当请求是一个 HTTP 时，我们使用 POST 方法，然后，会得到一个响应 (response)，通常大家会以 JSON 的方式获取并处理结果。该函数接收一个搜索词 search_term 和一个可选的搜索类型 search_type（默认为 "search"）作为参数，创建一个包含请求头信息的 headers 字典，包括 API 密钥和内容类型。接着，创建了一个包含查询参数的params字典，其中包括搜索词和其他关键字参数，发起一个 POST 请求到google.serper.dev，并传入请求头和查询参数，然后检查响应的状态码，如果状态码表示请求成功，将响应的 JSON 数据解析为字典，最终返回搜索结果的字典。

当然，你也可以使用async的方式进行操作，使用_async_google_serper_search_results函数，它会用异步非阻塞的方式，其过程是一致的。使用async会涉及到一些概念，其底层的原理机制，实际上是基于协程 (Coroutine)，在这里类似于Future，在将来的某个时刻你会完成操作，并在那时接收结果。对于具备编程经验的人来说，基本上都会理解这种直观的运行机制，因为你想要非阻塞，但又无法预知将来何时完成操作，因此使用了“as response”，因为你不确定何时会获得结果，所以使用了“await”，这样当你获取到结果时，再进行赋值。从理解的角度来看，大家不会遇到太多难点。我们关注的重点是LangChain 框架的执行方式，这里只是简单地展示一下异步的运行原理，它本身并不会带来太多困难。