LLM与LangChain整合：探索OpenAI Agent Function Calling的实现原理_convert_to_openai_tool

OpenAI函数调用的实现原理

OpenAI的函数调用功能依靠大型模型（LLM）的能力与外部工具或API的连接。

1. 首先，大型模型会接收用户输入，并根据输入内容判断何时需要调用外部函数。一旦确定需要调用函数，模型会生成一个包含所需调用函数信息和参数信息的请求消息。
2. 其次，模型会根据目标函数的描述生成符合要求的请求参数。这些参数可能直接从用户输入中提取，也可能是模型根据输入内容推断出来的。
3. 然后，生成的函数请求和相应参数会被传递给外部工具或API。外部工具或API接收到请求后，执行相应函数，并将结果返回给模型。
4. 最后，模型将外部工具或API返回的结果以适当方式展示给用户，例如通过自然语言回复用户或在特定应用界面上展示结果。

在判断何时调用外部函数时，大型模型可能采用以下方法：

1. 自然语言理解：通过分析输入文本中的词汇、语法、语义等信息，模型可以识别出是否包含特定任务、指令或需求，从而需要调用外部函数来实现。
2. 意图识别：模型会尝试理解用户的意图，例如，如果用户输入是询问产品价格，模型可能会识别出这是一个查询意图，并调用相应函数获取产品价格信息。
3. 上下文分析：模型会考虑用户之前的输入和对话历史，通过分析上下文，更好地理解用户的当前需求，并判断是否需要调用外部函数提供更准确或相关信息。
4. 模式匹配：模型可能会根据先前的学习经验，识别出输入内容中特定的模式或结构，从而触发模型调用相应的外部函数。
5. 置信度阈值：模型可能会评估自身生成回答的置信度。如果对某个问题的回答不够自信（即置信度低于某个阈值），模型可能会选择调用外部函数获取更准确或更可靠的信息。

LangChain函数调用

很多聊天模型（例如 OpenAI）具有函数调用 API，可让您描述函数及其参数，并让模型返回一个 JSON 对象，其中包含要调用的函数以及该函数的输入。函数调用对于构建使用工具的链和代理以及更普遍地从模型中获取结构化输出非常有用。

LangChain 附带了许多实用程序来简化函数调用。也就是说，它带有

• 将函数绑定到模型的简单语法
• 用于将各种类型的对象格式化为预期函数模式的转换器
• 用于从 API 响应中提取函数调用的输出解析器

要了解输出解析的工作原理，请查看OpenAI Tools 输出解析器。

定义函数

这里重点关注OpenAI 函数格式，OpenAI 是支持函数调用的主要模型提供者。LangChain 有一个内置的转换器，可以将 Python 函数、Pydantic 类和 LangChain Tools 转换为 OpenAI 函数格式。

Python function

import json
 
from langchain_core.utils.function_calling import convert_to_openai_tool
 
 
def multiply(a: int, b: int) -> int:
    """Multiply two integers together.
    Args:
        a: First integer
        b: Second integer
    """
    return a * b
 
 
print(json.dumps(convert_to_openai_tool(multiply), indent=2))

{
  "type": "function",
  "function": {
    "name": "multiply",
    "description": "Multiply two integers together.",
    "parameters": {
      "type": "object",
      "properties": {
        "a": {
          "type": "integer",
          "description": "First integer"
        },
        "b": {
          "type": "integer",
          "description": "Second integer"
        }
      },
      "required": [
        "a",
        "b"
      ]
    }
  }
} 

Pydantic class

from langchain_core.pydantic_v1 import BaseModel, Field
 
 
class multiply(BaseModel):
    """Multiply two integers together."""
 
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")
 
 
print(json.dumps(convert_to_openai_tool(multiply), indent=2))

 {
  "type": "function",
  "function": {
    "name": "multiply",
    "description": "Multiply two integers together.",
    "parameters": {
      "type": "object",
      "properties": {
        "a": {
          "description": "First integer",
          "type": "integer"
        },
        "b": {
          "description": "Second integer",
          "type": "integer"
        }
      },
      "required": [
        "a",
        "b"
      ]
    }
  }
}

LangChain Tool

from typing import Any, Type
 
from langchain_core.tools import BaseTool
 
 
class MultiplySchema(BaseModel):
    """Multiply tool schema."""
 
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")
 
 
class Multiply(BaseTool):
    args_schema: Type[BaseModel] = MultiplySchema
    name: str = "multiply"
    description: str = "Multiply two integers together."
 
    def _run(self, a: int, b: int, **kwargs: Any) -> Any:
        return a * b
 
 
# Note: we're passing in a Multiply object not the class itself.
print(json.dumps(convert_to_openai_tool(Multiply()), indent=2))

{
  "type": "function",
  "function": {
    "name": "multiply",
    "description": "Multiply two integers together.",
    "parameters": {
      "type": "object",
      "properties": {
        "a": {
          "description": "First integer",
          "type": "integer"
        },
        "b": {
          "description": "Second integer",
          "type": "integer"
        }
      },
      "required": [
        "a",
        "b"
      ]
    }
  }
} 

Binding functions

现在我们已经定义了一个函数，我们希望将其传递到我们的模型中。

from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-3.5-turbo")
llm.invoke("what's 5 times three", tools=[convert_to_openai_tool(multiply)])
 
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_JvOu9oUwMrQHiDekZTbpNCHY', 'function': {'arguments': '{\n  "a": 5,\n  "b": 3\n}', 'name': 'multiply'}, 'type': 'function'}]})

如果我们希望每次调用该工具时都传递该函数，则可以将其绑定到该工具：

llm_with_tool = llm.bind(tools=[convert_to_openai_tool(multiply)])
llm_with_tool.invoke("what's 5 times three")
 
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_cwRoTnD1ux1SnWXLrTj2KlWH', 'function': {'arguments': '{\n  "a": 5,\n  "b": 3\n}', 'name': 'multiply'}, 'type': 'function'}]})

我们还可以强制使用tool_choice参数调用工具 。

llm_with_tool = llm.bind(
    tools=[convert_to_openai_tool(multiply)],
    tool_choice={"type": "function", "function": {"name": "multiply"}},
)
llm_with_tool.invoke(
    "don't answer my question. no do answer my question. no don't. what's five times four"
)
 
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_sWjLyioSZAtYMQRLMTzncz1v', 'function': {'arguments': '{\n  "a": 5,\n  "b": 4\n}', 'name': 'multiply'}, 'type': 'function'}]})

ChatOpenAI 类甚至附带了一个辅助bind_tools函数，用于将类似函数的对象转换为 OpenAI 格式并为您绑定它们：

llm_with_tool = llm.bind_tools([multiply], tool_choice="multiply")
llm_with_tool.invoke("what's 5 times three")
 
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_LCdBa4cbhMJPRdtkhDzpRh7x', 'function': {'arguments': '{\n  "a": 5,\n  "b": 3\n}', 'name': 'multiply'}, 'type': 'function'}]})

输出解析与工具

• 输出解析：请参阅OpenAI Tools 输出解析器 和OpenAI Functions 输出解析器 ，了解如何将调用 API 响应的函数提取为各种格式。
• 工具使用：在这些指南中了解如何构建实际调用所调用工具的链和代理。

Function calling流程总结的步骤如下：

1. 自定义函数：根据用户需求，自定义函数chen_ming_algorithm，用于处理特定的任务。
2. 创建字典：根据自定义函数，创建一个字典chen_ming_function，其中包含自定义函数的信息。
3. 创建functions参数：将多个包含不同函数字典的list放入一个名为functions的参数中，以便后续使用。
4. 判断是否需要调用外部函数：根据用户提供的messages和functions参数，判断是否需要调用外部函数。

• 如果需要调用外部函数，执行以下操作：
  • 输出assistant_message_content：根据function_call提供的信息挑选合适的函数。
  • 将function call中的函数参数信息传递给函数。
  • 获取函数返回结果function_response。
  • 将输出结果包装为一条function message，添加到messages1中。
  • 将messages1和assistant_message合并成messages2。
  • 将messages2发送给外部函数库，获取second_response。
  • 根据second_response生成Final message，作为最终输出结果。
• 如果不需 要调用外部函数，将messages直接发送给模型，得到second_response。
  • 根据second_response生成Final message，作为最终输出结果。

用于构建Agent的function calling流程

1. **主函数get_chat_response的流程：**
   - 输入用户需求messages的对象。
   - 判断是否是text_response。
     - 如果是text_response：
       - 创建text_answer_message。
       - 返回text_answer_message。
     - 如果不是text_response：
       - 创建function_call_message。
       - 返回function_call_message。

2. **创建text answer message的流程：**
   - 输入text answer message。
   - 通过is_text_response_valid函数判断是否是开发模式。
     - 如果是开发模式：
       - 引导用户进行审查。
       - 执行修改步骤（根据用户的反馈进行修改）。
     - 如果不是开发模式：
       - 执行不修改步骤。
       - 返回new_messages。

3. **创建function call message的流程：**
   - 输入function call message。
   - 通过is_code_response_valid函数判断解析是否成功。
     - 如果解析失败：
       - 返回修改后的messages。
     - 如果解析成功：
       - 打印模型编写代码。
       - 通过is_text_response_valid函数判断是否是开发模式。
         - 如果是开发模式：
           - 引导用户进行审查。
           - 执行修改步骤（根据用户的反馈进行修改）。
         - 如果不是开发模式：
           - 执行不修改步骤。
           - 返回new_messages。

4. **创建function call message流程（继续）：**
   - 通过check_get_final_function_response函数解析JSON类型的函数参数。
   - 将function_call_message加入到初始messages中。
   - 判断function_response_message是否包含报错信息。
     - 如果包含报错信息：
       - 在增强模式下启动深度debug模式。
         - 创建多轮for循环，引导模型进行深度debug。
         - 在原始messages中添加prompt，并将messages带入get_chat_response并做出回答。
       - 在其他模式下启动有效debug模式。
         - 创建单次for循环，引导模型进行有效debug。
         - 在原始messages中添加prompt，并将messages带入get_chat_response并做出回答。
     - 如果不包含报错信息：
       - 返回new_messages。

这个流程描述了如何根据用户输入的消息来创建和返回不同的类型的消息，包括文本回答消息和函数调用消息。同时，它还描述了在开发模式下如何引导用户进行审查和修改，以及在出现报错时如何启动不同的debug模式来解决问题。