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

一、初識LangGraph

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

LangGraph的核心思想是將應(yīng)用程序的工作流程表示為圖結(jié)構(gòu)，其中包含節(jié)點和邊。節(jié)點代表工作流中的離散工作單元或計算任務(wù)，每個節(jié)點都是一個Python函數(shù)，負(fù)責(zé)處理當(dāng)前狀態(tài)并返回更新后的狀態(tài)。節(jié)點可以執(zhí)行諸如調(diào)用LLM、與工具或API交互以及操作數(shù)據(jù)等任務(wù)。而邊則連接節(jié)點，定義了執(zhí)行流程。邊可以是簡單的無條件轉(zhuǎn)換，也可以是基于節(jié)點輸出的條件分支邏輯，類似于if-else語句，從而實現(xiàn)工作流中的動態(tài)決策。

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

二、LangGraph的關(guān)鍵概念

（一）圖結(jié)構(gòu)

如前文所述，LangGraph的工作流程基于圖結(jié)構(gòu)，由節(jié)點和邊組成。節(jié)點是工作流的基本構(gòu)建塊，負(fù)責(zé)處理當(dāng)前狀態(tài)并返回更新后的狀態(tài)。邊則定義了節(jié)點之間的執(zhí)行順序，可以是簡單的直接轉(zhuǎn)換，也可以是基于條件的分支邏輯。

（二）狀態(tài)管理

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

（三）多代理系統(tǒng)

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

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

（四）持久化

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

檢查點類似于圖狀態(tài)的快照，其中包含以下內(nèi)容：

• 配置信息：在該步驟中使用的配置信息。
• 元數(shù)據(jù)：步驟的詳細(xì)信息，例如正在運行的節(jié)點。
• 值：該點的實際狀態(tài)值。
• 下一步：將要運行的下一個節(jié)點。
• 任務(wù)：即將執(zhí)行的任務(wù)或錯誤信息。

在執(zhí)行過程中，每個圖都需要一個線程ID來組織其檢查點?？梢酝ㄟ^配置提供此線程ID。例如：

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

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

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

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

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

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

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

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

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

（五）人機協(xié)作（Human-in-the-Loop）

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

要使用人機協(xié)作，需要定義一個檢查點，并在節(jié)點中添加??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支持流式輸出，這意味著在生成輸出的同時，用戶可以實時看到結(jié)果。這種方式不僅提高了用戶體驗，還使得應(yīng)用程序更具響應(yīng)性。LangGraph支持三種主要的數(shù)據(jù)類型流式輸出：工作流進(jìn)度、LLM令牌和自定義更新。

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

• ??"values"??：每個圖步驟后的完整狀態(tài)。
• ??"updates"??：每個節(jié)點后的更改。
• ??"custom"??：在節(jié)點中記錄的任何自定義數(shù)據(jù)。
• ??"messages"??：帶有元數(shù)據(jù)的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()??方法，這對于遷移大型應(yīng)用程序非常有用。

三、為什么選擇LangGraph？

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

（一）可靠且可控

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

（二）定制化和可擴(kuò)展性

LangGraph提供了低級工具，開發(fā)者可以根據(jù)自己的需求構(gòu)建代理，設(shè)計具有特定角色的代理系統(tǒng)。這種靈活性使得開發(fā)者能夠根據(jù)具體需求定制化解決方案。

（三）出色的流式輸出

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

四、構(gòu)建最簡單的LangGraph

現(xiàn)在我們已經(jīng)了解了LangGraph的關(guān)鍵組件，接下來讓我們通過一個簡單的例子來構(gòu)建一個包含三個節(jié)點和一個條件邊的基本圖。這個例子將展示如何使用狀態(tài)、節(jié)點和邊等關(guān)鍵概念來調(diào)用一個圖。

（一）定義圖狀態(tài)

狀態(tài)定義了在節(jié)點之間共享的數(shù)據(jù)結(jié)構(gòu)，它就像一個在圖中流動的共享內(nèi)存。我們可以使用Python的??TypedDict???來聲明狀態(tài)的結(jié)構(gòu)。在這個例子中，我們定義了一個名為??graph_state??的鍵，它存儲了一個字符串。

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

（二）創(chuàng)建節(jié)點

節(jié)點是簡單的Python函數(shù)，每個節(jié)點接收當(dāng)前狀態(tài)，對其進(jìn)行修改，并返回更新后的狀態(tài)。在這個例子中，我們定義了三個節(jié)點，每個節(jié)點都會在??graph_state??中添加不同的內(nèi)容。

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!"}

（三）添加條件邏輯

有時候我們希望工作流的行為是動態(tài)的，即下一步驟取決于某些邏輯或隨機性。條件邊可以實現(xiàn)這一點。在這個例子中，我們定義了一個函數(shù)??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"

（四）構(gòu)建圖

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

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

# 初始化圖并指定狀態(tài)結(jié)構(gòu)
builder = StateGraph(State)

# 添加節(jié)點到圖中
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()))

（五）調(diào)用圖

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

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

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

五、使用LangGraph構(gòu)建支持聊天機器人

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

（一）設(shè)置

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

!pip install -U langgraph langchain openai

這些包分別是：

• LangGraph：用于構(gòu)建圖結(jié)構(gòu)。
• LangChain：用于與OpenAI的語言模型進(jìn)行交互。
• OpenAI：用于使用OpenAI的模型（如GPT-4）。

我們需要安全地提供OpenAI API密鑰，以便應(yīng)用程序能夠認(rèn)證并使用GPT模型。以下函數(shù)會在環(huán)境變量中未設(shè)置密鑰時提示輸入。

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

（二）構(gòu)建基本聊天機器人

定義狀態(tài)

狀態(tài)定義了在圖的節(jié)點之間傳遞的數(shù)據(jù)結(jié)構(gòu)。在這個例子中，我們定義了一個名為??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]

創(chuàng)建圖構(gòu)建器

??StateGraph??對象是定義圖結(jié)構(gòu)的入口點。我們使用剛剛定義的狀態(tài)初始化它。

graph_builder = StateGraph(State)

添加聊天機器人節(jié)點

我們定義了一個名為??chatbot???的Python函數(shù)，它接收當(dāng)前狀態(tài)，調(diào)用OpenAI的GPT模型，并將模型的響應(yīng)作為更新返回到狀態(tài)的??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)

設(shè)置入口和出口點

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

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

編譯圖

定義完所有節(jié)點和邊后，編譯圖結(jié)構(gòu)。

graph = graph_builder.compile()

可視化（可選）

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

from IPython.display import Image, display

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

運行聊天機器人

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

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)

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

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

1. 添加工具集成

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

（1）安裝工具依賴

%pip install -U tavily-python langchain_community

（2）設(shè)置工具API密鑰

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

（3）定義工具

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

from langchain_community.tools.tavily_search import TavilySearchResults

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

2. 添加記憶功能

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

（1）添加檢查點

使用??MemorySaver??檢查點將對話狀態(tài)存儲在內(nèi)存中。在生產(chǎn)環(huán)境中，可以使用持久化后端，如SQLite或Postgres。

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

3. 添加人機協(xié)作

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

（1）定義人工協(xié)助工具

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暫停圖的執(zhí)行并等待輸入
    human_response = interrupt({"query": query})
    return human_response["data"]

這個工具會暫停圖的執(zhí)行，并等待人工輸入后再繼續(xù)。

（四）部署LangGraph應(yīng)用

構(gòu)建完LangGraph應(yīng)用后，下一步就是將其部署到本地機器或云平臺，以便進(jìn)行進(jìn)一步的開發(fā)和測試。LangGraph提供了多種部署選項，每種選項都有不同的工作流程和基礎(chǔ)設(shè)施。

• 云SaaS模型：一切由LangChain托管，用戶無需管理任何基礎(chǔ)設(shè)施。
• 自托管數(shù)據(jù)平面：用戶可以在自己的云環(huán)境中運行應(yīng)用，同時使用LangChain的控制平面。
• 自托管控制平面：用戶自行管理所有基礎(chǔ)設(shè)施。
• 獨立容器：使用Docker實現(xiàn)完全的靈活性。

六、LangGraph的用例

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

（一）改進(jìn)客戶服務(wù)

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

（二）AI研究助手

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

（三）個性化學(xué)習(xí)

LangGraph還可以用于構(gòu)建個性化的學(xué)習(xí)系統(tǒng)，根據(jù)學(xué)習(xí)者的特點調(diào)整內(nèi)容。它能夠幫助學(xué)習(xí)者識別薄弱環(huán)節(jié)，并推薦相應(yīng)的學(xué)習(xí)資源，從而提高學(xué)習(xí)效果。

（四）簡化業(yè)務(wù)流程

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

七、總結(jié)

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

LangGraph對于開發(fā)者來說是一個非常有價值的工具。它可以幫助我們創(chuàng)建強大的、由AI驅(qū)動的應(yīng)用程序。通過LangGraph，我們可以構(gòu)建靈活、適應(yīng)性強的系統(tǒng)，能夠處理復(fù)雜的任務(wù)。無論你是想構(gòu)建聊天機器人、研究助手還是個性化學(xué)習(xí)工具，LangGraph都提供了所需的結(jié)構(gòu)和工具，助力高效開發(fā)。

本文轉(zhuǎn)載自??Halo咯咯??    作者：基咯咯