近年、大規模言語モデル（LLM）の応用として、LLM ベース自律的 AI エージェント（以下 AI エージェント）が注目を集めています。この記事では、KARAKURI LM ベースの AI エージェントを動かして遊ぶ方法を紹介していきます！
AI エージェントとは？
自律的 AI エージェントとは、自分で考えて行動を繰り返すことでタスクを自律的に遂行するシステムのことです。このうち自分で考えて行動を選択する部分を AI が担っています。この AI として LLM を使用したのが、LLM ベース AI エージェントです。
https://www.promptingguide.ai/research/llm-agents
一般に、エージェントの行動は「ツール」として事前に定義されています。エージェントはこのツールを呼び出すことで行動を実行します。また、どのツールをどのような順番で呼び出すかは事前に人間が指示しません。人間はただ達成したいタスクを与えるだけで、それを元に LLM が現在の状況に応じて呼び出すべきツールを適切に判断することで人間に与えられたタスクを遂行します。
近年では AI エージェントは一般的になっており沢山の応用事例がありますが、初期的な試みとして有名なものとしては、汎用 AI エージェントを目指した AutoGPT や BabyAGI などがあります。
AI エージェントを作る
[注意！]
今回作成するエージェントはファイルの作成もします。重要なファイルがある場所などでやると予期せぬことになる可能性もあるので、その点は気をつけてください！
それでは実際に AI エージェントを作っていきましょう！今回は Ollama と LangChain を使ってエージェントを実装していきます。 Ollama はローカル環境で LLM を実行するためのツールです。LangChain は LLM を基盤にしたアプリケーションを構築するための有名なフレームワークの一つです。AI エージェントを実装に役立つモジュールも提供しているので、LangChain を使うことで簡単に AI エージェントを実装することができます。なお、エージェント実装を容易にするフレームワークには他にも AutoGen、Semantic Kernel、CrewAI、LlamaIndexなどがあります。
今回作るエージェントには、顧客の情報が与えられた csv ファイルを与えます。そして、1. ディレクトリ内のファイルの一覧表示、2. ファイルの読み込み、3. ファイルの書き出し、の 3 つの単純なツールを用いて、顧客情報を読み取りそこで気づいたことについて報告させるのが今回のタスクです。
具体的には、以下のような指示を与えます。
f"あなたには{os.getcwd()}/data/ に格納されたデータが与えられています。"
f"ここにある顧客データを読み取り気づいたデータのパターンを "
f"{os.getcwd()}/files/ にマークダウンファイルとして報告してください。"
今回使用するデータとしては Kaggle のサイトから、顧客の情報（Mall_Customers.csv）とオースティンの天気(austin_weather.csv)の二つの csv ファイルをダウンロードし、このファイルの中の一部を抜粋したものを ./data に格納しました。実際に使うのは Mall_Customers.csv だけなのですが、複数のファイルから関連するデータを見つけられることを確認するために、今回のタスクには関係のない austin_weather.csv もダウンロードしています。
準備
リポジトリの取得
この Techblog で必要なファイルやパッケージの情報を https://github.com/t46/karakuri-techblog-agent.git に公開していますので、まずはこのリポジトリをクローンしてきてください。特に断りの無い限り、以下ではこの中で作業をしていきます。
git clone https://github.com/t46/karakuri-techblog-agent.git
cd karakuri-techblog-agent
パッケージのインストール
パッケージと仮想環境の管理に今回は rye を使っていきます。rye が入ってない方はインストールしてください。
curl -sSf https://rye.astral.sh/get | bash
source "$HOME/.rye/env"
すでにパッケージの依存関係が記載された lock ファイルがリポジトリに同梱されていますので、以下のコマンドで仮想環境の作成とパッケージのインストールを実行できます。
rye pin 3.11
rye sync
以上でパッケージのインストールは終わりです！
KARAKURI LM の準備
次に KARAKURI LM を動かせるように準備をします。
KARAKURI LM の中で、ツール使用などに対応しているのが karakuri-lm-8x7b-instruct-v0.1なのでこれを使っていきます。
ただ、そのままだと大きすぎるので、mmnga さんが作ってくださった量子化モデルを使用します。今回は特に karakuri-lm-8x7b-instruct-v0.1-Q5_K_M を使うことにします。
wget https://huggingface.co/mmnga/karakuri-lm-8x7b-instruct-v0.1-gguf/resolve/main/karakuri-lm-8x7b-instruct-v0.1-Q5_K_M.gguf
Ollama の準備
KARAKURI LM をダウンロードしたら、これをOllama で 動かせる様にします。Ollama をまだインストールしていない方は、以下のコマンドでインストールしてください。
curl -fsSL https://ollama.com/install.sh | sh
Ollama では Modelfile と呼ばれるファイルにプロンプトテンプレートの情報を書き込み、これを元にモデルを作成します。
今回はとりあえず Hugging Face の karakuri-lm-8x7b-instruct-v0.1 のページに書かれているプロンプトテンプレートと、KARAKURI LM のプロンプトのベースになっている command R+ の Modelfile を参考に、簡易的な Modelfile を作成しました。
まず、Modelfile の FROM... の部分を、ダウンロードしてきた karakuri-lm-8x7b-instruct-v0.1-Q5_K_M.gguf へのパスに変更してください。
FROM /root/karakuri-techblog-agent/karakuri-lm-8x7b-instruct-v0.1-Q5_K_M.gguf
次に、この Modelfile を使って、以下のコマンドで Ollama で使用可能なモデルを作成します。
ollama create karakuri-lm-8x7b-instruct-v0.1-Q5_K_M -f ./Modelfile
これで Ollama で実行可能なモデルが準備できました。
最後に、Ollama サーバーを起動しておいてください。
ollama serve
環境変数の設定
.env.example を用意していますので、これをコピーして .env ファイルを作成します。
cp .env.example .env
LANGCHAIN_API_KEY には LangSmith の API キーを設定します。まだ取得されていない方は、API key を取得しておいてください。
LangSmith の API key の取得方法
ディレクトリの作成
エージェントの最終的な出力を files ディレクトリに出力させるように指示をしますので、ディレクトリを事前に作成しておきます。
mkdir files
実装する
いよいよエージェントの実装に入ります。LangChain は急速にアップデートが進んでいるフレームワークでありそれぞれごとに推奨される記法が異なりますが、今回は今年の 5 月にリリースされた LangChain v0.2 をもとに作成していきます。
エージェント関係で LangChain v0.1 以前と v0.2 以降で変わったことの一つが、LangGraph がエージェントを実装する際に推奨される方法になったという点です。それまでは AgentExecutor というクラスがエージェント実装の多くを担っていましたが、LangGraph ベースになりエージェントのカスタマイズがより容易になったとのことです。せっかくなので今回はこちらの LangGraph ベースで作成します。なお、ここで実装する内容はリポジトリに run_karakuri_agent.py としてアップロードしています。
事前準備
必要なパッケージをインストールし、設定した環境変数を load_dotenv() で呼び出します。
# 事前準備
import os
import sqlite3
from typing import Optional
from dotenv import load_dotenv
from langchain.tools import StructuredTool
from langchain_core.messages import HumanMessage
from langchain_ollama import ChatOllama
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent
from langsmith import traceable
# .env ファイルを読み込む
load_dotenv()
ツールの準備
次に、エージェントが使うことができるツールを準備します。今回は簡単のため 1. ディレクトリ内のファイルを一覧表示するツール、2. ファイルを読み込むツール、3. ファイルを書き出すツール、の 3 つだけを準備します。まずは通常 python の関数を書くのと同様に関数を定義します。そして StructuredTool を用いてそれらをエージェントが処理しやすい形にします。
# ツールの定義
@traceable(name="list_files_in_directory")
def list_files_in_directory(directory: Optional[str] = "./data") -> str:
try:
file_list = []
for root, dirs, files in os.walk(directory):
for file in files:
file_list.append(os.path.join(root, file))
return "\n".join(file_list)
except Exception as e:
return f"Error listing files: {str(e)}"
@traceable(name="read_data_file")
def read_data_file(data_file_path: str) -> str:
with open(data_file_path, "r") as file:
return file.read()
@traceable(name="write_file")
def write_file(file_path: str, content_of_file: str) -> str:
with open(file_path, "w") as file:
file.write(content_of_file)
return f"File {file_path} has been created."
# ツールの定義
tools = [
StructuredTool.from_function(
name="list_files_in_directory",
func=list_files_in_directory,
description="ディレクトリ内のファイルを一覧表示します。",
),
StructuredTool.from_function(
name="read_data_file",
func=read_data_file,
description="ファイルを読み込みます。",
),
StructuredTool.from_function(
name="write_file",
func=write_file,
description="ファイルを作成します。",
),
]
エージェントの作成
次に、エージェントを作成します。これは LangChain/LangGraph を使うとたった 3 行で書けます。
# エージェントの作成
memory = SqliteSaver(sqlite3.connect(":memory:", check_same_thread=False))
model = ChatOllama(model="karakuri-lm-8x7b-instruct-v0.1-Q5_K_M", request_timeout=240.0)
agent_executor = create_react_agent(model, tools, checkpointer=memory, debug=False)
エージェントの実行
エージェントやツールの準備ができたので、実際にエージェントを実行します。上で述べたように、今回はある顧客情報が書かれた csv ファイルを自分で読み取り、その内容を報告させます。具体的には以下のような指示を与えます：
f"あなたには{os.getcwd()}/data/ に格納されたデータが与えられています。ここにある顧客データを読み取り気づいたデータのパターンを {os.getcwd()}/files/ にマークダウンファイルとして報告してください。"
# エージェントの実行
config = {"configurable": {"thread_id": "abc123"}}
@traceable(name="run_agent")
def run_agent(query):
for chunk in agent_executor.stream(
{"messages": [HumanMessage(content=query)]}, config
):
print(chunk)
print("----")
if __name__ == "__main__":
query = (
f"あなたには{os.getcwd()}/data/ に格納されたデータが与えられています。"
f"ここにある顧客データを読み取り気づいたデータのパターンを "
f"{os.getcwd()}/files/ にマークダウンファイルとして報告してください。"
)
run_agent(query)
これで実装は完了です！これらをまとめたファイル全体は以下のようになります。
# 事前準備
import os
import sqlite3
from typing import Optional
from dotenv import load_dotenv
from langchain.tools import StructuredTool
from langchain_core.messages import HumanMessage
from langchain_ollama import ChatOllama
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent
from langsmith import traceable
# .env ファイルを読み込む
load_dotenv()
# ツールの定義
@traceable(name="list_files_in_directory")
def list_files_in_directory(directory: Optional[str] = "./data") -> str:
try:
file_list = []
for root, dirs, files in os.walk(directory):
for file in files:
file_list.append(os.path.join(root, file))
return "\n".join(file_list)
except Exception as e:
return f"Error listing files: {str(e)}"
@traceable(name="read_data_file")
def read_data_file(data_file_path: str) -> str:
with open(data_file_path, "r") as file:
return file.read()
@traceable(name="write_file")
def write_file(file_path: str, content_of_file: str) -> str:
with open(file_path, "w") as file:
file.write(content_of_file)
return f"File {file_path} has been created."
# ツールの定義
tools = [
StructuredTool.from_function(
name="list_files_in_directory",
func=list_files_in_directory,
description="ディレクトリ内のファイルを一覧表示します。",
),
StructuredTool.from_function(
name="read_data_file",
func=read_data_file,
description="ファイルを読み込みます。",
),
StructuredTool.from_function(
name="write_file",
func=write_file,
description="ファイルを作成します。",
),
]
# エージェントの作成
memory = SqliteSaver(sqlite3.connect(":memory:", check_same_thread=False))
model = ChatOllama(model="karakuri-lm-8x7b-instruct-v0.1-Q5_K_M", request_timeout=240.0)
agent_executor = create_react_agent(model, tools, checkpointer=memory, debug=False)
# エージェントの実行
config = {"configurable": {"thread_id": "abc123"}}
@traceable(name="run_agent")
def run_agent(query):
for chunk in agent_executor.stream(
{"messages": [HumanMessage(content=query)]}, config
):
print(chunk)
print("----")
if __name__ == "__main__":
query = (
f"あなたには{os.getcwd()}/data/ に格納されたデータが与えられています。"
f"ここにある顧客データを読み取り気づいたデータのパターンを "
f"{os.getcwd()}/files/ にマークダウンファイルとして報告してください。"
)
run_agent(query)
AI エージェントを動かす
エージェントの実行
いよいよエージェントを動かします！
まず、エージェントの出力が格納できるように files というディレクトリを作成します。
mkdir files
そしたら、あとは1行で python ファイルを実行するだけです！
rye run python run_karakuri_agent.py
結果
これを実行すると、うまくいった場合はこのような報告結果がまとまったマークダウンファイルが作成されます。（今回の Markdwon やツールの設定は必ずしも最適ではないためうまくいかない時も結構あります。）
この報告書を見ると、顧客の年齢に偏りがあること、年齢と出費額には相関がなさそうであること、などをエージェントが自分で発見したことがわかります。
LangSmith
今回は LangSmith を使ったので、エージェントの実際の行動選択のログをブラウザ上で確認することができます。
ブラウザのアドレスバーに https://smith.langchain.com/projects と打ち込んでこちらのページに飛ぶと、Projects と表示されたページが表示されるので、karakuri-agent をクリックします。
すると次のような画面が表示されます。「TRACE」というところにエージェントが行った処理を追跡した結果が表示されています。そして各ステップをクリックすると処理の詳細を見ることができます。
例えばこの画面では、エージェントが read_file 関数をツールとして呼び出し、 「Input」を見るとわかるように、それに対して .../karakuri-agent/data/Mall_Customers.csv を入力したことがわかります。そしてこの csv ファイルを読み取った結果の内容が「Rendered Output」に表示されています。
サンプル実行結果はこちらから見ることができます。
この結果を見ると、エージェントは（1回余計にやっていますが）まずデータファイルを探し（list_files_in_directory）、データを読み（read_data_file）、読み取った結果をマークダウンファイルとしてまとめている（write_file）ことがわかります。これらをどのような順番で実施するかを人間は与えていないにもかかわらず適切にやるべきことを考えて順次実行できているというエージェントの柔軟性が少し感じられたかなと思います！
【おまけ】AI エージェントのゼロからの実装
LangChain などはエージェントを非常に簡単に実装することができますが、その反面内部の処理が見えづらくなっています。そこで、エージェントがどのように動いているのかの雰囲気を掴んでもらうために、from scratch で実装するとどんな感じになるかを簡単に説明します。
今回は OpenAI の API を使用するので、.env ファイルを開いて、OPENAI_API_KEY を追加してください。
OPENAI_API_KEY=...
OPENAI_API_KEY をまだ取得していない人は公式ページの指示に従って取得してください。
こちらが、非常に単純な AI エージェントの from scratch での実装例です。これをベースに解説をしていきます。
from typing import Dict, Callable
from openai import OpenAI
from dotenv import load_dotenv
import re
# .env ファイルを読み込む
load_dotenv()
PROMPT_TEMPLATE = '''Answer the following questions as best you can. You have access to the following tools:
{formatted_tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {question}
Thought:'''
class Agent:
def __init__(self, tools: Dict[str, Callable]):
self.tools = tools # ツール名と関数の辞書
self.tool_names = ", ".join(tools.keys()) # ツール名のリスト
self.formatted_tools = self._format_tool_descriptions() # ツール名と説明の並んだ文字列
self.prompt_template = PROMPT_TEMPLATE
self.client = OpenAI()
def _format_tool_descriptions(self) -> str:
return "\n".join([f"{name}: {func.__doc__ or 'No description available.'}" for name, func in self.tools.items()])
def _parse_agent_response(self, response: str) -> tuple:
thought_match = re.search(r"Thought: (.+?)(?=\nAction:|$)", response, re.DOTALL)
action_match = re.search(r"Action: (\w+)", response)
action_input_match = re.search(r"Action Input: (.+)", response)
final_answer_match = re.search(r"Final Answer: (.+)", response)
thought = thought_match.group(1).strip() if thought_match else None
action = action_match.group(1) if action_match else None
action_input = action_input_match.group(1) if action_input_match else None
final_answer = final_answer_match.group(1) if final_answer_match else None
return thought, action, action_input, final_answer
def run(self, question: str, max_iterations: int = 5) -> str:
prompt = self.prompt_template.format(
formatted_tools=self.formatted_tools,
tool_names=self.tool_names,
question=question
)
for _ in range(max_iterations):
response = self.client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=[
{"role": "user", "content": prompt},
],
max_tokens=150,
stop=["Observation:", "Human:"]
)
agent_response = response.choices[0].message.content.strip()
thought, action, action_input, final_answer = self._parse_agent_response(agent_response)
if final_answer:
return final_answer
if action and action_input:
if action in self.tools:
observation = self.tools[action](action_input)
else:
observation = f"Error: {action} is not a valid tool. Valid tools are {self.tool_names}"
else:
observation = "Error: Invalid response format. Please provide both Action and Action Input."
prompt += f"{thought}\nAction: {action}\nAction Input: {action_input}\nObservation: {observation}\nThought:"
return "Error: Maximum iterations reached without a final answer."
if __name__ == "__main__":
# ツールの定義
def search(query: str) -> str:
"""Search the web for information."""
# 実際の検索機能の代わりにダミー関数を使用
return f"Search results for: {query}"
def calculator(expression: str) -> str:
"""Calculate the result of a mathematical expression."""
try:
return str(eval(expression))
except:
return "Error: Invalid expression"
tools = {
"Search": search,
"Calculator": calculator,
}
agent = Agent(tools)
question = "What is the 256 plus 16?"
answer = agent.run(question)
print(f"Question: {question}")
print(f"Answer: {answer}")
上で述べたように、LLM ベース AI エージェントとは行動を繰り返してタスクを達成するLLMベースのシステムです。
まず、AI エージェントにはプロンプトが与えられます。プロンプトでは、1. 達成するべきタスク（あるいは質問）、2. どのような形式でそれに答えるか、3. タスクに答えるために使えるツール、が記述されます。
ある基本的な AI エージェントに与えられるプロンプトの例を以下に載せています。このエージェントの例では、ある質問(Question)が与えられると、それに対して何をすべきかについての考え（Thought）、そしてそれに基づいた取るべき行動（Action）、を出力するという形式を取るように指示されています。そして前述の通り行動はツール、つまり関数で定義されてますので、行動には関数名が入ります。そして関数は引数を取りますのでその入力（Action Input）を次に出すように指示されています。その関数の実行結果である行動の結果（Observation）が取得できて、それを踏まえて次の考えを出して、ということを N 回繰り返して質問に答えていきます。質問に答えるために行動として使えるツールは {formatted_tools} に与えられています。
'''
以下の質問に最善を尽くして答えてください。あなたは以下のツールを使用できます：
{formatted_tools}
以下の形式を使用してください：
Question: あなたが答えなければならない入力された質問
Thought: 何をすべきか常に考えてください
Action: 取るべき行動、[{tool_names}]のいずれかである必要があります
Action Input: 行動に対する入力
Observation: 行動の結果
...（この思考/行動/行動入力/観察のパターンはN回繰り返すことができます）
Thought: これで最終的な答えがわかりました
Final Answer: 元の入力質問に対する最終的な答え
始めましょう！
Question: {question}
Thought:
'''
それでは、このプロンプトを与えられて LLM が実際にどの様にして質問に答えていくのかをコードを見ながら解説していきます。
まず、先ほどのプロンプトが与えられると LLM は出力を返します。ここで、LLM は Action と Action Input の内容は出力しますが、Obaservation には実際の行動結果（ツールにツールの引数を入れた出力）を代入することに注意してください。そのために、stop word に "Observation:" を指定することで、LLM の応答を Observation: という単語が出てきたタイミングで止めています。
response = self.client.chat.completions.create(
model="gpt-3.5-turbo-0125",
messages=[
{"role": "user", "content": prompt},
],
max_tokens=150,
stop=["Observation:"]
)
Observation を得るために、まずは LLM の出力をパースして "Action:" や "Action Input:" と書かれている部分を抽出します。
def _parse_agent_response(self, response: str) -> tuple:
thought_match = re.search(r"Thought: (.+?)(?=\nAction:|$)", response, re.DOTALL)
action_match = re.search(r"Action: (\w+)", response)
action_input_match = re.search(r"Action Input: (.+)", response)
final_answer_match = re.search(r"Final Answer: (.+)", response)
thought = thought_match.group(1).strip() if thought_match else None
action = action_match.group(1) if action_match else None
action_input = action_input_match.group(1) if action_input_match else None
final_answer = final_answer_match.group(1) if final_answer_match else None
return thought, action, action_input, final_answer
agent_response = response.choices[0].message.content.strip()
thought, action, action_input, final_answer = self._parse_agent_response(agent_response)
そして Action Input を引数として Action を実際に実行することで Observation を得ます。
if action and action_input:
if action in self.tools:
observation = self.tools[action](action_input)
else:
observation = f"Error: {action} is not a valid tool. Valid tools are {self.tool_names}"
else:
observation = "Error: Invalid response format. Please provide both Action and Action Input."
Final Answer が生成されたら終了です。
if final_answer:
return final_answer
Final Answer が出力されるまで、あるいは定められた回数までこれを繰り返します
for _ in range(max_iterations):
プロンプトに、とった行動/行動への入力/観測/思考を追加していくことで、過去にどのようなステップを踏んだかを踏まえて、次の思考を決めることができます。
prompt += f"{thought}\nAction: {action}\nAction Input: {action_input}\nObservation: {observation}\nThought:"
以上がエージェントの非常にシンプルな処理の流れです！エージェントがどのようにして動いているか少しイメージを掴んでいただけたら幸いです！
Discussion