在我刚开始接触 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=$1 无损,官方是 ¥7.3=$1,节省超过 85%
- 主流模型价格参考(2026年最新):
- GPT-4.1:$8 / 1M tokens output
- Claude Sonnet 4.5:$15 / 1M tokens output
- Gemini 2.5 Flash:$2.50 / 1M tokens output
- DeepSeek V3.2:$0.42 / 1M tokens output(性价比之王)
- 国内延迟:直连延迟 < 50ms,无需代理
- 充值方式:支持微信、支付宝直接充值
- 新手福利:注册即送免费额度
常见报错排查
错误 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. 变量命名规范
我踩过的坑告诉我,变量命名一定要规范:
- 使用下划线命名法:
user_name、product_list - 前缀标识来源:
start_user_age、llm_generated_text - 避免中文变量名:Dify 对中文支持有时不稳定
总结
通过本文,你应该已经掌握了 Dify 工作流变量传递的核心概念和实战技巧。记住三个关键点:
- 变量需要显式声明:在每个节点的输入配置中添加要使用的变量
- 注意变量类型匹配:数组、字符串、对象之间的转换要小心处理
- 避免循环引用:设计工作流时要规划好变量的流向
如果你在实操过程中遇到任何问题,欢迎在评论区留言,我会尽力解答。
👉 免费注册 HolySheep AI,获取首月赠额度