File size: 6,003 Bytes
58dc3e5
2201f72
1475fdc
 
 
2189307
32e2d9a
1475fdc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6b8b11d
1475fdc
6b8b11d
1475fdc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2189307
2711826
2189307
2711826
2189307
2711826
 
2189307
2711826
 
 
 
2189307
 
2711826
 
 
 
 
 
 
 
 
 
 
 
 
2189307
 
1475fdc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2711826
4bf1376
6557388
1475fdc
6557388
1475fdc
2711826
6557388
ff00b17
1475fdc
 
 
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
import os
import json
from typing import Optional
import gradio as gr
from gradio import Interface, Blocks
import networkx as nx
import pyvis
from pyvis.network import Network
from smolagents import CodeAgent, HfApiModel, tool
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 HF_TOKEN environment variable
os.environ["HF_TOKEN"] = os.getenv("HF_TOKEN")

# 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)

# Define examples
examples = [
    ["Highlight the Medici family's central role in the network using all three centrality metrics."],
    ["Focus on the Strozzi family's role in the network using betweenness centrality."],
    ["Compare family positions in the network using degree and closeness centrality."],
    ["Visualize the overall network structure and the relative positions of the families."],
    ["Explore the roles of the Florentine families with the highest centrality values."]
]

from smolagents import GradioUI
import gradio as gr

from smolagents import GradioUI
import gradio as gr

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()

# Define a tool for analyzing the Florentine Families graph
@tool
def analyze_florentine_graph(metric: str) -> str:
    """
    Analyzes the Florentine Families graph based on the chosen centrality metric.

    Args:
        metric: The centrality metric to calculate. Valid options: 'degree', 'betweenness', or 'closeness'.

    Returns:
        A textual analysis of the chosen metric for the graph with formatted numerical results.
    """
    graph = nx.florentine_families_graph()

    if metric == "degree":
        centrality = nx.degree_centrality(graph)
    elif metric == "betweenness":
        centrality = nx.betweenness_centrality(graph)
    elif metric == "closeness":
        centrality = nx.closeness_centrality(graph)
    else:
        return "Invalid metric. Please choose 'degree', 'betweenness', or 'closeness'."

    analysis = f"Analysis of {metric} centrality:\n"
    for node, value in centrality.items():
        analysis += f"- {node}: {value:.3f}\n"
    return analysis

from smolagents import tool

@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}"

from smolagents import tool

@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 to JSON format.

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

    Returns:
        str: The JSON representation of the graph.
    """
    graph = nx.node_link_graph(graph_data)
    json_output = json.dumps(nx.node_link_data(graph), indent=4)
    return json_output

# Initialize the agent with proper configuration
model = HfApiModel()
agent = CodeAgent(
    tools=[analyze_florentine_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"],
    add_base_tools=True
)

# Use the fixed GradioUIWithExamples
interface = GradioUIWithExamples(agent, examples=examples)
interface.launch()