Coze Studio API 多轮对话教程：用 Python 保持同一个 conversation_id

Coze Studio 开源版 API 的多轮对话，需要让两次调用使用同一个 conversation_id，并保持同一测试用户身份。它的官方 API 说明目前只支持流式对话响应，不能把云版 stream=false 的轮询教程直接套过来。下面用 Python 标准库完成创建会话、两轮提问和流式完成事件检查。

本文适用于已部署 Coze Studio、已配置可用聊天模型的开发者。端点、发布与令牌入口以 2026 年 10 月 1 日读取的开源版官方 API 参考为依据。代码做过本地模拟 HTTP 与 SSE 检查，未调用真实 Studio 账号；编号、响应和检查目标是演示。

先发布到 API，再生成本实例令牌

1. 登录自己的 Studio，进入项目开发，打开要调用的智能体。先在预览中确认能够正常回答。
2. 点击右上角发布，勾选API渠道，再发布。API 调用使用发布版本；编辑页刚改的配置要重新发布才用于这个渠道。
3. 点击左下角头像 → API 授权 → 个人访问令牌 → 添加新令牌，填写名称与有效期，生成后立即保存。使用 Studio 自己签发的令牌，不使用商业版 coze.cn 的令牌代替。
4. 记录本实例地址和智能体 ID。默认本机部署地址是 http://localhost:8888；如果部署在其他机器，使用那个实例的实际地址。智能体 ID 可从开发页 URL 中的 bot 对应数字取得。

令牌只交给后端程序，不放进网页 JavaScript 或代码仓库。本机 HTTP 示例只用于本机调试，非本机访问使用 HTTPS。不同实例里的 ID 和令牌不能混用。

三个标识分别做什么

仅把 user_id 固定，第二轮仍不传会话 ID，会得到新的会话环境。也不要让前端任意提交其他用户的 conversation_id：在应用后端保存“登录用户 → 本实例会话”的归属，再发起请求。平台的令牌权限不能代替你自己业务系统的用户鉴权。

保存并运行两轮客户端

将下面代码保存为 coze_two_turns.py，在终端执行 python coze_two_turns.py，按提示输入实例地址、智能体 ID 与令牌。程序会隐藏令牌输入；也能读取已有的 COZE_STUDIO_TOKEN 环境变量。无需安装第三方 HTTP 库。

程序先调用 /v1/conversation/create 创建会话，再把会话 ID 放入 /v3/chat?conversation_id=…。它只收集最终文本 answer，避免把 delta 与 completed 的完整回复重复拼接；失败、缺少正常结束事件或工具中断都会停止。

"""Coze Studio streaming conversation example; no automatic retries."""
import getpass
import json
import os
import urllib.error
import urllib.parse
import urllib.request


def request(base, token, path, payload):
    req = urllib.request.Request(
        base + path,
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": "Bearer " + token,
                 "Content-Type": "application/json"},
        method="POST",
    )
    return urllib.request.urlopen(req, timeout=90)


def sse_events(response):
    event = "message"
    data = []
    for raw in response:
        line = raw.decode("utf-8").rstrip("\r\n")
        if not line:
            if data:
                yield event, "\n".join(data)
            event, data = "message", []
        elif line.startswith("event:"):
            event = line[6:].lstrip(" ")
        elif line.startswith("data:"):
            data.append(line[5:].lstrip(" "))
    if data:
        yield event, "\n".join(data)


def parse_chat(response):
    answers = {}
    completed = False
    ended = False
    usage = None
    conversation_id = None
    for event, data in sse_events(response):
        if event == "done":
            ended = True
            break
        item = json.loads(data)
        if not isinstance(item, dict):
            raise RuntimeError("流式事件不是对象")
        if event in ("error", "conversation.chat.failed"):
            raise RuntimeError("服务返回失败事件，请核对服务端日志中的错误码")
        if event == "conversation.chat.requires_action":
            raise RuntimeError("需要外部工具回传，本示例不处理 requires_action")
        if event == "conversation.message.completed":
            if item.get("type") == "answer" and item.get("content_type") == "text":
                if not item.get("id") or not isinstance(item.get("content"), str):
                    raise RuntimeError("最终文本消息字段缺失")
                answers[item["id"]] = item["content"]
        if event == "conversation.chat.completed":
            error = item.get("last_error")
            if isinstance(error, dict) and error.get("code") not in (None, 0):
                raise RuntimeError("完成事件仍带有错误，不能视为成功")
            completed = True
            conversation_id = item.get("conversation_id")
            usage = item.get("usage")
    if not completed or not ended or not answers or not conversation_id:
        raise RuntimeError("缺少完成事件、结束事件、会话 ID 或最终文本答案")
    return {"answer": "\n".join(answers.values()),
            "conversation_id": conversation_id, "usage": usage}


def chat(base, token, bot_id, user_id, conversation_id, question):
    path = "/v3/chat?" + urllib.parse.urlencode({"conversation_id": conversation_id})
    payload = {
        "bot_id": bot_id, "user_id": user_id, "stream": True,
        "additional_messages": [{"role": "user", "content": question,
                                 "content_type": "text"}],
    }
    with request(base, token, path, payload) as response:
        if "text/event-stream" not in response.headers.get("Content-Type", ""):
            raise RuntimeError("响应不是 SSE，请核对实例地址与接口版本")
        result = parse_chat(response)
    if str(result["conversation_id"]) != str(conversation_id):
        raise RuntimeError("返回会话 ID 不一致，停止继续调用")
    return result


def main():
    base = input("Studio 地址，如 http://localhost:8888：").strip().rstrip("/")
    parsed = urllib.parse.urlparse(base)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        raise ValueError("请输入完整 Studio 地址")
    if parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1", "::1"):
        raise ValueError("非本机地址请使用 HTTPS，避免令牌明文传输")
    bot_id = input("已发布的智能体 ID：").strip()
    if not bot_id:
        raise ValueError("智能体 ID 不能为空")
    token = os.getenv("COZE_STUDIO_TOKEN") or getpass.getpass("Studio 个人访问令牌：")
    if not token:
        raise ValueError("令牌不能为空")
    with request(base, token, "/v1/conversation/create", {"bot_id": bot_id}) as response:
        item = json.load(response)
    if item.get("code") != 0 or not isinstance(item.get("data"), dict) or not item["data"].get("id"):
        raise RuntimeError("会话创建失败，请检查发布渠道、令牌和实例地址")
    conversation_id = str(item["data"]["id"])
    user_id = "studio-local-demo-27"
    questions = ["请记住本次测试编号是 CEDAR-27，仅回复已记住。",
                 "刚才的测试编号是什么？只返回编号。"]
    print("conversation_id:", conversation_id)
    for index, question in enumerate(questions, 1):
        result = chat(base, token, bot_id, user_id, conversation_id, question)
        print(f"第 {index} 轮：", result["answer"])


if __name__ == "__main__":
    try:
        main()
    except urllib.error.HTTPError as exc:
        print("HTTP 请求失败，状态码：", exc.code)
    except urllib.error.URLError:
        print("连接失败或超时；先核对远端会话，再决定是否继续，程序没有自动重发")
    except (ValueError, RuntimeError, TimeoutError) as exc:
        print("停止：", str(exc))

怎样验证第二轮确实读到了历史

目标结果是第二轮返回 CEDAR-27，而且第二次请求只问“刚才的测试编号是什么”，没有再提交编号。程序两轮使用相同的会话 ID，并检查返回 ID 一致。第一轮确认的措辞可以变化，第二轮编号必须与第一轮输入相同。

如果第二轮不知道编号，依次检查：两次请求的 conversation_id 是否相同、是否误换了实例或智能体、会话上下文是否被清除、智能体自身的人设是否拒绝或覆盖了这个测试任务。多轮字段正确也不保证每个模型一定正确记住内容，仍需检查实际回答。

不要把流打开当成执行成功

HTTP 200 只是连接成功。conversation.message.completed 表示一条消息完整，conversation.chat.completed 表示本轮对话完成，done 表示流正常结束；工具输出和推荐问题也会作为消息出现，不能统统当作答案。

本例需要最终文本答案、完成事件、结束事件和相符会话 ID 都存在，才进入下一轮。当前开源对话实现提供相应的流式消息、会话与完成数据。

这份客户端是单用户、两轮、纯文本示例，不实现生产级重连、任务恢复、多用户存储或外部工具回传。程序每次启动创建新会话；要跨启动续聊，可在后端保存本实例会话 ID 与用户归属，并复用有效记录。不能把本地模拟通过理解为真实账号额度、网络或模型效果已经验证。