LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@tkatsumata09

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

Posted at

はじめに

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

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

参考記事

LangGraphとは

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

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

image.png

LangGraphの基本構造

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

{7FD9794E-08D6-444F-AFDE-D87CE7C80921}.png

グラフ(Graph)

  • ノードとエッジで構成される全体構造
  • ワークフローやプロセスフローを表現

状態(State)

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

ノード(Node)

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

エッジ(Edge)

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

LangChainとLangGraphの関係

LangChainとLangGraphは相互に補完し合う関係にあります
{E6D6FB2C-CED4-4DA4-9995-A0DCEA1F9D86}.png

また、フローで比較するとLangchainが単純なフロー(チェーン構造)に対して、LangGraphは複雑なフローも表現できる。
{DBCD8600-814B-4E15-958D-5F06B6EF7ADB}.png

LangGraphの詳細

状態の定義

「typingモジュールのTypeDict」または「PydanticのBaseModel」が使用できます。

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アプリケーションの構築が可能になります。

実装例:シンプルなエージェントの構築

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

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

# 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. 検索結果の日付が明確でないため、この情報がいつの時点のものかは不明確です。

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

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?