# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能,如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块
import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv
import os # noqa
import time
# Get an access_token through personal access token oroauth.
api_coze_token = "xxxxx"
# os.getenv("COZE_API_TOKEN")
from cozepy import Coze, TokenAuth, Message, ChatStatus, MessageContentType, ChatEventType # noqa
from cozepy import COZE_CN_BASE_URL
# Init the Coze client through the access_token.
coze = Coze(auth=TokenAuth(token=api_coze_token), base_url=COZE_CN_BASE_URL)
# Create a bot instance in Coze, copy the last number from the web link as the bot's ID.
bot_id = "xxxxx"
# The user id identifies the identity of a user. Developers can use a custom business ID
# or a random string.
user_id = "1111"
def main(args):
chat = coze.chat.create(
bot_id=bot_id,
user_id=user_id,
additional_messages=[
Message.build_user_question_text(args)
],
)
# Assume the development allows at most one chat to run for 10 minutes. If it exceeds 10
# minutes, the chat will be cancelled.
# And when the chat status is not completed, poll the status of the chat once every second.
# After the chat is completed, retrieve all messages in the chat.
start = int(time.time())
timeout = 600
while chat.status == ChatStatus.IN_PROGRESS:
if int(time.time()) - start > timeout:
# too long, cancel chat
coze.chat.cancel(conversation_id=chat.conversation_id, chat_id=chat.id)
break
time.sleep(1)
# Fetch the latest data through the retrieve interface
chat = coze.chat.retrieve(conversation_id=chat.conversation_id, chat_id=chat.id)
# When the chat status becomes completed, all messages under this chat can be retrieved through the list messages interface.
messages = coze.chat.messages.list(conversation_id=chat.conversation_id, chat_id=chat.id)
for message in messages:
print(f"role={message.role}, content={message.content}")
return messages