LangGraph 条件付きエッジ


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

# 条件付きエッジとは

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


# 例

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

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

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

今回のケースで実行したい処理は3つあります。
- LLMと会話する
- 英語かどうかチェックする
- 英語に翻訳する

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

```python
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)と会話する
1. check_englishで、LLMからの返事が英語かどうかLLMに判定させる
1. 判定結果が英語なら終了。英語でなければ、translate_into_englishで翻訳させて再度判定。

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

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

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

```python
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からの返事によって処理が変わります。
<font color="Red">**ここで条件付きエッジを使いましょう。**</font>

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

#### 作ってみる

LLMからの返事にあたるstate["Message"]が"Yes"であれば"MessageIsEnglish"を返し、そうでなければ"MessageIsJapanese"を返す関数を定義します
```python
def check_openai_response(state: MyState):
    # LLMが"Yes"か"No"かで返してくれると信じて条件分岐(本当はLLMからの返事を固定するべき。)
    if "Yes" in state["check_english_reponse"]:
        return "MessageIsEnglish"
    return "MessageIsJapanese"
```

この関数さえ定義すれば、条件付きエッジが作れます。
```python
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と進む経路) 

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

### ここまでのコードまとめ

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

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

## グラフの確認

コンパイルしたグラフは、マーメイド形式で出力できます。
```python 
print(graph.get_graph().draw_mermaid())
```

以下が今回のグラフとなります。
```mermaid
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
        __start__([<p>__start__</p>]):::first
        conversation_with_openai(conversation_with_openai)
        check_english(check_english)
        translate_into_english(translate_into_english)
        __end__([<p>__end__</p>]):::last
        __start__ --> conversation_with_openai;
        conversation_with_openai --> check_english;
        translate_into_english --> check_english;
        check_english -. &nbsp;MessageIsEnglish&nbsp; .-> __end__;
        check_english -. &nbsp;MessageIsJapanese&nbsp; .-> translate_into_english;
        classDef default fill:#f2f0ff,line-height:1.2
        classDef first fill-opacity:0
        classDef last fill:#bfb6fc
```

## 実行する

### 実行結果1

実行してみましょう。まず、日本語で入力してみます。
以下のinvokeメソッドでも実行できるのですが、
```python
graph.invoke({"input": "今日はいい天気ですね"})
```
これだと途中経過が追いづらかったので、streamを使ってprintすることにしました。
※LangSmithを使えば、もっと途中経過が追いやすくなります。今回は簡単に確認するためにprintを使います。


```python
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

次は英語で入力してみます。
```python
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