LangChain (v0.2以降)の使い方確認

概要

昨年、LangChainにおけるAgent利用について中身を調査した記事を書きましたが、その後のLangChainのversion upによってほとんどのコードが新しいversionで動かない状況に…
色々と機能も追加されていると思うので、チュートリアルを元に新しいversionの使い方を確認しました。
(LLMについてはOpenAIのAPIを利用してGPT系のものを使用)

色々とチュートリアルページが用意されているので、ページごとに使い方とその詳細を見ていこうと思います。
今回は1番最初のチュートリアルである「Build a Simple LLM Application with LCEL」について確認。

Using Language Models

基本的な使い方は以下の通り。modelを設定してmessagesにSystemMessageとHumanMessageを入れてmodel.invoke()で生成させます。
System~が一般的に言われるプロンプトの部分で、Human~が実際の入力(問いかけ)になっている様子。
⇒ 公式では以下のように説明。

• SystemMessage：Message for priming AI behavior. The system message is usually passed in as the first of a sequence of input messages.
• HumanMessage：Message from a human. HumanMessages are messages that are passed in from a human to the model.

基本の使い方

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4")
messages = [
    SystemMessage(content="Translate the following from English into Italian"),
    HumanMessage(content="hi!"),
]
model.invoke(messages)

次にmodel.invoke()の詳細を確認。
ChatOpenAIクラスを確認すると、今回のmodelであるChatOpenAIはBaseChatOpenAI、BaseChatModelを継承しており、BaseChatModelの中のinvoke()が呼び出されていました。
invoke()の中身を確認すると、generate_prompt()から更にgenerate()が呼び出されて最終的な出力を生成している様子。

入力したmessagesは_convert_input()でChatPromptValueという形に変形された後、generate_prompt()の中のto_messages()で元の形に戻っていました。(なんのための変形？)

invoke()

def invoke(
    self,
    input: LanguageModelInput,
    config: Optional[RunnableConfig] = None,
    *,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> BaseMessage:
    config = ensure_config(config)
    return cast(
        ChatGeneration,
        # ↓ここが生成部分っぽい
        self.generate_prompt(
            [self._convert_input(input)],
            stop=stop,
            callbacks=config.get("callbacks"),
            tags=config.get("tags"),
            metadata=config.get("metadata"),
            run_name=config.get("run_name"),
            run_id=config.pop("run_id", None),
            **kwargs,
        ).generations[0][0],
    ).message

generate_prompt()

def generate_prompt(
    self,
    prompts: List[PromptValue],
    stop: Optional[List[str]] = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult:
    prompt_messages = [p.to_messages() for p in prompts]
    return self.generate(prompt_messages, stop=stop, callbacks=callbacks, **kwargs)

更に深堀りするとgenerate()の中ではlistのprompt_messagesを順番に_generate_with_cache()で処理。
_generate_with_cache()の中では_generate()にmessagesが入力として渡されていました。
_generate()の中のself.client.create(**payload)が最終的な生成部分？

generate()

def generate(
    self,
    messages: List[List[BaseMessage]],
    stop: Optional[List[str]] = None,
    callbacks: Callbacks = None,
    *,
    tags: Optional[List[str]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    run_name: Optional[str] = None,
    run_id: Optional[uuid.UUID] = None,
    **kwargs: Any,
) -> LLMResult:
    ``` 省略 ```
    results = []
    for i, m in enumerate(messages):
        try:
            results.append(
                self._generate_with_cache(
                    m,
                    stop=stop,
                    run_manager=run_managers[i] if run_managers else None,
                    **kwargs,
                )
            )
    ``` 省略 ```
    flattened_outputs = [
        LLMResult(generations=[res.generations], llm_output=res.llm_output)  # type: ignore[list-item]
        for res in results
    ]
    llm_output = self._combine_llm_outputs([res.llm_output for res in results])
    generations = [res.generations for res in results]
    output = LLMResult(generations=generations, llm_output=llm_output)  # type: ignore[arg-type]
    ``` 省略 ```
    return output

_generate_with_cache()

def _generate_with_cache(
    self,
    messages: List[BaseMessage],
    stop: Optional[List[str]] = None,
    run_manager: Optional[CallbackManagerForLLMRun] = None,
    **kwargs: Any,
) -> ChatResult:
    ``` 省略 ```
    else:
        if inspect.signature(self._generate).parameters.get("run_manager"):
            result = self._generate(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
        else:
            result = self._generate(messages, stop=stop, **kwargs)
    ``` 省略 ```
    return result

_generate()

def _generate(
    self,
    messages: List[BaseMessage],
    stop: Optional[List[str]] = None,
    run_manager: Optional[CallbackManagerForLLMRun] = None,
    **kwargs: Any,
) -> ChatResult:
    ``` 省略 ```
    payload = self._get_request_payload(messages, stop=stop, **kwargs)
    generation_info = None
    if "response_format" in payload:
        if self.include_response_headers:
            warnings.warn(
                "Cannot currently include response headers when response_format is "
                "specified."
            )
        payload.pop("stream")
        response = self.root_client.beta.chat.completions.parse(**payload)
    elif self.include_response_headers:
        raw_response = self.client.with_raw_response.create(**payload)
        response = raw_response.parse()
        generation_info = {"headers": dict(raw_response.headers)}
    else:
        response = self.client.create(**payload)
    return self._create_chat_result(response, generation_info)

更に、model.client.create(**payload)の部分の詳細を確認。
openai-pythonライブラリ中のcompletions.pyから呼び出されていました。
いつmodel.clientの方にCompletionクラスが入っているのかは掴み切れてません。
(Pydanticの仕様の理解が不十分なので怪しいけどここら辺?)
model.client.create()の中では_post()が呼び出されていますが、これはCompletionの継承元であるSyncAPIResourceを見ると、__init__()中でpost()が定義されていました。
post()から更に辿っていくと、request()→_request()→send()と続いています。

create()以降についてはopenai-pythonの中の話なので、invoke()の確認はここで終了。
チュートリアルの続きを見ていきます。

ParserやChain, PromptTemplateについて

出力の形式をParserで調整することが可能。
また|を使うことでchainとしてprompt, model, parserを繋げることも可能。
(色々とパーツごとにいじれるので便利そう)
ChatPromptTemplateを使えば形式的な入力の作成も出来る。

parser, chain

model = ChatOpenAI(model="gpt-4")
# ref
messages = [
    SystemMessage(content="Translate the following from English into Italian"),
    HumanMessage(content="hi!"),
]
model.invoke(messages)
# AIMessage(content='Ciao!', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 3, 'prompt_tokens': 20, 'total_tokens': 23}, 'model_name': 'gpt-3.5-turbo-0125', ~

# 出力のParser使用
parser = StrOutputParser()
result = model.invoke(messages)
parser.invoke(result)
# 'Ciao!'

# modelとparserのchain
chain = model | parser
chain.invoke(messages)
# 'Ciao!'

# promptも加えたchain
system_template = "Translate the following into {language}:"
prompt_template = ChatPromptTemplate.from_messages(
    [("system", system_template),
     ("user", "{text}")]
)
chain = prompt_template | model | parser
chain.invoke({"language":"italian",
              "text":"hi"}
            )
# 'Ciao!'

この|でChainが出来るのはLangChain側の実装みたいです。
parserのクラスであるStrOutputParserを辿っていくと、BaseOutputParserでRunnableSerializableを継承していることが分かります。
このRunnableSerializableが継承しているRunnableクラス中の__or__で|を使った時に呼び出される特殊メソッドとして定義しているようです。
(model側でも定義されているのかは未確認)

残りの部分についてはLangServeAPIを利用したアプリ化の話なので、確認は省略。
このページについては以上にしたいと思います。

まとめ

今回はLangChainの基本的な使い方について、ざっくりではありますが裏側の確認を行いました。
色々と発展的なことをするためなのか、かなり複雑な構造になっている印象です。
複雑過ぎるので必要に応じて自分用に作り直した方が、大幅変更あった際の影響が少なくて済むといった感じの記事をどこかで見かけたこともありますが、確かにそんな気にもなります…
他のチュートリアルも見ながら、使いたい機能を整理したいですね。