Buckets:
| from __future__ import annotations | |
| import json | |
| import os | |
| import time | |
| from typing import Any, Dict, List, Optional, Tuple, Union | |
| import matplotlib.pyplot as plt | |
| import networkx as nx | |
| import numpy as np | |
| import plotly.graph_objects as go | |
| import plotly.express as px | |
| from plotly.subplots import make_subplots | |
| import seaborn as sns | |
| from matplotlib.patches import FancyBboxPatch | |
| from matplotlib.colors import LinearSegmentedColormap | |
| from core.state import ThoughtState, StateType | |
| from core.tree import TreeOfThought | |
| from algorithms.search import SearchResult | |
| class TreeVisualizer: | |
| def __init__(self, output_dir: str = "visualizations"): | |
| self.output_dir = output_dir | |
| self._ensure_output_dir() | |
| plt.style.use("seaborn-v0_8") | |
| self.colors = { | |
| "initial": "#2E8B57", | |
| "intermediate": "#4682B4", | |
| "terminal": "#DC143C", | |
| "solution": "#32CD32", | |
| "dead_end": "#8B0000", | |
| "path": "#FFD700", | |
| "highlight": "#FF6347", | |
| } | |
| self.value_cmap = LinearSegmentedColormap.from_list( | |
| "value_cmap", ["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FFEAA7"] | |
| ) | |
| def _ensure_output_dir(self): | |
| if not os.path.exists(self.output_dir): | |
| os.makedirs(self.output_dir) | |
| def visualize_tree_structure( | |
| self, | |
| tree: TreeOfThought, | |
| title: str = "Tree of Thought Structure", | |
| save_path: Optional[str] = None, | |
| ) -> str: | |
| fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(20, 10)) | |
| self._plot_tree_graph(tree, ax1) | |
| ax1.set_title(f"{title} - Structure", fontsize=14, fontweight="bold") | |
| self._plot_tree_statistics(tree, ax2) | |
| ax2.set_title(f"{title} - Statistics", fontsize=14, fontweight="bold") | |
| plt.tight_layout() | |
| if save_path is None: | |
| timestamp = int(time.time()) | |
| save_path = os.path.join(self.output_dir, f"tree_structure_{timestamp}.png") | |
| plt.savefig(save_path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| return save_path | |
| def _plot_tree_graph(self, tree: TreeOfThought, ax): | |
| if not tree.root_id: | |
| ax.text( | |
| 0.5, 0.5, "Empty Tree", ha="center", va="center", transform=ax.transAxes | |
| ) | |
| return | |
| G = nx.DiGraph() | |
| for state in tree.state_manager.get_all_states(): | |
| G.add_node( | |
| state.state_id, | |
| content=( | |
| state.content[:30] + "..." | |
| if len(state.content) > 30 | |
| else state.content | |
| ), | |
| value=state.value, | |
| depth=state.depth, | |
| state_type=state.state_type, | |
| ) | |
| for action in tree.actions: | |
| if G.has_node(action.from_state) and G.has_node(action.to_state): | |
| G.add_edge( | |
| action.from_state, | |
| action.to_state, | |
| description=( | |
| action.description[:20] + "..." | |
| if len(action.description) > 20 | |
| else action.description | |
| ), | |
| ) | |
| if not G.nodes(): | |
| ax.text( | |
| 0.5, 0.5, "No Nodes", ha="center", va="center", transform=ax.transAxes | |
| ) | |
| return | |
| try: | |
| pos = nx.spring_layout(G, k=3, iterations=50) | |
| except: | |
| pos = nx.random_layout(G) | |
| node_colors = [] | |
| node_sizes = [] | |
| for node in G.nodes(): | |
| state_type = G.nodes[node]["state_type"] | |
| value = G.nodes[node]["value"] | |
| if state_type == StateType.INITIAL: | |
| node_colors.append(self.colors["initial"]) | |
| elif state_type == StateType.SOLUTION: | |
| node_colors.append(self.colors["solution"]) | |
| elif state_type == StateType.DEAD_END: | |
| node_colors.append(self.colors["dead_end"]) | |
| elif state_type == StateType.TERMINAL: | |
| node_colors.append(self.colors["terminal"]) | |
| else: | |
| node_colors.append(self.colors["intermediate"]) | |
| node_sizes.append(300 + value * 200) | |
| nx.draw_networkx_edges( | |
| G, pos, ax=ax, edge_color="gray", alpha=0.6, arrows=True, arrowsize=20 | |
| ) | |
| nx.draw_networkx_nodes( | |
| G, pos, ax=ax, node_color=node_colors, node_size=node_sizes, alpha=0.8 | |
| ) | |
| labels = {node: f"{node[:8]}\n{G.nodes[node]['content']}" for node in G.nodes()} | |
| nx.draw_networkx_labels(G, pos, labels, ax=ax, font_size=8) | |
| legend_elements = [ | |
| plt.Line2D( | |
| [0], | |
| [0], | |
| marker="o", | |
| color="w", | |
| markerfacecolor=self.colors["initial"], | |
| markersize=10, | |
| label="Initial", | |
| ), | |
| plt.Line2D( | |
| [0], | |
| [0], | |
| marker="o", | |
| color="w", | |
| markerfacecolor=self.colors["intermediate"], | |
| markersize=10, | |
| label="Intermediate", | |
| ), | |
| plt.Line2D( | |
| [0], | |
| [0], | |
| marker="o", | |
| color="w", | |
| markerfacecolor=self.colors["solution"], | |
| markersize=10, | |
| label="Solution", | |
| ), | |
| plt.Line2D( | |
| [0], | |
| [0], | |
| marker="o", | |
| color="w", | |
| markerfacecolor=self.colors["terminal"], | |
| markersize=10, | |
| label="Terminal", | |
| ), | |
| plt.Line2D( | |
| [0], | |
| [0], | |
| marker="o", | |
| color="w", | |
| markerfacecolor=self.colors["dead_end"], | |
| markersize=10, | |
| label="Dead End", | |
| ), | |
| ] | |
| ax.legend(handles=legend_elements, loc="upper right", bbox_to_anchor=(1.15, 1)) | |
| ax.set_aspect("equal") | |
| ax.axis("off") | |
| def _plot_tree_statistics(self, tree: TreeOfThought, ax): | |
| stats = tree.get_tree_statistics() | |
| categories = [ | |
| "Total States", | |
| "Total Actions", | |
| "Max Depth", | |
| "Solutions", | |
| "Branching Factor", | |
| ] | |
| values = [ | |
| stats.get("total_states", 0), | |
| stats.get("total_actions", 0), | |
| stats.get("max_depth", 0), | |
| stats.get("solutions_found", 0), | |
| round(stats.get("branching_factor", 0), 2), | |
| ] | |
| bars = ax.bar( | |
| categories, | |
| values, | |
| color=["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FFEAA7"], | |
| ) | |
| for bar, value in zip(bars, values): | |
| height = bar.get_height() | |
| ax.text( | |
| bar.get_x() + bar.get_width() / 2.0, | |
| height + 0.1, | |
| f"{value}", | |
| ha="center", | |
| va="bottom", | |
| fontweight="bold", | |
| ) | |
| ax.set_ylabel("Count/Value") | |
| ax.set_title("Tree Statistics") | |
| plt.setp(ax.get_xticklabels(), rotation=45, ha="right") | |
| def visualize_search_path( | |
| self, | |
| search_result: SearchResult, | |
| tree: TreeOfThought, | |
| title: str = "Search Path Visualization", | |
| save_path: Optional[str] = None, | |
| ) -> str: | |
| fig, ax = plt.subplots(figsize=(15, 10)) | |
| if not search_result.success or not search_result.path: | |
| ax.text( | |
| 0.5, | |
| 0.5, | |
| f"Search Failed: {search_result.error}", | |
| ha="center", | |
| va="center", | |
| transform=ax.transAxes, | |
| fontsize=16, | |
| ) | |
| ax.set_title(title) | |
| if save_path is None: | |
| timestamp = int(time.time()) | |
| save_path = os.path.join( | |
| self.output_dir, f"search_path_failed_{timestamp}.png" | |
| ) | |
| plt.savefig(save_path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| return save_path | |
| G = nx.DiGraph() | |
| path_state_ids = [state.state_id for state in search_result.path] | |
| for state in search_result.path: | |
| G.add_node( | |
| state.state_id, | |
| content=( | |
| state.content[:40] + "..." | |
| if len(state.content) > 40 | |
| else state.content | |
| ), | |
| value=state.value, | |
| depth=state.depth, | |
| is_in_path=True, | |
| ) | |
| for i in range(len(search_result.path) - 1): | |
| current_state = search_result.path[i] | |
| next_state = search_result.path[i + 1] | |
| G.add_edge(current_state.state_id, next_state.state_id, is_path_edge=True) | |
| try: | |
| pos = nx.spring_layout(G, k=2, iterations=50) | |
| except: | |
| pos = nx.random_layout(G) | |
| nx.draw_networkx_edges( | |
| G, pos, ax=ax, edge_color="gray", alpha=0.3, arrows=True, arrowsize=15 | |
| ) | |
| path_edges = [ | |
| (search_result.path[i].state_id, search_result.path[i + 1].state_id) | |
| for i in range(len(search_result.path) - 1) | |
| ] | |
| nx.draw_networkx_edges( | |
| G, | |
| pos, | |
| edgelist=path_edges, | |
| ax=ax, | |
| edge_color=self.colors["path"], | |
| width=3, | |
| alpha=0.8, | |
| arrows=True, | |
| arrowsize=20, | |
| ) | |
| node_colors = [] | |
| node_sizes = [] | |
| for node in G.nodes(): | |
| value = G.nodes[node]["value"] | |
| node_colors.append(self.value_cmap(value)) | |
| node_sizes.append(400 + value * 300) | |
| nx.draw_networkx_nodes( | |
| G, pos, ax=ax, node_color=node_colors, node_size=node_sizes, alpha=0.9 | |
| ) | |
| labels = {node: f"{node[:6]}\n{G.nodes[node]['content']}" for node in G.nodes()} | |
| nx.draw_networkx_labels(G, pos, labels, ax=ax, font_size=9) | |
| path_info = ( | |
| f"Path Length: {search_result.path_length}\n" | |
| f"Final Value: {search_result.final_value:.3f}\n" | |
| f"Nodes Explored: {search_result.nodes_explored}\n" | |
| f"Execution Time: {search_result.execution_time:.3f}s" | |
| ) | |
| ax.text( | |
| 0.02, | |
| 0.98, | |
| path_info, | |
| transform=ax.transAxes, | |
| fontsize=10, | |
| verticalalignment="top", | |
| bbox=dict(boxstyle="round", facecolor="wheat", alpha=0.8), | |
| ) | |
| ax.set_title(title, fontsize=16, fontweight="bold") | |
| ax.axis("off") | |
| if save_path is None: | |
| timestamp = int(time.time()) | |
| save_path = os.path.join(self.output_dir, f"search_path_{timestamp}.png") | |
| plt.savefig(save_path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| return save_path | |
| def visualize_performance_comparison( | |
| self, | |
| search_results: Dict[str, SearchResult], | |
| title: str = "Search Algorithm Performance Comparison", | |
| save_path: Optional[str] = None, | |
| ) -> str: | |
| fig, axes = plt.subplots(2, 2, figsize=(16, 12)) | |
| fig.suptitle(title, fontsize=16, fontweight="bold") | |
| algorithms = list(search_results.keys()) | |
| success_rates = [ | |
| 1 if result.success else 0 for result in search_results.values() | |
| ] | |
| path_lengths = [result.path_length for result in search_results.values()] | |
| final_values = [result.final_value for result in search_results.values()] | |
| nodes_explored = [result.nodes_explored for result in search_results.values()] | |
| execution_times = [result.execution_time for result in search_results.values()] | |
| ax1 = axes[0, 0] | |
| bars1 = ax1.bar( | |
| algorithms, | |
| success_rates, | |
| color=["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4"], | |
| ) | |
| ax1.set_title("Success Rate") | |
| ax1.set_ylabel("Success (1) / Failure (0)") | |
| ax1.set_ylim(0, 1.2) | |
| for bar, rate in zip(bars1, success_rates): | |
| ax1.text( | |
| bar.get_x() + bar.get_width() / 2.0, | |
| bar.get_height() + 0.05, | |
| f"{rate}", | |
| ha="center", | |
| va="bottom", | |
| fontweight="bold", | |
| ) | |
| ax2 = axes[0, 1] | |
| scatter = ax2.scatter( | |
| path_lengths, | |
| final_values, | |
| c=range(len(algorithms)), | |
| cmap="viridis", | |
| s=100, | |
| alpha=0.7, | |
| ) | |
| for i, alg in enumerate(algorithms): | |
| ax2.annotate( | |
| alg, | |
| (path_lengths[i], final_values[i]), | |
| xytext=(5, 5), | |
| textcoords="offset points", | |
| fontsize=9, | |
| ) | |
| ax2.set_xlabel("Path Length") | |
| ax2.set_ylabel("Final Value") | |
| ax2.set_title("Path Length vs Final Value") | |
| ax2.grid(True, alpha=0.3) | |
| ax3 = axes[1, 0] | |
| bars3 = ax3.bar( | |
| algorithms, | |
| nodes_explored, | |
| color=["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4"], | |
| ) | |
| ax3.set_title("Nodes Explored") | |
| ax3.set_ylabel("Number of Nodes") | |
| for bar, nodes in zip(bars3, nodes_explored): | |
| ax3.text( | |
| bar.get_x() + bar.get_width() / 2.0, | |
| bar.get_height() + max(nodes_explored) * 0.01, | |
| f"{nodes}", | |
| ha="center", | |
| va="bottom", | |
| fontsize=9, | |
| ) | |
| ax4 = axes[1, 1] | |
| bars4 = ax4.bar( | |
| algorithms, | |
| execution_times, | |
| color=["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4"], | |
| ) | |
| ax4.set_title("Execution Time") | |
| ax4.set_ylabel("Time (seconds)") | |
| for bar, time_val in zip(bars4, execution_times): | |
| ax4.text( | |
| bar.get_x() + bar.get_width() / 2.0, | |
| bar.get_height() + max(execution_times) * 0.01, | |
| f"{time_val:.3f}s", | |
| ha="center", | |
| va="bottom", | |
| fontsize=9, | |
| ) | |
| plt.tight_layout() | |
| if save_path is None: | |
| timestamp = int(time.time()) | |
| save_path = os.path.join( | |
| self.output_dir, f"performance_comparison_{timestamp}.png" | |
| ) | |
| plt.savefig(save_path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| return save_path | |
| def create_interactive_plot( | |
| self, | |
| tree: TreeOfThought, | |
| search_results: Optional[Dict[str, SearchResult]] = None, | |
| ) -> str: | |
| fig = make_subplots( | |
| rows=2, | |
| cols=2, | |
| subplot_titles=( | |
| "Tree Structure", | |
| "State Values Distribution", | |
| "Depth Analysis", | |
| "Performance Metrics", | |
| ), | |
| specs=[ | |
| [{"type": "scatter"}, {"type": "histogram"}], | |
| [{"type": "bar"}, {"type": "scatter"}], | |
| ], | |
| ) | |
| if tree.root_id: | |
| states = tree.state_manager.get_all_states() | |
| x_coords = [state.depth for state in states] | |
| y_coords = [hash(state.state_id) % 100 for state in states] | |
| values = [state.value for state in states] | |
| colors = [state.state_type.value for state in states] | |
| fig.add_trace( | |
| go.Scatter( | |
| x=x_coords, | |
| y=y_coords, | |
| mode="markers+text", | |
| marker=dict( | |
| size=10, | |
| color=values, | |
| colorscale="Viridis", | |
| colorbar=dict(title="Value"), | |
| ), | |
| text=[state.state_id[:8] for state in states], | |
| textposition="top center", | |
| name="States", | |
| ), | |
| row=1, | |
| col=1, | |
| ) | |
| if tree.state_manager.get_all_states(): | |
| values = [state.value for state in tree.state_manager.get_all_states()] | |
| fig.add_trace( | |
| go.Histogram(x=values, name="Value Distribution", nbinsx=20), | |
| row=1, | |
| col=2, | |
| ) | |
| if tree.state_manager.get_all_states(): | |
| depths = [state.depth for state in tree.state_manager.get_all_states()] | |
| depth_counts = {} | |
| for depth in depths: | |
| depth_counts[depth] = depth_counts.get(depth, 0) + 1 | |
| fig.add_trace( | |
| go.Bar( | |
| x=list(depth_counts.keys()), | |
| y=list(depth_counts.values()), | |
| name="States by Depth", | |
| ), | |
| row=2, | |
| col=1, | |
| ) | |
| if search_results: | |
| algorithms = list(search_results.keys()) | |
| final_values = [result.final_value for result in search_results.values()] | |
| fig.add_trace( | |
| go.Scatter( | |
| x=algorithms, | |
| y=final_values, | |
| mode="markers+lines", | |
| marker=dict(size=15), | |
| name="Final Values", | |
| ), | |
| row=2, | |
| col=2, | |
| ) | |
| fig.update_layout( | |
| title_text="Interactive Tree of Thought Analysis", | |
| showlegend=True, | |
| height=800, | |
| ) | |
| timestamp = int(time.time()) | |
| save_path = os.path.join(self.output_dir, f"interactive_plot_{timestamp}.html") | |
| fig.write_html(save_path) | |
| return save_path | |
| def export_tree_data( | |
| self, | |
| tree: TreeOfThought, | |
| search_results: Optional[Dict[str, SearchResult]] = None, | |
| ) -> str: | |
| data = { | |
| "tree_data": tree.export_tree(), | |
| "search_results": {}, | |
| "export_timestamp": time.time(), | |
| } | |
| if search_results: | |
| for name, result in search_results.items(): | |
| data["search_results"][name] = { | |
| "success": result.success, | |
| "path_length": result.path_length, | |
| "final_value": result.final_value, | |
| "nodes_explored": result.nodes_explored, | |
| "execution_time": result.execution_time, | |
| "error": result.error, | |
| } | |
| timestamp = int(time.time()) | |
| save_path = os.path.join(self.output_dir, f"tree_data_{timestamp}.json") | |
| with open(save_path, "w") as f: | |
| json.dump(data, f, indent=2, default=str) | |
| return save_path | |
| class PerformanceAnalyzer: | |
| def __init__(self, output_dir: str = "visualizations"): | |
| self.output_dir = output_dir | |
| self.visualizer = TreeVisualizer(output_dir) | |
| self.metrics_history: List[Dict[str, Any]] = [] | |
| def analyze_tree_performance( | |
| self, tree: TreeOfThought, search_results: Dict[str, SearchResult] | |
| ) -> Dict[str, Any]: | |
| analysis = { | |
| "tree_statistics": tree.get_tree_statistics(), | |
| "search_performance": {}, | |
| "recommendations": [], | |
| "timestamp": time.time(), | |
| } | |
| for name, result in search_results.items(): | |
| analysis["search_performance"][name] = { | |
| "success": result.success, | |
| "efficiency": result.final_value / max(1, result.nodes_explored), | |
| "speed": result.execution_time, | |
| "quality": result.final_value, | |
| "exploration": result.nodes_explored, | |
| } | |
| if search_results: | |
| best_quality = max(search_results.items(), key=lambda x: x[1].final_value) | |
| fastest = min(search_results.items(), key=lambda x: x[1].execution_time) | |
| most_efficient = max( | |
| search_results.items(), | |
| key=lambda x: x[1].final_value / max(1, x[1].nodes_explored), | |
| ) | |
| analysis["recommendations"] = [ | |
| f"Best quality: {best_quality[0]} (value: {best_quality[1].final_value:.3f})", | |
| f"Fastest: {fastest[0]} (time: {fastest[1].execution_time:.3f}s)", | |
| f"Most efficient: {most_efficient[0]} (efficiency: {most_efficient[1].final_value / max(1, most_efficient[1].nodes_explored):.4f})", | |
| ] | |
| self.metrics_history.append(analysis) | |
| return analysis | |
| def create_performance_report(self, analysis: Dict[str, Any]) -> str: | |
| timestamp = int(time.time()) | |
| report_path = os.path.join( | |
| self.output_dir, f"performance_report_{timestamp}.json" | |
| ) | |
| with open(report_path, "w") as f: | |
| json.dump(analysis, f, indent=2, default=str) | |
| return report_path | |
Xet Storage Details
- Size:
- 21.1 kB
- Xet hash:
- ed6a271698408d4fa497ff7b5d69ececaab5b4c9b258f00658bbc8cb1b246169
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.