iTranslated by AI
The content below is an AI-generated translation. This is an experimental feature, and may contain errors. View original article
❄️
Automated Flowchart Generation App with Streamlit in Snowflake (SiS) and Cortex AI
Introduction
In this article, I will introduce an app I created that combines Streamlit in Snowflake and Cortex AI to automatically generate flowcharts and architecture diagrams. There are still some points I want to improve, so there's a high possibility I'll write a Part 2 article later!
What is Streamlit in Snowflake (SiS)?
First of all, Streamlit is a Python library. With Streamlit, you can create a Web UI using only simple Python code. Originally, Web UIs had to be written using HTML / CSS / JavaScript, etc., but you can think of Streamlit as a tool that acts as a substitute for those functions.
This App Gallery contains samples posted by users, which makes it easy to visualize.
Streamlit in Snowflake (SiS) allows you to develop and run Streamlit web apps directly on Snowflake. One of its main attractions is the ease of use—as long as you have a Snowflake account, you can start using it. More importantly, the ability to easily integrate table data and other resources from Snowflake into a web app is a huge advantage.
About Streamlit in Snowflake (Snowflake Official Documentation)
What is Snowflake Cortex?
Snowflake Cortex is a suite of generative AI features within Snowflake. Among them, Cortex LLM is a feature that allows you to call large language models running on Snowflake, making it possible to use various types of LLMs through simple functions from SQL or Python.
Large Language Model (LLM) Functions (Snowflake Cortex) (Snowflake Official Documentation)
Functional Overview
Feature List
The app created this time implements the following features:
- Graphical editor for creating flowcharts and architecture diagrams
- Ability to interactively add and edit nodes and edges
- Customizable shapes and style settings
- Graph generation from the configured items
- Template function
- Graph generation from several types of templates
- Ability to reflect template content into the graphical editor
- DOT code direct editing function
- Direct input of Graphviz syntax in DOT code format
- Graph generation from the entered DOT code
- AI generation function
- Automatic generation of flowcharts from natural language descriptions
Finished Image
Overall view of the app
Graphical editor for nodes and edges
List display of nodes and edges
Graph and DOT code generated from the graphical editor
Template function
List display of template nodes and edges and DOT code
DOT code direct input function
Automatic graph generation function using Cortex AI
Graph generated by Cortex AI
Prerequisites
- Snowflake Account
- A Snowflake account where Cortex LLM is available (With the release of cross-region inference, cloud and region constraints have been almost eliminated).
- Streamlit in Snowflake installed packages
- python 3.11 or later
- snowflake-ml-python 1.7.4 or later
- python-graphviz 0.20.1 or later
*Refer to the Cortex LLM region availability table (Snowflake Official Documentation)
What is Graphviz?
Graphviz is an open-source library for visualizing graph structures. It is well-suited for drawing directed and undirected graphs consisting of nodes and edges, and can be used to create various diagrams such as network diagrams, flowcharts, organizational charts, and ER diagrams.
Graphviz has the following features:
- Uses a dedicated description language called DOT code
- Automatic layout functionality (automatically optimizes node placement)
- Supports a wide variety of node shapes and edge styles
Since the python-graphviz library can be easily installed in Streamlit in Snowflake, we will build a flowchart creation app using Graphviz.
Steps
Creating a New Streamlit in Snowflake App
Click on "Streamlit" in the left pane of Snowsight, then click the "+ Streamlit" button to create an SiS app.
Running the Streamlit in Snowflake App
In the editor screen of your Streamlit in Snowflake app, copy and paste the following code. (The code is quite long, so please open the accordion to view it).
Source Code
import streamlit as st
import graphviz
import pandas as pd
from snowflake.snowpark.context import get_active_session
from snowflake.cortex import Complete as CompleteText
# Application configuration
st.set_page_config(
layout="wide",
initial_sidebar_state="expanded"
)
# Get Snowflake session
session = get_active_session()
# Application header
st.title("Streamlit in Snowflake Flowchart Creation Tool")
st.markdown("A tool for creating architecture and data flow diagrams using Cortex AI and Graphviz.")
# Common function: Display node list
def display_nodes_table(nodes_list, is_template=False):
if not nodes_list:
st.info("No nodes have been added yet.")
return
nodes_data = {
"Node ID": [],
"Label": [],
"Shape": [],
"Style": [],
"Fill Color": [],
"Border Color": [],
"Details": []
}
for node in nodes_list:
# Convert node structure if it's a template
if is_template:
node_id = node["id"]
node_label = node["label"]
node_shape = node["attrs"].get("shape", "box")
node_style = "filled"
node_fillcolor = node["attrs"].get("fillcolor", "#D0E8FF")
node_color = "#000000"
nodes_data["Node ID"].append(node_id)
nodes_data["Label"].append(node_label)
nodes_data["Shape"].append(node_shape)
nodes_data["Style"].append(node_style)
nodes_data["Fill Color"].append(f"<span style='color:{node_fillcolor};'>■</span> {node_fillcolor}")
nodes_data["Border Color"].append(f"<span style='color:#000000;'>■</span> #000000")
nodes_data["Details"].append("-")
else:
nodes_data["Node ID"].append(node['id'])
nodes_data["Label"].append(node['label'])
nodes_data["Shape"].append(node['shape'])
nodes_data["Style"].append(node['style'])
nodes_data["Fill Color"].append(f"<span style='color:{node['fillcolor']};'>■</span> {node['fillcolor']}")
nodes_data["Border Color"].append(f"<span style='color:{node['color']};'>■</span> {node['color']}")
# Summarize advanced settings information
details = []
if 'peripheries' in node and node['peripheries'] > 1:
details.append(f"Peripheries: {node['peripheries']}")
if 'fontname' in node and node['fontname'] != 'sans-serif':
details.append(f"Font: {node['fontname']}")
if 'fontsize' in node and node['fontsize'] != 14:
details.append(f"Size: {node['fontsize']}")
if 'tooltip' in node and node['tooltip']:
details.append(f"Tooltip included")
nodes_data["Details"].append(", ".join(details) if details else "-")
nodes_df = pd.DataFrame(nodes_data)
st.write(nodes_df.to_html(escape=False), unsafe_allow_html=True)
# Common function: Display edge list
def display_edges_table(edges_list, is_template=False):
if not edges_list:
st.info("No edges have been added yet.")
return
edges_data = {
"Source": [],
"Target": [],
"Label": [],
"Style": [],
"Color": [],
"Direction": [],
"Arrow": [],
"Details": []
}
for edge in edges_list:
# Convert edge structure if it's a template
if is_template:
source = edge["source"]
target = edge["target"]
label = edge.get("label", "-")
# Get style information
style = "solid"
if "style" in edge["attrs"]:
style = edge["attrs"]["style"]
elif "penwidth" in edge["attrs"] and edge["attrs"]["penwidth"] == "2":
style = "bold"
color = edge["attrs"].get("color", "#666666")
arrow = edge["attrs"].get("arrowhead", "normal")
edges_data["Source"].append(source)
edges_data["Target"].append(target)
edges_data["Label"].append(label)
edges_data["Style"].append(style)
edges_data["Color"].append(f"<span style='color:{color};'>■</span> {color}")
edges_data["Direction"].append("forward")
edges_data["Arrow"].append(arrow)
edges_data["Details"].append("-")
else:
edges_data["Source"].append(edge['source'])
edges_data["Target"].append(edge['target'])
edges_data["Label"].append(edge['label'] if edge['label'] else "-")
style_text = edge['style'].split(" ")[0] if " " in edge['style'] else edge['style']
edges_data["Style"].append(style_text)
edges_data["Color"].append(f"<span style='color:{edge['color']};'>■</span> {edge['color']}")
edges_data["Direction"].append(edge['dir'])
edges_data["Arrow"].append(edge['arrow'])
details = []
if 'fontname' in edge and edge['fontname'] != 'sans-serif':
details.append(f"Font: {edge['fontname']}")
if 'fontsize' in edge and edge['fontsize'] != 10:
details.append(f"Size: {edge['fontsize']}")
if 'penwidth' in edge and edge['penwidth'] != 1.0:
details.append(f"Thickness: {edge['penwidth']}")
if 'weight' in edge and edge['weight'] != 1.0:
details.append(f"Weight: {edge['weight']}")
if 'minlen' in edge and edge['minlen'] != 1:
details.append(f"Min Length: {edge['minlen']}")
if 'tooltip' in edge and edge['tooltip']:
details.append(f"Tooltip included")
edges_data["Details"].append(", ".join(details) if details else "-")
edges_df = pd.DataFrame(edges_data)
st.write(edges_df.to_html(escape=False), unsafe_allow_html=True)
# Initialize session state
if 'active_tab' not in st.session_state:
st.session_state['active_tab'] = "Node and Edge Input"
if 'nodes' not in st.session_state:
st.session_state['nodes'] = []
if 'edges' not in st.session_state:
st.session_state['edges'] = []
if 'template_applied' not in st.session_state:
st.session_state['template_applied'] = False
if 'template_type' not in st.session_state:
st.session_state['template_type'] = ""
if 'template_previewed' not in st.session_state:
st.session_state['template_previewed'] = False
if 'current_template_data' not in st.session_state:
st.session_state['current_template_data'] = None
if 'dot_code' not in st.session_state:
st.session_state['dot_code'] = """digraph G {
rankdir=LR;
node [shape=box];
A [label=\"Start\"];
B [label=\"Process 1\"];
C [label=\"Process 2\"];
D [label=\"End\"];
A -> B [label=\"Flow 1\"];
B -> C [label=\"Flow 2\"];
C -> D [label=\"Flow 3\"];
}"""
if 'ai_prompt' not in st.session_state:
st.session_state['ai_prompt'] = """Please create a system architecture diagram for a web application.
Users access via a browser and reach multiple web servers through a load balancer.
The web servers communicate with application servers, which store data in a database server.
Also, a cache server is being used."""
# Callback function for template or style changes
def reset_template_preview():
"""Resets the preview state when the template or style is changed"""
st.session_state['template_previewed'] = False
# Callback function for tab switching
def on_tab_change():
"""Process when the tab is changed"""
# Reset template preview state
st.session_state['template_previewed'] = False
# State management function for template application
def on_template_apply_click(template_nodes, template_edges, template_node_shape, template_node_color, template_edge_style, template_edge_color, template_type):
"""Process when the apply template button is clicked"""
# Save pending template information to session
st.session_state['current_template_data'] = {
'nodes': template_nodes,
'edges': template_edges,
'node_shape': template_node_shape,
'node_color': template_node_color,
'edge_style': template_edge_style,
'edge_color': template_edge_color,
'type': template_type
}
# Set template applied flag
st.session_state['template_applied'] = True
# Switch active tab
st.session_state['active_tab'] = "Node and Edge Input"
# Function to apply template to node and edge inputs
def apply_template_to_session(template_nodes, template_edges, template_node_shape, template_node_color, template_edge_style, template_edge_color, template_type):
"""Applies the template to the node and edge inputs (Direct session update)"""
# Clear existing nodes and edges
st.session_state['nodes'] = []
st.session_state['edges'] = []
# Add template nodes
for node in template_nodes:
st.session_state['nodes'].append({
'id': node["id"],
'label': node["label"],
'shape': template_node_shape,
'style': "filled",
'fillcolor': template_node_color,
'color': "#000000",
'peripheries': 1,
'fontname': "sans-serif",
'fontsize': 14,
'fontcolor': "#000000",
'tooltip': ""
})
# Add template edges
for edge in template_edges:
edge_style = "solid (Solid Line)"
if template_edge_style == "点線":
edge_style = "dashed (Dashed Line)"
elif template_edge_style == "太線":
edge_style = "bold (Bold Line)"
elif template_edge_style == "矢印":
edge_style = "solid (Solid Line)"
# Get edge attributes
arrow_head = "normal"
if "arrowhead" in edge["attrs"]:
arrow_head = edge["attrs"]["arrowhead"]
st.session_state['edges'].append({
'source': edge["source"],
'target': edge["target"],
'label': edge.get("label", ""),
'style': edge_style,
'color': template_edge_color,
'dir': "forward",
'arrow': arrow_head,
'fontname': "sans-serif",
'fontsize': 10,
'fontcolor': "#000000",
'tooltip': "",
'penwidth': 2.0 if template_edge_style == "太線" else 1.0,
'weight': 1.0,
'constraint': True,
'minlen': 1
})
# Set template information
st.session_state['template_applied'] = True
st.session_state['template_type'] = template_type
# Switch tab
st.session_state['active_tab'] = "Node and Edge Input"
# Sidebar: Basic Settings
st.sidebar.title("Basic Settings")
graph_direction = st.sidebar.radio(
"Graph Direction",
["Left to Right (LR)", "Top to Bottom (TB)"]
)
# Sidebar: AI Settings
st.sidebar.title("Cortex AI Settings")
use_llm = st.sidebar.checkbox("Enable AI Generation", value=False)
if use_llm:
lang_model = st.sidebar.radio("Select Cortex AI Model",
("deepseek-r1",
"claude-4-sonnet", "claude-3-7-sonnet", "claude-3-5-sonnet",
"mistral-large2", "mixtral-8x7b", "mistral-7b",
"llama4-maverick",
"llama3.3-70b",
"llama3.2-1b", "llama3.2-3b",
"llama3.1-8b", "llama3.1-70b", "llama3.1-405b",
"snowflake-llama-3.1-405b", "snowflake-llama-3.3-70b",
"snowflake-arctic",
"reka-flash", "reka-core",
"jamba-instruct", "jamba-1.5-mini", "jamba-1.5-large",
"gemma-7b",
"mistral-large", "llama3-8b", "llama3-70b", "llama2-70b-chat"
),
index=1)
# Main content area
st.header("Graph Creation")
# Input method tabs
tab_options = ["Node and Edge Input", "Templates", "Direct DOT Code Input", "AI Generation"]
selected_tab = st.radio("Select Input Method", tab_options, horizontal=True, key="tab_selector", on_change=on_tab_change)
st.session_state['active_tab'] = selected_tab
# Tab 1: Node and Edge Input
if selected_tab == "Node and Edge Input":
# Check if a template was applied
if st.session_state.get('template_applied'):
st.success(f"'{st.session_state.get('template_type')}' template applied. Nodes: {len(st.session_state['nodes'])}, Edges: {len(st.session_state['edges'])}")
# Reset flag
st.session_state['template_applied'] = False
node_col, edge_col = st.columns(2)
# Node input form
with node_col:
st.subheader("Define Nodes")
with st.form(key="node_form"):
node_id = st.text_input("Node ID", placeholder="e.g., node1, server1")
node_label = st.text_input("Node Label", placeholder="e.g., Server 1, Database")
node_shape = st.selectbox(
"Node Shape",
["box", "ellipse", "circle", "diamond", "plaintext", "polygon", "triangle", "hexagon", "cylinder",
"folder", "component", "note", "tab", "house", "invhouse", "parallelogram", "record", "Mrecord"]
)
node_style = st.selectbox(
"Node Style",
["filled", "dashed", "dotted", "solid", "filled,rounded", "dashed,filled", "dotted,filled", "bold", "invis"],
index=0
)
node_fillcolor = st.color_picker("Node Fill Color", "#D0E8FF")
node_color = st.color_picker("Node Border Color", "#000000")
# Advanced Settings
with st.expander("Advanced Settings"):
peripheries = st.number_input("Number of Peripheries", min_value=1, max_value=10, value=1, help="Number of outlines around the node. 2 or more creates a double border.")
font_col1, font_col2 = st.columns(2)
with font_col1:
fontname = st.selectbox("Font Name",
["sans-serif", "serif", "Arial", "Helvetica", "Times-Roman", "Courier", "MS Gothic", "MS UI Gothic", "Meiryo"],
index=0)
fontcolor = st.color_picker("Font Color", "#000000")
with font_col2:
fontsize = st.number_input("Font Size", min_value=8, max_value=72, value=14)
tooltip = st.text_input("Tooltip", placeholder="Text shown on mouseover")
node_submit = st.form_submit_button("Add Node")
if node_submit and node_id and node_label:
# Check for existing node IDs
existing_ids = [node['id'] for node in st.session_state['nodes']]
if node_id in existing_ids:
st.error(f"Node ID '{node_id}' is already in use. Please choose a different ID.")
else:
style_str = node_style
st.session_state['nodes'].append({
'id': node_id,
'label': node_label,
'shape': node_shape,
'style': style_str,
'fillcolor': node_fillcolor,
'color': node_color,
'peripheries': peripheries,
'fontname': fontname,
'fontsize': fontsize,
'fontcolor': fontcolor,
'tooltip': tooltip
})
st.success(f"Added node '{node_id}'.")
# Edge input form
with edge_col:
st.subheader("Define Edges")
with st.form(key="edge_form"):
dir_option = st.selectbox(
"Edge Direction",
["forward (Directed)", "back (Reverse Directed)", "both (Bidirectional)", "none (Undirected)"],
index=0,
help="'forward' is an arrow from start to end, 'back' is from end to start, 'both' is arrows on both ends, and 'none' has no arrows."
)
# Get list of node IDs
node_ids = [node['id'] for node in st.session_state['nodes']]
if not node_ids:
st.warning("Please add nodes first.")
source_node = ""
target_node = ""
source_select = st.selectbox("Source Node", ["Please add nodes first"])
target_select = st.selectbox("Target Node", ["Please add nodes first"])
else:
source_select = st.selectbox("Source Node", node_ids)
target_select = st.selectbox("Target Node", node_ids)
source_node = source_select
target_node = target_select
edge_label = st.text_input("Edge Label", placeholder="e.g., Data flow, Call")
edge_style = st.selectbox(
"Basic Edge Style",
["solid (Solid Line)", "dashed (Dashed Line)", "dotted (Dotted Line)", "bold (Bold Line)"]
)
edge_color = st.color_picker("Edge Color", "#666666")
# Arrow shape selection (auto-adjusted based on direction)
if dir_option.startswith("none"):
arrow_shape = "none"
else:
arrow_shape = st.selectbox(
"Arrow Shape",
["normal", "vee", "tee", "dot", "diamond", "box", "crow", "inv", "invdot", "odot", "open", "halfopen", "none"]
)
# Advanced Settings
with st.expander("Advanced Settings"):
font_col1, font_col2 = st.columns(2)
with font_col1:
edge_fontname = st.selectbox("Font Name",
["sans-serif", "serif", "Arial", "Helvetica", "Times-Roman", "Courier", "MS Gothic", "MS UI Gothic", "Meiryo"],
index=0,
key="edge_fontname")
edge_fontcolor = st.color_picker("Font Color", "#000000", key="edge_fontcolor")
with font_col2:
edge_fontsize = st.number_input("Font Size", min_value=8, max_value=72, value=10, key="edge_fontsize")
edge_tooltip = st.text_input("Tooltip", placeholder="Text shown on mouseover", key="edge_tooltip")
edge_penwidth = st.slider("Line Thickness", min_value=0.5, max_value=10.0, value=1.0, step=0.5)
edge_weight = st.slider("Weight", min_value=0.1, max_value=10.0, value=1.0, step=0.1,
help="Higher values treat the connection as more important.")
edge_constraint = st.checkbox("Maintain Hierarchical Order", value=True,
help="Check to maintain structural hierarchy (Top to Bottom, Left to Right).")
edge_minlen = st.number_input("Minimum Length", min_value=1, max_value=10, value=1,
help="Specifies the minimum length of the edge.")
edge_submit = st.form_submit_button("Add Edge")
if edge_submit and source_node and target_node:
# Duplicate check
duplicate = False
for edge in st.session_state['edges']:
if edge['source'] == source_node and edge['target'] == target_node:
duplicate = True
break
if duplicate:
st.error(f"Edge '{source_node} -> {target_node}' already exists.")
else:
dir_value = dir_option.split(" ")[0] # Extract "forward" from "forward (Directed)"
st.session_state['edges'].append({
'source': source_node,
'target': target_node,
'label': edge_label,
'style': edge_style,
'color': edge_color,
'dir': dir_value,
'arrow': arrow_shape,
'fontname': edge_fontname,
'fontsize': edge_fontsize,
'fontcolor': edge_fontcolor,
'tooltip': edge_tooltip,
'penwidth': edge_penwidth,
'weight': edge_weight,
'constraint': edge_constraint,
'minlen': edge_minlen
})
st.success(f"Added edge '{source_node} -> {target_node}'.")
# Display list of added nodes and edges
st.subheader("List of Added Nodes and Edges")
nodes_tab, edges_tab = st.tabs(["Node List", "Edge List"])
# Display Node List
with nodes_tab:
if not st.session_state['nodes']:
st.info("No nodes have been added yet.")
else:
display_nodes_table(st.session_state['nodes'])
# Node deletion functionality
with st.form(key="node_delete_form"):
delete_node = st.selectbox("Select Node to Delete", [node['id'] for node in st.session_state['nodes']])
node_delete_submit = st.form_submit_button("Delete Selected Node")
if node_delete_submit:
node_to_delete = delete_node
st.session_state['nodes'] = [node for node in st.session_state['nodes'] if node['id'] != node_to_delete]
# Also delete related edges
st.session_state['edges'] = [
edge for edge in st.session_state['edges']
if edge['source'] != node_to_delete and edge['target'] != node_to_delete
]
st.success(f"Deleted node '{node_to_delete}' and related edges.")
st.rerun()
# Display Edge List
with edges_tab:
if not st.session_state['edges']:
st.info("No edges have been added yet.")
else:
display_edges_table(st.session_state['edges'])
# Edge deletion functionality
with st.form(key="edge_delete_form"):
edge_options = [f"{edge['source']} -> {edge['target']}" for edge in st.session_state['edges']]
delete_edge = st.selectbox("Select Edge to Delete", edge_options)
edge_delete_submit = st.form_submit_button("Delete Selected Edge")
if edge_delete_submit:
source, target = delete_edge.split(" -> ")
st.session_state['edges'] = [
edge for edge in st.session_state['edges']
if not (edge['source'] == source and edge['target'] == target)
]
st.success(f"Deleted edge '{delete_edge}'.")
st.rerun()
# Data clear button
if st.button("Clear All"):
st.session_state['nodes'] = []
st.session_state['edges'] = []
st.success("Cleared all nodes and edges.")
st.rerun()
# Tab 2: Templates
elif selected_tab == "Templates":
st.subheader("Select from Templates")
template_type = st.selectbox(
"Template",
["Simple System Architecture", "Microservices Architecture", "Data Pipeline",
"Network Topology", "Cloud Architecture"],
key="template_select",
on_change=reset_template_preview
)
st.info(f"Selected Template: {template_type}")
# Style customization
st.subheader("Template Style Customization")
st.caption("You can customize the appearance of the template")
template_style_col1, template_style_col2 = st.columns(2)
with template_style_col1:
template_node_shape = st.selectbox(
"Node Shape",
["box", "ellipse", "circle", "diamond", "plaintext", "triangle", "hexagon", "cylinder"],
key="node_shape_select",
on_change=reset_template_preview
)
template_node_color = st.color_picker(
"Node Color",
"#D0E8FF",
key="node_color_picker",
on_change=reset_template_preview
)
with template_style_col2:
template_edge_style = st.selectbox(
"Edge Style",
["Solid", "Dashed", "Bold", "Arrow"],
key="edge_style_select",
on_change=reset_template_preview
)
template_edge_color = st.color_picker(
"Edge Color",
"#666666",
key="edge_color_picker",
on_change=reset_template_preview
)
# Generate template preview
if st.button("Preview Template"):
st.session_state['template_previewed'] = True
# Display preview (only if button is clicked)
if st.session_state['template_previewed']:
try:
# Create temporary graph object
preview_graph = graphviz.Digraph()
# Graph direction setting
if graph_direction == "Left to Right (LR)":
preview_graph.attr(rankdir="LR")
else:
preview_graph.attr(rankdir="TB")
# Common template style configuration
node_attrs = {
"shape": template_node_shape,
"style": "filled",
"fillcolor": template_node_color
}
edge_attrs = {"color": template_edge_color}
if template_edge_style == "Dashed":
edge_attrs["style"] = "dashed"
elif template_edge_style == "Bold":
edge_attrs["penwidth"] = "2"
elif template_edge_style == "Arrow":
edge_attrs["arrowhead"] = "normal"
# Temp template node and edge definitions
template_nodes = []
template_edges = []
# Definitions by template type
if template_type == "Simple System Architecture":
template_nodes = [
{"id": "client", "label": "Client", "attrs": node_attrs},
{"id": "server", "label": "Server", "attrs": node_attrs},
{"id": "db", "label": "Database", "attrs": node_attrs}
]
template_edges = [
{"source": "client", "target": "server", "label": "Request", "attrs": edge_attrs},
{"source": "server", "target": "db", "label": "Query", "attrs": edge_attrs},
{"source": "db", "target": "server", "label": "Result", "attrs": edge_attrs},
{"source": "server", "target": "client", "label": "Response", "attrs": edge_attrs}
]
elif template_type == "Microservices Architecture":
template_nodes = [
{"id": "api", "label": "API Gateway", "attrs": node_attrs},
{"id": "auth", "label": "Auth Service", "attrs": node_attrs},
{"id": "user", "label": "User Service", "attrs": node_attrs},
{"id": "order", "label": "Order Service", "attrs": node_attrs},
{"id": "payment", "label": "Payment Service", "attrs": node_attrs},
{"id": "db_user", "label": "User DB", "attrs": node_attrs},
{"id": "db_order", "label": "Order DB", "attrs": node_attrs}
]
template_edges = [
{"source": "api", "target": "auth", "label": "Authenticate", "attrs": edge_attrs},
{"source": "api", "target": "user", "label": "User Management", "attrs": edge_attrs},
{"source": "api", "target": "order", "label": "Order Processing", "attrs": edge_attrs},
{"source": "order", "target": "payment", "label": "Payment Processing", "attrs": edge_attrs},
{"source": "user", "target": "db_user", "label": "Save Data", "attrs": edge_attrs},
{"source": "order", "target": "db_order", "label": "Save Data", "attrs": edge_attrs}
]
elif template_type == "Data Pipeline":
template_nodes = [
{"id": "source", "label": "Data Source", "attrs": node_attrs},
{"id": "ingest", "label": "Ingestion", "attrs": node_attrs},
{"id": "process", "label": "Processing", "attrs": node_attrs},
{"id": "analyze", "label": "Analysis", "attrs": node_attrs},
{"id": "store", "label": "Storage", "attrs": node_attrs},
{"id": "visual", "label": "Visualization", "attrs": node_attrs}
]
template_edges = [
{"source": "source", "target": "ingest", "label": "Extract", "attrs": edge_attrs},
{"source": "ingest", "target": "process", "label": "Preprocess", "attrs": edge_attrs},
{"source": "process", "target": "analyze", "label": "Analyze", "attrs": edge_attrs},
{"source": "analyze", "target": "store", "label": "Save", "attrs": edge_attrs},
{"source": "store", "target": "visual", "label": "Visualize", "attrs": edge_attrs}
]
elif template_type == "Network Topology":
template_nodes = [
{"id": "router", "label": "Router", "attrs": node_attrs},
{"id": "switch1", "label": "Switch 1", "attrs": node_attrs},
{"id": "switch2", "label": "Switch 2", "attrs": node_attrs},
{"id": "server1", "label": "Server 1", "attrs": node_attrs},
{"id": "server2", "label": "Server 2", "attrs": node_attrs},
{"id": "client1", "label": "Client 1", "attrs": node_attrs},
{"id": "client2", "label": "Client 2", "attrs": node_attrs}
]
template_edges = [
{"source": "router", "target": "switch1", "label": "Connect", "attrs": edge_attrs},
{"source": "router", "target": "switch2", "label": "Connect", "attrs": edge_attrs},
{"source": "switch1", "target": "server1", "label": "Connect", "attrs": edge_attrs},
{"source": "switch1", "target": "server2", "label": "Connect", "attrs": edge_attrs},
{"source": "switch2", "target": "client1", "label": "Connect", "attrs": edge_attrs},
{"source": "switch2", "target": "client2", "label": "Connect", "attrs": edge_attrs}
]
elif template_type == "Cloud Architecture":
template_nodes = [
{"id": "lb", "label": "Load Balancer", "attrs": node_attrs},
{"id": "web1", "label": "Web Server 1", "attrs": node_attrs},
{"id": "web2", "label": "Web Server 2", "attrs": node_attrs},
{"id": "app1", "label": "App Server 1", "attrs": node_attrs},
{"id": "app2", "label": "App Server 2", "attrs": node_attrs},
{"id": "db_master", "label": "DB Master", "attrs": node_attrs},
{"id": "db_slave", "label": "DB Slave", "attrs": node_attrs}
]
template_edges = [
{"source": "lb", "target": "web1", "label": "Forward", "attrs": edge_attrs},
{"source": "lb", "target": "web2", "label": "Forward", "attrs": edge_attrs},
{"source": "web1", "target": "app1", "label": "Request", "attrs": edge_attrs},
{"source": "web2", "target": "app2", "label": "Request", "attrs": edge_attrs},
{"source": "app1", "target": "db_master", "label": "Query", "attrs": edge_attrs},
{"source": "app2", "target": "db_master", "label": "Query", "attrs": edge_attrs},
{"source": "db_master", "target": "db_slave", "label": "Replicate", "attrs": edge_attrs}
]
# Add nodes and edges to graph
for node in template_nodes:
preview_graph.node(node["id"], label=node["label"], **node["attrs"])
for edge in template_edges:
preview_graph.edge(edge["source"], edge["target"], label=edge.get("label", ""), **edge["attrs"])
# Show preview
st.subheader("Template Preview")
st.graphviz_chart(preview_graph)
# Show template info in tables
st.subheader("Template Node and Edge Information")
template_nodes_tab, template_edges_tab = st.tabs(["Template Node List", "Template Edge List"])
with template_nodes_tab:
display_nodes_table(template_nodes, is_template=True)
with template_edges_tab:
display_edges_table(template_edges, is_template=True)
# Show DOT code
st.subheader("Generated DOT Code")
st.code(preview_graph.source, language="dot")
# Option to apply template to custom input
st.subheader("Apply Template to Custom Input")
# Execute template application as batch process via form
with st.form(key="apply_template_form"):
st.write(f"Applies the '{template_type}' template to the Node and Edge inputs.")
apply_submit = st.form_submit_button("Apply Template")
# Generate template data for processing
direction = "LR" if graph_direction == "Left to Right (LR)" else "TB"
if apply_submit:
# Directly apply template to session
apply_template_to_session(template_nodes, template_edges, template_node_shape, template_node_color, template_edge_style, template_edge_color, template_type)
st.success(f"Applied '{template_type}' template to the 'Node and Edge Input' tab.")
except Exception as e:
st.error(f"An error occurred while generating the template preview: {str(e)}")
# Tab 3: Direct DOT Code Input
elif selected_tab == "Direct DOT Code Input":
st.subheader("Direct DOT Code Input")
# Load DOT code from session state and update after input
dot_code = st.text_area(
"DOT Code",
st.session_state['dot_code'],
height=300
)
# Update session state
st.session_state['dot_code'] = dot_code
# Preview button
if st.button("Preview DOT Code"):
try:
# Draw DOT code using Graphviz
graph = graphviz.Source(dot_code)
st.subheader("Preview")
st.graphviz_chart(graph)
except Exception as e:
st.error(f"An error occurred while parsing the DOT code: {str(e)}")
# Tab 4: AI Generation
elif selected_tab == "AI Generation":
st.subheader("AI-Based Graphviz Generation")
if not use_llm:
st.warning("To use AI generation, please check 'Enable AI Generation' in the sidebar.")
dot_code = ""
else:
# AI prompt input (load from session state, update after input)
ai_prompt = st.text_area(
"Enter a description of the diagram you want to create",
st.session_state['ai_prompt'],
height=150
)
# Update session state
st.session_state['ai_prompt'] = ai_prompt
direction_for_ai = "LR" if "Left to Right" in graph_direction else "TB"
# Generate button
if st.button("Generate Graphviz Code with AI"):
with st.spinner("AI is generating Graphviz code..."):
try:
# Create AI prompt
ai_system_prompt = f"""
You are an expert in Graphviz DOT language. Based on the user's request, generate appropriate Graphviz DOT code.
Please follow these conditions when generating the code:
1. Create it in 'digraph' format.
2. Specify the direction using rankdir={direction_for_ai}.
3. Set appropriate shapes for each node (box, circle, ellipse, diamond, cylinder, etc.).
4. Set appropriate styles for edges (solid, dashed, arrows, etc.).
5. Use colors effectively.
6. Use Japanese or English labels based on context (default to English unless user implies Japanese).
7. Aim for a beautifully organized graph.
8. Return only the DOT code. No explanation needed.
User request: {ai_prompt}
DOT code:
"""
# Generate DOT code with AI
dot_code = CompleteText(lang_model, ai_system_prompt)
# Clean up response (remove extra backticks)
dot_code = dot_code.strip()
if dot_code.startswith("```dot"):
dot_code = dot_code[6:]
if dot_code.startswith("```"):
dot_code = dot_code[3:]
if dot_code.endswith("```"):
dot_code = dot_code[:-3]
# Show generated DOT code
st.subheader("Generated DOT Code")
st.code(dot_code, language="dot")
# Show preview
try:
graph = graphviz.Source(dot_code)
st.subheader("Preview")
st.graphviz_chart(graph)
except Exception as e:
st.error(f"An error occurred with the generated DOT code: {str(e)}")
except Exception as e:
st.error(f"An error occurred during AI generation: {str(e)}")
dot_code = ""
# Show current status in sidebar
st.sidebar.write(f"Active Input Method: {st.session_state['active_tab']}")
# Global Graph Generation Button
if selected_tab == "Node and Edge Input" and st.button("Generate Graph"):
try:
active_tab = st.session_state['active_tab']
st.write(f"Graph Generation: Generating in {active_tab} mode")
# Initialize directed graph
graph = graphviz.Digraph()
# Set graph direction
if graph_direction == "Left to Right (LR)":
graph.attr(rankdir="LR")
else:
graph.attr(rankdir="TB")
# Add nodes
for node in st.session_state['nodes']:
node_attrs = {
"label": node['label'],
"shape": node['shape'],
"style": node['style'],
"fillcolor": node['fillcolor'],
"color": node['color'],
"peripheries": str(node['peripheries']),
"fontname": node['fontname'],
"fontsize": str(node['fontsize']),
"fontcolor": node['fontcolor'],
"tooltip": node['tooltip']
}
graph.node(node['id'], **node_attrs)
# Add edges
for edge in st.session_state['edges']:
edge_attrs = {}
if edge['label']:
edge_attrs["label"] = edge['label']
# Style configuration
style = edge['style'].split(" ")[0] # Extract "solid" from "solid (Solid Line)"
if style == "dashed":
edge_attrs["style"] = "dashed"
elif style == "dotted":
edge_attrs["style"] = "dotted"
elif style == "bold":
edge_attrs["penwidth"] = "2"
# Color and arrow configuration
edge_attrs["color"] = edge['color']
if edge['dir'] != "none" and edge['arrow'] != "none":
edge_attrs["arrowhead"] = edge['arrow']
# Direction configuration
if 'dir' in edge and edge['dir'] != 'forward':
edge_attrs["dir"] = edge['dir']
# Font configuration
if 'fontname' in edge:
edge_attrs["fontname"] = edge['fontname']
if 'fontsize' in edge:
edge_attrs["fontsize"] = str(edge['fontsize'])
if 'fontcolor' in edge:
edge_attrs["fontcolor"] = edge['fontcolor']
# Additional settings
if 'tooltip' in edge and edge['tooltip']:
edge_attrs["tooltip"] = edge['tooltip']
if 'penwidth' in edge:
edge_attrs["penwidth"] = str(edge['penwidth'])
if 'weight' in edge:
edge_attrs["weight"] = str(edge['weight'])
if 'constraint' in edge:
edge_attrs["constraint"] = "true" if edge['constraint'] else "false"
if 'minlen' in edge:
edge_attrs["minlen"] = str(edge['minlen'])
graph.edge(edge['source'], edge['target'], **edge_attrs)
# Display graph
st.header("Generated Diagram")
st.graphviz_chart(graph)
# Show DOT code
st.subheader("Generated DOT Code")
st.code(graph.source, language="dot")
# Export instruction
st.info("To export the diagram, copy and save the DOT code shown above.")
except Exception as e:
st.error(f"An error occurred while generating the graph: {str(e)}")
# Help Information
with st.expander("Graphviz and Diagram Creation Help"):
st.markdown("""
### What is Graphviz?
Graphviz is an open-source tool for visualizing graph structures. You can create various types of graphs (directed/undirected).
### Basic Usage
1. Select the graph direction in the sidebar.
2. Choose an input method from 'Node and Edge Input', 'Templates', 'Direct DOT Code Input', or 'AI Generation'.
3. Click 'Generate Graph' to display the diagram.
### Node and Edge Form Input
- Nodes: Fill the form and click 'Add Node' to add it to the list.
- Edges: Fill the form and click 'Add Edge' to add it to the list.
- You can review added nodes and edges in the lists and delete unnecessary ones.
### Commonly Used Node Shapes
- box: Rectangle
- ellipse: Ellipse (default)
- circle: Circle
- diamond: Diamond
- cylinder: Cylinder (database)
- plaintext: Text only
- polygon: Polygon
- record: Record format (multiple fields)
### Commonly Used Node Styles
- filled: Filled background
- dashed: Dashed border
- dotted: Dotted border
- solid: Solid border
- filled,rounded: Rounded corners with fill
- dashed,filled: Dashed border and fill
- dotted,filled: Dotted border and fill
- bold: Thick border
- invis: Invisible (shows only connections)
### Commonly Used Edge Styles
- solid: Solid line (default)
- dashed: Dashed line
- dotted: Dotted line
- bold: Thick line (achieved with penwidth="2")
### Basic DOT Syntax
```
digraph G { // For directed graphs
A -> B; // Edge from A to B
B -> C; // Edge from B to C
}
```
### Examples of Node and Edge Attributes
```
digraph G {
// Node configuration
A [label=\"Start\", shape=box, style=filled, fillcolor=lightblue];
// Edge configuration
A -> B [label=\"Flow\", style=dashed, color=red];
}
```
### How to Use AI Generation
1. Check 'Enable AI Generation' in the sidebar.
2. Select the Cortex AI model to use.
3. Go to the 'AI Generation' tab and enter a description of the diagram.
4. Click 'Generate Graphviz Code with AI' to generate a diagram based on your description.
""")
# Footer
st.caption("Created by Tsubasa Kanno")
Usage Examples
This app provides the following four functions:
- Node and Edge Input: Interactively add nodes and edges from the UI
- Templates: Select and customize existing templates
- Direct DOT Code Input: Directly edit Graphviz syntax
- AI Generation: Automatically generate flowcharts from natural language descriptions
Particularly with the AI generation feature, diagrams can be generated simply by entering a prompt like the following:
Please create a system architecture diagram for a web application.
Users access via a browser and reach multiple web servers through a load balancer.
The web servers communicate with application servers, which store data in a database server.
Also, a cache server is being used.
Graph generated by Cortex AI
Conclusion
By combining the Graphviz library in Streamlit in Snowflake with Cortex AI, I was able to realize an application that allows users to create high-quality flowcharts and architecture diagrams without requiring deep technical knowledge.
While multimodal generative AI has recently made it possible to read and generate diagrams, it is still considered difficult to utilize them effectively without any adjustments at this stage. However, by combining them with tools like Graphviz that generate diagrams based on code, we can expect to absorb the fluctuations in generative AI's diagram reading and generation. I encourage everyone to try out this app and challenge yourselves with new ways to utilize generative AI.
Promotions
Delivered a virtual hands-on at SNOWFLAKE DISCOVER!
I delivered a practical virtual hands-on session titled "Next-Generation VoC (Voice of the Customer) Analysis with Snowflake Cortex AI" at the large-scale webinar for Snowflake engineers, "SNOWFLAKE DISCOVER Vol. 2," held on July 8-9, 2025. You can experience how to analyze customer voices, which is relevant across many industries, using Snowflake's latest features. If you are looking for hints on analyzing unstructured data, please check it out!
You can watch it on-demand immediately by registering through the link below.
Spoke at SNOWFLAKE DISCOVER!
I spoke in the very first session, "Starting from Zero with Snowflake: Building a Modern Data & AI Platform," at the large-scale webinar for Snowflake engineers, "SNOWFLAKE DISCOVER," held on April 24-25, 2025. I explained everything from the overview of Snowflake to its latest status as clearly as possible, so I'd be happy if you could use it to catch up!
You can watch it on-demand immediately by registering through the link below.
Spoke at the Generative AI Conf Webinar!
I gave a lightning talk (LT) on the theme of Data * AI as a Snowflake employee at a webinar titled "Platform Supporting the Generative AI Era," alongside speakers from NVIDIA and my former employer, AWS! Here is the video archive, so please take a look!
Sharing Snowflake What's New on X
I'm sharing update information for Snowflake's What's New on X, so please feel free to follow me.
日本語版
Snowflake What's New Bot (日本語版)
English Version
Snowflake What's New Bot (English Version)
Change Log
(2025/03/20) New post
(2025/04/29) Added link to the sequel article "Creating a Hand-drawn Flowchart Polishing App with Streamlit in Snowflake (SiS)"
(2025/05/08) Corrected promotion section
(2025/05/10) Added Cortex LLM model claude-3-7-sonnet to the list
(2025/05/24) Added Cortex LLM models claude-4-sonnet and llama4-maverick to the list
(2025/06/29) Corrected promotion section
(2025/09/27) Corrected promotion section
Discussion