Coze的流式与非流式开发入门

一、认识Coze，也叫扣子

Coze是一个AI大模型智能体开发平台，也称为扣子。用户可以通过Coze平台创建AI机器人，也称为Bot，这些机器人可以用于多种场景，如AI客服、知识助理、学习规划等。Coze平台集成了插件、知识库、数据库、记忆和工作流等多种功能，使得即使是非技术用户也能轻松创建AI Bot。

Coze平台的特点包括：

1. 免费提供多种功能，尽管海外版和国内的专业版可能已经收费。
2. 适合新手上手，不需要编程基础。
3. 支持将创建的Bot发布到多个平台，如Coze商店、豆包、飞书、抖音、微信等。
4. 内置多个大模型，如豆包·Function call 32k、通义千问-Max 8k、智普 GLM-4 128k等，这些模型提供了不同的文本处理能力。
5. 创建的Bot支持电脑端和手机端使用，提高了用户体验。

此外，Coze平台还提供了插件功能，允许用户调用各种工具或技能，如搜索、查询等，以满足不同的业务需求。平台插件数量超过1万+，覆盖图像类、文本类、搜索类、数据分析、语音识别等多个领域。

Coze平台还支持自定义插件开发，用户可以根据自己的需求开发插件，以实现个性化的功能。通过Coze平台，用户可以创建、开发、发布和上架自定义插件，以满足特定的业务需求。

二、CozeApi开发

Coze关于bot如何配置请移步coze官方使用手册，本文不多做赘述，专注于开发层面的思路研究。

Coze的对话分为流式跟非流式，想要打字机效果的可以研究流式，想要直接出完整结果的，可以研究非流式。

1.流式

首先看官方实例代码：

curl --location --request POST 'https://api.coze.cn/v3/chat?conversation_id=7374752000116113452' \ --header 'Authorization: Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****' \ --header 'Content-Type: application/json' \ --data-raw '{ "bot_id": "734829333445931****", "user_id": "123456789", "stream": true, "auto_save_history":true, "additional_messages":[ { "role":"user", "content":"2024年10月1日是星期几", "content_type":"text" } ] }'

官方没有Python版本，但是不要紧，看到curl了，立马找到咱们网站导航里面的CURL命令转换工具，一键转换成Python版本。

import requests

headers = {
    'Authorization': 'Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****',
    'Content-Type': 'application/json',
}

params = {
    'conversation_id': '7374752000116113452',
}

body = {
    'bot_id': '734829333445931****',
    'user_id': '123456789',
    'stream': True,
    'auto_save_history': True,
    'additional_messages': [
        {
            'role': 'user',
            'content': '2024年10月1日是星期几',
            'content_type': 'text',
        },
    ],
}

response = requests.post('https://api.coze.cn/v3/chat', params=params, headers=headers, json=body)

将Authorization和bot_id换成自己，先运行一下看看，conversation_id按理说不应该出现在这里，但是示例有咱们先带着，毕竟首次运行肯定会报错，果不其然，不管打印response.json()还是text，都是报错。

首先删除刚刚咱们看不顺眼的conversation_id，打印text发现出结果了，但是json还是不行，text内容如下：

现在需要找出如何取流式消息的方法，发现response中有一个iter_lines()方法可以实现，但这个是个列表，需要for循环一下取到里面的内容。

response = requests.post('https://api.coze.cn/v3/chat', headers=headers, json=json_data)
for line in response.iter_lines():
    print(line)

直接打印发现单独取出的line中，有空字节字符串，这里判断一下当line存在在获取，字节字符串需要转化为字符串，所以将line进行decode一下。

response = requests.post('https://api.coze.cn/v3/chat', headers=headers, json=json_data)
for line in response.iter_lines():
    if line:
        lineStr = line.decode('utf-8')
        print(lineStr)

这时候发现内容非常杂乱，并且一行字符串一行类似json格式代码的数据，现在我们要找我们目标数据，也就是包含content的数据，然后发现type为answer的才是我们需要的内容，但是看似json数据的前面还有个data，这样会导致直接json.loads失败，所以需要split一下。

response = requests.post('https://api.coze.cn/v3/chat', headers=headers, json=json_data)
for line in response.iter_lines():
    if line:
        lineStr = line.decode('utf-8')
        if 'content' in lineStr:
            jsData = json.loads(lineStr.split(':',1)[1])
            if jsData['type'] == "answer":
                print(jsData['content'])

结果成功输出，这里发现最后一个content是一个完整的答案，需要自己在做处理，这里就不多做赘述。

2.非流式

这里我们就不看官方实例代码了，因为官方给的都挺扯的，没一点参考价值。

非流式根据文档需要做到三步：

• 请求一个发起对话api，发起一个对话
• 请求一个查看对话详情api，检测对话是否生成完毕
• 请求一个查看对话消息详情api，拿到最终生成结果

这里发现，三个api的请求headers是一样的，请求body和params略有不同。两个查看的api都共同用到了两个参数conversation_id和chat_id，而这两个参数是发起对话Api返回中的数据，也就是说需要先发起对话获取这两个参数，然后再去查看对话详情Api做一个轮询，直到检测到某个状态表示对话生成完毕，再去请求查看对话消息详情Api获取最终内容。

headers

headers = {
    'Authorization': 'Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****',
    'Content-Type': 'application/json',
}

body

body = {
    'bot_id': '734829333445931****',
    'user_id': '123456789',
    'stream': False,
    'auto_save_history': True,
}

params

params = {
    'conversation_id': '*****',
    'chat_id': '******'
}

发起对话Api

api: https://api.coze.cn/v3/chat
type: post

发起对话因为需要初始带用户问题，所以需要再body的基础上额外添加部分内容，内容如下：

import requests

headers = {
    'Authorization': 'Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****',
    'Content-Type': 'application/json',
}

body = {
    'bot_id': '734829333445931****',
    'user_id': '123456789',
    'stream': False,
    'auto_save_history': True,
}

additional_messages = [
    {
        "role": "user",
        "content": '用户问题',
        "content_type": "text"
    }
]

body['additional_messages'] = additional_messages

response = requests.post('https://api.coze.cn/v3/chat', headers=headers, json=body)
data = response.json()
conversation_id = data['data']['conversation_id']
chat_id = data['data']['id']

拿到conversation_id和chat_id后，调用传输给查看详情Api

查看对话详情Api

api: https://api.coze.cn/v3/chat/retrieve
type: get

这里原本需要构造成一个函数供其他调用，单独讲解就不构造了，最后会有完整构造的代码。

import requests
import time

headers = {
    'Authorization': 'Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****',
    'Content-Type': 'application/json',
}

body = {
    'bot_id': '734829333445931****',
    'user_id': '123456789',
    'stream': False,
    'auto_save_history': True,
}

params = {
    'conversation_id': '*****',
    'chat_id': '******'
}

while True:
    response = requests.get('https://api.coze.cn/v3/chat/retrieve', params=params, headers=headers, json=body)
    data = response.json()
    if data['data']['status'] == 'completed':
        print("生成完毕")
    time.sleep(1)

因为这个是监控某个状态，所以需要一直轮询做请求，根据官方说法，每次请求需要间隔一秒，这点很重要，关乎是否会触发频繁并发导致请求失败。成功调用查看对话详情Api既可获取内容成功。

查看对话消息详情Api

api:  https://api.coze.cn/v3/chat/message/list
type: get

import requests
import time

headers = {
    'Authorization': 'Bearer pat_OYDacMzM3WyOWV3Dtj2bHRMymzxP****',
    'Content-Type': 'application/json',
}

body = {
    'bot_id': '734829333445931****',
    'user_id': '123456789',
    'stream': False,
    'auto_save_history': True,
}

params = {
    'conversation_id': '*****',
    'chat_id': '******'
}

response = requests.get(' https://api.coze.cn/v3/chat/message/list', params=params, headers=headers, json=body)
data = response.json()
contents = data['data']
if contents[0]['content'] is not None:
    print(contents[0]['content'])

以上就是非流式所有的过程，三个api需要自己构造成出函数来做逻辑调用。

PS:流式调用和非流式调用中有一个点很重要，就是body中有一个参数叫stream，设置为True，则是流式，为False，为非流式，在使用代码时自己需要注意区分。

三、完整代码