泉州网站的建设,笑傲网站建设,如何在亚马逊上开网店,网站媒体作风建设年工作总结文章目录 BaseTool 源码分析核心属性以 TavilySearchResults(BaseTool) 为例namedescriptionargs_schemaresponse_format查询选项属性 需要子类实现的抽象方法以 TavilySearchResults(BaseTool) 为例 核心方法arun()#xff1a;run()的异步执行版本invoke()和ainvoke() BaseTo… 文章目录 BaseTool 源码分析核心属性以 TavilySearchResults(BaseTool) 为例namedescriptionargs_schemaresponse_format查询选项属性 需要子类实现的抽象方法以 TavilySearchResults(BaseTool) 为例 核心方法arun()run()的异步执行版本invoke()和ainvoke() BaseTool 源码分析
BaseTool 是 LangChain 框架中定义 tools 的模板类
核心属性
name表示 tool 唯一名称的字符串用于识别description对如何 / 何时 / 为何使用该 tool 的描述帮助模型决定什么时候调用该 toolargs_schema验证工具输入参数的 Pydantic model 或 schemareturn_direct如果为True则立即返回 tool 的输出responcse_format定义 tool 的响应格式
以 TavilySearchResults(BaseTool) 为例
name
name: str tavily_search_results_jsondescription description: str (A search engine optimized for comprehensive, accurate, and trusted results. Useful for when you need to answer questions about current events. Input should be a search query.)args_schema
class TavilyInput(BaseModel):Input for the Tavily tool.query: str Field(descriptionsearch query to look up)# 输入将遵循 TavilyInput 类中定义的架构规则
# 同时args_schema的值必须是BaseModel派生类
args_schema: Type[BaseModel] TavilyInput按照TavilyInput的规则如果输入没有提供query值将抛出一个验证错误Field函数用于向字段添加元数据描述
response_format
response_format: Literal[content_and_artifact] content_and_artifact使用 Literal 来确保某些值被限制为特定文字
_LiteralSpecialForm
_tp_cache(typedTrue)
def Literal(self, *parameters):Special typing form to define literal types (a.k.a. value types).This form can be used to indicate to type checkers that the correspondingvariable or function parameter has a value equivalent to the providedliteral (or one of several literals):def validate_simple(data: Any) - Literal[True]: # always returns True...MODE Literal[r, rb, w, wb]def open_helper(file: str, mode: MODE) - str:...open_helper(/some/path, r) # Passes type checkopen_helper(/other/path, typo) # Error in type checkerLiteral[...] cannot be subclassed. At runtime, an arbitrary valueis allowed as type argument to Literal[...], but type checkers mayimpose restrictions.# There is no _type_check call because arguments to Literal[...] are# values, not types.parameters _flatten_literal_params(parameters)try:parameters tuple(p for p, _ in _deduplicate(list(_value_and_type_iter(parameters))))except TypeError: # unhashable parameterspassreturn _LiteralGenericAlias(self, parameters)查询选项属性
**max_results返回的最大结果数量默认为 5。**search_depth查询的深度可以是 basic 或 advanced默认是 advanced。**include_domains一个包含在结果中的域名列表默认为空即包含所有域名。exclude_domains一个排除在结果之外的域名列表。include_answer是否在结果中包含简短答案默认值为 False。include_raw_content是否返回 HTML 原始内容的解析结果默认关闭。include_images是否在结果中包含相关图片默认值为 False。
需要子类实现的抽象方法 abstractmethoddef _run(self, *args: Any, **kwargs: Any) - Any:Use the tool.Add run_manager: Optional[CallbackManagerForToolRun] Noneto child implementations to enable tracing.以 TavilySearchResults(BaseTool) 为例
api_wrapper: TavilySearchAPIWrapper Field(default_factoryTavilySearchAPIWrapper) # type: ignore[arg-type]api_wrapper 是一个 TavilySearchAPIWrapper 实例用于封装 API 调用的细节
class TavilySearchAPIWrapper(BaseModel):Wrapper for Tavily Search API.tavily_api_key: SecretStrmodel_config ConfigDict(extraforbid,)model_validator(modebefore)classmethoddef validate_environment(cls, values: Dict) - Any:Validate that api key and endpoint exists in environment.tavily_api_key get_from_dict_or_env(values, tavily_api_key, TAVILY_API_KEY)values[tavily_api_key] tavily_api_keyreturn valuesdef raw_results(self,query: str,max_results: Optional[int] 5,search_depth: Optional[str] advanced,include_domains: Optional[List[str]] [],exclude_domains: Optional[List[str]] [],include_answer: Optional[bool] False,include_raw_content: Optional[bool] False,include_images: Optional[bool] False,) - Dict:params {api_key: self.tavily_api_key.get_secret_value(),query: query,max_results: max_results,search_depth: search_depth,include_domains: include_domains,exclude_domains: exclude_domains,include_answer: include_answer,include_raw_content: include_raw_content,include_images: include_images,}response requests.post(# type: ignoref{TAVILY_API_URL}/search,jsonparams,)response.raise_for_status()return response.json()def results(self,query: str,max_results: Optional[int] 5,search_depth: Optional[str] advanced,include_domains: Optional[List[str]] [],exclude_domains: Optional[List[str]] [],include_answer: Optional[bool] False,include_raw_content: Optional[bool] False,include_images: Optional[bool] False,) - List[Dict]:Run query through Tavily Search and return metadata.Args:query: The query to search for.max_results: The maximum number of results to return.search_depth: The depth of the search. Can be basic or advanced.include_domains: A list of domains to include in the search.exclude_domains: A list of domains to exclude from the search.include_answer: Whether to include the answer in the results.include_raw_content: Whether to include the raw content in the results.include_images: Whether to include images in the results.Returns:query: The query that was searched for.follow_up_questions: A list of follow up questions.response_time: The response time of the query.answer: The answer to the query.images: A list of images.results: A list of dictionaries containing the results:title: The title of the result.url: The url of the result.content: The content of the result.score: The score of the result.raw_content: The raw content of the result.raw_search_results self.raw_results(query,max_resultsmax_results,search_depthsearch_depth,include_domainsinclude_domains,exclude_domainsexclude_domains,include_answerinclude_answer,include_raw_contentinclude_raw_content,include_imagesinclude_images,)return self.clean_results(raw_search_results[results])async def raw_results_async(self,query: str,max_results: Optional[int] 5,search_depth: Optional[str] advanced,include_domains: Optional[List[str]] [],exclude_domains: Optional[List[str]] [],include_answer: Optional[bool] False,include_raw_content: Optional[bool] False,include_images: Optional[bool] False,) - Dict:Get results from the Tavily Search API asynchronously.# Function to perform the API callasync def fetch() - str:params {api_key: self.tavily_api_key.get_secret_value(),query: query,max_results: max_results,search_depth: search_depth,include_domains: include_domains,exclude_domains: exclude_domains,include_answer: include_answer,include_raw_content: include_raw_content,include_images: include_images,}async with aiohttp.ClientSession() as session:async with session.post(f{TAVILY_API_URL}/search, jsonparams) as res:if res.status 200:data await res.text()return dataelse:raise Exception(fError {res.status}: {res.reason})results_json_str await fetch()return json.loads(results_json_str)async def results_async(self,query: str,max_results: Optional[int] 5,search_depth: Optional[str] advanced,include_domains: Optional[List[str]] [],exclude_domains: Optional[List[str]] [],include_answer: Optional[bool] False,include_raw_content: Optional[bool] False,include_images: Optional[bool] False,) - List[Dict]:results_json await self.raw_results_async(queryquery,max_resultsmax_results,search_depthsearch_depth,include_domainsinclude_domains,exclude_domainsexclude_domains,include_answerinclude_answer,include_raw_contentinclude_raw_content,include_imagesinclude_images,)return self.clean_results(results_json[results])def clean_results(self, results: List[Dict]) - List[Dict]:Clean results from Tavily Search API.clean_results []for result in results:clean_results.append({url: result[url],content: result[content],})return clean_resultsraw_results()同步调用 API。raw_results_async()异步调用 API。clean_results()清理和格式化查询结果。 def _run(self,query: str,run_manager: Optional[CallbackManagerForToolRun] None,) - Tuple[Union[List[Dict[str, str]], str], Dict]:Use the tool.# TODO: remove try/except, should be handled by BaseTooltry:raw_results self.api_wrapper.raw_results(query,self.max_results,self.search_depth,self.include_domains,self.exclude_domains,self.include_answer,self.include_raw_content,self.include_images,)except Exception as e:return repr(e), {}return self.api_wrapper.clean_results(raw_results[results]), raw_results传入查询参数调用 TavilySearchAPIWrapper 来获取结果。如果查询失败则返回错误信息。
核心方法
arun()run()的异步执行版本 async def _arun(self, *args: Any, **kwargs: Any) - Any:Use the tool asynchronously.Add run_manager: Optional[AsyncCallbackManagerForToolRun] Noneto child implementations to enable tracing.if kwargs.get(run_manager) and signature(self._run).parameters.get(run_manager):kwargs[run_manager] kwargs[run_manager].get_sync()return await run_in_executor(None, self._run, *args, **kwargs)若具有run_manager参数则转换为同步版本然后使用默认执行器异步运行 self._run 方法run_in_executor 是一个异步执行器它允许你在不同的执行器中运行同步代码而不会阻塞当前的事件循环
invoke()和ainvoke()
def invoke(self,input: Union[str, dict, ToolCall],config: Optional[RunnableConfig] None,**kwargs: Any,
) - Any:tool_input, kwargs _prep_run_args(input, config, **kwargs)return self.run(tool_input, **kwargs)async def ainvoke(self,input: Union[str, dict, ToolCall],config: Optional[RunnableConfig] None,**kwargs: Any,
) - Any:tool_input, kwargs _prep_run_args(input, config, **kwargs)return await self.arun(tool_input, **kwargs)充当执行工具逻辑的入口点准备输入参数并在内部调用run()或arun()
