把域名解析到其他网站的ip下载百度到桌面上

兜底降级-LangChain调用工具Tool异常处理

ToolException调用工具Tool异常处理介绍

• 智能体（Agent）在调用外部的Tool（APi）的时候会遇到不可预知的错误，那么如果说这个错误被LLM接收得到的话，那么大模型就会返回给用户一个不可预知的一个答案。（eg：报错一堆未编译代码等等）

  • 列如如下的场景
    • 网络请求失败（调用API无响应）
    • 权限调用不足（Api-Key错误）
    • 输入的参数不符合规范（如格式的错误）
    • 调用失败（调用次数的用尽）

• 解决的方式（ToolException）

  • 通过ToolException去捕获和处理工具的调用错误
  • ToolException的核心作用
    • 统一错误格式：将不同工具异常转换为标准的一个格式，方便智能体解析
    • 错误上下文传递：保留错误的原因，工具名称等关键的信息

案例<(￣︶￣)↗[GO!]

• 方法一：配置handle_tool_error=True默认是False

# 模拟搜索的工具API调用失败所造成的结果为空的一个情况处理
from langchain_core.tools import ToolException, StructuredTooldef search(query: str) -> str:"""执行搜索查询"""# 引发一个ToolException来模拟搜索结果为空的一个情况raise ToolException(f"相关搜索数据为空：{query}")# 自定义工具异常处理函数
def _handle_tool_error(error: ToolException) -> str:"""处理ToolException异常"""return f"搜索结果为空,自定义失败请重试：{error}"search_tool = StructuredTool.from_function(func=search,name="搜索引擎工具",handle_tool_error=True, # 开启# handle_tool_error=_handle_tool_error # 自定义异常处理函数去处理捕获到的error# handle_tool_error="搜索结果失败，请重试", # 开启,表示工具会处理内部的异常
)result = search_tool.invoke("腾讯的股票今天是多少？")print(result)

整体的一个LLM + Tool的整体方案代码

from langchain_community.utilities import SearchApiAPIWrapper
from langchain_core.tools import tool,StructuredTool, BaseTool
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import ToolMessage
import osos.environ["ALY_API-KEY"] = "sk-XXX"
os.environ["SEARCH_API_KEY"] = "XXX"# 创建一个搜索对象
search = SearchApiAPIWrapper(searchapi_api_key=os.environ["SEARCH_API_KEY"],engine="bing",
)# 切割工具
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=20)@tool("web_search", return_direct=True)
def web_search(query : str) -> str:"""当需要获取实时信息，最新事件或未知领域知识时候调用输入应为搜索关键词"""try:results = search.results(query) # 获取前三条数据return "\n\n".join([f"来源：{res['title']} \n内容：{res['snippet']}"for res in results['organic_results']])except Exception as e:return f"搜索结果失败：{str(e)}"# 用run的话# try:#     result = search_tool.run(query)#     return result# except Exception as e:#     return f"搜索结果失败：{str(e)}"# 创建一个工具对象
@tool
def multiply(a: int, b: int) -> int:"""Multiply two numbers together."""return a * b# 创建一个LLM对象
llm = ChatOpenAI(model_name="qwen-plus", base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key=os.environ.get("ALY_API-KEY"),temperature=0.7,
)# 创建聊天的模板
prompt = ChatPromptTemplate.from_messages([("system", "你是一个AI助手，名字叫关沐吖，请根据用户输入的查询问题，必要的时候可以调用工具去帮助用户解答"),("human", "{query}"),]
)# 定义字典工具
tool_dict = {"multiply": multiply,"web_search": web_search,
}# 从字典工具提取工具列表
tools = [tool_dict[tool_name] for tool_name in tool_dict]# 绑定工具到llm中
llm_with_tools = llm.bind_tools(tools)# 构建运行链e
chain = {"query":RunnablePassthrough()} | prompt | llm_with_tools# 问题
query="距离今日日本核污水排放情况如何了？"resp = chain.invoke({"query": query})
# print(resp)# 判断是否需要调用工具, content=为空的时候不一定需要调用工具而是根据tool_calls去判断的
tool_calls = resp.tool_callsif len(tool_calls)<=0:print("不需要调用工具")
else:# 合并历史消息history_messages = prompt.invoke(query).to_messages()history_messages.append(resp)print(history_messages)# 循环工具调用for tool_call in tool_calls:tool_name = tool_call.get("name")tool_args = tool_call.get("args")tool_response = tool_dict[tool_name].invoke(tool_args) tool_response_message = ToolMessage( tool_call_id=tool_call.get("id"), content=tool_response,name=tool_name)print(f"一次调用工具：{tool_name},参数：{tool_args}，结果：{tool_response}")history_messages.append(tool_response_message)print(f"历史全部消息：{history_messages}")result = llm_with_tools.invoke(history_messages)
print(f"最终结果：{result}")
print(f"工具调用结果：{result.content}")# 得到对应的答案# query = "距离今日日本的核污水事件怎么样了"
# 两种处理的方式
# search.result() 提取到的是元数据不是整理完之后的数据的
# result = search.run("今天日本的核污水排放怎么样了？")