Dify低代码Agent开发

慕鱼而网

48人浏览 · 2026-06-27 00:18:53

慕鱼而网 · 2026-06-27 00:18:53 发布

Dify 低代码 Agent 开发：Workflow、知识库与 Agent 编排一站式实战

导读： 不是每个团队都有精力从零搭建 LangChain/LangGraph 代码栈。Dify 作为国内最活跃的开源 LLM 应用平台，提供了可视化工作流编排、知识库管理、Agent 策略配置三大核心能力。本文将从 Workflow 节点拆解、知识库 RAG 原理、Agent 编排策略三个维度，带你用"低代码"的方式构建生产级 AI 应用。

目录

Dify 是什么？为什么选择它？
Dify 核心架构速览
Workflow：可视化编排 LLM 工作流
知识库：从文档到 RAG 的全链路
Agent 编排：Function Calling 与策略配置
实战一：构建智能客服系统（Workflow + 知识库）
实战二：构建数据分析 Agent（代码执行 + 多工具）
实战三：通过 API 将 Dify 嵌入业务系统
进阶：插件开发与私有化部署
总结与选型建议

一、Dify 是什么？为什么选择它？

1.1 Dify 的定位

┌─────────────────────────────────────────────────────────┐
│                    LLM 应用开发方式                        │
│                                                          │
│  纯代码                         低代码           无代码    │
│  ────────────┬──────────────┬──────────────┬───────────  │
│  LangChain   │  LangGraph   │    Dify      │   Coze     │
│  LlamaIndex  │  CrewAI      │  (开源/云)    │  (扣子)    │
│              │              │              │            │
│  最大灵活性   │              │  平衡效率与   │  最快上手   │
│  最多代码     │              │  灵活性       │  最少控制   │
└──────────────────────────────────────────────────────────┘

Dify 的核心价值主张是：用可视化编排 + 配置驱动，覆盖 80% 的 LLM 应用场景，同时保留 API 和插件扩展能力来覆盖剩下的 20%。

1.2 Dify vs 纯代码框架

维度	Dify	LangChain/LangGraph
开发效率	⭐⭐⭐⭐⭐ 拖拽式编排	⭐⭐ 需要编写大量代码
灵活性	⭐⭐⭐ 受限于平台能力	⭐⭐⭐⭐⭐ 完全自由
知识库/RAG	内置可视化配置	需要手动集成
多模型管理	统一管理，一键切换	代码中切换
运维监控	内置日志、费用统计	需要 LangSmith 等
协作	多人协作、权限管理	Git 协作
私有化部署	社区版免费部署	无额外依赖
适合人群	产品/运营/全栈	资深开发者

二、Dify 核心架构速览

2.1 四大核心模块

┌─────────────────────────────────────────────────────────┐
│                     Dify 平台                             │
│                                                          │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────────┐  │
│  │  应用层       │  │   Workflow   │  │   知识库       │  │
│  │              │  │              │  │               │  │
│  │ • 聊天助手    │  │ • 可视化编排  │  │ • 文档导入     │  │
│  │ • Agent      │  │ • 节点系统   │  │ • Embedding   │  │
│  │ • 文本生成    │  │ • 变量管理   │  │ • 向量检索     │  │
│  │ • API 服务   │  │ • 调试工具   │  │ • 分段策略     │  │
│  └──────┬───────┘  └──────┬───────┘  └───────┬───────┘  │
│         │                 │                   │          │
│         └─────────────────┼───────────────────┘          │
│                           │                              │
│  ┌────────────────────────┴──────────────────────────┐  │
│  │              模型供应商层                            │  │
│  │  OpenAI │ 通义千问 │ 文心一言 │ DeepSeek │ 本地模型  │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

2.2 应用类型

Dify 支持四种应用类型，按灵活性递增排列：

类型	适用场景	灵活性
聊天助手	简单的对话机器人	⭐
Agent	需要调用工具的智能助手	⭐⭐⭐
文本生成	翻译、摘要、写作等单次任务	⭐
Workflow	复杂的多步骤业务流程	⭐⭐⭐⭐⭐

三、Workflow：可视化编排 LLM 工作流

Workflow 是 Dify 最强大的功能。它允许你通过拖拽节点、连线、配置参数来构建复杂的 LLM 应用。

3.1 节点类型全览

┌─────────────────────────────────────────────────────────┐
│                 Dify Workflow 节点分类                    │
│                                                          │
│  【开始/结束】                                            │
│  ├── Start：定义输入变量                                 │
│  └── End：定义输出变量                                   │
│                                                          │
│  【LLM 相关】                                             │
│  ├── LLM：调用大语言模型                                 │
│  ├── Knowledge Retrieval：从知识库检索                   │
│  └── Question Classifier：问题分类                       │
│                                                          │
│  【逻辑控制】                                             │
│  ├── IF/ELSE：条件分支                                  │
│  ├── Iteration：循环（遍历数组）                         │
│  └── Variable Aggregator：变量聚合                      │
│                                                          │
│  【工具调用】                                             │
│  ├── Code：执行 Python/JS 代码                          │
│  ├── HTTP Request：调用外部 API                         │
│  ├── Tool：使用内置/自定义工具                          │
│  └── Parameter Extractor：从文本中提取结构化参数        │
│                                                          │
│  【数据处理】                                             │
│  ├── Template：Jinja-2 模板转换                         │
│  ├── Variable Assigner：变量赋值                        │
│  └── Doc Extractor：文档内容提取                        │
└─────────────────────────────────────────────────────────┘

3.2 核心节点深度解析

LLM 节点

这是最核心的节点，配置要点：

┌─────────────────────────────────────────┐
│           LLM 节点配置面板                │
│                                          │
│  模型：[下拉选择] GPT-4 / Qwen / ...     │
│                                          │
│  SYSTEM Prompt：                         │
│  ┌──────────────────────────────────┐   │
│  │ 你是一位{{role}}，擅长{{skill}}。  │   │
│  │ 请用{{language}}回答。            │   │
│  │              ↑ 变量引用           │   │
│  └──────────────────────────────────┘   │
│                                          │
│  USER Prompt：                           │
│  ┌──────────────────────────────────┐   │
│  │ {{#context#}}  ← 上下文（知识库）   │   │
│  │                                   │   │
│  │ 用户问题：{{#sys.query#}}         │   │
│  └──────────────────────────────────┘   │
│                                          │
│  Memory：[开启] 窗口大小：10 轮          │
│  Temperature：[0.7          ]           │
└─────────────────────────────────────────┘

Prompt 中的变量引用语法：

{# 输入变量 #}
{{#sys.query#}}              {# 用户原始输入 #}
{{#sys.files#}}              {# 用户上传的文件 #}

{# 上游节点输出 #}
{{#node_name.output_field#}} {# 引用特定节点的输出 #}

{# 会话变量 #}
{{#conversation_id#}}        {# 当前会话 ID #}

{# 环境变量 #}
{{#env.API_KEY#}}            {# 平台级环境变量 #}

Code 节点

Code 节点让你在 Workflow 中执行自定义 Python/JavaScript 代码，是扩展 Workflow 能力的关键：

# Dify Code 节点的 Python 代码示例
# 输入变量：上游节点传递的 data
# 输出：通过 return 返回

import json
from datetime import datetime

def main(data: dict) -> dict:
    """
    data 包含：
    - llm_output: LLM 节点的输出文本
    - user_info: 用户信息
    """
    # 解析 LLM 输出
    llm_text = data.get("llm_output", "")
    
    # 提取结构化信息
    result = {
        "processed": True,
        "timestamp": datetime.now().isoformat(),
        "word_count": len(llm_text),
        "summary": llm_text[:200] if len(llm_text) > 200 else llm_text,
    }
    
    # 尝试提取 JSON
    try:
        # 假设 LLM 输出中包含 JSON
        import re
        json_match = re.search(r'\{.*?\}', llm_text, re.DOTALL)
        if json_match:
            result["extracted_data"] = json.loads(json_match.group())
    except:
        result["extracted_data"] = None
    
    return result

HTTP Request 节点

将外部 API 的能力引入 Workflow：

# HTTP Request 节点配置示例

# 请求配置：
# Method: POST
# URL: https://api.weather.com/v1/forecast
# Headers: {"Authorization": "Bearer {{#env.WEATHER_API_KEY}}"}
# Body:
{
    "city": "{{#start.city}}",
    "date": "{{#start.date}}",
    "units": "metric"
}

# 响应处理：
# 响应会被自动解析为 JSON，下游节点可以用 {{#http_node.body.temperature}} 引用

IF/ELSE 条件节点

                 ┌──────────┐
                 │   LLM    │  (回答用户问题)
                 └────┬─────┘
                      │
              ┌───────┴───────┐
              │ IF/ELSE       │
              │               │
              │ 条件1：        │
              │ {{#llm.output │
              │  .contains(   │
              │  "投诉")}}    │
              │               │
              ├─── 是 ────────▶ [转人工客服]
              │               │
              │ 条件2：        │
              │ {{#llm.output │
              │  .contains(   │
              │  "购买")}}    │
              │               │
              ├─── 是 ────────▶ [创建订单]
              │               │
              └─── 否 ────────▶ [正常回复]

Iteration 循环节点

# Iteration 节点内的处理逻辑

# 输入：一个数组 ["Python", "JavaScript", "Go"]
# 每次迭代，当前元素通过 {{#iteration.item}} 引用

# 在迭代体中：
# LLM 节点 Prompt：请用一句话介绍 {{#iteration.item}} 语言。
# 输出收集到数组中，通过 Variable Aggregator 聚合

3.3 变量系统

Dify Workflow 的变量系统分为三层：

┌──────────────────────────────────────────────────────┐
│                  Dify 变量作用域                       │
│                                                       │
│  【环境变量】（全局）                                   │
│  ├── API Keys                                         │
│  ├── 系统配置常量                                       │
│  └── 通过 {{#env.NAME}} 引用                          │
│                                                       │
│  【会话变量】（跨轮次）                                  │
│  ├── 对话历史                                          │
│  ├── 用户身份信息                                       │
│  └── 通过会话变量节点设置/读取                           │
│                                                       │
│  【节点变量】（单次运行）                                │
│  ├── 各节点的输入/输出                                  │
│  └── 通过 {{#node_name.field}} 引用                   │
└──────────────────────────────────────────────────────┘

四、知识库：从文档到 RAG 的全链路

知识库是 Dify 解决 LLM 幻觉问题的核心武器。它让 LLM 能够基于你的私有文档回答问题。

4.1 知识库工作原理

┌─────────────────────────────────────────────────────────┐
│                   Dify 知识库全链路                        │
│                                                          │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌───────┐ │
│  │ 文档导入  │──▶│ 文本分段  │──▶│ 向量化   │──▶│ 索引  │ │
│  │          │   │          │   │          │   │       │ │
│  │ • PDF    │   │ • 自动分段│   │ • Embed  │   │• 向量  │ │
│  │ • Word   │   │ • 自定义  │   │ • 模型   │   │  数据库│ │
│  │ • Markdown│  │   分隔符  │   │   选择   │   │       │ │
│  │ • TXT    │   │ • 分段长度│   │          │   │       │ │
│  │ • 网页   │   │ • 重叠   │   │          │   │       │ │
│  └──────────┘   └──────────┘   └──────────┘   └───────┘ │
│                                                          │
│  检索阶段：                                               │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐             │
│  │ 用户问题  │──▶│ 向量化    │──▶│ 相似度    │             │
│  │          │   │ (Embed)  │   │ 搜索 TopK │             │
│  └──────────┘   └──────────┘   └────┬─────┘             │
│                                     │                    │
│                                     ▼                    │
│                              ┌──────────┐               │
│                              │ 重排序    │ (Rerank)      │
│                              │ (可选)   │               │
│                              └────┬─────┘               │
│                                   │                      │
│                                   ▼                      │
│                           返回最相关的文档片段            │
└─────────────────────────────────────────────────────────┘

4.2 分段策略：知识库效果的"胜负手"

分段（Chunking）是知识库最关键的配置，直接影响检索质量：

策略	适用场景	示例
自动分段	通用文档	按段落、标题自动切分
自定义分隔符	结构化文档	`\n\n`（段落）、`###`（Markdown 标题）
固定长度	代码、日志	每 500 token 一段，重叠 50 token
递归分段	复杂嵌套文档	先按大标题分，再按小标题分

# Dify 知识库 API 调用示例
import requests

API_BASE = "https://your-dify-instance.com/v1"
API_KEY = "dataset-xxx"

# 方式1: 通过 API 上传文档
def upload_document(file_path: str, dataset_id: str):
    """上传文档到知识库"""
    url = f"{API_BASE}/datasets/{dataset_id}/document/create-by-file"
    
    headers = {"Authorization": f"Bearer {API_KEY}"}
    
    with open(file_path, "rb") as f:
        files = {"file": f}
        # 注意：嵌套配置需要 JSON 序列化后作为表单字段传递
        import json
        data = {
            "indexing_technique": "high_quality",
            "process_rule": json.dumps({
                "mode": "custom",
                "rules": {
                    "segmentation": {
                        "separator": "\n",
                        "max_tokens": 500,
                        "chunk_overlap": 50,
                    }
                }
            })
        }
        response = requests.post(url, headers=headers, files=files, data=data)
    
    return response.json()


# 方式2: 直接通过文本创建
def create_document_by_text(text: str, dataset_id: str, title: str):
    """通过文本内容创建知识库文档"""
    url = f"{API_BASE}/datasets/{dataset_id}/document/create-by-text"
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    
    payload = {
        "name": title,
        "text": text,
        "indexing_technique": "high_quality",
        "process_rule": {
            "mode": "automatic",
        }
    }
    
    response = requests.post(url, headers=headers, json=payload)
    return response.json()

4.3 检索模式对比

Dify 支持三种检索模式，按精确度递增：

经济模式（Economy）               高质量模式（High Quality）
───────────────                 ────────────────
关键词匹配（BM25）               向量检索 + Rerank
模糊匹配，速度快                 语义匹配，更精准
适合：FAQ、精确匹配              适合：长文档、语义理解

4.4 知识库 Workflow 配置示例

在 Workflow 中使用知识库节点时的配置：

# 知识库检索节点配置（逻辑示意）
节点类型: Knowledge Retrieval
配置:
  知识库: [选择已创建的知识库]
  检索模式: high_quality
  查询变量: "{{#llm_query_node.output}}"  # 用 LLM 改写后的查询
  TopK: 5                                    # 返回最相关的 5 段
  分数阈值: 0.6                              # 低于此分数的结果丢弃
  重排序: 启用                                # 使用 Rerank 模型

# 在 LLM 节点中使用检索结果：
# {{#knowledge_node.result}}  → 包含检索到的文档片段列表

五、Agent 编排：Function Calling 与策略配置

Dify 的 Agent 应用类型，本质上是ReAct + Function Calling 的可视化配置版本。

5.1 Agent 策略选择

Dify Agent 提供两种核心策略：

策略	全称	适用场景
ReAct	Reasoning + Acting	通用场景，兼容性好
Function Calling	原生函数调用	模型本身支持 FC（GPT-4, Qwen-Max 等）

ReAct vs Function Calling 的差异：

ReAct:
  LLM 输出格式：Thought → Action → Action Input
  Prompt 工程驱动，所有模型通用
  输出格式可能不稳定（解析错误）

Function Calling:
  LLM 直接输出结构化的 tool_calls
  模型原生支持，格式可靠
  需要模型支持（不是所有模型都支持）

5.2 工具配置

在 Dify Agent 中可以配置三类工具：

┌──────────────────────────────────────────────────────┐
│                 Dify Agent 工具面板                    │
│                                                       │
│  【内置工具】                                          │
│  ├── 知识库检索（关联已创建的知识库）                    │
│  ├── 代码执行（Python/JS Sandbox）                    │
│  ├── 图片生成（DALL-E / Stable Diffusion）            │
│  └── 网页搜索（SerpAPI / Bing / Google）              │
│                                                       │
│  【自定义 API 工具】                                   │
│  ├── HTTP Request 工具                               │
│  └── OpenAPI / Swagger 导入                          │
│                                                       │
│  【自定义插件工具】                                     │
│  ├── 安装社区插件                                      │
│  └── 开发自定义插件（见第九节）                         │
└──────────────────────────────────────────────────────┘

5.3 Agent 配置最佳实践

# Dify Agent 的推荐配置（逻辑示意）
Agent 配置:
  模型: "qwen-max"              # 强力模型，Agent 对推理能力要求高
  策略: "Function Calling"      # 首选 FC，失败时降级到 ReAct
  
  工具:
    - knowledge_search:         # 知识库检索
        关联知识库: ["产品文档", "FAQ"]
        最大召回数: 5
    
    - code_executor:            # 代码执行
        语言: Python
        超时: 30s
    
    - custom_api:               # 自定义 API
        name: "查询订单"
        method: GET
        url: "https://api.example.com/orders/{{order_id}}"
        
  System Prompt: |
    你是羽哥电商的智能客服助手。
    
    处理流程：
    1. 如果用户询问产品信息，先检索知识库
    2. 如果用户询问订单状态，调用订单查询 API
    3. 如果需要计算（如折扣、运费），使用代码执行器
    4. 回复时引用来源，保持友好专业的语气
    
  Memory: 20 轮
  Temperature: 0.1              # Agent 建议低温

六、实战一：构建智能客服系统（Workflow + 知识库）

6.1 业务场景

为一个电商平台构建智能客服，需求如下：

用户询问产品信息 → 从知识库检索回答
用户询问订单状态 → 调用订单 API
用户投诉或要求退款 → 转人工客服
用户要求计算折扣 → 使用代码节点计算

6.2 Workflow 设计

                    ┌────────────┐
                    │   Start    │
                    │ user_query │
                    └─────┬──────┘
                          │
                          ▼
                  ┌───────────────┐
                  │   问题分类     │
                  │ (LLM节点)     │
                  │               │
                  │ 输出:         │
                  │ category:     │
                  │ product/order │
                  │ /complaint    │
                  │ /calculation  │
                  └───────┬───────┘
                          │
                ┌─────────┴─────────┐
                │   IF/ELSE         │
                │                   │
      ┌─────────┤ category=product  ├─────────┐
      │         │ category=order    │         │
      │         │ category=complaint│         │
      │         │ category=calc     │         │
      ▼         └───────────────────┘         │
┌──────────┐                                  │
│ 知识库检索│     ...  (其他分支)               │
└────┬─────┘                                  │
     │                                        │
     ▼                                        │
┌──────────┐                                  │
│ LLM:     │◀─────────────────────────────────┘
│ 回答问题  │    所有分支最终汇聚到这里
└────┬─────┘
     │
     ▼
┌──────────┐
│  End     │
└──────────┘

6.3 Step-by-Step 配置

Step 1：创建知识库

# 1. 在 Dify 界面：知识库 → 创建知识库
# 2. 上传文件：product_manual.pdf, faq.xlsx, shipping_policy.md
# 3. 设置分段策略：
#    - 模式：高质量
#    - 分隔符：\n\n### 
#    - 最大 token：500
#    - 重叠：50
# 4. 等待索引完成

Step 2：构建 Workflow

以下是具体节点的配置（代码形式展示，实际在 UI 中配置）：

# === 节点1：Start ===
输入变量：
  user_query: text     # 用户问题
  user_id: text        # 用户 ID（可选）

# === 节点2：LLM 分类器 ===
# 模型：qwen-plus (temperature=0.1)
SYSTEM:
"""
你是一个问题分类专家。请将用户问题分类为以下之一：
- product：关于产品信息、功能、价格的问题
- order：关于订单状态、物流、退换货的问题
- complaint：投诉、不满、要求人工服务
- calculation：需要数学计算的问题（折扣、总价等）

只返回分类结果，不要解释。
"""

USER:
"""
用户问题：{{#start.user_query}}
"""

# === 节点3：IF/ELSE 分支 ===
条件：
  - IF {{#classifier.text}} contains "product"
    → 进入知识库检索分支
  - ELIF {{#classifier.text}} contains "order"  
    → 进入订单查询分支
  - ELIF {{#classifier.text}} contains "complaint"
    → 进入转人工分支
  - ELSE
    → 进入计算分支

# === 节点4a：知识库检索 ===
配置：
  知识库：电商产品知识库
  检索模式：高质量
  查询变量：{{#start.user_query}}
  TopK：5
  分数阈值：0.6

# === 节点4b：HTTP Request (订单查询) ===
Method: GET
URL: https://your-api.com/orders?user_id={{#start.user_id}}
Headers: {"Authorization": "Bearer {{#env.ORDER_API_KEY}}"}

# === 节点4c：转人工 ===
Template 节点：
"""
已为您转接人工客服。
用户问题：{{#start.user_query}}
用户ID：{{#start.user_id}}
请稍候，客服人员将很快为您服务。
"""

# === 节点5：LLM 回答生成 ===
SYSTEM:
"""
你是羽哥电商的客服助手。

使用以下检索到的信息回答问题：
{{#knowledge_retrieval.result}}

使用以下订单信息（如有）：
{{#order_query.body}}

要求：
1. 基于提供的资料回答，不要编造
2. 回答简洁友好，200字以内
3. 如果是产品推荐，给出2-3个选项
"""

6.4 通过 API 测试 Workflow

import requests

# Workflow API 调用
def test_customer_service_workflow(user_query: str, user_id: str = "guest"):
    url = "https://your-dify-instance.com/v1/workflows/run"
    
    headers = {
        "Authorization": "Bearer app-xxx",  # Workflow 应用的 API Key
        "Content-Type": "application/json",
    }
    
    payload = {
        "inputs": {
            "user_query": user_query,
            "user_id": user_id,
        },
        "response_mode": "blocking",  # 同步模式
        "user": user_id,
    }
    
    response = requests.post(url, headers=headers, json=payload)
    return response.json()


# 测试
test_cases = [
    ("你们的蓝牙耳机续航多久？", "user_001"),
    ("订单 #12345 到哪了？", "user_002"),
    ("我要投诉，产品质量太差了！", "user_003"),
    ("买2件打8折，原价299，省了多少钱？", "user_004"),
]

for query, uid in test_cases:
    print(f"\n👤 用户：{query}")
    result = test_customer_service_workflow(query, uid)
    print(f"🤖 客服：{result.get('data', {}).get('outputs', {}).get('text', '无响应')}")

七、实战二：构建数据分析 Agent（代码执行 + 多工具）

7.1 场景设计

构建一个数据分析 Agent，能够：

接收用户上传的 CSV/Excel 文件
自动理解数据结构
根据用户需求执行 Python 分析代码
生成可视化图表
输出分析报告

7.2 Agent 配置

# Agent 应用配置

应用类型: Agent
模型: qwen-max (Function Calling 模式)

System Prompt: |
  你是一个数据分析专家 Agent。
  
  你可以使用以下工具：
  1. 代码执行器 - 运行 Python 代码进行数据分析
  2. 知识库检索 - 查找数据分析方法的最佳实践
  
  工作流程：
  1. 用户上传数据文件后，先用代码执行器查看数据结构
  2. 根据用户的分析需求，编写并执行 Python 代码
  3. 如果需要可视化，使用 matplotlib 生成图表
  4. 总结分析结果并给出建议
  
  重要规则：
  - 分析代码要包含错误处理
  - 图表保存后告知用户路径
  - 数据分析要严谨，标明数据的局限性

工具:
  - 代码执行器 (Python)
      超时: 120s
      预装库: pandas, numpy, matplotlib, seaborn, scipy
      
  - 知识库检索
      关联知识库: [数据分析方法知识库]
      TopK: 3
      
  - 自定义 API 工具 (可选):
      name: "查询行业基准数据"
      url: "https://api.benchmark.com/{{industry}}/{{metric}}"

7.3 知识库准备

为 Agent 准备一个"数据分析方法"知识库：

# 数据分析知识库文档示例


## 缺失值处理
- df.isnull().sum() 查看缺失值
- df.dropna() 删除含缺失值的行
- df.fillna(df.mean()) 用均值填充缺失值

## 异常值检测
- IQR 方法：Q1 - 1.5*IQR < normal < Q3 + 1.5*IQR
- Z-score 方法：|z| > 3 视为异常

## 常用分析方法
- df.describe() 描述性统计
- df.corr() 相关性分析
- df.groupby('column').agg(['mean', 'sum', 'count']) 分组聚合

## 可视化最佳实践
- 趋势用折线图：plt.plot()
- 分布用直方图：plt.hist()
- 类别对比用柱状图：plt.bar()
- 相关性用热力图：sns.heatmap()

7.4 与 Agent 交互示例

用户：上传了 sales_2024.csv，请帮我分析销售趋势

Agent：
💭 Thought: 用户上传了销售数据，我需要先查看数据结构。
🔧 Action: code_executor
📝 Code:
```python
import pandas as pd
df = pd.read_csv('sales_2024.csv')
print("数据概况：")
print(f"行数：{len(df)}，列数：{len(df.columns)}")
print(f"列名：{list(df.columns)}")
print(f"数据类型：\n{df.dtypes}")
print(f"\n前5行：\n{df.head()}")
print(f"\n描述性统计：\n{df.describe()}")

👁️ Observation: …
💭 Thought: 数据包含日期、销售额、产品类别等字段。让我分析趋势。
🔧 Action: code_executor
📝 Code:

import pandas as pd
import matplotlib.pyplot as plt
# ... 趋势分析代码

👁️ Observation: 图表已保存

📝 Final Answer:
根据对 sales_2024.csv 的分析，发现以下关键趋势：

全年销售额呈上升趋势，Q4环比增长35%
电子产品类目贡献了60%的营收
12月销售额达到峰值，与促销季吻合
建议：加大Q4库存准备，重点推广电子产品。


---

## 八、实战三：通过 API 将 Dify 嵌入业务系统

Dify 的每个应用都自动生成 API 端点，可以直接嵌入到你的业务系统中。

### 8.1 API 类型

| API 类型 | 端点 | 适用场景 |
|---------|------|---------|
| Chat (流式) | `POST /v1/chat-messages` | 对话应用，支持 SSE 流式输出 |
| Workflow (同步) | `POST /v1/workflows/run` | 工作流应用，一次性返回 |
| Completion | `POST /v1/completion-messages` | 文本生成应用 |
| Files | `POST /v1/files/upload` | 上传文件到应用 |

### 8.2 完整的 API 集成示例

```python
"""
Dify 业务集成 SDK 封装
适用于将 Dify 应用嵌入到 Flask/FastAPI/Django 项目中
"""
import requests
import json
from typing import Optional, Dict, Any, Generator


class DifyClient:
    """Dify API 客户端封装"""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
    
    # ========== 对话应用 ==========
    def chat(
        self,
        query: str,
        user: str,
        conversation_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """发送对话消息（阻塞模式）"""
        url = f"{self.base_url}/v1/chat-messages"
        
        payload = {
            "inputs": {},
            "query": query,
            "response_mode": "blocking",
            "user": user,
        }
        
        if conversation_id:
            payload["conversation_id"] = conversation_id
        
        resp = self.session.post(url, json=payload)
        resp.raise_for_status()
        return resp.json()
    
    def chat_stream(
        self,
        query: str,
        user: str,
        conversation_id: Optional[str] = None,
    ) -> Generator[str, None, None]:
        """发送对话消息（流式模式）"""
        url = f"{self.base_url}/v1/chat-messages"
        
        payload = {
            "inputs": {},
            "query": query,
            "response_mode": "streaming",
            "user": user,
        }
        
        if conversation_id:
            payload["conversation_id"] = conversation_id
        
        resp = self.session.post(url, json=payload, stream=True)
        resp.raise_for_status()
        
        for line in resp.iter_lines():
            if line:
                line_str = line.decode('utf-8')
                if line_str.startswith('data: '):
                    data = json.loads(line_str[6:])
                    if data.get('event') == 'message':
                        yield data.get('answer', '')
    
    # ========== Workflow 应用 ==========
    def run_workflow(
        self,
        inputs: Dict[str, Any],
        user: str,
        response_mode: str = "blocking",
    ) -> Dict[str, Any]:
        """运行 Workflow"""
        url = f"{self.base_url}/v1/workflows/run"
        
        payload = {
            "inputs": inputs,
            "response_mode": response_mode,
            "user": user,
        }
        
        resp = self.session.post(url, json=payload)
        resp.raise_for_status()
        return resp.json()
    
    # ========== 文件上传 ==========
    def upload_file(
        self,
        file_path: str,
        user: str,
    ) -> Dict[str, Any]:
        """上传文件（用于多模态应用）"""
        url = f"{self.base_url}/v1/files/upload"
        
        headers = {"Authorization": f"Bearer {self.api_key}"}
        # 注意：文件上传不用 Content-Type: application/json
        
        with open(file_path, 'rb') as f:
            files = {'file': f}
            data = {'user': user}
            resp = requests.post(url, headers=headers, files=files, data=data)
        
        resp.raise_for_status()
        return resp.json()
    
    # ========== 知识库管理 ==========
    def list_datasets(self, page: int = 1, limit: int = 20) -> Dict[str, Any]:
        """获取知识库列表"""
        url = f"{self.base_url}/v1/datasets"
        params = {"page": page, "limit": limit}
        resp = self.session.get(url, params=params)
        resp.raise_for_status()
        return resp.json()
    
    def search_knowledge(
        self,
        dataset_id: str,
        query: str,
        top_k: int = 5,
    ) -> Dict[str, Any]:
        """直接搜索知识库（不经过 LLM）"""
        url = f"{self.base_url}/v1/datasets/{dataset_id}/retrieve"
        
        payload = {
            "query": query,
            "retrieval_model": {
                "search_method": "hybrid_search",
                "top_k": top_k,
                "score_threshold": 0.5,
            }
        }
        
        resp = self.session.post(url, json=payload)
        resp.raise_for_status()
        return resp.json()


# ========== 在 FastAPI 中集成 ==========
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

# 初始化 Dify 客户端
dify_customer_service = DifyClient(
    base_url="https://your-dify.com",
    api_key="app-xxxx",
)

dify_data_analyst = DifyClient(
    base_url="https://your-dify.com",
    api_key="app-yyyy",
)


@app.post("/api/cs/chat")
async def customer_service_chat(query: str, user_id: str):
    """客服对话接口"""
    try:
        result = dify_customer_service.chat(query=query, user=user_id)
        return {
            "answer": result["answer"],
            "conversation_id": result.get("conversation_id"),
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/cs/chat/stream")
async def customer_service_chat_stream(query: str, user_id: str):
    """客服对话接口（流式）"""
    def generate():
        try:
            for chunk in dify_customer_service.chat_stream(query, user_id):
                yield f"data: {json.dumps({'content': chunk})}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")


@app.post("/api/analyst/analyze")
async def analyze_data(file_url: str, question: str, user_id: str):
    """数据分析接口"""
    result = dify_data_analyst.run_workflow(
        inputs={
            "file_url": file_url,
            "user_question": question,
        },
        user=user_id,
    )
    return result


# 运行：uvicorn app:app --reload

九、进阶：插件开发与私有化部署

9.1 Dify 插件开发

Dify 支持通过插件扩展 Tool 能力。以下是开发一个自定义 Tool 插件的基本流程：

# dify_plugin/weather_tool.py
"""
Dify 自定义工具插件示例：天气查询工具

打包后上传到 Dify 平台即可在 Agent 中使用。
"""
from typing import Any, Dict
from dify_plugin import ToolProvider


class WeatherTool(ToolProvider):
    """天气查询工具"""
    
    name = "weather_query"
    
    # 工具参数定义
    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，例如：北京、上海",
                },
                "date": {
                    "type": "string",
                    "description": "日期，格式 YYYY-MM-DD，默认为今天",
                },
            },
            "required": ["city"],
        }
    
    # 工具描述（LLM 据此判断何时使用）
    @property
    def description(self) -> Dict[str, Any]:
        return {
            "zh_Hans": "查询指定城市的天气情况，包括温度、湿度、天气状况、风力。",
            "en_US": "Query weather information for a specified city.",
        }
    
    # 工具执行逻辑
    def _invoke(self, parameters: Dict[str, Any]) -> str:
        city = parameters.get("city")
        date = parameters.get("date", "today")
        
        # 调用真实的天气 API
        import requests
        api_key = self.credentials.get("weather_api_key")
        
        response = requests.get(
            "https://api.weatherapi.com/v1/forecast.json",
            params={"key": api_key, "q": city, "dt": date},
        )
        
        data = response.json()
        current = data.get("current", {})
        
        return (
            f"城市：{city}\n"
            f"日期：{date}\n"
            f"天气：{current.get('condition', {}).get('text')}\n"
            f"温度：{current.get('temp_c')}℃\n"
            f"湿度：{current.get('humidity')}%\n"
            f"风力：{current.get('wind_kph')} km/h"
        )


# 注册插件
def register():
    return WeatherTool()

9.2 Docker 私有化部署

# 1. 克隆项目
git clone https://github.com/langgenius/dify.git
cd dify/docker

# 2. 配置环境变量
cp .env.example .env
# 编辑 .env，配置必要的密钥和数据库连接

# 3. 启动服务
docker compose up -d

# 4. 访问
# http://localhost:3000  → Web 控制台
# http://localhost:5001  → API 服务

# 5. 生产环境注意事项：
# - 配置 HTTPS
# - 使用外部 PostgreSQL / Redis
# - 配置对象存储（S3/MinIO）替代本地存储
# - 设置资源限制（特别是 Sandbox 容器）

十、总结与选型建议

10.1 本文回顾

知识点	核心要点
Workflow	可视化编排的核心，掌控 10+ 节点类型构建复杂流程
知识库	分段策略（分隔符/长度/重叠）是检索质量的生死线
Agent	Function Calling 模式优先，System Prompt 设计是关键
API 集成	每个应用自动生成 REST API，支持同步/流式
插件开发	自定义 Tool 扩展 Agent 能力，弥补低代码的灵活性短板

10.2 Dify 最适合的场景

Dify 特别适合：
✅ 需要快速原型验证的 LLM 项目
✅ 业务人员参与配置的 AI 应用
✅ 知识库 + 对话的企业内部应用
✅ 多模型统一管理的平台
✅ 需要私有化部署的开源方案

Dify 不太适合：
❌ 需要非常复杂的自定义控制流（考虑 LangGraph）
❌ 多 Agent 协作场景（考虑 CrewAI / AutoGen）
❌ 需要深度定制的训练/微调流程
❌ 对延迟有极致要求的实时场景（考虑直接调用 API）

💡 核心心法：Dify 解决的不是"能不能做"的问题（代码都能做），而是"做多快"和"谁来做"的问题。 对于中小团队，用 Dify 一天能上线的应用，纯代码开发可能需要一周。但当你的业务逻辑超出 Dify 节点的表达能力时，果断回到代码方案（LangChain/LangGraph），不要强行在 UI 上绕弯路。

本文基于 Dify v0.6+ 编写，部分界面和 API 可能随版本更新调整。建议结合 Dify 官方文档获取最新信息。

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

新手友好｜OpenClaw 本地自动化工具快速搭建流程（含安装包）

cover

提升办公效率 OpenClaw 本地智能体完整部署方案（含安装包）

cover

收藏！95%的程序员和打工人必看：如何驾驭AI，逆袭职场瓶颈期？

所有评论(0)

查看更多评论

慕鱼而网

已为社区贡献1条内容