Files
012-kaopeilian/docs/规划/全链路联调/Ai工作流/coze/Coze-API文档.md
111 998211c483 feat: 初始化考培练系统项目
- 从服务器拉取完整代码
- 按框架规范整理项目结构
- 配置 Drone CI 测试环境部署
- 包含后端(FastAPI)、前端(Vue3)、管理端

技术栈: Vue3 + TypeScript + FastAPI + MySQL
2026-01-24 19:33:28 +08:00

11 KiB
Raw Blame History

Coze API 使用文档

一、概述

Coze是字节跳动推出的AI对话平台提供强大的Bot开发和对话管理能力。本文档整理了考培练系统陪练功能需要使用的核心API。

官方资源

重要提示

⚠️ 从GitHub获取的源码和示例默认使用 COZE_COM_BASE_URL,使用前必须改为 COZE_CN_BASE_URL

二、认证方式

2.1 个人访问令牌 (Personal Access Token - 推荐)

获取方式

  1. 访问 https://www.coze.cn/open/oauth/pats
  2. 创建新的个人访问令牌
  3. 设置名称、有效期和权限
  4. 保存令牌(仅显示一次)

使用示例

from cozepy import Coze, TokenAuth, COZE_CN_BASE_URL

coze = Coze(
    auth=TokenAuth(token="pat_Sa5OiuUl0gDflnKstQTToIz0sSMshBV06diX0owOeuI1ZK1xDLH5YZH9fSeuKLIi"),
    base_url=COZE_CN_BASE_URL  # 重要:使用中国区域名
)

2.2 OAuth JWT认证 (生产环境推荐)

from cozepy import Coze, JWTAuth, COZE_CN_BASE_URL
from pathlib import Path

coze = Coze(
    auth=JWTAuth(
        client_id="your_client_id",
        private_key=Path("private_key.pem").read_text(),
        public_key_id="your_public_key_id",
        ttl=900  # Token有效期
    ),
    base_url=COZE_CN_BASE_URL
)

三、核心API功能

3.1 Bot对话 (Chat API)

流式对话 (推荐)

功能说明实时流式返回AI响应适合陪练对话场景

示例代码

from cozepy import Coze, TokenAuth, Message, ChatEventType, COZE_CN_BASE_URL

coze = Coze(auth=TokenAuth(token="your_token"), base_url=COZE_CN_BASE_URL)

# 创建流式对话
stream = coze.chat.stream(
    bot_id='7560643598174683145',  # 陪练Bot ID
    user_id='user_123',             # 用户ID业务系统的用户标识
    additional_messages=[
        Message.build_user_question_text("你好,我想练习轻医美产品咨询"),
    ],
    conversation_id='conv_abc',     # 可选关联对话ID
)

# 处理流式事件
for event in stream:
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        # 消息增量(实时打字效果)
        print(event.message.content, end="", flush=True)
    
    elif event.event == ChatEventType.CONVERSATION_MESSAGE_COMPLETED:
        # 消息完成
        print("\n消息完成")
    
    elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
        # 对话完成
        print("Token用量:", event.chat.usage.token_count)
        break
    
    elif event.event == ChatEventType.CONVERSATION_CHAT_FAILED:
        # 对话失败
        print("对话失败:", event.chat.last_error)
        break

非流式对话

chat = coze.chat.create(
    bot_id='bot_id',
    user_id='user_id',
    additional_messages=[
        Message.build_user_question_text('你好')
    ]
)
print(chat.content)

3.2 对话管理 (Conversation API)

创建对话

# 创建新对话
conversation = coze.conversations.create()
print("对话ID:", conversation.id)

获取对话列表

# 获取Bot的对话列表
conversations = coze.conversations.list(
    bot_id='bot_id',
    page_num=1,
    page_size=20
)

for conv in conversations.items:
    print(f"对话ID: {conv.id}, 创建时间: {conv.created_at}")

删除对话

# 删除指定对话
coze.conversations.delete(conversation_id='conversation_id')

3.3 消息历史

获取对话消息

# 获取指定对话的消息列表
messages = coze.conversations.messages.list(
    conversation_id='conversation_id',
    page_num=1,
    page_size=50
)

for msg in messages.items:
    print(f"{msg.role}: {msg.content}")

3.4 中断对话

# 中断正在进行的对话
result = coze.chat.cancel(
    conversation_id='conversation_id',
    chat_id='chat_id'
)

3.5 文件上传 (可选)

from pathlib import Path

# 上传文件(如音频文件)
uploaded_file = coze.files.upload(file=Path('audio.wav'))
print("文件ID:", uploaded_file.id)

# 在消息中使用文件
from cozepy import MessageObjectString

message = Message.build_user_question_objects([
    MessageObjectString.build_audio(file_id=uploaded_file.id)
])

四、事件类型说明

4.1 ChatEventType枚举

事件类型 说明 用途
CONVERSATION_CHAT_CREATED 对话创建 获取chat_id和conversation_id
CONVERSATION_MESSAGE_DELTA 消息增量 实时显示打字效果
CONVERSATION_MESSAGE_COMPLETED 消息完成 显示完整消息
CONVERSATION_CHAT_COMPLETED 对话完成 统计Token用量、清理状态
CONVERSATION_CHAT_FAILED 对话失败 错误处理、用户提示
CONVERSATION_AUDIO_DELTA 音频增量 实时语音播放(语音对话)

4.2 事件对象结构

# 消息增量事件
event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA
event.message.content  # 消息内容增量
event.message.role     # 消息角色user/assistant

# 对话完成事件
event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED
event.chat.id                    # 对话ID
event.chat.conversation_id       # 会话ID
event.chat.usage.token_count     # Token用量
event.chat.usage.input_count     # 输入Token数
event.chat.usage.output_count    # 输出Token数

# 对话失败事件
event.event == ChatEventType.CONVERSATION_CHAT_FAILED
event.chat.last_error            # 错误信息

五、消息构建方法

5.1 文本消息

from cozepy import Message

# 用户问题
user_msg = Message.build_user_question_text("你好,我想了解产品")

# 助手回答
assistant_msg = Message.build_assistant_answer("好的,我来为您介绍")

5.2 多轮对话

# 构建对话历史
messages = [
    Message.build_user_question_text("第一个问题"),
    Message.build_assistant_answer("第一个回答"),
    Message.build_user_question_text("第二个问题"),
]

stream = coze.chat.stream(
    bot_id='bot_id',
    user_id='user_id',
    additional_messages=messages
)

六、错误处理

6.1 常见错误

from cozepy.exception import CozePyError

try:
    chat = coze.chat.create(bot_id='bot_id', user_id='user_id')
except CozePyError as e:
    print(f"Coze API错误: {e}")
    # 处理错误

6.2 超时配置

import httpx
from cozepy import Coze, TokenAuth, SyncHTTPClient

# 自定义超时设置
http_client = SyncHTTPClient(timeout=httpx.Timeout(
    timeout=180.0,   # 总超时
    connect=5.0      # 连接超时
))

coze = Coze(
    auth=TokenAuth(token="your_token"),
    base_url=COZE_CN_BASE_URL,
    http_client=http_client
)

七、调试技巧

7.1 日志配置

import logging
from cozepy import setup_logging

# 启用DEBUG日志
setup_logging(level=logging.DEBUG)

7.2 获取LogID

# 每个请求都有唯一的logid用于排查问题
bot = coze.bots.retrieve(bot_id='bot_id')
print("LogID:", bot.response.logid)

stream = coze.chat.stream(bot_id='bot_id', user_id='user_id')
print("LogID:", stream.response.logid)

八、最佳实践

8.1 陪练对话场景建议

  1. 使用流式响应:提供更好的用户体验
  2. 传递对话上下文:使用conversation_id保持多轮对话
  3. 合理设置超时陪练对话建议180秒超时
  4. 错误重试机制:网络波动时自动重试
  5. Token计数统计监控API使用成本

8.2 用户ID设计

# 推荐使用业务系统的用户ID
user_id = f"trainee_{user.id}"  # trainee_123

# 对话ID可包含场景信息
conversation_id = f"practice_{scene_id}_{user_id}_{timestamp}"

8.3 场景参数传递

可以通过Bot的系统提示词Prompt或参数传递场景信息

# 方式1在用户消息中包含场景背景
scene_context = """
场景:轻医美产品咨询
背景客户是30岁女性关注面部抗衰
AI角色扮演挑剔的客户对价格敏感
"""

stream = coze.chat.stream(
    bot_id='bot_id',
    user_id='user_id',
    additional_messages=[
        Message.build_user_question_text(scene_context + "\n\n开始陪练")
    ]
)

九、性能优化

9.1 连接复用

# 全局初始化一次Coze客户端
coze = Coze(auth=TokenAuth(token="your_token"), base_url=COZE_CN_BASE_URL)

# 多次调用复用连接
def handle_chat(user_id, message):
    stream = coze.chat.stream(bot_id='bot_id', user_id=user_id, ...)
    return stream

9.2 异步并发

from cozepy import AsyncCoze, AsyncTokenAuth
import asyncio

async_coze = AsyncCoze(
    auth=AsyncTokenAuth(token="your_token"),
    base_url=COZE_CN_BASE_URL
)

async def concurrent_chats():
    tasks = [
        async_coze.chat.create(bot_id='bot_id', user_id=f'user_{i}')
        for i in range(10)
    ]
    results = await asyncio.gather(*tasks)
    return results

十、陪练系统专用配置

10.1 配置信息

# 考培练系统陪练Bot配置
COZE_API_BASE = "https://api.coze.cn"
COZE_API_TOKEN = "pat_Sa5OiuUl0gDflnKstQTToIz0sSMshBV06diX0owOeuI1ZK1xDLH5YZH9fSeuKLIi"
COZE_PRACTICE_BOT_ID = "7560643598174683145"

10.2 FastAPI集成示例

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from cozepy import Coze, TokenAuth, Message, ChatEventType, COZE_CN_BASE_URL
import json

app = FastAPI()
coze = Coze(auth=TokenAuth(token="your_token"), base_url=COZE_CN_BASE_URL)

@app.post("/api/v1/practice/start")
async def start_practice(user_id: str, message: str):
    """开始陪练对话SSE流式返回"""
    def generate_stream():
        stream = coze.chat.stream(
            bot_id=COZE_PRACTICE_BOT_ID,
            user_id=user_id,
            additional_messages=[Message.build_user_question_text(message)]
        )
        
        for event in stream:
            if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
                yield f"event: message.delta\ndata: {json.dumps({'content': event.message.content})}\n\n"
            elif event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
                yield f"event: done\ndata: [DONE]\n\n"
    
    return StreamingResponse(
        generate_stream(),
        media_type="text/event-stream"
    )

十一、参考资料


文档维护:本文档基于 Coze Python SDK v0.19.0 编写最后更新时间2025-10-13