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

ここまで

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

今日やること

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を受け渡す方法

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

先ずはライブラリのインポート

いっぱいなんで嫌んなる。

ライブラリのインポート

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は出なかった。

環境系の設定

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"

モデルの用意

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")
)

テンプレートの用意

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}
     ''')
])

辞書型を使う方法

いろいろと定義

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が入出力されるというコトをイメージして関数を書いていきます。

ノードになる関数を用意

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をモデルに渡して出力します。

テスト

関数が上手くつなげられるかテストしてみましょう

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

入力の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 <・・・中略・・・>'}

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

test(日英比較はダミーで実行）

state_4 = StateDict(topic='お茶', oogiri_en='英語で大喜利', oogiri_jp='日本語で大喜利', messages=['test1','test2'])
state_5 = comparison_template_func(state_4)
print("日英比較\n", state_5)

output

日英比較
 {'role': 'AIMessage', 'messages': ['大喜利とは、<・・・中略・・・>'], 'result': '大喜利とは、特定の<・・・中略・・・>ます。'}

上手くいった！ぃえぃ！

Chainを組む

今回はチェーンの最初と最後をset_entry_pointとset_entry_pointを使う書き方でやってみましょう。

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

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

実行

インプット用のstateを作る

initial_query = StateDict(topic='靴下')
print(initial_query)

推論

result = graph.invoke(initial_query)
print(result)

output

{'topic': '靴下',
 'oogiri_jp': 'もちろんです！靴下についての大喜利、いきますよ！\n\n1. 「靴下が<・・・中略・・・>',
 'oogiri_en': 'Sure! Oogiri is a form of Japanese <・・・中略・・・>',
 'result': '日本語の大喜利と英語の大喜利を比較することで、<・・・中略・・・>示しています。',
 'messages': [AIMessage(content='もちろんです！靴下についての大喜利、いきますよ！\n\n1. 「靴下が喋ったら<・・・中略・・・>', <・・・中略・・・>,
  HumanMessage(content='日本語の大喜利と英語の大喜利を比較することで、<・・・中略・・・>]}

結果を取り出すには単に辞書形式なのでこんな感じ

resultのアウトプット

print(result['topic'])
print(result['result'])

StateModelを使う方法

テンプレートは共用で行きましょう！（面倒だし）

いろいろと定義

stateの定義をします。

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

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
    

テスト

ほんじゃま、テスト。今回はシーケンシャルに実行します。

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

入力の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を使う書き方とちょっとだけ違う。
まぁ、気にしなくていいと思う。

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

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

実行

インプット用のstateを作る

initial_query2 = StateModel(topic_ja='靴下')
print(initial_query2)

こんな感じのインプットができました。

output

StateModel(topic_en='', topic_ja='靴下', oogiri_en='', oogiri_ja='', result='', messages=[])

推論

result2 = graph2.invoke(initial_query2)
print(result2)

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だとどうするんですかね？

終わりに

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