网上找了些资料,但是发现好多都是用的 Coze 的历史接口,现在新的 V3 接口处理方式包括字段和处理方式都不一样了,所以现在这个教程是基于最新版 V3 接口的。配置好 API 后,在影刀 RPA 中使用。
一、API 介绍
开发文档 : https://www.coze.cn/docs/developer_guides/coze_api_overview
目前 API 的调用是免费供开发者使用的,但是有一定的限制,包括频率和次数:
发起对话 API 的请求频率限制
QPS(每秒发送的请求数):2
QPM(每分钟发送的请求数):60
QPD(每天发送的请求数):3000
扣子基础版和专业版账号的 API 使用限额存在差异。
基础版:2024年8月15日之后,基础版为每个账号免费提供 100 次 API 调用,累计用量超出额度后无法使用扣子 API。
专业版:不限制调用 API 的频率和次数。调用发起对话 API, 按Token 消耗收取费用,其他接口免费。
二、准备工作
要使用 API 服务 ,在开始之前需要准备 3 个步骤:
1,将智能体发布为 API 服务。
2,添加访问令牌。
3,获取 bot_id
2.1) 发布 API 服务
在单个 Bot 的编排页面,点击「发布」按钮。
在发布页面,选择「Bot as API」
发布后,等平台审核通过,具体审核状态可以通过发布历史查看,具体为点击「发布记录」查看。
2.2 ) 添加访问令牌
登录 Coze 后,左侧菜单找到「扣子 API」
点击链接后,点击「授权」标签,个人访问信息,再添加新令牌。
勾选 Bot 管理的权限,注意添加时,令牌只会显示 1 次,所以一旦显示及时复制保存。
2.3 ) 获取 bot_id
从个人空间中,点击某个bot进去,在浏览器链接里面找到具体的id
经过上面几步准备工作,API 的准备工作就算完成了,如果想测试一下可以使用 CURL 命令,更改如下命令的
import requests
import timeAPI_URL = "https://api.coze.cn/v3/chat"
RETRIEVE_URL = "https://api.coze.cn/v3/chat/retrieve"
MESSAGE_LIST_URL = "https://api.coze.cn/v3/chat/message/list"
PAT_TOKEN = "" # 替换为你的 Personal Access Token
BOT_ID = "" # 替换为你的 bot_id
USER_ID = "123" # 替换为你的 user_id
POLLING_INTERVAL = 2
MAX_RETRIES = 30# 函数:发送消息
def send_message_to_coze(message):headers = {"Authorization": f"Bearer {PAT_TOKEN}","Content-Type": "application/json"}data = {"bot_id": BOT_ID,"user_id": USER_ID,"stream": False,"auto_save_history": True,"additional_messages": [{"role": "user", "content": message, "content_type": "text"}]}response = requests.post(API_URL, json=data, headers=headers)return response.json() if response.status_code == 200 else None# 函数:轮询获取对话状态
def poll_conversation_status(conversation_id, chat_id):headers = {"Authorization": f"Bearer {PAT_TOKEN}","Content-Type": "application/json"}params = {"conversation_id": conversation_id, "chat_id": chat_id}for _ in range(MAX_RETRIES):response = requests.get(RETRIEVE_URL, headers=headers, params=params)if response.status_code == 200:conversation_data = response.json()if conversation_data.get("code") == 0:status = conversation_data["data"].get("status")if status == "completed":return conversation_data["data"]elif status == "failed":print("对话失败,无法获取最终结果。")return Nonetime.sleep(POLLING_INTERVAL)else:print("API调用失败,错误信息:", conversation_data.get("msg"))else:print(f"Error: HTTP {response.status_code}")return Noneprint("达到最大轮询次数,对话可能仍在进行中。")return None# 函数:获取对话的详细信息
def get_conversation_details(conversation_id, chat_id):headers = {"Authorization": f"Bearer {PAT_TOKEN}","Content-Type": "application/json"}params = {"conversation_id": conversation_id, "chat_id": chat_id}response = requests.get(MESSAGE_LIST_URL, headers=headers, params=params)return response.json() if response.status_code == 200 else None# 主程序:与 Coze 智能体进行对话
def main(message):print("正在与 Coze 智能体对话...")initial_response = send_message_to_coze(message)if not initial_response:return "无法获取初始响应。"conversation_id = initial_response.get("data", {}).get("conversation_id")chat_id = initial_response.get("data", {}).get("id")if not conversation_id or not chat_id:return "未获取到 conversation_id 或 chat_id,无法进行轮询。"final_status = poll_conversation_status(conversation_id, chat_id)if not final_status:return "无法获取对话的最终状态。"conversation_details = get_conversation_details(conversation_id, chat_id)if conversation_details:for message in details_data.get('data', []):if message.get('role') == 'assistant' and message.get('type') == 'answer':content = message.get('content', '没有回复内容')return contentreturn "无法获取对话的详细信息。"`
`
`