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