❄️
Streamlit in Snowflake (SiS) と Cortex AI で実現するフロー図自動作成アプリ
はじめに
今回は Streamlit in Snowflake と Cortex AI を組み合わせて、フロー図やアーキテクチャ図を自動生成できるアプリを作成してみましたのでご紹介します。まだまだ改善したい点があるため、後日 Part2 の記事を書く可能性が高いです！
Streamlit in Snowflake (SiS) とは？
まず Stremlit は Python のライブラリです。Streamlit を使うとシンプルな Python の記述だけで Web UI を作成することができます。本来 Web UI は HTML / CSS / JavaScript などを用いて記述する必要があるのですが、Streamlit がそれらの機能を代替してくれるイメージとなります。
こちらの App Gallery にユーザーが投稿したサンプルがあるためイメージしやすいです。
Streamlit in Snowflake は Snowflake 上で Streamlit の Web アプリを開発したり実行することが可能です。Snowflake のアカウントさえあれば触れる手軽さも魅力ですが、何より Snowflake 上のテーブルデータなどを簡単に Web アプリに組み込める点が良いところだと思います。
Streamlit in Snowflake について (Snowflake 公式ドキュメント)
Snowflake Cortex とは？
Snowflake Cortex とは Snowflake の生成 AI の機能群です。その中の Cortex LLM は Snowflake 上で稼働する大規模言語モデルを呼び出せる機能であり、SQL や Python などからシンプルな関数で色んな種類の LLM を利用することが可能となります。
Large Language Model (LLM) Functions (Snowflake Cortex) (Snowflake 公式ドキュメント)
機能概要
機能一覧
今回作成したアプリでは以下の機能を実装しています。
- フロー図やアーキテクチャ図を作成するためのグラフィカルエディタ
- ノードとエッジを対話的に追加・編集できる機能
- カスタマイズ可能な形状やスタイル設定
- 設定した項目からグラフの生成
- テンプレート機能
- 数種類のテンプレートからのグラフ生成
- テンプレート内容をグラフィカルエディタに反映する機能
- DOTコード直接編集機能
- Graphvizの構文を DOT コード形式で直接入力
- 入力した DOT コードからグラフの生成
- AI生成機能
- 自然言語の説明からフロー図を自動生成
完成イメージ
アプリ全体図
ノードとエッジのグラフィカルエディタ
ノードとエッジの一覧表示
グラフィカルエディタから生成したグラフと DOT コード
テンプレート機能
テンプレートのノードとエッジの一覧表示と DOT コード
DOT コードの直接入力機能
Cortex AI によるグラフの自動生成機能
Cortex AI により生成されたグラフ
前提条件
- Snowflake アカウント
- Cortex LLM が利用できる Snowflake アカウント (クロスリージョン推論のリリースによりクラウドやリージョンの制約がほぼ無くなりました)
- Streamlit in Snowflake のインストールパッケージ
- python 3.11 以降
- snowflake-ml-python 1.7.4 以降
- python-graphviz 0.20.1 以降
※Cortex LLM のリージョン対応表 (Snowflake 公式ドキュメント)
Graphviz とは？
Graphviz は、グラフ構造を視覚化するためのオープンソースライブラリです。ノードとエッジで構成される有向・無向グラフを描画するのに適しており、ネットワーク図、フローチャート、組織図、ER 図など様々な図表の作成に利用できます。
Graphviz には以下のような特徴があります：
- DOT コードという専用の記述言語を使用
- 自動レイアウト機能 (ノードの配置を自動で最適化)
- 多様なノード形状とエッジスタイルをサポート
今回は Streamlit in Snowflake では python-graphviz というライブラリが簡単にインストールできますので、Graphviz を用いたフロー図作成アプリを作っていこうと思います。
手順
新規で Streamlit in Snowflake のアプリを作成
Snowsight の左ペインから『Streamlit』をクリックし、『+ Streamlit』ボタンをクリックし SiS アプリを作成します。
Streamlit in Snowflake のアプリを実行
Streamlit in Snowflake アプリの編集画面で以下コードをコピー＆ペーストで貼り付けて完了です。(コードが長くなってしまったためアコーディオンを開いて表示してください)
ソースコード
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
# アプリケーション設定
st.set_page_config(
layout="wide",
initial_sidebar_state="expanded"
)
# Snowflakeセッションの取得
session = get_active_session()
# アプリケーションヘッダー
st.title("Streamlit in Snowflake フロー図作成ツール")
st.markdown("Cortex AIとGraphvizを活用した、アーキテクチャ図やデータフロー図作成ツールです。")
# 共通関数：ノード一覧を表示する
def display_nodes_table(nodes_list, is_template=False):
if not nodes_list:
st.info("ノードがまだ追加されていません。")
return
nodes_data = {
"ノードID": [],
"ラベル": [],
"形状": [],
"スタイル": [],
"塗りつぶし色": [],
"枠線色": [],
"詳細設定": []
}
for node in nodes_list:
# テンプレートの場合はノードの構造が異なるので変換
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["ノードID"].append(node_id)
nodes_data["ラベル"].append(node_label)
nodes_data["形状"].append(node_shape)
nodes_data["スタイル"].append(node_style)
nodes_data["塗りつぶし色"].append(f"<span style='color:{node_fillcolor};'>■</span> {node_fillcolor}")
nodes_data["枠線色"].append(f"<span style='color:#000000;'>■</span> #000000")
nodes_data["詳細設定"].append("-")
else:
nodes_data["ノードID"].append(node['id'])
nodes_data["ラベル"].append(node['label'])
nodes_data["形状"].append(node['shape'])
nodes_data["スタイル"].append(node['style'])
nodes_data["塗りつぶし色"].append(f"<span style='color:{node['fillcolor']};'>■</span> {node['fillcolor']}")
nodes_data["枠線色"].append(f"<span style='color:{node['color']};'>■</span> {node['color']}")
# 詳細設定の情報をまとめる
details = []
if 'peripheries' in node and node['peripheries'] > 1:
details.append(f"輪郭線: {node['peripheries']}")
if 'fontname' in node and node['fontname'] != 'sans-serif':
details.append(f"フォント: {node['fontname']}")
if 'fontsize' in node and node['fontsize'] != 14:
details.append(f"サイズ: {node['fontsize']}")
if 'tooltip' in node and node['tooltip']:
details.append(f"ツールチップあり")
nodes_data["詳細設定"].append(", ".join(details) if details else "-")
nodes_df = pd.DataFrame(nodes_data)
st.write(nodes_df.to_html(escape=False), unsafe_allow_html=True)
# 共通関数：エッジ一覧を表示する
def display_edges_table(edges_list, is_template=False):
if not edges_list:
st.info("エッジがまだ追加されていません。")
return
edges_data = {
"始点": [],
"終点": [],
"ラベル": [],
"スタイル": [],
"色": [],
"方向": [],
"矢印": [],
"詳細設定": []
}
for edge in edges_list:
# テンプレートの場合はエッジの構造が異なるので変換
if is_template:
source = edge["source"]
target = edge["target"]
label = edge.get("label", "-")
# スタイル情報の取得
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["始点"].append(source)
edges_data["終点"].append(target)
edges_data["ラベル"].append(label)
edges_data["スタイル"].append(style)
edges_data["色"].append(f"<span style='color:{color};'>■</span> {color}")
edges_data["方向"].append("forward")
edges_data["矢印"].append(arrow)
edges_data["詳細設定"].append("-")
else:
edges_data["始点"].append(edge['source'])
edges_data["終点"].append(edge['target'])
edges_data["ラベル"].append(edge['label'] if edge['label'] else "-")
style_text = edge['style'].split(" ")[0] if " " in edge['style'] else edge['style']
edges_data["スタイル"].append(style_text)
edges_data["色"].append(f"<span style='color:{edge['color']};'>■</span> {edge['color']}")
edges_data["方向"].append(edge['dir'])
edges_data["矢印"].append(edge['arrow'])
details = []
if 'fontname' in edge and edge['fontname'] != 'sans-serif':
details.append(f"フォント: {edge['fontname']}")
if 'fontsize' in edge and edge['fontsize'] != 10:
details.append(f"サイズ: {edge['fontsize']}")
if 'penwidth' in edge and edge['penwidth'] != 1.0:
details.append(f"太さ: {edge['penwidth']}")
if 'weight' in edge and edge['weight'] != 1.0:
details.append(f"重み: {edge['weight']}")
if 'minlen' in edge and edge['minlen'] != 1:
details.append(f"最小長: {edge['minlen']}")
if 'tooltip' in edge and edge['tooltip']:
details.append(f"ツールチップあり")
edges_data["詳細設定"].append(", ".join(details) if details else "-")
edges_df = pd.DataFrame(edges_data)
st.write(edges_df.to_html(escape=False), unsafe_allow_html=True)
# セッション状態の初期化
if 'active_tab' not in st.session_state:
st.session_state['active_tab'] = "ノードとエッジの入力"
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="開始"];
B [label="処理1"];
C [label="処理2"];
D [label="終了"];
A -> B [label="フロー1"];
B -> C [label="フロー2"];
C -> D [label="フロー3"];
}"""
if 'ai_prompt' not in st.session_state:
st.session_state['ai_prompt'] = """Webアプリケーションのシステム構成図を作成してください。
ユーザーはブラウザからアクセスし、ロードバランサーを経由して複数のWebサーバーにアクセスします。
Webサーバーはアプリケーションサーバーと通信し、アプリケーションサーバーはデータベースサーバーにデータを格納します。
また、キャッシュサーバーも使用しています。"""
# テンプレートやスタイル変更時のコールバック関数
def reset_template_preview():
"""テンプレートやスタイルが変更されたときにプレビュー状態をリセットする"""
st.session_state['template_previewed'] = False
# タブ切り替え時のコールバック関数
def on_tab_change():
"""タブが変更されたときの処理"""
# テンプレートプレビュー状態をリセット
st.session_state['template_previewed'] = False
# テンプレート適用の状態管理関数
def on_template_apply_click(template_nodes, template_edges, template_node_shape, template_node_color, template_edge_style, template_edge_color, template_type):
"""テンプレート適用ボタンがクリックされたときの処理"""
# 保留中のテンプレート情報をセッションに保存
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
}
# テンプレート適用フラグを設定
st.session_state['template_applied'] = True
# アクティブタブを切り替え
st.session_state['active_tab'] = "ノードとエッジの入力"
# テンプレートをノードとエッジの入力に適用する関数
def apply_template_to_session(template_nodes, template_edges, template_node_shape, template_node_color, template_edge_style, template_edge_color, template_type):
"""テンプレートをノードとエッジの入力に適用する（直接セッション更新）"""
# 既存のノードとエッジをクリア
st.session_state['nodes'] = []
st.session_state['edges'] = []
# テンプレートのノードを追加
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': ""
})
# テンプレートのエッジを追加
for edge in template_edges:
edge_style = "solid (実線)"
if template_edge_style == "点線":
edge_style = "dashed (点線)"
elif template_edge_style == "太線":
edge_style = "bold (太線)"
elif template_edge_style == "矢印":
edge_style = "solid (実線)"
# エッジ属性を取得
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
})
# テンプレート情報を設定
st.session_state['template_applied'] = True
st.session_state['template_type'] = template_type
# タブの切り替え
st.session_state['active_tab'] = "ノードとエッジの入力"
# サイドバー：基本設定
st.sidebar.title("基本設定")
graph_direction = st.sidebar.radio(
"グラフの方向",
["左から右 (LR)", "上から下 (TB)"]
)
# サイドバー：AI設定
st.sidebar.title("Cortex AIの設定")
use_llm = st.sidebar.checkbox("AI生成機能を有効化", value=False)
if use_llm:
lang_model = st.sidebar.radio("Cortex AIのモデルを選択してください",
("deepseek-r1",
"claude-3-5-sonnet",
"mistral-large2", "mixtral-8x7b", "mistral-7b",
"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)
# メインコンテンツエリア
st.header("グラフ作成")
# 入力方法タブ
tab_options = ["ノードとエッジの入力", "テンプレート", "DOTコード直接入力", "AI生成"]
selected_tab = st.radio("入力方法を選択", tab_options, horizontal=True, key="tab_selector", on_change=on_tab_change)
st.session_state['active_tab'] = selected_tab
# タブ1: ノードとエッジの入力
if selected_tab == "ノードとエッジの入力":
# テンプレートが適用されたかチェック
if st.session_state.get('template_applied'):
st.success(f"「{st.session_state.get('template_type')}」テンプレートが適用されました。ノード数: {len(st.session_state['nodes'])}、エッジ数: {len(st.session_state['edges'])}")
# フラグをリセット
st.session_state['template_applied'] = False
node_col, edge_col = st.columns(2)
# ノード入力フォーム
with node_col:
st.subheader("ノードの定義")
with st.form(key="node_form"):
node_id = st.text_input("ノードID", placeholder="例: node1, server1")
node_label = st.text_input("ノードのラベル", placeholder="例: サーバー1, データベース")
node_shape = st.selectbox(
"ノードの形状",
["box", "ellipse", "circle", "diamond", "plaintext", "polygon", "triangle", "hexagon", "cylinder",
"folder", "component", "note", "tab", "house", "invhouse", "parallelogram", "record", "Mrecord"]
)
node_style = st.selectbox(
"ノードのスタイル",
["filled", "dashed", "dotted", "solid", "filled,rounded", "dashed,filled", "dotted,filled", "bold", "invis"],
index=0
)
node_fillcolor = st.color_picker("ノードの塗りつぶし色", "#D0E8FF")
node_color = st.color_picker("ノードの枠線色", "#000000")
# 詳細設定
with st.expander("詳細設定"):
peripheries = st.number_input("輪郭線の数", min_value=1, max_value=10, value=1, help="ノードの周りの輪郭線の数。2以上で二重線になります")
font_col1, font_col2 = st.columns(2)
with font_col1:
fontname = st.selectbox("フォント名",
["sans-serif", "serif", "Arial", "Helvetica", "Times-Roman", "Courier", "MS Gothic", "MS UI Gothic", "Meiryo"],
index=0)
fontcolor = st.color_picker("フォント色", "#000000")
with font_col2:
fontsize = st.number_input("フォントサイズ", min_value=8, max_value=72, value=14)
tooltip = st.text_input("ツールチップ", placeholder="マウスオーバー時に表示されるテキスト")
node_submit = st.form_submit_button("ノードを追加")
if node_submit and node_id and node_label:
# 既存のノードIDをチェック
existing_ids = [node['id'] for node in st.session_state['nodes']]
if node_id in existing_ids:
st.error(f"ノードID '{node_id}' は既に使用されています。別の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"ノード '{node_id}' を追加しました。")
# エッジ入力フォーム
with edge_col:
st.subheader("エッジの定義")
with st.form(key="edge_form"):
dir_option = st.selectbox(
"エッジの方向",
["forward (順方向/有向)", "back (逆方向/有向)", "both (双方向/有向)", "none (無向)"],
index=0,
help="forwardは始点から終点への矢印、backは終点から始点への矢印、bothは双方向の矢印、noneは矢印なし（無向）"
)
# ノードIDのリスト取得
node_ids = [node['id'] for node in st.session_state['nodes']]
if not node_ids:
st.warning("先にノードを追加してください。")
source_node = ""
target_node = ""
source_select = st.selectbox("始点ノード", ["先にノードを追加してください"])
target_select = st.selectbox("終点ノード", ["先にノードを追加してください"])
else:
source_select = st.selectbox("始点ノード", node_ids)
target_select = st.selectbox("終点ノード", node_ids)
source_node = source_select
target_node = target_select
edge_label = st.text_input("エッジのラベル", placeholder="例: データフロー, 呼び出し")
edge_style = st.selectbox(
"エッジの基本スタイル",
["solid (実線)", "dashed (点線)", "dotted (破線)", "bold (太線)"]
)
edge_color = st.color_picker("エッジの色", "#666666")
# 矢印形状の選択（方向によって自動調整）
if dir_option.startswith("none"):
arrow_shape = "none"
else:
arrow_shape = st.selectbox(
"矢印の形状",
["normal", "vee", "tee", "dot", "diamond", "box", "crow", "inv", "invdot", "odot", "open", "halfopen", "none"]
)
# 詳細設定
with st.expander("詳細設定"):
font_col1, font_col2 = st.columns(2)
with font_col1:
edge_fontname = st.selectbox("フォント名",
["sans-serif", "serif", "Arial", "Helvetica", "Times-Roman", "Courier", "MS Gothic", "MS UI Gothic", "Meiryo"],
index=0,
key="edge_fontname")
edge_fontcolor = st.color_picker("フォント色", "#000000", key="edge_fontcolor")
with font_col2:
edge_fontsize = st.number_input("フォントサイズ", min_value=8, max_value=72, value=10, key="edge_fontsize")
edge_tooltip = st.text_input("ツールチップ", placeholder="マウスオーバー時に表示されるテキスト", key="edge_tooltip")
edge_penwidth = st.slider("線の太さ", min_value=0.5, max_value=10.0, value=1.0, step=0.5)
edge_weight = st.slider("重み", min_value=0.1, max_value=10.0, value=1.0, step=0.1,
help="値が大きいほど重要な接続として扱われます")
edge_constraint = st.checkbox("階層順序を維持", value=True,
help="チェックすると階層構造（上から下、左から右）が維持されます")
edge_minlen = st.number_input("最小長さ", min_value=1, max_value=10, value=1,
help="エッジの最小長さを指定します")
edge_submit = st.form_submit_button("エッジを追加")
if edge_submit and source_node and target_node:
# 重複チェック
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"エッジ '{source_node} -> {target_node}' は既に存在します。")
else:
dir_value = dir_option.split(" ")[0] # "forward (順方向/有向)" から "forward" を抽出
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"エッジ '{source_node} -> {target_node}' を追加しました。")
# 追加したノードとエッジの一覧表示
st.subheader("追加したノードとエッジの一覧")
nodes_tab, edges_tab = st.tabs(["ノード一覧", "エッジ一覧"])
# ノード一覧表示
with nodes_tab:
if not st.session_state['nodes']:
st.info("ノードがまだ追加されていません。")
else:
display_nodes_table(st.session_state['nodes'])
# ノード削除機能
with st.form(key="node_delete_form"):
delete_node = st.selectbox("削除するノード", [node['id'] for node in st.session_state['nodes']])
node_delete_submit = st.form_submit_button("選択したノードを削除")
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]
# 関連エッジも削除
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"ノード '{node_to_delete}' と関連するエッジを削除しました。")
st.rerun()
# エッジ一覧表示
with edges_tab:
if not st.session_state['edges']:
st.info("エッジがまだ追加されていません。")
else:
display_edges_table(st.session_state['edges'])
# エッジ削除機能
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("削除するエッジ", edge_options)
edge_delete_submit = st.form_submit_button("選択したエッジを削除")
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"エッジ '{delete_edge}' を削除しました。")
st.rerun()
# データクリアボタン
if st.button("すべてクリア"):
st.session_state['nodes'] = []
st.session_state['edges'] = []
st.success("すべてのノードとエッジをクリアしました。")
st.rerun()
# タブ2: テンプレート
elif selected_tab == "テンプレート":
st.subheader("テンプレートから選択")
template_type = st.selectbox(
"テンプレート",
["シンプルなシステム構成", "マイクロサービスアーキテクチャ", "データパイプライン",
"ネットワーク構成", "クラウドアーキテクチャ"],
key="template_select",
on_change=reset_template_preview
)
st.info(f"選択されたテンプレート: {template_type}")
# スタイルカスタマイズ
st.subheader("テンプレートのスタイルカスタマイズ")
st.caption("テンプレートの見た目をカスタマイズできます")
template_style_col1, template_style_col2 = st.columns(2)
with template_style_col1:
template_node_shape = st.selectbox(
"ノードの形状",
["box", "ellipse", "circle", "diamond", "plaintext", "triangle", "hexagon", "cylinder"],
key="node_shape_select",
on_change=reset_template_preview
)
template_node_color = st.color_picker(
"ノードの色",
"#D0E8FF",
key="node_color_picker",
on_change=reset_template_preview
)
with template_style_col2:
template_edge_style = st.selectbox(
"エッジのスタイル",
["実線", "点線", "太線", "矢印"],
key="edge_style_select",
on_change=reset_template_preview
)
template_edge_color = st.color_picker(
"エッジの色",
"#666666",
key="edge_color_picker",
on_change=reset_template_preview
)
# テンプレートのプレビュー生成
if st.button("テンプレートをプレビュー"):
st.session_state['template_previewed'] = True
# プレビュー表示（ボタンが押された場合のみ表示）
if st.session_state['template_previewed']:
try:
# 一時的なグラフオブジェクトを作成
preview_graph = graphviz.Digraph()
# グラフの方向設定
if graph_direction == "左から右 (LR)":
preview_graph.attr(rankdir="LR")
else:
preview_graph.attr(rankdir="TB")
# テンプレート共通スタイル設定
node_attrs = {
"shape": template_node_shape,
"style": "filled",
"fillcolor": template_node_color
}
edge_attrs = {"color": template_edge_color}
if template_edge_style == "点線":
edge_attrs["style"] = "dashed"
elif template_edge_style == "太線":
edge_attrs["penwidth"] = "2"
elif template_edge_style == "矢印":
edge_attrs["arrowhead"] = "normal"
# 一時的なテンプレートノードとエッジの設定
template_nodes = []
template_edges = []
# テンプレート別のノード・エッジ定義
if template_type == "シンプルなシステム構成":
template_nodes = [
{"id": "client", "label": "クライアント", "attrs": node_attrs},
{"id": "server", "label": "サーバー", "attrs": node_attrs},
{"id": "db", "label": "データベース", "attrs": node_attrs}
]
template_edges = [
{"source": "client", "target": "server", "label": "リクエスト", "attrs": edge_attrs},
{"source": "server", "target": "db", "label": "クエリ", "attrs": edge_attrs},
{"source": "db", "target": "server", "label": "結果", "attrs": edge_attrs},
{"source": "server", "target": "client", "label": "レスポンス", "attrs": edge_attrs}
]
elif template_type == "マイクロサービスアーキテクチャ":
template_nodes = [
{"id": "api", "label": "APIゲートウェイ", "attrs": node_attrs},
{"id": "auth", "label": "認証サービス", "attrs": node_attrs},
{"id": "user", "label": "ユーザーサービス", "attrs": node_attrs},
{"id": "order", "label": "注文サービス", "attrs": node_attrs},
{"id": "payment", "label": "決済サービス", "attrs": node_attrs},
{"id": "db_user", "label": "ユーザーDB", "attrs": node_attrs},
{"id": "db_order", "label": "注文DB", "attrs": node_attrs}
]
template_edges = [
{"source": "api", "target": "auth", "label": "認証", "attrs": edge_attrs},
{"source": "api", "target": "user", "label": "ユーザー管理", "attrs": edge_attrs},
{"source": "api", "target": "order", "label": "注文処理", "attrs": edge_attrs},
{"source": "order", "target": "payment", "label": "決済処理", "attrs": edge_attrs},
{"source": "user", "target": "db_user", "label": "データ保存", "attrs": edge_attrs},
{"source": "order", "target": "db_order", "label": "データ保存", "attrs": edge_attrs}
]
elif template_type == "データパイプライン":
template_nodes = [
{"id": "source", "label": "データソース", "attrs": node_attrs},
{"id": "ingest", "label": "取り込み", "attrs": node_attrs},
{"id": "process", "label": "処理", "attrs": node_attrs},
{"id": "analyze", "label": "分析", "attrs": node_attrs},
{"id": "store", "label": "保存", "attrs": node_attrs},
{"id": "visual", "label": "可視化", "attrs": node_attrs}
]
template_edges = [
{"source": "source", "target": "ingest", "label": "抽出", "attrs": edge_attrs},
{"source": "ingest", "target": "process", "label": "前処理", "attrs": edge_attrs},
{"source": "process", "target": "analyze", "label": "分析処理", "attrs": edge_attrs},
{"source": "analyze", "target": "store", "label": "保存", "attrs": edge_attrs},
{"source": "store", "target": "visual", "label": "可視化", "attrs": edge_attrs}
]
elif template_type == "ネットワーク構成":
template_nodes = [
{"id": "router", "label": "ルーター", "attrs": node_attrs},
{"id": "switch1", "label": "スイッチ1", "attrs": node_attrs},
{"id": "switch2", "label": "スイッチ2", "attrs": node_attrs},
{"id": "server1", "label": "サーバー1", "attrs": node_attrs},
{"id": "server2", "label": "サーバー2", "attrs": node_attrs},
{"id": "client1", "label": "クライアント1", "attrs": node_attrs},
{"id": "client2", "label": "クライアント2", "attrs": node_attrs}
]
template_edges = [
{"source": "router", "target": "switch1", "label": "接続", "attrs": edge_attrs},
{"source": "router", "target": "switch2", "label": "接続", "attrs": edge_attrs},
{"source": "switch1", "target": "server1", "label": "接続", "attrs": edge_attrs},
{"source": "switch1", "target": "server2", "label": "接続", "attrs": edge_attrs},
{"source": "switch2", "target": "client1", "label": "接続", "attrs": edge_attrs},
{"source": "switch2", "target": "client2", "label": "接続", "attrs": edge_attrs}
]
elif template_type == "クラウドアーキテクチャ":
template_nodes = [
{"id": "lb", "label": "ロードバランサー", "attrs": node_attrs},
{"id": "web1", "label": "Webサーバー1", "attrs": node_attrs},
{"id": "web2", "label": "Webサーバー2", "attrs": node_attrs},
{"id": "app1", "label": "アプリサーバー1", "attrs": node_attrs},
{"id": "app2", "label": "アプリサーバー2", "attrs": node_attrs},
{"id": "db_master", "label": "DBマスター", "attrs": node_attrs},
{"id": "db_slave", "label": "DBスレーブ", "attrs": node_attrs}
]
template_edges = [
{"source": "lb", "target": "web1", "label": "転送", "attrs": edge_attrs},
{"source": "lb", "target": "web2", "label": "転送", "attrs": edge_attrs},
{"source": "web1", "target": "app1", "label": "リクエスト", "attrs": edge_attrs},
{"source": "web2", "target": "app2", "label": "リクエスト", "attrs": edge_attrs},
{"source": "app1", "target": "db_master", "label": "クエリ", "attrs": edge_attrs},
{"source": "app2", "target": "db_master", "label": "クエリ", "attrs": edge_attrs},
{"source": "db_master", "target": "db_slave", "label": "レプリケーション", "attrs": edge_attrs}
]
# グラフにノードとエッジを追加
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"])
# プレビュー表示
st.subheader("テンプレートのプレビュー")
st.graphviz_chart(preview_graph)
# テンプレートのノードとエッジの情報を表形式で表示
st.subheader("テンプレートのノードとエッジ情報")
template_nodes_tab, template_edges_tab = st.tabs(["テンプレートのノード一覧", "テンプレートのエッジ一覧"])
with template_nodes_tab:
display_nodes_table(template_nodes, is_template=True)
with template_edges_tab:
display_edges_table(template_edges, is_template=True)
# DOTコードの表示
st.subheader("生成されたDOTコード")
st.code(preview_graph.source, language="dot")
# テンプレートをカスタム入力に適用するオプション
st.subheader("テンプレートをカスタム入力に適用")
# フォームでテンプレート適用をバッチ処理として実行
with st.form(key="apply_template_form"):
st.write(f"{template_type}テンプレートをノードとエッジの入力に適用します。")
apply_submit = st.form_submit_button("テンプレートを適用する")
# 作業用のテンプレートデータを生成
# グラフの方向設定
direction = "LR" if graph_direction == "左から右 (LR)" else "TB"
if apply_submit:
# テンプレートを直接セッションに適用
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"{template_type}テンプレートを「ノードとエッジの入力」タブに適用しました。")
except Exception as e:
st.error(f"テンプレートのプレビュー生成中にエラーが発生しました: {str(e)}")
# タブ3: DOTコード直接入力
elif selected_tab == "DOTコード直接入力":
st.subheader("DOTコード直接入力")
# DOTコードをセッション状態から読み込み、入力後に更新
dot_code = st.text_area(
"DOTコード",
st.session_state['dot_code'],
height=300
)
# セッション状態を更新
st.session_state['dot_code'] = dot_code
# プレビューボタン
if st.button("DOTコードをプレビュー"):
try:
# DOTコードをGraphvizで描画
graph = graphviz.Source(dot_code)
st.subheader("プレビュー")
st.graphviz_chart(graph)
except Exception as e:
st.error(f"DOTコードの解析中にエラーが発生しました: {str(e)}")
# タブ4: AI生成
elif selected_tab == "AI生成":
st.subheader("AIによるGraphviz生成")
if not use_llm:
st.warning("AIによるGraphviz生成を使用するには、サイドバーで「AI生成機能を有効化」にチェックを入れてください。")
dot_code = ""
else:
# AIプロンプト入力（セッション状態から読み込み、入力後に更新）
ai_prompt = st.text_area(
"作成したい図表の説明を入力してください",
st.session_state['ai_prompt'],
height=150
)
# セッション状態を更新
st.session_state['ai_prompt'] = ai_prompt
direction_for_ai = "LR" if "左から右" in graph_direction else "TB"
# 生成ボタン
if st.button("AIでGraphvizコードを生成"):
with st.spinner("AIがGraphvizコードを生成しています..."):
try:
# AIプロンプト作成
ai_system_prompt = f"""
あなたはGraphvizのDOT言語の専門家です。ユーザーの要求に基づいて、適切なGraphvizのDOTコードを生成してください。
以下の条件に従ってDOTコードを生成してください：
1. digraph形式で作成すること
2. rankdir={direction_for_ai}で方向を指定すること
3. 各ノードに適切な形状（box, circle, ellipse, diamond, cylinder など）を設定すること
4. エッジに適切なスタイル（実線、点線、矢印など）を設定すること
5. 色を効果的に使用すること
6. 日本語のラベルを使用すること
7. 美しく整理されたグラフになるよう工夫すること
8. DOTコードだけを返してください。説明などは不要です。
ユーザーの要求: {ai_prompt}
DOTコード:
"""
# AIによるDOTコード生成
dot_code = CompleteText(lang_model, ai_system_prompt)
# 返答の整形（余分なバッククォートを削除）
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]
# 生成されたDOTコードを表示
st.subheader("生成されたDOTコード")
st.code(dot_code, language="dot")
# プレビュー表示
try:
graph = graphviz.Source(dot_code)
st.subheader("プレビュー")
st.graphviz_chart(graph)
except Exception as e:
st.error(f"生成されたDOTコードでエラーが発生しました: {str(e)}")
except Exception as e:
st.error(f"AIによる生成中にエラーが発生しました: {str(e)}")
dot_code = ""
# サイドバーに現在の状態を表示
st.sidebar.write(f"選択中の入力方法: {st.session_state['active_tab']}")
# グラフ生成ボタン
if selected_tab == "ノードとエッジの入力" and st.button("グラフを生成"):
try:
active_tab = st.session_state['active_tab']
st.write(f"グラフ生成: {active_tab}モードで生成します")
# 有向グラフの初期化
graph = graphviz.Digraph()
# グラフの方向設定
if graph_direction == "左から右 (LR)":
graph.attr(rankdir="LR")
else:
graph.attr(rankdir="TB")
# ノード追加
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)
# エッジ追加
for edge in st.session_state['edges']:
edge_attrs = {}
if edge['label']:
edge_attrs["label"] = edge['label']
# スタイル設定
style = edge['style'].split(" ")[0] # "solid (実線)" から "solid" を抽出
if style == "dashed":
edge_attrs["style"] = "dashed"
elif style == "dotted":
edge_attrs["style"] = "dotted"
elif style == "bold":
edge_attrs["penwidth"] = "2"
# 色と矢印設定
edge_attrs["color"] = edge['color']
if edge['dir'] != "none" and edge['arrow'] != "none":
edge_attrs["arrowhead"] = edge['arrow']
# 方向設定
if 'dir' in edge and edge['dir'] != 'forward':
edge_attrs["dir"] = edge['dir']
# フォント設定
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']
# 追加設定
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)
# グラフ表示
st.header("生成された図表")
st.graphviz_chart(graph)
# DOTコードの表示
st.subheader("生成されたDOTコード")
st.code(graph.source, language="dot")
# エクスポート案内
st.info("図表をエクスポートするには、上記のDOTコードをコピーして保存してください。")
except Exception as e:
st.error(f"グラフの生成中にエラーが発生しました: {str(e)}")
# ヘルプ情報
with st.expander("Graphvizと図表作成のヘルプ"):
st.markdown("""
### Graphvizとは
Graphvizはグラフ構造を視覚化するためのオープンソースツールです。さまざまな形式のグラフ（有向・無向）を作成できます。
### 基本的な使い方
1. サイドバーでグラフの方向を選択します
2. 「ノードとエッジの入力」「テンプレート」「DOTコード直接入力」「AI生成」から入力方法を選びます
3. 「グラフを生成」ボタンをクリックすると図表が表示されます
### ノードとエッジのフォーム入力
- ノード: フォームに情報を入力し「ノードを追加」をクリックすると、ノード一覧に追加されます
- エッジ: フォームに情報を入力し「エッジを追加」をクリックすると、エッジ一覧に追加されます
- 追加したノードとエッジは一覧で確認でき、不要なものは削除できます
### よく使用されるノード形状
- box: 四角形
- ellipse: 楕円形（デフォルト）
- circle: 円形
- diamond: ダイヤモンド形
- cylinder: シリンダー形（データベース）
- plaintext: テキストのみ
- polygon: 多角形
- record: レコード形式（複数フィールド）
### よく使用されるノードスタイル
- filled: 塗りつぶし
- dashed: 破線枠
- dotted: 点線枠
- solid: 実線枠
- filled,rounded: 角丸塗りつぶし
- dashed,filled: 破線枠と塗りつぶし
- dotted,filled: 点線枠と塗りつぶし
- bold: 太線枠
- invis: 非表示（接続のみを表示）
### よく使用されるエッジスタイル
- solid: 実線（デフォルト）
- dashed: 破線
- dotted: 点線
- bold: 太線（penwidth="2"で実現）
### DOT言語の基本構文
```
digraph G { // 有向グラフの場合
A -> B; // AからBへのエッジ
B -> C; // BからCへのエッジ
}
```
### ノードとエッジの属性設定例
```
digraph G {
// ノードの設定
A [label="開始", shape=box, style=filled, fillcolor=lightblue];
// エッジの設定
A -> B [label="フロー", style=dashed, color=red];
}
```
### AI生成の使い方
1. サイドバーで「AI生成機能を有効化」にチェックを入れます
2. 使用するCortex AIモデルを選択します
3. 「AI生成」タブを選択し、作成したい図表の説明を入力します
4. 「AIでGraphvizコードを生成」ボタンをクリックすると、説明に基づいた図表が生成されます
""")
# フッター
st.caption("Created by Tsubasa Kanno")
使い方の例
このアプリは以下4つの機能を提供しています：
- ノードとエッジの入力: UI から対話的にノードとエッジを追加
- テンプレート: 既存のテンプレートを選択してカスタマイズ
- DOT コード直接入力: Graphviz の構文を直接編集
- AI 生成: 自然言語の説明からフロー図を自動生成
特に AI 生成機能では、次のようなプロンプトを入力するだけで図表が生成されます：
Webアプリケーションのシステム構成図を作成してください。
ユーザーはブラウザからアクセスし、ロードバランサーを経由して複数のWebサーバーにアクセスします。
Webサーバーはアプリケーションサーバーと通信し、アプリケーションサーバーはデータベースサーバーにデータを格納します。
また、キャッシュサーバーも使用しています。
Cortex AI により生成されたグラフ
最後に
Streamlit in Snowflake の Graphviz ライブラリと Cortex AI を組み合わせることで、技術的な知識がなくても高品質なフロー図やアーキテクチャ図を作成できるアプリケーションを実現できました。
特に最近はマルチモーダルの生成 AI での図表の読み取りや生成などができるようになっていますが、現状は全く調整せずに活用するのは難しいと考えられます。しかし Graphviz のようなコードベースで図表を生成できるツールと組み合わせることで、生成 AI の図表の読み取りや生成の揺らぎを吸収することが期待できます。是非皆様も本アプリを試してみていただき、新しい生成 AI の活用方法にチャレンジしてみてください。
宣伝
生成AI Conf 様の Webinar で登壇しました！
『生成AI時代を支えるプラットフォーム』というテーマの Webinar で NVIDIA 様、古巣の AWS 様と共に Snowflake 社員としてデータ*AI をテーマに LTをしました！以下が動画アーカイブとなりますので是非ご視聴いただければ幸いです！
SNOWFLAKE WORLD TOUR TOKYO のオンデマンド配信中！
Snowflake の最新情報を知ることができる大規模イベント『SNOWFLAKE WORLD TOUR TOKYO』が2024/9/11-12@ANAインターコンチネンタル東京で開催されました。
現在オンデマンド配信中ですので数々の最新のデータ活用事例をご覧ください。
また私が登壇させていただいた『今から知ろう！Snowflakeの基本概要』では、Snowflakeのコアの部分を30分で押さえられますので、Snowflake をイチから知りたい方、最新の Snowflake の特徴を知りたい方は是非ご視聴いただければ嬉しいですmm
X で Snowflake の What's new の配信してます
X で Snowflake の What's new の更新情報を配信しておりますので、是非お気軽にフォローしていただければ嬉しいです。
日本語版
Snowflake の What's New Bot (日本語版)
English Version
Snowflake What's New Bot (English Version)
変更履歴
(20250320) 新規投稿
Discussion