はじめに
OpenAIFunctionsAgentに記憶を持たせて使う方法がよくわからなかったので、整理しました。
経緯として、OpenAIFunctionsAgentを使った対話アプリを作りたかったのですが、langchainのdocsでは、１ターンの実行方法しか書かれておらず、過去の対話履歴をAgentに渡す方法がよくわかりません。
一方、ConversationalAgentという対話用のエージェントもあるのですが、こちらはfunction callingせずに、文字列の応答を力技でパースしています。せっかくならfunction callingのほうが精度が高いはずなので、うまく組み合わせたいです。
そこで、実装例を探すと、langchainの本家が別リポジトリで公開しているchat-langchainという対話アプリで、OpenAIFunctionsCallingが使われていることを発見しました。
実装
get_agent
chat-langchain/main.pyのget_agentという関数を見ていきます。
まず関数の入力と出力を見ます。
def get_agent(llm, *, chat_history: Optional[list] = None):
chat_history = chat_history or []
...
return agent_executor
入力はllmとchat_historyです。つまり、agentの実行に使う言語モデルと対話履歴です。
出力はagent_executor、つまりAgentExecutorです。Agentそのものではないことに注意です。
chat_historyの初期化の続きから見ていきます。まずpromptのテンプレートを作っています。
prompt = OpenAIFunctionsAgent.create_prompt(
system_message=system_message,
extra_prompt_messages=[MessagesPlaceholder(variable_name="chat_history")],
)
system_messageのほかに、MessagesPlaceholderを渡しています。
MessagesPlaceholderはMessagePropmtTemplateを継承したクラスで、variableがすでにlistになっているときに使われます。今回は、chat_history変数にあとで、履歴のリストを挿入するのだと思われます。
次にmemoryを定義しています。
memory = AgentTokenBufferMemory(
memory_key="chat_history", llm=llm, max_token_limit=2000
)
for msg in chat_history:
if "question" in msg:
memory.chat_memory.add_user_message(str(msg.pop("question")))
if "result" in msg:
memory.chat_memory.add_ai_message(str(msg.pop("result")))
AgentTokenBufferMemoryを初期化したのち、chat_historyの要素をコピーしています。chat_historyの要素はquestionやresultなどのkeyをもつことを想定されているようです。
次にtoolsを取得してAgentを生成しています。
tools = get_tools()
agent = OpenAIFunctionsAgent(llm=llm, tools=tools, prompt=prompt)
agentにはllmとtoolsとpromptだけを与えており、memoryは与えていないようです。
最後にAgentExecutorを生成して返しています。
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
return_intermediate_steps=True,
)
return agent_executor
memoryはAgentExecutorに渡されています。
chat_endpoint
get_agentで作成されたAgentExecutorが対話アプリでどのように使われているかを見ます。
main.pyのchat_endpointの実装をみます。
@app.post("/chat")
async def chat_endpoint(request: Request):
global run_id, feedback_recorded, trace_url
run_id = None
trace_url = None
feedback_recorded = False
run_collector.traced_runs = []
data = await request.json()
question = data.get("message")
chat_history = data.get("history", [])
data.get("conversation_id")
print("Recieved question: ", question)
data.get("history")でchat_historyを取得しています。dataはrequestから作られているので、おそらくUI側がチャットの履歴を毎回送るように実装されていると思われます。
get_agentでは、chat_historyはquestionやresultをkeyにもつ辞書のリストであるとして処理されていましたが、そのような辞書になるようにUI側で定義していると思われます。
続きをみます。関数内の関数としてstreamが定義されています。
def stream() -> Generator:
global run_id, trace_url, feedback_recorded
q = Queue()
job_done = object()
llm = ChatOpenAI(
model="gpt-3.5-turbo-16k",
streaming=True,
temperature=0,
callbacks=[QueueCallback(q)],
)
ChatOpenAIが定義されています。特に変わったところはありません。
続きを見ます。stream関数内でさらにtaskという関数が定義されています。
def task():
agent = get_agent(llm, chat_history=chat_history)
agent.invoke(
{"input": question, "chat_history": chat_history},
config=runnable_config,
)
q.put(job_done)
get_agentでagent(agent_executor)を取得した後、agent.invokeが実行されています。invokeは__call__とほぼ同じ役割のメソッドです(参考)。invokeにquestionだけでなくchat_historyも渡されている意味はよくわかりません(履歴はAgentExecutorの初期化時に渡し済では？）。
このあとも続きますが、AgentExecutorの実行とは直接的には関係ないので省略します。
まとめ
get_agentとchat_endpointの処理をまとめると以下のようになります。
- Agentの作成
- MessagesPlaceholderを使って履歴挿入を可能にしたプロンプトテンプレートを作成し、これを渡してOpenAIFunctionsAgentを初期化する
- AgentExecutorの作成
- AgentTokenBufferMemory型のmemoryを作成し、chat_historyの内容をコピーする
- memoryとAgentを与え、AgentExecutorを初期化する
- AgentExecutorの実行
- AgentExecutor.invokeにinputとchat_historyを与える
chat-langchainではAgentExecutorを応答生成のたびに初期化しています。chat_historyはその都度与えています。
langchainにはChain.runの実行の際にmemoryに履歴を追加する機能もありますが、毎回初期化して都度chat_historyをセットし直しているのには何か理由があるのでしょうか？ おそらくですが、OpenAIFunctionsAgentの場合は、FunctionMessageが対話の都度生成される可能性がありますが、これを履歴には残したくないため、毎回初期化し、user_messageとai_messageだけを取り出している、ような気がします。
実験
chat-langchainの実装を参考に対話履歴を考慮したOpenAIFunctionsAgent利用のchainを作ってみます。
import os
from typing import Optional
from langchain.chat_models import ChatOpenAI
from langchain.prompts import MessagesPlaceholder
from langchain.schema.messages import SystemMessage
from langchain.agents import (
AgentExecutor,
)
from langchain.agents.openai_functions_agent.base import OpenAIFunctionsAgent
from langchain.agents.openai_functions_agent.agent_token_buffer_memory import (
AgentTokenBufferMemory,
)
from langchain.agents import load_tools
def get_agent(llm, *, chat_history: Optional[list] = None):
chat_history = chat_history or []
system_message = SystemMessage(
content="あなたは計算を助けるアシスタントです"
)
prompt = OpenAIFunctionsAgent.create_prompt(
system_message=system_message,
extra_prompt_messages=[MessagesPlaceholder(variable_name="chat_history")],
)
memory = AgentTokenBufferMemory(
memory_key="chat_history", llm=llm, max_token_limit=2000
)
for msg in chat_history:
if "question" in msg:
memory.chat_memory.add_user_message(str(msg.pop("question")))
if "result" in msg:
memory.chat_memory.add_ai_message(str(msg.pop("result")))
tools = load_tools(["llm-math"], llm=llm)
agent = OpenAIFunctionsAgent(llm=llm, tools=tools, prompt=prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
return_intermediate_steps=True,
)
return agent_executor
def chat(question, chat_history):
llm = ChatOpenAI(
model="gpt-3.5-turbo-16k",
temperature=0
)
agent = get_agent(llm, chat_history=chat_history)
return agent.invoke(
#{"input": question, "chat_history": chat_history}
{"input": question}
)
chat関数の末尾で、agent.invokeにchat_historyを与える必要はない気がしたので省いています（これでも動きました。）
計算を助ける対話エージェントになっています。
まず、toolと関係ない入力をしてみます。
res = chat("こんにちは!", [])
print("\n\n",res)
> Entering new AgentExecutor chain...
こんにちは！計算を手伝うことができます。どのような計算をお手伝いできますか？
> Finished chain.
{'input': 'こんにちは!', 'chat_history': [HumanMessage(content='こんにちは!', additional_kwargs={}, example=False), AIMessage(content='こんにちは！計算を手伝うことができます。どのような計算をお手伝いできますか？', additional_kwargs={}, example=False)], 'output': 'こんにちは！計算を手伝うことができます。どのような計算をお手伝いできますか？', 'intermediate_steps': []}
OKです。
次にtoolと関係ある入力をしてみます。
res = chat("1+1は？", [])
print("\n\n",res)
> Entering new AgentExecutor chain...
Invoking: `Calculator` with `1+1`
Answer: 21+1は2です。
> Finished chain.
{'input': '1+1は？', 'chat_history': [HumanMessage(content='1+1は？', additional_kwargs={}, example=False), AIMessage(content='', additional_kwargs={'function_call': {'name': 'Calculator', 'arguments': '{\n "__arg1": "1+1"\n}'}}, example=False), FunctionMessage(content='Answer: 2', additional_kwargs={}, name='Calculator'), AIMessage(content='1+1は2です。', additional_kwargs={}, example=False)], 'output': '1+1は2です。', 'intermediate_steps': [(_FunctionsAgentAction(tool='Calculator', tool_input='1+1', log='\nInvoking: `Calculator` with `1+1`\n\n\n', message_log=[AIMessage(content='', additional_kwargs={'function_call': {'name': 'Calculator', 'arguments': '{\n "__arg1": "1+1"\n}'}}, example=False)]), 'Answer: 2')]}
OKです。
最後に履歴を考慮しないと答えられない質問を与えてみます。
履歴で1+1を尋ねた上で質問で「さらに1足すと？」と聞いているので、回答として3が期待されます。
res = chat("さらに1足すと？", [{"question": "1+1は？"}, {"result": "2です"}])
print("\n\n",res)
> Entering new AgentExecutor chain...
Invoking: `Calculator` with `1+1+1`
Answer: 31+1+1は3です。
> Finished chain.
{'input': 'さらに1足すと？', 'chat_history': [HumanMessage(content='1+1は？', additional_kwargs={}, example=False), AIMessage(content='2です', additional_kwargs={}, example=False), HumanMessage(content='さらに1足すと？', additional_kwargs={}, example=False), AIMessage(content='', additional_kwargs={'function_call': {'name': 'Calculator', 'arguments': '{\n "__arg1": "1+1+1"\n}'}}, example=False), FunctionMessage(content='Answer: 3', additional_kwargs={}, name='Calculator'), AIMessage(content='1+1+1は3です。', additional_kwargs={}, example=False)], 'output': '1+1+1は3です。', 'intermediate_steps': [(_FunctionsAgentAction(tool='Calculator', tool_input='1+1+1', log='\nInvoking: `Calculator` with `1+1+1`\n\n\n', message_log=[AIMessage(content='', additional_kwargs={'function_call': {'name': 'Calculator', 'arguments': '{\n "__arg1": "1+1+1"\n}'}}, example=False)]), 'Answer: 3')]}
正しく3を回答できました。
おわりに
chat-langchainの実装を参考にできたおかげで、わりとあっさりOpenAIFunctionsAgentを対話履歴を考慮して使うことができました。
これでチャットボット開発が捗りそうです。