【Langchain/LangGraph】LangGraphを触ってみた

# はじめに
LangGraphを学ぶ機会があったので復習がてらまとめた記事です。
なお、Langchainについてはここでは解説していません(なのでLangchainについて既に知っている人向けの記事です)
また、Langchain及びLangGraphの更新は結構頻繁なので細かい仕様は公式で確認推奨です。
公式はドキュメントも充実しているので読みやすいです(ありがたい！！！)

サンプルコードはPython3で実装しています。
LangGraphのバージョンは、v0.4.3を使用

# 参考記事
- [LangGraph 公式トップページ](https://langchain-ai.github.io/langgraph/)
- [LangGraph 公式クイックスタート](https://langchain-ai.github.io/langgraph/tutorials/introduction/#langgraph-quickstart)

# LangGraphとは
LangChainアーキテクチャの一部として開発された、状態を保持するLLMアプリケーションを構築するためのフレームワークです。

複雑な制御フローとステートマシンの概念を取り入れることで、高度な対話型AIシステムやマルチエージェントシステムの実装を可能にします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3635188/4d2fc3cf-bf9f-47b0-b4fd-d96cbc172960.png)

# LangGraphの基本構造
LangGraphは有向グラフのモデルを採用しており、以下の主要な要素から構成されています

![{7FD9794E-08D6-444F-AFDE-D87CE7C80921}.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3635188/faf98252-aed5-4d74-9ebd-97f506c42359.png)

## グラフ(Graph)
- ノードとエッジで構成される全体構造
- ワークフローやプロセスフローを表現

## 状態(State)
- ノード間でデータを共有する手段として機能
- 各ノードが参照及び更新する
- 通常は辞書型で実装され、TypedDictやBaseModelで型定義される

## ノード(Node)
- 処理単位を表す要素
- 入力を受け取り、処理を実行し、出力を生成
- LangChainのチェーン、LLM、ツール、カスタム関数などを含む

## エッジ(Edge)
- ノード間の接続と情報の流れを定義
- 条件付きエッジにより分岐ロジックを実装可能

# LangChainとLangGraphの関係
LangChainとLangGraphは相互に補完し合う関係にあります
![{E6D6FB2C-CED4-4DA4-9995-A0DCEA1F9D86}.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3635188/6ee91319-a57e-420b-8280-87351b7d1666.png)

また、フローで比較するとLangchainが単純なフロー(チェーン構造)に対して、LangGraphは複雑なフローも表現できる。
![{DBCD8600-814B-4E15-958D-5F06B6EF7ADB}.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3635188/696ee7df-039d-4b50-94ce-6aba8a8d06f2.png)

# LangGraphの詳細
## 状態の定義
「typingモジュールのTypeDict」または「PydanticのBaseModel」が使用できます。
```python
from pydantic import BaseModel
​
# 状態の型定義
class ChatState(BaseModel):
    messages: list  # 会話履歴
    current_question: str  # 現在の質問
    research_results: dict  # 調査結果
    final_answer: str  # 最終回答
```

## ノードの例
```
# 1. 関数ノード - 最も基本的なノード
def simple_node(state):
    # 状態を処理して新しい状態を返す
    new_value = state.value + 100
    return {"value": new_value}
​
# 2. LLMノード - LangChainのLLMを利用するノード
def llm_node(state):
    llm = ChatOpenAI(temperature=0)
    prompt = ChatPromptTemplate.from_template("質問: {question}")
    chain = prompt | llm
    response = chain.invoke({"question": state["question"]})
    return {"final_answer": response.content}
​
# 3. ツールノード - 外部ツールやAPIを利用するノード
def tool_node(state):
    search_tool = GoogleSearchAPIWrapper()
    search_results = search_tool.run(state["query"])
    return {"search_results": search_results}
```

## ノードの追加と接続(エッジ)
```
# グラフにノードを追加
graph = StateGraph(State)
graph.add_node("analyzer", analyze_question)
graph.add_node("researcher", research)
graph.add_node("answerer", generate_answer)
​
# 簡単な接続（シーケンシャル）
graph.add_edge("analyzer", "researcher")
graph.add_edge("researcher", "answerer")
​
# より複雑な条件付き接続
def route_based_on_analysis(state):
    if state["complexity_score"] > 0.7:
        return "deep_research"
    else:
        return "quick_research"
​
graph.add_conditional_edges(
    "analyzer",
    route_based_on_analysis,
    {
        "deep_research": "deep_researcher",
        "quick_research": "quick_researcher"
    }
)
```

# LangGraphのメリット
自分としては以下かなと。
- **複雑なワークフローの実現**: 条件分岐、ループ、並列処理などを自然に表現
- **状態管理の明示化**: アプリケーションの状態を明示的に追跡・管理
- **デバッグの容易さ**: グラフ構造により処理の流れが視覚的に理解しやすい
- **スケーラビリティ**: ノードやエッジの追加・変更が簡単なので、小規模なコンポーネントから複雑なシステムへの拡張が容易

※シンプルな１方向だけのフローをLangGraphで実現するとLangChainだけで実現した方がコード量は少なかったりするので注意

# LangGraphのユースケース
- **高度な対話システム**: 複数のターンにわたる対話の文脈を管理
- **自動化されたワークフロー**: 複数のステップと条件分岐を含むタスク
- **マルチエージェントシステム**: 異なる役割を持つエージェント間の協調動作
- **自己修正システム**: 出力を評価し改善するフィードバックループ
- **複合的なAIサービス**: 複数のAI機能を組み合わせたアプリケーション

# まとめ
LangGraphは、LLMアプリケーション開発における「次のステップ」と言えるフレームワークです。
単純なシーケンシャルなチェーンから、複雑な状態管理と制御フローを持つアプリケーションへの移行を可能にします。
特に、以下のような場合にLangGraphの採用を検討すべきです：
- 複雑な条件分岐を含むLLMアプリケーション
- 状態を保持する必要がある対話システム
- 複数のLLMやツールを連携させる必要があるシステム
- 反復的な改善ループを実装するアプリケーション

LangGraphを活用することで、より洗練された、状態を意識したLLMアプリケーションの構築が可能になります。

# 実装例:シンプルなエージェントの構築
- 参考1：[LangGraph公式：ツールによるチャットボットの強化](https://langchain-ai.github.io/langgraph/concepts/why-langgraph/#part-2-enhancing-the-chatbot-with-tools)

本例ではTavilyというAIエージェント専用に構築された検索エンジンを利用。毎月1000APIコールまでは無料
Tavilyの使い方についてはこちらの方の記事が参考になりました。

https://note.com/npaka/n/n9fe8a607c56e

```python
import boto3
import json
import traceback

from typing import Annotated, Any, Literal
from dotenv import load_dotenv
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langchain_community.tools.tavily_search import TavilySearchResults
from pydantic import BaseModel
from IPython.display import Image, display

# 環境変数の読み込み(boto3/Tavily用)
# .envファイル等でAWS_PROFILEとTAVILY_API_KEYを設定しておく
load_dotenv()

# ツールのセットアップ
tool = TavilySearchResults(max_results=1)
tools = [tool]

bedrock_client = boto3.client(service_name="bedrock-runtime", region_name="us-east-1")


def create_bedrock_llm(
    temperature: float = 0,
    max_tokens: int = 1000,
    model_id: str = "anthropic.claude-3-5-sonnet-20240620-v1:0"
) -> ChatBedrock:
    """プロバイダーはBedrockを利用
    """
    model_params = {
        "temperature": temperature,
        "max_tokens": max_tokens
    }
    return ChatBedrock(
        client=bedrock_client,
        model_id=model_id,
        model_kwargs=model_params
    )

# モデルに検索エンジンツールをバインド
model = create_bedrock_llm()
model_with_tools = model.bind_tools(tools)


class State(BaseModel):
    # 既存のlistを上書きではなく、追加していく
    messages: Annotated[list, add_messages]


def call_model(state: State) -> dict[str, list[Any]]:
    print("========== モデルの実行 ==========")
    messages = state.messages
    response = model_with_tools.invoke(messages)
    print("response:\n", response)
    return {"messages": [response]}


# Note:自前で実装せずともlanggraph.prebuilt.ToolNode利用するだけでも今は実現できるが練習がてら
def call_tool(state: State) -> dict[str, list[Any]]:
    """ツールの実行
    """
    print("========== ツールの実行 ==========")
    messages = state.messages
    # 継続条件に基づき、最後のメッセージが関数呼び出しを含まれています
    last_message = messages[-1]
    outputs = []
    for tool_call in last_message.tool_calls:
        print("tool_call: ", tool_call)
        for tl in tools:
            if tl.name == tool_call["name"]:
                tool_result = tl.invoke(tool_call["args"])
                tool_result_str = json.dumps(tool_result)
                print("tool_result_str:\n", tool_result_str)
                outputs.append(ToolMessage(
                    content=tool_result_str,
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                ))
    return {"messages": outputs}


# Note:自前で実装せずともlanggraph.prebuilt.tools_condition利用するだけでも今は実現できるが練習がてら
def should_continue(state: State) -> Literal["end", "continue"]:
    """
    継続条件の判定
    エージェントがアクションを取るよう指示した場合、処理続行
    """
    print("========== 継続条件の判定 ==========")
    messages = state.messages
    ai_message = messages[-1]
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "continue"
    return "end"


# StateGraphオブジェクトの作成
graph_builder = StateGraph(State)
# ノード関数の登録
graph_builder.add_node("agent", call_model)
graph_builder.add_node("action", call_tool)
# エッジを定義：グラフを実行した際に、どの順番でノードを処理するか(START～END)
graph_builder.add_edge(START, "agent")
graph_builder.add_edge("action", "agent")
# 条件付きエッジの追加
graph_builder.add_conditional_edges(
    # 最初に、開始ノードを定義します。
    "agent",
    #  次に呼び出されるノードを決定する関数
    should_continue,
    # should_continueの出力とどのノードに行くかのマッピング
    {
        "continue": "action",
        "end": END
    }
)
graph = graph_builder.compile()

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        inputs = {"messages": [HumanMessage(content=user_input)]}
        ai_response = graph.invoke(inputs)
        print("AI: ", ai_response["messages"][-1].content)
    except Exception as e:
        print(e)
        traceback.print_exc()
        break
```

上記のフローをコードから可視化したのが以下
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3635188/d2fea90a-46ca-4de0-882c-64d4609fddc1.png)
```python
# get_graphメソッドからフローの可視化も可能
try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    print("エラー")
```

ちなみに実行結果例は以下の通りです。
質問：現在の日本の総理大臣は誰？
```
========== モデルの実行 ==========
response:
 content='この質問に正確に答えるために、最新の情報を検索する必要があります。日本の政治情勢は変化する可能性があるため、現在の情報を確認しましょう。～～～省略～～～
========== 継続条件の判定 ==========
========== ツールの実行 ==========
～～～省略～～～
========== モデルの実行 ==========
response:
 content='検索結果に基づいて、現在の日本の総理大臣についてお答えします。～～～省略～～～
========== 継続条件の判定 ==========
AI:  検索結果に基づいて、現在の日本の総理大臣についてお答えします。

現在の日本の総理大臣（内閣総理大臣）は石破茂（いしば しげる）氏です。石破氏は第103代内閣総理大臣として、2024年（令和6年）11月11日に就任しました。

ただし、この情報には注意が必要です：

1. Wikipediaの情報は常に最新とは限らず、誰でも編集できるため、時に不正確な情報が含まれる可能性があります。

2. 政治的な変化は急速に起こる可能性があるため、この情報が検索時点で最新であっても、現在は異なる可能性があります。

3. 検索結果の日付が明確でないため、この情報がいつの時点のものかは不明確です。

したがって、最新かつ正確な情報を得るためには、日本政府の公式ウェブサイトや信頼できる最新のニュースソースを確認することをお勧めします。現在の正確な情報を知りたい場合は、それらのソースで直接確認することが最善です。

```