Build sophisticated workflows visually • Import/Export JSON • Generate interactive UI for end-users
"""
)
# API Status Display
with gr.Accordion("🔌 API Status", open=False):
gr.Markdown(f"""
**Available APIs:**
- FRIENDLI_TOKEN: {'✅ Connected' if os.getenv("FRIENDLI_TOKEN") else '❌ Not found'}
- OPENAI_API_KEY: {'✅ Connected' if os.getenv("OPENAI_API_KEY") else '❌ Not found'}
- ANTHROPIC_API_KEY: {'✅ Connected' if os.getenv("ANTHROPIC_API_KEY") else '❌ Not found'}
**Libraries:**
- OpenAI: {'✅ Installed' if OPENAI_AVAILABLE else '❌ Not installed'}
- Anthropic: {'✅ Installed' if ANTHROPIC_AVAILABLE else '❌ Not installed'}
- Requests: {'✅ Installed' if REQUESTS_AVAILABLE else '❌ Not installed'}
*Note: Without API keys, the UI will simulate AI responses.*
""")
# State for storing workflow data
loaded_data = gr.State(None)
trigger_update = gr.State(False)
# ─── Dynamic Workflow Container ───
with gr.Column(elem_classes=["workflow-container"]):
@gr.render(inputs=[loaded_data, trigger_update])
def render_workflow(data, trigger):
"""동적으로 WorkflowBuilder 렌더링"""
workflow_value = data if data else {"nodes": [], "edges": []}
return WorkflowBuilder(
label="🎨 Visual Workflow Designer",
info="Drag from sidebar → Connect nodes → Edit properties",
value=workflow_value,
elem_id="main_workflow"
)
# ─── Import Section ───
with gr.Accordion("📥 Import Workflow", open=True):
with gr.Row():
with gr.Column(scale=2):
import_json_text = gr.Code(
language="json",
label="Paste JSON here",
lines=8,
value='{\n "nodes": [],\n "edges": []\n}'
)
with gr.Column(scale=1):
file_upload = gr.File(
label="Or upload JSON file",
file_types=[".json"],
type="filepath"
)
btn_load = gr.Button("📥 Load Workflow", variant="primary", size="lg")
btn_sample = gr.Button("🎯 Load Sample", variant="secondary")
# Status
status_text = gr.Textbox(
label="Status",
value="Ready",
elem_classes=["status-box"],
interactive=False
)
# ─── Export Section ───
gr.Markdown("## 💾 Export")
with gr.Row():
with gr.Column(scale=3):
export_preview = gr.Code(
language="json",
label="Current Workflow JSON",
lines=8
)
with gr.Column(scale=1):
btn_preview = gr.Button("👁️ Preview JSON", size="lg")
btn_download = gr.DownloadButton("💾 Download JSON", size="lg")
# ─── UI Execution Section ───
with gr.Column(elem_classes=["ui-execution-section"]):
gr.Markdown("## 🚀 UI Execution")
gr.Markdown("Generate an interactive UI from your workflow for end-users")
btn_execute_ui = gr.Button("▶️ Generate & Run UI", variant="primary", size="lg")
# UI execution state
ui_workflow_data = gr.State(None)
# Dynamic UI container
@gr.render(inputs=[ui_workflow_data])
def render_execution_ui(workflow_data):
if not workflow_data or not workflow_data.get("nodes"):
gr.Markdown("*Load a workflow first, then click 'Generate & Run UI'*")
return
gr.Markdown("### 📋 Generated UI")
# Extract input and output nodes
input_nodes = []
output_nodes = []
for node in workflow_data.get("nodes", []):
node_type = node.get("type", "")
if node_type in ["ChatInput", "textInput", "Input", "numberInput"]:
input_nodes.append(node)
elif node_type in ["ChatOutput", "textOutput", "Output"]:
output_nodes.append(node)
# Create input components
input_components = {}
if input_nodes:
gr.Markdown("#### 📥 Inputs")
for node in input_nodes:
node_id = node.get("id")
label = node.get("data", {}).get("label", node_id)
node_type = node.get("type")
# Get default value
template = node.get("data", {}).get("template", {})
default_value = template.get("input_value", {}).get("value", "")
if node_type == "numberInput":
input_components[node_id] = gr.Number(
label=label,
value=float(default_value) if default_value else 0
)
else:
input_components[node_id] = gr.Textbox(
label=label,
value=default_value,
lines=2,
placeholder="Enter your input..."
)
# Execute button
execute_btn = gr.Button("🎯 Execute", variant="primary")
# Create output components
output_components = {}
if output_nodes:
gr.Markdown("#### 📤 Outputs")
for node in output_nodes:
node_id = node.get("id")
label = node.get("data", {}).get("label", node_id)
output_components[node_id] = gr.Textbox(
label=label,
interactive=False,
lines=3
)
# Execution log
gr.Markdown("#### 📊 Execution Log")
log_output = gr.Textbox(
label="Log",
interactive=False,
lines=5
)
# Define execution handler
def execute_ui_workflow(*input_values):
# Create input dictionary
inputs_dict = {}
input_keys = list(input_components.keys())
for i, key in enumerate(input_keys):
if i < len(input_values):
inputs_dict[key] = input_values[i]
# Check API status
log = "=== Workflow Execution Started ===\n"
log += f"Inputs provided: {len(inputs_dict)}\n"
# API 상태 확인
friendli_token = os.getenv("FRIENDLI_TOKEN")
openai_key = os.getenv("OPENAI_API_KEY")
anthropic_key = os.getenv("ANTHROPIC_API_KEY")
log += "\nAPI Status:\n"
log += f"- FRIENDLI_TOKEN: {'✅ Found' if friendli_token else '❌ Not found'}\n"
log += f"- OPENAI_API_KEY: {'✅ Found' if openai_key else '❌ Not found'}\n"
log += f"- ANTHROPIC_API_KEY: {'✅ Found' if anthropic_key else '❌ Not found'}\n"
if not friendli_token and not openai_key and not anthropic_key:
log += "\n⚠️ No API keys found. Results will be simulated.\n"
log += "To get real AI responses, set API keys in environment variables.\n"
log += "\n--- Processing Nodes ---\n"
try:
results = execute_workflow_simple(workflow_data, inputs_dict)
# Prepare outputs
output_values = []
for node_id in output_components.keys():
value = results.get(node_id, "No output")
output_values.append(value)
# Log 길이 제한
display_value = value[:100] + "..." if len(str(value)) > 100 else value
log += f"\nOutput [{node_id}]: {display_value}\n"
log += "\n=== Execution Completed Successfully! ===\n"
output_values.append(log)
return output_values
except Exception as e:
error_msg = f"❌ Error: {str(e)}"
log += f"\n{error_msg}\n"
log += "=== Execution Failed ===\n"
return [error_msg] * len(output_components) + [log]
# Connect execution
all_inputs = list(input_components.values())
all_outputs = list(output_components.values()) + [log_output]
execute_btn.click(
fn=execute_ui_workflow,
inputs=all_inputs,
outputs=all_outputs
)
# ─── Event Handlers ───
# Load workflow (from text or file)
def load_workflow(json_text, file_obj):
data, status = load_json_from_text_or_file(json_text, file_obj)
if data:
return data, status, json_text if not file_obj else export_pretty(data)
else:
return None, status, gr.update()
btn_load.click(
fn=load_workflow,
inputs=[import_json_text, file_upload],
outputs=[loaded_data, status_text, import_json_text]
).then(
fn=lambda current_trigger: not current_trigger,
inputs=trigger_update,
outputs=trigger_update
)
# Auto-load when file is uploaded
file_upload.change(
fn=load_workflow,
inputs=[import_json_text, file_upload],
outputs=[loaded_data, status_text, import_json_text]
).then(
fn=lambda current_trigger: not current_trigger,
inputs=trigger_update,
outputs=trigger_update
)
# Load sample
btn_sample.click(
fn=lambda: (create_sample_workflow(), "✅ Sample loaded", export_pretty(create_sample_workflow())),
outputs=[loaded_data, status_text, import_json_text]
).then(
fn=lambda current_trigger: not current_trigger,
inputs=trigger_update,
outputs=trigger_update
)
# Preview current workflow
btn_preview.click(
fn=export_pretty,
inputs=loaded_data,
outputs=export_preview
)
# Download workflow
btn_download.click(
fn=export_file,
inputs=loaded_data
)
# Generate UI execution
btn_execute_ui.click(
fn=lambda data: data,
inputs=loaded_data,
outputs=ui_workflow_data
)
# Auto-update export preview when workflow changes
loaded_data.change(
fn=export_pretty,
inputs=loaded_data,
outputs=export_preview
)
# -------------------------------------------------------------------
# 🚀 실행
# -------------------------------------------------------------------
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", show_error=True)