本文主要是梳理我自己的实现过程(使用的是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)