こんにちは、CCCMKホールディングス TECH LABの三浦です。
4月になりました。近所の桜の木がようやく開花し、春っぽくなってきたなぁと感じています。桜が咲く頃は何日か雨が続く日があって、いつの間にか花が散ってしまうことが多いので、今年は忘れずに花見に行きたい！と思っています。
エージェント(Agent)は日本語では"代理人"という意味があり、Large Language Model: 大規模言語モデル(LLM)を推論エンジンにしたAgentは複雑なLLMアプリケーションを組むうえでかなり重要な要素だと考えられます。
LLM Agentはユーザーからの要求に対し、与えられたツール(取ることが出来る行動)の中で最適なものをユーザーの代理で選択することが出来ます。今のところLLM単体では外部のツールを実行することは出来ませんが、LLM Agentが選択したツールを実行し、その結果をLLM Agentに渡すことが出来る仕組みがあれば、ユーザーは最初に一回だけ指示を与えるだけで、あとはLLM Agentが代理で複雑なタスクをこなしてくれるようなシステムを作ることが可能になります。
LLM Agentを構築すること自体はそれほど難しいことではないのですが、ツールとの連携や全体を通じた状態の管理など、LLM Agentを組み込んだアプリケーションの構築は複雑になります。
LLM Agentアプリケーションを分かりやすく組むことが出来るライブラリ"LangGraph"をLangChainが開発していて、LangChainを使ったことがある経験があると、そのノウハウをスムーズにLLM Agentアプリケーションの開発に活かすことが出来ます。
最近LangGraphを使って色々と試していたのですが、少しずつコツ(?)を掴めてきたように感じています。今回はLangGraphを使った簡単なLLM Agentアプリケーションの作り方について、まとめてみます。
LangGraph
LangGraphはLangChainをベースにしたステートフルで、LLM Agentなどの複数の要素で構成されたアプリケーションを構築するためのPythonのライブラリです。
https://python.langchain.com/docs/langgraph
その名前からも分かるように、LangGraphではグラフ構造でアプリケーションを定義します。LangGraphではGraph, Node, Edgeがキーとなる要素になります。Graph, Node, Edgeはそれぞれ以下のような役割を持ちます。
Graph
アプリケーションそのものを表現。Node
Messageのリストで表現されたアプリケーションの現在の状態を受け取り、Messageを生成する。Edge
Nodeを接続する。接続されたNode間で状態の受け渡しが行われるようになる。条件に応じて接続先のNodeを切り替える条件付きのEdgeもある。
LangGraphを使った実装例はGithubにも上がっていますが、公式ドキュメントの内容を読むと一通り出来ることを把握することが出来ます。今回はそちらの内容を参考にLangGraphの使い方についてまとめるのですが、その前にもう一つ理解しておくと良い機能があります。
それがOpenAI(Azure OpenAI)のChatモデルに備わっているFunction Callingという機能です。
Function Calling
Function Callingとは、OpenAIのChatモデルを呼び出す際にクエリと一緒に1つ以上のToolの情報を渡すと、その結果としてクエリを解くために必要なToolとそれを呼び出すための引数を自動的に生成してくれる機能です。
※参考
https://platform.openai.com/docs/guides/function-calling
Chatモデルには外部APIなどを直接実行する機能はありませんが、このFunction CallingとToolの実行環境(Executor)を接続することでクエリから外部APIを実行させることが可能になります。
LangChainではFunction Calling機能を次の様に利用することが出来ます。ここではChatモデル(gpt-35-turbo-16k)に2つの整数の積を取る関数をToolとして与えています。
from langchain_core.utils.function_calling import convert_to_openai_tool from langchain_openai import AzureChatOpenAI def multiply(first: int, second: int): """ この関数は引数として2つのint型の値を受け取り、その積を計算し返します。 Args: first(int): 積を取る一つ目のint型の値 second(int): 積を取る二つ目のint型の値 Returns(int): first * second """ return first * second multiply_tool = convert_to_openai_tool(multiply) tools = [multiply_tool] model = AzureChatOpenAI( api_key=os.environ.get("AZURE_OPENAI_API_KEY"), azure_endpoint=os.environ.get("AZURE_OPENAI_API_BASE"), api_version=os.environ.get("AZURE_OPENAI_API_VERSION"), model="gpt-35-turbo-16k" ) model_with_tools = model.bind(tools=tools)
model_with_toolsはユーザーの質問に対し、必要に応じてToolmultiplyを呼び出すための情報を生成します。たとえば、"4と5をかけると？"という質問を与えてみます。
model_with_tools.invoke("4と5をかけると？")
実行結果は以下の様になります。
content='' additional_kwargs={'tool_calls': [{'id': 'call_x...........', 'function': {'arguments': '{\n "first": 4,\n "second": 5\n}', 'name': 'multiply'}, 'type': 'function'}]} ...
additional_kwargsというキーに使用するToolの情報と与える引数の情報が記述されていることが分かります。
Toolはユーザーの質問に対して必要に応じて使用されるため、Toolが必要ないと考えられる場合はToolを使用せず、普通の回答が生成されます。 たとえば、
model_with_tools.invoke("こんにちは")
すると今度はToolの呼び出し情報が生成されていないことが確認出来ます。
content='こんにちは！なにかお手伝いできますか？' response_metadata={'token_usage': {'completion_tokens': 16, ...
Toolと接続されたChatモデルを使うことで、たとえば次のような処理を組むことが出来ます。
import json from langchain_core.messages import ToolMessage def execute_multiply(query: str): """ AzureOpenAIのChatモデルを使って任意のクエリで必要に応じて掛け算を行う。 掛け算をする必要がない場合は普通の応答を返す。 Args: query(str): 入力クエリ Returns: クエリが掛け算の問題なら掛け算の結果、それ以外は応答 """ res = model_with_tools.invoke(query) tools = res.additional_kwargs.get("tool_calls",[]) if len(tools) == 0: return res else: for tool in tools: if tool["function"]["name"] == "multiply": print("---EXECUTE TOOL---") return ToolMessage( tool_call_id=tool["id"], content=str(multiply(**json.loads(tool["function"]["arguments"]))) ) return res
この処理(関数)に、次の様に入力してみます。
execute_multiply("一個100円のリンゴを5買うのに必要な金額は？")
結果を見ると、Toolが使用され、正しい計算結果を得ることが出来ています。
---EXECUTE TOOL--- ToolMessage(content='500', tool_call_id='call_xxxxxxxxxxxxxxxxxxxxxxxxxx')
一方で全然関係のない質問に対しても回答してくれます。
execute_multiply("日本で一番標高の高い山は？")
AIMessage(content='日本で一番標高の高い山は富士山です。標高は3,776メートルです。', response_metadata={'token_usage': {'completion_tokens': 35, ...
このように与えられた関数などのToolに対し、必要に応じて使うべきToolを選択し、それを使うための引数などの情報を生成する機能がFunction Callingです。LangGraphを使う場合、各NodeでFunction Callingによってこういった情報が生成され、Toolに紐づいたNodeに渡され、処理が実行されていきます。この辺りのイメージがあると、LangGraphで何をしているのかを把握することがしやすくなると思います。
LangGraphを使ってみる
それではLangGraphを使ってどのようにアプリケーションを作るのかをまとめてみます。ここで掲載するコードはLangGraphのドキュメントに掲載されているものです。
構築の流れは、まずアプリケーションで使用するToolを定義し、各Nodeの処理内容を実装し、最後にGraphを組み立てる、という内容で進めます。
Toolの定義
先ほどと同様、二つの整数の積を取るToolを使用します。
import json from typing import List from langchain_core.messages import ToolMessage, BaseMessage, HumanMessage from langchain_core.tools import tool from langchain_core.utils.function_calling import convert_to_openai_tool from langgraph.graph import END, MessageGraph @tool def multiply(first_number: int, second_number: int): """Multiplies two numbers together.""" return first_number * second_number
関数に@toolというデコレータを付与することで、簡単にLangChainのToolとして使用することが出来るようになります。例えば以下の様に呼び出すことが可能になります。
multiply.invoke(
{
first_number: 3,
second_number: 4
}
)
Nodeの処理の実装
2つのNodeを実装します。一つはユーザーからの入力を直接受け取り、必要に応じてFunction Callingを実行するLLMを用いたNodeで、もう一つはFunction Callingの実行結果をパースし、実際に関数を実行するNodeです。
def invoke_model(state: List[BaseMessage]): """Function Callingを実行する""" model_with_tools = model.bind( tools=[convert_to_openai_tool(multiply)] ) return model_with_tools.invoke(state) def invoke_tool(state: List[BaseMessage]): """stateに含まれる直近のmessageをパース、関数multiplyを実行する""" tool_calls = state[-1].additional_kwargs.get("tool_calls", []) multiply_call = None for tool_call in tool_calls: if tool_call.get("function").get("name") == "multiply": multiply_call = tool_call if multiply_call is None: raise Exception("No adder input found.") res = multiply.invoke( json.loads(multiply_call.get("function").get("arguments")) ) return ToolMessage( tool_call_id=multiply_call.get("id"), content=res )
Graphを組み立てる
次に各Nodeを繋げてGraphを組み上げていきます。Graphの構造は次のようになります。
ENDはあらかじめLangGraphに用意されている特殊なNodeで、ここにたどり着くとアプリケーションが終了します。またinvoke_modelがFunction Callingを実行したか否かでinvoke_toolを実行するかどうかの分岐をさせます。(図の赤線の部分です。)
これを実現するため、まずFunction Callingが実行されたかどうかを判定する関数routerを定義し、add_conditional_edgesでrouterの戻り値に応じて次のNodeのルーティングを行います。
# Graphの構築 graph = MessageGraph() graph.add_node("oracle", invoke_model) graph.add_node("multiply",invoke_tool) def router(state: List[BaseMessage]): """invoke_modelでFunction Callingを実行したかどうかを判定する""" tool_calls = state[-1].additional_kwargs.get("tool_calls", []) if len(tool_calls): return "multiply" else: return "end" graph.add_edge("multiply", END) graph.set_entry_point("oracle") ### routerの結果に応じて次のNodeを決める。 graph.add_conditional_edges("oracle", router, { "multiply": "multiply", "end": END, })
アプリケーションの実行
最後にgraphのコンパイルを行い、アプリケーションを実行します。
runnable = graph.compile()
runnable.invoke(HumanMessage("35x42は?"))
実行結果はGraphを通じて生成された状態(Messageのリスト)になります。リストの末尾に35x42の答えである1470が出力されています。
[HumanMessage(content='35x42は?', id='xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx'),
AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_xxxxxxxxxxxxxxxxx', 'function': {'arguments': '{\n "first_number": 35,\n "second_number": 42\n}', 'name': 'multiply'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 23, 'prompt_tokens': 67, 'total_tokens': 90}, 'model_name': 'gpt-35-turbo-16k', 'system_fingerprint': None, 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'tool_calls', 'logprobs': None, 'content_filter_results': {}}, id='xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx'),
ToolMessage(content='1470', id='xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx', tool_call_id='call_xxxxxxxxxxxxxxxxx')]
まとめ
今回はLangGraphを使ってLLM Agentアプリケーションを作ってみました。LangGraphを使うと順序立ててアプリケーションの実装を進めることが出来る印象を持ちました。併せて今回はOpenAIのモデルのFunction Callingの機能についても触れることが出来ました。これまでなんとなく目にしてきた機能ですが、改めて理解を深めることが出来て良かったです。
