LangChain Chain連鎖処理デモンストレーション

# プログラム名: LangChain Chain連鎖処理デモンストレーション
# 特徴技術名: LangChain
# 出典: Harrison Chase. (2022). LangChain [Computer software]. https://github.com/langchain-ai/langchain
# 特徴機能: Chain機能 - 複数のLLM呼び出しを連鎖的に実行し、前段階の出力を次段階の入力として自動的に受け渡す機能
# 学習済みモデル: Gemini（Google DeepMindのマルチモーダルAIモデル、APIキー経由でアクセス）
# 方式設計:
#   - 関連利用技術:
#     - ChatGoogleGenerativeAI: Gemini APIを呼び出すLLM実装
#     - PromptTemplate: プロンプトの雛形を管理し、変数を動的に挿入する機能
#     - SimpleSequentialChain: 複数のLLMChainを順次実行するチェーン実装
#   - 入力と出力: 入力: テキスト文字列、出力: コンソール出力（処理フローと結果）
#   - 処理手順:
#     1. 入力テキストを要約プロンプトに挿入
#     2. Gemini APIが要約を生成
#     3. 要約結果をタイトル生成プロンプトに挿入
#     4. Gemini APIがタイトルを生成
#   - 前処理、後処理: なし（デモンストレーション用のため）
#   - 追加処理: verbose=Trueによる処理過程の可視化
#   - 調整を必要とする設定値: GEMINI_API_KEY（Google AI Studioで取得したAPIキー）
# 将来方策: Chain機能の応用展開（複数段階の処理連鎖、条件分岐の追加等）
# その他の重要事項: このプログラムは教育目的のデモンストレーションであり、Gemini APIを使用して実際のAI推論を行う
# 前準備: pip install langchain langchain-core langchain-google-genai

# 必要ライブラリのインポート
from langchain.chains import SimpleSequentialChain, LLMChain
from langchain.prompts import PromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
import time
import datetime
import tkinter as tk
from tkinter import ttk, scrolledtext
import os

# 設定定数
PROCESSING_INTERVAL = 1  # 処理間隔（秒）
OUTPUT_FILE = 'result.txt'  # 結果保存ファイル名
DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'  # 日時表示形式

# プログラム開始時の説明表示
print('=== LangChain Chain連鎖処理デモンストレーション ===')
print('概要: LangChainのChain機能を使用して、複数の処理を連鎖的に実行します')
print('注意: このプログラムはGemini APIを使用した教育用デモです')
print('      実際のAI推論が行われます')
print()

# プロンプトテンプレート定義
SUMMARY_PROMPT = '以下の内容を簡潔に要約してください:\n{input}\n\n要約:'
TITLE_PROMPT = '以下の要約文に基づいて、適切なタイトルを生成してください:\n{input}\n\nタイトル:'

# 入力テキスト定義
INPUT_TEXT = ('LangChainは大規模言語モデルの出力を組み合わせて自動化するPythonライブラリです。'
              '開発者はプロンプトテンプレート（プロンプトの雛形）を使って複数のAI処理を連鎖させることができます。')

# 結果記録用リスト
results = []

# グローバル変数
api_key = ''
root = None
api_key_var = None
prompt1_text = None
prompt2_text = None
input_text = None
result_text = None
llm = None

# .envファイルからAPIキー読み込み
env_file_path = '.env'
if os.path.exists(env_file_path):
    try:
        with open(env_file_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    if line.startswith('GEMINI_API_KEY='):
                        api_key = line.split('=', 1)[1].strip()
    except FileNotFoundError:
        print(f'.envファイルが見つかりません: {env_file_path}')
    except PermissionError:
        print(f'.envファイルへのアクセス権限がありません: {env_file_path}')
    except Exception as e:
        print(f'.envファイルの読み込み中に予期しないエラーが発生しました: {str(e)}')


def log_with_timestamp(message):
    """タイムスタンプ付きでメッセージを記録"""
    timestamp = datetime.datetime.now().strftime(DATETIME_FORMAT)
    results.append(f'[{timestamp}] {message}')


# APIキー設定処理
def set_api_key():
    global api_key, llm
    api_key = api_key_var.get().strip()
    if api_key:
        os.environ['GOOGLE_API_KEY'] = api_key
        llm = ChatGoogleGenerativeAI(
            model='gemini-2.0-flash-exp',
            temperature=0
        )
        result_text.insert(tk.END, f'APIキーが設定されました\n')
    else:
        result_text.insert(tk.END, f'APIキーが空です\n')


# Chain実行処理
def execute_chain():
    result_text.delete('1.0', tk.END)
    result_text.insert(tk.END, 'Chain構成: 入力 → 要約生成 → タイトル生成\n\n')

    if not llm:
        result_text.insert(tk.END, 'エラー: APIキーが設定されていません\n')
        return

    # 入力テキストを取得
    input_text_value = input_text.get('1.0', tk.END).strip()
    if not input_text_value:
        result_text.insert(tk.END, 'エラー: 入力テキストが空です\n')
        return

    try:
        # プロンプトテンプレートインスタンス作成
        summary_prompt_text = prompt1_text.get('1.0', tk.END).strip()
        title_prompt_text = prompt2_text.get('1.0', tk.END).strip()

        try:
            summary_prompt = PromptTemplate(input_variables=['input'], template=summary_prompt_text)
        except Exception as e:
            raise ValueError(f'要約プロンプトテンプレートの作成に失敗しました: {str(e)}')

        try:
            title_prompt = PromptTemplate(input_variables=['input'], template=title_prompt_text)
        except Exception as e:
            raise ValueError(f'タイトルプロンプトテンプレートの作成に失敗しました: {str(e)}')

        # LLMChainインスタンス作成（LLM + プロンプトテンプレートの組み合わせ）
        try:
            summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key='summary')
        except Exception as e:
            raise RuntimeError(f'要約チェーンの作成に失敗しました: {str(e)}')

        try:
            title_chain = LLMChain(llm=llm, prompt=title_prompt, output_key='title')
        except Exception as e:
            raise RuntimeError(f'タイトルチェーンの作成に失敗しました: {str(e)}')

        # 連続Chain作成（複数のLLMChainを順次実行）
        try:
            sequential_chain = SimpleSequentialChain(chains=[summary_chain, title_chain], verbose=True)
        except Exception as e:
            raise RuntimeError(f'シーケンシャルチェーンの作成に失敗しました: {str(e)}')

        # 結果出力
        result_text.insert(tk.END, f'入力テキスト: {input_text_value}\n')
        log_with_timestamp(f'入力テキスト: {input_text_value}')
        result_text.insert(tk.END, '\n')

        result_text.insert(tk.END, '--- チェーン実行開始 ---\n')
        log_with_timestamp('チェーン実行開始')

        # 1秒間隔での処理状況表示
        time.sleep(PROCESSING_INTERVAL)
        result_text.insert(tk.END, '処理中: 要約生成...\n')
        log_with_timestamp('処理中: 要約生成')
        root.update()

        # Chain実行（入力テキストを連続処理）
        try:
            result = sequential_chain.run(input_text_value)
        except Exception as e:
            raise RuntimeError(f'チェーンの実行中にエラーが発生しました: {str(e)}')

        time.sleep(PROCESSING_INTERVAL)
        result_text.insert(tk.END, '処理中: タイトル生成...\n')
        log_with_timestamp('処理中: タイトル生成')
        root.update()

        result_text.insert(tk.END, '--- チェーン実行完了 ---\n')
        result_text.insert(tk.END, f'最終出力: {result}\n')
        log_with_timestamp(f'最終出力: {result}')

        result_text.insert(tk.END, '\n=== 重要事項 ===\n')
        result_text.insert(tk.END, 'この出力はGemini APIによる実際のAI推論結果です\n')
        result_text.insert(tk.END, 'Gemini APIが入力内容を理解して応答を生成しています\n')
        result_text.insert(tk.END, '実際のAI推論と内容理解が行われています\n')

        result_text.insert(tk.END, '\n=== 処理ステップの説明 ===\n')
        result_text.insert(tk.END, '1. 入力テキストがprompt1テンプレートに挿入される\n')
        result_text.insert(tk.END, '2. Gemini APIが要約を生成する\n')
        result_text.insert(tk.END, '3. その出力がprompt2テンプレートに挿入される\n')
        result_text.insert(tk.END, '4. Gemini APIがタイトルを生成する\n')
        result_text.insert(tk.END, '5. 連鎖的処理が完了し、最終結果が出力される\n')
        result_text.insert(tk.END, '\n注記：プロンプトの内容や入力テキストが応答生成に影響する\n')

        # 結果をファイルに保存
        try:
            with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
                f.write('\n'.join(results))
            result_text.insert(tk.END, f'\n{OUTPUT_FILE}に保存しました\n')
        except PermissionError:
            result_text.insert(tk.END, f'\nエラー: {OUTPUT_FILE}への書き込み権限がありません\n')
        except Exception as e:
            result_text.insert(tk.END, f'\nファイル保存中にエラーが発生しました: {str(e)}\n')

    except ValueError as e:
        result_text.insert(tk.END, f'\n設定エラー: {str(e)}\n')
    except RuntimeError as e:
        result_text.insert(tk.END, f'\n実行エラー: {str(e)}\n')
    except Exception as e:
        result_text.insert(tk.END, f'\n予期しないエラーが発生しました: {str(e)}\n')


# GUI作成
root = tk.Tk()
root.title('LangChain Chain連鎖処理デモンストレーション')
root.geometry('800x700')

main_frame = ttk.Frame(root, padding='10')
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))

# APIキー設定フレーム
api_frame = ttk.LabelFrame(main_frame, text='API設定', padding='5')
api_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))

ttk.Label(api_frame, text='APIキー:').grid(row=0, column=0, sticky=tk.W)
api_key_var = tk.StringVar()
if api_key:
    api_key_var.set(api_key)
api_entry = ttk.Entry(api_frame, textvariable=api_key_var, width=50)
api_entry.grid(row=0, column=1, padx=(5, 0))
ttk.Button(api_frame, text='APIキー設定', command=set_api_key).grid(row=0, column=2, padx=(5, 0))

# 入力テキストフレーム
input_frame = ttk.LabelFrame(main_frame, text='入力テキスト', padding='5')
input_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))

input_text = scrolledtext.ScrolledText(input_frame, height=3, width=70)
input_text.grid(row=0, column=0, columnspan=2)
input_text.insert('1.0', INPUT_TEXT)

# プロンプト設定フレーム
prompt_frame = ttk.LabelFrame(main_frame, text='プロンプト設定', padding='5')
prompt_frame.grid(row=2, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))

ttk.Label(prompt_frame, text='要約プロンプト:').grid(row=0, column=0, sticky=tk.W)
prompt1_text = scrolledtext.ScrolledText(prompt_frame, height=3, width=70)
prompt1_text.grid(row=1, column=0, columnspan=2, pady=(5, 10))
prompt1_text.insert('1.0', SUMMARY_PROMPT)

ttk.Label(prompt_frame, text='タイトル生成プロンプト:').grid(row=2, column=0, sticky=tk.W)
prompt2_text = scrolledtext.ScrolledText(prompt_frame, height=3, width=70)
prompt2_text.grid(row=3, column=0, columnspan=2, pady=(5, 0))
prompt2_text.insert('1.0', TITLE_PROMPT)

# 実行ボタン
ttk.Button(main_frame, text='実行開始', command=execute_chain).grid(row=3, column=0, columnspan=2, pady=(10, 10))

# 結果表示フレーム
result_frame = ttk.LabelFrame(main_frame, text='実行結果', padding='5')
result_frame.grid(row=4, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S))

result_text = scrolledtext.ScrolledText(result_frame, height=15, width=80)
result_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))

# グリッド設定
root.columnconfigure(0, weight=1)
root.rowconfigure(0, weight=1)
main_frame.columnconfigure(0, weight=1)
main_frame.rowconfigure(4, weight=1)
result_frame.columnconfigure(0, weight=1)
result_frame.rowconfigure(0, weight=1)

# APIキー自動設定
if api_key:
    set_api_key()

root.mainloop()