跳转至

02 - 应用构建

Dify应用构建流程图

应用创建、工作流设计、节点配置

📖 章节概述

本章将深入介绍如何在Dify中构建应用,包括应用创建、工作流设计、节点配置、测试和调试等内容。通过详细的代码示例和实践指导,帮助读者掌握Dify应用构建的核心技能。

🎯 学习目标

完成本章后,你将能够:

  • 深入理解Dify应用类型和适用场景
  • 掌握工作流设计的核心原理
  • 熟练配置各种节点类型
  • 实现复杂的应用逻辑
  • 掌握测试和调试技巧
  • 了解应用构建的最佳实践

1. 应用创建

1.1 应用类型详解

1.1.1 聊天应用(Chat Application)

技术原理: 聊天应用基于对话式AI技术,通过维护会话上下文实现多轮对话。核心组件包括:

  • 对话管理器:维护对话历史和上下文
  • 意图识别器:识别用户意图
  • 响应生成器:基于上下文生成回复
  • 记忆模块:存储长期和短期记忆

适用场景: - 智能客服 - 个人助理 - 教育辅导 - 咨询服务

代码示例 - 聊天应用配置

Python
import requests
import json

class ChatApplicationBuilder:
    """聊天应用构建器"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def create_chat_app(self, config):
        """
        创建聊天应用

        注意:应用需在Dify Web界面创建,Dify不提供通过API创建应用的接口
        以下代码展示应用创建后的配置概念

        Args:
            config: 应用配置字典
        """
        # 应用创建步骤:
        # 1. 登录Dify控制台 -> 创建应用 -> 选择“聊天助手”类型
        # 2. 配置模型、提示词、变量等
        # 3. 在“访问API”页面获取API Key
        # 4. 使用API Key调用应用
        print(f"请在Dify Web界面创建聊天应用:{config['name']}")
        print("创建后可通过Service API调用应用")
        return {"message": "请在Dify Web界面创建应用", "config": config}

    def test_chat_app(self, query, user_id="user_001"):
        """测试聊天应用(应用创建后调用)"""
        url = "https://api.dify.ai/v1/chat-messages"

        payload = {
            "inputs": {},
            "query": query,
            "response_mode": "blocking",
            "user": user_id
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    builder = ChatApplicationBuilder(api_key="your_api_key_here")

    # 应用配置(在Dify Web界面创建时使用)
    config = {
        "name": "智能客服",
        "description": "24小时在线的智能客服系统",
        "provider": "openai",
        "model": "gpt-4o-mini",
        "temperature": 0.7,
        "opening_statement": "您好!我是智能客服,很高兴为您服务。",
        "suggested_questions": [
            "你们的产品有哪些?",
            "如何联系人工客服?",
            "退款政策是什么?"
        ]
    }

    # 注意:应用需在Dify Web界面创建
    app = builder.create_chat_app(config)
    print(f"应用配置: {app}")

    # 创建应用并获取API Key后,可以测试聊天
    response = builder.test_chat_app("你好,我想了解你们的产品")
    print(f"回复: {response}")

1.1.2 文本生成应用(Text Completion Application)

技术原理: 文本生成应用专注于单次文本生成,不维护对话上下文。适用于:

  • 内容创作
  • 文档生成
  • 代码编写
  • 翻译任务

代码示例 - 文本生成应用

Python
class CompletionApplicationBuilder:
    """文本生成应用构建器"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def create_completion_app(self, config):
        """
        创建文本生成应用

        注意:应用需在Dify Web界面创建,Dify不提供通过API创建应用的接口
        """
        # 应用创建步骤:
        # 1. 登录Dify控制台 -> 创建应用 -> 选择“文本生成”类型
        # 2. 配置模型、提示词模板、变量等
        # 3. 在“访问API”页面获取API Key
        print(f"请在Dify Web界面创建文本生成应用:{config['name']}")
        return {"message": "请在Dify Web界面创建应用", "config": config}

    def configure_completion_app(self, config):
        """
        配置文本生成应用

        注意:模型配置需在Dify Web界面完成,不存在model-config API
        以下展示配置的概念结构
        """
        model_config = {
            "provider": config.get("provider", "openai"),
            "model_name": config.get("model", "gpt-4o-mini"),
            "temperature": config.get("temperature", 0.8),
            "top_p": config.get("top_p", 0.9),
            "max_tokens": config.get("max_tokens", 4096)
        }
        print(f"请在Dify Web界面配置模型参数:{model_config}")
        return model_config

    def generate_text(self, inputs, user_id="default_user"):
        """生成文本(用于Completion类型应用)"""
        url = "https://api.dify.ai/v1/completion-messages"

        payload = {
            "inputs": inputs,
            "response_mode": "blocking",
            "user": user_id
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    builder = CompletionApplicationBuilder(api_key="your_api_key_here")

    # 创建文章生成应用
    config = {
        "name": "文章生成器",
        "description": "根据主题生成高质量文章",
        "provider": "openai",
        "model": "gpt-4o-mini",
        "temperature": 0.8,
        "prompt_template": """
请根据以下要求写一篇文章:
主题:{{#topic#}}
风格:{{#style#}}
篇幅:{{#length#}}

文章结构:
1. 引言
2. 正文
3. 结论
"""
    }

    app = builder.create_completion_app(config)
    print(f"应用配置: {app}")

    # 应用创建并获取API Key后,可以生成文本
    result = builder.generate_text(
        inputs={"topic": "人工智能发展趋势", "style": "科普", "length": "长篇"}
    )
    print(f"生成的文章:{result.get('answer', '')}")

1.1.3 工作流应用(Workflow Application)

技术原理: 工作流应用是最复杂的应用类型,通过可视化节点连接实现复杂的业务逻辑。

核心特性: - DAG架构:有向无环图确保流程可预测 - 节点类型丰富:支持多种处理节点 - 数据流转:节点间通过变量传递数据 - 条件分支:支持复杂的条件判断 - 循环处理:支持迭代操作

代码示例 - 工作流应用

Python
class WorkflowApplicationBuilder:
    """工作流应用构建器"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def create_workflow_app(self, config):
        """
        创建工作流应用

        注意:应用需在Dify Web界面创建,工作流也需在Web界面拖拽设计
        以下代码展示配置的概念结构
        """
        # 工作流应用创建步骤:
        # 1. 登录Dify控制台 -> 创建应用 -> 选择“工作流”类型
        # 2. 在可视化编辑器中拖拽节点、配置连接
        # 3. 发布后在“访问API”页面获取API Key
        print(f"请在Dify Web界面创建工作流应用:{config['name']}")
        return {"message": "请在Dify Web界面创建应用", "config": config}

    def get_workflow_config(self, workflow_config):
        """
        获取工作流配置数据结构(仅作为配置参考)

        注意:工作流的创建和编辑需在Dify Web界面完成
        以下结构展示了工作流的数据模型
        """
        return {
            "graph": workflow_config["graph"],
            "environment_variables": workflow_config.get("environment_variables", {})
        }

    def run_workflow(self, app_id, inputs, user_id="default_user"):
        """运行工作流"""
        url = "https://api.dify.ai/v1/workflows/run"

        payload = {
            "inputs": inputs,
            "response_mode": "blocking",
            "user": user_id
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 工作流配置示例
def create_sample_workflow():
    """创建示例工作流配置"""
    workflow_config = {
        "graph": {
            "nodes": [
                {
                    "id": "start",
                    "type": "start",
                    "data": {
                        "title": "开始",
                        "variables": [
                            {
                                "variable": "user_input",
                                "label": "用户输入",
                                "type": "text-input",
                                "max_length": 1000,
                                "required": True
                            }
                        ]
                    }
                },
                {
                    "id": "llm_node",
                    "type": "llm",
                    "data": {
                        "title": "LLM处理",
                        "model": {
                            "provider": "openai",
                            "name": "gpt-4o-mini",
                            "mode": "chat"
                        },
                        "prompt_template": [
                            {
                                "role": "system",
                                "text": "你是一个专业的助手,请帮助用户解决问题。"
                            },
                            {
                                "role": "user",
                                "text": "{{#start.user_input#}}"
                            }
                        ],
                        "context": {
                            "enabled": False
                        }
                    }
                },
                {
                    "id": "end",
                    "type": "end",
                    "data": {
                        "title": "结束",
                        "outputs": [
                            {
                                "value_selector": ["llm_node", "text"],
                                "variable": "result"
                            }
                        ]
                    }
                }
            ],
            "edges": [
                {
                    "id": "start-llm_node",
                    "source": "start",
                    "target": "llm_node"
                },
                {
                    "id": "llm_node-end",
                    "source": "llm_node",
                    "target": "end"
                }
            ]
        },
        "environment_variables": {}
    }

    return workflow_config

# 使用示例
if __name__ == "__main__":
    builder = WorkflowApplicationBuilder(api_key="your_api_key_here")

    # 创建工作流应用
    config = {
        "name": "智能助手工作流",
        "description": "一个简单的工作流应用",
        "workflow": create_sample_workflow()
    }

    # 注意:应用需在Dify Web界面创建
    app = builder.create_workflow_app(config)
    print(f"工作流应用配置: {app}")

    # 查看工作流配置结构
    wf_config = builder.get_workflow_config(create_sample_workflow())
    print(f"工作流配置结构示例: {list(wf_config.keys())}")

    # 应用创建并发布后,可以通过API运行工作流
    result = builder.run_workflow(
        app_id="your_app_id",
        inputs={"user_input": "请介绍一下Python编程语言"}
    )
    print(f"工作流执行结果:{result.get('data', {}).get('outputs', {})}")

2. 工作流设计

2.1 节点类型详解

2.1.1 开始节点(Start Node)

功能:工作流的入口点,定义输入变量

配置参数

Python
def create_start_node(variables):
    """
    创建开始节点

    Args:
        variables: 变量列表
            [{
                "variable": "变量名",
                "label": "显示标签",
                "type": "变量类型",
                "max_length": 最大长度,
                "required": 是否必填,
                "default": 默认值
            }]
    """
    return {
        "id": "start",
        "type": "start",
        "data": {
            "title": "开始",
            "variables": variables
        }
    }

# 示例:创建带多个输入的开始节点
start_node = create_start_node([
    {
        "variable": "user_question",
        "label": "用户问题",
        "type": "paragraph",
        "max_length": 2000,
        "required": True
    },
    {
        "variable": "language",
        "label": "语言",
        "type": "select",
        "required": True,
        "options": ["中文", "英文", "日文"],
        "default": "中文"
    },
    {
        "variable": "tone",
        "label": "语气",
        "type": "select",
        "required": False,
        "options": ["正式", "友好", "幽默"],
        "default": "友好"
    }
])

2.1.2 LLM节点(LLM Node)

功能:调用大语言模型生成文本

配置参数

Python
def create_llm_node(node_id, model_config, prompt_template):
    """
    创建LLM节点

    Args:
        node_id: 节点ID
        model_config: 模型配置
        prompt_template: 提示词模板
    """
    return {
        "id": node_id,
        "type": "llm",
        "data": {
            "title": "LLM处理",
            "model": model_config,
            "prompt_template": prompt_template,
            "context": {
                "enabled": False,
                "variable_selector": []
            },
            "vision": {
                "enabled": False
            },
            "memory": {
                "enabled": False
            }
        }
    }

# 示例:创建LLM节点
llm_node = create_llm_node(
    node_id="llm_1",
    model_config={
        "provider": "openai",
        "name": "gpt-4o-mini",
        "mode": "chat",
        "completion_params": {
            "temperature": 0.7,
            "top_p": 0.9,
            "max_tokens": 2048
        }
    },
    prompt_template=[
        {
            "role": "system",
            "text": "你是一个专业的{{#tone#}}助手。"
        },
        {
            "role": "user",
            "text": "请用{{#language#}}回答以下问题:\n{{#user_question#}}"
        }
    ]
)

2.1.3 知识库节点(Knowledge Base Node)

功能:从知识库中检索相关信息

配置参数

Python
def create_knowledge_node(node_id, dataset_config):
    """
    创建知识库节点

    Args:
        node_id: 节点ID
        dataset_config: 知识库配置
    """
    return {
        "id": node_id,
        "type": "knowledge-retrieval",
        "data": {
            "title": "知识库检索",
            "dataset_ids": dataset_config["dataset_ids"],
            "retrieval_mode": dataset_config.get("retrieval_mode", "single"),
            "multiple_retrieval_config": {
                "top_k": dataset_config.get("top_k", 3),
                "score_threshold": dataset_config.get("score_threshold", 0.5)
            }
        }
    }

# 示例:创建知识库节点
knowledge_node = create_knowledge_node(
    node_id="kb_1",
    dataset_config={
        "dataset_ids": ["dataset_id_1", "dataset_id_2"],
        "retrieval_mode": "multiple",
        "top_k": 5,
        "score_threshold": 0.6
    }
)

2.1.4 条件节点(Condition Node)

功能:根据条件选择不同的执行路径

配置参数

Python
def create_condition_node(node_id, conditions):
    """
    创建条件节点

    Args:
        node_id: 节点ID
        conditions: 条件列表
            [{
                "variable_selector": ["节点ID", "变量名"],
                "comparison_operator": "比较操作符",
                "value": 比较值
            }]
    """
    return {
        "id": node_id,
        "type": "if-else",
        "data": {
            "title": "条件判断",
            "cases": [
                {
                    "case_id": "case_1",
                    "conditions": conditions,
                    "logical_operator": "and"  # and / or
                }
            ]
        }
    }

# 示例:创建条件节点
condition_node = create_condition_node(
    node_id="condition_1",
    conditions=[
        {
            "variable_selector": ["llm_1", "sentiment"],
            "comparison_operator": "contains",
            "value": "positive"
        }
    ]
)

2.1.5 代码节点(Code Node)

功能:执行自定义Python代码

配置参数

Python
def create_code_node(node_id, code, outputs):
    """
    创建代码节点

    Args:
        node_id: 节点ID
        code: Python代码
        outputs: 输出变量列表
    """
    return {
        "id": node_id,
        "type": "code",
        "data": {
            "title": "代码执行",
            "code": code,
            "outputs": outputs
        }
    }

# 示例:创建代码节点
code_node = create_code_node(
    node_id="code_1",
    code="""
import json

def main(user_input: str) -> dict:
    # 处理用户输入
    result = {
        "processed": user_input.strip(),
        "length": len(user_input),
        "words": len(user_input.split())
    }
    return result
""",
    outputs=[
        {
            "variable": "processed",
            "type": "string"
        },
        {
            "variable": "length",
            "type": "number"
        },
        {
            "variable": "words",
            "type": "number"
        }
    ]
)

2.1.6 HTTP节点(HTTP Node)

功能:调用外部API

配置参数

Python
def create_http_node(node_id, http_config):
    """
    创建HTTP节点

    Args:
        node_id: 节点ID
        http_config: HTTP配置
    """
    return {
        "id": node_id,
        "type": "http-request",
        "data": {
            "title": "HTTP请求",
            "method": http_config["method"],  # GET / POST / PUT / DELETE
            "url": http_config["url"],
            "headers": http_config.get("headers", {}),
            "params": http_config.get("params", {}),
            "body": http_config.get("body", {}),
            "timeout": http_config.get("timeout", 60)
        }
    }

# 示例:创建HTTP节点
http_node = create_http_node(
    node_id="http_1",
    http_config={
        "method": "POST",
        "url": "https://api.example.com/v1/predict",
        "headers": {
            "Content-Type": "application/json",
            "Authorization": "Bearer {{#api_key#}}"
        },
        "body": {
            "text": "{{#user_input#}}"
        },
        "timeout": 30
    }
)

2.1.7 结束节点(End Node)

功能:工作流的出口,定义输出变量

配置参数

Python
def create_end_node(outputs):
    """
    创建结束节点

    Args:
        outputs: 输出变量列表
            [{
                "value_selector": ["节点ID", "变量名"],
                "variable": "输出变量名"
            }]
    """
    return {
        "id": "end",
        "type": "end",
        "data": {
            "title": "结束",
            "outputs": outputs
        }
    }

# 示例:创建结束节点
end_node = create_end_node([
    {
        "value_selector": ["llm_1", "text"],
        "variable": "response"
    },
    {
        "value_selector": ["kb_1", "context"],
        "variable": "retrieved_context"
    },
    {
        "value_selector": ["code_1", "processed"],
        "variable": "processed_input"
    }
])

2.2 节点连接方式

2.2.1 顺序连接

Python
def create_sequential_edge(source_id, target_id):
    """创建顺序连接"""
    return {
        "id": f"{source_id}-{target_id}",
        "source": source_id,
        "target": target_id,
        "type": "custom"
    }

# 示例:顺序连接
edges = [
    create_sequential_edge("start", "llm_1"),
    create_sequential_edge("llm_1", "end")
]

2.2.2 条件分支连接

Python
def create_conditional_edge(source_id, target_id, case_id):
    """创建条件分支连接"""
    return {
        "id": f"{source_id}-{target_id}",
        "source": source_id,
        "target": target_id,
        "sourceHandle": case_id,
        "type": "custom"
    }

# 示例:条件分支连接
edges = [
    create_sequential_edge("start", "condition_1"),
    create_conditional_edge("condition_1", "llm_positive", "case_1"),
    create_conditional_edge("condition_1", "llm_negative", "case_2"),
    create_sequential_edge("llm_positive", "end"),
    create_sequential_edge("llm_negative", "end")
]

3. 完整应用示例

示例1:智能问答系统

Python
import requests
import json

class IntelligentQAApp:
    """智能问答系统"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def build_workflow(self):
        """构建完整的工作流"""
        workflow = {
            "graph": {
                "nodes": [
                    # 开始节点
                    {
                        "id": "start",
                        "type": "start",
                        "data": {
                            "title": "开始",
                            "variables": [
                                {
                                    "variable": "question",
                                    "label": "问题",
                                    "type": "paragraph",
                                    "max_length": 1000,
                                    "required": True
                                }
                            ]
                        }
                    },
                    # 知识库检索节点
                    {
                        "id": "kb_retrieval",
                        "type": "knowledge-retrieval",
                        "data": {
                            "title": "知识库检索",
                            "dataset_ids": ["your_dataset_id"],
                            "retrieval_mode": "multiple",
                            "multiple_retrieval_config": {
                                "top_k": 3,
                                "score_threshold": 0.5
                            }
                        }
                    },
                    # LLM节点
                    {
                        "id": "llm_answer",
                        "type": "llm",
                        "data": {
                            "title": "生成答案",
                            "model": {
                                "provider": "openai",
                                "name": "gpt-4o-mini",
                                "mode": "chat"
                            },
                            "prompt_template": [
                                {
                                    "role": "system",
                                    "text": "你是一个专业的问答助手。请基于以下检索到的上下文回答用户的问题。如果上下文中没有相关信息,请诚实地说明。"
                                },
                                {
                                    "role": "user",
                                    "text": """上下文信息:
{{#kb_retrieval.context#}}

用户问题:{{#start.question#}}

请基于上下文信息回答问题。"""
                                }
                            ],
                            "context": {
                                "enabled": True,
                                "variable_selector": ["kb_retrieval", "context"]
                            }
                        }
                    },
                    # 结束节点
                    {
                        "id": "end",
                        "type": "end",
                        "data": {
                            "title": "结束",
                            "outputs": [
                                {
                                    "value_selector": ["llm_answer", "text"],
                                    "variable": "answer"
                                },
                                {
                                    "value_selector": ["kb_retrieval", "context"],
                                    "variable": "context"
                                }
                            ]
                        }
                    }
                ],
                "edges": [
                    {"id": "start-kb", "source": "start", "target": "kb_retrieval"},
                    {"id": "kb-llm", "source": "kb_retrieval", "target": "llm_answer"},
                    {"id": "llm-end", "source": "llm_answer", "target": "end"}
                ]
            }
        }
        return workflow

    def create_app(self):
        """
        创建应用

        注意:Dify不提供通过API创建应用的接口
        应用需在Dify Web界面创建:
        1. 登录Dify控制台 -> 创建应用 -> 选择"工作流"类型
        2. 在可视化编辑器中设计工作流(可参考build_workflow()的结构)
        3. 发布后在"访问API"页面获取API Key
        """
        app_config = {
            "name": "智能问答系统",
            "description": "基于知识库的智能问答",
            "mode": "workflow"
        }
        print(f"请在Dify Web界面创建应用: {app_config['name']}")
        return {"message": "请在Dify Web界面创建应用", "config": app_config}

    def query(self, app_id, question):
        """查询"""
        url = "https://api.dify.ai/v1/workflows/run"

        payload = {
            "inputs": {"question": question},
            "response_mode": "blocking",
            "user": "user_001"
        }

        try:
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

# 使用示例
if __name__ == "__main__":
    # 注意:应用需在Dify Web界面创建
    # 创建后获取API Key
    app = IntelligentQAApp(api_key="your_api_key_here")

    # 查看工作流配置(在Web界面设计时参考)
    workflow = app.build_workflow()
    print(f"工作流包含 {len(workflow['graph']['nodes'])} 个节点")

    # 应用创建并发布后,可以通过API查询
    result = app.query(
        app_id="your_app_id_here",
        question="什么是人工智能?"
    )

    answer = result.get("data", {}).get("outputs", {}).get("answer", "")
    print(f"答案: {answer}")

示例2:多轮对话系统

Python
class MultiTurnChatApp:
    """多轮对话系统"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.conversation_history = {}

    def create_chat_app(self):
        """
        创建聊天应用

        注意:Dify不提供通过API创建应用的接口
        应用需在Dify Web界面创建:
        1. 登录Dify控制台 -> 创建应用 -> 选择"聊天助手"类型
        2. 配置模型、提示词等
        3. 发布后在"访问API"页面获取API Key
        """
        app_config = {
            "name": "多轮对话系统",
            "description": "支持上下文的多轮对话",
            "mode": "chat"
        }
        print(f"请在Dify Web界面创建聊天应用: {app_config['name']}")
        return {"message": "请在Dify Web界面创建应用", "config": app_config}

    def chat(self, app_id, user_id, message):
        """发送消息"""
        url = "https://api.dify.ai/v1/chat-messages"

        # 获取会话ID
        conversation_id = self.conversation_history.get(user_id)

        payload = {
            "inputs": {},
            "query": message,
            "response_mode": "blocking",
            "conversation_id": conversation_id,
            "user": user_id
        }

        try:  # try/except捕获异常
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            result = response.json()

            # 保存会话ID
            self.conversation_history[user_id] = result.get("conversation_id")

            return {
                "answer": result.get("answer"),
                "conversation_id": result.get("conversation_id")
            }
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

    def get_history(self, user_id, limit=10):
        """获取对话历史"""
        url = "https://api.dify.ai/v1/messages"

        conversation_id = self.conversation_history.get(user_id)
        if not conversation_id:
            return {"error": "No conversation found"}

        params = {
            "conversation_id": conversation_id,
            "limit": limit
        }

        try:
            response = requests.get(url, headers=self.headers, params=params)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}

    def clear_history(self, user_id):
        """清除对话历史"""
        if user_id in self.conversation_history:
            del self.conversation_history[user_id]
        return {"message": "History cleared"}

# 使用示例
if __name__ == "__main__":
    # 注意:应用需在Dify Web界面创建,创建后获取API Key
    app = MultiTurnChatApp(api_key="your_api_key_here")

    user_id = "user_001"
    app_id = "your_app_id_here"  # 在Dify Web界面创建应用后获取

    # 多轮对话
    messages = [
        "你好,我想了解Python",
        "Python有什么特点?",
        "Python适合做什么?",
        "推荐一些学习资源"
    ]

    for msg in messages:
        response = app.chat(app_id, user_id, msg)
        print(f"用户: {msg}")
        print(f"助手: {response.get('answer', '')}\n")

4. 练习题

基础练习

  1. 创建问答应用
  2. 创建简单的问答应用
  3. 配置LLM节点
  4. 测试应用

  5. 设计工作流

  6. 设计包含多个节点的工作流
  7. 配置节点连接
  8. 测试工作流

进阶练习

  1. 构建智能客服系统
  2. 集成知识库
  3. 实现多轮对话
  4. 添加情感分析

  5. 开发内容生成工具

  6. 设计复杂工作流
  7. 集成多个数据源
  8. 实现批量处理

5. 最佳实践

✅ 推荐做法

  1. 模块化设计
  2. 将复杂流程分解为多个节点
  3. 每个节点负责单一功能
  4. 便于维护和调试

  5. 充分测试

  6. 测试每个节点
  7. 测试整个工作流
  8. 边界条件测试

  9. 错误处理

  10. 添加异常处理节点
  11. 提供友好的错误提示
  12. 记录详细日志

  13. 性能优化

  14. 合理使用缓存
  15. 优化节点顺序
  16. 减少不必要的计算

❌ 避免做法

  1. 过度复杂
  2. 保持工作流简洁
  3. 避免不必要的节点
  4. 优化流程

  5. 硬编码

  6. 使用变量和参数
  7. 提高灵活性
  8. 便于维护

  9. 忽视安全

  10. 验证用户输入
  11. 保护敏感信息
  12. 实施访问控制

6. 常见问题

Q1: 如何调试工作流?

A: 调试技巧: - 使用调试模式查看每个节点的输出 - 添加日志节点记录中间结果 - 逐步测试每个节点 - 使用测试数据验证逻辑

Q2: 如何优化工作流性能?

A: 优化方法: - 减少不必要的节点 - 优化节点顺序 - 使用缓存减少重复计算 - 选择合适的模型

Q3: 如何处理长文本?

A: 处理方法: - 使用知识库节点 - 分块处理 - 提取关键信息 - 使用摘要生成

7. 总结

本章深入介绍了Dify应用构建的核心内容,包括:

  1. 应用类型:聊天应用、文本生成应用、工作流应用
  2. 节点类型:开始、LLM、知识库、条件、代码、HTTP、结束节点
  3. 工作流设计:节点配置、连接方式、数据流转
  4. 完整示例:智能问答系统、多轮对话系统
  5. 最佳实践:模块化设计、充分测试、错误处理、性能优化

通过本章的学习,你应该能够构建复杂的Dify应用了。

8. 下一步

继续学习03-工作流设计,深入了解工作流的高级设计技巧。


最后更新日期:2026-02-12 适用版本:Dify实战教程 v2026