こちらの続きです。
こちらの🛠️ Enhancing the Chatbot with Toolsを実行します。
パート2: 🛠️ ツールを使ってチャットボットを強化する
チャットボットが「記憶から」答えられないクエリに対処するために、ウェブ検索ツールを統合します。このツールを使用して、関連情報を見つけ、より良い応答を提供できるようにします。
要件
始める前に、必要なパッケージがインストールされており、APIキーが設定されていることを確認してください：
まず、Tavily検索エンジンを使用するための要件をインストールし、TAVILY_API_KEYを設定します。
セットアップ
まず、必要なパッケージをインストールし、環境を設定します：
%%capture --no-stderr
%pip install -U langgraph langsmith langchain_openai openai tavily-python langchain_community
%restart_python
import os
os.environ["OPENAI_API_KEY"] = dbutils.secrets.get(scope="demo-token-takaaki.yayoi", key="openai_api_key")
# TavilyのAPIキー
os.environ["TAVILY_API_KEY"] = "<TavilyのAPIキー>"
次にツールを定義します。
from langchain_community.tools.tavily_search import TavilySearchResults
tool = TavilySearchResults(max_results=2)
tools = [tool]
tool.invoke("LangGraphにおける「ノード」とは？")
[{'url': 'https://papanoyang.github.io/langgraph/20240605002/',
'content': 'LangGraphではノードとノードの間、情報を共有していてそれをステート（State）と呼ぶ。 なのでパラメータの名前はstateかstateがついた名前になっている。 次の二つのエイジェントは同じのような内容なので次のように作成しよう。'},
{'url': 'https://speakerdeck.com/knishioka/langgraphnonodoetuziruteinguwoshen-ku-ri',
'content': 'Graph • LangGraphにおけるワークフローの中心的な表現方法で、ノードとエッジで 構成される • グラフはステートマシンとして機能し、共有状態を持つマルチアクターのア プリケーションを柔軟に構築可能 • グラフの主要コンポーネント:'}]
結果は、チャットボットが質問に答えるために使用できるページの要約です。
次に、グラフの定義を開始します。以下はパート1と同じですが、LLMにbind_toolsを追加しました。これにより、LLMが検索エンジンを使用したい場合に正しいJSON形式を認識できるようになります。
from typing import Annotated
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
llm = ChatOpenAI(model="gpt-4o-mini")
# 変更: LLMに使用できるツールを伝える
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
次に、ツールが呼び出された場合に実際に実行する関数を作成する必要があります。これを行うために、ツールを新しいノードに追加します。
以下では、状態の最新メッセージをチェックし、メッセージにtool_callsが含まれている場合にツールを呼び出すBasicToolNodeを実装します。これは、Anthropic、OpenAI、Google Gemini、その他多くのLLMプロバイダーで利用可能なLLMのtool_callingサポートに依存しています。
後でLangGraphの既製のToolNodeに置き換えてスピードアップを図りますが、最初に自分で構築することは有益です。
import json
from langchain_core.messages import ToolMessage
class BasicToolNode:
"""最後のAIMessageで要求されたツールを実行するノード。"""
def __init__(self, tools: list) -> None:
self.tools_by_name = {tool.name: tool for tool in tools}
def __call__(self, inputs: dict):
if messages := inputs.get("messages", []):
message = messages[-1]
else:
raise ValueError("入力にメッセージが見つかりません")
outputs = []
for tool_call in message.tool_calls:
tool_result = self.tools_by_name[tool_call["name"]].invoke(
tool_call["args"]
)
outputs.append(
ToolMessage(
content=json.dumps(tool_result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": outputs}
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
ツールノードが追加されたので、conditional_edgesを定義できます。
edgesは、あるノードから次のノードへの制御フローをルートすることを思い出してください。Conditional edgesは通常、「if」文を含み、現在のグラフの状態に応じて異なるノードにルートします。これらの関数は現在のグラフのstateを受け取り、次に呼び出すノードを示す文字列または文字列のリストを返します。
以下では、チャットボットの出力にtool_callsがあるかどうかをチェックするroute_toolsというルータ関数を定義します。この関数をadd_conditional_edgesを呼び出してグラフに提供し、chatbotノードが完了するたびにこの関数をチェックして次にどこに行くかを確認します。
条件は、ツールコールが存在する場合はtoolsにルートし、存在しない場合はENDにルートします。
後で、より簡潔にするために組み込みのtools_conditionに置き換えますが、最初に自分で実装することでより明確になります。
def route_tools(
state: State,
):
"""
最後のメッセージにツール呼び出しがある場合にToolNodeにルーティングするために
conditional_edgeで使用します。そうでない場合は終了にルーティングします。
"""
if isinstance(state, list):
ai_message = state[-1]
elif messages := state.get("messages", []):
ai_message = messages[-1]
else:
raise ValueError(f"入力状態にメッセージが見つかりません: {state}")
if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
return "tools"
return END
# `tools_condition`関数は、チャットボットがツールを使用するように要求した場合に"tools"を返し、
# 直接応答する場合は"END"を返します。この条件付きルーティングはメインエージェントループを定義します。
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
# 次の辞書は、条件の出力を特定のノードとして解釈するようにグラフに指示するためのものです
# デフォルトではアイデンティティ関数ですが、
# "tools"以外の名前のノードを使用したい場合は、
# 辞書の値を別のものに更新できます
# 例: "tools": "my_tools"
{"tools": "tools", END: END},
)
# ツールが呼び出されるたびに、次のステップを決定するためにチャットボットに戻ります
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
条件付きエッジは単一のノードから始まることに注意してください。これはグラフに「'chatbot'ノードが実行されるたびに、ツールを呼び出す場合は'tools'に進み、直接応答する場合はループを終了する」と指示します。
組み込みのtools_conditionと同様に、ツール呼び出しが行われない場合、関数はEND文字列を返します。グラフがENDに遷移すると、完了するタスクがなくなり、実行が停止します。条件がENDを返す可能性があるため、今回は明示的にfinish_pointを設定する必要はありません。私たちのグラフにはすでに終了する方法があります！
構築したグラフを視覚化してみましょう。次の関数には、このチュートリアルには重要でない追加の依存関係があります。
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
これで、トレーニングデータ外の質問をボットに尋ねることができます。
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
while True:
try:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
except:
# input()が利用できない場合のフォールバック
user_input = "LangGraphについて何を知っていますか？"
print("User: " + user_input)
stream_graph_updates(user_input)
break
会計ソフトの弥生の記事が該当しているのはご愛嬌で。
**おめでとうございます！**langgraphで会話エージェントを作成し、必要に応じて検索エンジンを使用して最新情報を取得できるようになりました。これで、より幅広いユーザーの質問に対応できます。エージェントが実行したすべてのステップを確認するには、このLangSmithトレースをチェックしてください。
私たちのチャットボットはまだ過去の対話を自分で記憶することができず、一貫したマルチターンの会話を行う能力が制限されています。次のパートでは、この問題に対処するためにメモリを追加します。
このセクションで作成したグラフの完全なコードは以下に再現されており、BasicToolNodeを既製のToolNodeに置き換え、route_tools条件を既製のtools_conditionに置き換えています。
from typing import Annotated
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()
こちらに続きます。