LoginSignup
0
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?

@MichaelHashimoto

LangGraph 条件付きエッジ

Last updated at Posted at 2025-02-11

LangGraphはAgent作成のためのフレームワークです。
LangGraphで複雑なAgentを作るためには、"条件付きエッジ"を使いこなす必要があります。
この記事では、簡単な例を使って条件付きエッジを説明します。

条件付きエッジとは

「一つ前のノードから〜という返答が来たらAノードへ、〜という返答が来たらBノードへ...」というように、ノードの遷移方法に条件分けを行うエッジのことを、条件付きエッジと呼びます。

例

では、以下のようなケースを考えます。

  1. LLMと会話する。
  2. 会話の返事が英語かどうかチェックし、英語ならOK。英語ではなければ、翻訳させて再度チェック。

処理を抽出し、Nodeを定義する

今回のケースで実行したい処理は3つあります。

  • LLMと会話する
  • 英語かどうかチェックする
  • 英語に翻訳する

まずはこれらに対応するNodeを定義します。

llm = ChatOpenAI(model="gpt-4o")

# Graph state
class MyState(TypedDict):
    input: str
    conversation_response: str
    check_english_response: str

# Node
def conversation_with_openai(state: MyState) -> MyState:
    msg = llm.invoke(state["input"]) # inputをllmに渡す
    return {"conversation_response": msg.content} 
    # llmからの返事をconversation_responseに格納

# Node
def check_english(state: MyState) -> MyState:
    # conversation_responseが英語で書かれているかllmに判定してもらう
    msg = llm.invoke(f'Is this written in English : {state["conversation_response"]} . If it is English, please answer "Yes". If not, please answer "No".')
    # 判定の返事をcheck_english_responseとして追加して返す
    return {"conversation_response": state["conversation_response"],
            "check_english_response":msg.content}

# Node
def translate_into_english(state: MyState) -> MyState:
    # conversation_responseをllmに英語に変換してもらう
    msg = llm.invoke(f'Translate this into English : {state["conversation_response"]}. ')
    # 返事を新たなconversation_responseとする
    return {"conversation_response": msg.content}


# Build workflow
workflow = StateGraph(MyState) # MyState型を引数に渡し、schemaとしてMyStateを使うことを宣言。

# Add nodes
# "conversation_with_openai"ノード
workflow.add_node("conversation_with_openai", conversation_with_openai)  
# "check_english"ノード
workflow.add_node("check_english", check_english)  
# "translate_into_english"ノード
workflow.add_node("translate_into_english", translate_into_english)

ここまでで以下の3つのノードを定義しました。

  • conversation_with_openai
  • check_english
  • translate_into_english
    これらのノードを使って、次のようにしたいと思います。
  1. conversation_with_openaiでLLM(openai)と会話する
  2. check_englishで、LLMからの返事が英語かどうかLLMに判定させる
  3. 判定結果が英語なら終了。英語でなければ、translate_into_englishで翻訳させて再度判定。

ノード同士をつなぎ、エッジを作る

エッジの作成 (START → 1 → 2)

ここは単純にnode同士をつなぎましょう。

workflow.add_edge(START, "conversation_with_openai")  # START→1
workflow.add_edge("conversation_with_openai", "check_english")  # 1→2

条件付きエッジの作成 (2 → 3 → END あるいは　2 → 3 →　2)

3では、2のLLMからの返事によって処理が変わります。
ここで条件付きエッジを使いましょう。

StateGraphのadd_conditional_edgesメソッドを使うことで作成できます。

作ってみる

LLMからの返事にあたるstate["Message"]が"Yes"であれば"MessageIsEnglish"を返し、そうでなければ"MessageIsJapanese"を返す関数を定義します

def check_openai_response(state: MyState):
    # LLMが"Yes"か"No"かで返してくれると信じて条件分岐(本当はLLMからの返事を固定するべき。)
    if "Yes" in state["check_english_reponse"]:
        return "MessageIsEnglish"
    return "MessageIsJapanese"

この関数さえ定義すれば、条件付きエッジが作れます。

workflow.add_conditional_edges("check_english", check_openai_response, 
    path_map={"MessageIsEnglish":END, "MessageIsJapanese":"translate_into_english"}
)

第一引数 source: 始点はcheck_englishノードです。
第二引数 path: 定義したcheck_openai_response関数です。"check_english"の返事で返る値が変わります。
第三引数 path_map: check_openai_responseの返り値とノードをマッピングします。今回なら、"MessageIsEnglish"が返ってくればENDノードに、"MessageIsJapanese"が返ってくればtranslate_into_englishノードに進むようにマッピングしました

今回のケースでは、条件付きエッジでtranslate_into_englsihに進んだ場合、再度判定させることにしていました。以下のエッジも追加します。 (2 → 3 → 2と進む経路)

workflow.add_edge("translate_into_english", "check_english")

ここまでのコードまとめ

ここまで作ってきたノード, エッジのコードを結合し、graphをコンパイルするコードが以下です。

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class MyState(TypedDict):
    input: str
    conversation_response: str
    check_english_response: str


llm = ChatOpenAI(model="gpt-4o")

# Node
def conversation_with_openai(state: MyState) -> MyState:
    msg = llm.invoke(state["input"])
    return {"conversation_response": msg.content}

# Node
def check_english(state: MyState) -> MyState:
    msg = llm.invoke(f'Is this written in English : {state["conversation_response"]} . If it is English, please answer "Yes". If not, please answer "No".')
    return {"conversation_response": state["conversation_response"], "check_english_response":msg.content}

# Node
def translate_into_english(state: MyState) -> MyState:
    msg = llm.invoke(f'Translate this into English : {state["conversation_response"]}. ')
    return {"conversation_response": msg.content}

def check_openai_response(state: MyState):
    if "Yes" in state["check_english_response"]:
        return "MessageIsEnglish"
    return "MessageIsJapanese"
    

# Build graph
workflow = StateGraph(MyState)
workflow.add_node("conversation_with_openai", conversation_with_openai)  # "conversation_with_openai"ノード
workflow.add_node("check_english", check_english)  # "check_english"ノード
workflow.add_node("translate_into_english", translate_into_english)  # translate_into_englishノード

workflow.add_edge(START, "conversation_with_openai")
workflow.add_edge("conversation_with_openai", "check_english")
workflow.add_conditional_edges("check_english", check_openai_response, {"MessageIsEnglish":END, "MessageIsJapanese":"translate_into_english"})
workflow.add_edge("translate_into_english", "check_english")

graph = workflow.compile()

グラフの確認

コンパイルしたグラフは、マーメイド形式で出力できます。

print(graph.get_graph().draw_mermaid())

以下が今回のグラフとなります。

実行する

実行結果1

実行してみましょう。まず、日本語で入力してみます。
以下のinvokeメソッドでも実行できるのですが、

graph.invoke({"input": "今日はいい天気ですね"})

これだと途中経過が追いづらかったので、streamを使ってprintすることにしました。
※LangSmithを使えば、もっと途中経過が追いやすくなります。今回は簡単に確認するためにprintを使います。

for event in graph.stream({"input": "今日はいい天気ですね"}):
    print(event)

print結果です

{'conversation_with_openai': 
    {
    'conversation_response': 'そうですね！天気が良いと気分も上がりますよね。こんな日は外に出かけたり、散歩したりするのが気持ち良さそうです。何か特別な予定はありますか？'
    }
}
{'check_english': 
    {
    'conversation_response': 'そうですね！天気が良いと気分も上がりますよね。こんな日は外に出かけたり、散歩したりするのが気持ち良さそうです。何か特別な予定はありますか？', 
    'check_english_reponse': 'No.'
    }
}
{'translate_into_english': 
    {
    'conversation_response': 'Sure! "That\'s right! Good weather really lifts your spirits. On a day like this, it feels great to go out or take a walk. Do you have any special plans?"'
    }
}
{'check_english': 
    {
    'conversation_response': 'Sure! "That\'s right! Good weather really lifts your spirits. On a day like this, it feels great to go out or take a walk. Do you have any special plans?"', 
    'check_english_reponse': 'Yes.'
    }
}

printから流れを確認すると

conversation_with_openai
→ 日本語で返事が来た
→ check_english
→ 返事で"No."が返ってきた
→ 条件付きエッジにより、translate_into_english
→ check_english
→ 返事で"Yes."が返ってきた
→ 条件付きエッジにより、END

というように動いたことが分かります

実行結果2

次は英語で入力してみます。

for event in graph.stream({"input": "It's a nice day today."}):
    print(event)

print結果です

{'conversation_with_openai': 
    {
    'conversation_response': "I'm glad to hear that! What do you have planned for the day?"
    }
}
{'check_english': 
    {
    'conversation_response': "I'm glad to hear that! What do you have planned for the day?", 
    'check_english_reponse': 'Yes.'
    }
}

printから流れを確認すると

conversation_with_openai
→ 英語で返事が来た
→ check_english
→ 返事で"Yes."が返ってきた
→ 条件付きエッジにより、END

0
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
0
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?