Chainlitについて学ぶ

まずはインストール
いつものようにuvを使う

uv add chainlit

uv run chainlit hello

で簡単に起動する。

tsutsutaku

app.py

import chainlit as cl


@cl.on_message
async def main(message: cl.Message):
    # Your custom logic goes here...

    # Send a response back to the user
    await cl.Message(
        content=f"Received: {message.content}",
    ).send()

ユーザーのインプットがあるたびにmain関数が呼ばれる。

LangChainを使ってみる

uv add langchain_openai

app.py

import chainlit as cl
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = llm | StrOutputParser()


@cl.on_message
async def main(message: cl.Message):
    # Your custom logic goes here...
    result = chain.invoke(message.content)
    # Send a response back to the user
    await cl.Message(
        content=result,
    ).send()

chainlit run app.py -w

で起動。
-wをつけることでオートリロードされる。

これだけで最低限のチャットボットができた。
簡単に使えていい

tsutsutaku

ストリーミング出力してみよう

app.py

import chainlit as cl
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = llm | StrOutputParser()


@cl.on_message
async def main(message: cl.Message):
    msg = await cl.Message(content="").send()

    async for token in chain.astream(message.content):
        await msg.stream_token(token)

    await msg.send()

でできた。

tsutsutaku

会話履歴を付けてみる。

app.py

import chainlit as cl
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

store = {}


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


prompt_template = ChatPromptTemplate.from_messages(
    [
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

chain = prompt_template | llm

runnable_with_history = RunnableWithMessageHistory(
    runnable=chain,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

session_id = "123"


@cl.on_message
async def main(message: cl.Message):
    msg = await cl.Message(content="").send()

    try:
        async for chunk in runnable_with_history.astream(
            {"input": message.content},
            config={"configurable": {"session_id": session_id}},
        ):
            if hasattr(chunk, "content"):
                content = chunk.content
            else:
                content = str(chunk)

            await msg.stream_token(content)

        await msg.send()
    except Exception as e:
        await cl.Message(content=f"エラーが発生しました: {str(e)}").send()

でできた。

参考