从Dify到Flowise：LLMOps开源平台落地全景路线图​​

# 版本化文档存储模型
from sqlalchemy import Column, JSON, String
class WorkflowVersion(Base):
    __tablename__ = "workflow_versions"
    id = Column(String(36), primary_key=True)
    config = Column(JSON)  # 存储YAML工作流配置
    snapshot = Column(JSON)  # 节点状态快照
    created_at = Column(DateTime, default=datetime.utcnow)
# 版本回滚操作
def rollback_version(workflow_id, version_id):
    version = session.query(WorkflowVersion).filter_by(id=version_id).first()
    load_config(version.config)
    restore_snapshot(version.snapshot)

from langchain_plantuml import activity_diagram_callback
callback = activity_diagram_callback()
agent.run(query, callbacks=[callback])
callback.save_uml_content("agent_flow.puml")

# workflow_config.yaml
nodes:
  - id: data_loader
    type: CsvFile
    params: 
      path: "sales_data.csv"
  - id: analyzer
    type: PythonAnalyzer
    ｓｃｒｉｐｔ: |
      import pandas as pd
      df = pd.read_csv(input)
      return df.describe()
edges:
  - source: data_loader/output
    target: analyzer/input
triggers:
  - cron: "0 9 * * *"  # 每天9点自动执行

from langgraph.graph import Graph
workflow = Graph.from_yaml("workflow_config.yaml")
workflow.build()
results = workflow.invoke({})

// ReactFlow边配置
const edgeTypes = {
  "conditional": ConditionalEdge,  // 条件边
  "dataflow": DataFlowEdge,        // 数据流边
  "fallback": FallbackEdge          // 异常处理边
};
// 条件边实现
if (sourceNode.type === "DECISION") {
  return <ConditionalEdge 
    condition={data.condition} 
    label={data.label} 
  />;
}

import { useDagreLayout } from "@reactflow/dagre-layout";
const { nodes, edges } = useDagreLayout({
  nodes: rawNodes,
  edges: rawEdges,
  options: {
    rankdir: "TB",       // 布局方向(Top-Bottom)
    align: "UL",         // 对齐方式(Upper-Left)
    ranksep: 100,        // 层级间距
    nodesep: 50          // 节点间距
  }
});

def workflow_to_tool(workflow):
    tool_desc = f"""
    {workflow.name}工具功能：
    1. 输入：{workflow.input_params}
    2. 处理步骤：
    """
    for node in workflow.nodes:
        tool_desc += f"    - {node.type}: {node.description}\n"
    return Tool(
        name=workflow.name,
        func=workflow.run,
        description=tool_desc
    )
# 注册到Agent
agent.tools.append(workflow_to_tool(sales_analysis_flow))

<template>
  <div class="container">
    <WorkflowCanvas :nodes="nodes" :edges="edges"/>
    <ToolPalette @drag-start="handleDragStart"/>
    <PropertyPanel :selected-node="selectedNode"/>
  </div>
</template>
<ｓｃｒｉｐｔ>
import { VueFlow, useVueFlow } from '@vue-flow/core'
export default {
  components: { VueFlow },
  setup() {
    const { addNodes, onNodeDrag } = useVueFlow()
    // 节点拖拽事件处理
    onNodeDrag((event, node) => { /* 实时更新位置 */ })
  }
}
</ｓｃｒｉｐｔ>

nodes:
  - id: fetch_transaction
    type: APILoader
    endpoint: "/transactions"
  - id: fraud_detect
    type: LLMChain
    prompt: "分析交易{data}是否存在欺诈模式"
  - id: alert
    type: Webhook
    url: "https://alert.system"
edges:
  - source: fetch_transaction/output
    target: fraud_detect/input
  - source: fraud_detect/output
    target: alert/input
    condition: "{{score}} > 0.9"