998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
141 lines
4.1 KiB
Python
141 lines
4.1 KiB
Python
"""
|
|
Coze 集成示例
|
|
演示如何使用 Coze 服务
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# 设置环境变量（实际使用时应通过 .env 文件或环境配置）
|
|
os.environ.update({
|
|
"COZE_API_BASE": "https://api.coze.cn",
|
|
"COZE_WORKSPACE_ID": "your_workspace_id",
|
|
"COZE_API_TOKEN": "your_api_token",
|
|
"COZE_CHAT_BOT_ID": "7489905095250526219",
|
|
"COZE_TRAINING_BOT_ID": "7494965200370581538",
|
|
"COZE_EXAM_BOT_ID": "7492288561950195724"
|
|
})
|
|
|
|
from app.services.ai.coze import (
|
|
get_coze_service,
|
|
CreateSessionRequest,
|
|
SendMessageRequest,
|
|
EndSessionRequest,
|
|
SessionType
|
|
)
|
|
|
|
|
|
async def demo_course_chat():
|
|
"""演示课程对话功能"""
|
|
print("\n=== 课程对话示例 ===")
|
|
|
|
service = get_coze_service()
|
|
|
|
# 1. 创建会话
|
|
create_request = CreateSessionRequest(
|
|
session_type=SessionType.COURSE_CHAT,
|
|
user_id="demo-user-123",
|
|
course_id="python-basics-101",
|
|
metadata={"course_name": "Python 基础教程"}
|
|
)
|
|
|
|
session_response = await service.create_session(create_request)
|
|
print(f"创建会话成功: {session_response.session_id}")
|
|
|
|
# 2. 发送消息
|
|
message_request = SendMessageRequest(
|
|
session_id=session_response.session_id,
|
|
content="请解释一下 Python 的装饰器是什么？",
|
|
stream=True
|
|
)
|
|
|
|
print("\n助手回复:")
|
|
async for event in service.send_message(message_request):
|
|
if event.event.value == "message_delta":
|
|
print(event.content, end="", flush=True)
|
|
elif event.event.value == "message_completed":
|
|
print(f"\n\n[消息完成] Token 使用: {event.data.get('usage', {})}")
|
|
|
|
# 3. 获取历史消息
|
|
messages = await service.get_session_messages(session_response.session_id)
|
|
print(f"\n会话消息数: {len(messages)}")
|
|
|
|
|
|
async def demo_training_session():
|
|
"""演示陪练会话功能"""
|
|
print("\n\n=== 陪练会话示例 ===")
|
|
|
|
service = get_coze_service()
|
|
|
|
# 1. 创建陪练会话
|
|
create_request = CreateSessionRequest(
|
|
session_type=SessionType.TRAINING,
|
|
user_id="demo-user-456",
|
|
training_topic="客诉处理技巧",
|
|
metadata={"difficulty": "intermediate"}
|
|
)
|
|
|
|
session_response = await service.create_session(create_request)
|
|
print(f"创建陪练会话成功: {session_response.session_id}")
|
|
|
|
# 2. 进行多轮对话
|
|
scenarios = [
|
|
"客户抱怨产品质量有问题，要求退款",
|
|
"我会先向客户道歉，然后了解具体问题",
|
|
"客户说产品使用一周就坏了，非常生气"
|
|
]
|
|
|
|
for i, content in enumerate(scenarios):
|
|
print(f"\n--- 第 {i+1} 轮对话 ---")
|
|
print(f"用户: {content}")
|
|
|
|
message_request = SendMessageRequest(
|
|
session_id=session_response.session_id,
|
|
content=content,
|
|
stream=False # 非流式响应
|
|
)
|
|
|
|
print("助手: ", end="")
|
|
async for event in service.send_message(message_request):
|
|
if event.event.value == "message_completed":
|
|
print(event.content)
|
|
|
|
# 模拟思考时间
|
|
await asyncio.sleep(1)
|
|
|
|
# 3. 结束会话
|
|
end_request = EndSessionRequest(
|
|
reason="练习完成",
|
|
feedback={
|
|
"rating": 5,
|
|
"helpful": True,
|
|
"comment": "模拟场景很真实，学到了很多处理技巧"
|
|
}
|
|
)
|
|
|
|
end_response = await service.end_session(session_response.session_id, end_request)
|
|
print(f"\n会话结束:")
|
|
print(f"- 时长: {end_response.duration_seconds} 秒")
|
|
print(f"- 消息数: {end_response.message_count}")
|
|
|
|
|
|
async def main():
|
|
"""运行所有示例"""
|
|
try:
|
|
# 运行课程对话示例
|
|
await demo_course_chat()
|
|
|
|
# 运行陪练会话示例
|
|
await demo_training_session()
|
|
|
|
except Exception as e:
|
|
print(f"\n错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 运行示例
|
|
asyncio.run(main())
|