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