LangChainの使い方 - うぇぶすく

LangChainは、言語モデルを活用して、チャットボットや自動化されたテキスト生成タスクを簡単に作成、管理、実行できるフレームワークです。

このツールは特に、OpenAIのGPTモデルなどの強力なAI言語技術をバックエンドに使用して、ユーザーの質問や入力に対するインテリジェントな応答を生成するのに役立ちます。

LangChainはプロンプト設計、出力の解析、対話の管理といった複雑なプロセスを簡単に構築できるようにデザインされており、開発者はより効果的な対話システムを迅速に開発できます。

1. LangChainのメリット

LangChainを使用する主なメリットは、開発の速度と柔軟性です。

コーディングの専門知識が少なくても、LangChainの直感的なAPIと組み合わせることで、複雑な自然言語理解タスクが容易になります。

さらに、LangChainを利用することで、カスタマーサポート、データ分析、コンテンツ生成といった幅広い用途において、自動化とスケーラビリティが大幅に向上します。

これにより、企業や開発者はリソースをより創造的な作業に集中させることができ、全体的な生産性の向上に寄与します。

2. LangChainの基本コンセプト

LangChainは、複数の主要コンポーネントを組み合わせて、柔軟かつ強力な言語処理アプリケーションを構築します。

その核となるのは「OpenAIプロンプト」と「出力パーサー」です。

2-1. OpenAIプロンプト

OpenAIプロンプトは、ユーザーの入力や以前の対話履歴から適切なクエリを生成し、AIモデルに送信するためのテキストを形成します。

このプロンプトは、AIがより正確かつ関連性の高い回答を生成するための文脈を提供します。

開発者はこのプロンプトをカスタマイズすることで、特定の応答スタイルやトーン、対話の流れを制御できます。

2-2. 出力パーサー

出力パーサーは、AIモデルからの生のテキスト応答を解析し、アプリケーションが処理しやすい形式に変換します。

これにより、応答から重要な情報を抽出したり、特定のフォーマットに整形する作業が効率化されます。

2-3. 問題解決の例

LangChainを使用することで解決可能な典型的な問題には、以下のようなものがあります：

カスタマーサポートの自動化：
LangChainは顧客からのよくある質問に自動的に応答し、サポートチームの負担を軽減します。
AIが対話を通じて問題を解決できるため、顧客満足度が向上します。
データ抽出と分析：
大量のテキストデータから重要な情報を自動的に抽出し、それを構造化された形で分析可能にします。
これは研究、市場分析、競合調査など多岐にわたる分野で利用できます。
コンテンツ生成：
LangChainを使用して記事やレポート、マーケティング素材などのテキストコンテンツを自動生成します。これにより、コンテンツ制作のコストを削減し、生産性を向上させることができます。

これらのコンポーネントと機能を組み合わせることで、LangChainは多様なニーズに応じたカスタマイズが可能であり、各種ビジネスプロセスの効率化と自動化に貢献します。

3. 実践的な例

LangChainの柔軟性とパワーを実証するために、いくつかの具体的な応用例を紹介します。

これには、カスタマーサポート用チャットボットの開発、データ分析プロジェクト、自動応答システムの構築が含まれます。

3-1. カスタマーサポート用チャットボット - ケーススタディ：オンライン小売業者

オンライン小売業者は、顧客からの問い合わせに迅速かつ効率的に対応するためにカスタマーサポート用チャットボットを導入しました。

LangChainを活用して、顧客の問い合わせに基づいて適切な回答を生成するシステムを構築。

このチャットボットは、製品情報の提供、注文状況の確認、配送オプションの説明など、多岐にわたるタスクを自動で処理します。

この結果、顧客満足度が向上し、サポートチームの負担も大幅に軽減されました。

3-2. データ分析 - ケーススタディ：マーケティング会社

あるマーケティング会社は、市場調査のためのデータ分析プロジェクトにLangChainを使用しました。

膨大な量の顧客フィードバックとオンラインレビューから重要なインサイトを自動的に抽出し、これを基に効果的なマーケティング戦略を策定。

LangChainの自然言語処理能力を活用して、感情分析やキーワード抽出を行い、データドリブンな意思決定を支援しました。

3-3. 自動応答システム - ケーススタディ：教育機関

大学が学生からの頻繁な質問に対処するために、自動応答システムを導入することを決定しました。

LangChainを利用して、授業スケジュール、試験日、登録手続きなどの情報を提供するシステムを構築。

学生がウェブサイトやモバイルアプリから簡単に情報を取得できるようになり、事務の効率化が図られ、学生サービスの質が向上しました。

これらの例からわかるように、LangChainは多種多様な業種や業務での応用が可能です。

その柔軟性と強力な言語処理能力により、多くの企業や機関が効率的なオペレーションと向上したユーザー体験を実現しています。

4. 基本的な使い方 - シンプルなチャットボット

LangChainを利用するための最初のステップは、簡単なプロジェクトのセットアップから始まります。

ここでは、LangChainとSteamlitを使った簡単なチャットシステムの例を通じて、基本的な流れをわかりやすく説明します。

4-1. 環境準備

LangChainを使用するためには、Pythonがインストールされている環境が必要です。

Pythonと必要なパッケージをインストールした後、LangChain関連のライブラリもインストールします。

pip install openai==1.29.0
pip install streamlit==1.33.0
pip install langchain==0.1.6
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3

#anthropicを使用する場合
pip install langchain-anthropic==0.1.11

#googleを使用する場合
pip install langchain-google-genai==1.0.3

4-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY=◯◯◯◯◯」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
main.py … ここにコードを入力します

💡

AIPキーが漏洩して悪用されると大金が課金される可能性があるので、ご注意ください。

4-3. 全部のコード

# 必要なライブラリをインポート
import streamlit as st
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# main関数の定義：アプリのメインロジックを含む
def main():
    # Streamlitページの基本設定
    st.set_page_config(
        page_title="My Great ChatGPT",  # ウェブページのタイトル
        page_icon="🤗"  # ウェブページのアイコン（絵文字使用）
    )
    st.header("My Great ChatGPT🤗")  # ページのヘッダーにタイトルを表示

    # セッションステートでチャット履歴を管理
    if "message_history" not in st.session_state:
        # チャット履歴が存在しない場合、初期化する
        st.session_state.message_history = [
            ("system", "You are a helpful assistant.")  # システムからの最初のメッセージ
        ]
    
    # LangChainの設定：ChatGPTモデルのインスタンスを作成
    llm = ChatOpenAI(temperature=0)  # 温度パラメータを0に設定して確定的な回答を求める

    # チャットプロンプトの設定
    prompt = ChatPromptTemplate.from_messages([
        *st.session_state.message_history,  # 現在のチャット履歴をプロンプトに含める
        ("user", "{user_input}")  # ユーザー入力を受け取るためのプレースホルダー
    ])

    # 出力パーサーの設定
    output_parser = StrOutputParser()  # 文字列としてモデルの出力を解析する

    # チャット処理の連鎖を設定
    chain = prompt | llm | output_parser  # プロンプト、モデル、出力パーサーをパイプで繋ぐ

    # ユーザーの入力を受け取る
    if user_input := st.chat_input("聞きたいことを入力してね"):
        with st.spinner("ChatGPT is typing ..."):  # 処理中のインジケーターを表示
            # チェーンを実行し、モデルからの回答を取得
            response = chain.invoke({"user_input": user_input})
        
        # ユーザーの質問を履歴に追加
        st.session_state.message_history.append(("user", user_input))
        # ChatGPTの回答を履歴に追加
        st.session_state.message_history.append(("ai", response))

    # チャット履歴を表示
    for role, message in st.session_state.get("message_history", []):
        st.chat_message(role).markdown(message)  # Streamlitのchat_messageを使ってメッセージを表示

# Pythonスクリプトとして実行された場合にmain関数を呼び出す
if __name__ == '__main__':
    main()

4-4. コードの解説

このコードは、Streamlitを使ってウェブベースのChatGPT対話システムを構築するためのものです。

主要な部分について説明します。

STEP 1. ライブラリのインポート

import streamlit as st
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

streamlit as st: Streamlitライブラリをインポートし、ウェブアプリのフロントエンドを簡単に構築します。
ChatOpenAI: LangChainライブラリの一部で、OpenAIのAPIを活用して言語モデルを操作します。
ChatPromptTemplate: ユーザーの入力や過去の対話を基にプロンプトを生成するためのクラス。
StrOutputParser: モデルの出力を文字列として解析するパーサー。

STEP 2. main関数の定義

この関数はアプリの主要なロジックを含んでいます。

st.set_page_config(page_title="My Great ChatGPT", page_icon="🤗")
st.header("My Great ChatGPT🤗")

set_page_config: ページのタイトルとアイコンを設定。
header: ページのヘッダーにタイトルを表示。

STEP 3. セッションステートの管理

if "message_history" not in st.session_state:
		st.session_state.message_history = [("system", "You are a helpful assistant.")]

Streamlitのセッションステート機能を使用して、チャット履歴を保存し、セッション間で状態を保持します。

STEP 4. LangChainの設定

llm = ChatOpenAI(temperature=0)
prompt = ChatPromptTemplate.from_messages([
    *st.session_state.message_history,
    ("user", "{user_input}")
])
output_parser = StrOutputParser()
chain = prompt | llm | output_parser

ChatOpenAI: 温度パラメータを0に設定して確定的な回答を求めるChatGPTモデルを初期化します。
ChatPromptTemplate: チャット履歴とユーザーの入力からプロンプトを生成します。
output_parser: モデルからの応答を文字列として解析します。
chain: これらの要素をパイプで連結し、連続的な処理を行うためのチェーンを構成します。

STEP 5. ユーザーの入力処理と応答の生成

if user_input := st.chat_input("聞きたいことを入力してね"):
    with st.spinner("ChatGPT is typing ..."):
        response = chain.invoke({"user_input": user_input})
    st.session_state.message_history.append(("user", user_input))
    st.session_state.message_history.append(("ai", response))

chat_input: ユーザーからの入力を受け取ります。
spinner: 処理中にスピナーを表示。
chain.invoke: 入力されたユーザーの質問に対してLangChainを使用して応答を生成します。

STEP 6. チャット履歴の表示

for role, message in st.session_state.get("message_history", []):
    st.chat_message(role).markdown(message)

chat_message: Streamlitの機能を使用して、ロールに基づきメッセージを適切にフォーマットして表示します。

STEP 7. スクリプトの実行

if __name__ == '__main__':
    main()

スクリプトが直接実行された場合にmain関数を呼び出します。

5. 機能を追加したシンプルなチャットボット

4で実装した機能に加えて、以下の機能を追加します。

モデルをGPT、Gemini、Claudiaから選択できる
消費したトークンが把握できる

5-1. 環境準備

pip install openai==1.29.0
pip install anthropic ==0.25.7
pip install google-generativeai ==0.5.2

pip install tiktoken==0.7.0
pip install streamlit==1.33.0

pip install langchain==0.1.6
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3
pip install langchain-anthropic==0.1.11
pip install langchain-google-genai==1.0.3

5-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY="◯◯◯◯◯”」「GOOGLE_API_KEY = "◯◯◯◯◯”」「ANTHROPIC_API_KEY = “◯◯◯◯◯”」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
main.py … ここにコードを入力します

💡

AIPキーが漏洩して悪用されると大金が課金される可能性があるので、ご注意ください。

5-3. 全部のコード

import tiktoken
import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# LangChainのモデルインポート
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# モデルの使用料金設定
MODEL_PRICES = {
    "input": {
        "gpt-3.5-turbo": 0.5 / 1_000_000,
        "gpt-4o": 5 / 1_000_000,
        "claude-3-5-sonnet-20240620": 3 / 1_000_000,
        "gemini-1.5-pro-latest": 3.5 / 1_000_000
    },
    "output": {
        "gpt-3.5-turbo": 1.5 / 1_000_000,
        "gpt-4o": 15 / 1_000_000,
        "claude-3-5-sonnet-20240620": 15 / 1_000_000,
        "gemini-1.5-pro-latest": 10.5 / 1_000_000
    }
}

def init_page():
    # Streamlitページの基本設定
    st.set_page_config(
        page_title="My Great ChatGPT",
        page_icon="🤗"
    )
    st.header("My Great ChatGPT 🤗")
    st.sidebar.title("Options")

def init_messages():
    # チャット履歴の初期化と管理
    clear_button = st.sidebar.button("Clear Conversation", key="clear")
    if clear_button or "message_history" not in st.session_state:
        st.session_state.message_history = [
            ("system", "You are a helpful assistant.")
        ]

def select_model():
    # モデル選択と設定のインターフェース
    temperature = st.sidebar.slider("Temperature:", min_value=0.0, max_value=2.0, value=0.0, step=0.01)
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    if model == "GPT-3.5":
        return ChatOpenAI(temperature=temperature, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(temperature=temperature, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(temperature=temperature, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(temperature=temperature, model="gemini-1.5-pro-latest")

def init_chain():
    # チェーンの初期化と設定
    st.session_state.llm = select_model()
    prompt = ChatPromptTemplate.from_messages([
        *st.session_state.message_history,
        ("user", "{user_input}")  # ここに後でユーザーの入力が入る
    ])
    output_parser = StrOutputParser()
    return prompt | st.session_state.llm | output_parser

def get_message_counts(text):
    # メッセージのトークン数を計算
    if "gemini" in st.session_state.model_name:
        return st.session_state.llm.get_num_tokens(text)
    else:
        encoding = tiktoken.encoding_for_model(st.session_state.model_name if "gpt" in st.session_state.model_name else "gpt-3.5-turbo")
        return len(encoding.encode(text))

def calc_and_display_costs():
    # 会話のコスト計算と表示
    output_count = 0
    input_count = 0
    for role, message in st.session_state.message_history:
        token_count = get_message_counts(message)
        if role == "ai":
            output_count += token_count
        else:
            input_count += token_count

    if len(st.session_state.message_history) == 1:
        return  # まだ会話が始まっていない場合は何もしない

    input_cost = MODEL_PRICES['input'][st.session_state.model_name] * input_count
    output_cost = MODEL_PRICES['output'][st.session_state.model_name] * output_count
    if "gemini" in st.session_state.model_name and (input_count + output_count) > 128000:
        input_cost *= 2  # Geminiの場合、大量のトークン使用には追加料金がかかる
        output_cost *= 2

    cost = output_cost + input_cost
    st.sidebar.markdown("## Costs")
    st.sidebar.markdown(f"**Total cost: ${cost:.5f}**")
    st.sidebar.markdown(f"- Input cost: ${input_cost:.5f}")
    st.sidebar.markdown(f"- Output cost: ${output_cost:.5f}")

def main():
    # メイン関数: アプリの流れを管理
    init_page()
    init_messages()
    chain = init_chain()

    # チャット履歴の表示
    for role, message in st.session_state.get("message_history", []):
        st.chat_message(role).markdown(message)

    # ユーザーの入力を監視し、対応する応答を生成
    if user_input := st.chat_input("聞きたいことを入力してね！"):
        with st.chat_message('ai'):
            response = st.write_stream(chain.stream({"user_input": user_input}))
        st.session_state.message_history.append(("user", user_input))
        st.session_state.message_history.append(("ai", response))

    # コスト計算と表示
    calc_and_display_costs()

if __name__ == '__main__':
    main()

5-4. コードの解説

上記のコードは、Streamlitを使用して構築された対話型ウェブアプリケーションで、LangChainを活用して複数のAIモデルからの応答を管理し、それに関連するコストを計算するものです。

以下に主要な部分を詳しく説明します。

STEP 1. ライブラリのインポート

import tiktoken
import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

tiktoken: トークンの数をカウントするために使用するライブラリ。
streamlit: ウェブアプリを作成するためのライブラリ。
langchain_coreモジュール内のクラス: チャットプロンプトの生成とモデル出力の解析に使用。
langchain_openai, langchain_anthropic, langchain_google_genai: LangChainを通じてOpenAI, Anthropic, GoogleのAIモデルにアクセスするためのクラス。

STEP 2. モデルとコストの設定

MODEL_PRICES = { ... }

各AIモデルの入力と出力のコストを定義する辞書。
これはAPIリクエストの単位コストを指定しており、アプリの料金計算に使用されます。

STEP 3. ヘルパー関数

init_page(): Streamlitページの基本設定を行います（タイトル、アイコン、ヘッダー）。
init_messages(): チャット履歴を管理し、“Clear Conversation”ボタンが押された時に履歴をクリアします。
select_model(): Streamlitサイドバーでモデルと温度パラメータを選択し、選択に基づいて対応するLangChainモデルオブジェクトを返します。
init_chain(): チャットプロンプト、選択されたモデル、出力パーサーを連結して処理チェーンを構築します。
get_message_counts(): メッセージのトークン数を計算します。これは、コスト計算に不可欠です。
calc_and_display_costs(): 会話の入力と出力に基づいてコストを計算し、結果をサイドバーに表示します。

STEP 4. メインロジック (main() 関数)

ページ初期化とチャット履歴の初期化を行います。
モデルと処理チェーンの初期化を通じて、AIモデルを選択し、プロンプトを設定します。
ユーザーの入力監視: ユーザーがチャット入力を提供すると、選択されたAIモデルを使用して応答を生成し、応答をリアルタイムで表示します。
コスト計算と表示: ユーザーとAIのやりとりに基づいてコストを計算し、サイドバーに表示します。

STEP 5. スクリプト実行エントリポイント

if __name__ == '__main__':
    main()

このコードがスクリプトとして直接実行された場合に、main()関数を呼び出してアプリケーションを起動します。

6. Webページを要約するアプリ

6-1. 環境準備

pip install openai==1.29.0
pip install anthropic ==0.25.7
pip install google-generativeai ==0.5.2

pip install streamlit==1.33.0
pip install requests==2.31.0
pip install beautifulsoup4==4.12.3
pip install langchain_text_splitters==0.0.1

pip install langchain==0.1.6
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3
pip install langchain-anthropic==0.1.11
pip install langchain-google-genai==1.0.3

6-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY="◯◯◯◯◯”」「GOOGLE_API_KEY = "◯◯◯◯◯”」「ANTHROPIC_API_KEY = “◯◯◯◯◯”」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
main.py … ここにコードを入力します

💡

AIPキーが漏洩して悪用されると大金が課金される可能性があるので、ご注意ください。

6-3. 全部のコード

import traceback
import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# AIモデルをインポート
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

import requests
from bs4 import BeautifulSoup
from urllib.parse import urlparse

# 環境変数を扱うためのdotenvライブラリ。dotenvが存在しない場合に警告を表示。
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    import warnings
    warnings.warn("dotenv not found. Please make sure to set your environment variables manually.", ImportWarning)

# 要約するコンテンツを指定するプロンプトテンプレート
SUMMARIZE_PROMPT = """以下のコンテンツについて、内容を300文字程度でわかりやすく要約してください。

========

{content}

========

日本語で書いてね！
"""

def init_page():
    # Streamlitページの初期設定
    st.set_page_config(
        page_title="Website Summarizer",
        page_icon="🤗"
    )
    st.header("Website Summarizer 🤗")
    st.sidebar.title("Options")

def select_model(temperature=0):
    # モデルの選択と設定を行う
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    # 選択されたモデルに応じて対応するLangChainモデルオブジェクトを返す
    if model == "GPT-3.5":
        return ChatOpenAI(temperature=temperature, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(temperature=temperature, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(temperature=temperature, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(temperature=temperature, model="gemini-1.5-pro-latest")

def init_chain():
    # 設定したモデルとプロンプトを連結し、処理チェーンを初期化
    llm = select_model()
    prompt = ChatPromptTemplate.from_messages([
        ("user", SUMMARIZE_PROMPT),
    ])
    output_parser = StrOutputParser()
    chain = prompt | llm | output_parser
    return chain

def validate_url(url):
    """ URLが有効かどうかを判定する関数 """
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False

def get_content(url):
    # URLからコンテンツを取得し、解析する
    try:
        with st.spinner("Fetching Website ..."):
            response = requests.get(url)
            soup = BeautifulSoup(response.text, 'html.parser')
            # メインコンテンツを優先的に取得
            if soup.main:
                return soup.main.get_text()
            elif soup.article:
                return soup.article.get_text()
            else:
                return soup.body.get_text()
    except:
        st.write(traceback.format_exc())  # エラー内容を表示
        return None

def main():
    init_page()
    chain = init_chain()

    # ユーザーのURL入力を監視
    if url := st.text_input("URL: ", key="input"):
        is_valid_url = validate_url(url)
        if not is_valid_url:
            st.write('Please input valid url')
        else:
            if content := get_content(url):
                st.markdown("## Summary")
                st.write_stream(chain.stream({"content": content}))
                st.markdown("---")
                st.markdown("## Original Text")
                st.write(content)

    # コスト計算表示は別の章で説明（コメントアウトされている部分）

if __name__ == '__main__':
    main()

6-3. コードの説明

STEP 1. モデルとプロンプトの設定

SUMMARIZE_PROMPT = """以下のコンテンツについて、内容を300文字程度でわかりやすく要約してください。
{content}
日本語で書いてね！
"""

STEP 2. ヘルパー関数

init_page(), select_model(), init_chain() は、Streamlitページの設定、AIモデルの選択、処理チェーンの初期化を行います。

STEP 3. URLの検証とコンテンツ取得

def validate_url(url):
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False

与えられたURLが有効かどうかを検証します。

def get_content(url):
    try:
        with st.spinner("Fetching Website ..."):
            response = requests.get(url)
            soup = BeautifulSoup(response.text, 'html.parser')
            return soup.main.get_text() if soup.main else soup.article.get_text() if soup.article else soup.body.get_text()
    except:
        st.write(traceback.format_exc())
        return None

指定されたURLからWebページを取得し、そのテキスト内容を解析して返します。

特定のHTMLタグ (main, article, body) に基づいてテキストを抽出します。

STEP 3. メイン関数

def main():
    init_page()
    chain = init_chain()
    if url := st.text_input("URL: ", key="input"):
        if validate_url(url) and (content := get_content(url)):
            st.markdown("## Summary")
            st.write_stream(chain.stream({"content": content}))
            st.markdown("## Original Text")
            st.write(content)
        else:
            st.write('Please input valid url')

7. YouTube動画を要約するアプリ

7-1. 環境準備

pip install openai==1.29.0
pip install anthropic ==0.25.7
pip install google-generativeai ==0.5.2

pip install streamlit==1.33.0
pip install youtube-transcript-api==0.6.2
pip install pytube==15.0.0

pip install langchain==0.1.6
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3
pip install langchain-anthropic==0.1.11
pip install langchain-google-genai==1.0.3

7-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY="◯◯◯◯◯”」「GOOGLE_API_KEY = "◯◯◯◯◯”」「ANTHROPIC_API_KEY = “◯◯◯◯◯”」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
main.py … ここにコードを入力します

💡

AIPキーが漏洩して悪用されると大金が課金される可能性があるので、ご注意ください。

7-3. 全部のコード

import traceback
import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# AIモデルをインポート
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

from urllib.parse import urlparse
from langchain_community.document_loaders import YoutubeLoader  # Youtube用のドキュメントローダー

# 環境変数を管理するdotenvの利用。利用できない場合は警告を発する
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    import warnings
    warnings.warn("dotenv not found. Please make sure to set your environment variables manually.", ImportWarning)

# 要約する内容のプロンプトを定義
SUMMARIZE_PROMPT = """以下のコンテンツについて、内容を300文字程度でわかりやすく要約してください。

========

{content}

========

日本語で書いてね！
"""

def init_page():
    # Streamlitページの設定
    st.set_page_config(
        page_title="Youtube Summarizer",
        page_icon="🤗"
    )
    st.header("Youtube Summarizer 🤗")
    st.sidebar.title("Options")

def select_model(temperature=0):
    # AIモデルの選択肢を設定し、ユーザーが選ぶ
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    # 選択に応じたモデルのインスタンスを返す
    if model == "GPT-3.5":
        return ChatOpenAI(temperature=temperature, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(temperature=temperature, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(temperature=temperature, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(temperature=temperature, model="gemini-1.5-pro-latest")

def init_chain():
    # プロンプトとモデルを結合し、処理チェーンを初期化
    llm = select_model()
    prompt = ChatPromptTemplate.from_messages([("user", SUMMARIZE_PROMPT)])
    output_parser = StrOutputParser()
    chain = prompt | llm | output_parser
    return chain

def validate_url(url):
    # URLが有効かどうかを確認
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False

def get_content(url):
    # URLからコンテンツを取得し、整形して返す
    with st.spinner("Fetching Youtube ..."):
        loader = YoutubeLoader.from_youtube_url(
            url,
            add_video_info=True,  # タイトルや再生数も取得
            language=['en', 'ja']  # 字幕の言語設定
        )
        res = loader.load()
        try:
            if res:
                content = res[0].page_content
                title = res[0].metadata['title']
                return f"Title: {title}\n\n{content}"
            else:
                return None
        except:
            st.write(traceback.format_exc())
            return None

def main():
    # メイン関数：ページと処理チェーンの初期化、ユーザーの入力監視
    init_page()
    chain = init_chain()
    if url := st.text_input("URL: ", key="input"):
        is_valid_url = validate_url(url)
        if not is_valid_url:
            st.write('Please input valid url')
        else:

7-4. コードの説明

STEP 1. ライブラリのインポート

import traceback
import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

from urllib.parse import urlparse
from langchain_community.document_loaders import YoutubeLoader

STEP 2. ページの初期設定

def init_page():
    st.set_page_config(
        page_title="Youtube Summarizer",
        page_icon="🤗"
    )
    st.header("Youtube Summarizer 🤗")
    st.sidebar.title("Options")

Streamlitでのページ設定を行い、ページのタイトル、アイコン、ヘッダーを設定します。サイドバーにはオプションのタイトルが追加されます。

STEP 3. AIモデルの選択

def select_model(temperature=0):
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    if model == "GPT-3.5":
        return ChatOpenAI(temperature=temperature, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(temperature=temperature, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(temperature=temperature, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(temperature=temperature, model="gemini-1.5-pro-latest")

Streamlitのサイドバーでユーザーがモデルを選択できるように設定します。

選択されたモデルに応じて、対応するLangChain AIモデルのインスタンスを生成し、返却します。

STEP 4. 処理チェーンの初期化

def init_chain():
    llm = select_model()
    prompt = ChatPromptTemplate.from_messages([("user", SUMMARIZE_PROMPT)])
    output_parser = StrOutputParser()
    chain = prompt | llm | output_parser
    return chain

選択されたAIモデル、プロンプトテンプレート、出力パーサーを連結して、処理チェーンを構築します。これにより、入力された動画の内容を要約する処理が実行されます。

STEP 5. URLのバリデーションとコンテンツ取得

def validate_url(url):
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False

def get_content(url):
    with st.spinner("Fetching Youtube ..."):
        loader = YoutubeLoader.from_youtube_url(
            url,
            add_video_info=True,
            language=['en', 'ja']
        )
        res = loader.load()
        try:
            if res:
                content = res[0].page_content
                title = res[0].metadata['title']
                return f"Title: {title}\n\n{content}"
            else:
                return None
        except:
            st.write(traceback.format_exc())
            return None

validate_url 関数は、与えられたURLが有効かどうかを検証します。
get_content 関数は、指定されたYouTube URLから動画の内容とメタデータを取得します。取得したコンテンツは要約のために後続の処理に渡されます。

STEP 6. メイン関数

def main():
    init_page()
    chain = init_chain()
    if url := st.text_input("URL: ", key="input"):
        is_valid_url = validate_url(url)
        if not is_valid_url:
            st.write('Please input valid url')
        else:
            if content := get_content(url):
                st.markdown("## Summary")
                st.write_stream(chain.stream({"content": content}))
                st.markdown("## Original Text")
                st.write(content)

if __name__ == '__main__':
    main()

メイン関数ではページの初期設定、処理チェーンの初期化を行い、ユーザーからのURL入力を受け取ります。

URLが有効であれば、そのURLからコンテンツを取得し、AIモデルを使用して要約を生成し、結果を表示します。無効なURLが入力された場合は、エラーメッセージを表示します。

8. PDFファイルを読み込んで、Q＆Aを行う

8-1. 環境準備

pip install openai==1.29.0
pip install anthropic ==0.25.7
pip install google-generativeai ==0.5.2

pip install streamlit==1.33.0
pip install PyMuPDF==1.23.25
pip install faiss-cpu==1.7.4

pip install langchain==0.1.6
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3
pip install langchain-anthropic==0.1.11
pip install langchain-google-genai==1.0.3

8-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY="◯◯◯◯◯”」「GOOGLE_API_KEY = "◯◯◯◯◯”」「ANTHROPIC_API_KEY = “◯◯◯◯◯”」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
pdf_qa
- main.py … ここにコードを入力します
- pages
  - 1 📄 Upload PDF(s).py
  - 2 🧐 PDF QA.py

8-3. 全部のコード

main.py

import streamlit as st

def init_page():
    st.set_page_config(
        page_title="Ask My PDF(s)",
        page_icon="🧐"
    )

def main():
    init_page()

    st.sidebar.success("👆のメニューから進んでね")

    st.markdown(
    """
    ### Ask My PDF(s) にようこそ！

    - このアプリでは、アップロードしたPDFに対して質問をすることができます。
    - まずは左のメニューから `📄 Upload PDF(s)` を選択してPDFをアップロードしてください。
    - PDFをアップロードしたら `🧐 PDF QA` を選択して質問をしてみましょう😇
    """
    )

if __name__ == '__main__':
    main()

1 📄 Upload PDF(s).py

import fitz  # PyMuPDFをインポート。PDFの読み込みや操作を行うライブラリです。
import streamlit as st
from langchain_community.vectorstores import FAISS  # ベクトル検索を効率的に行うためのライブラリ
from langchain_openai import OpenAIEmbeddings  # OpenAIの埋め込みモデルを利用するためのクラス
from langchain_text_splitters import RecursiveCharacterTextSplitter  # テキストをチャンクに分割するクラス

def init_page():
    # Streamlitページの基本設定を行います。
    st.set_page_config(
        page_title="Upload PDF(s)",
        page_icon="📄"
    )
    st.sidebar.title("Options")  # サイドバーにタイトルを設置

def init_messages():
    # サイドバーに「Clear DB」というボタンを配置し、押された場合はベクトルデータベースをクリアします。
    clear_button = st.sidebar.button("Clear DB", key="clear")
    if clear_button and "vectorstore" in st.session_state:
        del st.session_state.vectorstore

def get_pdf_text():
    # Streamlitのファイルアップローダーを用いてPDFファイルをアップロードします。
    pdf_file = st.file_uploader(
        label='Upload your PDF 😇',
        type=['pdf']  # PDFファイルのみ許可
    )
    if pdf_file:
        pdf_text = ""
        with st.spinner("Loading PDF ..."):
            # PyMuPDFを使用してPDFファイルを読み込み、テキストを抽出します。
            pdf_doc = fitz.open(stream=pdf_file.read(), filetype="pdf")
            for page in pdf_doc:
                pdf_text += page.get_text()  # 各ページのテキストを追加

        # テキストを適切なサイズのチャンクに分割します。
        text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            model_name="text-embedding-3-small",
            chunk_size=500,
            chunk_overlap=0,
        )
        return text_splitter.split_text(pdf_text)  # 分割したテキストを返します
    else:
        return None

def build_vector_store(pdf_text):
    with st.spinner("Saving to vector store ..."):
        # ベクトルデータベースにテキストを追加します。セッション状態に`vectorstore`が存在しない場合は新しく作成します。
        if 'vectorstore' in st.session_state:
            st.session_state.vectorstore.add_texts(pdf_text)
        else:
            st.session_state.vectorstore = FAISS.from_texts(
                pdf_text,
                OpenAIEmbeddings(model="text-embedding-3-small")
            )

def page_pdf_upload_and_build_vector_db():
    st.title("PDF Upload 📄")
    pdf_text = get_pdf_text()
    if pdf_text:
        build_vector_store(pdf_text)

def main():
    init_page()
    page_pdf_upload_and_build_vector_db()

if __name__ == '__main__':
    main()

2 🧐 PDF QA.py

import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# models
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

def init_page():
    st.set_page_config(
        page_title="Ask My PDF(s)",
        page_icon="🧐"
    )
    st.sidebar.title("Options")

def select_model(temperature=0):
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    if model == "GPT-3.5":
        return ChatOpenAI(
            temperature=temperature,
            model_name="gpt-3.5-turbo"
        )
    elif model == "GPT-4":
        return ChatOpenAI(
            temperature=temperature,
            model_name="gpt-4o"
        )
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(
            temperature=temperature,
            model_name="claude-3-5-sonnet-20240620"
        )
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(
            temperature=temperature,
            model="gemini-1.5-pro-latest"
        )

def init_qa_chain():
    llm = select_model()
    prompt = ChatPromptTemplate.from_template("""
    以下の前提知識を用いて、ユーザーからの質問に答えてください。

    ===
    前提知識
    {context}

    ===
    ユーザーからの質問
    {question}
    """)
    retriever = st.session_state.vectorstore.as_retriever(
        # "mmr",  "similarity_score_threshold" などもある
        search_type="similarity",
        # 文書を何個取得するか (default: 4)
        search_kwargs={"k":10}
    )
    chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return chain

def page_ask_my_pdf():
    chain = init_qa_chain()

    if query := st.text_input("PDFへの質問を書いてね: ", key="input"):
        st.markdown("## Answer")
        st.write_stream(chain.stream(query))

def main():
    init_page()
    st.title("PDF QA 🧐")
    if "vectorstore" not in st.session_state:
        st.warning("まずは 📄 Upload PDF(s) からPDFファイルをアップロードしてね")
    else:
        page_ask_my_pdf()

if __name__ == '__main__':
    main()

PDFがアップロードできない時は、「--server.enableXsrfProtection false 」というコードを使うとアップロードできるようになります。

streamlit run pdf_qa/main.py --server.enableXsrfProtection false

8-4. コード「1 📄 Upload PDF(s).py」の解説

STEP 1. ライブラリのインポート

import fitz  # PyMuPDF
import streamlit as st
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

fitz (PyMuPDF): PDF ファイルを操作し、その内容を読み込むために使用します。
streamlit as st: ウェブアプリのインターフェイスを構築するためのライブラリです。
FAISS: 大規模なベクトルデータを高速に検索するためのライブラリです。
OpenAIEmbeddings: OpenAIが提供する埋め込みモデルを活用し、テキストをベクトル化するために使用します。
RecursiveCharacterTextSplitter: テキストを一定のサイズのチャンクに分割するためのユーティリティです。

STEP 2. Streamlitページの初期設定

def init_page():
    st.set_page_config(
        page_title="Upload PDF(s)",
        page_icon="📄"
    )
    st.sidebar.title("Options")

この関数は、アプリのページ設定を行います。page_title と page_icon でウェブページのタイトルとアイコンを設定し、サイドバーに「Options」というタイトルを表示します。

STEP 3. セッション状態の初期化

def init_messages():
    clear_button = st.sidebar.button("Clear DB", key="clear")
    if clear_button and "vectorstore" in st.session_state:
        del st.session_state.vectorstore

この関数は、サイドバーに「Clear DB」というボタンを設置し、押された場合にセッション状態から vectorstore オブジェクトを削除することでデータベースをクリアします。

STEP 4. PDFテキストの取得

def get_pdf_text():
    pdf_file = st.file_uploader(
        label='Upload your PDF 😇',
        type=['pdf']
    )
    if pdf_file:
        pdf_text = ""
        with st.spinner("Loading PDF ..."):
            pdf_doc = fitz.open(stream=pdf_file.read(), filetype="pdf")
            for page in pdf_doc:
                pdf_text += page.get_text()
        text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            model_name="text-embedding-3-small",
            chunk_size=500,
            chunk_overlap=0,
        )
        return text_splitter.split_text(pdf_text)
    else:
        return None

この関数では、Streamlitの file_uploader を使用してPDFファイルをアップロードします。

PDFがアップロードされると、PyMuPDFを使ってPDFを開き、各ページからテキストを抽出して結合します。

その後、RecursiveCharacterTextSplitter を用いてテキストを小さなチャンクに分割します。

STEP 5. ベクトルデータベースの構築

def build_vector_store(pdf_text):
    with st.spinner("Saving to vector store ..."):
        if 'vectorstore' in st.session_state:
            st.session_state.vectorstore.add_texts(pdf_text)
        else:
            st.session_state.vectorstore = FAISS.from_texts(
                pdf_text,
                OpenAIEmbeddings(model="text-embedding-3-small")
            )

この関数は、抽出されたテキストをベクトルデータベースに保存します。

セッション状態に vectorstore が存在する場合は、新しいテキストを既存のデータベースに追加します。

存在しない場合は新しい FAISS データベースを作成してテキストを追加します。

STEP 6. メイン機能の統合

def main():
    init_page()
    page_pdf_upload_and_build_vector_db()

if __name__ == '__main__':
    main()

main 関数では、ページの初期設定を行い、PDFファイルをアップロードしてテキストをベクトルデータベースに保存するプロセスを実行します。

これにより、ユーザーはPDFファイルをアップロードし、その内容がデータベースに保存される一連の流れをウェブインターフェイス上で行うことができます。

8-5. コード「2 🧐 PDF QA.py」の解説

STEP 1. ライブラリのインポート

import streamlit as st
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

streamlit as st: Streamlitを使ってウェブアプリを構築します。
ChatPromptTemplate, RunnablePassthrough, StrOutputParser: LangChainライブラリを使い、プロンプトの生成、入力の受け渡し、出力の解析を行います。
ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI: 異なるAIプラットフォームからのモデルを利用します。

STEP 2. Streamlitページの初期設定

def init_page():
    st.set_page_config(
        page_title="Ask My PDF(s)",
        page_icon="🧐"
    )
    st.sidebar.title("Options")

Streamlitのページ設定を行い、ページのタイトルとアイコンを設定し、サイドバーにオプションを表示します。

STEP 3. AIモデルの選択

def select_model(temperature=0):
    models = ("GPT-3.5", "GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro")
    model = st.sidebar.radio("Choose a model:", models)
    if model == "GPT-3.5":
        return ChatOpenAI(temperature=temperature, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(temperature=temperature, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(temperature=temperature, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(temperature=temperature, model="gemini-1.5-pro-latest")

サイドバーを使用してモデルを選択し、選んだモデルに応じて適切なLangChainモデルインスタンスを作成します。

STEP 4. 質問応答チェーンの初期化

def init_qa_chain():
    llm = select_model()
    prompt = ChatPromptTemplate.from_template("""
    以下の前提知識を用いて、ユーザーからの質問に答えてください。

    ===
    前提知識
    {context}

    ===
    ユーザーからの質問
    {question}
    """)
    retriever = st.session_state.vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k":10}
    )
    chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return chain

PDFの内容に基づいてユーザーからの質問に答えるために、適切なLangChain AIチェーンを設定します。

これには、文書検索、プロンプト作成、AIモデル、出力解析が含まれます。

STEP 5. ユーザーインタラクションページ

def page_ask_my_pdf():
    chain = init_qa_chain()
    if query := st.text_input("PDFへの質問を書いてね: ", key="input"):
        st.markdown("## Answer")
        st.write_stream(chain.stream(query))

ユーザーがテキスト入力を通じてPDFに対して質問を投げかけることができ、その質問に対する回答を動的に表示します。

STEP 6. メイン関数

def main():
    init_page()
    st.title("PDF QA 🧐")
    if "vectorstore" not in st.session_state:
        st.warning("まずは 📄 Upload PDF(s) からPDFファイルをアップロードしてね")
    else:
        page_ask_my_pdf()

if __name__ == '__main__':
    main()

アプリのメイン関数です。

PDFのアップロードがまだの場合は警告を表示し、アップロード後は質問ページを表示します。

9. Web検索をしてくれるAIエージェント

9-1. 環境準備

pip install openai==1.29.0
pip install anthropic ==0.25.7
pip install google-generativeai ==0.5.2

pip install streamlit==1.33.0
pip install duckduckgo-search==4.2
pip install html2text==2020.1.16
pip install 'lxml[html_clean]'==5.2.1
pip install readability-lxml==0.8.1
pip install langsmith==0.1.54

pip install langchain==0.1.16
pip install langchain-community==0.0.34
pip install langchain-core==0.1.46
pip install langchain-openai==0.1.3
pip install langchain-anthropic==0.1.11
pip install langchain-google-genai==1.0.0

9-2. ファイル構成

.streamlit
- secrets.toml … ここに「OPENAI_API_KEY="◯◯◯◯◯”」「GOOGLE_API_KEY = "◯◯◯◯◯”」「ANTHROPIC_API_KEY = “◯◯◯◯◯”」という形で各モデルのAPIキーを収納ください。
.gitignore … ここに「.streamlit/」と入力して、APIキーがクラウド上にアップされないようにします
main.py
tools
- fetch_page.py
- search_ddg.py

💡

AIPキーが漏洩して悪用されると大金が課金される可能性があるので、ご注意ください。

9-3. 全部のコード

main.py

import streamlit as st  # Streamlitライブラリのインポート
from langchain.agents import create_tool_calling_agent, AgentExecutor  # LangChainのエージェント機能をインポート
from langchain.memory import ConversationBufferWindowMemory  # 会話のメモリ管理用
from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate  # プロンプトテンプレートの構成用
from langchain_core.runnables import RunnableConfig  # 実行設定の構成用
from langchain_community.callbacks import StreamlitCallbackHandler  # Streamlitでのコールバック処理用

# 使用するモデルのインポート
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# カスタムツールのインポート
from tools.search_ddg import search_ddg  # DuckDuckGoを使った検索機能
from tools.fetch_page import fetch_page  # Webページの取得機能

# 環境変数の読み込み。dotenvライブラリがない場合は警告を表示
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    import warnings
    warnings.warn("dotenv not found. Please make sure to set your environment variables manually.", ImportWarning)

# システムプロンプトの定義。エージェントの動作説明と指示
CUSTOM_SYSTEM_PROMPT = """
あなたは、ユーザーのリクエストに基づいてインターネットで調べ物を行うアシスタントです。
利用可能なツールを使用して、調査した情報を説明してください。
既に知っていることだけに基づいて答えないでください。回答する前にできる限り検索を行ってください。
(ユーザーが読むページを指定するなど、特別な場合は、検索する必要はありません。)

検索結果ページを見ただけでは情報があまりないと思われる場合は、次の2つのオプションを検討して試してみてください。
- 検索結果のリンクをクリックして、各ページのコンテンツにアクセスし、読んでみてください。
- 1ページが長すぎる場合は、3回以上ページ送りしないでください（メモリの負荷がかかるため）。
- 検索クエリを変更して、新しい検索を実行してください。
- 検索する内容に応じて検索に利用する言語を適切に変更してください。例えば、プログラミング関連の質問については英語で検索するのがいいでしょう。

ユーザーは非常に忙しく、あなたほど自由ではありません。
そのため、ユーザーの労力を節約するために、直接的な回答を提供してください。

=== 悪い回答の例 ===
- これらのページを参照してください。
- これらのページを参照してコードを書くことができます。
- 次のページが役立つでしょう。

=== 良い回答の例 ===
- これはサンプルコードです。 -- サンプルコードをここに --
- あなたの質問の答えは -- 回答をここに --

回答の最後には、参照したページのURLを**必ず**記載してください。（これにより、ユーザーは回答を検証することができます）

ユーザーが使用している言語で回答するようにしてください。
ユーザーが日本語で質問した場合は、日本語で回答してください。ユーザーがスペイン語で質問した場合は、スペイン語で回答してください。
"""

def init_page():
    # Streamlitページの基本設定
    st.set_page_config(
        page_title="Web Browsing Agent",  # ページタイトル
        page_icon="🤗"  # ページアイコン
    )
    st.header("Web Browsing Agent 🤗")  # ヘッダーの設定
    st.sidebar.title("Options")  # サイドバーのタイトル

def init_messages():
    # 初期メッセージの設定と会話のクリア機能
    clear_button = st.sidebar.button("Clear Conversation", key="clear")  # 会話をクリアするボタン
    if clear_button or "messages" not in st.session_state:
        # 会話をクリアまたは初期化
        st.session_state.messages = [
            {"role": "assistant", "content": "こんにちは！なんでも質問をどうぞ！"}
        ]
        st.session_state['memory'] = ConversationBufferWindowMemory(
            return_messages=True,  # メッセージを返す設定
            memory_key="chat_history",  # メモリキーの設定
            k=10  # 保持するメッセージ数
        )

def select_model():
    # 使用するモデルの選択
    models = ("GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro", "GPT-3.5 (not recommended)")
    model = st.sidebar.radio("Choose a model:", models)  # ラジオボタンでモデル選択
    if model == "GPT-3.5 (not recommended)":
        return ChatOpenAI(
            temperature=0, model_name="gpt-3.5-turbo")
    elif model == "GPT-4":
        return ChatOpenAI(
            temperature=0, model_name="gpt-4o")
    elif model == "Claude 3.5 Sonnet":
        return ChatAnthropic(
            temperature=0, model_name="claude-3-5-sonnet-20240620")
    elif model == "Gemini 1.5 Pro":
        return ChatGoogleGenerativeAI(
            temperature=0, model="gemini-1.5-pro-latest")

def create_agent():
    # エージェントの作成
    tools = [search_ddg, fetch_page]  # 使用するツールのリスト
    prompt = ChatPromptTemplate.from_messages([
        ("system", CUSTOM_SYSTEM_PROMPT),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
    llm = select_model()  # モデルの選択
    agent = create_tool_calling_agent(llm, tools, prompt)  # エージェントの作成
    return AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        memory=st.session_state['memory']
    )

def main():
    # メイン関数
    init_page()
    init_messages()
    web_browsing_agent = create_agent()  # エージェントの作成

    for msg in st.session_state['memory'].chat_memory.messages:
        st.chat_message(msg.type).write(msg.content)

    if prompt := st.chat_input(placeholder="2023 FIFA 女子ワールドカップの優勝国は？"):
        st.chat_message("user").write(prompt)

        with st.chat_message("assistant"):
            # コールバック関数の設定 (エージェントの動作の可視化用)
            st_cb = StreamlitCallbackHandler(
                st.container(), expand_new_thoughts=True)

            # エージェントを実行
            response = web_browsing_agent.invoke(
                {'input': prompt},
                config=RunnableConfig({'callbacks': [st_cb]})
            )
            st.write(response["output"])

if __name__ == '__main__':
    main()

tools/fetch_page.py


import requests  # HTTPリクエストを扱うためのライブラリ
import html2text  # HTMLをプレーンテキストに変換するライブラリ
from readability import Document  # HTMLドキュメントを読みやすくするライブラリ
from langchain_core.tools import tool  # LangChainのデコレーター、ツールとして関数を登録
from langchain_core.pydantic_v1 import BaseModel, Field  # Pydanticを用いたデータ検証
from langchain_text_splitters import RecursiveCharacterTextSplitter  # テキストをチャンクに分割するためのツール


class FetchPageInput(BaseModel):
    url: str = Field()  # 取得するウェブページのURL
    page_num: int = Field(0, ge=0)  # 取得するページ番号、0からスタート

@tool(args_schema=FetchPageInput)
def fetch_page(url, page_num=0, timeout_sec=10):
    """
    指定されたURLからウェブページのコンテンツを取得するツール。
    取得したコンテンツは、タイトルと本文に分けて返されます。
    返されるデータは、状態コード、ページ内容（タイトル、内容、次ページ有無）を含みます。

    Parameters
    ----------
    url : str
        取得するページのURL。
    page_num : int, optional
        取得するページ番号、デフォルトは0。
    timeout_sec : int, optional
        リクエストのタイムアウト秒数、デフォルトは10秒。

    Returns
    -------
    Dict[str, Any]:
        - status : int
            HTTPステータスコードまたは内部エラーコード。
        - page_content : dict
            - title : str
                ページのタイトル。
            - content : str
                ページの内容（指定されたページ番号のチャンク）。
            - has_next : bool
                次のページ番号が存在するかどうか。
    """
    try:
        response = requests.get(url, timeout=timeout_sec)  # URLからコンテンツを取得
        response.encoding = 'utf-8'  # エンコーディングをUTF-8に設定
    except requests.exceptions.Timeout:
        return {
            "status": 500,
            "page_content": {'error_message': 'タイムアウトによりページが取得できませんでした。他のページを試してください。'}
        }

    if response.status_code != 200:
        return {
            "status": response.status_code,
            "page_content": {'error_message': 'ページが取得できませんでした。他のページを試してください。'}
        }
    
    try:
        doc = Document(response.text)  # レスポンスからDocumentオブジェクトを生成
        title = doc.title()  # ページのタイトルを取得
        html_content = doc.summary()  # 読みやすい要約を生成
        content = html2text.html2text(html_content)  # HTMLをテキストに変換
    except:
        return {
            "status": 500,
            "page_content": {'error_message': 'ページの解析に失敗しました。他のページを試してください。'}
        }

tools/search_ddg.py


from itertools import islice  # 部分的なイテレーションに使用されるモジュール
from duckduckgo_search import DDGS  # DuckDuckGo検索APIのラッパーライブラリ
from langchain_core.tools import tool  # LangChainのツールデコレータを使用
from langchain_core.pydantic_v1 import BaseModel, Field  # Pydanticを使用したデータモデルとフィールド定義

# DuckDuckGoの検索結果のサンプル
"""
Sample Response of DuckDuckGo python library
--------------------------------------------
[
    {
        'title': '日程・結果｜Fifa 女子ワールドカップ オーストラリア&ニュージーランド 2023｜なでしこジャパン｜日本代表｜Jfa｜日本サッカー協会',
        'href': 'https://www.jfa.jp/nadeshikojapan/womensworldcup2023/schedule_result/',
        'body': '日程・結果｜FIFA 女子ワールドカップ オーストラリア&ニュージーランド 2023｜なでしこジャパン｜日本代表｜JFA｜日本サッカー協会. FIFA 女子ワールドカップ. オーストラリア&ニュージーランド 2023.'
    }, ...
]
"""

class SearchDDGInput(BaseModel):
    query: str = Field(description="検索したいキーワードを入力してください")  # 検索クエリの入力を定義

@tool(args_schema=SearchDDGInput)  # LangChainのツールとしてこの関数を登録
def search_ddg(query, max_result_num=5):
    """
    DuckDuckGo検索を行うツールです。
    指定されたクエリでDuckDuckGoを使ってウェブ検索を行い、結果を返します。
    返される情報には、ページのタイトル、概要（スニペット）、そしてURLが含まれます。

    Parameters
    ----------
    query : str
        検索を行うためのクエリ文字列。
    max_result_num : int
        返される結果の最大数。

    Returns
    -------
    List[Dict[str, str]]:
        検索結果のリスト。各辞書オブジェクトには以下が含まれます。
        - title: タイトル
        - snippet: ページの概要
        - url: ページのURL

    この関数は、プログラミングに関連する質問など、特定の質問に最適な言語で検索を行うことを推奨します。
    また、検索結果だけでは十分でない場合は、実際のページ内容を取得するために追加のツールを使用することをお勧めします。
    """
    # DuckDuckGo検索の実行
    res = DDGS().text(query, region='wt-wt', safesearch='off', backend="lite")
    # 最大結果数までの結果を返す
    return [
        {
            "title": r.get('title', ""),  # タイトルの取得、デフォルトは空文字
            "snippet": r.get('body', ""),  # スニペットの取得、デフォルトは空文字
            "url": r.get('href', "")  # URLの取得、デフォルトは空文字
        }
        for r in islice(res, max_result_num)  # 指定された数の結果だけをイテレート
    ]

9-4. コード「main.py」の解説

STEP 1. ライブラリのインポート

streamlit: Webアプリを構築するためのライブラリ。
langchain: 人工知能による会話エージェントやツール呼び出し機能を提供するライブラリ。

STEP 2. システムプロンプトの定義

CUSTOM_SYSTEM_PROMPT = """..."""

この長い文字列はエージェントの振る舞いを定義するもので、ユーザーからのリクエストに基づいて情報を検索し、直接的な回答を提供するよう指示します。

また、検索結果が不十分な場合の対処法や、どの言語で検索すべきかについてのガイドラインも含まれています。

STEP 3. 初期設定関数

init_page … Streamlitでページの設定を行います。タイトルやアイコンの設定を含む基本的なページ設定を行います。

init_messages … 会話を管理するためのメッセージとメモリの初期化を行います。ユーザーが「会話をクリア」ボタンを押すと、会話の履歴がリセットされます。

STEP 4. モデル選択関数 select_model

models = ("GPT-4", "Claude 3.5 Sonnet", "Gemini 1.5 Pro", "GPT-3.5 (not recommended)")
model = st.sidebar.radio("Choose a model:", models)

STEP 5. エージェント生成関数 create_agent

この関数はカスタムツールと選択したモデルを組み合わせてエージェントを生成します。

def create_agent():
    # 使用するツールのリスト
    tools = [search_ddg, fetch_page]
    
    # チャットプロンプトテンプレートの設定
    prompt = ChatPromptTemplate.from_messages([
        ("system", CUSTOM_SYSTEM_PROMPT),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
    
    # モデルの選択
    llm = select_model()
    
    # エージェントの作成
    agent = create_tool_calling_agent(llm, tools, prompt)
    
    # エージェントの実行環境を設定
    return AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        memory=st.session_state['memory']
    )

エージェントの構成要素

ツールの設定: search_ddg と fetch_page はカスタムツールで、それぞれDuckDuckGoを使ったウェブ検索とウェブページの取得機能を提供します。これらはエージェントがクエリに応じて情報を収集する際に使用されます。
プロンプトテンプレートの利用: ChatPromptTemplate はエージェントがユーザーとの対話を管理する際に使用するテンプレートです。このテンプレートには、エージェントの説明、過去のメッセージ履歴、ユーザーからの現在の入力が含まれます。これにより、エージェントはコンテキストを保持しつつ対話を進めることができます。
モデルの選択: select_model 関数によって選ばれたモデル（例えばGPT-4やClaude 3.5 Sonnetなど）がエージェントの核となります。このモデルはユーザーからの入力に基づいて適切な回答や情報を生成します。

エージェントの実行環境

生成されたエージェントは AgentExecutor によって実行管理されます。
これにはツールの実行、対話の管理、メモリの利用状態の追跡が含まれ、エージェントの振る舞いが具体的に制御されます。

STEP 5. メイン関数 main

この関数はアプリケーションの主要な流れを制御し、Streamlitのウェブインターフェースを設定し、ユーザー入力に基づいてエージェントを実行します。

def main():
    # Streamlitページの基本設定
    init_page()
    
    # 初期メッセージの設定と会話メモリの初期化
    init_messages()
    
    # エージェントの生成
    web_browsing_agent = create_agent()
    
    # Streamlitのチャットインターフェースにメッセージを表示
    for msg in st.session_state['memory'].chat_memory.messages:
        st.chat_message(msg.type).write(msg.content)
    
    # ユーザーからの入力を受け取る
    if prompt := st.chat_input(placeholder="2023 FIFA 女子ワールドカップの優勝国は？"):
        st.chat_message("user").write(prompt)
        
        with st.chat_message("assistant"):
            # エージェントの動作の可視化用コールバック
            st_cb = StreamlitCallbackHandler(
                st.container(), expand_new_thoughts=True)
            
            # エージェントの実行と結果の表示
            response = web_browsing_agent.invoke(
                {'input': prompt},
                config=RunnableConfig({'callbacks': [st_cb]})
            )
            st.write(response["output"])

if __name__ == '__main__':
    main()

インターフェースの設定

ページ設定: Streamlitの set_page_config でページの基本設定を行います。
メッセージの初期化: init_messages 関数を呼び出して、会話の初期状態やメモリを設定します。

エージェントの動作

エージェントの生成: create_agent を呼び出し、ユーザーのインタラクションに応じて動作するエージェントを準備します。
ユーザー入力の受付: Streamlitの chat_input を使用してユーザーからのクエリを受け取ります。
エージェントの実行と結果の表示: ユーザーからの入力をエージェントに渡し、AgentExecutor で処理させます。StreamlitCallbackHandler を通じてエージェントの処理過程をリアルタイムでユーザーに可視化し、最終的な回答を表示します。

この設計により、ユーザーは直感的なインターフェースを通じて複雑な情報検索タスクを効率的に行うことができ、エージェントは背後で動的にデータを処理し、適切な回答を提供します。

9-5. コード「tools/fetch_page.py」の解説

このコードはウェブページのコンテンツを取得し、それを処理してテキスト形式で出力するための関数 fetch_page を定義しています。以下に各部分の詳細な説明を行います。

コードの構成

ライブラリのインポート

requests: HTTPリクエストを行うためのライブラリ。ウェブからデータを取得する際に使用します。
html2text: HTMLコンテンツをプレーンテキストに変換するライブラリ。ウェブページのHTMLを読みやすい形式に変換します。
readability: HTMLドキュメントを解析し、読みやすくするためのライブラリ。主要なコンテンツのみを抽出する機能を提供します。
tool: LangChainライブラリのデコレーターで、関数を特定のツールとして登録します。
BaseModel, Field: Pydanticライブラリを用いたデータモデル定義。入力データの検証に使用します。
RecursiveCharacterTextSplitter: テキストを一定のサイズに分割するツール（このコードでは実際には使われていませんが、テキスト処理に使うことができます）。

入力データモデルの定義 (FetchPageInput)

url: 取得するウェブページのURLを指定するためのフィールド。
page_num: ページ分割されているコンテンツの特定のページ番号を指定するフィールド。初期値は0で、非負の整数を受け入れるように設定されています。

関数の定義 (fetch_page)

この関数はtoolデコレーターによりLangChainのエージェントが使用可能なツールとして登録されています。
引数にはurlとpage_num、timeout_sec（タイムアウト秒数、デフォルトは10秒）があります。

処理の流れ

HTTPリクエストの実行

requests.getを使用して指定されたurlからウェブページのデータを取得します。timeout_secによりリクエストのタイムアウト時間を制限します。

エラーハンドリング

タイムアウトやその他のリクエストエラー（HTTPステータスコードが200以外）が発生した場合、適切なエラーメッセージとともにステータスコードを返します。

コンテンツの解析と変換

readability.Documentを使用してHTMLから主要なコンテンツを抽出し、html2textでHTMLをプレーンテキストに変換します。
解析中にエラーが発生した場合は、それに対応するエラーメッセージを含むレスポンスを返します。

この関数は、ウェブページのデータを取得し、主要なテキストコンテンツを抽出して整形するための処理を行います。

これにより、ウェブページからの情報をさまざまなアプリケーションで使用しやすくなります。

9-6. コード「tools/search_ddg.py」の解説

このコードは、DuckDuckGoの検索APIを活用してウェブ検索を行い、結果を取得するツールを定義しています。各部分の詳細な解説を以下に行います。

コードの主要な部分

ライブラリのインポート

itertools.islice: イテラブル（反復可能オブジェクト）から指定された数の要素を効率的に抽出するための関数です。
duckduckgo_search.DDGS: DuckDuckGoの検索APIをラップしてPythonで使いやすくするライブラリです。
langchain_core.tools.tool: LangChainフレームワークの一部で、特定の関数をエージェントが利用可能なツールとして登録するデコレーターです。
langchain_core.pydantic_v1.BaseModel, Field: Pydanticライブラリを使用して入力データのバリデーションを行うためのクラスとフィールド定義です。

入力データモデルの定義

SearchDDGInput: 検索を行うためのクエリ文字列を受け取るためのデータモデルです。このクラスはPydanticの BaseModel を継承しており、検索クエリを定義するフィールドがあります。

検索ツールの定義

@tool(args_schema=SearchDDGInput): このデコレータは、関数 search_ddg をLangChainのツールとして登録し、引数の構造を SearchDDGInput クラスに基づいてバリデーションを行うよう指定しています。
search_ddg: この関数はDuckDuckGoを使用してウェブ検索を行い、検索結果を整形して返します。検索クエリとして文字列を受け取り、結果の最大数も指定できます。

検索処理の流れ

検索の実行: DDGS().text() メソッドを使用して、指定されたクエリでDuckDuckGoを通じて検索を行います。検索パラメータとして region や safesearch, backend を指定しています。
検索結果の処理: 検索結果は res に格納され、それを islice でイテレートして指定された数の結果を抽出します。各結果は辞書形式で、タイトル (title), スニペット (snippet), そして URL (url) を含んでいます。これらはリスト内包表記を使用してリストに変換され、関数の戻り値として返されます。

用途と推奨

このツールは、特にLangChainエージェントに組み込んで使用されることを想定しています。

エージェントはこのツールを呼び出してクエリに応じた情報をウェブから取得し、ユーザーに対して直接的な回答を提供する際に使用します。

プログラミングや特定の専門知識が必要なクエリに対しては、最適な言語で検索を行うことが推奨されています。

また、検索結果だけでは不十分な場合、追加のツールを用いてページ内容を詳細に取得することも提案されています。

まとめ

LangChainを活用することは、多方面にわたる長期的な利点を提供します。

このツールは、強力なAIモデルを使って複雑な言語処理タスクを簡単に、効率的に行う能力を提供します。

LangChainを使うことで、開発者は時間を節約し、リソースをより創造的な取り組みに集中させることができます。また、LangChainはスケーラビリティと柔軟性を兼ね備えているため、小規模プロジェクトから大規模な企業用アプリケーションまで、あらゆる規模のニーズに応じてカスタマイズ可能です。

さらに、LangChainを用いることで、AIの最新進歩を容易に統合し、進化し続けるテクノロジーの波に乗ることができます。これにより、ユーザーエクスペリエンスを向上させ、ビジネスの競争力を保つことが可能になります。

例えば、カスタマーサポートの自動化、リアルタイムのデータ分析、パーソナライズされたコンテンツ生成など、LangChainの応用は無限大です。

このブログ記事を通じて、LangChainの基本的な使い方から具体的な応用例に至るまでを解説しましたが、これはほんの入り口に過ぎません。

ぜひLangChainを実際に使ってみることで、その真価を体験し、自身のプロジェクトやビジネスにどのように役立てるかを探ってみてください。

技術の力を最大限に活用し、新しい創造的な解決策を開発することが、これからの時代を生き抜く鍵となるでしょう。