LangGraph Quick Start

LangGraphを使うときの定番のインストール

%%capture --no-stderr
%pip install -U langgraph langsmith
%pip install -U langchain_openai

import os
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")

Koichiro Mori

LLMノードを一つ追加した簡単なワークフロー

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from IPython.display import Image, display


class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

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


# chatbotノード
def chatbot(state: State):
    print("@chatbot")
    print("***", state)

    # 過去のメッセージ履歴をすべてLLMに入力して出力を生成
    response = llm.invoke(state["messages"])

    # 今回の結果だけ返せば add_messages が動いてStateに追加される？ => YES!!!
    return {"messages": [response]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

Koichiro Mori

1回目のワークフローの呼び出し

graph.invoke() には初期Stateを渡す
Stateの定義にあるようにmessages（List）を渡す
LangChainのHumanMessageに自動的に変換されている

response = graph.invoke({"messages": [("user", "こんにちは!"), ("user", "私はスミスです。")]})
print(type(response))
print(len(response["messages"]))
print(response["messages"][0].content)
print(response["messages"][1].content)
print(response["messages"][2].content)

戻り値はStateなのか？
messagesにはアクセスできる
入力だけではなく生成した値が最後に追加された状態で返る（自分でmessagesにappendとかしてはいけない）

@chatbot
*** {'messages': [HumanMessage(content='こんにちは!', id='0fa91483-7ee1-4b87-a1d4-22f54e97b660'), HumanMessage(content='私はスミスです。', id='ae7a0750-a503-4e19-9e54-26739596662b')]}
<class 'langgraph.pregel.io.AddableValuesDict'>
3
こんにちは!
私はスミスです。
こんにちは、スミスさん！今日はどんなお手伝いをしましょうか？

状態でただのlistにしてしまうと自動的に追加されない

class State(TypedDict):
    messages: list

Koichiro Mori

2回目のワークフローの呼び出し

1回目の呼び出しの状態はクリアされている
ENDまで到達したら状態がクリアされる仕様？

# 前回の呼び出しのStateは消えてしまう
# __end__まで行ったら全部クリアされる？
response = graph.invoke({"messages": ("user", "私の名前は？")})
len(response["messages"])
print(response["messages"][0].content)
print(response["messages"][1].content)

@chatbot
*** {'messages': [HumanMessage(content='私の名前は？', id='ffc2edf0-9fc1-4b55-9d7a-1f70ea5ee0ae')]}
私の名前は？
申し訳ありませんが、あなたの名前はわかりません。何かお手伝いできることがあれば教えてください。

Koichiro Mori

状態にプロパティを追加

class State(TypedDict):
    messages: Annotated[list, add_messages]
    user_name: str
    age: int

def chatbot(state: State):
    print("@chatbot")
    print("***", state)

    # 過去のメッセージ履歴をすべてLLMに入力して出力を生成
    response = llm.invoke(state["messages"])

    # 今回の結果だけ返せば add_messages が動いてStateに追加される？ => YES!!!
    return {"messages": [response], "age": state["age"] * 100}

response = graph.invoke({"messages": [("user", "こんにちは!")],
                         "user_name": "mori",
                         "age": 3392})
print(response.keys())
print(response["messages"])   # ノードで返した値が追加されている
print(response["user_name"])
print(response["age"])

@chatbot
*** {'messages': [HumanMessage(content='こんにちは!', id='b69799ad-9342-47d9-a80c-eea09a23c5bd')], 'user_name': 'mori', 'age': 3392}
dict_keys(['messages', 'user_name', 'age'])
[HumanMessage(content='こんにちは!', id='b69799ad-9342-47d9-a80c-eea09a23c5bd'), AIMessage(content='こんにちは! 今日はどんなことをお手伝いできますか？', response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 9, 'total_tokens': 24}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_bc2a86f5f5', 'finish_reason': 'stop', 'logprobs': None}, id='run-446ad07e-283f-4feb-aec4-8d050b049bfe-0', usage_metadata={'input_tokens': 9, 'output_tokens': 15, 'total_tokens': 24})]
mori
339200

Koichiro Mori

ワークフローをまたいで状態を記憶する

Part3に書いてある
モデルのcompile時にmemoryを渡す
ターンをまたいでStateが継続される

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from IPython.display import Image, display
from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

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


# chatbotノード
def chatbot(state: State):
    print("@chatbot")
    print("***", state)

    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile(checkpointer=memory)
display(Image(graph.get_graph().draw_mermaid_png()))

1回目のターン

import uuid
config = {"configurable": {"thread_id": uuid.uuid4()}}
response = graph.invoke({"messages": [("user", "私はスミスです。")]}, config)
for i, message in enumerate(response["messages"]):
    print(i, message.content)

@chatbot
*** {'messages': [HumanMessage(content='私はスミスです。', id='89c52b9d-2d2e-4e8d-956f-dc1a2628450a')]}
0 私はスミスです。
1 こんにちは、スミスさん。今日はどんなお手伝いができるでしょうか？

2回目のターン

invokeしたときのStateに過去の状態が保存されている
今回はmessagesなので過去の会話履歴がすべて維持される

# 2回目のターン
# 会話を引き継ぐために前回と同じthread_idのconfigを渡す
response = graph.invoke({"messages": [("user", "私の名前はわかりますか？")]}, config)
for i, message in enumerate(response["messages"]):
    print(i, message.content)

@chatbot
*** {'messages': [HumanMessage(content='私はスミスです。', id='89c52b9d-2d2e-4e8d-956f-dc1a2628450a'), AIMessage(content='こんにちは、スミスさん。今日はどんなお手伝いができるでしょうか？', response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 13, 'total_tokens': 33}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_4e2b2da518', 'finish_reason': 'stop', 'logprobs': None}, id='run-f9adab83-c32a-4fd4-b585-6f38af548c51-0', usage_metadata={'input_tokens': 13, 'output_tokens': 20, 'total_tokens': 33}), HumanMessage(content='私の名前はわかりますか？', id='e35d6caf-695f-4a48-8034-3f342a94bf86')]}
0 私はスミスです。
1 こんにちは、スミスさん。今日はどんなお手伝いができるでしょうか？
2 私の名前はわかりますか？
3 はい、スミスさんですね。何かお手伝いできることがあれば教えてください。

別のスレッドID

会話は別になる

# 別のスレッドIDを渡す
config = {"configurable": {"thread_id": uuid.uuid4()}}
response = graph.invoke({"messages": [("user", "私の名前はわかりますか？")]}, config)
for i, message in enumerate(response["messages"]):
    print(i, message.content)

@chatbot
*** {'messages': [HumanMessage(content='私の名前はわかりますか？', id='8e2c5b3a-2274-4312-8ca8-44c759d1d405')]}
0 私の名前はわかりますか？
1 すみませんが、個々のユーザーの名前や個人情報を把握することはできません。何か他にお手伝いできることがあれば教えてください。

Koichiro Mori

ノード間で状態を伝搬する

ノード1でユーザの入力から名前と年齢を抽出して状態にセット
ノード2でユーザに対してあいさつする

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from IPython.display import Image, display
import json
from langchain_core.pydantic_v1 import BaseModel

class State(TypedDict):
    messages: Annotated[list, add_messages]
    name: str
    age: int


class User(BaseModel):
    name: str
    age: int

graph_builder = StateGraph(State)

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


# chatbotノード
def node1(state: State):
    print("@node1")
    print("***", state)

    prompt = ChatPromptTemplate([
        ("system", "ユーザの名前と年齢を抽出してJSONで出力してください。キーはnameとageです。"),
        ("human", "{input}")
    ])
    chain = prompt | llm.with_structured_output(User)
    user_data = chain.invoke({"input": state["messages"]})
    return {"name": user_data.name, "age": user_data.age}


def node2(state: State):
    print("@node2")
    print("***", state)

    prompt = ChatPromptTemplate([
        ("system", "このユーザの名前と年齢を呼んであいさつしてください。\nユーザの名前は{name}、年齢は{age}です。"),
    ])
    chain = prompt | llm
    response = chain.invoke({"name": state["name"], "age": state["age"]})
    return {"messages": [response]}


graph_builder.add_node("名前と年齢を抽出する", node1)
graph_builder.add_node("ご挨拶", node2)
graph_builder.add_edge(START, "名前と年齢を抽出する")
graph_builder.add_edge("名前と年齢を抽出する", "ご挨拶")
graph_builder.add_edge("ご挨拶", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

response = graph.invoke({"messages": [("user", "私はスミスです。20歳です。")]})
print(response["messages"][-1].content)

出力

@node1
*** {'messages': [HumanMessage(content='私はスミスです。20歳です。', id='baa89498-0f1e-4b0e-be7f-7a2686b8b81f')], 'name': None, 'age': None}
@node2
*** {'messages': [HumanMessage(content='私はスミスです。20歳です。', id='baa89498-0f1e-4b0e-be7f-7a2686b8b81f')], 'name': 'スミス', 'age': 20}
こんにちは、スミスさん！20歳のお祝いを申し上げます。今日も素晴らしい一日をお過ごしください。何かお手伝いできることがあれば教えてくださいね。

ノード1で名前と年齢が抽出できるまで繰り返すというループがあるとよい

このスクラップは4ヶ月前にクローズされました