Dify API 多轮对话怎么调用？用 Python 保存 conversation_id

Dify API 的多轮对话靠两项信息连接：同一用户持续发送相同的 user，下一轮带回上一轮返回的 conversation_id。第一次可以传空会话 ID，后续不能每次又传空字符串，否则是在不断开启新会话。

下面用 Python 标准库调用已发布的聊天应用，先告诉它一个测试编号，再问刚才的编号。代码对 Chatbot、Chatflow 使用阻塞调用，对 Agent 使用流式调用。本文按 2026 年 10 月 1 日官方接口整理，代码做过本地模拟请求与响应检查，未调用真实 Dify 工作空间；模拟检查不证明某个模型记忆效果或用户账户额度。

先选对应用类型和密钥

blocking 会等本轮回答完成后返回 JSON，适合检查短问题。官方 Chatflow API 文档和 Chatbot API 文档说明了对应接口；Legacy Agent 与新版 Agent 仅支持流式响应。下面的代码先读应用类型再选择调用方式，新版行为可查 Agent API 文档。

1. 准备一个已经 Publish 的 Chatbot、Chatflow 或 Agent，暂时不要加必须暂停等待人工输入的节点。
2. 在应用内部创建该应用的 API Key。它是 Dify 应用密钥，不是模型供应商的 DeepSeek、OpenAI 等密钥。
3. 确认 API Base URL：Dify Cloud 为 https://api.dify.ai/v1；自部署用自己实例的服务 API 地址，不能直接拿 Web App 聊天链接替代。
4. 本例不设置额外必填输入变量，因此 inputs 是空对象。如果应用有必填字段，应按应用参数将实际字段放进 inputs。

密钥范围、地址和后端调用要求见 Dify API 入门文档；发送聊天消息接口列出了 query、inputs、user、conversation_id 和 response_mode 等字段。

Chatflow 还要让回答节点能看到上一轮

一个 conversation_id 说明两次请求属于同一会话，不保证每个 LLM 节点都使用历史。使用 Chatflow 时，在负责回答的 LLM 节点检查 Memory，按需要启用；User 消息绑定当前聊天输入，Answer 节点绑定该 LLM 的 text。官方 LLM 文档说明，记忆是节点级配置，同一对话里的之前交互可纳入后续提示词。

本次测试应用的 System 可写：“根据当前问题和本会话历史回答。用户要求记录测试编号时确认记录，询问刚才编号时只返回编号；没有历史依据就说明无法确认。”不要把编号直接写死到 System，否则第二轮答对也不能证明记忆有效。

保存并运行 Python 示例

把下面代码保存为 dify_two_turns.py。不需要安装第三方 HTTP 库；执行时会隐藏输入 API Key，也可以从已有的 DIFY_API_KEY 环境变量读取。示例只保存在本次进程里的会话 ID，长期产品应按“应用、登录用户、会话”分别保存。

import argparse
import getpass
import json
import os
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


def call_api(base, key, path, payload=None):
    data = None if payload is None else json.dumps(payload).encode('utf-8')
    request = Request(
        base.rstrip('/') + path,
        data=data,
        headers={
            'Authorization': 'Bearer ' + key,
            'Content-Type': 'application/json',
        },
        method='GET' if payload is None else 'POST',
    )
    try:
        with urlopen(request, timeout=120) as response:
            return json.load(response)
    except HTTPError as exc:
        try:
            detail = json.loads(exc.read().decode('utf-8'))
        except (ValueError, UnicodeError):
            detail = {}
        code = detail.get('code', 'unknown_error')
        raise RuntimeError(f'HTTP {exc.code}: {code}') from None
    except URLError as exc:
        raise RuntimeError('Network failure; verify the run before retrying') from exc


def ask(base, key, user, query, conversation_id=''):
    result = call_api(base, key, '/chat-messages', {
        'inputs': {},
        'query': query,
        'response_mode': 'blocking',
        'conversation_id': conversation_id,
        'user': user,
    })
    answer = result.get('answer')
    cid = result.get('conversation_id')
    if not isinstance(answer, str) or not answer.strip():
        raise RuntimeError('No answer returned; inspect the app output')
    if not isinstance(cid, str) or not cid:
        raise RuntimeError('No conversation_id returned')
    return answer, cid


def ask_stream(base, key, user, query, conversation_id=''):
    payload = {'inputs': {}, 'query': query, 'response_mode': 'streaming',
               'conversation_id': conversation_id, 'user': user}
    request = Request(base.rstrip('/') + '/chat-messages',
                      data=json.dumps(payload).encode('utf-8'),
                      headers={'Authorization': 'Bearer ' + key,
                               'Content-Type': 'application/json'}, method='POST')
    answer, cid, completed = '', '', False
    with urlopen(request, timeout=120) as response:
        for raw_line in response:
            line = raw_line.decode('utf-8').strip()
            if not line.startswith('data:'):
                continue
            event = json.loads(line[5:].strip())
            kind = event.get('event')
            cid = event.get('conversation_id') or cid
            if kind == 'error':
                raise RuntimeError('Stream error: ' + event.get('code', 'unknown'))
            if kind == 'agent_message':
                answer += event.get('answer', '')
            elif kind == 'message':
                # For Agent apps, this closing event contains the full answer.
                answer = event.get('answer', '')
            elif kind == 'message_end':
                completed = True
    if not completed or not answer.strip() or not cid:
        raise RuntimeError('Stream ended without a complete answer')
    return answer, cid


def run_two_turns(base, key, user):
    info = call_api(base, key, '/info')
    mode = info.get('mode')
    if mode not in {'chat', 'advanced-chat', 'agent', 'agent-chat'}:
        raise RuntimeError('This example supports chat-family apps only')
    send = ask_stream if mode in {'agent', 'agent-chat'} else ask
    first, cid = send(base, key, user,
                     '请记住，本轮测试编号是 CEDAR-27。只回复已记录。')
    second, second_cid = send(base, key, user,
                            '刚才的测试编号是什么？只回复编号。', cid)
    if second_cid != cid:
        raise RuntimeError('Conversation changed; inspect session handling')
    return first, second, cid


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--base', default='https://api.dify.ai/v1')
    parser.add_argument('--user', default='demo-student-27')
    args = parser.parse_args()
    key = os.environ.get('DIFY_API_KEY') or getpass.getpass('Dify app API key: ')
    if not key.strip():
        raise SystemExit('API key is required')
    try:
        first, second, cid = run_two_turns(args.base, key.strip(), args.user)
    except (RuntimeError, HTTPError, URLError, TimeoutError, ValueError) as exc:
        raise SystemExit(f'Call failed: {exc}') from None
    print('First answer:', first)
    print('Second answer:', second)
    print('Conversation:', cid)

使用 Python 3 运行：

python dify_two_turns.py

自部署实例可以显式指定服务 API 地址，例如：

python dify_two_turns.py --base https://dify.example.com/v1 --user demo-student-27

dify.example.com 是示意域名，必须替换成自己的地址。程序先调用 /info 验证应用类型，之后发出两个 /chat-messages 请求；Chatbot 和 Chatflow 等待 JSON，Agent 与 Legacy Agent 读取 SSE 流。工作流和文本生成应用不适用，将停止运行。

Agent 流式响应中，agent_message 是回复增量；新版 Agent 最后的 message 带完整答案，代码用它替换已收集文本，避免把答案重复拼接。未出现 message_end、遇到 error 事件或没有答案，均按失败处理。本例的解析适用于官方示例中一行一个 JSON 的 data 事件，不实现人工暂停、重连续读或通用多行 SSE 解析。

怎样判断第二轮真的用上了历史

目标结果是第一轮确认记录，第二轮回答 CEDAR-27。措辞可以不同，但第二轮必须能复述编号，而且第二次请求没有再次提供编号。再对照两次请求的 user 和返回的 conversation_id，确认它们一致。

如果 conversation_id 一致，但第二轮说不知道，先检查 Chatflow 中实际回答的 LLM 节点是否开启 Memory、当前用户输入是否正确传入。不能只因会话 ID 相同就认定模型确实读取了历史。

还可以把代码第一轮的编号改成 CEDAR-48，重新运行新会话；若第二轮仍答旧编号，检查提示词是否写死了测试数据、是否使用了错误会话或缓存。编号变化是本例的验证条件，不是效果数字。

user 是范围标识，不是登录凭证

官方 终端用户身份文档说明，Dify 不会验证你传入的 user 是否真的是那个人。生产系统应在后端验证登录身份，再生成稳定 user，并检查当前用户是否有权使用对应 conversation_id；不要接受浏览器随意传来的他人标识。

演示中的 demo-student-27 只用于单人测试。同一个固定 user 不能作为所有真实用户的共同身份，否则会话范围无法按人区分。API 会话与 Dify Web App 会话也相互独立，不能期待网页聊天记录自动出现在 API 返回里。

接口失败时先处理原因

错误类别可查 Dify 错误处理说明。代码设置 120 秒客户端超时，不会改变云端代理和服务的超时限制；长生成任务需要另外设计流式处理，不应不断提高等待时间。

怎样让程序重启后继续上次会话？

把返回的 conversation_id 与用户身份一起保存到自己的会话存储，下次读取它再传给接口。本例为了展示两轮连接，只在内存里保存，不实现数据库、登录和会话选择界面。

只跑通这个例子就能上线聊天网站吗？

还不能。它只验证短问题请求、会话连接和基本错误处理。网页产品还需要后端鉴权、会话归属、流式显示、限流和断线后的状态核对；不能把应用 API Key 放到浏览器代码中。