最近我在DIFY上搭工作流,遇到一个特别常见的需求:开始节点里收了一堆信息,有用户输入的问题、过往的聊天记录、上传文档抽出来的文本,还有几个配置参数,但后面的模型提示词只想要一份拼好的完整材料。一开始我直接在LLM节点前串一长串模板替换,结果变量一多,模板就乱成麻。后来我换成一个代码执行节点,在入模型之前把所有内容合并成一段干净的文本,问题瞬间清爽。这篇文章就把我在DIFY中使用代码执行节点合并开始节点各元素内容的完整代码和配置过程梳理出来,给同样被提示词模板折磨的人参考。
1. 合并开始节点内容的两种思路,为什么我选了代码节点
先别急着写代码,想清楚为什么需要这一步,比复制代码更重要。很多人第一次搭工作流时,习惯把所有变量直接甩给大模型,觉得模型能理解多源输入。但实际跑起来会发现,变量一多,提示词模板就成了一个巨大的拼接现场,改一处就崩一片。我是在踩了一轮坑之后,才决定把“合并”这一步独立出来,用代码执行节点统一收口。
1.1 开始节点里的变量到底有多乱
真实项目里,开始节点绝对不只是接收一句用户问题那么简单。以我做的客服工单自动摘要为例:开始节点要接收客户填的问题、订单号、历史会话数组、内部备注,有时候还要挂一个上传的截图或者文档。这些元素类型完全不同——有些是短文本,有些是数组,有些甚至是从知识库检索出来的长段落。如果直接往后端大模型节点扔,模型收到的是七零八落的变量:历史会话要我在提示词里用循环去拼?不行,提示词模板不支持复杂循环。而且每个变量都可能为空,空值处理逻辑在提示词里写起来非常痛苦。
合并的核心价值,是把多个变量收敛成一个“上下文包”。我经常拿打包行李来类比:你不希望把衣服、充电器、护照散落在箱子各个角落,而是把它们归置进几个收纳袋,再放到一个行李箱里。DIFY里的开始节点就是散落的行李,代码执行节点就是那个收纳袋。它接收上游所有变量,统一做清洗、去空、拼接,最后输出给下游。这样一来,LLM节点碰到的始终是一段结构清晰、顺序可控的文本,而不是一堆需要它自己去拼凑的碎片。
1.2 模板拼接 vs 代码执行节点
在决定用代码节点前,我其实先试过直接模板拼接。把问题、历史、文档分别放进提示词的几个位置,再用DIFY的变量引用去填。这个方法对于两三个变量还行,一旦超过四个,维护成本就上来了。具体对比如下:
| 对比项 | 直接在提示词里拼接 | 用代码执行节点合并 |
|---|---|---|
| 可维护性 | 变量一多,模板又长又乱 | 逻辑集中在代码里,改一处就行 |
| 条件判断 | 只能靠节点分支,无法做细粒度控制 | 代码里可以灵活判断空值、过滤无意义内容 |
| 循环处理 | 提示词模板基本不支持 | Python里用 for 循环轻松搞定 |
| 调试能力 | 只能看最终提示词,不好定位问题 | 运行节点能看到输入输出,问题定位快 |
所以我在面对包含数组、JSON、可能为空的长文本这类场景时,会优先选代码执行节点。它用Python3写,可以处理字符串、数组、字典,还能用json模块解析结构化数据。对于“把多个元素合并成一段内容”这种轻量数据整形工作,无论是性能还是灵活性,都比在提示词里硬扛要靠谱得多。
1.3 用代码节点前,先记住一条边界
代码执行节点虽然好用,但不是用来做重活的地方。DIFY的代码运行环境有超时限制,第三方库支持也有限,如果你试图在里面跑爬虫、做复杂的机器学习推理,或者处理几百MB的文件,基本都会遇到超时或内存问题。它更适合做“轻量级的数据整形”:合并文本、转换格式、提取字段、做简单的条件过滤。
我给自己定了一条规律:超过20行并且涉及外部依赖的逻辑,尽量拆到更合适的节点里;只是拼接、清洗、判断,放心交给代码节点。合并开始节点元素这个需求,恰好落在边界内,所以用代码节点是非常合适的选择。
2. 手写合并代码:变量定义、核心函数和输出设计
接下来进入正题。DIFY的代码执行节点不像本地Python脚本那么随意,它有严格的函数约定:必须定义main函数,参数名要与你在节点里配置的输入变量名完全一致,返回值必须是一个dict。我下面给出的代码,可以直接粘到DIFY的代码执行节点里用。
2.1 先想清楚输入输出结构
写代码之前,先明确输入输出。我这里按最常见的场景设计了四个输入变量:
| 变量名 | 类型 | 含义 |
|---|---|---|
| question | string | 用户输入的问题或请求 |
| history | array[string] | 历史消息列表,可能为空 |
| doc_text | string | 文档提取出来的文本,可能为空 |
| meta | string | 其他配置信息,可能是JSON字符串 |
这四个变量都来自开始节点。你在代码执行节点里配置输入时,变量名必须跟这里一致,而且最好是英文小写加下划线,不要用中文名,减少不必要的麻烦。
输出我也固定为三个字段:
| 输出字段 | 类型 | 含义 |
|---|---|---|
| merged_content | string | 合并后的完整文本 |
| message_count | number | 历史消息条数 |
| has_doc | boolean | 是否有文档内容 |
输出字段之所以要拆开,是为了方便下游节点做条件分支。比如has_doc为 true 时,让大模型优先参考文档;message_count可以用来判断是否要做多轮摘要。这样代码节点不只是输出一段文本,还能给后续流程提供路由依据。
2.2 核心代码:把四个变量拼成一段干净文本
直接上代码。这段代码我跑在DIFY 1.x社区版的代码执行节点里,用Python3运行时,没有任何问题:
import json def main(question, history, doc_text, meta): question = question or "" history = history or [] doc_text = doc_text or "" meta = meta or "" if not isinstance(history, list): history = [history] parts = [] question = question.strip() if question: parts.append(f"## 用户提问\n{question}") history_lines = [] for idx, item in enumerate(history, start=1): item = (item or "").strip() if item: history_lines.append(f"{idx}. {item}") if history_lines: parts.append("## 历史消息\n" + "\n".join(history_lines)) if doc_text.strip(): parts.append(f"## 参考文档\n{doc_text.strip()}") if meta.strip(): try: meta_obj = json.loads(meta) if isinstance(meta_obj, dict): meta_lines = [f"{k}: {v}" for k, v in meta_obj.items()] parts.append("## 配置信息\n" + "\n".join(meta_lines)) else: parts.append("## 配置信息\n" + meta.strip()) except Exception: parts.append("## 配置信息\n" + meta.strip()) return { "merged_content": "\n\n".join(parts), "message_count": len(history), "has_doc": bool(doc_text and doc_text.strip()) }这段代码的逻辑其实很简单,但几个细节值得展开说说。
第一,所有输入变量都做了or ""处理。这是我从DIFY的坑里学来的:开始节点里只要用户没填某个字段,DIFY传给代码节点的往往不是空字符串,而是None。如果直接对question调用.strip(),会直接报AttributeError。所以先统一兜底成空字符串或空列表。
第二,history做了isinstance判断。我在调试时遇到过一种情况:上游数组里只有一个元素时,DIFY在测试界面传进来的可能不是数组,而是单个字符串。这时候如果直接for item in history,会把字符串拆成一个个字符,结果完全错乱。所以先判断,不是列表就包成列表。
第三,enumerate从1开始,是为了给历史消息编号。这样下游模型看到的不是一堆挤在一起的文字,而是带序号的列表,能更清楚理解对话顺序。这个细节在单轮和多轮对话都试用过,确实能提升摘要质量。
第四,meta尝试用json.loads解析。如果传入的是标准JSON字符串,就展开成键值对;如果解析失败,就直接按原文拼接。我遇到过接口传过来的 meta 其实是一段普通文本,不解析反而更好看。这里用 try/except 兜底,不会因为解析失败导致整个节点报错。
2.3 为什么我选择Markdown二级标题做分隔
有人可能会问:合并文本为什么不用逗号分隔,或者用一条长拼接字符串?我实际对比过,模型对结构化的内容更敏感。我在每一段内容前加了## 用户提问、## 历史消息这种Markdown小标题,而不是简单用横线或换行。原因很简单:小标题给了模型一个语义锚点,它知道后面跟的是问题还是文档内容,生成时就不会把不同来源的信息混在一起。
段落之间我用空行隔开,而不是硬编码换行。因为LLM上下文里的文本,空行是天然的段落分隔符,比\r\n之类的符号更通用。很多人在本地写脚本时会用+ "\n" +来拼接,但在DIFY代码节点里,我建议用"\n\n".join(parts),最终效果更干净,也方便后续做截断处理。
3. 在DIFY工作流里一步步配好代码执行节点
代码写完,接下来就是把它放到DIFY的工作流里。这一步看着简单,但很多新手会在变量绑定上翻车。我按完整流程拆开讲。
3.1 在开始节点里添加元素变量
第一步,新建一个空白工作流,点击开始节点。在DIFY的节点配置面板里,找到“输入变量”或“表单字段”区域,开始添加变量。
我添加了四个变量:
question,类型选“文本”history,类型选“数组”,数组内的子类型选“文本”doc_text,类型选“段落”meta,类型选“文本”
注意,DIFY的变量有“显示名称”和“变量名/键名”之分。显示名称可以写中文,方便人看;但变量名一定用英文,而且要和后面代码里的参数名完全一致。我甚至不建议用questionText这种驼峰命名,统一小写加下划线更省心:question、history、doc_text、meta。
如果开始节点要接外部API,这些变量名就是接口入参的字段名。换句话说,谁调用这个工作流,就得传这几个参数。所以我通常会在开始节点里给每个变量写默认值,比如问题填“请帮我总结”,历史数组留空,这样测试时不至于报空指针。
3.2 添加代码执行节点并绑定输入
第二步,从左侧节点列表里拖入“代码执行”节点。打开节点配置,选择运行环境为Python3,然后重点来了:在“输入变量”区域,添加question、history、doc_text、meta四个变量,并把它们分别绑定到开始节点的对应变量上。
很多人会漏掉这一步,以为只要代码里写了参数名就行。但DIFY代码节点不是自动识别你心里的变量名的,输入变量的绑定关系是显式配置的。你在代码里写了def main(question, ...),就一定要在节点配置里创建一个叫question的输入变量,并连接上游。
接下来把上面的代码粘贴进编辑器。保存后,在“输出变量”区域添加merged_content、message_count、has_doc,类型分别选String、Number、Boolean。有些DIFY版本可以自动根据返回值生成输出变量,但我习惯手动配一遍,避免类型不匹配。
3.3 调试运行和下游对接
配置完成后,先做一轮调试。点击DIFY运行按钮,填一组测试数据,比如question="订单怎么查进度"、history=["客服:您好", "用户:想知道物流", "客服:稍等"]、doc_text=""、meta='{"order_id":"123456","channel":"app"}',然后运行。
运行完看代码节点的日志,你会看到返回结果像这样:
{ "merged_content": "## 用户提问\n订单怎么查进度\n\n## 历史消息\n1. 客服:您好\n2. 用户:想知道物流\n3. 客服:稍等\n\n## 配置信息\norder_id: 123456\nchannel: app", "message_count": 3, "has_doc": false }看到这个结构,说明代码没白写。接下来在LLM节点里,把提示词中的上下文变量替换成merged_content。DIFY的LLM节点提示词支持插入节点输出,你只需要在变量选择面板里找到代码执行节点的merged_content字段,它就会自动生成一个引用。
下游需要做条件分支时,可以接一个“条件分支”节点,判断has_doc是否为 true。为 true 时,在提示词里加一句“如果参考文档非空,请优先依据文档内容回答”;为 false 时,可以用另一套更精简的提示词。message_count也能用于判断是否属于多轮对话,比如超过2条历史记录就走深层次分析分支。
4. 运行报错和输出不对?这份排查手册直接抄
代码执行节点运行起来之后,最耗时间的其实是排查问题。我把常见问题整理成一张速查表,后面再展开讲几个我踩得最深的坑。
4.1 高频报错速查表
| 现象 | 可能原因 | 解决办法 |
|---|---|---|
| 保存时报变量未定义 | 代码的main参数名与输入变量名不一致 | 把参数名改成输入变量名,并重新保存 |
| 运行返回为空 | 所有输入变量都为空,parts列表为空 | 填一组测试数据;检查输入变量绑定 |
| 报“expecting value” | meta字段是普通文本,json.loads失败 | 代码里加try/except,失败后按原文拼接 |
| message_count返回字符串 | 返回dict里用了str(len(history)) | 确保返回的是int,不要手动转字符串 |
| 输出变量找不到 | 节点返回的dict缺了某个key | 检查return语句的key和输出变量配置 |
| 代码执行超时 | 合并的文本太长或循环太复杂 | 给文本加截断,或把大计算从节点里拆出去 |
这张表里的问题,我基本都在真实项目里遇到过。尤其是变量未定义,DIFY的报错信息有时并不直接说“main里没有这个参数”,而是告诉你代码保存失败,一开始很迷惑,后来发现就是大小写不一致。
4.2 空值处理:所有变量都可能传None
这是文本合并最容易翻车的地方。DIFY里如果你把开始节点的某个变量留空,代码节点里拿到的不是空字符串,而是None。比如question没有填,你直接写question.strip(),运行必挂。
我处理这类问题的方案其实很简单:每个输入变量进来之后,先统一做一次兜底转换。字符串用or "",数组用or [],然后再做进一步判断。如果你不确定某个变量到底是什么类型,可以在返回值里临时加一个debug_info字段,把type(question)和repr(question)都打出来。调试完之后再删掉这段,避免污染下游。
另外,数组里的空值也不能放过。history列表里某个元素可能是None,也可能是空字符串。我在代码里对每个item都做了(item or "").strip(),这样即使历史消息中间漏了一条,也不会导致整个节点崩溃。
4.3 合并结果过长:截断函数和保留策略
客服工单场景下,历史消息往往很多,文档内容也可能很长。合并文本长度一旦超过模型上下文限制,LLM节点就会报错。别等报错才处理,我建议在代码里加一个截断函数。
def truncate(text, max_len=4000): if len(text) <= max_len: return text return text[:max_len] + "\n...[已截断]"然后在返回前调用:
final_text = truncate("\n\n".join(parts))截断策略我选择保前4000字符,因为合并后的文本开头是用户提问,这部分最重要;后面的历史消息超过长度时,只保留前面部分,再加上一个明确的“已截断”标记。这样模型能知道内容不完整,不会误以为信息缺失是它的问题。
我试过直接截断不保留标记,结果模型在摘要时会编造“用户已提出退货”这类不存在的内容。加了截断标记后,模型会谨慎很多,至少在生成时会提示“根据可用的历史记录”。这个细节建议你也加上。
5. 还能怎么扩展:动态拼接、JSON解析和文件内容合并
合并的基本代码搞定后,这个代码执行节点其实可以变成一个多用途的“内容装配工厂”。我根据实际项目经验,补充几个高频扩展场景。
5.1 用开关变量动态决定拼不拼接某段
有些工作流,用户通过勾选“是否需要内部备注”来决定要不要把备注内容传给模型。这时候可以在开始节点加一个布尔变量include_note,然后在代码里用if控制拼接:
if include_note: parts.append(f"## 内部备注\n{internal_note}")这样比在DIFY里搭一堆条件分支再分别接提示词模板要简洁得多。尤其当可选内容有五六段时,用代码节点做“开关”,比用可视化分支节点画出一堆线清晰多了。
我自己的经验是,不要把这种开关逻辑散落在多个节点里。全部收拢到代码执行节点后,下游只需要面对一个merged_content,无论上游怎么组合,LLM节点都感知不到变化。后期要加新的可选项,只改代码节点里的 if 逻辑就行,不需要动整个工作流结构。
5.2 文件内容别硬读,先转成文本再接进来
有人遇到开始节点里有上传的文件,想在代码执行节点里直接读取文件内容。我的建议是:不要这么做。DIFY代码执行节点的文件变量在不同版本里表现不完全一样,直接读文件名和路径很容易踩版本坑,我早期在找文件路径上浪费过很多时间。
更稳妥的链路是:开始节点的文件变量先接一个“文档抽取”节点,或者先经过知识库检索,把文件内容转成纯文本,再把那段文本作为doc_text传给代码执行节点。
我验证过这套链路:开始节点上传PDF -> 文档抽取节点拿到文本 -> 文本接入代码执行节点的doc_text-> 合并后喂给LLM。这样代码节点只处理字符串,逻辑稳定,也好调试。反正最终目的都是拿到文本内容,没必要在代码节点里依赖文件API。
5.3 同时输出JSON结构化数据,方便下游做分支
有些下游节点需要的不是一段Markdown文本,而是结构化数据。比如表单里包含订单号、渠道、客服ID,你可能希望在代码节点解析meta后,把这些字段单独输出去,供后续节点做逻辑判断。
代码可以长这样:
meta_obj = {} if meta.strip(): try: meta_obj = json.loads(meta) except Exception: meta_obj = {} return { "merged_content": final_text, "message_count": message_count, "has_doc": has_doc, "structured_meta": json.dumps(meta_obj, ensure_ascii=False) }如果DIFY版本支持Object类型,你可以直接把meta_obj作为输出;如果不支持,输出一个JSON字符串也没问题。我实际更倾向于输出JSON字符串,因为后面通常只是做几个字段的分支判断,在条件分支节点里用“包含”或“等于”规则就能处理,不一定要真正的对象结构。
这个扩展思路的好处是,代码执行节点从一个“合并器”升级成了“数据前置处理层”。开始节点再乱、再复杂,只要在这里把文本和结构化字段都准备好,后面的工作流就顺很多。
说实话,这段合并代码本身门槛不高,真正的难点在变量类型和DIFY节点的坑。我在实际使用中最大的体会是:代码执行节点里写代码,不要假设上游一定按你设想的数据类型传值,所有输入都要做防御式处理。另一个小技巧是调试时在返回结果里放一个debug_info字段,把所有入参的类型和值打印一遍,调通了再删掉,这比盯着报错干瞪眼快得多。把我上面这段代码跑通,再按自己的字段名改一改,你就能在DIFY里把开始节点那一堆乱七八糟的元素,干干净净地统一成一份稳定上下文。