openJiuwen agent-core DLTransformer converters 模块解析：DL 节点到平台 DSL 的九类转换器实战指南

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

openJiuwen agent-core 的openjiuwen.dev_tools.agent_builder.builders.workflow.dl_transformer.converters模块为工作流可视化编排（DL 格式）到平台 DSL 的转换提供了逐节点转换能力。本文围绕该模块的BaseConverter抽象基类与九个具体转换器，结合仓库源码与测试用例，深入讲解其设计、转换流程、节点类型注册机制以及可复用的调用方式，帮助读者掌握如何将一份 DL 节点描述转换成可执行的平台工作流 DSL。

模块定位：DL 到 DSL 转换的"逐节点翻译层"

在 openJiuwen agent-core 的agent_builder工作流构建链路中，画布上的可视化节点以DL（Diagram Language）形式描述，最终需要被翻译成平台可执行的工作流 DSL。converters子模块正是承担这一"逐节点翻译"的职责：

• 抽象基类BaseConverter封装所有节点共有的转换流程（公共配置 → 子类专属配置 → 边）；
• 九个具体转换器分别处理开始、结束、大模型、意图识别、提问、代码、插件、输出与分支节点；
• DLTransformer依据节点type从注册表中选取对应转换器并统一调度。

模块对外导出：BaseConverter、StartConverter、EndConverter、LLMConverter、IntentDetectionConverter、QuestionerConverter、CodeConverter、PluginConverter、OutputConverter、BranchConverter（见 converters/init.py）。

BaseConverter：公共转换流程的抽象基类

BaseConverter定义在 converters/base.py 中，是所有具体转换器的基类。子类只需实现_convert_specific_config()一个抽象方法，即可复用整套公共转换逻辑。

构造函数参数

class BaseConverter( node_data: Dict[str, Any], nodes_dict: Dict[str, Any], resource: Optional[Dict[str, Any]] = None, position: Position = Position(0, 0), )

在__init__中，基类会依据node_data["type"]从NodeType枚举反查得到 DSL 类型码，并预先构造好一个中间Node对象以及空的edges列表：

node_type_enum = NodeType[node_data["type"]] self.node = Node( id=node_data["id"], type=node_type_enum.dsl_type ) self.edges: List[Edge] = [] self._variable_index: int = 0

convert()：三段式转换主流程

convert()是每个转换器的统一入口，依次执行三步：

def convert(self) -> None: self.convert_common_config() # 1. 公共配置 self._convert_specific_config() # 2. 子类专属配置 self.convert_edges() # 3. 边转换

调用完成之后，DLTransformer会把converter.node追加到Workflow.nodes、把converter.edges合并进Workflow.edges，最终产出 DSL JSON（见后文"DLTransformer 调度"一节）。

convert_common_config()：公共配置转换

将 DL 通用字段落到中间节点上：

self.node.id = self.node_data["id"] self.node.meta = { "position": {"x": self.position.x, "y": self.position.y} } self.node.data.title = self.node_data["description"]

即：节点id原样保留，画布坐标写入meta.position，DL 中的description作为节点标题data.title。

convert_edges()：默认单出口边转换

基类默认处理最简单的"单出口"情形——DL 节点通过next字段指向后继节点：

if "next" in self.node_data and self.node_data["next"]: self.edges.append(Edge( source_node_id=self.node_data["id"], target_node_id=self.node_data["next"] ))

多分支节点（Branch、IntentDetection）以及需要携带端口信息（Code节点）的场景会重写该方法，下文逐一说明。

公共工具方法：输入变量与输出字段

_convert_input_variables(inputs)将 DL 的输入列表转换为InputVariable字典，核心逻辑是识别${node.variable}引用语法：

• 值中包含${时，调用ConverterUtils.convert_ref_variable()解析为type: "ref"的引用变量；
• 否则作为常量处理，type: "constant"，并带上schema = {"type": item.get("type") or "string"}；
• 每个变量附带extra={"index": self._variable_index}序号，序号随转换过程自增。

_convert_outputs_field(outputs)则把输出列表转换为OutputsField（type="object"、properties={}、required=[]）。这里有一个值得注意的细节：输出变量名采用xxx_of_yyy的路径式命名，name.split("_of_")[::-1]反转切分后即可表达嵌套结构，通过OutputPropertySpec逐层构建出嵌套的properties树（对应 models.py 中OutputsField.add_property的递归实现）。

NodeType：DL 类型名与 DSL 类型码的桥接

转换器选择依赖 models.py 中定义的NodeType枚举。每个枚举成员同时持有两个字符串：

• dl_type：DL 节点type字段的取值（如Start、LLM）；
• dsl_type：平台 DSL 使用的类型码。

各节点类型的完整映射如下（来自NodeType枚举定义）：

BaseConverter.__init__中的NodeType[node_data["type"]].dsl_type正是利用该枚举把 DL 类型名翻译成 DSL 类型码写入node.type。

九个具体转换器逐个剖析

九个具体转换器均继承BaseConverter，由DLTransformer依据节点type从注册表中选取，构造函数签名与BaseConverter一致（PluginConverter通常需要传入resource）。转换完成后，节点与边会被追加到Workflow图结构中，详细调度见 dl_transformer.md。

StartConverter：定义工作流入口与初始输出

开始节点的核心产物是初始输出变量：

outputs = self._convert_outputs_field(self.node_data["parameters"]["outputs"]) self.node.data.outputs = outputs if outputs.properties: self.node.data.outputs.required = list(outputs.properties.keys())

即：把 DL 中声明的outputs（如用户输入query）转换为 DSL 输出字段，并全部标记为required。集成测试 test_converters_integration.py 中构造了type: "Start"、outputs: [{"name": "query", "description": "用户输入"}]的节点数据，转换后断言node.id == "node_start"、node.type == NodeType.Start.dsl_type。

EndConverter：汇聚工作流最终输入

结束节点只处理输入侧：

self.node.data.inputs = InputsField( input_parameters=self._convert_input_variables( self.node_data["parameters"]["inputs"] ) )

将 DL 中的inputs（通常是引用上游节点变量的${...}表达式）转换为InputsField下的输入参数。

LLMConverter：输出格式的智能推断

大模型节点的转换逻辑集中在两处：输出格式推断与LLM 参数组装。

outputs_count = len(outputs_list) output_format = self.node_data["parameters"]["configs"].get("output_format", "text") if outputs_count > 1 and output_format in ("text", "markdown"): output_format = "json" if outputs_count == 1 and output_format not in ("text", "markdown", "json"): output_format = "text" self.node.data.output_format = output_format

可见默认输出格式为text，并存在两条自动修正规则：

• 声明了多个输出变量时，若格式仍为text/markdown，强制改为json（多输出只能靠结构化 JSON 承载）；
• 只有一个输出变量但格式既不是text/markdown也不是json时，回落为text。

LLM 参数通过ConverterUtils.convert_llm_param(system_prompt, user_prompt)生成systemPrompt/prompt模板对，并额外注入response_format = {"type": output_format}：

llm_param = ConverterUtils.convert_llm_param( self.node_data["parameters"]["configs"]["system_prompt"], self.node_data["parameters"]["configs"]["user_prompt"] ) llm_param["response_format"] = {"type": output_format} self.node.data.inputs = InputsField( input_parameters=self._convert_input_variables(self.node_data["parameters"]["inputs"]), llm_param=llm_param, ) self.node.data.outputs = self._convert_outputs_field(self.node_data["parameters"]["outputs"])

IntentDetectionConverter：意图识别与多端口边

意图识别节点把 DL 条件中的"xxx contain 意图名"表达式解析为意图列表，并为每个分支建立端口：

parts = cond["expression"].split(" contain ") if len(parts) > 1: intent_name = parts[1] intents.append({ "name": intent_name, "id": cond.get("intent_id", f"intent_{uuid.uuid4().hex[:8]}") })

• 输入侧：DL 的configs.prompt被组装成llm_param（systemPrompt为空模板、prompt为实际提示词、model使用ConverterUtils.LLM_MODEL_CONFIG），输入变量统一改名为query；
• 输出侧：固定输出一个classification_id（type: "integer"）并加入required；
• 边侧：重写convert_edges()，default分支连接到端口"0"，其余分支以对应intent_id作为source_port_id建立边。

QuestionerConverter：提问节点

提问节点用于在流程中向用户收集输入。其转换逻辑包括：

• 用configs.prompt生成llm_param（第二个参数用户提示为空串）；
• 输出字段中若缺少user_response与output会自动补全（均为string类型）；
• 若 DL 未声明任何输入变量，自动注入默认引用node_start.query：

if not input_parameters: input_parameters["input"] = InputVariable( type=SourceType.ref.value, content="node_start.query", extra={"index": 0} )

• 固定设置history_enable=False、max_response=3；
• required列表排除名为output的字段。

CodeConverter：代码执行节点与异常配置

代码节点将 DL 的configs.code写入inputs（语言固定python），输出字段全部标记为必填，并注入异常处理配置：

CODE_EXCEPTION_CONFIG = { "retryTimes": 3, "timeoutSeconds": 30, "processType": "break", "executeStep": { "defaultStep": "0", "errorStep": "1" } }

此外它重写了convert_edges()，为默认边显式携带source_port_id="0"，对应异常配置中defaultStep: "0"的正常执行出口。

PluginConverter：插件/工具节点与默认输出补全

插件节点依赖resource定位插件元数据：

tool_id = dl_configs.get("tool_id", "") plugins = self.resource.get("plugins", []) if self.resource else [] plugin_info = next((p for p in plugins if p.get("tool_id") == tool_id), {}) plugin_param = self._convert_plugin_info(plugin_info)

_convert_plugin_info把插件信息映射为 DSL 的toolID、toolName、pluginID、pluginName、pluginVersion（版本缺失时默认"draft"）。

输出侧采用"用户输出 + 默认输出"的合并策略：先转换 DL 声明的输出，再补上三个固定字段并全部置为必填：

PLUGIN_DEFAULT_OUTPUTS = { "error_code": {"type": "integer", "extra": {"index": 1}}, "error_message": {"type": "string", "extra": {"index": 2}}, "data": {"type": "object", "extra": {"index": 3}, "properties": {}} } result.required = ["error_code", "error_message", "data"]

另外，PluginConverter还提供了两个静态判定方法：is_local_code_plugin（依据language/code字段判断是否为本地代码插件）与is_cloud_plugin（依据path/method字段判断是否为云端插件）。

OutputConverter：输出节点模板

输出节点把 DL 的configs.template转为content模板字段：

self.node.data.inputs = InputsField( input_parameters=self._convert_input_variables(self.node_data["parameters"]["inputs"]), content={ "type": "template", "content": self.node_data["parameters"]["configs"]["template"] } )

BranchConverter：条件分支与操作符映射

分支节点是整个模块中映射关系最丰富的转换器，包含两套映射表：

比较操作符映射（BRANCH_OPERATOR_MAP，DL 表达式 → DSL 操作符）：

条件逻辑映射（BRANCH_LOGIC_MAP）：or→1，and→2。

分支条件转换规则（_convert_branches）：

• 条件带expressions列表时，逐条转换表达式并按operator映射逻辑值，绑定branchId；
• 单条expression非default时，转换为单条件分支；
• expression == "default"时输出空条件列表，作为兜底分支。

表达式解析（_convert_expression）：扫描字符串找出命中的操作符，按操作符切分左右两侧；两侧值通过_build_side处理——含${视为引用变量（type: "ref"），否则作为常量并附带schema.extra.weak: True弱类型标记。

边转换同样被重写：每条条件生成一条边，source_port_id取条件的branch标识：

for cond in self.node_data["parameters"]["conditions"]: self.edges.append(Edge( source_node_id=self.node_data["id"], target_node_id=cond["next"], source_port_id=cond["branch"] ))

DLTransformer 如何调度转换器

转换器的实际调度发生在 dl_transformer.py 的DLTransformer.transform_to_dsl()中，整个调用链如下：

1. 若传入resource，先调用collect_plugin()依据tool_id_map把原始插件字典规整为plugins列表（含plugin_id、plugin_name、tool_id、tool_name、ori_inputs、ori_outputs，以及可选的language/code/path/method字段）；
2. 从 DL 文本中抽取 JSON 数组并校验类型（非数组抛ValueError）；
3. 构建nodes_dict（id → 节点映射）；
4. 遍历每个节点，从_dsl_converter_registry按node["type"]取转换器类：

_dsl_converter_registry: Dict[str, type] = { NodeType.Start.dl_type: StartConverter, NodeType.End.dl_type: EndConverter, NodeType.LLM.dl_type: LLMConverter, NodeType.IntentDetection.dl_type: IntentDetectionConverter, NodeType.Questioner.dl_type: QuestionerConverter, NodeType.Code.dl_type: CodeConverter, NodeType.Plugin.dl_type: PluginConverter, NodeType.Output.dl_type: OutputConverter, NodeType.Branch.dl_type: BranchConverter, }

5. 实例化转换器——Plugin类型额外传入resource，其余节点使用默认position；
6. 调用node_converter.convert()，把node与edges分别汇入Workflow图；
7. 每处理完一个节点，画布坐标x += 20、y += 20向右下递增；
8. 最终经ConverterUtils.convert_to_dict()（递归剔除所有None值）序列化为 DSL JSON 字符串。

未注册的节点类型会被记录warning日志后跳过，不影响整体转换。transform_to_mermaid()则走SimpleIrToMermaid分支，将 DL 转为 Mermaid 流程图用于可视化预览。get_dsl_converter_registry()类方法返回注册表副本，方便外部检查或扩展。

支撑模型与工具类速览

转换过程依赖 models.py 中的一组数据类与 converter_utils.py 中的工具函数：

• Workflow：图容器，含nodes: List[Node]与edges: List[Edge]；
• Node：中间节点，含id、type、meta、data（DataConfig，内含title、inputs、outputs、branches、exception_config）；
• Edge：边，含source_node_id、target_node_id、可选的source_port_id；
• SourceType：变量来源枚举ref/constant；
• ConverterUtils.extract_variable()：用正则\$\{\s*(\w+)\.(\w+)\s*\}从${node.variable}中提取节点名与变量名；
• ConverterUtils.convert_ref_variable()：把引用表达式解析为{"type": "ref", "content": [节点名, ...变量路径]}；
• ConverterUtils.convert_llm_param()：生成systemPrompt/prompt模板与mode模型配置（默认模型为ConverterUtils.LLM_MODEL_CONFIG，即{"id": "52", "name": "siliconf-qwen3-8b", "type": "Qwen/Qwen3-8B"}）。

通过 DLTransformer 使用的完整示例

日常使用中无需直接实例化具体转换器，统一通过DLTransformer入口即可。原文档给出的最小验证示例：

>>> from openjiuwen.dev_tools.agent_builder.builders.workflow.dl_transformer import DLTransformer >>> t = DLTransformer() >>> callable(t.transform_to_dsl) True

一份完整的最小可执行流程如下——先准备 DL JSON 数组文本，再调用transform_to_dsl得到平台 DSL：

from openjiuwen.dev_tools.agent_builder.builders.workflow.dl_transformer import DLTransformer dl_content = ''' [ {"id": "node_start", "type": "Start", "description": "开始节点", "parameters": {"outputs": [{"name": "query", "description": "用户输入"}]}, "next": "node_llm"}, {"id": "node_llm", "type": "LLM", "description": "大模型节点", "parameters": { "inputs": [{"name": "query", "value": "${node_start.query}"}], "outputs": [{"name": "output", "description": "输出"}], "configs": {"system_prompt": "You are helpful", "user_prompt": "{{query}}"} }, "next": "node_end"}, {"id": "node_end", "type": "End", "description": "结束节点", "parameters": {"inputs": [{"name": "result", "value": "${node_llm.output}"}]}} ] ''' transformer = DLTransformer() dsl_json = transformer.transform_to_dsl(dl_content) print(dsl_json) # 可视化预览（无需资源） mermaid_code = transformer.transform_to_mermaid(dl_content)

其中transform_to_dsl的第二个可选参数resource用于插件类节点，结构为{"plugins": [...], "plugin_dict": {...}, "tool_id_map": {"tool_1": "plugin_1"}}，传入后DLTransformer会自动规整插件信息供PluginConverter使用（对应collect_plugin的测试见 test_dl_transformer.py）。如需检查某个节点类型是否支持转换，可调用：

registry = DLTransformer.get_dsl_converter_registry() assert "Start" in registry and "LLM" in registry

测试验证

仓库提供了两层测试覆盖本模块行为：

• 单元测试 test_dl_transformer.py：验证collect_plugin的插件收集（含缺失工具、空列表、带language/code的本地代码插件等分支）、注册表包含全部 9 种节点类型、注册表值为类对象，以及DLTransformer初始化；
• 集成测试 test_converters_integration.py：逐一构造Start、End、LLM、Branch、Plugin、Code、Questioner、IntentDetection、Output节点的 DL 数据并实例化对应转换器，断言node_data透传以及convert()后节点id、type的转换正确性。

这些测试同时为读者提供了各种节点 DL 数据结构的真实样例，可作为编写自定义 DL 的参考模板。

扩展你自己的节点转换器

若需支持新的节点类型，只需遵循两条约定：一是继承BaseConverter并实现_convert_specific_config()（必要时重写convert_edges()支持多端口）；二是在NodeType枚举中登记dl_type/dsl_type对，并将转换器类注册进DLTransformer._dsl_converter_registry。基类提供的_convert_input_variables、_convert_outputs_field等公共工具足以覆盖大多数节点的输入输出转换需求，这正是该模块"开放扩展、统一流程"的设计意图。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习