前に以下の記事で書いた内容をChainlitで実装し直しました。

ChainlitだとChatに特化しているだけあって、シンプルにUI実装できて非常にやりやすかったです。自動でLangChainの内部処理も画面表示できるようになっており、(UIでなく)機能にフォーカスしたPoCなどは非常にやりやすそうです。画面に細かく手を入れたい場合(例: 内部処理の非表示制御)は、難しそうだと予測していますが、調べていないのでわかりません。

サマリ

今回使っている主なモデルなどです。

種類	内容	備考
Chat Model	ChatOpenAI	今回のLLM(GPT3.5-Turbo)
Prompt Template	hwchase17/openai-functions-agent	会話履歴ありのOpen AI Function Agent
Tool	DuckDuckGo Search	検索エンジンとして使用
Tool	Wikipedia	Wiki検索に使用
Callback	LangchainCallbackHandler	ChainlitへCallback

画面

画面はこんな感じです。

Took N stepsを開くと詳細を見ることができます。

実装

事前準備

APIキー取得

OPEN AI とLangSmithでキーを取得し、ファイル".env"に書き込みます。Pythonプログラムと同じディレクトリに置きます。キー取得方法は省略します。LangSmithは個人勉強でやっているので無償でした。
キー以外にもモデルやTemperatureなど設定しています。

.env

OPENAI_API_KEY=<key>
OPENAI_API_MODEL=gpt-3.5-turbo
OPENAI_API_TEMPERATURE=0.5
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=<key>

Package Install

以下のPythonパッケージをインストールしています。ちなみにPythonは3.12.2を使っています。

Package	Version	備考
chainlit	1.0.506
openai	1.23.6	langchain-openaiがあれば不要？
langchain	0.1.16
langchain-openai	0.1.4
langchain-community	0.0.34
langchainhub	0.1.15
python-dotenv	1.0.1	.envファイルのロード
duckduckgo-search	5.3.0
wikipedia	1.4.0

Python Program

Pythonプログラムの全体です。

import os

from dotenv import load_dotenv
from langchain import hub
from langchain.agents import AgentExecutor, load_tools, create_openai_functions_agent
from langchain.memory import ConversationBufferMemory
from langchain.schema.runnable.config import RunnableConfig
from langchain_openai import ChatOpenAI

import chainlit as cl
import wikipedia


load_dotenv()

@cl.on_chat_start
async def on_chat_start():
    await cl.Message(f"How can I help you?").send()
    model = ChatOpenAI(
        model_name=os.environ["OPENAI_API_MODEL"],
        temperature=os.environ["OPENAI_API_TEMPERATURE"],
        streaming=True,
    )
    prompt = hub.pull("hwchase17/openai-functions-agent")
    tools = load_tools(["ddg-search", "wikipedia"])
    wikipedia.set_lang('ja')
    _agent = create_openai_functions_agent(model, tools, prompt)

    memory = ConversationBufferMemory(return_messages=True, 
                                      memory_key="chat_history",
                                      output_key='output')

    agent_executor = AgentExecutor(agent=_agent,
                                        tools=tools, 
                                        verbose=True, 
                                        memory=memory
                                        )
    
    cl.user_session.set("agent_executor", agent_executor)
    

@cl.on_message
async def on_message(message: cl.Message):
    agent_executor = cl.user_session.get("agent_executor")
    msg = cl.Message(content="")

    async for chunk in agent_executor.astream(
        {"input": message.content},
        config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()],
                              configurable={"session_id": "any"}),
    ):
        # Agent Action
        if "actions" in chunk:
            for action in chunk["actions"]:
                print(f"Calling Tool: `{action.tool}` with input `{action.tool_input}`")
        # Observation
        elif "steps" in chunk:
            for step in chunk["steps"]:
                print(f"Tool Result: `{step.observation}`")
        # Final result
        elif "output" in chunk:
            print(f'Final Output: {chunk["output"]}')
            await msg.stream_token(chunk['output'])
        print("---")
        
    await msg.send()

上記の全体のうち、分割して少し説明。

初期実行

@on_chat_startで初回実行されます。Agent Executorを非同期で作っています。

@cl.on_chat_start
async def on_chat_start():
    await cl.Message(f"How can I help you?").send()
    model = ChatOpenAI(
        model_name=os.environ["OPENAI_API_MODEL"],
        temperature=os.environ["OPENAI_API_TEMPERATURE"],
        streaming=True,
    )
    prompt = hub.pull("hwchase17/openai-functions-agent")
    tools = load_tools(["ddg-search", "wikipedia"])
    wikipedia.set_lang('ja')
    _agent = create_openai_functions_agent(model, tools, prompt)

    memory = ConversationBufferMemory(return_messages=True, 
                                      memory_key="chat_history",
                                      output_key='output')

    agent_executor = AgentExecutor(agent=_agent,
                                        tools=tools, 
                                        verbose=True, 
                                        memory=memory
                                        )
    
    cl.user_session.set("agent_executor", agent_executor)

メッセージPost時実行

@cl.on_messageでメッセージをPostしたときに動くロジックです。
最後の条件分岐はターミナルに出すだけの確認用ロジックが多く、実質的な機能として動いているのはoutput時に出力する部分のみです。

async def on_message(message: cl.Message):
    agent_executor = cl.user_session.get("agent_executor")
    msg = cl.Message(content="")

    async for chunk in agent_executor.astream(
        {"input": message.content},
        config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()],
                              configurable={"session_id": "any"}),
    ):
        # Agent Action
        if "actions" in chunk:
            for action in chunk["actions"]:
                print(f"Calling Tool: `{action.tool}` with input `{action.tool_input}`")
        # Observation
        elif "steps" in chunk:
            for step in chunk["steps"]:
                print(f"Tool Result: `{step.observation}`")
        # Final result
        elif "output" in chunk:
            print(f'Final Output: {chunk["output"]}')
            await msg.stream_token(chunk['output'])
        print("---")
        
    await msg.send()