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

条件付きエッジとは

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

例

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

LLMと会話する。
会話の返事が英語かどうかチェックし、英語なら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
これらのノードを使って、次のようにしたいと思います。

conversation_with_openaiでLLM(openai)と会話する
check_englishで、LLMからの返事が英語かどうかLLMに判定させる
判定結果が英語なら終了。英語でなければ、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