998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试cozepy 0.2.0的conversation续接功能
|
||
"""
|
||
from cozepy import Coze, TokenAuth, COZE_CN_BASE_URL, Message, ChatEventType
|
||
|
||
# 配置
|
||
TOKEN = "pat_Sa5OiuUl0gDflnKstQTToIz0sSMshBV06diX0owOeuI1ZK1xDLH5YZH9fSeuKLIi"
|
||
BOT_ID = "7560643598174683145"
|
||
|
||
def test_conversation_continuation():
|
||
"""测试对话续接"""
|
||
print("=" * 60)
|
||
print("测试cozepy 0.2.0对话续接功能")
|
||
print("=" * 60)
|
||
|
||
# 初始化客户端
|
||
coze = Coze(
|
||
auth=TokenAuth(token=TOKEN),
|
||
base_url=COZE_CN_BASE_URL
|
||
)
|
||
|
||
# 第1步：创建conversation
|
||
print("\n1. 创建conversation...")
|
||
conversation = coze.conversations.create()
|
||
conversation_id = conversation.id
|
||
print(f"✅ Conversation已创建: {conversation_id}")
|
||
|
||
# 第2步：发送第一条消息
|
||
print("\n2. 发送第一条消息...")
|
||
user_id = "test_user_001"
|
||
message1 = "我的名字是张三，请记住我的名字"
|
||
|
||
stream1 = coze.chat.stream(
|
||
bot_id=BOT_ID,
|
||
user_id=user_id,
|
||
additional_messages=[Message.build_user_question_text(message1)],
|
||
conversation_id=conversation_id
|
||
)
|
||
|
||
print(f"用户消息: {message1}")
|
||
print("AI回复: ", end="", flush=True)
|
||
|
||
ai_reply1 = ""
|
||
for event in stream1:
|
||
if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
|
||
ai_reply1 += event.message.content
|
||
print(event.message.content, end="", flush=True)
|
||
elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
|
||
break
|
||
|
||
print(f"\n✅ 第一轮对话完成")
|
||
|
||
# 第3步：发送第二条消息（测试是否记住）
|
||
print("\n3. 发送第二条消息（测试是否记住名字）...")
|
||
message2 = "我叫什么名字？"
|
||
|
||
stream2 = coze.chat.stream(
|
||
bot_id=BOT_ID,
|
||
user_id=user_id,
|
||
additional_messages=[Message.build_user_question_text(message2)],
|
||
conversation_id=conversation_id # 使用同一个conversation_id
|
||
)
|
||
|
||
print(f"用户消息: {message2}")
|
||
print("AI回复: ", end="", flush=True)
|
||
|
||
ai_reply2 = ""
|
||
for event in stream2:
|
||
if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
|
||
ai_reply2 += event.message.content
|
||
print(event.message.content, end="", flush=True)
|
||
elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
|
||
break
|
||
|
||
print(f"\n✅ 第二轮对话完成")
|
||
|
||
# 分析结果
|
||
print("\n" + "=" * 60)
|
||
print("测试结果分析")
|
||
print("=" * 60)
|
||
print(f"conversation_id: {conversation_id}")
|
||
print(f"\n第一轮AI回复: {ai_reply1[:100]}...")
|
||
print(f"\n第二轮AI回复: {ai_reply2}")
|
||
|
||
# 检查AI是否记住了名字
|
||
if "张三" in ai_reply2:
|
||
print("\n✅ 成功！AI记住了名字，对话续接正常工作")
|
||
else:
|
||
print("\n❌ 失败！AI没有记住名字，对话续接可能不工作")
|
||
print(" 这说明cozepy 0.2.0的conversation_id参数可能不起作用")
|
||
|
||
if __name__ == "__main__":
|
||
test_conversation_continuation()
|
||
|