DatabricksでLangGraphのクイックスタートを動かしてみる(その2)

こちらの続きです。

https://qiita.com/taka_yayoi/items/d6e5fd6a724eb9b65fbc

こちらの🛠️ Enhancing the Chatbot with Toolsを実行します。

https://langchain-ai.github.io/langgraph/tutorials/introduction/#part-2-enhancing-the-chatbot-with-tools

# パート2: 🛠️ ツールを使ってチャットボットを強化する
チャットボットが「記憶から」答えられないクエリに対処するために、ウェブ検索ツールを統合します。このツールを使用して、関連情報を見つけ、より良い応答を提供できるようにします。

**要件**
始める前に、必要なパッケージがインストールされており、APIキーが設定されていることを確認してください：

まず、[Tavily検索エンジン](https://python.langchain.com/docs/integrations/tools/tavily_search/)を使用するための要件をインストールし、[TAVILY_API_KEY](https://tavily.com/)を設定します。

## セットアップ

まず、必要なパッケージをインストールし、環境を設定します：

```py
%%capture --no-stderr
%pip install -U langgraph langsmith langchain_openai openai tavily-python langchain_community
%restart_python
```

```py
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キー>"
```

次にツールを定義します。

```py
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**](https://qiita.com/taka_yayoi/items/d6e5fd6a724eb9b65fbc)と同じですが、LLMに`bind_tools`を追加しました。これにより、LLMが検索エンジンを使用したい場合に正しいJSON形式を認識できるようになります。

```py
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](https://langchain-ai.github.io/langgraph/reference/prebuilt/#toolnode)に置き換えてスピードアップを図りますが、最初に自分で構築することは有益です。

```py
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`に置き換えますが、最初に自分で実装することでより明確になります。

```py
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`を設定する必要はありません。私たちのグラフにはすでに終了する方法があります！

構築したグラフを視覚化してみましょう。次の関数には、このチュートリアルには重要でない追加の依存関係があります。

```py
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
```
![download.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ccff0b18-0ef3-2d30-862d-986f10034212.png)

これで、トレーニングデータ外の質問をボットに尋ねることができます。

```py
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
```
![Screenshot 2025-01-29 at 20.42.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/10e16c9f-722e-2bcb-afd3-b79afeb8758f.png)

会計ソフトの弥生の記事が該当しているのはご愛嬌で。

![Screenshot 2025-01-29 at 20.43.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cdf39728-d87e-67d6-ba82-83cf365a6900.png)

**おめでとうございます！**langgraphで会話エージェントを作成し、必要に応じて検索エンジンを使用して最新情報を取得できるようになりました。これで、より幅広いユーザーの質問に対応できます。エージェントが実行したすべてのステップを確認するには、この[LangSmithトレース](https://smith.langchain.com/public/4fbd7636-25af-4638-9587-5a02fdbb0172/r)をチェックしてください。

私たちのチャットボットはまだ過去の対話を自分で記憶することができず、一貫したマルチターンの会話を行う能力が制限されています。次のパートでは、この問題に対処するために**メモリ**を追加します。

このセクションで作成したグラフの完全なコードは以下に再現されており、`BasicToolNode`を既製の[ToolNode](https://langchain-ai.github.io/langgraph/reference/prebuilt/#toolnode)に置き換え、`route_tools`条件を既製の[tools_condition](https://langchain-ai.github.io/langgraph/reference/prebuilt/#tools_condition)に置き換えています。

```py
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()
```

こちらに続きます。

https://qiita.com/taka_yayoi/items/255b124aae94df2e04ec


### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)