1. LCEL（LangChain Expression Language）の紹介

LCEL（LangChain Expression Language）は、複雑なチェーンを構築するためのシンプルで使いやすいフレームワークです。統一されたインターフェースと合成プリミティブを提供し、チェーンの構築を容易にします。各LCELオブジェクトはRunnableインターフェースを実装し、invoke、batch、stream、ainvokeなどの一般的に使用される呼び出しメソッドのセットを定義しています。そのため、LCELオブジェクトのチェーンも自動的にこれらの呼び出しメソッドをサポートし、各LCELオブジェクト自体がLCELオブジェクトとなります。

2. 呼び出し

2.1. LCELを使用しない場合

LCELを使用せず、以下のコードスニペットを使用して、トピック文字列を渡し、ジョーク文字列を取得できます。

from typing import List
import openai

prompt_template = "{topic}についてのジョークを教えて"
client = openai.OpenAI()

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

def invoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return call_chat_model(messages)

invoke_chain("アイスクリーム")

2.2. LCELを使用する場合

それに対し、LCELを使用すると、同じ機能をより簡潔に実現できます。次のコードスニペットは、LCELオブジェクトを使用して簡単にチェーンを構築する方法を示しています。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "{topic}についてのジョークを教えて"
)
output_parser = StrOutputParser()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.invoke("アイスクリーム")

3. ストリーミング

3.1. LCELを使用しない場合

以下のコードスニペットは、LCELを使用せずに結果をストリーミング処理する方法を示しています。

from typing import Iterator

def stream_chat_model(messages: List[dict]) -> Iterator[str]:
    stream = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
    )
    for response in stream:
        content = response.choices[0].delta.content
        if content is not None:
            yield content

def stream_chain(topic: str) -> Iterator[str]:
    prompt_value = prompt.format(topic=topic)
    stream = stream_chat_model([{"role": "user", "content": prompt_value}])
    for chunk in stream:
        print(chunk, end="", flush=True)

stream_chain("アイスクリーム")

3.2. LCELを使用する場合

LCELを使用して結果をストリーミング処理する方が便利です。次のコードスニペットは、LCELを使用したストリーミング方法を示しています。

for chunk in chain.stream("アイスクリーム"):
    print(chunk, end="", flush=True)

4. バッチ処理

4.1. LCELを使用しない場合

以下のコードスニペットは、LCELを使用せずに複数の入力を並列処理する方法を示しています。

from concurrent.futures import ThreadPoolExecutor

def batch_chain(topics: list) -> list:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(invoke_chain, topics))

batch_chain(["アイスクリーム", "スパゲッティ", "餃子"])

4.2. LCELを使用する場合

LCELを使用したバッチ処理は非常に簡単です。次のコードスニペットは、LCELを使用したバッチ処理の方法を示しています。

chain.batch(["アイスクリーム", "スパゲッティ", "餃子"])