Dify：开源LLM应用平台

import jsonfrom typing import List, Dict, Any  class SimpleDifyWorkflow:    """简化版Dify工作流引擎演示（模拟核心节点类型）"""        def __init__(self):        self.nodes = {}        self.edges = []        self.variables = {}        def add_node(self, node_id: str, node_type: str, config: Dict):        """添加工作流节点"""        self.nodes[node_id] = {            "type": node_type,            "config": config        }        def add_edge(self, from_node: str, to_node: str):        """添加节点连线（有向边）"""        self.edges.append((from_node, to_node))        def _get_next_nodes(self, current_node: str) -> List[str]:        """获取当前节点的下游节点"""        return [to for frm, to in self.edges if frm == current_node]        def _execute_llm_node(self, config: Dict, inputs: Dict) -> Dict:        """模拟LLM节点执行"""        prompt_template = config.get("prompt", "")        model = config.get("model", "gpt-4")        prompt = prompt_template        for key, value in inputs.items():            prompt = prompt.replace("{{" + key + "}}", str(value))        simulated_response = f"[{model}响应] 基于输入生成的回答...（Prompt: {prompt[:50]}...）"        return {"text": simulated_response, "model": model, "prompt_used": prompt}        def _execute_retrieval_node(self, config: Dict, inputs: Dict) -> Dict:        """模拟知识检索节点执行"""        query = inputs.get("query", "")        knowledge_base = config.get("knowledge_base", "default")        top_k = config.get("top_k", 3)        mock_docs = [            {"title": "文档1", "content": f"关于{query}的相关内容片段A...", "score": 0.92},            {"title": "文档2", "content": f"关于{query}的相关内容片段B...", "score": 0.85},            {"title": "文档3", "content": f"关于{query}的相关内容片段C...", "score": 0.78},        ]        return {            "query": query,            "knowledge_base": knowledge_base,            "documents": mock_docs[:top_k],            "context": "\n\n".join([d["content"] for d in mock_docs[:top_k]])        }        def _execute_conditional_node(self, config: Dict, inputs: Dict) -> Dict:        """模拟条件分支节点执行"""        variable = config.get("variable", "")        operator = config.get("operator", ">")        threshold = config.get("threshold", 0)        value = inputs.get(variable, 0)        try:            if operator == ">":                result = float(value) > float(threshold)            elif operator == "<":                result = float(value) < float(threshold)            elif operator == "==":                result = value == threshold            else:                result = True        except (ValueError, TypeError):            result = False        branch = "true_branch" if result else "false_branch"        return {"condition_met": result, "branch": branch, "value": value}        def _execute_code_node(self, config: Dict, inputs: Dict) -> Dict:        """模拟代码执行节点（安全沙箱）"""        result = {"processed": True, "input_keys": list(inputs.keys())}        if "text" in inputs:            result["text_length"] = len(str(inputs["text"]))            result["text_upper"] = str(inputs["text"]).upper()        return result        def _execute_http_node(self, config: Dict, inputs: Dict) -> Dict:        """模拟HTTP请求节点"""        url = config.get("url", "")        method = config.get("method", "GET")        return {            "status_code": 200,            "url": url,            "method": method,            "response": {"success": True, "data": f"模拟{url}的返回结果"}        }        def execute(self, start_node: str, initial_inputs: Dict) -> Dict:        """执行工作流（简化版拓扑排序执行）"""        self.variables = dict(initial_inputs)        executed = set()        execution_order = []        queue = [start_node]        while queue:            node_id = queue.pop(0)            if node_id in executed:                continue            node = self.nodes.get(node_id)            if not node:                continue            node_type = node["type"]            config = node["config"]            if node_type == "llm":                result = self._execute_llm_node(config, self.variables)            elif node_type == "retrieval":                result = self._execute_retrieval_node(config, self.variables)            elif node_type == "conditional":                result = self._execute_conditional_node(config, self.variables)            elif node_type == "code":                result = self._execute_code_node(config, self.variables)            elif node_type == "http":                result = self._execute_http_node(config, self.variables)            elif node_type == "output":                result = {"output": self.variables}            else:                result = {}            for key, value in result.items():                self.variables[f"{node_id}.{key}"] = value            for key, value in result.items():                if key not in self.variables:                    self.variables[key] = value            executed.add(node_id)            execution_order.append(node_id)            next_nodes = self._get_next_nodes(node_id)            queue.extend(next_nodes)        return {            "execution_order": execution_order,            "final_variables": self.variables,            "nodes_executed": len(executed)        }  def demo_workflow():    """演示一个简单的RAG问答工作流"""    wf = SimpleDifyWorkflow()    wf.add_node("start", "output", {"desc": "开始节点，接收用户输入"})    wf.add_node("retrieval", "retrieval", {        "knowledge_base": "product_faq",        "top_k": 3,        "search_mode": "hybrid"    })    wf.add_node("llm", "llm", {        "model": "gpt-4",        "prompt": "你是客服助手，请根据以下上下文回答用户问题。\n\n上下文：{{context}}\n\n用户问题：{{query}}\n\n回答：",        "temperature": 0.7    })    wf.add_node("conditional", "conditional", {        "variable": "score",        "operator": "<",        "threshold": 0.7,        "desc": "置信度低于0.7转人工"    })    wf.add_node("output", "output", {"desc": "输出最终回答"})    wf.add_edge("start", "retrieval")    wf.add_edge("retrieval", "llm")    wf.add_edge("llm", "conditional")    wf.add_edge("conditional", "output")    print("=== Dify工作流演示 ===")    print("\n工作流节点：")    for nid, node in wf.nodes.items():        print(f"  {nid}: {node['type']}")    print(f"\n连线数: {len(wf.edges)}")    print()    result = wf.execute("start", {"query": "怎么申请退款？", "score": 0.85})    print("执行顺序:", " → ".join(result["execution_order"]))    print(f"执行节点数: {result['nodes_executed']}")    print()    print("=== 关键输出 ===")    print(f"检索结果: {result['final_variables'].get('retrieval.documents', 'N/A')}")    print(f"LLM输出: {result['final_variables'].get('llm.text', 'N/A')[:80]}...")    print(f"条件判断: 置信度0.85 {'满足' if result['final_variables'].get('conditional.condition_met') else '不满足'}阈值要求")  if __name__ == '__main__':    demo_workflow()