工具错误处理

%pip install --upgrade --quiet langchain langchain-openai

import getpass
import os
 
os.environ["OPENAI_API_KEY"] = getpass.getpass()
 
# 如果您想使用LangSmith，请取消下面的注释：
# os.environ["LANGCHAIN_TRACING_V2"] = "true"
# os.environ["LANGCHAIN_API_KEY"] = getpass.getpass()

# 定义工具
from langchain_core.tools import tool
 
 
@tool
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:
    """使用复杂的工具做一些复杂的事情。"""
    return int_arg * float_arg

# 定义模型并绑定工具
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
model_with_tools = model.bind_tools(
    [complex_tool],
    tool_choice="complex_tool",
)

# 定义链式调用
from operator import itemgetter
 
from langchain.output_parsers import JsonOutputKeyToolsParser
from langchain_core.runnables import Runnable, RunnableLambda, RunnablePassthrough
 
chain = (
    model_with_tools
    | JsonOutputKeyToolsParser(key_name="complex_tool", first_tool_only=True)
    | complex_tool
)

chain.invoke(
    "使用复杂的工具。参数为5、2.1、空字典。不要忘记dict_arg"
)

ValidationError: 调用工具时出错。以下是引发的错误:

<class 'pydantic.v1.error_wrappers.ValidationError'>: 1个验证错误 for complex_toolSchemaSchema
dict_arg
  field required (type=value_error.missing)

from typing import Any
 
from langchain_core.runnables import RunnableConfig
 
 
def try_except_tool(tool_args: dict, config: RunnableConfig) -> Runnable:
    try:
        complex_tool.invoke(tool_args, config=config)
    except Exception as e:
        return f"使用以下参数调用工具：\n\n{tool_args}\n\n引发以下错误：\n\n{type(e)}: {e}"
 
 
chain = (
    model_with_tools
    | JsonOutputKeyToolsParser(key_name="complex_tool", first_tool_only=True)
    | try_except_tool
)

print(
    chain.invoke(
        "使用复杂的工具。参数为5、2.1、空字典。不要忘记dict_arg"
    )
)

使用以下参数调用工具：

{'int_arg': 5, 'float_arg': 2.1}

引发以下错误：

<class 'pydantic.v1.error_wrappers.ValidationError'>: 1个验证错误 for complex_toolSchemaSchema
dict_arg
  field required (type=value_error.missing)

chain = (
    model_with_tools
    | JsonOutputKeyToolsParser(key_name="complex_tool", first_tool_only=True)
    | complex_tool
)
better_model = ChatOpenAI(model="gpt-4-1106-preview", temperature=0).bind_tools(
    [complex_tool], tool_choice="complex_tool"
)
better_chain = (
    better_model
    | JsonOutputKeyToolsParser(key_name="complex_tool", first_tool_only=True)
    | complex_tool
)
 
chain_with_fallback = chain.with_fallbacks([better_chain])
chain_with_fallback.invoke(
    "使用复杂工具。参数为 5, 2.1, 空字典。不要忘记 dict_arg。"
)

import json
from typing import Any
 
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
 
 
class CustomToolException(Exception):
    """自定义 LangChain 工具异常。"""
 
    def __init__(self, tool_call: dict, exception: Exception) -> None:
        super().__init__()
        self.tool_call = tool_call
        self.exception = exception
 
 
def tool_custom_exception(tool_call: dict, config: RunnableConfig) -> Runnable:
    try:
        return complex_tool.invoke(tool_call["args"], config=config)
    except Exception as e:
        raise CustomToolException(tool_call, e)
 
 
def exception_to_messages(inputs: dict) -> dict:
    exception = inputs.pop("exception")
    tool_call = {
        "type": "function",
        "function": {
            "name": "complex_tool",
            "arguments": json.dumps(exception.tool_call["args"]),
        },
        "id": exception.tool_call["id"],
    }
 
    # 将历史消息添加到原始输入中，以便模型知道它在最后一次工具调用中犯了一个错误。
    messages = [
        AIMessage(content="", additional_kwargs={"tool_calls": [tool_call]}),
        ToolMessage(tool_call_id=tool_call["id"], content=str(exception.exception)),
        HumanMessage(
            content="上次工具调用引发了异常。请尝试使用更正的参数再次调用工具。"
        ),
    ]
    inputs["last_output"] = messages
    return inputs
 
 
# 在提示中添加一个 last_output MessagesPlaceholder，如果不传递它，它不会对提示产生任何影响，但是我们可以选择插入一个任意的消息列表到提示中，如果需要的话。我们将在重试时使用此功能插入错误消息。
prompt = ChatPromptTemplate.from_messages(
    [("human", "{input}"), MessagesPlaceholder("last_output", optional=True)]
)
chain = (
    prompt
    | model_with_tools
    | JsonOutputKeyToolsParser(
        key_name="complex_tool", return_id=True, first_tool_only=True
    )
    | tool_custom_exception
)
 
# 如果初始的链式调用失败，我们将使用传入的异常作为消息重新运行它。
self_correcting_chain = chain.with_fallbacks(
    [exception_to_messages | chain], exception_key="exception"
)

self_correcting_chain.invoke(
    {
        "input": "使用复杂工具。参数为 5, 2.1, 空字典。不要忘记 dict_arg"
    }
)