Dify 与 LangFlow 框架详解

目标：掌握 Dify 低代码平台和 LangFlow 可视化工作流工具

---

目录

• Dify 框架详解
• LangFlow 框架详解
• 对比与选择
• 实战案例
• 面试高频问法
• 记忆要点

---

Dify 框架详解

1. Dify 概述

【定义】

Dify 是一个开源的 LLM 应用开发平台，提供可视化的
应用构建、部署和管理能力。

【核心特性】

1. 可视化编排
   - 拖拽式工作流设计
   - 直观的节点连接
   - 实时预览效果

2. 内置组件
   - LLM 模型接入
   - 向量数据库集成
   - RAG 模板
   - Agent 编排

3. 灔合生态
   - 模型市场
   - 插件系统
   - API 管理
   - 数据集管理

4. 一键部署
   - Docker 容器化
   - K8s 编排
   - 云平台发布

2. Dify 架构

┌─────────────────────────────────────────────┐
│                  用户界面层                    │
│  ┌──────────┬──────────┬──────────┐        │
│  │  工作流编辑 │  应用管理 │ 模型市场│      │
│  │     器       │   │        │        │
│  └──────────┴──────────┴──────────┘        │
└────────────────┬────────────────────────────┘
                  │
┌────────────────┴────────────────────────────┐
│                  API 网关层                      │
│  ┌────────┬──────────┬──────────┐        │
│  │ workflow  │  dataset  │ model   │      │
│  │     │    │     │    │            │
│  └────────┴──────────┴──────────┘        │
└────────────────┬────────────────────────────┘
                  │
┌────────────────┴────────────────────────────┐
│                  应用层                        │
│  ┌────────┬──────────┬──────────┐        │
│  │  Agent    │  RAG       │  Chat    │      │
│  │          │           │         │            │
│  └────────┴──────────┴──────────┘        │
└────────────────┬────────────────────────────┘
                  │
┌────────────────┴────────────────────────────┐
│                  基础设施层                    │
│  ┌────────┬──────────┬──────────┐        │
│  │ Vector DB │  Redis │  Queue   │      │
│  └────────┴──────────┴──────────┘        │
└─────────────────────────────────────────────┘

3. Dify 核心组件

3.1 Workflow（工作流）

# Dify Workflow 定义示例
"""
Dify 工作流 JSON 定义
{
  "name": "RAG Chatbot",
  "version": "0.1.0",
  "mode": "chat",
  "environment": {
    "variable_declarations": [
      {
        "id": "vector_db_name",
        "value": "my_documents",
        "name": "Vector Database Collection Name"
      },
      {
        "id": "embedding_model",
        "value": "text-embedding-3-small",
        "name": "Embedding Model"
      }
    ]
  },
  "conversation_variables": [
    {
      "id": "history",
      "value": "[]",
      "name": "Conversation History"
    }
  ],
  "nodes": [
    {
      "id": "context",
      "position": {
        "x": 30,
        "y": 200
      },
      "data": {
        "type": "knowledge-retrieval",
        "desc": "知识检索节点，使用向量数据库检索相关文档",
        "title": "Knowledge Retieval",
        "query_variable_selector": [
          "sys"
        ],
        "embedding_model_selector": [
          "embedding_model"
        ],
        "top_k_selector": [
          "3"
        ],
        "return_values_selector": [
          "context_str"
        ],
        "vector_database_variable_selector": [
          "vector_db_name"
        ],
        "retrieval_mode": "chunk",
        "retrieval_resource": [
          {
            "desc": "知识文档数据集"
          }
        ]
      }
    },
    {
      "id": "answer",
      "position": {
        "x": 30,
        "y": 400
      },
      "data": {
        "type": "llm",
        "desc": "基于检索到的上下文，使用 LLM 生成回答",
        "title": "Answer Generation",
        "model": {
          "provider": "openai",
          "name": "gpt-4",
          "mode": "chat",
          "completion_params": {
            "temperature": 0.7
          }
        },
        "prompt_template": [
          "{{sys.system_intro}}",
          "根据以下内容回答用户的问题：\n\n",
          "{{context_str}}",
          "\n\n用户的问题：{{sys.user_query}}",
          "\n\n请只输出答案，不要输出其他内容。"
        ],
        "context_variable_selector": [
          "history"
        ],
        "memory": {
          "window": {
            "enabled": true,
            "size": 10
          },
          "type": "simple_chat"
        },
        "output_selector": [
          "sys"
        ]
      }
    }
  ]
}
"""

3.2 RAG 模板

# Dify RAG 模板使用
{
  "name": "Document QA System",
  "version": "0.1.0",
  "mode": "chat",
  "nodes": [
    {
      "id": "context_retrieval",
      "data": {
        "type": "knowledge-retrieval",
        "title": "Context Retrieval",
        "query_variable_selector": ["sys"],
        "top_k_selector": ["5"],
        "retrieval_resource": [{
          "desc": "技术文档库"
        }],
        "embedding_model_selector": ["embedding_model"],
        "vector_database_variable_selector": ["vector_db"],
        "return_values_selector": ["retrieved_context"]
      }
    },
    {
      "id": "prompt_generator",
      "data": {
        "type": "template-transformer",
        "title": "Prompt Generator",
        "template": "根据以下信息回答问题：\n\n{{retrieved_context}}\n\n问题：{{sys.user_query}}"
      }
    },
    {
      "id": "llm",
      "data": {
        "type": "llm",
        "title": "LLM",
        "model": {
          "provider": "openai",
          "name": "gpt-4"
        },
        "prompt_template_selector": ["prompt_generator"],
        "context_variable_selector": ["history"],
        "memory": {
          "window": {"enabled": true, "size": 10}
        }
      }
    }
  ]
}

3.3 Agent 编排

# Dify Agent Workflow
{
  "name": "Multi-Agent Research",
  "version": "0.1.0",
  "mode": "chat",
  "nodes": [
    {
      "id": "intent_classifier",
      "data": {
        "type": "if-else",
        "desc": "根据用户意图路由到不同的 Agent",
        "title": "Intent Classifier",
        "conditions": [
          {
            "id": "research",
            "variable_selector": [
              "sys",
              "user_intent"
            ],
            "output_selector": [
              "sys",
              "selected_agent"
            ],
            "case_id": "research"
          },
          {
            "id": "code",
            "variable_selector": [
              "sys",
              "user_intent"
            ],
            "output_selector": [
              "sys",
              "selected_agent"
            ],
            "case_id": "code"
          },
          {
            "id": "general",
            "variable_selector": [
              "sys",
              "user_intent"
            ],
            "output_selector": [
              "sys",
              "selected_agent"
            ],
            "case_id": "general"
          }
        ]
      }
    },
    {
      "id": "research_agent",
      "data": {
        "type": "tool",
        "desc": "研究 Agent：负责信息搜集",
        "title": "Research Agent",
        "provider_id": "research",
        "provider_type": "api",
        "tool_id": "search",
        "tool_name": "Web Search",
        "tool_configurations": {},
        "tool_parameters": {
          "url": "{{sys.search_query}}"
        },
        "output_selector": ["sys"]
      }
    },
    {
      "id": "code_agent",
      "data": {
        "type": "tool",
        "desc": "代码 Agent：负责代码生成",
        "title": "Code Agent",
        "provider_id": "code",
        "provider_type": "api",
        "tool_id": "code_generate",
        "tool_name": "Code Generator",
        "tool_parameters": {
          "description": "{{sys.code_description}}"
        },
        "output_selector": ["sys"]
      }
    },
    {
      "id": "general_agent",
      "data": {
        "type": "llm",
        "desc": "通用 Agent：处理一般对话",
        "title": "General Agent",
        "model": {
          "provider": "openai",
          "name": "gpt-4"
        }
      }
    }
  ]
}

4. Dify API 使用

import requests
import json
from typing import Dict, Any, Optional

class DifyClient:
    """Dify 平台客户端"""

    def __init__(
        self,
        base_url: str = "http://localhost/v1",
        api_key: Optional[str] = None
    ):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.session = requests.Session()

        if api_key:
            self.session.headers.update({
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            })

    def create_workflow(self, workflow: Dict[str, Any]) -> Dict[str, Any]:
        """创建工作流"""
        url = f"{self.base_url}/workflows"
        response = self.session.post(url, json=workflow)
        response.raise_for_status()
        return response.json()

    def run_workflow(
        self,
        workflow_id: str,
        inputs: Dict[str, Any],
        user: str = "default"
    ) -> Dict[str, Any]:
        """运行工作流"""
        url = f"{self.base_url}/workflows/{workflow_id}/run"
        payload = {
            "inputs": inputs,
            "user": user,
            "response_mode": "blocking"
        }
        response = self.session.post(url, json=payload)
        response.raise_for_status()
        return response.json()

    def stream_workflow(
        self,
        workflow_id: str,
        inputs: Dict[str, Any],
        user: str = "default"
    ):
        """流式运行工作流"""
        url = f"{self.base_url}/workflows/{workflow_id}/run"
        payload = {
            "inputs": inputs,
            "user": user,
            "response_mode": "streaming"
        }
        response = self.session.post(url, json=payload, stream=True)
        response.raise_for_status()

        for line in response.iter_lines():
            if line.startswith("data: "):
                data = line[5:]
                if data == "[DONE]":
                    break
                yield json.loads(data)

    def list_workflows(self, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
        """列出工作流"""
        url = f"{self.base_url}/workflows"
        params = {
            "page": page,
            "page_size": page_size
        }
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()

    def get_workflow(self, workflow_id: str) -> Dict[str, Any]:
        """获取工作流详情"""
        url = f"{self.base_url}/workflows/{workflow_id}"
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()

    def update_workflow(
        self,
        workflow_id: str,
        workflow: Dict[str, Any]
    ) -> Dict[str, Any]:
        """更新工作流"""
        url = f"{self.base_url}/workflows/{workflow_id}"
        response = self.session.put(url, json=workflow)
        response.raise_for_status()
        return response.json()

    def delete_workflow(self, workflow_id: str) -> None:
        """删除工作流"""
        url = f"{self.base_url}/workflows/{workflow_id}"
        response = self.session.delete(url)
        response.raise_for_status()
        return None

    def get_chat_history(
        self,
        workflow_id: str,
        session_id: str
    ) -> list:
        """获取对话历史"""
        url = f"{self.base_url}/chats/{workflow_id}/messages"
        params = {"session_id": session_id}
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json().get("data", [])

    def get_datasets(self) -> list:
        """获取数据集列表"""
        url = f"{self.base_url}/datasets"
        response = self.session.get(url)
        response.raise_for_status()
        return response.json().get("data", [])

    def create_dataset(
        self,
        name: str,
        indexing_technique: str = "high_quality",
        permission: str = "only_me"
    ) -> Dict[str, Any]:
        """创建数据集"""
        url = f"{self.base_url}/datasets"
        payload = {
            "name": name,
            "indexing_technique": indexing_technique,
            "permission": permission
        }
        response = self.session.post(url, json=payload)
        response.raise_for_status()
        return response.json()

    def upload_document(
        self,
        dataset_id: str,
        file_path: str,
        process_rule: Optional[Dict] = None
    ) -> Dict[str, Any]:
        """上传文档到数据集"""
        url = f"{self.base_url}/datasets/{dataset_id}/document/create_by_text"

        with open(file_path, "r") as f:
            content = f.read()

        payload = {
            "name": file_path.split("/")[-1],
            "text": content,
            "indexing_technique": "high_quality",
            "process_rule": process_rule or {}
        }

        response = self.session.post(url, json=payload)
        response.raise_for_status()
        return response.json()

# 使用示例
if __name__ == "__main__":
    client = DifyClient(
        base_url="http://localhost/v1",
        api_key="your-api-key"
    )

    # 创建数据集
    dataset = client.create_dataset(
        name="技术文档库",
        indexing_technique="high_quality"
    )
    print(f"创建数据集: {dataset['id']}")

    # 上传文档
    upload_result = client.upload_document(
        dataset_id=dataset["id"],
        file_path="./python_guide.md"
    )
    print(f"上传文档: {upload_result['id']}")

    # 创建 RAG 工作流
    workflow = {
        "name": "Python 助手",
        "version": "0.1.0",
        "mode": "chat",
        "nodes": [
            {
                "id": "context_retrieval",
                "data": {
                    "type": "knowledge-retrieval",
                    "title": "Context Retrieval",
                    "query_variable_selector": ["sys"],
                    "top_k_selector": ["5"],
                    "retrieval_resource": [{
                        "desc": "Python 文档",
                        "id": dataset["id"]
                    }],
                    "return_values_selector": ["retrieved_context"]
                }
            },
            {
                "id": "llm",
                "data": {
                    "type": "llm",
                    "title": "Answer",
                    "model": {
                        "provider": "openai",
                        "name": "gpt-4"
                    },
                    "prompt_template": [
                        "根据以下信息回答：",
                        "{{retrieved_context}}",
                        "\n问题：{{sys.user_query}}"
                    ]
                }
            }
        ]
    }

    # 保存工作流
    created_workflow = client.create_workflow(workflow)
    print(f"创建工作流: {created_workflow['id']}")

    # 运行工作流
    result = client.run_workflow(
        workflow_id=created_workflow["id"],
        inputs={"sys.user_query": "什么是 Python 的装饰器？"}
    )
    print(f"回答: {result['data']['outputs'][0]['text']}")

---

LangFlow 框架详解

1. LangFlow 概述

【定义】

LangFlow 是一个可视化的 LangChain 工作流设计工具，
通过拖拽节点和连接边来构建复杂的 LLM 应用。

【核心特性】

1. 可视化设计
   - 节点拖拽式设计
   - 实时预览
   - 流程图展示

2. 节点组件
   - LLM 节点
   - 工具节点
   - 条件节点
   - 数据节点

3. 导出功能
   - 导出为 LangChain 代码
   - 导出为 JSON 配置
   - 部署运行

4. 可扩展性
   - 自定义节点
   - 自定义工具
   - 集成扩展

2. LangFlow 架构

┌─────────────────────────────────────────────┐
│                  Canvas 画布                   │
│  ┌──────────┬──────────┬──────────┐        │
│  │  Node    │  Connector │   Flow   │        │
│  │  Palette  │   Panel  │   Controls│        │
│  └──────────┴──────────┴──────────┘        │
└────────────────┬────────────────────────────┘
                  │
┌────────────────┴────────────────────────────┐
│              Node Engine                    │
│  ┌──────────┬──────────┬──────────┐        │
│  │ LLM Nodes│ Tool Nodes│ Logic Nodes│     │
│  └──────────┴──────────┴──────────┘        │
└────────────────┬────────────────────────────┘
                  │
┌────────────────┴────────────────────────────┐
│            Code Generator                     │
│  ┌──────────┬──────────┬──────────┐        │
│  │ Python  │ TypeScript │ JSON  │        │
│  └──────────┴──────────┴──────────┘        │
└─────────────────────────────────────────────┘

3. LangFlow 节点类型

3.1 LLM 节点

{
  "id": "llm_node",
  "type": "llm",
  "data": {
    "model": "gpt-4",
    "temperature": 0.7,
    "max_tokens": 1000,
    "prompt": "你是一个有帮助的助手。"
  },
  "position": {
    "x": 200,
    "y": 100
  }
}

3.2 工具节点

{
  "id": "search_tool",
  "type": "tool",
  "data": {
    "name": "search",
    "description": "搜索网络信息",
    "parameters": {
      "query": {
        "type": "string",
        "required": true
      }
    },
    "input": "{{user_query}}"
  }
}

3.3 条件节点

{
  "id": "router",
  "type": "if-else",
  "data": {
    "condition": "{{intent_type}}",
    "branches": {
      "search": "search_node_id",
      "chat": "llm_node_id"
    }
  }
}

3.4 数据节点

{
  "id": "input_data",
  "type": "data",
  "data": {
    "value": "Hello, how can I help you?"
  }
}

4. LangFlow 流程定义

{
  "name": "Customer Service Bot",
  "description": "客户服务聊天机器人",
  "version": "1.0.0",
  "nodes": [
    {
      "id": "input",
      "type": "data",
      "data": {
        "value": "{{user_input}}"
      },
      "position": {
        "x": 100,
        "y": 100
      }
    },
    {
      "id": "intent_classifier",
      "type": "llm",
      "data": {
        "model": "gpt-4",
        "prompt": "分析用户意图，分类为：search(搜索)、complaint(投诉)、chat(聊天)。\n\n用户输入：{{input}}\n\n只输出类别，不要输出其他内容。",
        "temperature": 0.1
      },
      "position": {
        "x": 300,
        "y": 100
      }
    },
    {
      "id": "router",
      "type": "if-else",
      "data": {
        "condition": "{{intent_classifier_output}}",
        "branches": {
          "search": "search_tool",
          "complaint": "escalate_tool",
          "chat": "response_llm"
        }
      },
      "position": {
        "x": 500,
        "y": 150
      }
    },
    {
      "id": "search_tool",
      "type": "tool",
      "data": {
        "name": "web_search",
        "description": "搜索网络",
        "input": "{{user_input}}"
      },
      "position": {
        "x": 700,
        "y": 250
      }
    },
    {
      "id": "escalate_tool",
      "type": "tool",
      "data": {
        "name": "escalate",
        "description": "升级到人工",
        "input": "{{user_input}}"
      },
      "position": {
        "x": 700,
        "y": 350
      }
    },
    {
      "id": "response_llm",
      "type": "llm",
      "data": {
        "model": "gpt-4",
        "prompt": "回答用户问题。\n\n用户：{{user_input}}\n\n提供一个友好、专业的回答。",
        "temperature": 0.7
      },
      "position": {
        "x": 700,
        "y": 200
      }
    },
    {
      "id": "output",
      "type": "data",
      "data": {
        "value": "{{router_output}}"
      },
      "position": {
        "x": 900,
        "y": 300
      }
    }
  ],
  "edges": [
    {
      "source": "input",
      "target": "intent_classifier"
    },
    {
      "source": "intent_classifier",
      "target": "router",
      "data": {
        "variable": "intent_classifier_output"
      }
    },
    {
      "source": "router",
      "target": "output",
      "data": {
        "variable": "router_output"
      }
    },
    {
      "source": "router",
      "target": "search_tool",
      "condition": "search"
    },
    {
      "source": "router",
      "target": "escalate_tool",
      "condition": "complaint"
    },
    {
      "source": "router",
      "target": "response_llm",
      "condition": "chat"
    }
  ]
}

5. LangFlow 代码生成

# langflow_code_generator.py
"""
将 LangFlow 配置转换为 LangChain 代码
"""
import json
from typing import Dict, List

class LangFlowToLangChain:
    def __init__(self, flow_config: Dict):
        self.config = flow_config

    def generate_code(self) -> str:
        """生成 LangChain 代码"""
        nodes = {node["id"]: node for node in self.config["nodes"]}

        # 生成导入
        code = self._generate_imports()

        # 生成工具
        code += self._generate_tools(nodes)

        # 生成链
        code += self._generate_chains(nodes)

        # 生成入口
        code += self._generate_main(nodes)

        return code

    def _generate_imports(self) -> str:
        """生成导入语句"""
        return '''from langchain.chains import SequentialChain, ParallelChain
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import BaseOutputParser, OutputParser
from langchain.llms import ChatOpenAI
import os

# 配置
os.environ["OPENAI_API_KEY"] = "your-api-key"

'''

    def _generate_tools(self, nodes: Dict) -> str:
        """生成工具定义"""
        tool_nodes = [n for n in nodes.values() if n.get("type") == "tool"]

        code = "# 定义工具\n"
        for node in tool_nodes:
            tool_name = node["data"]["name"]
            code += f'def {tool_name}_func({tool_name}_input: str) -> str:\n'
            code += f'    """执行 {tool_name} 工具"""\n'
            code += f'    # 实现 {tool_name} 逻辑\n'
            code += f'    return f"执行 {tool_name}: {{{tool_name}_input}}"\n\n'

        return code

    def _generate_chains(self, nodes: Dict) -> str:
        """生成链定义"""
        llm_nodes = [n for n in nodes.values() if n.get("type") == "llm"]

        code = "# 定义提示模板\n"
        for node in llm_nodes:
            node_id = node["id"]
            prompt = node["data"].get("prompt", "")
            if prompt:
                code += f'{node_id}_prompt = ChatPromptTemplate.from_messages([\n'
                code += f'    ("system", "{prompt}"),\n'
                code += f'    ("user", "{input}"),\n'
                code += f'    MessagesPlaceholder(variable_name="input")\n'
                code += f'])\n\n'

        return code

    def _generate_main(self, nodes: Dict) -> str:
        """生成主函数"""
        edges = self.config.get("edges", [])
        code = "\n# 主函数\n"
        code += "def main(user_input: str):\n"
        code += '    print(f"用户输入: {user_input}")\n'

        # 简化边生成执行逻辑
        for edge in edges:
            source = nodes[edge["source"]]
            target = nodes[edge["target"]]

            if source.get("type") == "llm" and target.get("type") == "data":
                # LLM 到数据节点
                output_var = edge["data"].get("variable", "output")
                code += f'    # 执行 LLM\n'
                code += f'    chain = LLMChain(llm=llm, prompt={source["id"]}_prompt)\n'
                code += f'    {output_var} = chain.run(user_input)\n'

        code += '    print(f"最终输出: {output_var}")\n'
        code += '\nif __name__ == "__main__":\n'
        code += '    main("Hello, how can I help you?")\n'

        return code

# 使用示例
if __name__ == "__main__":
    # 加载 LangFlow 配置
    with open("workflow.json", "r") as f:
        flow_config = json.load(f)

    # 生成代码
    generator = LangFlowToLangChain(flow_config)
    code = generator.generate_code()

    print("生成的 LangChain 代码：")
    print(code)

    # 保存代码
    with open("generated_chain.py", "w") as f:
        f.write(code)

---

对比与选择

Dify vs LangFlow 对比

维度	Dify	LangFlow
定位	应用平台	工作流工具
部署	一键部署	代码生成后手动部署
存储	内置向量库	需要外部配置
协作	团队协作	适合个人开发
集成	丰富插件	LangChain 原生集成
导出	JSON API	多种语言代码
适用场景	快速构建生产应用	可视化设计复杂流程

选择建议

【选择 Dify 的场景】

1. 需要快速部署
   - 内置部署支持
   - 云平台发布
   - Docker/K8s 支持

2. 团队协作
   - 多人协作编辑
   - 权限管理
   - 版本控制

3. 数据管理
   - 需要内置向量库
   - 文档上传管理
   - 知识库管理

4. 低代码需求
   - 非技术人员也能使用
   - 可视化配置
   - 预制模板

【选择 LangFlow 的场景】

1. 灔度定制
   - 完全控制代码
   - 自定义组件
   - 灵活扩展

2. 现有 LangChain 项目
   - 无缝集成
   - 代码复用
   - 版本管理

3. 原型开发
   - 本地开发调试
   - IDE 集成
   - 版本控制

4. 复杂流程设计
   - 可视化复杂逻辑
   - 条件分支清晰
   - 实时预览效果

---

实战案例

案例：Dify RAG 问答系统

# dify_rag_system.py
"""
构建一个基于 Dify 的 RAG 问答系统
"""

class DifyRAGBuilder:
    """Dify RAG 构建器"""

    def __init__(self, api_key: str):
        self.client = DifyClient(
            base_url="http://localhost/v1",
            api_key=api_key
        )

    def build_rag_workflow(
        self,
        name: str,
        vector_db_name: str,
        top_k: int = 3,
        model_name: str = "gpt-4"
    ) -> str:
        """构建 RAG 工作流"""
        workflow = {
            "name": name,
            "version": "0.1.0",
            "mode": "chat",
            "environment": {
                "variable_declarations": [
                    {
                        "id": "top_k",
                        "value": str(top_k),
                        "name": "Top K"
                    },
                    {
                        "id": "model",
                        "value": model_name,
                        "name": "Model"
                    }
                ]
            },
            "nodes": [
                {
                    "id": "context_retrieval",
                    "data": {
                        "type": "knowledge-retrieval",
                        "desc": "知识检索",
                        "title": "Context Retrieval",
                        "query_variable_selector": ["sys"],
                        "top_k_selector": ["top_k"],
                        "retrieval_resource": [{
                            "desc": "知识库",
                            "id": vector_db_name
                        }],
                        "return_values_selector": ["retrieved_context"]
                    }
                },
                {
                    "id": "prompt_engine",
                    "data": {
                        "type": "template-transformer",
                        "desc": "提示词模板",
                        "title": "Prompt Engine",
                        "template": "你是一个专业的问答助手。"
                    }
                },
                {
                    "id": "llm",
                    "data": {
                        "type": "llm",
                        "desc": "LLM 生成",
                        "title": "Answer Generation",
                        "model": {
                            "provider": "openai",
                            "name": "{{model}}"
                        },
                        "prompt_template": [
                            "{{prompt_engine.output}}",
                            "\n",
                            "基于以下信息回答用户问题：",
                            "{{retrieved_context}}",
                            "\n",
                            "用户问题：{{sys.user_query}}",
                            "\n",
                            "请直接给出答案，不要包含其他内容。"
                        ],
                        "memory": {
                            "window": {
                                "enabled": True,
                                "size": 10
                            },
                            "type": "simple_chat"
                        }
                    }
                },
                {
                    "id": "answer",
                    "data": {
                        "type": "answer",
                        "desc": "答案输出",
                        "title": "Answer",
                        "output_selector": ["llm"]
                    }
                }
            ]
        }

        return workflow

# 使用示例
if __name__ == "__main__":
    builder = DifyRAGBuilder(api_key="your-api-key")

    # 构建工作流
    workflow = builder.build_rag_workflow(
        name="技术问答系统",
        vector_db_name="="tech_docs",
        top_k=5,
        model_name="gpt-4"
    )

    # 保存工作流
    import json
    with open("rag_workflow.json", "w") as f:
        json.dump(workflow, f, indent=2, ensure_ascii=False)

    # 提交到 Dify
    # client = DifyClient(api_key="...")
    # created = client.create_workflow(workflow)
    print("RAG 工作流已创建")

---

面试高频问问法

Q1: Dify 和 LangFlow 的选择？

【标准回答】

选择标准：

1. 团队协作场景 → Dify
   - 团队成员都需要访问
   - 需要统一的平台管理
   - 权限控制重要

2. 快速部署需求 → Dify
   - 内置部署功能
   - 云平台集成
   - 监控告警完善

3. 开发灵活需求 → LangFlow
   - 完全代码控制
   - 本地调试方便
   - IDE 集成

4. 复杂流程设计 → LangFlow
   - 可视化复杂逻辑
   - 实时预览效果
   - 支持条件分支

混合方案：
- 使用 LangFlow 设计流程
- 导出 LangChain 代码
- 集成到现有项目
- 或发布到 Dify 平台

---

记忆要点

【Dify】

开源平台
可视化编排
一键部署
数据集管理
向量库集成

【LangFlow】

工作流工具
代码生成
可视化设计
LangChain 集成

【对比】

Dify：团队协作、快速部署
LangFlow：个人开发、灵活定制

【选择】

团队协作 → Dify
快速部署 → Dify
代码控制 → LangFlow
复杂流程 → LangFlow

---

文档版本: 1.0
最后更新: 2026-01-21