DifyでLLMが出力したプログラムをローカルで実行

以下のQiita記事で、DIfyでLLMが出力したテキストを、ローカルに保存する手順を紹介している。

Difyからローカル環境のフォルダにファイル出力・読込 - Qiita

この記事では、ローカルに保存したプログラムをDifyから呼び出し、ローカルで実行するフローを作成する。

workflow

以下のようなフローを構築する。

このフローで行われる処理をフロー図で表すと以下のようになる。

環境構築

以下の記事の環境構築を行う。
（WIndowsであれば、基本的に１コマンドで構築可能な想定。pythonとDifyは別途インストールが必要）

Difyからローカル環境のフォルダにファイル出力・読込 - Qiita

Difyの設定

「開始」ノード

input_text：出力するプログラムの概要（例：正常したことを出力するプログラム）
folder_name：出力先フォルダ（例：sample）
file_name：出力するファイル名（例：sample.py）

「LLM（プログラム作成）」ノード

「プログラムの要件」に基づいたpythonのプログラムを作成してください。
「python XXX.py」コマンドで実行できるプログラムにしてください。

「プログラムの要件」：{{#1715394478158.input_text#}}

「LLM（プログラム抽出）」ノード

「プログラムを含んだテキスト」からpythonのプログラムのみを出力してください。
「````python」と「```」が含まれている場合は削除してください。

「プログラムを含んだテキスト」：{{#1716114640530.text#}}

「HTTPリクエスト（upload）」ノード

{
  "folder_name": "{{#1715394478158.folder_name#}}",
  "file_name": "{{#1715394478158.file_name#}}",
  "file_content": "{{#1725366501715.text#}}"
}

「HTTPリクエスト（execute）」ノード

{
  "folder_name": "{{#1715394478158.folder_name#}}",
  "file_name": "{{#1715394478158.file_name#}}"
}

後処理ノード

import json

def main(arg1: str) -> dict:

    # JSONデータをパース
    parsed_data = json.loads(arg1)

    return {
        "result": str(parsed_data)
    }

「終了」ノード

flask

プログラム

以下の記事の環境構築を実施する。

Difyからローカル環境のフォルダにファイル出力・読込 - Qiita

上記の手順で作成された「app.py」を以下のプログラムで修正し、「python app.py」で実行する。

# -*- coding: utf-8 -*- 
from flask import Flask, request, jsonify, send_file 
import os 
import sys
from io import StringIO
import contextlib

app = Flask(__name__) 

@app.route('/upload', methods=['POST']) 
def upload_file(): 
    # JSONデータの取得 
    data = request.json 
    
    # フォルダ名とファイル名の取得 
    folder_name = data.get('folder_name') 
    file_name = data.get('file_name') 
    file_content = data.get('file_content') 
    
    # フォルダ名とファイル名が存在するかチェック 
    if not folder_name or not file_name: 
        return jsonify({'error': 'Folder name and file name are required'}), 400 
    
    # 出力先のディレクトリを作成（存在しない場合） 
    output_dir = os.path.join('output', folder_name) 
    if not os.path.exists(output_dir): 
        os.makedirs(output_dir) 
    
    # ファイル内容を指定されたディレクトリに保存 
    file_path = os.path.join(output_dir, file_name) 
    try: 
        with open(file_path, 'w', encoding='utf-8') as f: 
            f.write(file_content) 
    except Exception as e: 
        return jsonify({'error': str(e)}), 500 
    
    return jsonify({'message': 'File saved successfully', 'path': file_path}), 200 

@app.route('/retrieve', methods=['GET']) 
def retrieve_file(): 
    # JSONデータの取得 
    data = request.json 
    
    # クエリパラメータからフォルダ名とファイル名を取得 
    folder_name = data.get('folder_name') 
    file_name = data.get('file_name') 
    
    # フォルダ名とファイル名が存在するかチェック 
    if not folder_name or not file_name: 
        return jsonify({'error': 'Folder name and file name are required'}), 400 
    
    # ファイルパスを構築 
    file_path = os.path.join('output', folder_name, file_name) 
    
    # ファイルが存在するかチェック 
    if not os.path.exists(file_path): 
        return jsonify({'error': 'File not found'}), 404 
    
    # ファイルを送信 
    return send_file(file_path, as_attachment=True) 

@app.route('/execute', methods=['POST'])
def execute_file():
    # JSONデータの取得
    data = request.json
    
    # フォルダ名とファイル名の取得
    folder_name = data.get('folder_name')
    file_name = data.get('file_name')
    
    # フォルダ名とファイル名が存在するかチェック
    if not folder_name or not file_name:
        return jsonify({'error': 'Folder name and file name are required'}), 400
    
    # ファイルパスを構築
    file_path = os.path.join('output', folder_name, file_name)
    
    # ファイルが存在するかチェック
    if not os.path.exists(file_path):
        return jsonify({'error': 'File not found'}), 404
    
    # ファイルの内容を読み込む
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            code = f.read()
    except Exception as e:
        return jsonify({'error': f'Error reading file: {str(e)}'}), 500
    
    # コードの実行と出力のキャプチャ
    output = StringIO()
    with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
        try:
            exec(code)
        except Exception as e:
            print(f"Error: {str(e)}")
    
    return jsonify({'result': output.getvalue()}), 200

if __name__ == '__main__': 
    app.run(host='0.0.0.0', port=5000, debug=True)

結果

以下のように、正常に終了している。

Dify

フォルダ

ファイル