AI Agent编排框架Dify完全教程

一、Dify框架概述与架构设计

Dify是一个开源的LLM应用开发平台，提供从Agent原型到生产级部署的完整工具链。其核心价值在于：通过可视化编排降低AI应用开发门槛，同时保留足够的灵活性供开发者深度定制。

整体架构

┌─────────────────────────────────────────────┐
│                  前端控制台                    │
│   (应用管理 / Workflow编辑器 / 知识库管理)      │
├─────────────────────────────────────────────┤
│                 API Gateway                  │
├──────────┬──────────┬──────────┬────────────┤
│ Workflow │  Agent   │ RAG引擎  │  工具管理   │
│  引擎    │  运行时   │          │            │
├──────────┴──────────┴──────────┴────────────┤
│              模型接入层 (LLM Providers)        │
├─────────────────────────────────────────────┤
│     存储层 (PostgreSQL / Redis / Weaviate)    │
└─────────────────────────────────────────────┘

后端：Python (Flask) + Celery异步任务 前端：Next.js + React 向量数据库：支持Weaviate/Qdrant/Milvus/PGVector 消息队列：Redis 主数据库：PostgreSQL

---

二、核心概念

2.1 四大应用类型

类型	特点	适用场景
Chatflow	多轮对话、支持记忆	客服、咨询、聊天机器人
Workflow	单次执行、DAG编排	数据处理、内容生成、自动化流程
Agent	自主推理、工具调用	复杂问答、研究分析
Completion	单轮文本生成	翻译、摘要、文案

2.2 节点（Node）

Workflow由节点组成，每个节点是一个处理单元：

• LLM节点：调用大语言模型
• 知识检索节点：从知识库中检索相关内容
• 问题分类节点：对输入进行分类路由
• 条件分支节点（IF/ELSE）：基于条件走不同路径
• 代码执行节点：运行自定义Python/JS代码
• HTTP请求节点：调用外部API
• 变量赋值节点：设置和修改变量
• 模板转换节点：Jinja2模板渲染
• 迭代节点：循环处理列表数据
• 参数提取节点：从文本中提取结构化参数

2.3 变量系统

Dify的变量是节点间数据传递的纽带：

• 环境变量：全局配置，如API Key
• 会话变量：Chatflow中跨轮次持久化的变量
• 节点输出变量：每个节点的输出可被下游节点引用
• 系统变量：sys.query（用户输入）、sys.user_id等

---

三、环境搭建与本地部署

3.1 Docker Compose一键部署

# 克隆仓库
git clone https://github.com/langgenius/dify.git
cd dify/docker

# 复制环境配置
cp .env.example .env

# 编辑.env文件，配置关键参数
# SECRET_KEY=your-random-secret-key
# CONSOLE_WEB_URL=http://localhost:3000
# API_KEY=sk-xxx (如需接入OpenAI)

# 启动所有服务
docker compose up -d

# 查看运行状态
docker compose ps

启动后访问：

• 控制台：http://localhost:3000
• API服务：http://localhost:5001

3.2 环境变量配置

.env文件中的关键配置项：

# 模型提供商配置
OPENAI_API_KEY=sk-your-key
ANTHROPIC_API_KEY=sk-ant-your-key

# 向量数据库（默认使用Weaviate）
VECTOR_STORE=weaviate
WEAVIATE_ENDPOINT=http://weaviate:8080

# 存储配置
DB_USERNAME=postgres
DB_PASSWORD=your-db-password
REDIS_HOST=redis

# 文件存储
STORAGE_TYPE=local
STORAGE_LOCAL_PATH=/app/storage

3.3 源码开发模式

# 后端
cd api
pip install -r requirements.txt
flask db upgrade
flask run --host 0.0.0.0 --port 5001

# 前端
cd web
npm install
npm run dev

---

四、Workflow工作流编排实战

4.1 构建文章摘要Workflow

以"输入URL → 抓取内容 → 生成摘要 → 翻译成英文"为例：

Step 1：创建Workflow应用

在控制台创建新应用，选择"Workflow"类型。

Step 2：添加开始节点

开始节点定义输入变量：

变量名: url
类型: String
必填: true

Step 3：HTTP请求节点 - 抓取网页

{
  "url": "{{#start.url#}}",
  "method": "GET",
  "headers": {
    "User-Agent": "Mozilla/5.0"
  },
  "timeout": 10
}

Step 4：代码节点 - 提取正文

import re
from html.parser import HTMLParser

class TextExtractor(HTMLParser):
    def __init__(self):
        super().__init__()
        self.text = []
        self.skip_tags = {'script', 'style', 'nav', 'header', 'footer'}

    def handle_starttag(self, tag, attrs):
        if tag in self.skip_tags:
            self._skip = True

    def handle_endtag(self, tag):
        if tag in self.skip_tags:
            self._skip = False

    def handle_data(self, data):
        if not getattr(self, '_skip', False):
            stripped = data.strip()
            if stripped:
                self.text.append(stripped)

def main(html: str) -> dict:
    extractor = TextExtractor()
    extractor.feed(html)
    content = '\n'.join(extractor.text)
    # 限制长度避免超出token
    if len(content) > 8000:
        content = content[:8000]
    return {"content": content}

Step 5：LLM节点 - 生成摘要

配置提示词：

请将以下内容总结为3-5个要点，每个要点一句话：

{{#code_node.content#}}

Step 6：LLM节点 - 翻译

Translate the following summary to English:

{{#summary_node.text#}}

Step 7：结束节点

输出变量：

summary: {{#summary_node.text#}}
summary_en: {{#translate_node.text#}}

4.2 通过API调用Workflow

import requests

API_BASE = "http://localhost:5001/v1"
API_KEY = "app-your-api-key"

def run_workflow(inputs: dict) -> dict:
    response = requests.post(
        f"{API_BASE}/workflows/run",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "inputs": inputs,
            "response_mode": "blocking",
            "user": "user-001"
        }
    )
    return response.json()

# 执行
result = run_workflow({"url": "https://example.com/article"})
print(result['data']['outputs'])

---

五、Agent智能体配置与调试

5.1 创建Agent应用

Agent的核心能力是推理-行动循环（ReAct）：LLM根据问题决定是否调用工具，观察工具返回结果后继续推理，直到得出最终答案。

5.2 配置工具

Dify内置了多种工具，也支持自定义：

内置工具示例：

• Google搜索 / Bing搜索
• 网页抓取
• 维基百科
• 数学计算
• 代码执行器

自定义API工具（通过OpenAPI Schema定义）：

openapi: "3.0.0"
info:
  title: 天气查询API
  version: "1.0"
servers:
  - url: https://api.weather.com
paths:
  /v1/current:
    get:
      operationId: getCurrentWeather
      summary: 获取当前天气
      parameters:
        - name: city
          in: query
          required: true
          schema:
            type: string
          description: 城市名称
      responses:
        "200":
          description: 天气信息
          content:
            application/json:
              schema:
                type: object
                properties:
                  temperature:
                    type: number
                  description:
                    type: string

5.3 Agent提示词设计

你是一个专业的研究助手。你可以：
1. 搜索网络获取最新信息
2. 查询天气数据
3. 执行数学计算

## 工作原则
- 回答要有数据支撑，注明信息来源
- 不确定的信息要明确标注
- 复杂问题分步骤推理
- 用中文回答

## 输出格式
回答时先列出关键事实，再给出结论。

5.4 调试技巧

在Dify控制台的"日志"功能中，可以查看Agent的完整推理链：

[思考] 用户问的是今天的天气，我需要先确定用户所在城市
[行动] 调用天气API，参数: city=北京
[观察] 温度28°C，晴天
[思考] 已获得天气信息，可以回答
[回答] 北京今天天气晴朗，气温28°C...

常见调试问题：

• 工具调用失败：检查API Schema定义是否正确，参数类型是否匹配
• 推理循环：限制最大迭代次数（默认5次），避免无限循环
• 回答质量差：优化系统提示词，增加few-shot示例

---

六、知识库管理与RAG集成

6.1 创建知识库

支持多种数据源导入：

• 文本文件（TXT/MD）
• PDF文档
• 网页抓取
• Notion同步
• 自定义API同步

6.2 分块策略配置

分块模式: 自动 / 自定义
最大分块长度: 500 tokens
分块重叠: 50 tokens
分隔符: \n\n（段落级）

分块策略选择指南：

• 通用文档：自动模式，500 token分块
• FAQ类：按问答对分块，每对为一个分块
• 技术文档：按标题层级分块，保留上下文
• 对话记录：按轮次分块

6.3 检索模式

Dify提供三种检索模式：

向量检索：语义相似度匹配

适合: 模糊查询、同义词场景
示例: "退货政策" 能匹配到 "退款流程"

全文检索：关键词匹配

适合: 精确术语、编号查询
示例: "订单号12345" 精确匹配

混合检索（推荐）：向量 + 全文 + Rerank

适合: 大多数场景
配置:
  - 向量权重: 0.7
  - 全文权重: 0.3
  - Rerank模型: bge-reranker-v2-m3
  - TopK: 5
  - 相似度阈值: 0.5

6.4 在Workflow中使用知识检索

节点类型: 知识检索
输入变量: {{#start.query#}}
知识库: 选择已创建的知识库
检索模式: 混合检索
TopK: 5
输出变量: result (包含检索到的文档片段列表)

在后续LLM节点中引用检索结果：

根据以下参考资料回答用户问题。如果资料中没有相关信息，请如实说明。

参考资料：
{{#knowledge_retrieval.result#}}

用户问题：{{#start.query#}}

---

七、自定义工具与API集成

7.1 通过OpenAPI Schema添加工具

在控制台"工具 → 自定义工具 → 创建自定义工具"中，粘贴OpenAPI 3.0 Schema。

示例：集成企业内部CRM系统

openapi: "3.0.0"
info:
  title: CRM客户管理
  version: "1.0"
servers:
  - url: https://crm.internal.com/api
security:
  - BearerAuth: []
paths:
  /customers/search:
    get:
      operationId: searchCustomer
      summary: 搜索客户信息
      parameters:
        - name: keyword
          in: query
          required: true
          schema:
            type: string
          description: 搜索关键词（姓名/手机号/邮箱）
        - name: limit
          in: query
          schema:
            type: integer
            default: 10
      responses:
        "200":
          description: 客户列表
          content:
            application/json:
              schema:
                type: object
                properties:
                  customers:
                    type: array
                    items:
                      type: object
                      properties:
                        name:
                          type: string
                        phone:
                          type: string
                        email:
                          type: string
                        orders_count:
                          type: integer
components:
  securitySchemes:
    BearerAuth:
      type: http
      scheme: bearer

7.2 代码节点实现复杂逻辑

代码节点支持Python和JavaScript，可实现工具无法覆盖的逻辑：

# 数据格式转换示例
def main(api_response: str, format_type: str) -> dict:
    import json
    data = json.loads(api_response)

    if format_type == "table":
        headers = list(data[0].keys()) if data else []
        rows = [list(item.values()) for item in data]
        table = "| " + " | ".join(headers) + " |\n"
        table += "| " + " | ".join(["---"] * len(headers)) + " |\n"
        for row in rows:
            table += "| " + " | ".join(str(v) for v in row) + " |\n"
        return {"output": table}

    elif format_type == "summary":
        count = len(data)
        return {"output": f"共找到 {count} 条记录"}

    return {"output": json.dumps(data, ensure_ascii=False, indent=2)}

7.3 Webhook触发

Dify支持通过Webhook触发Workflow执行：

import requests

def trigger_workflow(webhook_url: str, data: dict):
    """通过Webhook触发Dify Workflow"""
    response = requests.post(
        webhook_url,
        json={
            "inputs": data,
            "response_mode": "blocking"
        },
        headers={"Content-Type": "application/json"}
    )
    return response.json()

# 场景：新订单创建时触发自动处理
trigger_workflow(
    "https://dify.internal.com/v1/webhooks/xxx",
    {
        "order_id": "ORD-20240101-001",
        "customer": "张三",
        "items": "iPhone 15 x1",
        "total": 9999
    }
)

---

八、变量与条件分支控制

8.1 变量类型详解

变量类型	作用域	持久性	使用场景
环境变量	全局	永久	API Key、配置参数
会话变量	单会话	会话期间	用户偏好、对话状态
节点输出	当前执行	执行期间	节点间数据传递
系统变量	全局	自动	用户ID、当前时间等

8.2 条件分支实战

场景：根据用户问题类型走不同处理流程

节点: 问题分类器
输入: {{#sys.query#}}
分类:
  - "订单相关" → 订单处理流程
  - "产品咨询" → 产品知识检索
  - "投诉建议" → 投诉处理流程
  - "其他" → 通用回复

IF/ELSE条件分支：

# 条件表达式示例
# 条件1: 优先客户
{{#customer.level#}} == "VIP"

# 条件2: 紧急问题
{{#sentiment.score#}} < 0.3 and {{#query.urgency#}} == "high"

# 条件3: 包含订单号
{{#sys.query#}} contains "订单"

8.3 迭代节点处理批量数据

节点: 迭代
输入: {{#api_response.items#}}  (列表变量)

循环体:
  - LLM节点: 对每个item生成描述
  - 代码节点: 整理结果

输出: 所有迭代结果的汇总列表

代码节点示例（循环体内）：

def main(item: dict, index: int) -> dict:
    """处理单个迭代项"""
    processed = {
        "序号": index + 1,
        "名称": item.get("name", ""),
        "摘要": item.get("description", "")[:100],
        "价格": f"¥{item.get('price', 0)}"
    }
    return {"processed_item": processed}

---

九、Chatflow vs Workflow选择

9.1 核心区别

特性	Chatflow	Workflow
交互模式	多轮对话	单次执行
记忆能力	支持会话记忆	无记忆
用户交互	等待用户输入	自动执行到底
适用场景	客服、咨询、教育	数据处理、报告生成
输出方式	流式输出	阻塞/流式

9.2 Chatflow特有功能

会话变量：在Chatflow中，可以定义跨轮次持久化的变量：

变量名: user_name
类型: String
默认值: ""

变量名: conversation_stage
类型: String
默认值: "greeting"

对话开场白：

你好！我是你的AI助手 👋
我可以帮你：
• 查询订单状态
• 解答产品问题
• 处理售后服务
请问有什么可以帮你的？

下一步问题建议： 基于对话上下文自动生成建议问题，提升用户体验。

9.3 选择决策树

需要多轮对话？
├── 是 → 用户需要多次交互完成任务？
│   ├── 是 → Chatflow
│   └── 否 → Workflow + 对话开场
└── 否 → 需要自主决策？
    ├── 是 → Agent
    └── 否 → Workflow / Completion

---

十、企业级应用案例

10.1 智能客服系统（Chatflow）

架构设计：

用户消息 → 意图分类 → ┌─ 订单查询 → 订单API → 回复
                      ├─ 产品咨询 → 知识库检索 → 回复
                      ├─ 投诉建议 → 情感分析 → 转人工/自动回复
                      └─ 闲聊    → LLM直接回复

关键配置：

• 知识库：产品手册、FAQ、退换货政策
• 工具：订单查询API、CRM系统、工单系统
• 会话变量：客户等级、问题类型、工单号
• 条件分支：VIP客户优先处理、高敏感度自动转人工

10.2 数据分析助手（Agent）

系统提示词:
你是一个数据分析助手。用户会提供数据或描述分析需求。
你可以：
1. 使用代码执行器运行Python进行数据分析
2. 使用搜索工具查询行业基准数据
3. 生成可视化图表

分析步骤：
1. 理解分析需求
2. 检查数据质量
3. 选择合适的分析方法
4. 执行分析并生成结论
5. 给出可操作的建议

工具配置：

• 代码执行器（Python，预装pandas/numpy/matplotlib）
• SQL查询工具（连接企业数据仓库）
• 图表生成工具

10.3 内容生成工厂（Workflow）

流程：

输入(主题) → 竞品分析(搜索) → 大纲生成(LLM) → 
→ 逐章节生成(LLM×迭代) → 润色(LLM) → SEO优化 → 输出

迭代节点配置：

# 章节列表
chapters = [
    "引言",
    "市场分析",
    "产品对比",
    "使用场景",
    "价格分析",
    "结论与建议"
]

每个章节通过LLM节点生成，最后由代码节点合并为完整文章。

---

十一、性能调优与生产部署

11.1 检索优化

分块策略调优：

• 文档分块太小 → 丢失上下文，回答不完整
• 文档分块太大 → 检索不精确，噪音多
• 建议：500-800 tokens，50-100 tokens重叠

Rerank模型：

配置路径: 知识库设置 → 检索设置 → Rerank模型
推荐: bge-reranker-v2-m3 (多语言，效果好)
TopN: 3-5 (太多会增加延迟和成本)

混合检索权重调优：

# 场景化权重建议
精确查询为主（订单号、产品编号）: 向量0.3 + 全文0.7
语义查询为主（概念、描述）: 向量0.8 + 全文0.2
通用场景: 向量0.6 + 全文0.4

11.2 LLM调优

Prompt优化：

• 明确角色和任务边界
• 提供few-shot示例
• 指定输出格式（JSON/Markdown）
• 设置合理的temperature（客服场景建议0.3-0.5）

Token优化：

# 通过代码节点控制上下文长度
def main(full_context: str, max_tokens: int = 3000) -> dict:
    # 按token估算截断（中文约1.5字/token）
    max_chars = max_tokens * 1.5
    if len(full_context) > max_chars:
        # 保留最新的上下文
        full_context = full_context[-int(max_chars):]
    return {"trimmed_context": full_context}

11.3 生产部署配置

Docker Compose生产配置：

# docker-compose.prod.yml
services:
  api:
    image: langgenius/dify-api:latest
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 4G
          cpus: '2'
    environment:
      - MODE=production
      - LOG_LEVEL=WARNING
    depends_on:
      - db
      - redis

  worker:
    image: langgenius/dify-api:latest
    command: celery -A app.celery worker
    deploy:
      replicas: 2

  web:
    image: langgenius/dify-web:latest
    deploy:
      replicas: 2

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl

Nginx配置：

upstream dify_api {
    server api:5001;
    keepalive 32;
}

upstream dify_web {
    server web:3000;
    keepalive 32;
}

server {
    listen 443 ssl;
    server_name dify.yourcompany.com;

    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;

    client_max_body_size 50M;

    location /api/ {
        proxy_pass http://dify_api/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 300s;
    }

    location / {
        proxy_pass http://dify_web;
        proxy_set_header Host $host;
    }
}

11.4 监控与告警

# 自定义监控指标采集
import time
from functools import wraps

class DifyMetrics:
    def __init__(self):
        self.workflow_runs = 0
        self.llm_calls = 0
        self.errors = 0
        self.latencies = []

    def track_workflow(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                DifyMetrics.workflow_runs += 1
                return result
            except Exception as e:
                DifyMetrics.errors += 1
                raise
            finally:
                DifyMetrics.latencies.append(time.time() - start)
        return wrapper

    def get_stats(self) -> dict:
        return {
            "total_runs": self.workflow_runs,
            "error_rate": self.errors / max(self.workflow_runs, 1),
            "p50_latency": sorted(self.latencies)[len(self.latencies)//2]
                          if self.latencies else 0,
            "p99_latency": sorted(self.latencies)[int(len(self.latencies)*0.99)]
                          if self.latencies else 0,
        }

---

十二、与LangChain/LlamaIndex对比分析

12.1 定位差异

维度	Dify	LangChain	LlamaIndex
定位	低代码AI应用平台	LLM编程框架	数据索引框架
用户	业务人员+开发者	开发者	数据工程师
开发方式	可视化拖拽	代码编写	代码编写
学习曲线	低	中高	中
灵活性	中（受限于节点类型）	高（完全可编程）	高
部署	Docker一键	自行部署	自行部署

12.2 技术栈对比

Dify优势：

• 可视化编排，快速原型
• 内置知识库管理、多模型切换
• 开箱即用的API服务
• 团队协作支持
• 内置日志和监控

LangChain优势：

• 生态最丰富，集成最多
• 完全可编程，无限制
• LangSmith提供专业调试工具
• 社区活跃，文档完善
• 支持最前沿的Agent架构（ReAct/Plan-and-Execute/LATS）

LlamaIndex优势：

• 数据索引和检索最专业
• 多种索引结构（树/关键词/向量/知识图谱）
• 复杂文档解析能力强
• 与数据源集成最深

12.3 如何选择

需要快速上线AI应用？
├── 是 → 团队有前端/全栈能力？
│   ├── 否 → Dify（低代码）
│   └── 是 → 需要深度定制？
│       ├── 否 → Dify
│       └── 是 → LangChain + 自建前端
└── 否 → 核心需求是数据检索？
    ├── 是 → LlamaIndex
    └── 否 → LangChain

12.4 混合使用方案

在实际项目中，三者并不互斥：

# Dify作为编排层 + LangChain处理复杂逻辑
# 通过Dify的API节点调用LangChain服务

# LangChain服务端
from fastapi import FastAPI
from langchain.agents import initialize_agent, Tool
from langchain.chat_models import ChatOpenAI

app = FastAPI()

@app.post("/analyze")
async def analyze(data: dict):
    llm = ChatOpenAI(model="gpt-4o")
    tools = [...]  # 自定义工具
    agent = initialize_agent(tools, llm, agent="openai-functions")
    result = await agent.arun(data["query"])
    return {"result": result}

在Dify Workflow中，通过HTTP请求节点调用上述LangChain服务，实现"可视化编排 + 代码级灵活性"的最佳组合。

---

总结

Dify作为AI应用开发平台，最大的价值是降低了AI应用的构建门槛。通过可视化Workflow编排，非技术背景的业务人员也能快速搭建AI应用；通过开放的API和代码节点，开发者可以实现深度定制。

关键实践建议：

1. 从Chatflow/Workflow开始，验证业务价值后再考虑自研
2. 善用知识库+RAG，让AI基于企业私有数据回答
3. Prompt工程是核心，投入足够时间优化提示词
4. 监控和迭代，持续分析bad case，优化流程
5. 生产环境重视安全，API Key管理、输入过滤、输出审核不可少