Coze插件开发教程：从零开发一个调用Azure的GPT-4o接口插件

发表于 分类于 ， ，

前言

目的

COZE不能接入gpt-4o，因此需要自己开发一个插件来调用gpt-4o的api，Azure是一个很好的选择。

因此，我打算利用COZE的插件来调用Azure的gpt-4o的api，来实现这个功能。

目标

开发一个可以调用gpt-4o的插件。
下面就是记录开发的全过程，并附上代码。

插件工具创建模式

基于已有服务创建，也就是提供api地址和认证方式来让coze调用已经开发的服务。

开发语言

python

技术栈

fastapi为框架
使用serivce的授权方式
通过http的方式请求Azure的api

插件开发步骤

创建conda环境

1
conda create -n useAzureGPT python=3.10

安装所需的包

1
2
pip install fastapi uvicorn python-multipart
pip install httpx

编写代码

新建main.py文件，并编辑代码。
代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from fastapi import FastAPI, Depends, HTTPException, Request, Form

from gpt import Gpt4o
from pydantic import BaseModel
from typing import Optional, List, Dict

# 创建FastAPI实例
app = FastAPI()

# 用于作为service验证用
TOKEN = os.getenv("API_TOKEN", "default_token")

# 调用Azure api所需参数
AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY", "default_api_key")
AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT", "https://default.endpoint")
DEPLOYMENT_NAME = "GPT-4o"
MODEL_VERSION = "2024-02-15-preview"
MODEL_NAME = "gpt-4o"

# 验证token的函数
async def verify_token(request: Request):
    token = request.headers.get('token')
    if token != TOKEN:
        raise HTTPException(status_code=401, detail="Invalid or missing token")

# 定义请求体模型，就是可以接收的参数
class GPTRequest(BaseModel):
    prompt: List[Dict] # 请求所需的prompt
    system: Optional[str] = None # 设置system的prompt
    max_tokens: Optional[int] = 4000
    temperature: Optional[float] = 0.7
    top_p: Optional[float] = 0.95
    frequency_penalty: Optional[float] = 0
    presence_penalty: Optional[float] = 0

# 调用LLMs的API
@app.post('/apis/gpt4o', dependencies=[Depends(verify_token)])
async def call_gpt4o_api(request: GPTRequest):
    gpt = Gpt4o(AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, DEPLOYMENT_NAME, MODEL_VERSION)
    response = gpt.call_gpt4o_api(
        request.prompt,
        request.system,
        request.max_tokens,
        request.temperature,
        request.top_p,
        request.frequency_penalty,
        request.presence_penalty
    )
    return {"response": response}

新建gpt4o.py文件，并编辑
代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import httpx
import json

class Gpt4o():
    def __init__(self, endpoint, api_key, model_name, model_version):
        self.endpoint = endpoint
        self.api_key = api_key
        self.model_name = model_name
        self.model_version = model_version

    def call_gpt4o_api(self, messages_json, system=None, max_tokens=4000, temperature=0.7, top_p=0.95, frequency_penalty=0, presence_penalty=0):
        url = f"{self.endpoint}openai/deployments/{self.model_name}/chat/completions?api-version={self.model_version}"

        headers = {
            "Content-Type": "application/json",
            "api-key": self.api_key
        }

        # 如果 system 等于 None，设置默认值
        if system is None:
            system = "You are an AI assistant that helps people find information."

        system_data = {
            "role": "system",
            "content": [
                {
                    "type": "text",
                    "text": system
                }
            ]
        }

        # 将 system_data 和 messages_json 合并到 messages 列表中
        messages = [system_data] + messages_json

        payload = {
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "top_p": top_p,
            "frequency_penalty": frequency_penalty,
            "presence_penalty": presence_penalty,
        }

        timeout = httpx.Timeout(None)

        try:
            response = httpx.post(url, headers=headers, json=payload, timeout=timeout)
            response.raise_for_status()  # 检查请求是否成功
            res = response.json()
            return res['choices'][0]['message']['content']
        except httpx.RequestError as exc:
            print(f"An error occurred while requesting {exc.request.url!r}.")
            return ""
        except httpx.HTTPStatusError as exc:
            print(f"Error response {exc.response.status_code} while requesting {exc.request.url!r}.")
            return ""

测试

启动应用

1
2
# 启动命令
uvicorn main:app --reload

image.png

调用本地接口

调用工具：我这里使用apipost，你也可以是哟postman等顺手的工具
调用参数：
url：http://127.0.0.1:8000/apis/gpt4o
Header
token：用于认证的token值，main.py中的TOKEN值
Content-Type：application/json
Body
选择类型
image.png
提交信息示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
    "prompt": [
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "图片中有几个人?在干什么？"
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "http://xxxxxxcms/content/202407/5a1265428e91f.png"
                    }
                }
            ]
        }
    ]
}

具体的请求参数如何可以参考openai的文档中gpt-4o的请求说明

返回信息
image.png
本地测试完毕

部署

我使用的是宝塔面板，因此部署也是基于宝塔的部署方式
使用的宝塔版本是免费版9.1.0
image.png

创建文件夹并上传文件

创建文件夹用于放置python文件
image.png
上传main.py和gpt4o.py两个文件。到刚刚创建的文件夹

创建requirements.txt文件

用于安装必要的模块

1
2
3
uvicorn==0.30.3
fastapi==0.111.1
httpx==0.27.0

添加一个python项目

在宝塔界面点击网站

image.png

再选择python项目

image.png

选择python版本，使用和本地环境相同的python版本

image.png

点击添加python项目

image.png

填写后

image.png

自定义命令

1
gunicorn -k uvicorn.workers.UvicornWorker main:app --bind 127.0.0.1:8000

确认后会先安装所需模块，安装完成后，状态变为运行中

image.png

设置项目信息

现在项目启动了，但是也只能内网访问，因此我们需要配置成外网访问。

先添加域名

image.png

启动外网映射

image.png

配置反向代理信息

image.png

系统会自动生成配置信息

测试项目

使用apipost

URL：https://xxxx(域名)/apis/gpt4o

请求参数与本地相同

image.png

能够返回正常的内容。

添加到coze插件

新建插件

image.png

选择基于已有服务撞见

插件URL：使用刚刚创建项目的url

授权方式使用service

image.png

apikey，就是main.py中的TOKEN的值

创建工具

填写基本信息

image.png

配置参数

先配置两个基本参数测试

prompt

按照测试内容的格式添加

image.png

system

类型是string

配置输出参数

image.png

填写测试参数来查看调试结果

image.png

运行后的调试结果

查看request

image.png

查看response

image.png

如果成功了，就点击完成

测试成功后点击发布按钮以便在工作流或bots中调用

测试coze

创建一个工作流来测试插件节点

工作流具体如何设置这里就不细说了，总之需要一个代码节点组合请求的参数

代码如下：

这里我使用的是python代码，使用javascript也可以，道理是一样的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import Optional, List, Dict
async def main(args: Args) -> Output:
    params = args.params

    prompt = params["prompt"]
    image = params["image"]
    system = "作为一个图片分析助手，帮我分析图片中的各种信息和细节。"

    messages:List[Dict] = [
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": prompt
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": image
                    }
                }
            ]
        }
    ]


    ret: Output = {
        "prompt": messages,
        "system": system,
    }
    return ret

结果如下：

image.png

再添加一个刚刚的插件节点

输入参数就引用刚刚代码节点处理后的结果

点击试运行

image.png

image.png

回答正常

在bots中测试

编写好调用prompt后

先添加图片

image.png

image.png

image.png

image.png

成功！！！

参考文档

https://platform.openai.com/docs/api-reference/chat/create

https://learn.microsoft.com/zh-cn/azure/ai-services/openai/how-to/migration?tabs=python-new%2Cdalle-fix

https://github.com/openai/openai-python