LangChainは、大規模な言語モデルを使用したアプリケーションの作成を簡素化するためのフレームワークです。言語モデル統合フレームワークとして、LangChainの使用ケースは、文書の分析や要約、チャットボット、コード分析を含む、言語モデルの一般的な用途と大いに重なっています。

LangChainは、PythonとJavaScriptの2つのプログラミング言語に対応しています。LangChainを使って作られているアプリケーションには、AutoGPT、LaMDA、CodeAnalyzerなどがあります。

AutoGPTは、文章生成、翻訳、コード生成などの機能を持つアプリケーションです。
LaMDAは、対話や文章生成を行うチャットボットです。
CodeAnalyzerは、コードを分析するアプリケーションです。

記事概要

この記事では、LangChainでのLLMについて紹介します。LangChainのオフィシャルサイトは以下の通りです。

LLMは以下の真ん中のPredictの部分です。

LangChainは、OpenAIやAzure OpenAIなどでのモデルを提供するのではなく、これらモデルを利用するため、共通のインタフェースを提供しています。

2種類のモデルのためのインターフェイスを提供しています：

LLMs: テキスト文字列を入力として受け取り、テキスト文字列を返すモデル
チャットモデル: 言語モデルに基づいているが、入力としてチャットメッセージのリストを受け取り、チャットメッセージを返すモデル

はじめに

from langchain.llms import OpenAI
llm = OpenAI()

該当クラス内に、以下の関数があるので、

__call__: string in -> string out

直接呼び出すことができます：

llm("面白いジョックを教えてください。")

出力結果：

'\n\nQ.なぜキツネはイルカを見るのがきらいなの？\nA.なぜなら、イルカは「わんわん！」と言っているから、キツネは「おかしい！」と思うからです！'

また、generateを利用して、複数回の呼び出しが可能です。

llm_result = llm.generate(["日本について面白いことを教えてください。", "美しい漢詩を作ってください"]*15)

出力結果が30です。

len(llm_result.generations)

print(llm_result.generations[0])

出力結果が：

[Generation(text='\n\n日本は世界有数の文化が息づく国であり、古くから続く伝統的な文化や食文化などを楽しむことができます。また、折り紙、茶道、お稽古などの伝統的な文化もあります。また、近年ではアニメやマンガなどのポップカルチャーも世界的に非常に人気があります。', generation_info={'finish_reason': 'stop', 'logprobs': None})]

次のは：

print(llm_result.generations[1])

出力結果が：

[Generation(text='\n\n春空に銀の虹架け\n憧れの光に想い抱け\n静かな風に心揺らす\n色鮮やかな世界を抱く', generation_info={'finish_reason': 'stop', 'logprobs': None})]

また、この呼び出しのToken数なども出力できます：

llm_result.llm_output

出力結果は：

{'token_usage': {'total_tokens': 4236,
  'prompt_tokens': 435,
  'completion_tokens': 3801},
 'model_name': 'text-davinci-003'}

非同期API

LangChainは、asyncioライブラリを活用して、LLMのための非同期サポートを提供しています。

非同期サポートは、特に複数のLLMを同時に呼び出す場面で非常に役立ちます。これらの呼び出しはネットワークに依存しているためです。現在、OpenAI、PromptLayerOpenAI、ChatOpenAI、Anthropic、Cohereがサポートされています。

import time  # 時間計測
import asyncio  # 非同期処理用

from langchain.llms import OpenAI  # langchain.llmsライブラリからOpenAIクラスをインポート

# 同期方法で文字列を生成する関数を定義
def generate_serially():
    llm = OpenAI(temperature=0.9)  
    for _ in range(10):  
        resp = llm.generate(["Hello, how are you?"])  # 同期でgenerate方法で文字列を作成
        print(resp.generations[0][0].text)  

# 非同期方法で文字列を生成する関数を定義
async def async_generate(llm):
    resp = await llm.agenerate(["Hello, how are you?"])  # 非同期でgenerate方法で文字列を作成
    print(resp.generations[0][0].text)  

# 非同期方法で文字列を生成する関数を定義
async def generate_concurrently():
    llm = OpenAI(temperature=0.9)  
    tasks = [async_generate(llm) for _ in range(10)]  # 非同期タスクを10個作る
    await asyncio.gather(*tasks)  # asyncio.gatherを利用して全ての非同期タスクの完了を待つ

非同期方法：

s = time.perf_counter()
# 非同期方法で文字列作成を実施
# Jupyter以外の場合はasyncio.run(generate_concurrently())を利用
await generate_concurrently()
# 実施時間を計測
elapsed = time.perf_counter() - s
print("\033[1m" + f"Concurrent executed in {elapsed:0.2f} seconds." + "\033[0m")

非同期方法で、合計4.8秒がかかりました。

同期方法：

s = time.perf_counter()
# 同期方法で文字列作成を実施
generate_serially()
# 実施時間を計測
elapsed = time.perf_counter() - s
print("\033[1m" + f"Serial executed in {elapsed:0.2f} seconds." + "\033[0m")

同期方法で、合計8.04秒がかかりました。

LLMのカスタマイズ

LangChainで独自のLLMを使用したい場合に、カスタムLLMラッパーを作成することができます。カスタムLLMが実装する必要があるものは1つだけです：

_callメソッド: 文字列を受け取り、オプションでストップワードを指定して、文字列を返す_callメソッド。

#入力の最初のN文字だけを返す非常にシンプルなカスタムLLMを実装しましょう。
from typing import Any, List, Mapping, Optional
from langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.llms.base import LLM

#このクラス CustomLLM は、LLM クラスを継承し、新しいクラス変数 n を追加しています。
#_llm_type と _identifying_params の2つの property デコレータ付きのメソッドがあり、これらのメソッドはいくつかの固定の属性値を返します。
#_call メソッドは、入力された prompt 文字列を処理し、最初の n 文字を返します。stop パラメータが提供された場合、例外が発生します。
# LLM から継承した CustomLLM クラス
class CustomLLM1(LLM):

    # クラス変数、整数を示す
    n: int

    # _llm_type の値を取得するための属性デコレータ
    @property
    def _llm_type(self) -> str:
        # _llm_type の値として "custom" 文字列を返す
        return "custom"

    # ある操作を処理するための _call メソッド
    def _call(
        self,
        prompt: str,  # 入力されたプロンプト文字列
        stop: Optional[List[str]] = None,  # オプションの停止文字列のリスト、デフォルトは None
        run_manager: Optional[CallbackManagerForLLMRun] = None,  # オプションのコールバックマネージャ、デフォルトは None
    ) -> str:
        # stop パラメータが None でない場合、ValueError 例外を発生させる
        if stop is not None:
            raise ValueError("stop kwargsは許可されていません。")
        # prompt 文字列の最初の n 文字を返す
        return prompt[: self.n]

    # _identifying_params の値を取得するための属性デコレータ
    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        # このメソッドのドキュメント文字列、このメソッドの機能は識別パラメータを取得することを示す
        """識別パラメータを取得する。"""
        # n の値を含む辞書を返す
        return {"n": self.n}

llm = CustomLLM1(n=10)

実際にこのカスタマイズしたLLMを利用してみる

llm("This is a foobar thing")

出力結果が：

'This is a '

print(llm)

出力結果（@propertyでの内容）：

CustomLLM1
Params: {'n': 10}

フェイクのLLM

LangChainはテスト用に使用できるフェイクのLLMクラスを提供しています。これにより、LLMへの呼び出しをモックアウトし、LLMが特定の方法で応答した場合に何が起こるかをシミュレートできます。

特にテストする際に、LLM利用料金の節約が期待できます。

# langchain.llms.fakeモジュールからFakeListLLMクラスをインポートする。このクラスは何らかの動作を模倣または偽造するために使用される可能性がある
from langchain.llms.fake import FakeListLLM

# langchain.agentsモジュールからload_tools、initialize_agent、AgentTypeをインポートする
from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain.agents import AgentType

# "python_repl"という名前のツールをロードするためにload_tools関数を呼び出す
tools = load_tools(["python_repl"])

# LLMの期待されるレスポンスを模倣するレスポンスのリストを定義する
responses = ["Action: Python REPL\nAction Input: print(2 + 2)", "Final Answer: 6"]

# 上で定義されたresponsesを使用してFakeListLLMオブジェクトを初期化する
llm = FakeListLLM(responses=responses)

# initialize_agent関数を呼び出し、上記のtoolsとllm、指定されたエージェントタイプとverboseパラメータを使用してエージェントを初期化する
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)

# エージェントのrunメソッドを呼び出し、"whats 2 + 2"という文字列を入力として渡し、エージェントに2プラス2の結果を尋ねる
agent.run("お名前は何ですか")

上記のagentがフェイクのLLMなので、固定の結果'6'になります。

[1m> Entering new AgentExecutor chain...[0m
[32;1m[1;3mAction: Python REPL
Action Input: print(2 + 2)[0m
Observation: Python REPL is not a valid tool, try one of [Python_REPL].
Thought:[32;1m[1;3mFinal Answer: 6[0m

[1m> Finished chain.[0m

'6'

キャッシング

以下の二つ理由で、LangChainはLLMにオプションのキャッシュ層を提供しています。

同じ完成を何度もリクエストする場合、LLMプロバイダーへのAPI呼び出しの回数を減らすことでお金を節約できます。
LLMプロバイダーへのAPI呼び出しの回数を減らすことで、アプリケーションの速度を上げることができます。

メモリキャッシュ内利用

# メモリキャッシュ内で
import langchain
from langchain.llms import OpenAI
import time
# キャッシングを明確にするため、遅いモデルを使用しましょう。
llm = OpenAI(model_name="text-davinci-002", n=2, best_of=2)
from langchain.cache import InMemoryCache
langchain.llm_cache = InMemoryCache()

start_time = time.time()  # 開始時間を記録
# 初めての場合、キャッシュにはないので、時間がかかるはずです
print(llm.predict("冗談を言って"))
end_time = time.time()  # 終了時間を記録
elapsed_time = end_time - start_time  # 合計時間を計算
print(f"Predictメソッドは{elapsed_time:.4f}秒かかりました。")

出力結果で：Predictメソッドは3.2378秒かかりました。

start_time = time.time()  # 開始時間を記録
# 2回目なので、速くなります
print(llm.predict("冗談を言って"))
end_time = time.time()  # 終了時間を記録
elapsed_time = end_time - start_time  # 合計時間を計算
print(f"Predictメソッドは{elapsed_time:.4f}秒かかりました。")

出力結果で、Predictメソッドは0.0000秒かかりました。

かなり早くなりました。

SQLiteデータベースキャッシュを使用する

import langchain
from langchain.llms import OpenAI
import time
from langchain.llms.base import LLM

# SQLiteデータベースキャッシュを使用する
# SQLiteキャッシュで同じことができます
from langchain.cache import SQLiteCache
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

start_time = time.time()  # 開始時間を記録
# 最初の時はキャッシュにないので、時間がかかるはずです
print(llm.predict("日本語でクレヨンしんちゃんの面白いことを言ってください"))
end_time = time.time()  # 終了時間を記録
elapsed_time = end_time - start_time  # 合計時間を計算
print(f"Predictメソッドは{elapsed_time:.4f}秒かかりました。")

初回なので、Predictメソッドは3.1467秒かかりました。

次の回では、早くなります。

start_time = time.time()  # 開始時刻を記録
# 2回目なので、速くなる
print(llm.predict("日本語でクレヨンしんちゃんの面白いことを言ってください"))
end_time = time.time()  # 終了時刻を記録
elapsed_time = end_time - start_time  # 合計時間を計算
print(f"Predictメソッドの実行には{elapsed_time:.4f}秒かかりました。")

今回は：Predictメソッドの実行には0.0020秒かかりました。

シリアライゼーション

ディスクへのLLM設定の書き込みと読み込みの方法を説明しています。これは、特定のLLMの設定（例：プロバイダ、温度など）を保存したい場合に便利です。

from langchain.llms import OpenAI
from langchain.llms.loading import load_llm
llm = load_llm("llmstore/llm.json")

llm.json

    {
        "model_name": "text-davinci-003",
        "temperature": 0.7,
        "max_tokens": 256,
        "top_p": 1.0,
        "frequency_penalty": 0.0,
        "presence_penalty": 0.0,
        "n": 1,
        "best_of": 1,
        "request_timeout": null,
        "_type": "openai"
    }

取り出し：

llm

出力結果が：

OpenAI(client=<class 'openai.api_resources.completion.Completion'>, openai_api_key='自身のAPIキー', openai_api_base='', openai_organization='', openai_proxy='')

保存

現在のLLMもファイルに保存することができます。jsonとyamlどちらでもOKです。

llm.save("llmsave.json")
llm.save("llmsave.yaml")

ストリーミング

一部のLLMはストリーミングレスポンスを提供し、完全なレスポンスを待たずにすぐに処理できます。OpenAIやChatOpenAIなどの多くのLLM実装でストリーミングがサポートされており、その機能を活用するための方法として、on_llm_new_tokenを実装したCallbackHandlerを使用することが示されています。

from langchain.llms import OpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler


llm = OpenAI(streaming=True, callbacks=[StreamingStdOutCallbackHandler()], temperature=0)
resp = llm("流れている水について、歌詞を作ってください")

これで、出力結果が一気に出されたではなく、ChatGPTでの出力と同じように完全な出力でなくても、一部づつ表示されるようになりました。

流れる水は 心を洗い流す
涙を拭いて 心を癒す
清らかな水が 流れる様に
心も洗い流して 新しい日を迎える

最終結果のみを受け取ることも可能です。

resp = llm.generate(["流れている水について、歌詞を作ってください"])
print(resp)

出力結果：

generations=[[Generation(text='\n\n流れる水は 心を洗い流す\n涙を拭いて 心を癒す\n清らかな水が 流れる様に\n心も洗い流して 新しい日を迎える', generation_info={'finish_reason': 'stop', 'logprobs': None})]] llm_output={} run=None

トークン使用状況の追跡

特定の呼び出しに対するトークンの使用を追跡する方法を解説しています。現在、OpenAI APIにのみ実装されています。

まず、単一のLLM呼び出しに対するトークン使用の追跡の非常にシンプルな例を見てみましょう。

from langchain.llms import OpenAI
from langchain.callbacks import get_openai_callback
llm = OpenAI(model_name="text-davinci-002", n=2, best_of=2,cache = None)

with get_openai_callback() as cb:
    result = llm("面白いことを教えてください。")
    print(cb)

出力結果：

Tokens Used: 294
	Prompt Tokens: 19
	Completion Tokens: 275
Successful Requests: 1
Total Cost (USD): $0.005880000000000001

コンテストも追跡できます：

with get_openai_callback() as cb:
    result2 = llm("面白い歌詞を教えてください。")
    result3 = llm("面白いジョックを教えてください。")
    print(cb)

出力結果：

Tokens Used: 887
	Prompt Tokens: 43
	Completion Tokens: 844
Successful Requests: 2
Total Cost (USD): $0.017740000000000002

多くのステップを持つChainやAgentを使用する場合、それはすべてのステップを追跡します。

from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.llms import OpenAI

llm = OpenAI(temperature=0)
tools = load_tools(["llm-math"], llm=llm)
agent = initialize_agent(
    tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True
)
with get_openai_callback() as cb:
    response = agent.run(
        "孫正義はおいくつですか？"
    )
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Prompt Tokens: {cb.prompt_tokens}")
    print(f"Completion Tokens: {cb.completion_tokens}")
    print(f"Total Cost (USD): ${cb.total_cost}")

出力結果：

> Entering new AgentExecutor chain...
 I need to figure out how old Sun Zhengyi is.
Action: Calculator
Action Input: Sun Zhengyi's birth year (1954)
Observation: Answer: 1954
Thought: I now know Sun Zhengyi's age.
Final Answer: Sun Zhengyi is 66 years old.

> Finished chain.
Total Tokens: 690
Prompt Tokens: 618
Completion Tokens: 72
Total Cost (USD): $0.0138

最後に

これでLLMsを例として説明したが、Chat modelsでも同じ特徴があります。詳細はオフィシャルサイトを参照します。