在當今數字化時代，利用大型語言模型（LLMs）構建應用程序變得越來越受歡迎。這些強大的工具能夠幫助我們創建出智能且交互性強的系統，比如聊天機器人、智能助手等。然而，隨著應用復雜度的提升，尤其是當多個LLMs協同工作時，問題也隨之而來。如何管理這些模型之間的信息流動？如何確保它們能夠順暢地理解和執行任務？今天，就讓我們一起走進LangGraph的世界，看看它是如何解決這些問題的。

一、初識LangGraph

LangGraph是LangChain工具集中的一個重要組成部分，它是一個強大的庫，專門用于簡化LLMs的集成過程。通過LangGraph，開發者可以輕松地構建和管理包含多個LLM代理的應用程序，確保這些代理能夠無縫協作，高效地完成各種任務。

LangGraph的核心思想是將應用程序的工作流程表示為圖結構，其中包含節點和邊。節點代表工作流中的離散工作單元或計算任務，每個節點都是一個Python函數，負責處理當前狀態并返回更新后的狀態。節點可以執行諸如調用LLM、與工具或API交互以及操作數據等任務。而邊則連接節點，定義了執行流程。邊可以是簡單的無條件轉換，也可以是基于節點輸出的條件分支邏輯，類似于if-else語句，從而實現工作流中的動態決策。

LangGraph特別適合使用有向無環圖（DAG）來處理直線任務，但由于它支持循環結構，能夠回溯，因此可以構建更復雜、更靈活的系統。比如，一個智能代理可以根據新信息重新思考并更新響應，或者改變決策路徑。

二、LangGraph的關鍵概念

（一）圖結構

如前文所述，LangGraph的工作流程基于圖結構，由節點和邊組成。節點是工作流的基本構建塊，負責處理當前狀態并返回更新后的狀態。邊則定義了節點之間的執行順序，可以是簡單的直接轉換，也可以是基于條件的分支邏輯。

（二）狀態管理

在多代理系統中，跟蹤任務的當前狀態至關重要。LangGraph通過自動管理狀態來解決這一問題。它維護一個主狀態對象，并在代理執行任務時實時更新該對象。狀態對象中存儲了重要的信息，比如聊天歷史記錄、用戶偏好、過去的動作、外部數據等。這些信息可以幫助代理在執行任務時做出更明智的決策。

（三）多代理系統

多代理系統由多個獨立的代理組成，它們可以合作或競爭以實現共同的目標。這些代理利用LLMs進行決策，并控制應用程序的流程。隨著代理數量和任務復雜度的增加，系統可能會面臨諸如決策質量下降、上下文管理困難等問題。多代理系統通過將系統分解為專注于特定任務的小型代理來解決這些問題，例如規劃或研究。

多代理系統的主要優勢在于模塊化、專業化和控制能力。模塊化使得開發、測試和維護變得更加容易；專業化確保了每個代理都能在其特定領域發揮專長，從而提高整體性能；控制能力則確保了開發者可以清晰地定義代理之間的通信方式。

（四）持久化

持久化是指保存流程的進度，以便在中斷后能夠從中斷點繼續執行。LangGraph通過檢查點（checkpoints）來實現持久化。在執行過程中，LangGraph會在每個主要步驟后保存圖的狀態，每個保存的狀態稱為一個檢查點。這些檢查點被組織在一個線程（特定運行的對話歷史）中。

檢查點類似于圖狀態的快照，其中包含以下內容：

• 配置信息：在該步驟中使用的配置信息。
• 元數據：步驟的詳細信息，例如正在運行的節點。
• 值：該點的實際狀態值。
• 下一步：將要運行的下一個節點。
• 任務：即將執行的任務或錯誤信息。

在執行過程中，每個圖都需要一個線程ID來組織其檢查點。可以通過配置提供此線程ID。例如：

config = {"configurable": {"thread_id": "1"}}

要獲取線程中的最新狀態，可以使用以下代碼：

graph.get_state({"configurable": {"thread_id": "1"}})

要獲取特定檢查點，可以使用以下代碼：

graph.get_state({
  "configurable": {
    "thread_id": "1", 
    "checkpoint_id": "your_checkpoint_id"
  }
})

要獲取狀態歷史記錄或所有先前狀態，可以使用以下代碼：

history = graph.get_state_history({"configurable": {"thread_id": "1"}})

還可以在任何時間手動更新或編輯狀態，使用以下代碼：

graph.update_state(
    cnotallow={"configurable": {"thread_id": "1"}},
    values={"foo": "new_value"}
)

（五）人機協作（Human-in-the-Loop）

人機協作允許在自動化LangGraph工作流的關鍵步驟中加入人工反饋。這對于某些任務至關重要，因為LLMs可能會生成不確定或有風險的輸出，例如在工具調用、內容生成或決策制定中。LangGraph的??interrupt()???函數使得這一過程成為可能，它可以通過暫停圖的執行，將數據呈現給人類，并使用??Command(resume=value)??方法根據人類的輸入恢復執行。這種方式支持多種模式，如批準/拒絕、編輯狀態、提供輸入或多輪對話。

要使用人機協作，需要定義一個檢查點，并在節點中添加??interrupt()??。以下是一個示例代碼：

from langgraph.types import interrupt, Command

def human_node(state):
    value = interrupt({"text_to_revise": state["some_text"]})
    return {"some_text": value}

graph = graph_builder.compile(checkpointer=checkpointer)
graph.invoke(some_input, cnotallow={"configurable": {"thread_id": "some_id"}})
graph.invoke(Command(resume="Edited text"), cnotallow={"configurable": {"thread_id": "some_id"}})

（六）流式輸出（Streaming）

LangGraph支持流式輸出，這意味著在生成輸出的同時，用戶可以實時看到結果。這種方式不僅提高了用戶體驗，還使得應用程序更具響應性。LangGraph支持三種主要的數據類型流式輸出：工作流進度、LLM令牌和自定義更新。

可以使用??.stream()???（同步）或??.astream()???（異步）方法來流式輸出結果，并通過設置??stream_mode??來控制輸出的內容：

• ??"values"??：每個圖步驟后的完整狀態。
• ??"updates"??：每個節點后的更改。
• ??"custom"??：在節點中記錄的任何自定義數據。
• ??"messages"??：帶有元數據的LLM令牌流。
• ??"debug"??：運行過程中的所有信息。

可以同時傳遞多個模式，例如：

for stream_type, data in graph.stream(inputs, stream_mode=["updates", "messages"]):
    if stream_type == "messages":
        print(data[0].content)  # AIMessageChunk
    elif stream_type == "updates":
        print(data)  # State update

如果需要完整的事件流，可以使用??.astream_events()??方法，這對于遷移大型應用程序非常有用。

三、為什么選擇LangGraph？

LangGraph非常適合開發智能且靈活的AI代理，原因如下：

（一）可靠且可控

LangGraph允許開發者添加內容審核檢查和人工審批，確保在長時間任務中保持上下文的連貫性。這對于需要高可靠性和可控性的應用場景至關重要。

（二）定制化和可擴展性

LangGraph提供了低級工具，開發者可以根據自己的需求構建代理，設計具有特定角色的代理系統。這種靈活性使得開發者能夠根據具體需求定制化解決方案。

（三）出色的流式輸出

LangGraph支持實時流式輸出，開發者可以實時查看每個令牌和步驟，跟蹤代理的思考過程。這對于需要實時反饋的應用場景非常有幫助。

四、構建最簡單的LangGraph

現在我們已經了解了LangGraph的關鍵組件，接下來讓我們通過一個簡單的例子來構建一個包含三個節點和一個條件邊的基本圖。這個例子將展示如何使用狀態、節點和邊等關鍵概念來調用一個圖。

（一）定義圖狀態

狀態定義了在節點之間共享的數據結構，它就像一個在圖中流動的共享內存。我們可以使用Python的??TypedDict???來聲明狀態的結構。在這個例子中，我們定義了一個名為??graph_state??的鍵，它存儲了一個字符串。

from typing_extensions import TypedDict
class State(TypedDict):
    graph_state: str

（二）創建節點

節點是簡單的Python函數，每個節點接收當前狀態，對其進行修改，并返回更新后的狀態。在這個例子中，我們定義了三個節點，每個節點都會在??graph_state??中添加不同的內容。

def node_1(state):
    print("---Node 1---")
    return {"graph_state": state['graph_state'] + " I am"}

def node_2(state):
    print("---Node 2---")
    return {"graph_state": state['graph_state'] + " extremely happy!"}

def node_3(state):
    print("---Node 3---")
    return {"graph_state": state['graph_state'] + " extremely sad!"}

（三）添加條件邏輯

有時候我們希望工作流的行為是動態的，即下一步驟取決于某些邏輯或隨機性。條件邊可以實現這一點。在這個例子中，我們定義了一個函數??decide_mood???，它隨機選擇??node_2???或??node_3??，模擬一個簡單的情緒選擇器。

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:
    if random.random() < 0.5:
        return "node_2"
    return "node_3"

（四）構建圖

接下來，我們使用LangGraph的??StateGraph??類將所有組件組合在一起。在這個類中，我們定義了完整的圖結構。

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# 初始化圖并指定狀態結構
builder = StateGraph(State)

# 添加節點到圖中
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 添加邊以定義流程
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# 編譯并可視化圖
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

（五）調用圖

最后，我們可以通過??invoke()???方法運行這個圖。我們將??graph_state??初始化為“Hi, this is Janvi.”，然后觀察圖的執行過程。

graph.invoke({"graph_state" : "Hi, this is Janvi."})

這個簡單的例子展示了狀態如何在圖的每個步驟中流動和更新。

五、使用LangGraph構建支持聊天機器人

在上一部分中，我們已經學會了如何構建一個簡單的LangGraph。接下來，我們將通過一個更實際的例子來展示如何使用LangGraph構建一個支持聊天機器人。我們將從基本功能開始，逐步添加功能，如網頁搜索、記憶功能和人機協作。在這個過程中，我們將看到LangGraph的核心概念是如何發揮作用的。

（一）設置

在開始構建聊天機器人之前，我們需要安裝必要的包。

!pip install -U langgraph langchain openai

這些包分別是：

• LangGraph：用于構建圖結構。
• LangChain：用于與OpenAI的語言模型進行交互。
• OpenAI：用于使用OpenAI的模型（如GPT-4）。

我們需要安全地提供OpenAI API密鑰，以便應用程序能夠認證并使用GPT模型。以下函數會在環境變量中未設置密鑰時提示輸入。

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")

（二）構建基本聊天機器人

定義狀態

狀態定義了在圖的節點之間傳遞的數據結構。在這個例子中，我們定義了一個名為??messages??的鍵，它將存儲對話消息的列表。

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

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

創建圖構建器

??StateGraph??對象是定義圖結構的入口點。我們使用剛剛定義的狀態初始化它。

graph_builder = StateGraph(State)

添加聊天機器人節點

我們定義了一個名為??chatbot???的Python函數，它接收當前狀態，調用OpenAI的GPT模型，并將模型的響應作為更新返回到狀態的??messages??鍵中。

import openai

# 初始化OpenAI GPT模型
openai.api_key = os.environ["OPENAI_API_KEY"]

def chatbot(state: State):
    response = openai.Completion.create(
        model="gpt-4",  # 也可以使用"gpt-3.5-turbo"或其他OpenAI模型
        prompt=state["messages"],
        max_tokens=150
    )
    return {"messages": [response.choices[0].text.strip()]}

graph_builder.add_node("chatbot", chatbot)

設置入口和出口點

定義圖的入口點（START）和出口點（END）。

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

編譯圖

定義完所有節點和邊后，編譯圖結構。

graph = graph_builder.compile()

可視化（可選）

LangGraph支持可視化編譯后的圖結構，這有助于理解執行流程。我們可以使用工具如pygraphviz或mermaid來可視化圖。

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    pass  # 可選可視化

運行聊天機器人

設置一個循環與聊天機器人進行交互。它接收用戶輸入，將其打包成預期的狀態格式（??{"messages": [...]}???），并使用??graph.stream???執行圖。??stream??方法會返回圖執行過程中的事件，我們打印出助手的最終消息。

def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

# 與機器人聊天的循環
whileTrue:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:  # 適用于沒有輸入功能的環境
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

（三）增強聊天機器人功能

1. 添加工具集成

為了讓聊天機器人能夠獲取最新的信息，我們可以為其添加網頁搜索工具（例如Tavily）。這需要讓LLM能夠請求工具的使用，并在圖中添加處理這些工具執行的組件。

（1）安裝工具依賴

%pip install -U tavily-python langchain_community

（2）設置工具API密鑰

_set_env("TAVILY_API_KEY")  # 使用前面定義的函數

（3）定義工具

實例化Tavily搜索工具，限制返回2個結果。這個工具將被LLM和圖使用。

from langchain_community.tools.tavily_search import TavilySearchResults

# 創建Tavily搜索工具實例，限制返回2個結果
tool = TavilySearchResults(max_results=2)
tools = [tool]  # 機器人可以使用的工具列表

2. 添加記憶功能

為了讓聊天機器人能夠進行多輪對話并記住之前的對話內容，我們需要引入LangGraph的檢查點功能。

（1）添加檢查點

使用??MemorySaver??檢查點將對話狀態存儲在內存中。在生產環境中，可以使用持久化后端，如SQLite或Postgres。

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

3. 添加人機協作

有時，AI代理可能需要在繼續之前獲得人工輸入。我們可以通過創建一個暫停圖流程的工具來實現這一點。

（1）定義人工協助工具

from langchain_core.tools import tool
from langgraph.types import interrupt

@tool
def human_assistance(query: str) -> str:
    print(f"Pausing for human assistance regarding: {query}")
    # interrupt暫停圖的執行并等待輸入
    human_response = interrupt({"query": query})
    return human_response["data"]

這個工具會暫停圖的執行，并等待人工輸入后再繼續。

（四）部署LangGraph應用

構建完LangGraph應用后，下一步就是將其部署到本地機器或云平臺，以便進行進一步的開發和測試。LangGraph提供了多種部署選項，每種選項都有不同的工作流程和基礎設施。

• 云SaaS模型：一切由LangChain托管，用戶無需管理任何基礎設施。
• 自托管數據平面：用戶可以在自己的云環境中運行應用，同時使用LangChain的控制平面。
• 自托管控制平面：用戶自行管理所有基礎設施。
• 獨立容器：使用Docker實現完全的靈活性。

六、LangGraph的用例

LangGraph可用于構建交互式和智能的AI代理，以下是一些常見的用例：

（一）改進客戶服務

LangGraph能夠開發高級的客戶支持聊天機器人。這些聊天機器人能夠回憶過去的購買記錄和客戶偏好，從而更快地解決客戶問題。當需要人工干預時，它們還可以將對話轉接給人工客服。

（二）AI研究助手

利用LangGraph，可以創建一個研究助手，它能夠搜索學術文章并突出顯示重要信息。研究人員和學生可以利用這些信息，從各個領域獲取更多見解。

（三）個性化學習

LangGraph還可以用于構建個性化的學習系統，根據學習者的特點調整內容。它能夠幫助學習者識別薄弱環節，并推薦相應的學習資源，從而提高學習效果。

（四）簡化業務流程

LangGraph可以幫助自動化業務流程，例如文檔審批和項目管理。它還可以用于數據分析，提高生產力，減少人為錯誤，讓團隊專注于更高層次的任務。

七、總結

在本篇LangGraph新手教程中，我們學習了如何構建交互式的AI系統。這些系統不僅僅是簡單的問答機器人，而是可以通過LangGraph管理狀態、集成多個代理，并允許人工輸入。我們通過構建一個支持聊天機器人，展示了LangGraph如何處理網頁搜索、記住過去的互動，甚至涉及人工干預。

LangGraph對于開發者來說是一個非常有價值的工具。它可以幫助我們創建強大的、由AI驅動的應用程序。通過LangGraph，我們可以構建靈活、適應性強的系統，能夠處理復雜的任務。無論你是想構建聊天機器人、研究助手還是個性化學習工具，LangGraph都提供了所需的結構和工具，助力高效開發。

本文轉載自??Halo咯咯??    作者：基咯咯