LangChain v0.3　その６ ~LangGraphの基礎~

# ここまで
前回まで、`ChatPromptTemplate`や`LCEL`、`LangGraph`に繋げるための`Pydantic`についてやってきました。
今回はいよいよ`LangGraph`です。
ぜひ、`class`と`Pydantic`をしっかり理解しておけば難しくなさそうです。

https://qiita.com/ikedachin/items/7ae52f1c497860bb3624

# 今日やること
LangGraphを使って、日本語と英語で大喜利させて、その結果から日本語文化と英語文化を比較することをやってみます。
`LangGraph`は`node`と`edge`と`state`の理解が必要です。
- `node`：何かしら処理をするところ
- `state`：`node`で処理した内容を記憶した状態
- `edge`：`node`と`node`をつなぎ、`state`を受け渡すところ

なので、日本語で大喜利する`node`と、英語で大喜利する`node`、これらを合わせて文化を比較するノードが必要になりますね。


# バージョン関連
Python 3.10.8
langchain==0.3.7
python-dotenv
langchain-openai==0.2.5
langgraph>0.2.27
langchain-core

※LLMのAPIはAzureOpenAIのgpt-4o-miniを使いました

# 復習
その５において、`Pydantic`の使い方をやりましたが、`TypedDict`を使う方法と、`BaseModel`を使う方法がありました。
TypedDict: なんやら辞書形式でstateを受け渡す方法
BaseModel: なんやらJson形式っぽいのでstateを受け渡す方法

それぞれ、やっていきたいと思います。

# 先ずはライブラリのインポート
いっぱいなんで嫌んなる。

```Python:ライブラリのインポート
import os
from dotenv import load_dotenv
from typing import Annotated
from typing_extensions import TypedDict

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

from langchain_core.messages import AIMessage
from langchain_core.runnables import RunnableLambda
from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import AzureChatOpenAI, AzureOpenAI

from langgraph.graph import START, END, StateGraph
from langgraph.graph.message import add_messages
```
`from langchain_core.pydantic_v1 import BaseModel`は`from pydantic import BaseModel`に置き換えろって警告が出ます。僕の場合、Windowsだけ。Macは出なかった。


```Python:環境系の設定
load_dotenv('.env')
os.environ["LANGCHAIN_TRACING_V2"]="true"
os.environ["LANGCHAIN_ENDPOINT"]="https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"]=os.getenv("LANG_SMITH_API")
os.environ["LANGCHAIN_PROJECT"]="langchain_test"
```
```Python:モデルの用意
model = AzureChatOpenAI(
    azure_deployment='gpt-4o-mini',
    azure_endpoint=os.getenv("END_POINT"),
    api_version=os.getenv("API_VERSION"),
    api_key=os.getenv("API_KEY")
)
```

```Python:テンプレートの用意
prompt_template0 = ChatPromptTemplate([
    ('system', 'あなたは偉大な漫才師です。'),
    ('user', '{topic}について大喜利してください。')
])

topic2english = ChatPromptTemplate([
    ('system', 'あなたは優秀な英語への翻訳者です。翻訳した内容だけを回答してください。'),
    ('user', '{topic}を翻訳してください')
])

oogiri_prompt = ChatPromptTemplate([
    ('system', 'You are a great comedian'),
    ('user', 'Please create OOGIRI about {topic} in English')
])

template_compare = ChatPromptTemplate([
    ('system', 'あなたはお笑いを客観的に分析ができる優秀な評論家です。'),
    ('user', 
     '''ここまでの会話で、英語で書かれた大喜利と、日本語で掛かれた大喜利が出てきました。
     これらの大喜利の一部を引用しながら、推測可能な英語社会と日本語社会の文化の違いを推測し、解説してください。
     ### 日本語の大喜利
     {oogiri_ja}
     ### 英語の大喜利
     {oogiri_en}
     ''')
])
```
# 辞書型を使う方法
### いろいろと定義
```Python:stateの型定義
class StateDict(TypedDict): # ここで辞書にしたから後がtopic=state['topic']
    topic: str = Field(..., description='topic for oogiri')
    oogiri_jp: str = Field(default='', description='oogiri in Japanese')
    oogiri_en: str = Field(default='', description='oogiri in English')
    result: str = Field(default='', description='result')
    messages: Annotated[list[str], add_messages] = Field(default=[], description='履歴')
```
その５でやりましたね。説明をしておくと・・・
- `topic` 大喜利をするお題を入れます。文字型
- `oogiri_jp`　生成された日本語の大喜利を入れます。文字型
- `oogiri_en`　生成された英語の大喜利を入れます。文字型
- `result`　日本語と英語の大喜利を元に日本語と英語の文化を比較した結果を入れます。文字型
- `messages`　会話履歴を残します。リスト型で要素は文字型

こんな設定にしてみました。
この`state`が入出力されるというコトをイメージして関数を書いていきます。

```Python:ノードになる関数を用意
pass_through_node = RunnableLambda(lambda x: x) # 受け取った値をそのまま出力する関数


def oogiri_japanese(state: StateDict): # 日本語で大喜利する関数（後でノードとして定義します）
    print(state)
    topic = state['topic']
    oogiri_chain_jp = prompt_template0 | model
    output = oogiri_chain_jp.invoke(topic)

    return {"role":"AIMessage", 'messages': [output], 'oogiri_jp': output_parser.invoke(output)} 

def oogiri_english(state: StateDict): # 英語で大喜利する関数（後でノードとして定義します）
    print(state)
    topic = state['topic']
    oogiri_chain_english = topic2english | model | output_parser | oogiri_prompt | model
    output = oogiri_chain_english.invoke(topic)
    return {"role":"AIMessage", 'messages': [output], 'oogiri_en': output_parser.invoke(output)}


def comparison_template_func(state: StateDict): # 文化を比較するする関数（後でノードとして定義します）
    print(state['oogiri_jp'], state['oogiri_en'])
    input = template_compare.invoke({'oogiri_ja': state["oogiri_jp"], 'oogiri_en': state["oogiri_en"]})
    input = input.messages[-1].content
    comparison_chain =  model | output_parser
    output = comparison_chain.invoke(output_parser.invoke(input))
    return {"role":"AIMessage", 'messages': [output], 'result': output}

output_parser = StrOutputParser() # 出力のうち、テキストだけを取り出す関数
```

簡単に解説
`oogiri_japanese`と`oogiri_english`
それぞれの関数の引数は`state`です。`state`は辞書型で、定義した通りの`key`があります。
そのうちの`topic`が今回大喜利をする話題になります。
それを`prompt_template0`に渡してプロンプトを作り、モデルに渡します。その結果を返すそれだけです。
`oogiri_english`の場合は英語に翻訳する`LCEL`が増えてるだけです。

そして、`comparison_template_func`関数では`state`の中から日本語と英語の大喜利を抜き取って、テンプレートにいれて、その`content`をモデルに渡して出力します。

### テスト
関数が上手くつなげられるかテストしてみましょう
```python:test
state_1 = StateDict(topic='お茶')
print("入力のstateを作る\n", state_1)
state_2 = oogiri_japanese(state_1)
print("日本語で大喜利\n", state_2)
state_3 = oogiri_english(state_1)
print("英語で大喜利\n", state_3)
```
```output:output
入力のstateを作る
 {'topic': 'お茶'}
日本語で大喜利
 {'topic': 'お茶', 'role': 'AIMessage', 'messages': [AIMessage(content='もちろん！お茶についての大喜利、<・・・中略・・・>)], 'oogiri_jp': 'もちろん！お茶についての大喜利、<・・・中略・・・>！'}
英語で大喜利
 {'topic': 'お茶', 'role': 'AIMessage', 'messages': [AIMessage(content='Sure! Oogiri is a form of Japanese comedy that <・・・中略・・・>], 'oogiri_en': 'Sure! Oogiri is a form of Japanese comedy that <・・・中略・・・>'}

```

インプットを分岐して日本語の大喜利と英語の大喜利を行うので、合体したデータをダミーでつくって実行させます。

```Python:test(日英比較はダミーで実行）
state_4 = StateDict(topic='お茶', oogiri_en='英語で大喜利', oogiri_jp='日本語で大喜利', messages=['test1','test2'])
state_5 = comparison_template_func(state_4)
print("日英比較\n", state_5)
```
```output:output
日英比較
 {'role': 'AIMessage', 'messages': ['大喜利とは、<・・・中略・・・>'], 'result': '大喜利とは、特定の<・・・中略・・・>ます。'}

```
上手くいった！ぃえぃ！


### Chainを組む
今回はチェーンの最初と最後を`set_entry_point`と`set_entry_point`を使う書き方でやってみましょう。
```Python:Chainを組む
graph_builder = StateGraph(StateDict)

# nodeの定義
graph_builder.add_node('pass_through', pass_through_node)
graph_builder.add_node("oogiri_japanese", oogiri_japanese)
graph_builder.add_node("oogiri_english", oogiri_english)
graph_builder.add_node("comparison_template_func", comparison_template_func)
# 日本語大喜利
graph_builder.set_entry_point('pass_through')
graph_builder.add_edge('pass_through', "oogiri_japanese")
# 英語大喜利
graph_builder.add_edge("pass_through", "oogiri_english")
# merge
graph_builder.add_edge("oogiri_japanese", "comparison_template_func")
graph_builder.add_edge("oogiri_english", "comparison_template_func")
graph_builder.set_finish_point('comparison_template_func')

graph = graph_builder.compile()
```

```Python:Chainの可視化
from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass
```

![output.jpeg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3655320/cae1b7cb-806a-d66c-bcfd-5e9853f5f4e0.jpeg)


### 実行
```Python:インプット用のstateを作る
initial_query = StateDict(topic='靴下')
print(initial_query)
```
```Python:推論
result = graph.invoke(initial_query)
print(result)
```
```output:output
{'topic': '靴下',
 'oogiri_jp': 'もちろんです！靴下についての大喜利、いきますよ！\n\n1. 「靴下が<・・・中略・・・>',
 'oogiri_en': 'Sure! Oogiri is a form of Japanese <・・・中略・・・>',
 'result': '日本語の大喜利と英語の大喜利を比較することで、<・・・中略・・・>示しています。',
 'messages': [AIMessage(content='もちろんです！靴下についての大喜利、いきますよ！\n\n1. 「靴下が喋ったら<・・・中略・・・>', <・・・中略・・・>,
  HumanMessage(content='日本語の大喜利と英語の大喜利を比較することで、<・・・中略・・・>]}
```


結果を取り出すには単に辞書形式なのでこんな感じ
```python:resultのアウトプット
print(result['topic'])
print(result['result'])
```

# StateModelを使う方法
テンプレートは共用で行きましょう！（面倒だし）
### いろいろと定義
stateの定義をします。
```python
class StateModel(BaseModel):
    topic_en: str = Field(default='', description='topic for oogiri')
    topic_ja: str = Field(default='', description='話題')
    oogiri_en: str = Field(default='', description='English')
    oogiri_ja: str = Field(default='', description='Japanese')
    result: str = Field(default='', description='result')

    messages: Annotated[list, add_messages] = Field(default=[], description='履歴')
    # 今回は入出力のメタデータを残すので`[str]`を削除
```

ノードとなる関数は書き換えが必要
stateからの取り出しはJson風に書きます。
例：辞書の場合`state['topic']` -> BaseModelの場合`state.topic`
```python
def oogiri_japanese(state: StateModel):
    input = prompt_template0.invoke(state.topic_ja)
    output = model.invoke(input)
    
    state.messages += input.messages
    state.messages += [output]
    state.oogiri_ja = output.content
    return state

def oogiri_english(state: StateModel):
    # topicの英訳
    translate = topic2english | model | output_parser
    topic_en = translate.invoke(state.topic_ja)
    # プロンプトの作成
    input = oogiri_prompt.invoke(topic_en)
    # 推論
    output = model.invoke(input.messages)

    # stateに保存
    state.messages += input.messages
    state.messages += [output]
    state.oogiri_en = output.content
    state.topic_en = topic_en
    return state

def comparison_template_func(state: StateModel):
    # 日本語と英語の大喜利を取り出してプロンプトを作る
    input = template_compare.invoke({'oogiri_ja': state.oogiri_ja, 'oogiri_en': state.oogiri_en})
    # 推論
    output = model.invoke(input)
    # stateに保存
    state.messages += input.messages
    state.messages += [output]
    state.result = output_parser.invoke(output)
    return state
    
```

### テスト
ほんじゃま、テスト。今回はシーケンシャルに実行します。
```python:test
state_6 = StateModel(topic_ja='お茶')
print("入力のstateを作る\n", state_6)
state_6 = oogiri_japanese(state_6)
print("日本語で大喜利\n", state_6)
state_6 = oogiri_english(state_6)
print("英語で大喜利\n", state_6)
state_6 = comparison_template_func(state_6)
print("日英比較\n", state_6)
```

よしよし
簡単に言ってるけど、実は`state.massages`に履歴を入れる方法が面倒でした。

```output:output
入力のstateを作る
 topic_en='' topic_ja='お茶' oogiri_en='' oogiri_ja='' result='' messages=[]
日本語で大喜利
 topic_en='' topic_ja='お茶' oogiri_en='' oogiri_ja='もちろん、<・・・中略・・・>' result='' messages=[SystemMessage(content='あなたは<・・・中略・・・>), AIMessage(content='もちろん、お茶について大喜利を楽しんで<・・・中略・・・>')]
英語で大喜利
 topic_en='Tea' topic_ja='お茶' oogiri_en='Sure! Here are a few OOGIRI-style j<・・・中略・・・>' oogiri_ja='もちろん、お茶について大喜利を楽しんで<・・・中略・・・>' result='' messages=[SystemMessage(content='You are a <・・・中略・・・>]
日英比較
 topic_en='Tea' topic_ja='お茶' oogiri_en='Sure! Here are a <・・・中略・・・>' oogiri_ja='もちろん、お茶について大喜利<・・・中略・・・>' result="日本語の大喜利と英語の大喜利には、それぞれの文化や価値観が反映されています。以下に、いくつかの具体的な例を挙げながら、文化の違いを<・・・中略・・・>" messages=[SystemMessage(content='あなたはお笑いを客観的に<・・・中略・・・>]
```

### Chainを組む

今回のグラフは`START`と`END`を使う書き方。
`set_entry_point`と`set_entry_point`を使う書き方とちょっとだけ違う。
まぁ、気にしなくていいと思う。

```python
graph_builder2 = StateGraph(StateModel)

graph_builder2.add_node("oogiri_japanese", oogiri_japanese)
graph_builder2.add_node("oogiri_english", oogiri_english)
graph_builder2.add_node("comparison_template_func", comparison_template_func)


graph_builder2.add_edge(START, "oogiri_japanese")
graph_builder2.add_edge("oogiri_japanese", "oogiri_english")
graph_builder2.add_edge("oogiri_english", "comparison_template_func")
graph_builder2.add_edge("comparison_template_func", END)
graph2 = graph_builder2.compile()
```
```python
from IPython.display import Image, display

try:
    display(Image(graph2.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass
```
![output2.jpeg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3655320/09f38446-e420-167a-d875-207a1b3dd956.jpeg)



### 実行

```python:インプット用のstateを作る
initial_query2 = StateModel(topic_ja='靴下')
print(initial_query2)
```
こんな感じのインプットができました。
```output:output
StateModel(topic_en='', topic_ja='靴下', oogiri_en='', oogiri_ja='', result='', messages=[])
```


```python:推論
result2 = graph2.invoke(initial_query2)
print(result2)
```
```output:output
{'topic_en': 'Socks',
 'topic_ja': '靴下',
 'oogiri_en': 'Sure! OOGIRI is a type of Japanese wordplay or joke format<・・・中略・・・>)',
 'oogiri_ja': 'もちろんです！靴下についての大喜利いきますね。\n\n1. 「靴下の裏<・・・中略・・・)',
 'result': '日本語の大喜利と英語の大喜利を比較すると、文化的な違いがいくつか見え<・・・中略・・・)',
 'messages': [SystemMessage(content='あなたは偉大な漫才師です。', additional_<・・・中略・・・),
  HumanMessage(content='靴下について大喜利してください。', additional_kwargs=<・・・中略・・・),
  AIMessage(content='もちろんです！靴下についての大喜利いきますね。\n\n1. 「靴<・・・中略・・・),
  SystemMessage(content='You are a great comedian', additional_kwargs={}<・・・中略・・・),
  HumanMessage(content='Please create OOGIRI about Socks in English''<・・・中略・・・),
  AIMessage(content='Sure! OOGIRI is a type of Japanese wordplay or <・・・中略・・・),
  SystemMessage(content='あなたはお笑いを客観的に分析ができる優秀な評論<・・・中略・・・),
  HumanMessage(content='ここまでの会話で、英語で書かれた大喜利と、日本語<・・・中略・・・),
  AIMessage(content='日本語の大喜利と英語の大喜利を比較すると、文化的な<・・・中略・・・)]
}
```
おお、なぜか出力は**辞書形式**。ｗ
めっちゃ落とし穴やん


# 二つの方法の`state`について
良く検証ができていませんが、`TypedDict`と`BaseModel`を使う方法がありましたが、挙動が少し違う気がします。
TypedDictの方ではstateの変数の型定義がリスト型であれば辞書でstateを出力したらそのまま変数にアペンドされる感じ。そのstateはどこかのエッヂにのっているわけではなくて、別のメモリーにいるような挙動。
BaseModelはエッヂを伝ってstateが移動していくような挙動をしています。

BaseModelで二手に分岐した場合はこんなエラーが出ました。
```
Can receive only one value per step. Use an Annotated key to handle multiple values.
訳：1回のステップで1つの値しか受け取れません。複数の値を処理するには、注釈付きのキーを使用してください。
```
このエラー表示と共に出力されるURLには辞書の対応方法だけ記載されてました。
BaseModelだとどうするんですかね？

https://langchain-ai.github.io/langgraph/troubleshooting/errors/INVALID_CONCURRENT_GRAPH_UPDATE/




# 終わりに
これでLangGraphの基礎的な理解ができました。
何と言っても肝は`state`、`node`、`edge`この三つ
特に`state`の理解には`PYdantic`の理解が必要でした。
さて次はtoolsをやってみるか、チャットボットをやってみるか。
チャットボットはトークン数の制限。
toolsはいろんな機能をやらせること。
いずれもやってみたいと思ってますが、1㎜もドキュメントを読んでないのでしばしお待ちを。