作者 | 崔皓

審校 | 重樓

摘要

本文介紹了一種基于LangChain的新技術(shù)LangGraph，它通過(guò)循環(huán)圖協(xié)調(diào)大模型和外部工具，解決復(fù)雜任務(wù)。首先，介紹了LangChain的DAG模型處理簡(jiǎn)單任務(wù)，以及LangGraph使用循環(huán)圖處理復(fù)雜任務(wù)的原理。然后，詳細(xì)闡述了LangGraph的三個(gè)核心組成部分：狀態(tài)圖StateGraph、節(jié)點(diǎn)Nodes和邊Edges。最后，通過(guò)一個(gè)實(shí)例演示了如何使用LangGraph查詢北京2024年春節(jié)的旅游情況，包括創(chuàng)建代理、定義圖狀態(tài)、節(jié)點(diǎn)和邊，以及執(zhí)行工作流。

開篇

在當(dāng)今的技術(shù)領(lǐng)域，大型語(yǔ)言模型（Large Language Models, LLMs）已經(jīng)逐漸成為我們?nèi)粘Ｉ詈凸ぷ髦械牡昧χ?。無(wú)論是寫作輔助、編程調(diào)試，還是簡(jiǎn)單的問(wèn)答系統(tǒng)，LLMs都能夠提供快速且準(zhǔn)確的服務(wù)。然而，當(dāng)面對(duì)一些更為復(fù)雜的任務(wù)時(shí)，LLMs可能就顯得力不從心。為了解決這些挑戰(zhàn)，我們引入了LangGraph，這是一種結(jié)合了人工智能代理（AI Agents）的新技術(shù)，旨在處理更為復(fù)雜的任務(wù)和交互。

LangGraph是建立在LangChain之上，與其生態(tài)系統(tǒng)完全兼容的新庫(kù)。它通過(guò)引入循環(huán)圖的方法，協(xié)調(diào)大模型和外部工具，從而解決應(yīng)用場(chǎng)景中的復(fù)雜問(wèn)題。

在本篇文章中，我們將通過(guò)一個(gè)簡(jiǎn)單的例子——詢問(wèn)北京2024年春節(jié)期間的旅游情況——來(lái)展示如何創(chuàng)建一個(gè)屬于自己的LangGraph應(yīng)用。我們將一步步地引導(dǎo)您了解LangGraph的概念、組成部分，以及如何將其應(yīng)用于實(shí)際場(chǎng)景中。通過(guò)這個(gè)例子，您將能夠直觀地看到LangGraph如何提高工作效率，解決復(fù)雜問(wèn)題。

什么需要使用LangGraph

使用過(guò)LangChain的朋友都知道，LangChain的核心優(yōu)勢(shì)在于其能夠輕松構(gòu)建自定義鏈，這些鏈通常是線性的，類似于有向無(wú)環(huán)圖（DAG）。DAG是一種數(shù)據(jù)結(jié)構(gòu)，其中任務(wù)按照一定的順序執(zhí)行，每個(gè)任務(wù)只有一個(gè)輸出和一個(gè)后續(xù)任務(wù)，形成一個(gè)沒(méi)有循環(huán)的線性流程。例如，當(dāng)我們從向量庫(kù)中搜索內(nèi)容時(shí)，我們首先輸入提示詞，然后通過(guò)向量比對(duì)進(jìn)行搜索，并返回結(jié)果，這個(gè)過(guò)程就是一個(gè)典型的DAG，每個(gè)步驟都嚴(yán)格按順序執(zhí)行。

有向無(wú)環(huán)圖（DAG）是數(shù)據(jù)編排和工作流管理系統(tǒng)中的一個(gè)基本概念。它代表了一組具有依賴關(guān)系和關(guān)聯(lián)關(guān)系的任務(wù)，指明了執(zhí)行的順序。在DAG中，任務(wù)被視為節(jié)點(diǎn)，節(jié)點(diǎn)之間的有向邊表示了它們之間的依賴關(guān)系，確保了一個(gè)任務(wù)只有在它的前驅(qū)任務(wù)成功完成后才會(huì)運(yùn)行。

例如，一個(gè)基本的有向無(wú)環(huán)圖可能會(huì)定義任務(wù)A、B、C和D，并明確指出它們的執(zhí)行順序和依賴關(guān)系。這種結(jié)構(gòu)不僅指出了哪些任務(wù)必須先于其他任務(wù)執(zhí)行，而且還指定了調(diào)度參數(shù)，比如從明天開始每5分鐘運(yùn)行一次DAG，或者從2024年1月1日開始每天運(yùn)行一次。

DAG的主要關(guān)注點(diǎn)并不是任務(wù)內(nèi)部的運(yùn)作機(jī)制，而是它們應(yīng)該如何執(zhí)行，包括執(zhí)行的順序、重試邏輯、超時(shí)以及其他操作方面的問(wèn)題。這種抽象使得創(chuàng)建復(fù)雜的工作流變得容易管理和監(jiān)控。

然而，并非所有任務(wù)都如此簡(jiǎn)單。在遇到復(fù)雜任務(wù)時(shí)，比如第一次搜索沒(méi)有找到想要的內(nèi)容，我們可能需要進(jìn)行第二次、第三次搜索，甚至可能需要調(diào)用網(wǎng)絡(luò)搜索來(lái)完成。在這種情況下，順序執(zhí)行的任務(wù)（DAG）顯然無(wú)法滿足需求。此時(shí)，請(qǐng)求方和搜索方之間需要經(jīng)歷多次來(lái)回溝通，請(qǐng)求方可能會(huì)要求搜索方根據(jù)反饋調(diào)整搜索策略，這種多次的循環(huán)溝通才能逐步逼近最終答案。

這種情況下，我們需要的不再是DAG，而是一個(gè)循環(huán)圖，它能夠描述多個(gè)參與者之間的多輪對(duì)話和互動(dòng)，以確認(rèn)最終的答案。這種循環(huán)圖能夠處理更模糊、更復(fù)雜的用例，因?yàn)樗试S系統(tǒng)根據(jù)反饋進(jìn)行調(diào)整和迭代。那么，在循環(huán)圖的運(yùn)行模式就是智能代理，也就是AI Agent。

AI Agent，即人工智能代理，是一種基于強(qiáng)化學(xué)習(xí)理論設(shè)計(jì)的系統(tǒng)。如下圖所示，強(qiáng)化學(xué)習(xí)是一種機(jī)器學(xué)習(xí)方法，它使智能體（Agent）能夠根據(jù)環(huán)境的不同狀態(tài)（State）采取行動(dòng)（Action），目的是獲取最大程度的獎(jiǎng)勵(lì)（Reward）。這種學(xué)習(xí)過(guò)程涉及到智能體與環(huán)境的不斷互動(dòng)，通過(guò)嘗試和錯(cuò)誤來(lái)學(xué)習(xí)哪種行動(dòng)策略能夠帶來(lái)最佳的結(jié)果。

例如，在微信的跳一跳游戲中，智能體每次成功跳上平臺(tái)都會(huì)獲得獎(jiǎng)勵(lì)，而未能跳上平臺(tái)則會(huì)受到懲罰。通過(guò)這種獎(jiǎng)勵(lì)和懲罰機(jī)制，智能體能夠?qū)W習(xí)如何調(diào)整跳躍的力量和方向，以獲得更高的分?jǐn)?shù)。這個(gè)過(guò)程就是強(qiáng)化學(xué)習(xí)的一個(gè)簡(jiǎn)單體現(xiàn)，智能體通過(guò)不斷的互動(dòng)來(lái)優(yōu)化其行動(dòng)策略。

在LangGraph中，AI Agent的工作原理類似。LLM（大型語(yǔ)言模型）用于確定要采取的行動(dòng)和向用戶提供的響應(yīng)，然后執(zhí)行這些行動(dòng)，并返回到第一步。這個(gè)過(guò)程會(huì)重復(fù)進(jìn)行，直到生成最終的響應(yīng)。這就是LangChain中核心AgentExecutor的工作循環(huán)原理。

然而，在實(shí)際應(yīng)用過(guò)程中，我們發(fā)現(xiàn)需要對(duì)智能代理進(jìn)行更多的控制。例如，我們可能希望智能代理始終首先調(diào)用特定工具，或者我們可能希望對(duì)工具的調(diào)用方式有更多的控制，甚至可能希望根據(jù)智能代理的狀態(tài)使用不同的提示。為了解決這些問(wèn)題，LangGraph提出了“狀態(tài)機(jī)”的概念。通過(guò)狀態(tài)機(jī)為圖創(chuàng)建對(duì)應(yīng)的狀態(tài)機(jī)，這種方法可以更好地控制智能代理的行動(dòng)流程，使其更加靈活和有效地處理復(fù)雜任務(wù)。

LangGraph的組成部分

在介紹LangGraph如何通過(guò)應(yīng)用“狀態(tài)機(jī)”來(lái)實(shí)現(xiàn)AI Agent功能時(shí)，有幾個(gè)重要的概念我們需要理解，它們也是LangGraph的重要組成部分。

StateGraph（狀態(tài)圖）

首先，需要理解StateGraph這個(gè)核心概念。StateGraph是一個(gè)類，它負(fù)責(zé)表示整個(gè)圖的結(jié)構(gòu)。我們通過(guò)傳入一個(gè)狀態(tài)定義來(lái)初始化這個(gè)類，這個(gè)狀態(tài)定義代表了一個(gè)中心狀態(tài)對(duì)象，它會(huì)在執(zhí)行過(guò)程中不斷更新。這個(gè)狀態(tài)對(duì)象由圖中的節(jié)點(diǎn)更新，節(jié)點(diǎn)會(huì)以鍵值對(duì)的形式，返回對(duì)狀態(tài)屬性的操作。

狀態(tài)對(duì)象的屬性可以通過(guò)兩種方式更新：

1. 覆蓋更新：如果一個(gè)屬性需要被新的值替換，我們可以讓節(jié)點(diǎn)返回這個(gè)新值。

2. 增量更新：如果一個(gè)屬性是一個(gè)動(dòng)作列表（或類似的操作），我們可以在原有的列表上添加新的動(dòng)作。

在創(chuàng)建狀態(tài)定義時(shí)，我們需要指定屬性的更新方式，是覆蓋還是增量。

如果StateGraph的概念不好理解，可以想象一下你正在組織一次旅行。你設(shè)定了旅行的一些基本信息，比如確定目的地、預(yù)定航班和預(yù)定酒店。這些信息就像是一個(gè)中心狀態(tài)對(duì)象，隨著你計(jì)劃的進(jìn)展，它會(huì)不斷更新。比如，你可能會(huì)添加新的活動(dòng)到你的行程中，或者修改你的預(yù)算。這些更新就像是圖中的節(jié)點(diǎn)，它們對(duì)你的旅行計(jì)劃狀態(tài)對(duì)象進(jìn)行操作。

在LangGraph中，StateGraph類就是這樣的旅行計(jì)劃，而節(jié)點(diǎn)就像是規(guī)劃旅行的不同步驟，比如確定目的地、預(yù)定航班和預(yù)定酒店。每個(gè)步驟都會(huì)更新你的旅行計(jì)劃，可能是完全替換舊的計(jì)劃，也可能是添加新的信息到現(xiàn)有的計(jì)劃中。

Nodes（節(jié)點(diǎn)）

說(shuō)完了StateGraph，我們來(lái)關(guān)注圖的節(jié)點(diǎn)部分。在創(chuàng)建了StateGraph之后，我們需要向其中添加Nodes（節(jié)點(diǎn)）。添加節(jié)點(diǎn)是通過(guò)`graph.add_node(name, value)`語(yǔ)法來(lái)完成的。

其中，`name`參數(shù)是一個(gè)字符串，用于在添加邊時(shí)引用這個(gè)節(jié)點(diǎn)。`value`參數(shù)應(yīng)該可以是函數(shù)或者LCEL（LangChain Expression Language）可運(yùn)行的實(shí)例，它們將在節(jié)點(diǎn)被調(diào)用時(shí)執(zhí)行。它們可以接受一個(gè)字典作為輸入，這個(gè)字典的格式應(yīng)該與State對(duì)象相同，在執(zhí)行完畢之后也會(huì)輸出一個(gè)字典，字典中的鍵是State對(duì)象中要更新的屬性。說(shuō)白了，Nodes（節(jié)點(diǎn)）的責(zé)任是“執(zhí)行”，在執(zhí)行完畢之后會(huì)更新StateGraph的狀態(tài)。

接著，上面旅行計(jì)劃的例子，Nodes（節(jié)點(diǎn)）就好像旅行計(jì)劃中需要完成的任務(wù)，例如：預(yù)定航班、預(yù)訂酒店。Nodes（節(jié)點(diǎn)）接受旅行計(jì)劃（State對(duì)象）作為輸入，并輸出一個(gè)更新后的任務(wù)狀態(tài)，例如：完成酒店的預(yù)訂。

換句話說(shuō)為了完成復(fù)雜任務(wù)，我們會(huì)在StateGraph中添加很多Nodes（節(jié)點(diǎn)）。每個(gè)節(jié)點(diǎn)都代表一個(gè)任務(wù)，它們執(zhí)行的結(jié)果會(huì)影響StateGraph的狀態(tài)。這些節(jié)點(diǎn)通過(guò)邊相互連接，形成了一個(gè)有向無(wú)環(huán)圖（DAG），確保了任務(wù)的正確執(zhí)行順序。

Edges（邊）

說(shuō)了StateGraph之后就不得不提到Edge（邊）了。在LangGraph中，Edges（邊）是連接Nodes（節(jié)點(diǎn)）并定義StateGraph（狀態(tài)圖）中節(jié)點(diǎn)執(zhí)行順序的關(guān)鍵部分。添加節(jié)點(diǎn)后，我們可以添加邊來(lái)構(gòu)建整個(gè)圖。邊有幾種類型：

• 起始邊（Starting Edge）：這個(gè)邊確定了圖的開始，比如在旅行計(jì)劃中，起始邊就是確定你的目的地。一旦目的地被確定，你的旅行計(jì)劃就可以開始執(zhí)行了。
• 普通邊（Normal Edges）：這些邊表示一個(gè)節(jié)點(diǎn)總是要在另一個(gè)節(jié)點(diǎn)之后被調(diào)用。在旅行計(jì)劃中，普通邊就像是確定了任務(wù)執(zhí)行的順序。例如，在找到合適的航班之后，你可能會(huì)決定預(yù)訂酒店。這個(gè)順序確保了任務(wù)的有序執(zhí)行。
• 條件邊（Conditional Edges）：使用函數(shù)（通常由LLM提供）來(lái)確定首先調(diào)用哪個(gè)節(jié)點(diǎn)。在旅行計(jì)劃中，條件邊就像是根據(jù)你的喜好或者天氣情況來(lái)決定你的下一步行動(dòng)。比如，如果你發(fā)現(xiàn)沒(méi)有合適的航班，你可能會(huì)選擇推遲預(yù)訂酒店，而去查找火車車票。條件邊提供了靈活性，使得系統(tǒng)可以根據(jù)不同的情況來(lái)調(diào)整執(zhí)行的順序。

在LangGraph中，邊（Edges）是連接節(jié)點(diǎn)（Nodes）并定義圖（Graph）中節(jié)點(diǎn)執(zhí)行順序的關(guān)鍵部分。邊可以看作是對(duì)節(jié)點(diǎn)的控制和鏈接，它們確保了圖中的任務(wù)按照預(yù)定的順序執(zhí)行。

在LangGraph中，邊定義了節(jié)點(diǎn)之間的依賴關(guān)系和執(zhí)行順序。起始邊確定了圖的開始，普通邊確保了任務(wù)的正確執(zhí)行順序，而條件邊則根據(jù)特定的條件來(lái)決定下一步的操作。

LangGraph 實(shí)戰(zhàn)

前面對(duì)LangGraph的設(shè)計(jì)原理以及基本組成部分有了簡(jiǎn)單的了解， 接下來(lái)，我們通過(guò)一個(gè)例子來(lái)感受一下，如何實(shí)際使用LangGraph查詢北京2024年春節(jié)的旅游情況。熟悉大模型的朋友可能知道，大模型的短板就是對(duì)實(shí)時(shí)的信息一無(wú)所知，如果要攝入新的知識(shí)必須經(jīng)過(guò)新數(shù)據(jù)集的訓(xùn)練才行。因此，我們會(huì)使用LangGraph調(diào)用網(wǎng)絡(luò)搜索功能獲取實(shí)時(shí)信息。

創(chuàng)建LangChain 代理

在LangChain中代理是利用大模型作為推理引擎來(lái)確定采取的行動(dòng)序列，與在鏈中硬編碼的一系列行動(dòng)不同。說(shuō)白了，就是執(zhí)行這次任務(wù)的“關(guān)鍵人物”， 他作為“查詢北京2024年春節(jié)的旅游情況”任務(wù)的總負(fù)責(zé)，最后給用戶提供結(jié)果。

# 從langchain包中導(dǎo)入hub對(duì)象，該對(duì)象用于內(nèi)容管理和檢索
from langchain import hub
# 從langchain包中導(dǎo)入創(chuàng)建OpenAI函數(shù)代理的函數(shù)
from langchain.agents import create_openai_functions_agent
# 從langchain_openai包中導(dǎo)入ChatOpenAI類，用于與OpenAI聊天模型交互
from langchain_openai.chat_models import ChatOpenAI
# 從langchain_community包中導(dǎo)入TavilySearchResults類，提供搜索功能
from langchain_community.tools.tavily_search import TavilySearchResults

# 創(chuàng)建一個(gè)工具列表，其中包含TavilySearchResults實(shí)例，限制最大搜索結(jié)果為1
tools = [TavilySearchResults(max_results=1)]
# 通過(guò)hub對(duì)象的pull方法檢索語(yǔ)句提示，langchain hub 維護(hù)了很多prompt，這些prompt 是針對(duì)不同應(yīng)用場(chǎng)景而創(chuàng)建的
# 作為用戶，你也可以在hub中上傳你構(gòu)建的prompt
prompt = hub.pull("hwchase17/openai-functions-agent")
# 打印獲取的提示
print(prompt)
# 初始化一個(gè)Large Language Model（LLM）聊天實(shí)例，使用GPT-3.5 Turbo模型，并啟用流模式和自定義API基礎(chǔ)URL
llm = ChatOpenAI(model="gpt-3.5-turbo-1106", streaming=True)
# 使用LLM，工具和提示信息創(chuàng)建OpenAI函數(shù)代理，形成一個(gè)可運(yùn)行的代理
agent_runnable = create_openai_functions_agent(llm, tools, prompt)

針對(duì)上述代碼進(jìn)行解釋如下：

1. `from langchain import hub` - 從`langchain`包中導(dǎo)入`hub`對(duì)象，langchain hub 維護(hù)了很多prompt，這些prompt 是針對(duì)不同應(yīng)用場(chǎng)景而創(chuàng)建的。作為用戶，你也可以在hub中上傳你構(gòu)建的prompt。

2. `from langchain.agents import create_openai_functions_agent` - 從`langchain`包中導(dǎo)入`create_openai_functions_agent`函數(shù)，這個(gè)函數(shù)用來(lái)創(chuàng)建基于OpenAI函數(shù)的代理(agent)。

3. `from langchain_openai.chat_models import ChatOpenAI` - 從`langchain_openai`包中導(dǎo)入`ChatOpenAI`類，這個(gè)類用于與OpenAI聊天模型進(jìn)行交互。

4. `from langchain_community.tools.tavily_search import TavilySearchResults` - 從`langchain_community`包中導(dǎo)入`TavilySearchResults`類，該類用于提供搜索結(jié)果的功能。

5. `tools = [TavilySearchResults(max_results=1)]` - 創(chuàng)建一個(gè)列表`tools`，其中包含一個(gè)`TavilySearchResults`實(shí)例，這個(gè)實(shí)例限制搜索結(jié)果的數(shù)量為最多1個(gè)。這里我們需要通過(guò)Tavily的工具實(shí)現(xiàn)互聯(lián)網(wǎng)搜索。

6. `prompt = hub.pull("hwchase17/openai-functions-agent")` - 通過(guò)`hub`對(duì)象的`pull`方法檢索一個(gè)名為`hwchase17/openai-functions-agent`的語(yǔ)句提示。

7. `print(prompt)` - 打印出獲取的提示內(nèi)容。

8. `llm = ChatOpenAI(model="gpt-3.5-turbo-1106", streaming=True)` - 初始化一個(gè)Large Language Model（LLM）的聊天實(shí)例，指定使用`gpt-3.5-turbo-1106`模型，并啟用流模式。

9. `agent_runnable = create_openai_functions_agent(llm, tools, prompt)` - 創(chuàng)建一個(gè)OpenAI函數(shù)代理，它能夠根據(jù)給定的LLM實(shí)例、工具列表以及提示信息運(yùn)行。

這段代碼的目的是初始化并配置一個(gè)與OpenAI 聊天模型進(jìn)行交互的環(huán)境。首先，它導(dǎo)入了所需的模塊和類，包括內(nèi)容管理的`hub`對(duì)象、創(chuàng)建OpenAI 函數(shù)代理的`create_openai_functions_agent`函數(shù)、與OpenAI 聊天模型交互的`ChatOpenAI`類和提供搜索功能的`TavilySearchResults`類。然后，代碼創(chuàng)建了工具列表，并從`langchain hub`中拉取了一個(gè)預(yù)先定義的提示。

這里我們將prompt 的打印結(jié)果，貼圖如下：

這個(gè)prompt來(lái)自于LangChain hub，用于發(fā)現(xiàn)、分享不同的大模型提示詞。這里我們使用的就是別人定義好的提示詞模版。它定義了用于生成代理(agent)操作的輸入變量和信息類型（messages）。下面逐個(gè)解釋顯示的`prompt`內(nèi)容：

1. `input_variables=['agent_scratchpad', 'input']` - 這里指定了輸入變量列表，包括`agent_scratchpad`和`input`。

2. `input_types` - 這是一個(gè)定義了不同輸入類型的字典，包括以下鍵和對(duì)應(yīng)的類型：

• chat_history: 表述聊天歷史記錄的類型，是一個(gè)包含各種消息類型的列表。這些消息類型可能包括AI消息、人類發(fā)送的消息、聊天消息、系統(tǒng)消息、函數(shù)消息和工具消息。
• `agent_scratchpad`: 代表代理的“草稿本”，用來(lái)記錄過(guò)程中的信息，類型與`chat_history`相同，包含多種可能的消息類型。

3. `messages` - 這是一個(gè)包含多個(gè)不同類型模板的列表：

• `SystemMessagePromptTemplate` - 包含一個(gè)`SystemMessage`類型模板，用于定義系統(tǒng)消息。這里的``You are a helpful assistant``是指示agent的角色和期望行為的提醒。
• `MessagesPlaceholder` - 是一個(gè)占位符，與 `chat_history` 變量相關(guān)聯(lián)，并且它是可選的，這意味著在輸入中不一定需要提供聊天歷史。
• `HumanMessagePromptTemplate` - 包含一個(gè)`HumanMessage`類型的模板，用來(lái)定義來(lái)自用戶的輸入。這里的模板`{input}`將會(huì)被用戶輸入的實(shí)際文本替換。
• `MessagesPlaceholder` - 另一個(gè)占位符，與 `agent_scratchpad` 變量相關(guān)聯(lián)。

它定義了代理(agent)如何處理消息和交互。當(dāng)使用代理執(zhí)行任務(wù)時(shí)，這個(gè)模板將被用來(lái)生成符合預(yù)設(shè)格式的輸入，從而促進(jìn)代理的正確響應(yīng)和操作。

而當(dāng)你使用`langchain hub.pull("hwchase17/openai-functions-agent")`從`langchain hub`中拉取這個(gè)prompt template時(shí)，你可以使用這個(gè)模板來(lái)配置你的代理(agent)，使其理解并處理定義好的輸入類型和消息格式，以充當(dāng)一個(gè)有效和有用的助手。用戶也可以根據(jù)自己的需求創(chuàng)造新的prompt并上傳至`hub`，以適應(yīng)不同的場(chǎng)景。

創(chuàng)建圖狀態(tài)

在創(chuàng)建完代理之后，接著就是定義圖狀態(tài)。傳統(tǒng)LangChain代理的狀態(tài)具有如下屬性：

• 輸入：這是代表用戶主要請(qǐng)求的輸入字符串，作為輸入傳遞。
• 聊天歷史：這是任何先前的對(duì)話消息，也作為輸入傳遞。
• 中間步驟：代理采取的行動(dòng)和相應(yīng)的觀察。每次代理迭代時(shí)都會(huì)更新這個(gè)列表。
• 代理結(jié)果：代理的響應(yīng)，可以是AgentAction或AgentFinish。當(dāng)這是AgentFinish時(shí)，AgentExecutor應(yīng)該結(jié)束，否則應(yīng)該調(diào)用請(qǐng)求的工具。

代碼如下：

from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator

# 定義一個(gè)類型字典，用于表示代理的狀態(tài)信息
class AgentState(TypedDict):
    # 輸入字符串
    input: str
    # 對(duì)話中之前消息的列表
    chat_history: list[BaseMessage]
    # 代理調(diào)用產(chǎn)生的結(jié)果，可能為None表示開始時(shí)沒(méi)有結(jié)果
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # 動(dòng)作和對(duì)應(yīng)觀察結(jié)果的列表，使用operator.add注釋說(shuō)明這些狀態(tài)應(yīng)該被添加到現(xiàn)有值上
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

上面代碼需要說(shuō)明的是：

• agent_outcome: Union[AgentAction, AgentFinish, None]：這個(gè)字段存儲(chǔ)代理在某次調(diào)用后的結(jié)果狀態(tài)。說(shuō)明我們需要關(guān)注AgentAction對(duì)象（代表代理執(zhí)行了某個(gè)動(dòng)作）和AgentFinish對(duì)象（表明代理完成了其任務(wù)），以及None（表示代理尚未開始處理任務(wù)或沒(méi)有需要返回的結(jié)果）。
• intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]：這是動(dòng)作和觀察結(jié)果組成的元組列表，通過(guò)Annotated和operator.add進(jìn)行了注釋。這表明要更新intermediate_steps這個(gè)字段時(shí)，應(yīng)該是將新的動(dòng)作和觀察結(jié)果加到列表中，而不是覆蓋現(xiàn)有的列表。operator.add用于提示這個(gè)字段的更新操作是添加性質(zhì)的。

定義節(jié)點(diǎn)

在介紹了如何創(chuàng)建圖狀態(tài)之后，接下來(lái)要關(guān)注的是如何定義節(jié)點(diǎn)。在LangGraph中，節(jié)點(diǎn)可以是函數(shù)或可運(yùn)行實(shí)體。針對(duì)我們的任務(wù)，我們需要定義兩個(gè)主要節(jié)點(diǎn)：

• 代理節(jié)點(diǎn)：負(fù)責(zé)決定要采取什么行動(dòng)。
• 調(diào)用工具的函數(shù)：如果代理決定采取行動(dòng)，那么這個(gè)節(jié)點(diǎn)將執(zhí)行該行動(dòng)。這里的行動(dòng)可以理解為通過(guò)Tavily進(jìn)行網(wǎng)絡(luò)搜索，獲取“北京2024年春節(jié)的旅游情況”。

除了定義節(jié)點(diǎn)，我們還需要定義一些邊。其中一些邊可能是有條件的。這些邊之所以有條件，是因?yàn)楦鶕?jù)節(jié)點(diǎn)的輸出，可能會(huì)采取幾條不同的路徑。而具體路徑需要在運(yùn)行節(jié)點(diǎn)時(shí)由大模型來(lái)確定。

有條件邊：確定在調(diào)用代理后，如何動(dòng)作。如果代理需要行動(dòng)，那么應(yīng)該調(diào)用工具；如果代理認(rèn)為已經(jīng)完成任務(wù)，那么它就應(yīng)該結(jié)束。

普通邊：在調(diào)用工具后，它應(yīng)該始終返回到代理，以決定接下來(lái)要做什么。

現(xiàn)在，讓我們定義節(jié)點(diǎn)，以及函數(shù)來(lái)決定采取哪種有條件邊。函數(shù)將根據(jù)代理節(jié)點(diǎn)的輸出，決定是調(diào)用工具還是結(jié)束流程。通過(guò)這種方式，可以構(gòu)建一個(gè)靈活的圖，能夠根據(jù)不同的情況選擇合適的路徑。

# 導(dǎo)入AgentFinish類，用于標(biāo)識(shí)代理執(zhí)行完成狀態(tài)
from langchain_core.agents import AgentFinish
# 導(dǎo)入ToolExecutor類，用于執(zhí)行工具和處理代理動(dòng)作
from langgraph.prebuilt.tool_executor import ToolExecutor

# 實(shí)例化ToolExecutor，它是一個(gè)輔助類，能夠基于代理動(dòng)作調(diào)用相應(yīng)的工具并返回結(jié)果
tool_executor = ToolExecutor(tools)

# 定義代理執(zhí)行邏輯的函數(shù)
def run_agent(data):
    # 調(diào)用代理的可執(zhí)行對(duì)象，傳入數(shù)據(jù)data，執(zhí)行代理邏輯并得到結(jié)果
    agent_outcome = agent_runnable.invoke(data)
    # 將代理的執(zhí)行結(jié)果封裝成字典形式返回
    return {"agent_outcome": agent_outcome}

# 定義執(zhí)行工具的函數(shù)
def execute_tools(data):
    # 從數(shù)據(jù)中獲取最新的代理執(zhí)行結(jié)果（agent_outcome）
    agent_action = data["agent_outcome"]
    # 調(diào)用工具執(zhí)行器，并基于代理動(dòng)作執(zhí)行工具，得到輸出結(jié)果
    output = tool_executor.invoke(agent_action)
    # 將執(zhí)行的工具及其輸出結(jié)果作為中間步驟打包成元組，然后封裝成字典格式返回
    return {"intermediate_steps": [(agent_action, str(output))]}

# 定義邏輯函數(shù)，用于根據(jù)數(shù)據(jù)決定流程是否繼續(xù)或結(jié)束
def should_continue(data):
    # 如果代理的執(zhí)行結(jié)果是AgentFinish，表示結(jié)果滿意，代表流程結(jié)束
    if isinstance(data["agent_outcome"], AgentFinish):
        # 返回字符串'end'，在設(shè)置流程圖時(shí)用于標(biāo)識(shí)流程的終點(diǎn)
        return "end"
    # 如果代理的執(zhí)行結(jié)果不是AgentFinish類的實(shí)例，則代表代理流程應(yīng)繼續(xù)執(zhí)行
    else:
        # 返回字符串'continue'，在設(shè)置流程圖時(shí)用于標(biāo)識(shí)流程的繼續(xù)點(diǎn)
        return "continue"

從上面的代碼可以看出，定義了三個(gè)函數(shù)。

• run_agent：用來(lái)接收數(shù)據(jù)并且執(zhí)行代理。
• execute_tools：接收數(shù)據(jù)執(zhí)行具體操作，在本例中用來(lái)通過(guò)搜索引擎搜索2024年春節(jié)與北京旅游相關(guān)的信息。同時(shí)，還通過(guò)intermediate_steps 記錄執(zhí)行的結(jié)果。由于在實(shí)際工作環(huán)境中，agent 都是發(fā)號(hào)施令的一方，而具體干活的是tools。每次tools 完成工作（本例是網(wǎng)絡(luò)搜索）之后，都會(huì)返回信息，此時(shí)agent都會(huì)對(duì)返回的信息進(jìn)行判斷，如果不滿意tools還要繼續(xù)“工作”。基于這樣的情況，每次tools 工作完畢都將結(jié)果記錄到intermediate_steps中，做為存檔或者是依據(jù)。
• should_continue：用來(lái)判斷工作流是否繼續(xù)，如果得到了結(jié)果data["agent_outcome"]，那么就結(jié)束工作，否則繼續(xù)工作，直到拿到結(jié)果。

定義工作流（圖和邊）

到這里我們有了代理、圖狀態(tài)、節(jié)點(diǎn)等信息了。為了把上述信息串聯(lián)起來(lái)，需要構(gòu)建一個(gè)基于狀態(tài)的工作流（workflow）。工作流中包含的是有條件的和線性的節(jié)點(diǎn)轉(zhuǎn)換，用于根據(jù)某些條件決定程序的下一步執(zhí)行哪個(gè)操作。這個(gè)過(guò)程模擬了一個(gè)基于事件的系統(tǒng)，其中不同的狀態(tài)（或稱為節(jié)點(diǎn)）可以根據(jù)特定邏輯進(jìn)行轉(zhuǎn)換。代碼如下：

from langgraph.graph import END, StateGraph
#通過(guò)stategraph 初始化工作流
workflow = StateGraph(AgentState)
# 定義兩個(gè)節(jié)點(diǎn)，在實(shí)際場(chǎng)景中agent下命令，action完成任務(wù)返回結(jié)果。
# 結(jié)果如果不滿意，agent 會(huì)繼續(xù)要action完成新任務(wù)。
# 看上去它們的交互式不斷循環(huán)切換的
workflow.add_node("agent", run_agent)  # 添加節(jié)點(diǎn)"agent"，其運(yùn)行函數(shù)為 run_agent
workflow.add_node("action", execute_tools)  # 添加節(jié)點(diǎn)"action"，其運(yùn)行函數(shù)為 execute_tools

# 設(shè)置入口節(jié)點(diǎn)為 `agent`
# 這意味著首先調(diào)用的節(jié)點(diǎn)是`agent`
workflow.set_entry_point("agent")

# 接下來(lái)添加有條件的邊
workflow.add_conditional_edges(
    # 定義起始節(jié)點(diǎn)，這里使用`agent`
    "agent",
    # 判斷是否繼續(xù)的函數(shù)節(jié)點(diǎn)，用于決定下一個(gè)節(jié)點(diǎn)是哪個(gè)的函數(shù)
    should_continue,
    # 定義映射字典
    # 鍵為字符串，值為其他節(jié)點(diǎn)
    # END 是結(jié)束節(jié)點(diǎn)
    # 調(diào)用`should_continue`后，輸出將根據(jù)這個(gè)映射找到匹配的鍵
    # 根據(jù)匹配結(jié)果，接下來(lái)會(huì)調(diào)用相應(yīng)的節(jié)點(diǎn)
    {
        # 如果是`continue`，則調(diào)用工具節(jié)點(diǎn)`action`
        "continue": "action",
        # 否則結(jié)束流程
        "end": END,
    },
)

# 添加普通邊從`action`到`agent`
# 這意味著在`action`被調(diào)用后，下一個(gè)調(diào)用的節(jié)點(diǎn)是`agent`
workflow.add_edge("action", "agent")

# 對(duì)工作流進(jìn)行編譯
# 編譯成 LangChain 可運(yùn)行對(duì)象
# 可以這個(gè)對(duì)象
app = workflow.compile()

代碼比較長(zhǎng)，我們對(duì)其進(jìn)行拆解并解釋如下：

1. 定義工作流：使用StateGraph創(chuàng)建了一個(gè)名為workflow的新工作流實(shí)例，這個(gè)工作流將要基于一種名為AgentState的狀態(tài)。
2. 添加節(jié)點(diǎn)：向工作流中添加了兩個(gè)節(jié)點(diǎn)（"agent"和"action"），它們分別關(guān)聯(lián)了各自的處理函數(shù)run_agent和execute_tools。
3. 設(shè)置入口點(diǎn)："agent"被設(shè)置為工作流的入口點(diǎn)，這意味著它是工作流開始執(zhí)行時(shí)第一個(gè)被調(diào)用的節(jié)點(diǎn)。
4. 添加有條件的邊：通過(guò)add_conditional_edges方法，為從"agent"節(jié)點(diǎn)出發(fā)的邊添加了條件。
5. 使用should_continue函數(shù)來(lái)評(píng)估條件，根據(jù)返回值決定下一個(gè)執(zhí)行哪個(gè)節(jié)點(diǎn)。
6. 如果should_continue返回"continue"，流程將移動(dòng)到"action"節(jié)點(diǎn)；如果返回"end"，則工作流將結(jié)束。
7. 添加普通邊：通過(guò)add_edge方法添加了一個(gè)從"action"到"agent"的單向邊。這個(gè)邊代表在"action"節(jié)點(diǎn)執(zhí)行后，下一步將回到"agent"節(jié)點(diǎn)繼續(xù)執(zhí)行。
8. 編譯工作流：最后，通過(guò)調(diào)用compile方法，將之前定義的工作流編譯成一個(gè)可執(zhí)行的對(duì)象app。

這段代碼定義了工作流，加入了節(jié)點(diǎn)，并且設(shè)計(jì)了邊，我們通過(guò)下面這張圖來(lái)理解其原理。Agent作為整個(gè)工作流進(jìn)入的節(jié)點(diǎn)，在最左側(cè)。Agent和Action組成了一個(gè)普通邊，這個(gè)很好理解，由Agent發(fā)起搜索任務(wù)，而Action是來(lái)完成這個(gè)任務(wù)的。接著，Agent 與should_continue 函數(shù)，通過(guò)設(shè)置條件創(chuàng)建了一個(gè)條件邊。Should_continue函數(shù)本身不是節(jié)點(diǎn)，只是一個(gè)判斷條件，從函數(shù)內(nèi)容中可以看出，如果Action得到結(jié)果就結(jié)束工作流（End），否則就繼續(xù)執(zhí)行搜索任務(wù)（continue）。為了便于理解，我們用棱形來(lái)表示，當(dāng)Action執(zhí)行之后通過(guò)should_continue 進(jìn)行判斷，然后選擇正確的分支。如果選擇continue這條，就意味著整個(gè)工作流形成了一個(gè)環(huán)，如果Action的工作一直無(wú)法should_cointnue ，也就是一直不能讓Agent滿意的話，就需要不斷重復(fù)執(zhí)行搜索工作，直到超過(guò)執(zhí)行次數(shù)從而退出。

執(zhí)行搜索任務(wù)

通過(guò)上面一頓操作，工作流已經(jīng)創(chuàng)建完畢，接著就可以執(zhí)行它了。上代碼：

inputs = {"input": "2024年春節(jié)北京的旅游情況如何", "chat_history": []}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

我們?cè)儐?wèn)“2024年春節(jié)北京的旅游情況如何”，通過(guò)工作流實(shí)例輸出結(jié)果如下圖所示：

第一個(gè)agent_outcome，很明顯在agent發(fā)出命令的時(shí)候，tavily搜索工具開始工作。

接著，tavily 搜索工具記錄搜索的中間步驟，包括從什么網(wǎng)站地址獲取了相關(guān)信息。此時(shí)第二個(gè)agent_outcome 的內(nèi)容標(biāo)志為AgentFinish，意思是Agent對(duì)結(jié)果是滿意的，因此可以結(jié)束任務(wù)了，同時(shí)給出了搜索的最終結(jié)果，和我們的預(yù)期保持一致。

總結(jié)

LangGraph技術(shù)在處理復(fù)雜任務(wù)方面具有明顯優(yōu)勢(shì)，通過(guò)循環(huán)圖的方式，使大模型和外部工具協(xié)同工作，實(shí)現(xiàn)多輪對(duì)話和調(diào)整，從而更好地逼近最終答案。該技術(shù)具有廣闊的應(yīng)用前景，能夠顯著提高工作效率和解決問(wèn)題的能力。不過(guò)，創(chuàng)建LangGraph應(yīng)用也需要一定的技術(shù)門檻，需要理解狀態(tài)圖、節(jié)點(diǎn)和邊的概念，并編寫相應(yīng)的代碼實(shí)現(xiàn)。未來(lái)，LangGraph有望成為處理復(fù)雜任務(wù)的重要技術(shù)手段。

作者介紹

崔皓，51CTO社區(qū)編輯，資深架構(gòu)師，擁有18年的軟件開發(fā)和架構(gòu)經(jīng)驗(yàn)，10年分布式架構(gòu)經(jīng)驗(yàn)。