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

# プログラム名: LangChain Chain連鎖処理デモンストレーションプログラム
# 特徴技術名: LangChain
# 出典: Harrison Chase. (2022). LangChain [Computer software]. https://github.com/langchain-ai/langchain
# 特徴機能: LCELによる連鎖処理（前段の出力を次段の入力に受け渡す）
# 学習済みモデル: Gemini（Google DeepMindのマルチモーダルAIモデル、APIキー経由でアクセス）
# 方式設計:
#   - 関連利用技術:
#     - ChatGoogleGenerativeAI: Gemini APIを呼び出すLLM実装
#     - PromptTemplate: プロンプトの雛形を管理し、変数を動的に挿入する機能
#     - LCEL: パイプ演算子（|）による段階実行
#   - 入出力: 入力: テキスト文字列、出力: GUI表示（処理フロー、中間要約、最終結果）
#   - 追加機能: 生成パラメータUI（temperature, top_p, max_output_tokens）
#   - 前処理、後処理: なし
#   - 調整値: GEMINI_API_KEY（Google AI Studioで取得）
# 前準備: pip install langchain langchain-core langchain-google-genai

from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
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のLCELを使用して、複数の処理を連鎖的に実行する')
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
summary_text = None
result_text = None
llm = None

# パラメータUI用変数
temperature_var = None
top_p_var = None
max_tokens_var = 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 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}')

def apply_llm_settings():
    """UIパラメータに基づくLLM再構築"""
    global llm
    if not api_key:
        return False
    os.environ['GOOGLE_API_KEY'] = api_key
    try:
        llm = ChatGoogleGenerativeAI(
            model='gemini-2.0-flash-exp',
            temperature=float(temperature_var.get()),
            top_p=float(top_p_var.get()),
            max_output_tokens=int(max_tokens_var.get())
        )
        return True
    except Exception as e:
        if result_text is not None:
            result_text.insert(tk.END, f'パラメータ適用中にエラーが発生: {str(e)}\n')
        return False

def set_api_key():
    global api_key
    api_key = api_key_var.get().strip()
    if api_key:
        ok = apply_llm_settings()
        if ok and result_text is not None:
            result_text.insert(tk.END, 'APIキーが設定された\n')
        elif result_text is not None:
            result_text.insert(tk.END, 'APIキーは設定されたがLLM初期化に失敗した\n')
    else:
        if result_text is not None:
            result_text.insert(tk.END, 'APIキーが空である\n')

def build_summary_chain(summary_prompt_text: str):
    summary_prompt = PromptTemplate(input_variables=['input'], template=summary_prompt_text)
    return summary_prompt | llm | StrOutputParser()

def build_title_chain(title_prompt_text: str):
    title_prompt = PromptTemplate(input_variables=['input'], template=title_prompt_text)
    return title_prompt | llm | StrOutputParser()

def execute_chain():
    result_text.delete('1.0', tk.END)
    summary_text.delete('1.0', tk.END)
    result_text.insert(tk.END, 'Chain構成: 入力 → 要約生成 → タイトル生成（LCEL段階実行）\n\n')

    if not api_key:
        result_text.insert(tk.END, 'エラー: APIキーが設定されていない\n')
        return
    if not apply_llm_settings():
        result_text.insert(tk.END, 'エラー: LLMの初期化に失敗した\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()

        summary_chain = build_summary_chain(summary_prompt_text)
        title_chain = build_title_chain(title_prompt_text)

        result_text.insert(tk.END, f'入力テキスト: {input_text_value}\n')
        result_text.insert(tk.END, f'使用パラメータ: temperature={temperature_var.get()}, top_p={top_p_var.get()}, max_output_tokens={max_tokens_var.get()}\n\n')
        log_with_timestamp(f'入力テキスト: {input_text_value}')
        log_with_timestamp(f'使用パラメータ: temperature={temperature_var.get()}, top_p={top_p_var.get()}, max_output_tokens={max_tokens_var.get()}')

        # 要約生成
        result_text.insert(tk.END, '--- 要約生成を開始 ---\n')
        log_with_timestamp('要約生成開始')
        time.sleep(PROCESSING_INTERVAL)
        root.update()

        try:
            summary_result = summary_chain.invoke({'input': input_text_value})
        except Exception as e:
            raise RuntimeError(f'要約生成中にエラーが発生: {str(e)}')

        summary_text.insert(tk.END, summary_result + '\n')
        log_with_timestamp(f'要約出力: {summary_result}')
        result_text.insert(tk.END, '--- 要約生成が完了 ---\n\n')

        # タイトル生成
        result_text.insert(tk.END, '--- タイトル生成を開始 ---\n')
        log_with_timestamp('タイトル生成開始')
        time.sleep(PROCESSING_INTERVAL)
        root.update()

        try:
            final_output = title_chain.invoke({'input': summary_result})
        except Exception as e:
            raise RuntimeError(f'タイトル生成中にエラーが発生: {str(e)}')

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

        # 処理ステップ
        result_text.insert(tk.END, '\n=== 処理ステップの説明 ===\n')
        result_text.insert(tk.END, '1. 入力テキストが要約プロンプトに挿入される\n')
        result_text.insert(tk.END, '2. Gemini APIが要約を生成し、要約タブに表示される\n')
        result_text.insert(tk.END, '3. 要約がタイトル生成プロンプトに挿入される\n')
        result_text.insert(tk.END, '4. Gemini APIがタイトルを生成し、最終結果タブに表示される\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 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('800x800')  # 高さを800に設定

main_frame = ttk.Frame(root, padding='8')  # 余白を抑制
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, 6))
ttk.Label(api_frame, text='APIキー:').grid(row=0, column=0, sticky=tk.W)
api_key_var = tk.StringVar(value=api_key if api_key else '')
api_entry = ttk.Entry(api_frame, textvariable=api_key_var, width=48)
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=(6, 0))

# 生成パラメータ
params_frame = ttk.LabelFrame(main_frame, text='生成パラメータ', padding='5')
params_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 6))
temperature_var = tk.DoubleVar(value=0.0)
top_p_var = tk.DoubleVar(value=0.95)
max_tokens_var = tk.IntVar(value=512)
ttk.Label(params_frame, text='temperature (0.0-1.0):').grid(row=0, column=0, sticky=tk.W)
ttk.Scale(params_frame, from_=0.0, to=1.0, orient='horizontal', variable=temperature_var).grid(row=0, column=1, sticky=(tk.W, tk.E), padx=(5, 0))
ttk.Label(params_frame, text='top_p (0.0-1.0):').grid(row=1, column=0, sticky=tk.W)
ttk.Scale(params_frame, from_=0.0, to=1.0, orient='horizontal', variable=top_p_var).grid(row=1, column=1, sticky=(tk.W, tk.E), padx=(5, 0))
ttk.Label(params_frame, text='max_output_tokens:').grid(row=2, column=0, sticky=tk.W)
tk.Spinbox(params_frame, from_=1, to=8192, increment=1, textvariable=max_tokens_var, width=8).grid(row=2, column=1, sticky=tk.W, padx=(5, 0))

# 入力テキスト
input_frame = ttk.LabelFrame(main_frame, text='入力テキスト', padding='5')
input_frame.grid(row=2, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 6))
input_text = scrolledtext.ScrolledText(input_frame, height=3, width=70)  # 高さを削減
input_text.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E))
input_text.insert('1.0', INPUT_TEXT)

# プロンプト設定（2段→各2行に短縮）
prompt_frame = ttk.LabelFrame(main_frame, text='プロンプト設定', padding='5')
prompt_frame.grid(row=3, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 6))
ttk.Label(prompt_frame, text='要約プロンプト:').grid(row=0, column=0, sticky=tk.W)
prompt1_text = scrolledtext.ScrolledText(prompt_frame, height=2, width=70)  # 縦縮小
prompt1_text.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(4, 6))
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=2, width=70)  # 縦縮小
prompt2_text.grid(row=3, column=0, columnspan=2, sticky=(tk.W, tk.E))
prompt2_text.insert('1.0', TITLE_PROMPT)

# 実行ボタン（行を節約・同列配置）
ttk.Button(main_frame, text='実行開始', command=execute_chain).grid(row=4, column=0, columnspan=2, pady=(6, 6))

# 出力領域をタブ化して縦占有を削減
output_notebook = ttk.Notebook(main_frame)
output_notebook.grid(row=5, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S))

summary_tab = ttk.Frame(output_notebook)
result_tab = ttk.Frame(output_notebook)
output_notebook.add(summary_tab, text='要約結果')
output_notebook.add(result_tab, text='最終結果')

summary_text = scrolledtext.ScrolledText(summary_tab, height=10, width=80)  # 高さを抑制
summary_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
result_text = scrolledtext.ScrolledText(result_tab, height=10, 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.columnconfigure(1, weight=1)
main_frame.rowconfigure(5, weight=1)  # Notebook が余白を吸収
summary_tab.columnconfigure(0, weight=1)
summary_tab.rowconfigure(0, weight=1)
result_tab.columnconfigure(0, weight=1)
result_tab.rowconfigure(0, weight=1)

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

root.mainloop()