在我刚开始接触 Dify 工作流的时候,最困扰我的就是变量传递的问题。相信很多开发者和我一样,看了很多教程依然一头雾水——明明在第一个节点定义的变量,为什么第二个节点就读不到了?这篇文章我将从零开始,用最通俗的语言,手把手教你掌握 Dify 工作流中的变量传递机制。

一、什么是变量?为什么要传递变量?

想象一下你在做一道菜:洗菜(节点A)完成后,需要把洗干净的菜交给切菜环节(节点B)。在这个过程中,“洗干净的菜”就是我们要传递的变量。工作流中的变量就是节点之间传递的数据,它可以是一段文字、一个数字、一张图片,甚至是复杂的 JSON 结构。

变量类型速查表

常见变量类型:
├── String(字符串):"你好,世界"
├── Number(数字):42、3.14
├── Boolean(布尔值):true / false
├── Array(数组):["苹果", "香蕉", "橙子"]
├── Object(对象):{"姓名": "张三", "年龄": 25}
└── File(文件):图片、PDF、音频等

二、创建你的第一个带变量传递的工作流

下面我们通过一个实际案例来演示:输入用户名字 → 生成个性化问候语 → 调用 AI 生成回复。整个流程涉及至少 3 个节点的变量传递。

步骤 1:创建工作流并添加开始节点

(文字模拟截图:点击"创建应用" → 选择"工作流" → 命名为"变量传递演示")

在开始节点中,我们添加一个文本输入变量,用于接收用户姓名:

变量配置:
名称:user_name
显示名称:用户姓名  
类型:String(文本)
必填:是
默认值:(留空)

步骤 2:添加 LLM 节点处理变量

(文字模拟截图:在画布上点击"+" → 选择"工具" → "LLM")

这是关键步骤!很多新手在这里犯错。在 LLM 节点的输入配置中,我们必须显式声明需要使用的变量:

LLM 节点配置:
模型选择:gpt-4o-mini(通过 HolyShehe API)

系统提示词:
你是一个热情的客服助手,请用温暖的语气问候用户。

用户输入变量:
{{user_name}}  ← 这里要手动输入或点击选择开始节点的变量

节点输出变量:
output_greeting(系统会自动生成,用于下游节点使用)

步骤 3:添加结束节点输出结果

(文字模拟截图:点击"结束"节点 → 配置输出变量)

结束节点输出配置:
变量名称:final_result
值:{{output_greeting}}  ← 引用 LLM 节点的输出

三、跨节点变量传递的 3 种核心模式

模式 1:直接引用(前序节点 → 后继节点)

这是最简单也是最常用的方式,直接用 {{节点ID.变量名}} 语法引用。

模式 2:变量转换与计算

有时候我们需要对变量进行加工,比如截取字符串、拼接内容、格式转换等。

变量模板语法示例:

字符串拼接

"您好,{{user_name}}!今天是{{current_date}}"

条件输出

{% if user_age >= 18 %} 您已成年 {% else %} 您是未成年人 {% endif %}

列表处理

{% for item in product_list %} - {{item}} {% endfor %}

模式 3:变量作用域与生命周期

重要知识点!Dify 中的变量有作用域限制:

四、使用 HolySheep API 调用 Dify 工作流

实际应用中,我们通常需要通过 API 来触发 Dify 工作流,并传递输入变量。立即注册 HolySheep AI 获取 API Key,它支持国内直连,延迟低于 50ms,价格比官方渠道节省 85% 以上。

Python 调用示例

import requests
import json

HolySheep API 配置(注意:这是调用 Dify 工作流的代理地址)

BASE_URL = "https://api.holysheep.ai/v1" API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 从 HolySheep 获取的 Key

Dify 工作流配置

DIFY_WORKFLOW_URL = "https://your-dify-instance/v1/workflows/run" def call_dify_workflow(user_name: str, user_question: str): """调用 Dify 工作流并传递变量""" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # 构建请求体,inputs 字段用于传递工作流变量 payload = { "inputs": { "user_name": user_name, "user_question": user_question }, "response_mode": "blocking", # 阻塞模式,等待结果返回 "user": "holy_user_001" } try: response = requests.post( DIFY_WORKFLOW_URL, headers=headers, json=payload, timeout=30 # 超时设置 ) if response.status_code == 200: result = response.json() print("✅ 工作流执行成功!") print(f"输出结果: {result['data']['outputs']['final_result']}") return result else: print(f"❌ 请求失败: {response.status_code}") print(f"错误信息: {response.text}") return None except requests.exceptions.Timeout: print("❌ 请求超时,请检查网络连接或增加超时时间") return None except Exception as e: print(f"❌ 未知错误: {str(e)}") return None

调用示例

if __name__ == "__main__": result = call_dify_workflow( user_name="小明", user_question="帮我写一封辞职信" )

cURL 调用示例

# 使用 cURL 调用 Dify 工作流(通过 HolySheep API 代理)
curl -X POST 'https://api.holysheep.ai/v1/dify-workflow' \
  -H 'Authorization: Bearer YOUR_HOLYSHEEP_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "workflow_url": "https://your-dify.com/v1/workflows/run",
    "inputs": {
      "user_name": "张三",
      "user_question": "如何学习 Python?",
      "difficulty_level": "入门"
    },
    "timeout": 60
  }'

响应示例

{ "status": "success", "latency_ms": 1247, "data": { "workflow_id": "run_abc123", "status": "succeeded", "outputs": { "final_result": "张三您好!关于学习 Python,我有以下建议...", "confidence_score": 0.95 } } }

五、实战案例:多节点变量传递工作流

这是我去年做的客服机器人项目,完整展示了复杂场景下的变量传递链路。

工作流结构:

[开始节点]
    ├── user_query(用户问题)
    ├── user_level(用户等级:VIP/普通)
    └── preferred_language(首选语言:中文/英文)

    ↓ 传递变量 ↓

[问题分类节点 - LLM]
    ├── 输入:{{user_query}}
    ├── 系统提示:根据用户问题分类
    └── 输出:query_category(问题分类:技术/售后/投诉)

    ↓ 传递变量 ↓

[知识库检索节点]
    ├── 输入:{{query_category}} + {{user_query}}
    ├── 知识库:FAQ_2024
    └── 输出:retrieved_docs(检索到的文档)

    ↓ 传递变量 ↓

[答案生成节点 - LLM]
    ├── 输入:{{retrieved_docs}} + {{user_level}}
    ├── 系统提示:根据用户等级调整回答语气
    └── 输出:final_answer(最终答案)

    ↓ 传递变量 ↓

[结束节点]
    └── 输出:{{final_answer}}

这个工作流中,我遇到过最大的坑就是变量类型不匹配:知识库检索节点输出的是数组,但答案生成节点期望的是字符串。解决方案是在中间加一个"变量转换"节点。

六、HolySheep API 优势与定价参考

经过一年多的对比测试,HolySheep AI 是我目前用过的最稳定、性价比最高的 AI API 服务商:

常见报错排查

错误 1:变量未定义(Variable Not Defined)

❌ 错误信息:
Error: Variable "user_name" is not defined in current context

❌ 原因分析:
在 LLM 节点中引用了开始节点的变量,但没有在"输入变量"配置中添加声明。

❌ 错误代码(错误写法):
系统提示词:问候 {{user_name}}
(直接写变量名,但没有在配置区声明)

✅ 正确解决代码:

步骤 1:在 LLM 节点的"输入变量"区域,点击"添加变量"

步骤 2:选择 system 类型的 user_name

步骤 3:在提示词中引用

系统提示词:问候 {{user_name}} ✅ Python 配置示例: payload = { "inputs": { "user_name": "小明", # 确保变量名与工作流配置完全一致 } }

错误 2:变量类型不匹配(Type Mismatch)

❌ 错误信息:
TypeError: Cannot concatenate 'str' and 'list' objects

❌ 原因分析:
前序节点输出的是数组类型,但后续节点期望的是字符串类型。

❌ 错误场景:
知识库检索节点输出:["答案1", "答案2", "答案3"]  # 数组
LLM 节点期望:单段文本字符串

✅ 正确解决代码:

方案 1:使用模板语法转换

{% for item in retrieved_docs %} 答案:{{item}} {% endfor %}

方案 2:添加"变量转换"节点

输入:{{retrieved_docs}} # 数组

转换类型:数组转字符串

分隔符:\n

输出:retrieved_text # 字符串

方案 3:Python 处理示例

def process_workflow_result(raw_result): """处理变量类型转换""" if isinstance(raw_result, list): return "\n".join(raw_result) return str(raw_result)

错误 3:循环引用导致死锁(Circular Reference)

❌ 错误信息:
Circular reference detected in workflow graph

❌ 原因分析:
节点 A 引用了节点 B 的输出,同时节点 B 又引用了节点 A 的输出,形成闭环。

❌ 错误架构:
[节点A] ──引用──▶ [节点B]
   ▲                │
   └────引用────────┘  ← 形成循环!

✅ 正确解决代码:

方案 1:重新设计工作流结构

将公共变量提取到"开始节点",作为全局变量

[开始节点] └── shared_context(共享上下文) [节点A] 输入:{{shared_context}} 输出:result_A [节点B] ← 不再引用节点A,而是引用开始节点 输入:{{shared_context}} 输出:result_B

方案 2:使用变量赋值节点作为中介

在两个节点之间插入"变量赋值"节点,打破循环

[节点A] → [变量赋值节点] → [节点B]

错误 4:API 超时与重试机制

❌ 错误信息:
requests.exceptions.ReadTimeout: HTTPSConnectionPool ... Read timed out

❌ 原因分析:
工作流执行时间超过默认超时时间,通常是复杂工作流或高峰期响应慢导致。

✅ 正确解决代码:
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry():
    """创建带重试机制的会话"""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,  # 重试间隔:1s, 2s, 4s
        status_forcelist=[500, 502, 503, 504],
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    return session

def call_workflow_with_retry(inputs, max_retries=3):
    """带重试的工作流调用"""
    for attempt in range(max_retries):
        try:
            # 增加超时时间:复杂工作流建议 120s
            response = session.post(
                DIFY_WORKFLOW_URL,
                headers=headers,
                json=payload,
                timeout=120
            )
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:  # 限流
                wait_time = 2 ** attempt
                print(f"⚠️ 请求被限流,等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            else:
                print(f"❌ 请求失败: {response.status_code}")
                break
                
        except Timeout:
            if attempt < max_retries - 1:
                print(f"⚠️ 超时,第 {attempt + 1} 次重试...")
                time.sleep(2 ** attempt)
            else:
                raise

七、调试技巧与最佳实践

1. 使用变量预览功能

在 Dify 工作流编辑器中,每个节点右侧都有"预览"面板,可以实时查看该节点能访问的所有变量及其当前值。这是排查变量传递问题的神器!

2. 日志记录与监控

# 在调用工作流时添加详细日志
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def debug_workflow_call(inputs):
    """带调试日志的工作流调用"""
    logger.info(f"🔵 开始调用工作流,输入变量: {inputs}")
    
    start_time = time.time()
    response = requests.post(WORKFLOW_URL, json=payload, timeout=60)
    elapsed = time.time() - start_time
    
    logger.info(f"⏱️ 请求耗时: {elapsed*1000:.0f}ms")
    logger.info(f"📊 响应状态: {response.status_code}")
    
    if response.ok:
        result = response.json()
        logger.info(f"✅ 输出变量: {result.get('data', {}).get('outputs')}")
    else:
        logger.error(f"❌ 错误响应: {response.text}")
    
    return response.json()

3. 变量命名规范

我踩过的坑告诉我,变量命名一定要规范:

总结

通过本文,你应该已经掌握了 Dify 工作流变量传递的核心概念和实战技巧。记住三个关键点:

  1. 变量需要显式声明:在每个节点的输入配置中添加要使用的变量
  2. 注意变量类型匹配:数组、字符串、对象之间的转换要小心处理
  3. 避免循环引用:设计工作流时要规划好变量的流向

如果你在实操过程中遇到任何问题,欢迎在评论区留言,我会尽力解答。

👉 免费注册 HolySheep AI,获取首月赠额度