(self)
| 107 | return eg |
| 108 | |
| 109 | def draw(self): |
| 110 | import pygraphviz as pgv |
| 111 | G = pgv.AGraph(directed=True) |
| 112 | G.add_nodes_from([str(node) for node in self.nodes.values()]) |
| 113 | VIS_CONFIG={ |
| 114 | 'system':{'shape':'plaintext'}, |
| 115 | 'user': {'fillcolor':'yellow','style':'filled','shape':'circle'}, |
| 116 | 'tool': { |
| 117 | # 'fillcolor':'red','style':'filled', |
| 118 | 'shape':'diamond'}, |
| 119 | 'assistant': { |
| 120 | # 'fillcolor':'green','style':'filled', |
| 121 | 'shape':'box'} |
| 122 | } |
| 123 | def wrap_text(text:str, width=20): |
| 124 | wrapped_text = '' |
| 125 | for i in range(0, min(width*5,len(text)), width): |
| 126 | wrapped_text += text[i:i+width] + '\n' |
| 127 | escaped_chars = re.findall(r'\\[nrt\'"\\]', wrapped_text) |
| 128 | for escaped_char in escaped_chars: |
| 129 | wrapped_text = wrapped_text.replace(escaped_char, '') |
| 130 | return wrapped_text |
| 131 | |
| 132 | def set_node_vis(gnode,node:ExecutionNode): |
| 133 | for k,v in VIS_CONFIG[node.role].items(): |
| 134 | gnode.attr[k] = v |
| 135 | |
| 136 | if node.role == 'system': |
| 137 | gnode.attr['label']='SystemPrompt' |
| 138 | elif node.role == 'tool': |
| 139 | if node.message['name'] == 'Finish': |
| 140 | # args = json.loads(node.message['arguments']) |
| 141 | args = str(node.message['arguments']) |
| 142 | idx = args.find('return_type') |
| 143 | |
| 144 | if 'give_answer' in args[idx:idx+30]: |
| 145 | gnode.attr['fillcolor'] = 'green' |
| 146 | # gnode.attr['xlabel'] = f"{wrap_text(args.get('final_answer',''))}" |
| 147 | gnode.attr['label'] = wrap_text(args[args.find('final_answer'):]) |
| 148 | else: |
| 149 | gnode.attr['fillcolor'] = 'red' |
| 150 | gnode.attr['label'] = 'restart' |
| 151 | gnode.attr['style'] = 'filled' |
| 152 | gnode.attr['shape'] = 'ellipse' |
| 153 | else: |
| 154 | gnode.attr['label'] = f"tool: {wrap_text(node.message['name'])}" |
| 155 | gnode.attr['xlabel'] = f"{wrap_text(node.message['response'])}" |
| 156 | elif node.role =='assistant': |
| 157 | gnode.attr['label'] = node.role.upper() +'\n'+ wrap_text(str(node.message)) |
| 158 | else: |
| 159 | gnode.attr['xlabel'] = wrap_text(str(node.message)) |
| 160 | gnode.attr['label'] = node.role.upper() |
| 161 | |
| 162 | for node in self.nodes.values(): |
| 163 | gnode = G.get_node(str(node)) |
| 164 | set_node_vis(gnode,node) |
| 165 | to_nodes = list(self.edges.get(node.node_id,{}).keys()) |
| 166 | G.add_edges_from([(str(node),str(to_node)) for to_node in to_nodes]) |
nothing calls this directly
no outgoing calls
no test coverage detected