文档备案控制台登录立即注册
首页文章问答视频活动下载资源开发者计划团队号镜像站
发布

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

陈少文
陈少文
开源
火山方舟大模型智能内容
  1. 大模型与 Langchain

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

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

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

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

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

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

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

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

  1. 对话直接调用函数

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

2.1 先看看效果

  • 执行程序

        
          
python function_bot.py
  • 交互测试

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

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

2.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 实现:


        
        
            

          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 函数。

上面的完整示例代码中:


        
          
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 会给出一个更加完整的响应。

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

  1. 总结

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

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

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

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

0
0
0
0
关于作者
陈少文
陈少文
关于作者
陈少文
陈少文

文章

0

获赞

0

收藏

0

相关资源
字节跳动大数据容器化构建与落地实践
随着字节跳动旗下业务的快速发展，数据急剧膨胀，原有的大数据架构在面临日趋复杂的业务需求时逐渐显现疲态。而伴随着大数据架构向云原生演进的行业趋势，字节跳动也对大数据体系进行了云原生改造。本次分享将详细介绍字节跳动大数据容器化的演进与实践。
点击下载 
相关产品
推荐阅读
VS Code又又出新版本1.89了，新姿势用起来真是YYDS的快
Jetbrains Fleet 又爆出了一个王炸！！！
Goland 2025.2 迎来重大升级，又整新活！
从0死磕全栈第八天：使用nest.js五分钟搭建后端开发环境
Gin 实现统一异常处理和封装统一返回结果
评论
未登录
看完啦，登录分享一下感受吧～
暂无评论