使用 OpenAI 和 Langchain 通过对话直接调用函数

1. 大模型与 Langchain

很多人可能没有机会训练、甚至微调大模型，但对大模型的使用却是未来趋势。那么，我们应该如何拥抱这一变化呢？答案就是 Langchain。

大模型提供的是一种泛而通用的基础能力，目前，我看到的有两种主要落地方式：

基于生成能力的 AIGC，各种剧本、代码、二维码、短视频、分子结构层出不穷
基于理解能力的 AutoGPT，结合执行引擎，直接修改机器状态，进行自动化控制

目前，在我们的工作场景中，大模型常常还不足以大量直接替代人的工作。原因有两个：

很多私有的数据，没有提供给大模型进行训练。需要微调、结合知识库之后，才能达到较好的效果。这是大模型的基础定位决定的
大模型出来的时间还不够长，ToB 效率工具类服务是以十年为周期的，市场还没有形成

大模型的落地是必然，现在大模型是我们的 Copilot，以后我们可能就是大模型的 Copilot。结合大模型、贴合业务场景，开发出一些 Copilot 工具、提升效率，是我最近在思考的问题。

开发应用少不了各种框架。Langchain 的定位就是提供开发大模型应用的框架，以解决大模型落地过程中的一些通用问题。比如，对接多种大模型，Prompt 管理，上下文，外部文档加载，向量库对接，Chains 任务链等。

Langchain 已经由之前的个人项目转为商业公司运作，2023 年还进行了多轮融资。从中可以看到，创投行业对基于大模型的应用开发是非常看好的。既然投资人已经帮我们做出了判断，我们只需要多学习和使用 Langchain 即可。

2. 对话直接调用函数

在 2023 年 6 月份，OpenAI 和 Langchain 相继发布了版本，支持直接调用函数。这意味着，大模型不仅仅可以用来聊天，还可以用来触发一些业务逻辑。

2.1 先看看效果

执行程序

交互测试

manual_input:获取 default 这个命名空间的全部 pod
function_bot: ["pod1", "pod2", "pod3"]
manual_input:获取 c1 这个集群的全部节点
function_bot: ["node1", "node2", "node3"]

输入自然语言，自动执行函数，并返回结果。这里举了两个例子，一个是获取 default 命名空间下全部 Pod，一个是获取 c1 集群下全部节点。

2.2 代码实现

设置环境变量

1
2

export OPENAI_API_BASE="https://api.openai.com/v1"
export OPENAI_API_KEY="xxx"

在使用 OpenAI API 时，会自动读取环境变量中设置的 API KEY。

完整代码

# -*- coding: utf-8 -*-
import json
from typing import Type
from pydantic import BaseModel, Field, create_model
from typing import Optional
from langchain.tools import BaseTool
from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain.tools import format_tool_to_openai_function
import openai


class GetClusterNodes(BaseTool):
    name: str = "get_cluster_nodes"
    description: str = "get all nodes in kubernetes cluster"

    args_schema: Type[BaseModel] = create_model(
        "GetClusterNodesArgs",
        cluster=(str, Field(
            description="the cluster of you want to query", type="string")),
    )

    def _run(
        self, query: str,
        run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        return json.dumps(["node1", "node2", "node3"])

    async def _arun(
        self, query: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        return json.dumps(["node1", "node2", "node3"])


class GetClusterPodsByNamespaces(BaseTool):
    name: str = "get_cluster_pods_by_namespace"
    description: str = "get special pods in kubernetes special namespace"

    args_schema: Type[BaseModel] = create_model(
        "GetClusterPodsByNamespacesArgs",
        namespace=(str, Field(
            description="the namespace of you want to query", type="string")),
    )

    def _run(
        self, query: str,
        run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        return json.dumps(["pod1", "pod2", "pod3"])

    async def _arun(
        self, query: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        return json.dumps(["pod1", "pod2", "pod3"])


functions_list: list = [GetClusterNodes, GetClusterPodsByNamespaces]
functions_map: dict = {fun().name: fun for fun in functions_list}


def run(msg: str):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": msg}],
        functions=[
            format_tool_to_openai_function(t()) for t in functions_list],
        function_call="auto",
    )
    message = response["choices"][0]["message"]
    if message.get("function_call"):
        function_name = message["function_call"]["name"]
        function_response = functions_map[function_name]().run(
            message["function_call"]["arguments"])
        return function_response

if __name__ == "__main__":
    while True:
        user_input = input("manual_input:")

        if user_input == "exit":
            break

        print("function_bot:", run(user_input))

这里为了简化实现，_run 都直接进行了返回，没有实际调用函数。在实际生产中，我们需要去根据输入的 query，调用函数，返回结果。

2.3 逐步解析

核心代码

def run(msg: str):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": msg}],
        functions=[
            format_tool_to_openai_function(t()) for t in functions_list],
        function_call="auto",
    )
    message = response["choices"][0]["message"]
    if message.get("function_call"):
        function_name = message["function_call"]["name"]
        function_response = functions_map[function_name]().run(
            message["function_call"]["arguments"])
        return function_response

在 openai.ChatCompletion.create 设置两个参数:

function_call 设置为 auto，默认即 auto，由模型自己决定是否调用函数。这里并不是真的调用，而是返回一些函数元信息。

functions 是一个列表对象，OpenAI 根据传入的函数描述加用户输入的内容 msg 做出判断，返回函数的名字、获取到的参数。

自定义函数

有两种写法的定义: 一种是直接使用列表对象拼接，一种是继承 BaseTool 实现。

下面是列表对象拼接:

[
  {
    "name": "get_cluster_nodes",
    "description": "get all nodes in kubernetes cluster"
  },
  {
    "name": "get_cluster_pods_by_namespace",
    "description": "get special pods in kubernetes special namespace",
    "parameters": {
      "type": "object",
      "properties": {
        "namespace": {
          "type": "string",
          "description": "filter pods in namespace"
        }
      },
      "required": ["namespace"]
    }
  }
]

上面的完整示例代码中使用的就是继承 BaseTool 实现:

1
2

class GetClusterNodes(BaseTool):
class GetClusterPodsByNamespaces(BaseTool):

继承 BaseTool 的方式其实最终还是需要使用 format_tool_to_openai_function 提取自定义函数中的信息生成一个列表，但使用 BaseTool 管理函数方法是一个更加清晰的方式。

函数参数定义非常重要

如果不详细描述参数，那么 OpenAI 识别到的参数格式很有可能是这样的:

"function_call": {
    "name": "get_cluster_nodes",
    "arguments": "{\n\"__arg1\": \"c1\"\n}"
}

而如果设置了 properties 或者 args_schema 之后，OpenAI 返回的函数参数就非常符合预期了。

"function_call": {
    "name": "get_cluster_nodes",
    "arguments": "{\n\"cluster\": \"c1\"\n}"
}

哪里实现具体业务逻辑

如果使用直接拼接列表的形式，那么直接写在函数即可。如果继承 BaseTool，那么就需要实现其同步调用 _run 函数，异步调用 _arun 函数。

上面的完整示例代码中:

1
2

function_response = functions_map[function_name]().run(
            message["function_call"]["arguments"])

直接将返回的参数，传给被调用的函数，这里调用的就是 _run 函数。在 _run 函数中，通过 json.loads(query) 可以获取到符合 args_schema 定义的参数。

[可选]通过 OpenAI 再次整理消息响应

def format(msg: str, function_response: str):
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "user", "content": msg},
            {
                "role": "function",
                "name": "get_cluster_nodes",
                "content": function_response,
            },
        ],
    )["choices"][0]["message"]["content"]

代码如上，在获取到函数的响应之后，如果还需要对响应的格式、内容进行二次整理，可以设置一个 function 角色的消息，附加上函数的响应，加上用户的输入，一起发送给 OpenAI。此时，OpenAI 会给出一个更加完整的响应。

但这步不是必须，如果函数的响应已经符合预期，那么可以直接返回。

3. 总结

本篇主要是借助 OpenAI 和 Langchain 实现了一个直接使用自然语言调用函数的示例。

大模型不仅仅可以用来聊天，还可以用来触发一些业务逻辑。在我们开发 Copilot 时，经常需要这种胶水功能，粘合大模型和业务逻辑。

大模型生态的建设有两部分，一个是认知，一个是执行。认知依赖于大模型的参数规模、网络结构、训练数据；执行主要依赖于外部连接的情况。

我认为，即使参与不了大模型的训练，也可以尝试着整理一下行业知识库，还有机会参与到执行部分。围绕执行我们可以将产品的 API 开放出来，增加大模型连接系统的触点；还可以开发一些 SDK、工具包，帮助开发者快速接入大模型，比如整理一个 BaseTool 类库，封装各种 API 、脚本功能；当然，还可以根据大模型的思考方式，重新设计业务流程、执行逻辑。

推荐订阅源

陈少文的网站