ChainlitとLangChain(LCEL)でGPT連携(wikipediaとDuckDuckGoも)

前に以下の記事で書いた内容を[**Chainlit**](https://docs.chainlit.io/get-started/overview)**で実装し直し**ました。

https://qiita.com/FukuharaYohei/items/a11628c1ae081c5b01ac

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134273/94551899-de8d-3b01-1bb4-c96711dc1f8b.png)

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

|種類|内容|備考|
|:-:|:--|:--|
|Chat Model|[ChatOpenAI](https://api.python.langchain.com/en/latest/chat_models/langchain_openai.chat_models.base.ChatOpenAI.html)|今回のLLM(GPT3.5-Turbo)
|Prompt Template|[hwchase17/openai-functions-agent](https://smith.langchain.com/hub/hwchase17/openai-functions-agent)|会話履歴ありのOpen AI Function Agent
|Tool|[DuckDuckGo Search](https://python.langchain.com/docs/integrations/tools/ddg/)|検索エンジンとして使用
|Tool|[Wikipedia](https://python.langchain.com/docs/integrations/tools/wikipedia/)|Wiki検索に使用
|Callback|[LangchainCallbackHandler](https://api.python.langchain.com/en/latest/callbacks/langchain_community.callbacks.streamlit.streamlit_callback_handler.StreamlitCallbackHandler.html)|ChainlitへCallback

# 画面
画面はこんな感じです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134273/ec7031aa-d6e9-261f-0db3-4b8c741df33e.png)

`Took N steps`を開くと詳細を見ることができます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134273/d95f7775-876c-0789-8521-7a086a5c0d67.png)

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

```text:.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プログラムの全体です。

```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`](https://docs.chainlit.io/api-reference/lifecycle-hooks/on-chat-start)で初回実行されます。Agent Executorを非同期で作っています。

```python
@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`](https://docs.chainlit.io/api-reference/lifecycle-hooks/on-message)でメッセージをPostしたときに動くロジックです。
最後の条件分岐はターミナルに出すだけの確認用ロジックが多く、実質的な機能として動いているのは`output`時に出力する部分のみです。

```python
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()
```