File size: 16,362 Bytes
96b6673
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
import hashlib
import copy
import json
from citekit.prompt.prompt import Prompt, ALCEDocPrompt,DocPrompt,NewALCEVanillaPrompt

try:
    example_results = json.load(open('res.json'))
except:
    example_results = []


def weight_check(module, target):
    if module.model_type == 'verifier':
        test_module = copy.deepcopy(module)
        test_module.last_message = True
        true_target = test_module.send()
        if str(target) == str(true_target) or ('output' in str(target).lower() and 'output' in str(true_target).lower()):
            return 1
        test_module.last_message = False
        true_target = test_module.send()
        if str(target) == str(true_target) or ('output' in str(target).lower() and 'output' in str(true_target).lower()):
            return -1
        
    return 0

def get_params(module):
    import copy
    componets = copy.deepcopy(module.prompt_maker.components if hasattr(module, 'prompt_maker') and hasattr(module.prompt_maker, 'components') else None)
    if componets:
        if isinstance(componets, tuple):
            componets = componets[0]
        componets.update(module.self_prompt)
    destination = module.get_destinations()
    if len(destination) == 1:
        destination = str(destination[0])
    elif not destination:
        destination = 'N/A'
    else:
        destination = 'Not Available for multiple destinations'

    if hasattr(module, 'iterative'):
        mode = 'iterative' if module.iterative else ('parallel' if module.parallel else None) 
    else:
        mode = None
    
    if hasattr(module, 'if_add_output_to_head') and hasattr(module, 'head_key'):
        global_prompt = 'N/A' if module.if_add_output_to_head == False else module.head_key
    else:
        global_prompt = 'N/A'


    params = {
        'type': module.model_type,
        'Model': getattr(module, 'model_name', None),
        'Mode': mode,
        'Max Turn': getattr(module, 'max_turn', None),
        'Destination': destination,
        'Prompt': module.prompt_maker.make_prompt(componets) if hasattr(module,'prompt_maker') and module.prompt_maker else None,
        'Global Prompt': global_prompt,
    }
    print(params)
    non_empty_params = {}
    for k, v in params.items():
        if v:
            non_empty_params[k] = v
    return non_empty_params

class PipelineGraph:

    def __init__(self, pipeline = None):
        self.pipeline = pipeline
        self.nodes = {'input': {}, 'output': {}}
        self.edges = {}  # 邻接表:{from_node: {to_node: weight}}
        self.node_count = 0
        if pipeline:
            self.load_pipeline(pipeline)
            print('FINDING ATTR:', [str(module) for module in pipeline.module])

    def get_auto_node_name(self):
        """生成自动节点名。"""
        self.node_count += 1
        return f"Node{self.node_count}"
    
    def update(self):
        self.__init__(pipeline=self.pipeline)

    def load_pipeline(self, pipeline):
        """从现有Pipeline对象加载图。"""
        initial_module = pipeline.get_initial_module()
        processed_nodes = []
        if not initial_module:
            initial_module = pipeline.llm
        self.add_node(name = str(initial_module), 
                      params = get_params(module=initial_module))
        self.add_edge(from_node='input', to_node=str(initial_module), weight=0)
        processed_nodes.append(initial_module)
        subnodes = initial_module.get_destinations()
        if not subnodes:
            self.add_edge(from_node=str(initial_module), to_node='output', weight=0)
            return
        # print([str(subnode) for subnode in subnodes])
        for subnode in subnodes:
            weight = weight_check(initial_module, subnode)
            self.load_node(subnode, processed_nodes=processed_nodes)
            self.add_edge(from_node=str(initial_module), to_node=str(subnode), weight=0)
        if hasattr(initial_module,'output_cond') and initial_module.output_cond:
            weight = weight_check(initial_module, 'output')
            self.add_edge(from_node=str(initial_module), to_node='output', weight=weight)
            

    def load_node(self, module, processed_nodes = None):
        if module in processed_nodes:
            return
        self.add_node(name = str(module),
                      params = get_params(module=module))
        processed_nodes.append(module)
        subnodes = module.get_destinations()
        if not subnodes:
            self.add_edge(from_node=str(module), to_node='output', weight=0)
            return
        for subnode in subnodes:
            self.load_node(subnode, processed_nodes)
            weight = weight_check(module, subnode)
            self.add_edge(from_node=str(module), to_node=str(subnode), weight=weight)
        if hasattr(module,'output_cond') and module.output_cond:
            weight = weight_check(module, 'output')
            self.add_edge(from_node=str(module), to_node='output', weight=0)
            

    def export(self):
        """将图结构导出为字典格式,包含节点和边的信息。"""
        nodes_export = {name: params.copy() for name, params in self.nodes.items()}
        edges_export = []
        for from_node, connections in self.edges.items():
            for to_node, weight in connections.items():
                edges_export.append({
                    'from': from_node,
                    'to': to_node,
                    'weight': weight
                })
        return {
            'nodes': nodes_export,
            'edges': edges_export
        }

    @classmethod
    def import_from_dict(cls, data):
        """从字典数据创建新的PipelineGraph实例。"""
        graph = cls()  # 创建新实例,此时包含默认的input/output节点
        # 清空默认节点(假设导入数据包含完整节点信息)
        graph.nodes = {}
        # 导入节点
        for name, params in data['nodes'].items():
            graph.add_node(name, params)
        # 导入边
        for edge in data['edges']:
            graph.add_edge(edge['from'], edge['to'], edge['weight'])
        return graph

    def add_node(self, name, params=None):
        """添加或更新节点,参数可选。"""
        if params is None:
            params = {}
        self.nodes[name] = params.copy()

    def remove_node(self, name):
        """删除节点,自动移除相关边。不能删除input/output节点。"""
        if name in ['input', 'output']:
            raise ValueError("Cannot remove 'input' or 'output' nodes.")
        if name not in self.nodes:
            return
        del self.nodes[name]
        # 移除该节点作为起点的边
        if name in self.edges:
            del self.edges[name]
        # 移除其他节点到该节点的边
        for from_node in list(self.edges.keys()):
            if name in self.edges[from_node]:
                del self.edges[from_node][name]
                # 若该节点无其他出边,删除空字典以保持整洁
                if not self.edges[from_node]:
                    del self.edges[from_node]

    def add_edge(self, from_node, to_node, weight):
        """添加或更新边,权重必须是0、-1或+1。"""
        if weight not in {0, -1, 1}:
            raise ValueError("Weight must be 0, -1, or 1.")
        if from_node not in self.nodes:
            raise KeyError(f"Node '{from_node}' does not exist.")
        if to_node not in self.nodes:
            raise KeyError(f"Node '{to_node}' does not exist.")
        if from_node not in self.edges:
            self.edges[from_node] = {}
        self.edges[from_node][to_node] = weight

    def remove_edge(self, from_node, to_node):
        """删除指定边。"""
        if from_node in self.edges and to_node in self.edges[from_node]:
            del self.edges[from_node][to_node]
            # 清理空字典
            if not self.edges[from_node]:
                del self.edges[from_node]

    def get_nodes(self):
        """返回所有节点名。"""
        return list(self.nodes.keys())

    def get_edges(self):
        """返回所有边的列表,形式为(from, to, weight)。"""
        edges = []
        for from_node in self.edges:
            for to_node, weight in self.edges[from_node].items():
                edges.append((from_node, to_node, weight))
        return edges

    def update_node_params(self, name, params):
        """更新节点的参数表。"""
        if name not in self.nodes:
            raise KeyError(f"Node '{name}' does not exist.")
        self.nodes[name].update(params.copy())

    def visualize(self, use_graphviz=True):
        """
        可视化图结构:
        - 默认尝试用graphviz生成矢量图(需安装graphviz)
        - 若未安装graphviz,则用文本模式输出
        """
        # 尝试用graphviz生成专业图表
        if use_graphviz:
            try:
                import graphviz
            except ImportError:
                print("Graphviz未安装,切换到文本模式。安装命令:pip install graphviz")
                return self._visualize_text()
            
            # 创建有向图,设置排版方向为从左到右
            dot = graphviz.Digraph(comment='Pipeline Graph', graph_attr={'rankdir': 'LR'})
            
            # 添加所有节点(包含参数信息)
            for node_name in self.nodes:
                params = self.nodes[node_name]
                # 将参数字典转换为易读字符串(例如 "size=2, color=red")
                params_str = ", ".join([f"{k}={v}" for k, v in params.items()])
                
                # 特殊样式标记input/output节点
                if node_name in ('input', 'output'):
                    dot.node(
                        node_name, 
                        label=f"{node_name}\n({params_str})" if params_str else node_name,
                        shape='doublecircle',  # 双圆表示端点
                        style='filled',        # 填充颜色
                        fillcolor='#e6f3ff'   # 浅蓝色背景
                    )
                else:
                    dot.node(
                        node_name, 
                        label=f"{node_name}\n({params_str})" if params_str else node_name,
                        shape='box',           # 方框表示普通节点
                        style='rounded,filled',# 圆角填充
                        fillcolor='#f0f0f0'   # 浅灰色背景
                    )
            
            # 添加所有带权重的边
            for from_node, connections in self.edges.items():
                for to_node, weight in connections.items():
                    dot.edge(
                        from_node, 
                        to_node, 
                        label=str(weight),    # 显示权重
                        color='#666666',      # 边颜色
                        fontcolor='#ff5555'   # 权重文字颜色
                    )
            
            # 生成并自动打开可视化文件(Mac默认用Preview打开PDF)
            dot.render('pipeline_graph', view=True, cleanup=True)
            return
        
        # 文本模式回退方案
        self._visualize_text()
    
    def _visualize_text(self):
        """纯文本模式的可视化(备用方案)"""
        print("="*40 + "\nPipeline Graph 文本视图\n" + "="*40)
        
        # 列出所有节点及其参数
        print("\n[节点列表]")
        for node, params in self.nodes.items():
            param_desc = " ".join([f"{k}={v}" for k, v in params.items()])
            print(f"· {node:10} {param_desc}")
        
        # 列出所有边及其权重
        print("\n[边列表]")
        if not self.edges:
            print("(暂无边连接)")
        else:
            for from_node, connections in self.edges.items():
                for to_node, weight in connections.items():
                    arrow = {
                        1: "──(+)──>",   # +1权重用(+)表示
                        -1: "──(-)──>",  # -1权重用(-)表示
                        0: "───────>"    # 0权重无特殊标记
                    }[weight]
                    print(f"{from_node:10} {arrow} {to_node}")
        print("="*40 + "\n")



    def generate_html(self, results = example_results):
        assert isinstance(results, str)

        import json
        result_path = results
        results = json.load(open(results))
        """生成交互式可视化HTML,使用D3.js力导向图布局"""
        nodes_data = []
        for name in self.nodes:
            params = self.nodes[name]
            nodes_data.append({
                "id": name,
                "type": params.get("type", name),
                "params": params
            })

        edges_data = []
        for from_node, to_nodes in self.edges.items():
            for to_node, weight in to_nodes.items():
                edges_data.append({
                    "source": from_node,
                    "target": to_node,
                    "weight": weight
                })

        nodes_js = json.dumps(nodes_data)
        edges_js = json.dumps(edges_data)

        with open('htmls/html_templates/pipeline_template.txt', 'r', encoding='utf-8') as f:
            template = f.read()
        
        template = template.replace('<NODE_JS>', nodes_js).replace('<EDGE_JS>', edges_js).replace('<FILE_PATH>', result_path).replace('<OPTIONS>', "".join(f'<option value="{i}">Result {i+1}</option>' for i in range(len(results))))
        return template
    

    def get_nodes(self):
        nodes_data = []
        for name in self.nodes:
            params = self.nodes[name]
            nodes_data.append({
                "id": name,
                "type": params.get("type", name),
                "params": params
            })
        return json.dumps(nodes_data)

    def get_edges(self):
        edges_data = []
        for from_node, to_nodes in self.edges.items():
            for to_node, weight in to_nodes.items():
                edges_data.append({
                    "source": from_node,
                    "target": to_node,
                    "weight": weight
                })
        return json.dumps(edges_data)
    
    def get_json(self):
        return {
            'nodes': self.get_nodes(),
            'edges': self.get_edges()
        }
    
    def generate_html_embed(self, results = example_results):
        assert isinstance(results, str) or isinstance(results, list)

        import json
        result_path = results
        if isinstance(results, str):
            results = json.load(open(results))
        """生成交互式可视化HTML,使用D3.js力导向图布局"""
        nodes_data = []
        for name in self.nodes:
            params = self.nodes[name]
            nodes_data.append({
                "id": name,
                "type": params.get("type", name),
                "params": params
            })

        edges_data = []
        for from_node, to_nodes in self.edges.items():
            for to_node, weight in to_nodes.items():
                edges_data.append({
                    "source": from_node,
                    "target": to_node,
                    "weight": weight
                })

        nodes_js = json.dumps(nodes_data)
        edges_js = json.dumps(edges_data)

        with open('htmls/html_templates/pipeline_template_emb.txt', 'r', encoding='utf-8') as f:
            template = f.read()
        
        template = template.replace('<NODE_JS>', nodes_js).replace('<EDGE_JS>', edges_js).replace('<RESULTS>', json.dumps(results)).replace('<OPTIONS>', "".join(f'<option value="{i}">Result {i+1}</option>' for i in range(len(results))))
        return template
    
if __name__ == '__main__':
    # 初始化图
    graph = PipelineGraph()

    # 添加节点
    graph.add_node('processor', {'type': 'transform', 'rate': 0.8})
    graph.add_node('validator', {'threshold': 0.5})

    # 添加边
    graph.add_edge('input', 'processor', 1)
    graph.add_edge('processor', 'validator', -1)
    graph.add_edge('validator', 'output', 0)

    # 更新参数
    graph.update_node_params('processor', {'rate': 1.0})

    graph.visualize()