Zenn
Open8

LangChain Expression Language勉強メモ

mah_labmah_lab

最大トークン越えの場合の例外処理

from langchain.chat_models import ChatOpenAI
from openai import BadRequestError

openai_llm = ChatOpenAI(model="gpt-4")

try:
    print(openai_llm.invoke(f"次の文章を要約してください: {text}"))
except BadRequestError as error:
    message = error.response.json()["error"]["message"]
    print("Hit error: ", message)

Hit error:  This model's maximum context length is 8192 tokens. However, your messages resulted in 36534 tokens. Please reduce the length of the messages.
mah_labmah_lab

LCELでAgentを作成するための基本的な要素

  • ツールの作成／ロード：ビルドインツールやカスタムツールを読み込む。
  • ツールレンダリング：ツールの定義情報をプロンプト内にレンダリング。
  • 出力パーサの設定：AgentAction/AgentFinishでレスポンスする内容のフォーマット。
  • スクラッチパッドのフォーマット：エージェントのタイプに従って、「中間ステップ」の内容を適切なフォーマットに変換する。「中間ステップ」は配列で、エージェントの実行ステップ毎にLLMのアウトプットとツールのアウトプットとのタプルが追加される。
  • プロンプト：LangChain Hubからロードするか、テンプレートを作成する。
  • メモリ：チャット履歴の保存。ConversationBufferMemoryとか使う。
  • エージェント実行：whileループで中間ステップを手動で追加するか、AgentExecutorを利用する。
mah_labmah_lab

エージェントの基本的な動作

動作の全貌を理解するにはwhileループでエージェントを動かしたときのコードを見るのが早い。AgentExecutorの例を見ると、やっぱり中がブラックボックスなので分からなくなる。

from langchain.schema.agent import AgentFinish
from langchain.agents import tool
from langchain.tools.render import format_tool_to_openai_function
from langchain.agents.format_scratchpad import format_to_openai_functions
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chat_models import ChatOpenAI

# 文字を数えるツールを定義
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)

tools = [get_word_length]

# プロンプトを定義
# agent_scratchpadという変数名で中間ステップをプロンプトに突っ込めるようにする
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a useful assistant."),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# GPT-4をfunction callingで呼び出せるようにする
llm = ChatOpenAI(temperature=0, model='gpt-4-0613')
llm_with_tools = llm.bind(
    functions=[format_tool_to_openai_function(t) for t in tools]
)

# LCELでユーザー入力と中間ステップをプロンプトに埋め込んで処理するように定義
agent = {
    "input": lambda x: x["input"],
    "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'])
} | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()

# 中間ステップはここに保存
intermediate_steps = []

# エージェントループ
while True:
    output = agent.invoke({
        "input": "how many letters in the word zundamon?",
        "intermediate_steps": intermediate_steps
    })
    # 処理が完了したらAgentFinishのインスタンスが返る
    if isinstance(output, AgentFinish):
        final_result = output.return_values["output"]
        break
    else:
        print(output.tool, output.tool_input)
        tool = {
            "get_word_length": get_word_length
        }[output.tool]
        observation = tool.run(output.tool_input)
        intermediate_steps.append((output, observation))

print(final_result)

実行すると以下が出力される：

get_word_length {'word': 'zundamon'}
The word "zundamon" has 8 letters.
mah_labmah_lab

エラー処理を含めたエージェントの実装例

ローカルのテキストファイルについて処理するエージェントを定義。テキストファイルの中身が最大コンテキスト長以上でAPIエラーが返ってきたときでも、エラー内容からLLMが分かりやすいメッセージを返すようにしている。

こういうエラー処理もwith_fallbacksを使ってカッコ良く書けるかな？と思ったけど、with_fallbacksではフォールバック先にたらい回しするだけだったので、肝心のエラーメッセージを引数に繋げることができなかった。BadRequestErrorのときに、より大きなコンテキスト長を処理できるモデルにすげ替える、といったことはできる。

from langchain.schema.agent import AgentFinish
from langchain.tools.render import format_tool_to_openai_function
from langchain.agents.format_scratchpad import format_to_openai_functions
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.schema.output_parser import StrOutputParser
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chat_models import ChatOpenAI
from langchain.agents.agent_toolkits import FileManagementToolkit
from pathlib import Path

# ワーキングディレクトリを指定
working_directory = Path('./working')

# ファイルの読み書きのためのツールを設定
tools = FileManagementToolkit(
    root_dir=str(working_directory.name),
    selected_tools=["read_file", "write_file", "list_directory"],
).get_tools()

# エージェントプロンプト
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a useful assistant."),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# エラーメッセージを説明するためのプロンプト
fallback_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are the AI that tells the user what the error is in plain Japanese. Since the error occurs at the end of the step, you must guess from the process flow and the error message, and communicate the error message to the user in an easy-to-understand manner."),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# モデルを設定
llm = ChatOpenAI(temperature=0, model='gpt-4-0613')
llm_with_tools = llm.bind(
    functions=[format_tool_to_openai_function(t) for t in tools]
)

# エージェントを設定
agent = {
    "input": lambda x: x["input"],
    "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'])
} | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()

# エラーが発生した場合のLLMを設定
fallback_chain = {
    "input": lambda x: x["input"],
    "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'][:-1])
} | fallback_prompt | llm | StrOutputParser()

intermediate_steps = []

while True:
    try:
        output = agent.invoke({
            "input": "作業用ディレクトリ内にある.txtファイルを読み出し、箇条書きで要約した後に、[ファイル名].summarize.txtという名前で保存して下さい。",
            "intermediate_steps": intermediate_steps
        })
    except BadRequestError as error:
        message = error.response.json()["error"]["message"]
        # intermediate_stepsの最後のステップのタプルのobservationを空にする
        intermediate_steps[-1] = (intermediate_steps[-1][0], "")
        final_result = fallback_chain.invoke({
            "input": message,
            "intermediate_steps": intermediate_steps
        })
        break
    if isinstance(output, AgentFinish):
        final_result = output.return_values["output"]
        break
    else:
        print('== selected tool ==')
        print(output.tool, output.tool_input, "\n")
        read_file, write_file, list_directory = tools
        tool = {
            "read_file": read_file,
            "write_file": write_file,
            "list_directory": list_directory,
        }[output.tool]
        observation = tool.run(output.tool_input)
        print('== observation ==')
        if output.tool == "read_file":
            print('word counts: ', len(observation))
        else:
            print(observation)
        print("\n")
        intermediate_steps.append((output, observation))

print('== FINISH ==')
print(final_result)

実行結果：

== selected tool ==
list_directory {'dir_path': '.'} 

== observation ==
20231106.txt
20231027.txt

== selected tool ==
read_file {'file_path': '20231106.txt'} 

== observation ==
word counts:  37613

== FINISH ==
このモデルの最大コンテキスト長は8192トークンですが、あなたのメッセージは合計で36783トークン（メッセージ内で36651トークン、関数内で132トークン）になっています。メッセージや関数の長さを短くしてください。
mah_labmah_lab

ジェネレータで呼び出せるようにしてみる

外から呼び出しにくい実装になっているのでfor文でループを回せるようジェネレータで実装してみる。

from langchain.schema.agent import AgentFinish
from langchain.tools.render import format_tool_to_openai_function
from langchain.agents.format_scratchpad import format_to_openai_functions
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.schema.output_parser import StrOutputParser
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chat_models import ChatOpenAI
from langchain.agents.agent_toolkits import FileManagementToolkit
from pathlib import Path

class Agent:
    def __init__(self, input_message):
        self.input_message = input_message
        self.intermediate_steps = []

        # ワーキングディレクトリを指定
        working_directory = Path('./working')

        # ファイルの読み書きのためのツールを設定
        self.tools = FileManagementToolkit(
            root_dir=str(working_directory.name),
            selected_tools=["read_file", "write_file", "list_directory"],
        ).get_tools()

        # エージェントプロンプト
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a useful assistant."),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])

        # エラーメッセージを説明するためのプロンプト
        fallback_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are the AI that tells the user what the error is in plain Japanese. Since the error occurs at the end of the step, you must guess from the process flow and the error message, and communicate the error message to the user in an easy-to-understand manner."),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        # モデルを設定
        llm = ChatOpenAI(temperature=0, model='gpt-4-0613')
        llm_with_tools = llm.bind(
            functions=[format_tool_to_openai_function(t) for t in tools]
        )

        # エージェントを設定
        self.agent = {
            "input": lambda x: x["input"],
            "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'])
        } | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()

        # エラーが発生した場合のLLMを設定
        self.fallback_chain = {
            "input": lambda x: x["input"],
            "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'][:-1])
        } | fallback_prompt | llm | StrOutputParser()

    def run(self):
        while True:
            try:
                output = self.agent.invoke({
                    "input": self.input_message,
                    "intermediate_steps": self.intermediate_steps
                })
            except BadRequestError as error:
                message = error.response.json()["error"]["message"]
                self.intermediate_steps[-1] = (self.intermediate_steps[-1][0], "")
                final_result = self.fallback_chain.invoke({
                    "input": message,
                    "intermediate_steps": self.intermediate_steps
                })
                yield final_result
                return

            if isinstance(output, AgentFinish):
                yield output.return_values["output"]
                return
            else:
                messages = [
                    "== selected tool ==",
                    output.tool,
                    str(output.tool_input)
                ]
                read_file, write_file, list_directory = self.tools
                tool = {
                    "read_file": read_file,
                    "write_file": write_file,
                    "list_directory": list_directory,
                }[output.tool]
                observation = tool.run(output.tool_input)
                messages += [
                    "== observation ==",
                    f"word counts: {len(observation)}" if output.tool == "read_file" else observation
                ]
                self.intermediate_steps.append((output, observation))
                yield "\n".join(messages)

こんな感じで呼び出す。

input_message = "作業用ディレクトリ内にある.txtファイルを読み出し、箇条書きで要約した後に、[ファイル名].summarize.txtという名前で保存してください。"
agent = Agent(input_message)
step_count = 0

# エージェントを実行
for step in agent.run():
    step_count += 1
    print(f"***** step {step_count} *****")
    print(step)
    print("\n")

実行結果：

***** step 1 *****
== selected tool ==
list_directory
{'dir_path': '.'}
== observation ==
20231106.txt
20231027.txt

***** step 2 *****
== selected tool ==
read_file
{'file_path': '20231106.txt'}
== observation ==
word counts: 37613

***** step 3 *****
このモデルの最大コンテキスト長は8192トークンですが、あなたのメッセージは合計で36782トークン（メッセージ内で36650トークン、関数内で132トークン）になっています。メッセージや関数の長さを短くしてください。
mah_labmah_lab

会話型のエージェントにしてみる

イマドキチャットができないエージェントなんて流行らない。というわけで会話履歴を持つように。

https://python.langchain.com/docs/expression_language/cookbook/memory

この辺の例を参照するとRunnablePassthroughだったりRunnableLambdaを使っているけど、よく分からない。普通に無名関数でself.memory.load_variables()を呼び出すことで実装できたが、後で違いを調べてみる。
（ドキュメント読んでてもいきなり説明なしに新しいクラスが出てきたりして、心象が良くない）

from langchain.schema.agent import AgentFinish
from langchain.tools.render import format_tool_to_openai_function
from langchain.agents.format_scratchpad import format_to_openai_functions
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain.schema.output_parser import StrOutputParser
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chat_models import ChatOpenAI
from langchain.agents.agent_toolkits import FileManagementToolkit
from langchain.memory import ConversationBufferMemory
from openai import BadRequestError
from pathlib import Path


class Agent:
    def __init__(self):
        self.intermediate_steps = []

        # ワーキングディレクトリを指定
        working_directory = Path('./working')

        # ファイルの読み書きのためのツールを設定
        self.tools = FileManagementToolkit(
            root_dir=str(working_directory.name),
            selected_tools=["read_file", "write_file", "list_directory"],
        ).get_tools()

        # エージェントプロンプト
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a useful assistant."),
            MessagesPlaceholder(variable_name="chat_history"),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])

        # エラーメッセージを説明するためのプロンプト
        fallback_prompt = ChatPromptTemplate.from_messages([
            ("system",
             "You are the AI that tells the user what the error is in plain Japanese. Since the error occurs at the end of the step, you must guess from the process flow and the error message, and communicate the error message to the user in an easy-to-understand manner."),
            MessagesPlaceholder(variable_name="chat_history"),
            ("user", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])

        # モデルを設定
        llm = ChatOpenAI(temperature=0, model="gpt-4-0613")
        llm_with_tools = llm.bind(
            functions=[format_tool_to_openai_function(t) for t in self.tools]
        )

        # メモリを設定
        self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
        
        # エージェントを設定
        agent_assigns = {
            "input": lambda x: x["input"],
            "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps']),
            "chat_history": lambda x: self.memory.load_memory_variables({})["chat_history"]
        }
        self.agent = agent_assigns | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()

        # エラーが発生した場合のLLMを設定
        fallback_assigns = {
            "input": lambda x: x["input"],
            "agent_scratchpad": lambda x: format_to_openai_functions(
                x['intermediate_steps'][:-1]),
            "chat_history": lambda x: self.memory.load_memory_variables({})["chat_history"]
        }
        self.fallback_chain = fallback_assigns | fallback_prompt | llm | StrOutputParser()

    def run(self, input_message):
        while True:
            try:
                output = self.agent.invoke({
                "input": input_message,
                "intermediate_steps": self.intermediate_steps
            })
            except BadRequestError as error:
                self.intermediate_steps[-1] = (self.intermediate_steps[-1][0], "")
                final_result = self.fallback_chain.invoke({
                    "input": error.response.json()["error"]["message"],
                    "intermediate_steps": self.intermediate_steps
                })
                yield final_result
                return

            if isinstance(output, AgentFinish):
                message = output.return_values["output"]
                self.memory.save_context({"input": input_message}, {"output": message})
                yield message
                return
            else:
                messages = [
                    "== selected tool ==",
                    output.tool,
                    str(output.tool_input)
                ]
                read_file, write_file, list_directory = self.tools
                tool = {
                    "read_file": read_file,
                    "write_file": write_file,
                    "list_directory": list_directory,
                }[output.tool]
                observation = tool.run(output.tool_input)
                messages += [
                    "== observation ==",
                    f"word counts: {len(observation)}" if output.tool == "read_file" else observation
                ]
                self.intermediate_steps.append((output, observation))
                yield "\n".join(messages)

こんな感じで呼び出す。

# エージェントのインスタンスを作成
agent = Agent()

# 一つ目のメッセージでエージェントを実行
for step in agent.run("作業用ディレクトリ内にある.txtファイルをリストアップしてください"):
    print(step)

実行結果：

== selected tool ==
list_directory
{'dir_path': '.'}
== observation ==
20231106.txt
20231027.txt
作業用ディレクトリ内にある.txtファイルは以下の通りです：

1. 20231106.txt
2. 20231027.txt

ちゃんと記憶していることを確認してみる。

for step in agent.run("もう一度ファイルのリストを教えて下さい"):
    print(step)

ツールを使わずに返答しているので、メモリから返答していることが分かる。

作業用ディレクトリ内にある.txtファイルは以下の通りです：

1. 20231106.txt
2. 20231027.txt