概要

langchainのv0.1がリリースされたので、そのコア機能であるLCEL（LangChain　Expression Language）の使い方を練習します。

練習テーマ

選択肢問題をGPTに直接解かせたり、RAGで解かせたりしてみます。

データセット

JAQKET: クイズを題材にした日本語QAデータセットのvalidationを使います。
huggingfaceからダウンロードします。

pip install datasets

from datasets import load_dataset

dataset = load_dataset("kumapo/JAQKET", name="v1.0", split="validation")

中身を確認してみます。

QUESTION_NUM = 10
subset = dataset.select(range(QUESTION_NUM))
for item in subset:
    print(f'id: {item["qid"]}, question: {item["question"]}, answer: {item["answer_entity"]}, options: {item["answer_candidates"]}')

id: QA20CAPR-0006, question: 「パイプスライダー」や「そり立つ壁」などの関門がある、TBS系列で不定期に放送されている視聴者参加型のTV番組は何でしょう?, answer: SASUKE, options: ['最強の男は誰だ!壮絶筋肉バトル!!スポーツマンNo.1決定戦', 'SASUKE', '松本vs浜田・対決シリーズ/罰ゲーム', '島田紳助がオールスターの皆様に芸能界の厳しさ教えますスペシャル!', '激突 なりきりスター天下とったる歌合戦!']
id: QA20CAPR-0008, question: 東京都内では最も古い歴史を持つ寺院でもある、入口にある「雷門」で有名な観光名所は何でしょう?, answer: 浅草寺, options: ['下谷神社', '浅草寺', '浄閑寺', '天龍寺 (新宿区)', '源覚寺 (文京区)']
id: QA20CAPR-0012, question: よくオムライスの中身としても使われる、細かく切った鶏肉やタマネギとごはんを炒め、トマトケチャップで味付けしたものを何というでしょう?, answer: チキンライス, options: ['チキンライス', '冷やし麺', '野菜炒め', '炸醤麺', 'キムチチゲ']
id: QA20CAPR-0022, question: 東京23区の中で、名前に体の一部を表す漢字が入っているのは、目黒区とどこでしょう?, answer: 足立区, options: ['足立区', '墨田区', '西東京市', '台東区', '豊島区']
id: QA20CAPR-0030, question: アフリカの国で国名に「アフリカ」と入るのは、南アフリカ共和国とどこでしょう?, answer: 中央アフリカ共和国, options: ['ベナン', 'コンゴ民主共和国', 'ニジェール', '中央アフリカ共和国', 'ガボン']
id: QA20CAPR-0052, question: 女性アナウンサーとタモリが舞台となる街を歩きながらその歴史を探っていくという内容の、NHKの紀行番組は何でしょう?, answer: ブラタモリ, options: ['サキどり↑', '探検バクモン', 'もふもふモフモフ', 'ブラタモリ', 'これでわかった! 世界のいま']
id: QA20CAPR-0062, question: その頭の形から出産はほぼ帝王切開で行われるのが特徴である、牛と闘うために改良されたことから名が付いた犬の品種は何でしょう?, answer: ブルドッグ, options: ['ミニチュア・シュナウザー', 'シーズー', 'プードル', 'ブルドッグ', 'パピヨン (犬)']
id: QA20CAPR-0066, question: 英語では「rickshaw」といい、現在は観光地などで見られる、人が引っ張って動かす車のことを何というでしょう?, answer: 人力車, options: ['人力車', 'ベカ車', 'キャリッジ', 'ハックニーキャリッジ', 'リヤカー']
id: QA20CAPR-0068, question: 美人のこれは「富士」にたとえられ、せまい場所は「猫」のものにたとえられる顔の部分は何でしょう?, answer: 額, options: ['手首', '胸', '耳介', 'かかと', '額']
id: QA20CAPR-0086, question: ジャズの演奏ではウッドベースと呼ばれることが多い、オーケストラで使われる最大の弦楽器は何でしょう?, answer: コントラバス, options: ['コントラバス', 'ホルン', 'トロンボーン', 'ティンパニ', 'バストロンボーン']

定数の準備

選択肢をプロンプト用に整形する関数を作ります。

def option_list_to_string(options:List[str])->str:
    option_string = "\n".join([f"{i}. {option}" for i, option in enumerate(options,1)])
    return option_string

print(option_list_to_string(subset[0]["answer_candidates"]))

1. 最強の男は誰だ!壮絶筋肉バトル!!スポーツマンNo.1決定戦
2. SASUKE
3. 松本vs浜田・対決シリーズ/罰ゲーム
4. 島田紳助がオールスターの皆様に芸能界の厳しさ教えますスペシャル!
5. 激突 なりきりスター天下とったる歌合戦!

テスト用に1問目の情報を取り出して定数に格納します。
また、LLMのパラメータも定義します。

LLM_MODEL = "gpt-3.5-turbo"
TEMPERATURE = 0
QUESTION = subset[0]["question"]
OPTION_STR = option_list_to_string(subset[0]["answer_candidates"])

各要素を個別に触る

LCELを構成するいくつかの代表的な要素を直接触ってみます。
まず必要なライブラリを後で使うものも含めてimportします。

import pandas as pd
from langchain_openai import ChatOpenAI

from langchain_community.utils.openai_functions import convert_pydantic_to_openai_function

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser

from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

from typing import List, Dict, Any

from operator import itemgetter
import dotenv
dotenv.load_dotenv()

LCELのパイプでつなげる要素はRunnable（またはそれを継承したクラス）か、callable（関数）、辞書です。辞書の場合は、全てのvalueがRunnableかcallbleである必要があります。langchainのLCELのチュートリアルで使われているようなクラスは基本、Runnableクラスを継承しています。Runnableクラスはinvokeメソッドをもっており、パイプの前後では基本的にこのinvokeメソッドが呼び出されます。invokeが取れる引数の型はクラスごとに異なるため、各クラスでinvokeの入出力型を理解しておくことが重要です。

prompt

プロンプト（ChatPromptValue）をつかってみます。
invokeの入力はDict[str, str]、出力はChatPromptValueです。

prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた選択肢問題に回答してください"),
    ("user", "問題文:\n{question}\n\n選択肢:\n{option_str}")
    ]
)

prompt.invoke({"question":QUESTION, "option_str":OPTION_STR}).

ChatPromptValue(messages=[SystemMessage(content='与えられた選択肢問題に回答してください'), HumanMessage(content='問題文:\n「パイプスライダー」や「そり立つ壁」などの関門がある、TBS系列で不定期に放送されている視聴者参加型のTV番組は何でしょう?\n\n選択肢:\n1. 最強の男は誰だ!壮絶筋肉バトル!!スポーツマンNo.1決定戦\n2. SASUKE\n3. 松本vs浜田・対決シリーズ/罰ゲーム\n4. 島田紳助がオールスターの皆様に芸能界の厳しさ教えますスペシャル!\n5. 激突 なりきりスター天下とったる歌合戦!')])

LLM

LLM（ChatOpenAI）を使ってみます。
invokeの入力はstrやChatPromptValue、出力はBaseMessageです。

llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE)

llm.invoke("こんにちは")

AIMessage(content='こんにちは！何かお手伝いできますか？')

ChatPromptValueも引数に取れることを確認しておきます。

prompt_value = prompt.invoke({"question":QUESTION, "option_str":OPTION_STR})
llm.invoke(prompt_value)

AIMessage(content='選択肢2. SASUKE')

OutputParser

StrOutputParserを使ってみます。LLMの出力であるBaseMessageをstrに変換するために使います。
invokeの入力はstrやBaseMessage、出力はstrです。

strを入力した場合、同じstrが出力されます。

output_parser = StrOutputParser()
output_parser.invoke("こんにちは")

'こんにちは'

BaseMessageを入力した場合、content要素が取り出されます。

llm_response = llm.invoke("こんにちは")
output_parser.invoke(llm_response)

'こんにちは！何かお手伝いできますか？'

チェインを作ってみる

Runnableを継承したクラスをパイプ（|）でつなぐことでチェインを作れます。
パイプをまたぐ処理ではinvokeメソッドが実行されるので、パイプの左の要素のinvoke出力と、右の要素のinvoke入力の型が揃っている必要があります。

promptとllm

promptのinvoke出力とllmのinvoke入力が一致するので、これらをパイプでつなげてchainを作れます。
chainのinvokeの引数はパイプの一番左の要素（今回はprompt）のinvokeの引数と同じです。

chain = prompt | llm
chain.invoke({"question":QUESTION, "option_str":OPTION_STR})

AIMessage(content='選択肢2. SASUKE')

llmとoutput_parser

llm.invokeの出力とoutput_parser.invokeの入力が一致するので、これらをパイプでつなぐことができます。
chain.invokeの入力はパイプの一番左の要素（今回はllm）のinvokeの入力と同じです。

chain = llm | output_parser
chain.invoke("こんにちは")

'こんにちは！何かお手伝いできますか？'

promptとllmとoutput_parser

3つの要素を繋げてみます。

chain = prompt | llm | output_parser
chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

'選択肢2. SASUKE'

出力形式の指定（function_calling）

openai_functionsとJsonOutputFunctionsParserを使って出力形式を指定してみます。

pydantic.BaseModelで形式を定義して、convert_pydantic_to_openai_function関数でfunction_calling用の形式に変換します。

class Response(BaseModel):
    """Answer to the question."""
    answer: int = Field(description="correct option number for the answer")
openai_functions= [convert_pydantic_to_openai_function(Response)]
print(openai_functions)

[{'name': 'Response', 'description': 'Answer to the question.', 'parameters': {'title': 'Response', 'description': 'Answer to the question.', 'type': 'object', 'properties': {'answer': {'title': 'Answer', 'description': 'correct option number for the answer', 'type': 'integer'}}, 'required': ['answer']}}]

出力形式を指定したchainを作ります。
ポイントはllmをfunction_calling用にbindしていること、output_parserをJsonOutputFunctionsParserにしていることです。
llm実行時にとる引数を指定するメソッドです。functionsにfunction_callingの型指定を、function_callに使用する関数の名前を指定することで、必ずその関数の型で出力を得ることができます。
JsonOutputFunctionsParserはfunction_callの戻り値を辞書に変換するためのoutput_parserです。

prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた選択肢問題に回答してください"),
    ("user", "問題文:\n{question}\n\n選択肢:\n{option_str}")
    ]
)

llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(functions = openai_functions, function_call={"name": "Response"})

output_parser = JsonOutputFunctionsParser()

chain = prompt | llm | output_parser

chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

[{'name': 'Response', 'description': 'Answer to the question.', 'parameters': {'title': 'Response', 'description': 'Answer to the question.', 'type': 'object', 'properties': {'answer': {'title': 'Answer', 'description': 'correct option number for the answer', 'type': 'integer'}}, 'required': ['answer']}}]

途中の処理状況を確認するために部分的なchainを作って、invokeしてみます。

(prompt | llm).invoke({"question": QUESTION, "option_str":OPTION_STR})

AIMessage(content='', additional_kwargs={'function_call': {'arguments': '{\n  "answer": 2\n}', 'name': 'Response'}})

きちんとfunction_callできています。

Chain of Thought

Chain of Thoughtで回答させてみます。
出力形式にthoughtを含めることで実現します。function_callingも結局jsonの文字列を1文字ずつ出力しているだけなので、thought, answerの順番に値を出力させれば、実質的にChain of thoughtになるはずです。

class Response(BaseModel):
    """Answer to the question."""
    thought: str = Field(description="thought or evidence to answer to the question correctly")
    answer: int = Field(description="correct option number for the answer")
openai_functions= [convert_pydantic_to_openai_function(Response)]
print(openai_functions)

[{'name': 'Response', 'description': 'Answer to the question.', 'parameters': {'title': 'Response', 'description': 'Answer to the question.', 'type': 'object', 'properties': {'thought': {'title': 'Thought', 'description': 'thought or evidence to answer to the question correctly', 'type': 'string'}, 'answer': {'title': 'Answer', 'description': 'correct option number for the answer', 'type': 'integer'}}, 'required': ['thought', 'answer']}}]

thoughtというpropertyを増やしました。

prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた選択肢問題に回答してください"),
    ("user", "問題文:\n{question}\n\n選択肢:\n{option_str}")
    ]
)

llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(
    functions = openai_functions, function_call={"name": "Response"}
)

output_parser = JsonOutputFunctionsParser()

chain = prompt | llm | output_parser

chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

{'thought': 'TBS系列で不定期に放送されている視聴者参加型のTV番組で、関門があるといえば「SASUKE」です。', 'answer': 2}

thoughtとanswerを出力することができました。

answerだけ取り出したいケースもあると思うので、answerだけ取り出してみます。
標準関数のoperator.itemgetterを使うことで、dictから特定のkeyに値を取り出す関数を作ることができます。これをoutput_parserの後ろにつけると良いです。

chain = prompt | llm | output_parser | itemgetter("answer")

chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

検索して回答させる（RAG）

Wikipediaで検索した結果も踏まえて回答するchainを作ってみます。

まずwikipediaを検索するためのツールを作ります。

api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)
tool = WikipediaQueryRun(api_wrapper=api_wrapper)

tool.invoke({"query": "langchain"})

'Page: LangChain\nSummary: LangChain is a framework designed to simplify the creation of applications '

toolもRunnableであるため、invokeメソッドがあり、chainの要素とすることができます。
入力型はqueryというkeyをもつ辞書か、strです。出力はstrです。
strも入力できることを確かめておきます。

tool.invoke("langchain")

'Page: LangChain\nSummary: LangChain is a framework designed to simplify the creation of applications '

toolをchainに含めてみます。
出力形式を作ります。

class Response(BaseModel):
    """Answer to the question."""
    answer: int = Field(description="correct option number for the answer")
openai_functions= [convert_pydantic_to_openai_function(Response)]

promptを作ります。question, option_strに加え、参考情報を格納するcontextという変数も加えます。

prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた選択肢問題に回答してください。"),
    ("user", "問題文:\n{question}\n\n選択肢:\n{option_str}\n\n参考情報(検索結果):\n{context}")
    ]
)

llm、output_parserを作ります。これはそのままです。

llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(
    functions = openai_functions, function_call={"name": "Response"}
)

output_parser = JsonOutputFunctionsParser()

chainを作ります。

chain = (
    {
        "context": itemgetter("question") | tool,
        "question": itemgetter("question"),
        "option_str": itemgetter("option_str")
    }
    | prompt
    | llm
    | output_parser
)
chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

{'answer': 2}

挙動確認

chainはできましたが、最初の辞書が何をやっているのかよくわからないので、少し挙動を確認してみます。
検証のためにRunnablePassthroughクラスを使います。RunnablePassthrough.invokeは入力を無加工で出力します。つまり、RunnablePassthroughがchainの要素にあるとき、入力をそのまま次のパイプに渡すことができます。

RunnablePassThrough().invoke("a")

'a'

次に辞書をchainの要素に使ったときの入出力を確認します。
結論からいうと、辞書の各valueに直前のinvokeの出力がそのまま渡されます。したがってvalueはRunnableである必要があります。各valueの出力がkeyと対応付けられて、出力されます。

これは以下のように確認できます。辞書単体ではinvokeメソッドがなくchainにできないので、後ろにRunnablePassthrough()をパイプでつなげることで、辞書自体がchainの中でどのような出力とみなされているかがわかります。

chain = ({
        "a": RunnablePassthrough(),
        "b": RunnablePassthrough()
    } | RunnablePassthrough())
chain.invoke({"c":"c", "d":"d"})

{'a': {'c': 'c', 'd': 'd'}, 'b': {'c': 'c', 'd': 'd'}}

辞書のaにもbにもchain.invokeの入力がまるごと渡されていることがわかります。

chainへの入力のうち、特定のkeyだけ取り出して辞書のvalueに与えたい場合、itemgetterを使えばよいです。

chain = ({
        "a": itemgetter("c"),
        "b": itemgetter("d")
    } | RunnablePassthrough())
chain.invoke({"c":"c", "d":"d"})

{'a': 'c', 'b': 'd'}

今回のケースではquestion、option_str、contextの３つをpromptに渡す必要があります。
また、contextはquestionをtoolに渡した結果である必要があります。
この処理を以下のように実現できます。

chain = ({
        "context": itemgetter("question") | tool,
        "question": itemgetter("question"),
        "option_str": itemgetter("option_str")
    }
    | prompt)
chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

ChatPromptValue(messages=[SystemMessage(content='与えられた選択肢問題に回答してください'), HumanMessage(content='問題文:\n「パイプスライダー」や「そり立つ壁」などの関門がある、TBS系列で不定期に放送されている視聴者参加型のTV番組は何でしょう?\n\n選択肢:\n1. 最強の男は誰だ!壮絶筋肉バトル!!スポーツマンNo.1決定戦\n2. SASUKE\n3. 松本vs浜田・対決シリーズ/罰ゲーム\n4. 島田紳助がオールスターの皆様に芸能界の厳しさ教えますスペシャル!\n5. 激突 なりきりスター天下とったる歌合戦!')])

chainへの入力はquestionとoption_strのみです。これをpromptに渡す段階では、context, question, option_strの3つにする必要があります。questionとoption_strはitemgetterで要素を取り出してそのまま渡しています。contextについてはitemgetterでquestionを取り出しそれをtoolに渡す結果を得る処理をしています。

promptができて以降の処理はいつものchain処理となります。

クエリをLLMに考えさせてからRAG

最後に、questionをそのままクエリとするのではなく、クエリをLLMに考えさせる処理も前段に設けてみたいと思います。

イメージとしてはtoolに渡す前に処理を入れることになります。したがって、query_generation_chainというものが存在するとして、chainの第一要素のcontext部分が以下のようになれば良さそうです。

chain = (
    {
        "context": {"question": itemgetter("question")} | query_generation_chain | tool,
        ...

query_generation_chain

クエリを生成するためのchainを作ります。

class Query(BaseModel):
    """query"""
    query: str = Field(description="query to search database e.g. 'langchain'")
query_functions = [convert_pydantic_to_openai_function(Query)]

query_prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた問題に回答するため、データベースを検索するクエリを出力してください"),
    ("user", "問題文:\n{question}")
])

query_llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(
    functions = query_functions, function_call={"name": "Query"}
)

query_output_parser = JsonOutputFunctionsParser()

query_generation_chain = query_prompt | query_llm | query_output_parser

query_generation_chain.invoke({"question": QUESTION})

{'query': 'パイプスライダー そり立つ壁 TBS'}

よさそうです。
strを出力できたほうがtoolに渡しやすいので、itemgetterを最後に付けます。

query_generation_chain = query_prompt | query_llm | query_output_parser | itemgetter("query")
query_generation_chain.invoke({"question": QUESTION})

'パイプスライダー そり立つ壁 TBS'

toolと繋げてみます。

(query_generation_chain | tool).invoke({"question": QUESTION})

'No good Wikipedia Search Result was found'

No good resultになってしまいましたが、動作確認ができれば良いので、続けます。

回答生成用のchain要素も定義し、最後にchainを作って実装します。


class Response(BaseModel):
    """Answer to the question."""
    answer: int = Field(description="correct option number for the answer")
response_functions= [convert_pydantic_to_openai_function(Response)]


response_prompt = ChatPromptTemplate.from_messages([
    ("system", "与えられた選択肢問題に回答してください。"),
    ("user", "問題文:\n{question}\n\n選択肢:\n{option_str}\n\n参考情報(検索結果):\n{context}")
    ]
)


response_llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(
    functions = response_functions, function_call={"name": "Response"}
)

query_llm = ChatOpenAI(model = LLM_MODEL, temperature=TEMPERATURE).bind(
    functions = query_functions, function_call={"name": "Query"}
)

output_parser = JsonOutputFunctionsParser()

chain = (
    {
        "context": {"question": itemgetter("question")} | query_generation_chain | tool,
        "question": itemgetter("question"),
        "option_str": itemgetter("option_str")
    }
    | prompt
    | llm
    | output_parser
)
chain.invoke({"question": QUESTION, "option_str":OPTION_STR})

{'answer': 2}

おわりに

langchainのv0.1およびLCELに習熟するため、RAGをテーマとしていろいろ実装してみました。
感想としては、従来のlegacyな方法より、記法も直感的で、途中出力の確認がしやすく、よかったです。

多くの文書を処理すると、function_callingに失敗するとoutput_parserでエラーが起こってしまうので、エラー処理をうまくやる方法は今後の課題です。

個人的なポイントを以下にまとめます。

パイプの前後で、invokeの入出力を揃える
デバッグや動作確認では部分的なchainを作ってinvokeする
辞書をchainの要素として使う場合、各valueには直前の出力がそのまま渡される。要素の一部だけ渡したい場合はitemgetterを前段にいれる

以上です。

参考資料