前言

正文

基本使用與介紹

參考文章：https://langchain-ai.github.io/langgraph/tutorials/

from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph.message import add_messages


class State(TypedDict):
    # Messages have the type "list". The `add_messages` function
    # in the annotation defines how this state key should be updated
    # (in this case, it appends messages to the list, rather than overwriting them)
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)

首先建立一個 StateGraph。 StateGraph 物件繼承State物件，在state class中會定義在這整個系統中，會使用到的各種變數，State可以說是整個變數的集合，我們等等將新增節點來表示可以呼叫的 llm 和函數，並新增邊來展示function之間轉換。

from langchain_openai import ChatOpenAI
import os
from langgraph.graph import StateGraph, START, END

os.environ["OPENAI_API_KEY"] = ""
llm = ChatOpenAI(model_name="gpt-4o")

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}


# The first argument is the unique node name
# The second argument is the function or object that will be called whenever
# the node is used.
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

我們定義了一個chatbot function，這個函式會去呼叫LLM，並且將此message回傳成為StateGraph的狀態變數message的內容

畫出工作流程圖

# draw
try:
    image_data = graph.get_graph().draw_mermaid_png()  # 二進制資料
    with open('day28_LangGraph_workflow.png', 'wb') as f:
        f.write(image_data)
except Exception as e:
    print(f"意外: {e}")
    pass

最後我們可以將這個系統的工作流程圖畫出來

完整程式碼

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


class State(TypedDict):
    # Messages have the type "list". The `add_messages` function
    # in the annotation defines how this state key should be updated
    # (in this case, it appends messages to the list, rather than overwriting them)
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)

os.environ["OPENAI_API_KEY"] = ""
llm = ChatOpenAI(model_name="gpt-4o")


def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}


# The first argument is the unique node name
# The second argument is the function or object that will be called whenever
# the node is used.
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
# draw
try:
    image_data = graph.get_graph().draw_mermaid_png()  # 二進制資料
    with open('day28_LangGraph_workflow.png', 'wb') as f:
        f.write(image_data)
except Exception as e:
    print(f"意外: {e}")
    pass

工作流程圖