alternate.py

import os
import json
from typing import Optional
import gradio as gr
from gradio import Interface, Blocks
import networkx as nx
import pandas as pd
import matplotlib.pyplot as plt
import community as community_louvain
import pyvis
from pyvis.network import Network
from smolagents import CodeAgent, HfApiModel, tool, GradioUI
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from openinference.instrumentation.smolagents import SmolagentsInstrumentor

# Set up telemetry
PHOENIX_API_KEY = os.getenv("PHOENIX_API_KEY")
api_key = f"api_key={PHOENIX_API_KEY}"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = api_key
os.environ["PHOENIX_CLIENT_HEADERS"] = api_key
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com"

# Updated endpoint from local to cloud
endpoint = "https://app.phoenix.arize.com/v1/traces"
trace_provider = TracerProvider()
trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint)))
SmolagentsInstrumentor().instrument(tracer_provider=trace_provider)

examples = [
    ["Analyze the degree, betweenness, and closeness centrality metrics for all families in the network, and highlight families with the highest values for each metric."],
    ["Identify families with significant betweenness centrality and discuss their potential influence in the network."],
    ["Compare the top three families by degree centrality with their closeness centrality rankings, and explain the differences."],
    ["Visualize the network structure, emphasizing families with high centrality values using color or size variations."],
    ["Explore the roles of families with above-average centrality values across all metrics and discuss their positions in the network."]
]

class GradioUIWithExamples(GradioUI):
    def __init__(self, agent, examples=None, **kwargs):
        super().__init__(agent, **kwargs)
        self.examples = examples

    def build_interface(self):
        with gr.Blocks() as demo:
            gr.Markdown("## Florentine Families Network Analysis")

            # Main Input/Output
            input_box = gr.Textbox(
                label="Your Question",
                placeholder="Type your question about the Florentine Families graph...",
            )
            output_box = gr.Textbox(
                label="Agent's Response",
                placeholder="Response will appear here...",
                interactive=False,
            )
            submit_button = gr.Button("Submit")

            # Link submit button to agent logic
            submit_button.click(
                self.agent.run,
                inputs=input_box,
                outputs=output_box,
            )

            # Add Examples
            if self.examples:
                gr.Markdown("### Examples")
                for example in self.examples:
                    gr.Button(example[0]).click(
                        lambda x=example[0]: x,  # Populate input box
                        inputs=[],
                        outputs=input_box,
                    )
        return demo

    def launch(self):
        # Use the custom-built interface instead of the base class's logic
        demo = self.build_interface()
        demo.launch()

# Initialize graph
graph = nx.florentine_families_graph()
#graph = nx.les_miserables_graph()

@tool
def analyze_graph(graph: nx.Graph, metrics: Optional[str] = None, visualize: Optional[bool] = False) -> dict:
    """
    Performs an in-depth analysis of the Florentine families graph, a predefined social network representing relationships between Renaissance Florentine families. This graph has already been initialized and should be used for all analyses unless another graph is explicitly provided.

    Args:
        graph: A networkx graph object to analyze. This is a required argument.
        metrics: A comma-separated string of centrality metrics to calculate. 
            Valid options include: 'degree', 'betweenness', 'closeness', 'eigenvector', 
            'density', 'clustering_coefficient'. If None, all metrics will be calculated.
        visualize: A boolean indicating whether to generate visualizations for the graph and its metrics.

    Returns:
        A dictionary containing:
        - 'metrics': Numerical results for the requested centrality metrics.
        - 'graph_summary': High-level statistics about the graph (number of nodes, edges, density, etc.).
        - 'community_info': Detected communities, if applicable.
        - 'visualizations': Paths to generated visualization files, if visualize is True.

    Note:
        - This tool defaults to analyzing the Florentine families graph. If a different graph is provided, it will override the default.
        - Ensure that the 'metrics' argument contains valid options to avoid errors.
    """

    if metrics:
        metrics = [metric.strip() for metric in metrics.split(',')]
    else:
        metrics = ['degree', 'betweenness', 'closeness', 'eigenvector', 'density', 'clustering_coefficient']

    # Graph summary
    graph_summary = {
        "number_of_nodes": graph.number_of_nodes(),
        "number_of_edges": graph.number_of_edges(),
        "density": nx.density(graph),
        "average_clustering": nx.average_clustering(graph),
        "connected_components": len(list(nx.connected_components(graph))),
    }

    # Compute requested metrics
    computed_metrics = {}
    if 'degree' in metrics:
        computed_metrics['degree_centrality'] = nx.degree_centrality(graph)
    if 'betweenness' in metrics:
        computed_metrics['betweenness_centrality'] = nx.betweenness_centrality(graph)
    if 'closeness' in metrics:
        computed_metrics['closeness_centrality'] = nx.closeness_centrality(graph)
    if 'eigenvector' in metrics:
        computed_metrics['eigenvector_centrality'] = nx.eigenvector_centrality(graph)
    if 'density' in metrics:
        computed_metrics['density'] = nx.density(graph)
    if 'clustering_coefficient' in metrics:
        computed_metrics['clustering_coefficient'] = nx.average_clustering(graph)

    # Community detection
    communities = community_louvain.best_partition(graph)

    # Visualizations
    visualizations = []
    if visualize:
        pos = nx.spring_layout(graph)
        plt.figure(figsize=(10, 8))
        nx.draw(
            graph,
            pos,
            with_labels=True,
            node_size=700,
            node_color=list(communities.values()),
            cmap=plt.cm.rainbow,
        )
        plt.title("Graph Visualization - Communities")
        viz_path = "graph_communities.png"
        plt.savefig(viz_path)
        visualizations.append(viz_path)

    return {
        "metrics": computed_metrics,
        "graph_summary": graph_summary,
        "community_info": communities,
        "visualizations": visualizations if visualize else "Visualizations not generated.",
    }

@tool
def save_html_to_file(html_content: str, file_path: str) -> str:
    """
    Saves the provided HTML content to a file.

    Args:
        html_content: The HTML content to save.
        file_path: The path where the HTML file will be saved.

    Returns:
        A confirmation message upon successful saving.
    """
    with open(file_path, 'w', encoding='utf-8') as file:
        file.write(html_content)
    return f"HTML content successfully saved to {file_path}"

@tool
def read_html_from_file(file_path: str) -> str:
    """
    Reads HTML content from a file.

    Args:
        file_path: The path of the HTML file to read.

    Returns:
        The HTML content as a string.
    """
    with open(file_path, 'r', encoding='utf-8') as file:
        html_content = file.read()
    return html_content

@tool
def export_graph_to_json(graph_data: dict) -> str:
    """
    Exports a NetworkX graph represented as a dictionary in node-link format to JSON.

    Args:
        graph_data: The graph data in node-link format.

    Returns:
        str: The JSON representation of the graph.

    """
    try:
        graph = nx.node_link_graph(graph_data, edges="edges")
        json_output = json.dumps(nx.node_link_data(graph), indent=4)
        return json_output
    except Exception as e:
        return f"Error exporting graph to JSON: {str(e)}"

model = HfApiModel()
agent = CodeAgent(
    tools=[analyze_graph, save_html_to_file, read_html_from_file, export_graph_to_json],
    model=model,
    additional_authorized_imports=["gradio","networkx","community_louvain","pyvis","matplotlib","json", "pandas"],
    add_base_tools=True
)

interface = GradioUIWithExamples(agent, examples=examples)
interface.launch()