耿雨飞的个人博客

[[ slide_text ]]

[[ item.c ]]

Dify 系列教程 | 第 5 篇：通过 API 集成 Dify 到你的产品中

发布时间：2026-04-08 作者：gengyufei 来源：gengyufei个人博客

dify 教程

> **Dify 系列教程** | 模块一：认识 Dify -- 从零开始 | 第 5 篇 / 共 30 篇
>
> **适合读者**：有基本编程经验的开发者
>
> **学习目标**：掌握 Dify Service API 的使用方法；学会在自己的应用中调用 Dify 接口；了解 SDK 的使用和最佳实践

---

## 开篇：从 WebApp 到产品级集成

在上一篇教程中，我们创建了第一个 AI 聊天助手，并通过 Dify 自动生成的 WebApp 链接分享给了用户。但在实际产品开发中，你更可能需要把 AI 能力**嵌入到已有的产品中** -- 比如在你的 App 里加一个智能客服、在后台系统里增加文档分析功能，或者在微信小程序里接入 AI 对话。

这就需要用到 Dify 的 **Service API**。Dify 提供了一套完整的 RESTful API，让你可以用任何编程语言调用 AI 应用的能力，就像调用一个普通的后端接口一样简单。

本篇将带你从零掌握 Dify API 的完整使用方法。

---

## 一、理解 Dify 的三层 API 体系

Dify 提供了三套不同用途的 API，它们各自面向不同的使用者。理解这三者的区别，是正确使用 API 的第一步。

### 1.1 API 体系总览

| API 类型        | URL 前缀       | 鉴权方式                    | 使用者            | 用途                                     |
| --------------- | -------------- | --------------------------- | ----------------- | ---------------------------------------- |
| **Console API** | `/console/api` | 用户登录 Session            | Dify 控制台管理员 | 管理应用、配置模型、管理知识库等后台操作 |
| **Service API** | `/v1`          | App API Key (Bearer Token)  | 产品开发者        | 在自己的产品中调用 AI 应用能力           |
| **Web API**     | `/api`         | App Access Token / Passport | WebApp 终端用户   | Dify 自动生成的 WebApp 前端调用          |

**你需要关注的是 Service API** -- 它是对外集成的核心接口。

### 1.2 Service API 的注册与路由

在后端源码中，Service API 通过 Flask Blueprint 注册（`api/controllers/service_api/__init__.py`）：

```python
bp = Blueprint("service_api", __name__, url_prefix="/v1")
api = ExternalApi(bp, version="1.0", title="Service API")
```

所有 Service API 端点都以 `/v1` 为前缀。完整的 API URL 格式为：

```
https://your-dify-domain/v1/{endpoint}
```

如果是本地部署，默认为 `http://localhost/v1/{endpoint}`（Nginx 代理到后端 Flask 服务的 5001 端口）。

---

## 二、获取 API Key 与鉴权机制

### 2.1 获取 API Key

每个 Dify 应用都有独立的 API Key。获取方式：

1. 进入 Dify 控制台，打开目标应用
2. 点击左侧导航栏的 **"访问 API"**
3. 在页面顶部找到 **API 密钥** 区域
4. 点击 **"创建密钥"** 生成新的 API Key
5. 复制并安全保存（只显示一次）

> **安全提示**：API Key 是访问你 AI 应用的唯一凭证。不要将它硬编码在前端代码中、不要提交到 Git 仓库、不要通过不安全的渠道传输。建议使用环境变量管理。

### 2.2 鉴权机制详解

Service API 使用 **Bearer Token** 鉴权。每次请求都需要在 HTTP 头中携带 API Key：

```
Authorization: Bearer {your_api_key}
```

在后端，鉴权由 `validate_and_get_api_token()` 函数处理（`api/controllers/service_api/wraps.py`），验证流程如下：

```
请求到达
  ↓
提取 Authorization 头
  ↓
验证格式是否为 "Bearer {token}"
  ↓
查询 Redis 缓存（ApiTokenCache）
  ↓ 命中      ↓ 未命中
返回 Token   查询数据库 → 写入缓存 → 返回 Token
  ↓
验证关联的 App 状态（是否正常、是否启用 API）
  ↓
验证 Tenant/Workspace 状态
  ↓
创建或获取 EndUser → 执行请求
```

Token 信息存储在数据库的 `api_tokens` 表中，包含 `token`（密钥字符串）、`app_id`（关联应用）、`tenant_id`（所属工作空间）、`last_used_at`（最后使用时间）等字段。

### 2.3 用户标识（user 参数）

几乎所有 Service API 请求都需要传递一个 `user` 参数，用于标识终端用户。这个参数有两个重要作用：

1. **对话隔离**：不同 user 的对话历史互相独立
2. **用量统计**：Dify 控制台可以按用户维度查看使用数据

`user` 是一个字符串，你可以使用任何能唯一标识用户的值，比如用户 ID、邮箱哈希、设备 ID 等。

```json
{
  "user": "user-abc-123"
}
```

---

## 三、核心 API 接口详解

Service API 按功能分为以下几类端点。以下基于后端 `api/controllers/service_api/` 目录下的实际路由注册。

### 3.1 对话消息 API（Chat）

适用于**聊天助手**和 **Agent 智能体**类型的应用。

**发送消息**

```
POST /v1/chat-messages
```

请求体：

| 参数                 | 类型    | 必填 | 说明                                        |
| -------------------- | ------- | ---- | ------------------------------------------- |
| `query`              | string  | 是   | 用户发送的消息内容                          |
| `inputs`             | object  | 是   | 应用中定义的变量值，无变量传 `{}`           |
| `response_mode`      | string  | 否   | `"blocking"` 阻塞或 `"streaming"` 流式      |
| `conversation_id`    | string  | 否   | 会话 ID，首次对话不传，后续传入实现多轮对话 |
| `user`               | string  | 是   | 终端用户标识                                |
| `files`              | array   | 否   | 上传的文件引用列表                          |
| `auto_generate_name` | boolean | 否   | 是否自动生成会话标题，默认 `true`           |

请求示例：

```json
{
  "query": "帮我写一篇关于 AI 的公众号文章",
  "inputs": {
    "writing_type": "公众号文章"
  },
  "response_mode": "blocking",
  "user": "user-123"
}
```

阻塞模式响应：

```json
{
  "event": "message",
  "message_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "conversation_id": "f1e2d3c4-b5a6-7890-abcd-ef0987654321",
  "answer": "好的，我来帮你写一篇关于 AI 的公众号文章...",
  "metadata": {
    "usage": {
      "prompt_tokens": 150,
      "completion_tokens": 800,
      "total_tokens": 950
    }
  },
  "created_at": 1712534400
}
```

**停止生成**

```
POST /v1/chat-messages/{task_id}/stop
Content-Type: application/json

{"user": "user-123"}
```

### 3.2 文本补全 API（Completion）

适用于**文本生成**类型的应用。

```
POST /v1/completion-messages
```

请求体：

| 参数            | 类型   | 必填 | 说明                          |
| --------------- | ------ | ---- | ----------------------------- |
| `inputs`        | object | 是   | 应用中定义的变量值            |
| `query`         | string | 否   | 补充的查询文本                |
| `response_mode` | string | 否   | `"blocking"` 或 `"streaming"` |
| `user`          | string | 是   | 终端用户标识                  |
| `files`         | array  | 否   | 文件引用列表                  |

请求示例：

```json
{
  "inputs": {
    "text": "Dify is an open-source LLM application development platform."
  },
  "response_mode": "blocking",
  "user": "user-456"
}
```

### 3.3 工作流执行 API（Workflow）

适用于**工作流**类型的应用。

**执行工作流**

```
POST /v1/workflows/run
```

请求体：

| 参数            | 类型   | 必填 | 说明                          |
| --------------- | ------ | ---- | ----------------------------- |
| `inputs`        | object | 是   | 工作流输入变量                |
| `response_mode` | string | 否   | `"blocking"` 或 `"streaming"` |
| `user`          | string | 是   | 终端用户标识                  |
| `files`         | array  | 否   | 文件引用列表                  |

阻塞模式响应：

```json
{
  "workflow_run_id": "run-abc-123",
  "status": "succeeded",
  "outputs": {
    "result": "分析完成，共识别 5 个关键主题...",
    "confidence": 0.95
  },
  "total_tokens": 500,
  "created_at": 1712534400
}
```

**查询执行状态**

```
GET /v1/workflows/run/{workflow_run_id}
```

**查询执行日志**

```
GET /v1/workflows/logs?status=succeeded&page=1&limit=20
```

### 3.4 文件上传 API

在发送消息前，如果需要附带文件（如图片、文档），需要先通过此接口上传。

```
POST /v1/files/upload
Content-Type: multipart/form-data
```

参数：

| 参数   | 类型   | 必填 | 说明                      |
| ------ | ------ | ---- | ------------------------- |
| `file` | file   | 是   | 要上传的文件              |
| `user` | string | 是   | 终端用户标识（form 字段） |

响应（201 Created）：

```json
{
  "id": "file-id-123",
  "name": "document.pdf",
  "size": 102400,
  "extension": ".pdf",
  "mime_type": "application/pdf",
  "created_by": "user-123",
  "created_at": 1712534400
}
```

上传成功后，在对话请求的 `files` 参数中引用该文件：

```json
{
  "files": [
    {
      "type": "document",
      "transfer_method": "local_file",
      "upload_file_id": "file-id-123"
    }
  ]
}
```

也可以直接传递远程 URL，无需先上传：

```json
{
  "files": [
    {
      "type": "image",
      "transfer_method": "remote_url",
      "url": "https://example.com/image.jpg"
    }
  ]
}
```

**支持的文件类型**：PDF、DOCX、TXT、CSV、JSON、JPG、PNG、GIF、WEBP、MP3、WAV 等。默认大小限制 10MB。

### 3.5 会话管理 API

| 接口                          | 方法   | 说明                 |
| ----------------------------- | ------ | -------------------- |
| `/v1/conversations`           | GET    | 获取会话列表         |
| `/v1/conversations/{id}`      | DELETE | 删除会话             |
| `/v1/conversations/{id}/name` | POST   | 重命名会话           |
| `/v1/messages`                | GET    | 获取会话中的消息列表 |
| `/v1/messages/{id}/feedbacks` | POST   | 消息反馈（点赞/踩）  |
| `/v1/messages/{id}/suggested` | GET    | 获取推荐后续问题     |

**获取会话列表示例**：

```
GET /v1/conversations?user=user-123&limit=20&sort_by=-updated_at
```

响应：

```json
{
  "limit": 20,
  "has_more": true,
  "data": [
    {
      "id": "conv-id-123",
      "name": "关于 AI 的讨论",
      "inputs": {},
      "status": "normal",
      "created_at": 1712534400,
      "updated_at": 1712534500
    }
  ]
}
```

### 3.6 应用信息 API

| 接口             | 方法 | 说明                 |
| ---------------- | ---- | -------------------- |
| `/v1/info`       | GET  | 获取应用基本信息     |
| `/v1/parameters` | GET  | 获取应用输入参数配置 |
| `/v1/meta`       | GET  | 获取应用元数据       |

这些接口在初始化集成时非常有用 -- 可以动态获取应用配置的变量列表，据此渲染用户输入表单。

---

## 四、流式响应（SSE）与阻塞式响应

这是 API 集成中最重要的选择之一。理解两种模式的区别，直接影响用户体验。

### 4.1 阻塞式响应（Blocking）

```json
{"response_mode": "blocking"}
```

**工作方式**：发送请求后，服务端等待 AI 模型完整生成回答，然后一次性返回全部内容。

**优点**：

- 实现简单，一个 HTTP 请求/响应即可
- 适合后端对后端的集成
- 方便做后续处理（如存储、分析）

**缺点**：

- 用户等待时间长（可能 10~30 秒）
- 没有实时反馈，用户体验差
- 长文本生成可能超时

**适用场景**：后台任务、批量处理、不需要实时展示的场景。

### 4.2 流式响应（Streaming / SSE）

```json
{"response_mode": "streaming"}
```

**工作方式**：发送请求后，服务端通过 Server-Sent Events（SSE）协议逐步返回内容片段，实现"打字机"效果。

响应头：

```
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
```

**SSE 事件格式**：

每个事件由 `event:` 和 `data:` 两行组成，事件之间以空行分隔：

```
event: message
data: {"event": "message", "message_id": "msg-123", "answer": "你", "created_at": 1712534400}

event: message
data: {"event": "message", "message_id": "msg-123", "answer": "好", "created_at": 1712534400}

event: message_end
data: {"event": "message_end", "message_id": "msg-123", "metadata": {"usage": {"total_tokens": 950}}}
```

### 4.3 SSE 事件类型速查

**Chat/Completion 类应用**：

| 事件类型          | 说明                                  |
| ----------------- | ------------------------------------- |
| `message`         | AI 回答的内容片段（增量）             |
| `message_end`     | 消息生成完成，附带 token 用量等元数据 |
| `message_replace` | 内容审核替换（敏感内容被过滤）        |
| `message_file`    | 文件附件引用                          |
| `tts_message`     | 语音合成音频（base64 编码 MP3）       |
| `tts_message_end` | 语音流结束                            |
| `error`           | 流式生成过程中的错误                  |
| `ping`            | 心跳包，每 10 秒发送一次，保持连接    |

**Workflow 类应用**额外支持：

| 事件类型            | 说明                         |
| ------------------- | ---------------------------- |
| `workflow_started`  | 工作流开始执行               |
| `node_started`      | 节点开始执行                 |
| `node_finished`     | 节点执行完成，包含输出和耗时 |
| `workflow_finished` | 工作流执行完成               |

### 4.4 如何选择？

```
面向终端用户的对话界面？→ 流式（体验好）
后台自动化任务？→ 阻塞（实现简单）
微信小程序 / 钉钉机器人？→ 看平台是否支持 SSE
需要完整响应后做处理？→ 阻塞，或流式拼接完再处理
```

---

## 五、使用 SDK 快速集成

Dify 官方提供了 **Node.js SDK**（`dify-client` npm 包），可以大幅简化集成工作。

### 5.1 Node.js SDK

**安装**

```bash
npm install dify-client
```

**SDK 提供的客户端类**（`sdks/nodejs-client/src/index.ts`）：

| 客户端类              | 用途                                                       |
| --------------------- | ---------------------------------------------------------- |
| `DifyClient`          | 基础客户端，提供应用信息查询、消息反馈、文件上传等通用方法 |
| `ChatClient`          | 聊天助手/Agent 应用的对话操作                              |
| `CompletionClient`    | 文本生成应用的补全操作                                     |
| `WorkflowClient`      | 工作流应用的执行操作                                       |
| `KnowledgeBaseClient` | 知识库管理（需要 Dataset API Key）                         |

**基础用法**

```javascript
import { ChatClient, CompletionClient, WorkflowClient } from 'dify-client'

const API_KEY = 'app-your-api-key'

// 创建客户端
const chatClient = new ChatClient(API_KEY)
const completionClient = new CompletionClient(API_KEY)
const workflowClient = new WorkflowClient(API_KEY)
```

如果是自部署 Dify，需要指定 Base URL：

```javascript
const chatClient = new ChatClient({
  apiKey: 'app-your-api-key',
  baseUrl: 'http://your-dify-server/v1'
})
```

**发送聊天消息（阻塞模式）**

```javascript
const response = await chatClient.createChatMessage({
  inputs: { writing_type: '公众号文章' },
  query: '帮我写一篇关于 AI 的文章',
  user: 'user-123',
  response_mode: 'blocking'
})

console.log(response.data.answer)
```

**发送聊天消息（流式模式）**

```javascript
const stream = await chatClient.createChatMessage({
  inputs: {},
  query: '你好，介绍一下你自己',
  user: 'user-123',
  response_mode: 'streaming'
})

// 方式一：逐事件迭代
for await (const event of stream) {
  if (event.event === 'message') {
    process.stdout.write(event.data.answer)
  } else if (event.event === 'message_end') {
    console.log('\n[完成]', event.data.metadata)
  }
}

// 方式二：直接获取完整文本
// const text = await stream.toText()
```

**执行工作流**

```javascript
const result = await workflowClient.run({
  inputs: { query: '分析这组销售数据' },
  user: 'user-789',
  response_mode: 'blocking'
})

console.log(result.data.outputs)
```

**停止生成**

```javascript
await chatClient.stopChatMessage('task-id-xxx', 'user-123')
```

**会话管理**

```javascript
// 获取会话列表
const conversations = await chatClient.getConversations('user-123', null, 20)

// 获取历史消息
const messages = await chatClient.getConversationMessages('user-123', 'conv-id-xxx')

// 删除会话
await chatClient.deleteConversation('conv-id-xxx', 'user-123')
```

### 5.2 SDK 配置选项

Node.js SDK 支持以下配置（`sdks/nodejs-client/src/http/client.ts`）：

```javascript
const client = new ChatClient({
  apiKey: 'app-xxx',
  baseUrl: 'http://localhost/v1',    // 自部署地址
  timeout: 60000,                    // 超时时间（毫秒），默认 60s
  maxRetries: 3,                     // 最大重试次数，默认 3
  retryDelay: 1000,                  // 重试间隔（毫秒），默认 1s
  enableLogging: false               // 是否启用日志
})
```

### 5.3 SDK 错误处理

SDK 提供了层级化的错误类型（`sdks/nodejs-client/src/errors/dify-error.ts`）：

```javascript
import { DifyError, AuthenticationError, RateLimitError } from 'dify-client'

try {
  const response = await chatClient.createChatMessage({...})
} catch (error) {
  if (error instanceof AuthenticationError) {
    console.error('API Key 无效或已过期')
  } else if (error instanceof RateLimitError) {
    console.error('请求频率超限，请稍后重试')
    console.error(`重试等待：${error.retryAfter}秒`)
  } else if (error instanceof DifyError) {
    console.error(`API 错误 [${error.statusCode}]:`, error.message)
  }
}
```

---

## 六、API 错误处理与限流

### 6.1 错误响应格式

Dify Service API 的所有错误都返回统一的 JSON 格式（由 `api/libs/external_api.py` 处理）：

```json
{
  "code": "error_code_snake_case",
  "message": "人类可读的错误描述",
  "status": 400
}
```

### 6.2 常见错误码速查

| 错误码                        | HTTP 状态 | 说明                                   | 处理建议                               |
| ----------------------------- | --------- | -------------------------------------- | -------------------------------------- |
| `unauthorized`                | 401       | API Key 无效或缺失                     | 检查 Authorization 头格式和 Key 有效性 |
| `not_chat_app`                | 400       | 对非聊天应用调用了 Chat 接口           | 检查应用类型是否匹配                   |
| `not_completion_app`          | 400       | 对非文本生成应用调用了 Completion 接口 | 使用正确的接口                         |
| `not_workflow_app`            | 400       | 对非工作流应用调用了 Workflow 接口     | 使用正确的接口                         |
| `app_unavailable`             | 400       | 应用配置异常（如未配置模型）           | 在控制台检查应用配置                   |
| `provider_not_initialize`     | 400       | 模型供应商凭证未配置                   | 在控制台配置模型供应商                 |
| `provider_quota_exceeded`     | 400       | 模型供应商配额用尽                     | 充值或更换模型                         |
| `model_currently_not_support` | 400       | 模型不支持当前操作                     | 更换模型                               |
| `rate_limit_error`            | 429       | 请求频率超限                           | 降低请求频率或等待后重试               |
| `invalid_param`               | 400       | 请求参数无效                           | 检查请求体格式                         |
| `not_found`                   | 404       | 资源不存在                             | 检查 ID 是否正确                       |
| `internal_server_error`       | 500       | 服务端内部错误                         | 稍后重试，持续出现请查看日志           |

当收到 401 响应时，响应头中会包含：

```
WWW-Authenticate: Bearer realm="api"
```

### 6.3 限流机制

Dify 通过 Redis 实现活跃请求数限流（`api/core/app/features/rate_limiting/rate_limit.py`）：

- **并发限制**：同一应用的并发活跃请求数有上限
- **实现方式**：Redis Hash 跟踪每个客户端的活跃请求
- **超限响应**：返回 `429 Too Many Requests`

**限流处理最佳实践**：

```python
import time

def call_with_retry(func, max_retries=3, base_delay=1):
    """指数退避重试"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if hasattr(e, 'status_code') and e.status_code == 429:
                delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s
                print(f"限流，{delay}秒后重试...")
                time.sleep(delay)
            else:
                raise
    raise Exception("重试次数已用尽")
```

### 6.4 生产环境错误处理建议

1. **超时处理**：设置合理的超时时间（建议 60~120 秒），长文本生成需要更长时间
2. **重试策略**：对 429（限流）和 5xx（服务端错误）实施指数退避重试
3. **熔断降级**：连续失败达到阈值时，暂时切换到备用方案或提示用户稍后重试
4. **日志记录**：记录每次 API 调用的请求 ID、响应时间、token 用量，便于问题排查
5. **用量监控**：定期检查 token 消耗和费用，设置告警阈值

---

## 七、实践：用 curl 调用 Dify Chat API

### 7.1 准备工作

确保你已经：

1. 部署了 Dify 并创建了一个聊天助手应用（参考第 2~4 篇）
2. 获取了该应用的 API Key
3. 应用已发布（点击过"发布"按钮）

设置环境变量：

```bash
export DIFY_API_KEY="app-your-api-key-here"
export DIFY_BASE_URL="http://localhost/v1"
```

### 7.2 发送第一条消息（阻塞模式）

```bash
curl -X POST "${DIFY_BASE_URL}/chat-messages" \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "你好，请介绍一下你自己",
    "inputs": {},
    "response_mode": "blocking",
    "user": "test-user-001"
  }'
```

你会收到类似以下的响应：

```json
{
  "event": "message",
  "message_id": "a1b2c3d4-...",
  "conversation_id": "e5f6g7h8-...",
  "answer": "你好！我是 AI 写作助手...",
  "metadata": {
    "usage": {
      "prompt_tokens": 120,
      "completion_tokens": 200,
      "total_tokens": 320
    }
  },
  "created_at": 1712534400
}
```

### 7.3 继续多轮对话

使用上一步返回的 `conversation_id` 继续对话：

```bash
curl -X POST "${DIFY_BASE_URL}/chat-messages" \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "帮我写一篇关于 Dify 的文章吧",
    "inputs": {},
    "response_mode": "blocking",
    "conversation_id": "e5f6g7h8-上一步返回的ID",
    "user": "test-user-001"
  }'
```

### 7.4 体验流式响应

```bash
curl -X POST "${DIFY_BASE_URL}/chat-messages" \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -H "Content-Type: application/json" \
  -N \
  -d '{
    "query": "用三句话解释什么是机器学习",
    "inputs": {},
    "response_mode": "streaming",
    "user": "test-user-001"
  }'
```

> `-N` 参数禁用 curl 的输出缓冲，让你能看到 SSE 事件逐条输出。

你会看到类似以下的流式输出：

```
event: message
data: {"event": "message", "answer": "机器", ...}

event: message
data: {"event": "message", "answer": "学习", ...}

event: message
data: {"event": "message", "answer": "是", ...}

...

event: message_end
data: {"event": "message_end", "metadata": {"usage": {"total_tokens": 150}}}
```

### 7.5 上传文件并发送

```bash
# 步骤 1：上传文件
curl -X POST "${DIFY_BASE_URL}/files/upload" \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -F "file=@/path/to/document.pdf" \
  -F "user=test-user-001"

# 步骤 2：在对话中引用（使用上一步返回的文件 id）
curl -X POST "${DIFY_BASE_URL}/chat-messages" \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "帮我总结一下这份文档的要点",
    "inputs": {},
    "response_mode": "blocking",
    "user": "test-user-001",
    "files": [
      {
        "type": "document",
        "transfer_method": "local_file",
        "upload_file_id": "上一步返回的文件ID"
      }
    ]
  }'
```

---

## 八、进阶实践：用 Python 构建 Dify 客户端

下面我们用 Python 的 `requests` 库从零构建一个 Dify API 客户端，涵盖阻塞调用、流式调用和多轮对话。

### 8.1 基础客户端

```python
import requests
import json
import sseclient  # pip install sseclient-py

class DifyChat:
    """Dify Chat API 客户端"""

def __init__(self, api_key: str, base_url: str = "http://localhost/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

def send_message(
        self,
        query: str,
        user: str,
        inputs: dict | None = None,
        conversation_id: str | None = None,
        streaming: bool = False,
    ) -> dict | None:
        """发送聊天消息"""
        payload = {
            "query": query,
            "inputs": inputs or {},
            "response_mode": "streaming" if streaming else "blocking",
            "user": user,
        }
        if conversation_id:
            payload["conversation_id"] = conversation_id

if streaming:
            return self._send_streaming(payload)
        else:
            return self._send_blocking(payload)

def _send_blocking(self, payload: dict) -> dict:
        """阻塞模式调用"""
        response = requests.post(
            f"{self.base_url}/chat-messages",
            headers=self.headers,
            json=payload,
            timeout=120,
        )
        response.raise_for_status()
        return response.json()

def _send_streaming(self, payload: dict) -> None:
        """流式模式调用，逐字打印"""
        response = requests.post(
            f"{self.base_url}/chat-messages",
            headers=self.headers,
            json=payload,
            stream=True,
            timeout=120,
        )
        response.raise_for_status()

client = sseclient.SSEClient(response)
        full_answer = ""
        message_id = None
        conversation_id = None

for event in client.events():
            if event.event == "message":
                data = json.loads(event.data)
                chunk = data.get("answer", "")
                full_answer += chunk
                message_id = data.get("message_id")
                conversation_id = data.get("conversation_id")
                print(chunk, end="", flush=True)

elif event.event == "message_end":
                data = json.loads(event.data)
                usage = data.get("metadata", {}).get("usage", {})
                print(f"\n\n--- 生成完成 ---")
                print(f"Token 用量: {usage.get('total_tokens', 'N/A')}")

elif event.event == "error":
                data = json.loads(event.data)
                print(f"\n[错误] {data.get('message', '未知错误')}")
                return None

return {
            "answer": full_answer,
            "message_id": message_id,
            "conversation_id": conversation_id,
        }

def get_conversations(self, user: str, limit: int = 20) -> dict:
        """获取会话列表"""
        response = requests.get(
            f"{self.base_url}/conversations",
            headers=self.headers,
            params={"user": user, "limit": limit},
            timeout=30,
        )
        response.raise_for_status()
        return response.json()

def get_messages(
        self, user: str, conversation_id: str, limit: int = 20
    ) -> dict:
        """获取会话消息历史"""
        response = requests.get(
            f"{self.base_url}/messages",
            headers=self.headers,
            params={
                "user": user,
                "conversation_id": conversation_id,
                "limit": limit,
            },
            timeout=30,
        )
        response.raise_for_status()
        return response.json()
```

### 8.2 使用示例：多轮对话

```python
import os

# 初始化客户端
client = DifyChat(
    api_key=os.environ["DIFY_API_KEY"],
    base_url=os.environ.get("DIFY_BASE_URL", "http://localhost/v1"),
)

user_id = "python-demo-user"
conversation_id = None  # 第一轮对话不传

# 第一轮对话
print("=" * 50)
print("发送: 你好，请介绍一下你自己")
print("=" * 50)
result = client.send_message(
    query="你好，请介绍一下你自己",
    user=user_id,
    inputs={"writing_type": "公众号文章"},
)
print(f"\nAI: {result['answer'][:200]}...")
conversation_id = result["conversation_id"]

# 第二轮对话（传入 conversation_id 实现多轮）
print("\n" + "=" * 50)
print("发送: 帮我写一个关于 AI 的文章大纲")
print("=" * 50)
result = client.send_message(
    query="帮我写一个关于 AI 的文章大纲",
    user=user_id,
    conversation_id=conversation_id,
)
print(f"\nAI: {result['answer'][:200]}...")
```

### 8.3 使用示例：流式输出

```python
# 流式模式 -- 实时输出
print("=" * 50)
print("流式输出测试")
print("=" * 50)
client.send_message(
    query="用五句话解释什么是深度学习",
    user=user_id,
    streaming=True,
)
```

### 8.4 带错误处理的完整版

```python
import time
import requests

def chat_with_retry(
    client: DifyChat,
    query: str,
    user: str,
    max_retries: int = 3,
    **kwargs,
) -> dict | None:
    """带重试机制的对话调用"""
    for attempt in range(max_retries):
        try:
            return client.send_message(query=query, user=user, **kwargs)

except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code if e.response else 0
            error_body = {}
            try:
                error_body = e.response.json() if e.response else {}
            except ValueError:
                pass

if status_code == 401:
                print(f"[认证失败] API Key 无效: {error_body.get('message')}")
                return None  # 不重试

elif status_code == 429:
                delay = 2 ** attempt
                print(f"[限流] 第 {attempt + 1} 次重试，等待 {delay}s...")
                time.sleep(delay)

elif status_code >= 500:
                delay = 2 ** attempt
                print(f"[服务端错误 {status_code}] 第 {attempt + 1} 次重试...")
                time.sleep(delay)

else:
                code = error_body.get("code", "unknown")
                msg = error_body.get("message", str(e))
                print(f"[API 错误] {code}: {msg}")
                return None  # 客户端错误不重试

except requests.exceptions.Timeout:
            print(f"[超时] 第 {attempt + 1} 次重试...")
            time.sleep(2 ** attempt)

except requests.exceptions.ConnectionError:
            print(f"[连接失败] 请检查 Dify 服务是否运行")
            return None

print("[失败] 已达最大重试次数")
    return None
```

---

## 九、API 集成架构建议

当你准备在生产环境中集成 Dify API 时，以下架构建议值得参考：

### 9.1 推荐架构

```
┌──────────┐     ┌──────────────┐     ┌──────────────┐
│  你的前端  │────→│  你的后端服务  │────→│  Dify Server │
│  App/Web  │←────│  (API 代理)   │←────│  Service API │
└──────────┘     └──────────────┘     └──────────────┘
```

**不要在前端直接调用 Dify API**。正确的做法是：

1. **前端** → 调用你自己的后端接口
2. **你的后端** → 携带 API Key 调用 Dify Service API
3. **Dify** → 返回结果给你的后端
4. **你的后端** → 转发（或加工后）返回给前端

这样做的好处：

- API Key 安全存储在后端，不暴露给前端
- 可以在中间层做用户鉴权、日志记录、费用统计
- 可以对 Dify 响应做二次加工（如格式转换、敏感词过滤）
- 可以实现缓存、限流等策略

### 9.2 流式代理示例（Node.js）

```javascript
// Express 中间件：代理 Dify 流式响应给前端
app.post('/api/chat', async (req, res) => {
  const { query, conversationId } = req.body
  const userId = req.user.id  // 从你的认证中获取

res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Cache-Control', 'no-cache')
  res.setHeader('Connection', 'keep-alive')

const stream = await chatClient.createChatMessage({
    inputs: {},
    query,
    user: userId,
    conversation_id: conversationId,
    response_mode: 'streaming'
  })

for await (const event of stream) {
    res.write(`event: ${event.event}\ndata: ${JSON.stringify(event.data)}\n\n`)
  }

res.end()
})
```

---

## 小结与预告

本篇我们深入学习了 Dify 的 API 集成方法，你现在掌握了：

1. **三层 API 体系**的区别：Console API（管理）、Service API（集成）、Web API（WebApp）
2. **Bearer Token 鉴权**机制和 API Key 的安全管理
3. **四类核心接口**：Chat（对话）、Completion（补全）、Workflow（工作流）、File Upload（文件上传）
4. **流式 vs 阻塞**响应的选择策略和 SSE 事件类型
5. **Node.js SDK** 的完整使用方法
6. **Python 客户端**的从零构建，包括阻塞/流式/多轮对话/错误处理
7. **生产环境**的架构建议和最佳实践

到这里，模块一"认识 Dify"全部完成。你已经从零学会了：理解 Dify → 部署 Dify → 配置模型 → 创建应用 → API 集成。

接下来我们进入**模块二：工作流 -- Dify 的核心引擎**。工作流是 Dify 最强大的功能之一，它让你可以用可视化方式编排复杂的 AI 逻辑，而不只是简单的一问一答。

**下一篇：《工作流入门 -- 用可视化方式编排 AI 逻辑》** 将带你理解 Dify 工作流的设计理念，掌握工作流编辑器的基本操作，并创建你的第一个多步骤工作流。

---

**今日互动**：你计划把 Dify 集成到什么产品中？是移动端 App、微信小程序、企业内部系统，还是独立的 AI 产品？在评论区分享你的集成场景，一起探讨最佳方案！

上一篇：Dify 系列教程 | 第 4 篇：5 分钟创建你的第一个 AI 聊天助手已经是最后一篇啦

你觉得文章怎么样