本文主要是梳理我自己的实现过程(使用的是python),各位朋友看了图一乐就行。(关于API使用的前期工作我就不讲了)
内容导读
- 使用/工作流程
- 具体代码实现
- 源码(不想看文章的可以直接使用莫得关系)
1.使用/工作流程
我使用的是“对话”这个接口,通过阅读官方文档可以知道要实现“携带上下文”大致分为5个步骤(第一个步骤要不要看自己需要就行)。
- 01 使用 列表 准备好传给大模型的上下文
-
02 通过会话接口创建一个会话,将准备好的上下文传过去,并记录响应回来的会话 ID。
- 03 通过对话接口,向服务器发起请求,将你现在要问的内容和上面记录的会话ID一起传递过去,并记录新的会话ID与对话ID。
- 04 使用查看对话详情的接口,向服务器发起轮询(官方指导间隔时间为1s/次我是2s/次),获取大模型的处理进度(不然直接获取可能会导致信息缺失)。
- 05 使用查看对话消息详情的接口来获取最终结果。
2.具体代码实现
01.1 使用 列表 按照文档要求准备好上下文history_list。
#历史对话
history_list = [
{
"role":"user",
"content":'你好我是荣析木',
"content_type":"text"
},
]
02.1使用 字典 构造请求头(这个请求头是每个接口都可以用的)和请求体,并将 01.1 中的上下文history_list嵌入到键'messages'中。
#请求头(这个请求头是全部接口都可以使用的)
headers={
'Authorization':'Bearer pat_cWordjRcjozvC0Dt4exxxxxxxxxxxxxxxxxxx',#使用自己令牌
'Content-Type':'application/json'
}
#dialogue会话的请求体
dia_data={
'meta_data':{#这个没什么必要自己设置就行
'uuid':'12344',
},
'messages':history_list
}
02.2 向创建会话的接口发送请求获取会话ID并记录到dia_id里。
# 不要忘记导入请求模块啊歪
import requests
url_dialogue='https://api.coze.cn/v1/conversation/create'#会话API(创建会话)
#dialogue会话请求
dia = requests.post(url=url_dialogue, json=dia_data, headers=headers)
#这里是链式调用使用先使用.json()将响应的JSON内容解析为Python字典,在通过get()方法取值,因为id嵌套在data里所以get两次。
dia_id = dia.json().get('data',{}).get('id', '没有取到会话id')
03.1 使用字典构造对话的请求体data,将 01.1 中的上下文history_list嵌入到"additional_messages"中,并将上一步的会话IDdia_id与对话(V3)的接口进行拼接得到url_temp,然后获取用户的输入content并将其按照特定格式加到history_list里面。
url='https://api.coze.cn/v3/chat'#对话API
# 构造对话的请求体(使用官方文档按需构造)
data={
'bot_id':'739021380000000',#输入自己机器人的id
'user_id':'1234',# 这里的user_id是必填的参数,但可以自己任意取值(我用的是时间戳)
'stream':False,
"auto_save_history":True,
"additional_messages":history_list
}
# 将上面得到的会话ID与对话(V3)的接口进行拼接
url_temp = url + '?' + 'conversation_id:' + dia_id
# 输入你想问的问题
content = input('内容:')
# 将你的问题按照下面的格式添加到history_list上下文的后面
history_list.append({"role":"user","content":content,"content_type":"text"})
03.2 使用拼接过后的对话(V3)接口url_temp与请求体data向服务器发起请求并记入新的会话IDconversation_id与对话IDchat_id
import requests
# 向服务器发起请求,请求头可以直接使用最上面的就行
response = requests.post(url=url_temp,json=data,headers=headers)
#获取会话ID
conversation_id = response.json().get('data', {}).get('conversation_id', '没有取到会话id')#测试一下与前面的dia_id是不一样的
#获取对话ID
chat_id = response.json().get('data', {}).get('id', '没有取到chat_id')
04.1 使用03.2获得的会话IDconversation_id与对话IDchat_id构造GET参数params,并向查看对话详情接口发起轮询,将完成状态存入state_data(当响应data里的status=completed时代表大模型处理完成)。
import requests
import time #时间模块
# 查看对话详情接口(获取状态)
url_state='https://api.coze.cn/v3/chat/retrieve'
#轮询间隔时间
wait = 2
#轮询
while True:
params={# 构造GET请求的参数
'conversation_id':conversation_id,
'chat_id':chat_id,
}
# 向查看对话详情接口发起请求
state = requests.get(url=url_state, params=params, headers=headers)
try:# 避免错误导致程序直接崩(这种情况一般是在短时间内多次访问被限流了)
state_error = state.json().get('data', {}).get('last_error', None)
if(state_error is not None):
print(f"错误码:--{state_error['code']}--信息:--{state_error['msg']}")
break # 跳出循环
except:
pass
# 获取模型的完成状态
state_data = state.json().get('data', {}).get('status', '服务器状态异常')
# 暂停2秒
time.sleep(wait)
05.1 确认完成状态state_data当值为'completed'时使用上一步的params参数,向查看对话消息详情接口发起请求来获取大模型的最终响应,并存入info_data。
import requests
# 查看对话消息详情接口
url_info = 'https://api.coze.cn/v3/chat/message/list'
# 当state_data=='completed'时则代表可以去获取信息了
if(state_data=='completed'):
# 向对话消息详情接口发起请求
info = requests.get(url=url_info, params=params, headers=headers)
info_data = info.json().get('data', '没有取到东西哦!')
05.2 遍历info_data内的元素将'type'键的值为'answer'的元素中的内容按需打印,并将特定内容存入到上下文history_list中。
# 遍历info_data里的元素判断其中的'type'是否为'answer'(这个是回答内容的标识)
for item in info_data:
if (item['type'] == 'answer'):
temp_info = item
# 将回答内容存入history_list上下文中保持模型的记忆,并将结果打印出来
history_list.append({"role":temp_info['role'],"content":temp_info['content'],"content_type":temp_info['content_type']})
# 打印大模型回答的内容
print(f"回答:{temp_info['content']}")
3.源码展示(要个赞不过分吧)
'''
# Author: 荣鑫宇
# Update: 2024-08-20
# Content: 扣子对话API携带上下文的使用
# Version: 0.0.2
'''
import requests
import time
import sys
'''-----前期准备工作-----'''
#历史对话
history_list = [
{
"role":"user",
"content":'你好我是荣析木',
"content_type":"text"
},
]
#请求头(这个请求头是全部的接口都可以使用的)
headers={
'Authorization':'Bearer pat_cWordjRcjozvC0xxxxxxxxxxxxxxxxxxxxxxxxxxxxx',# 扣子的令牌
'Content-Type':'application/json'
}
#dialogue会话的请求体
dia_data={
'meta_data':{# 这个uuid有没有关系不大
'uuid':'12344',
},
'messages':history_list
}
uid = int(time.time())#时间戳id
# 构造对话的请求体(使用官方文档按需构造)
data={
'bot_id':'73902000000000',# 自己bot的id
'user_id':str(uid),# 注意这里的user_id是必填的参数并且要使用str()方法进行转换,但可以自己任意取值(我用的是时间戳)
'stream':False,
"auto_save_history":True,
"additional_messages":history_list
}
#对话API
url='https://api.coze.cn/v3/chat'
#会话接口(创建会话)
url_dialogue='https://api.coze.cn/v1/conversation/create'
# 查看对话详情接口(获取状态)
url_state='https://api.coze.cn/v3/chat/retrieve'
# 查看对话消息详情接口
url_info = 'https://api.coze.cn/v3/chat/message/list'
'''-----创建会话-----'''
#dialogue会话请求
dia=requests.post(url=url_dialogue, json=dia_data, headers=headers)
dia_id = dia.json().get('data',{}).get('id', str('没有取到会话id'))#使用get方法没有取到id时不抛出keys Error
print(f'---dialogue会话请求获得的dia_id:{dia_id}---\n')
'''-----创建对话-----'''
# 将上面得到的会话ID与对话(V3)的接口进行拼接
url_temp = url + '?' + 'conversation_id:' + dia_id
while True:
print(f'---上下文history_list:{history_list}---\n')
# 输入你想问的问题
content = input('问题内容:')
# 将你的问题按照下面的格式添加到history_list上下文的后面
history_list.append({"role":"user","content":content,"content_type":"text"})
print(f'---会话ID与对话(V3)的接口的拼接{url_temp}---\n')
# 向服务器发起请求,请求头可以直接使用最上面的就行
response = requests.post(url=url_temp,json=data,headers=headers)
sys.stdout.write('=--V3的响应'+response.text+'--='+'\n')
#获取会话ID
conversation_id = response.json().get('data', {}).get('conversation_id', '没有取到会话id')#测试过与前面的dia_id是不一样的
#获取对话ID
chat_id = response.json().get('data', {}).get('id', '没有取到chat_id')
print(f'---对话(V3)的接口获得的conversation_id会话ID:{conversation_id}--chat_id对话ID{chat_id}---\n')
'''-----轮询获取状态-----'''
#轮询间隔时间
wait = 2
#轮询
while True:
print('---发起轮询---')#
params={# 构造GET请求的参数
'conversation_id':conversation_id,
'chat_id':chat_id,
}
# 向查看对话详情接口发起请求
state = requests.get(url=url_state, params=params, headers=headers)
try:# 避免错误导致程序直接崩(这种情况一般是在短时间内多次访问被限流了)
state_error = state.json().get('data', {}).get('last_error', None)
if(state_error is not None):
print(f"===错误码:--{state_error['code']}--信息:--{state_error['msg']}===\n")
sys.exit('我直接退出') # ++
except:
pass
# 获取模型的完成状态
state_data = state.json().get('data', {}).get('status', '服务器状态异常')
'''-----获取最终结果-----'''
# 当state_data=='completed'时则代表可以去获取信息了
if(state_data=='completed'):
# 向对话消息详情接口发起请求
info = requests.get(url=url_info, params=params, headers=headers)
info_data = info.json().get('data', '没有取到东西哦!')
# 遍历info_data里的元素判断其中的'type'是否为'answer'(这个是回答内容的标识)
for item in info_data:
if (item['type'] == 'answer'):
temp_info = item
# 将回答内容存入history_list上下文中保持模型的记忆,并将结果打印出来
history_list.append({"role":temp_info['role'],"content":temp_info['content'],"content_type":temp_info['content_type']})
print(f"回复的内容:{temp_info['content']}\n")
break #退出for循环
break #退出while循环
time.sleep(wait)
Google Chrome
Windows 10
Android 10