1. 前言

這篇文章將從如何搭建一個帶網絡搜索功能的聊天機器人工作流，帶你初步了解 LangGraph。

?2. 前提條件

• 已搭建 Python 開發環境，使用 3.11 以上版本。

• 已熟悉 Python 基礎語法。可參考：【LLM】Python 基礎語法_llm python入門-CSDN博客文章瀏覽閱讀1k次，點贊18次，收藏28次。Python 作為最佳的機器學習和深度學習語言，如果你想入門大模型（如 GPT、BERT 等）與自然語言處理（NLP），那么 Python 一定是最佳選擇，這篇文章將帶你入門Python語法。_llm python入門https://blog.csdn.net/u013176571/article/details/144773581?spm=1001.2014.3001.5501

3. 什么是LangGraph

LangGraph 是由 LnagChain 創造者 LangChain Inc 開發的一款基于圖結構的開源框架，其專注于構建有狀態、多角色的應用程序，特別是與大型語言模型（LLMs）一起使用時，用于創建智能體和多智能體工作流。

3.1 為什么選擇LangGraph

LangGraph 作為底層基礎設施，支持 LLM 應用中的工作流（Workflow）和智能代理（Agentic System），提供持久化、流式處理和調試部署能力，助力構建更強大、更可控的 LLM 應用。其有以下三大核心優勢：

1. 持久化（Persistence）??

• 記憶能力（Memory）：支持存儲應用狀態，實現對話記憶和跨用戶交互的狀態管理。

• 人機協作（Human-in-the-loop）：狀態檢查點化，支持執行中斷與恢復，便于人工決策、驗證和修正。 ?

2. 流式處理（Streaming）??

• 支持工作流/代理狀態的流式傳輸，使用戶或開發者能在執行過程中獲取實時反饋。 ?

• 允許流式傳輸事件（如工具調用反饋）和 LLM 生成的 tokens，以提升交互體驗。 ?

3. 調試與部署（Debugging and Deployment）

• 通過 LangGraph Platform 提供簡便的測試、調試和部署能力。 ?

• 通過 LangGraph Studio 提供可視化工作流，支持交互式調試。

• 提供多種部署選項，方便應用上線。 ?

3.2?和LangChain的關系

LangGraph 是 LangChain 生態中的一個子項目，專門用于構建圖（Graph）結構的 LLM 工作流。它可以被認為是 LangChain 的補充，用于管理復雜的流程控制。它可以獨立于LangChain使用。

3.3?LangGraph核心概念

LangGraph 框架中，會將智能體工作流（agent workflows）建模為圖（Graphs）,構建和運行圖（Graphs）有三個核心概念：

1. State（狀態）：

• 通常為 TypedDict（字典）或 Pydantic BaseModel（對象）， 用于存儲圖運行時的狀態信息。它在圖的執行過程中被傳遞給各個節點和條件函數，用于存儲和傳遞數據。你可以將 State 理解為各節點和條件函數間的上下文。

2. Nodes（節點）：

• 是一個 Python 函數，用來執行特定的操作或邏輯（比如數據處理、調用外部服務等）。節點接收當前的 State 作為入參，在執行一系列業務邏輯后，并返回更新后的 State。

3. Edges（邊）：

• 也是一個 Python 函數，用于定義節點之間的連接關系，控制圖的執行流程。邊可以是無條件的，也可以是條件的。

簡而言之，節點用于執行工作，邊指示下一步做什么。

4. 聊天機器人實現

4.1 安裝LangGraph

這個例子中需要使用 LangChain 構建與大模型對話的 ChatModel，所以我們需要安裝 LangChain 相關包。我使用的是 0.3.0 版本。

# 安裝 LangChain 和 LangChain OpenAI
pip install langchain==0.3.0?
pip install -U langchain-openai# 安裝最新版 LangGraph
pip install -U langgraph# 安裝tavily，tavily是一個網絡檢索工具?
pip install -U tavily-python

4.2 準備工作

4.2.1 選擇模型

選擇一個支持函數調（Function Calling）用的模型，并創建一個 ChatModel，我使用的是月之暗面的 moonshot-v1-8k 模型。

from langchain_openai import ChatOpenAI# 構建一個chatModel
llm = ChatOpenAI(model="moonshot-v1-8k",openai_api_base="https://api.moonshot.cn/v1",openai_api_key="your_api_key",temperature=0.7,streaming=True,
)

4.2.2 申請tavily秘鑰

tavily 是一個網絡檢索工具，我們需要在tavily官網申請一個秘鑰來使用它（tavily 每月免費提供1000條搜索返回）。地址：Tavily AI。

定義 Tavily Search 工具

tavily_tool = TavilySearchResults(# 返回的最大搜索條數max_results=2,tavily_api_key="your_tavily_api_key",
)

4.3 代碼實現

4.3.1 流程圖

工作流主要包含兩個節點和兩個條件邊：

• chatbot：聊天節點，負責與 LLM 交互。

• tools：工具節點，負責調用 Tavily 搜索工具。

• edge1：條件邊（使用虛線表示），LLM會根據用戶輸入的內容自動判斷是否調用tools節點，如果不需要調用會直接返回。

• edge2：普通邊，tools節點完成搜索任務后，返回chatbot節點繼續對話。

4.3.2?完整代碼

以下是整個帶網絡搜索功能的聊天機器人的完整代碼

from typing import Annotatedfrom langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from typing_extensions import TypedDict# 定義一個狀態類型，用于保存工作流中的上下文信息，為 TypedDict 類型
class State(TypedDict):# messages 用于記錄與AI對話的記錄，為 list 類型# add_messages 用于定義這個值的更新方式，即 將對話消息追加到列表中，而不是覆蓋messages: Annotated[list, add_messages]# 定義一個工作流，使用 State 類型作為輸入和輸出
working_flow = StateGraph(State)# 定義 TavilySearchResults 工具
tavily_tool = TavilySearchResults(max_results=2,tavily_api_key="",
)
# 定義 ChatOpenAI 模型，并綁定工具
llm = ChatOpenAI(model="moonshot-v1-8k",openai_api_base="https://api.moonshot.cn/v1",openai_api_key="",temperature=0.7,streaming=True,
)
# 綁定工具
tools = [tavily_tool]
llm_with_tools = llm.bind_tools(tools)# 聊天節點邏輯
def chatbot(state: State):# 調用大模型，LLM 會自動決定是否調用Tavily工具# 返回值是更新后的聊天記錄return {"messages": [llm_with_tools.invoke(state["messages"])]}# 添加節點
working_flow.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=tools)
working_flow.add_node("tools", tool_node)# 添加條件邊和普通邊
working_flow.add_conditional_edges("chatbot", tools_condition, )
working_flow.add_edge("tools", "chatbot")# 設置起點
working_flow.set_entry_point("chatbot")# 創建MemorySaver,用于存儲工作流的對話歷史
memory = MemorySaver()
# 編譯工作流
graph = working_flow.compile(checkpointer=memory)

4.3.3 運行測試

以下代碼用于使用工作流并進行問答：

# 配置參數，包括唯一的線程ID，用于記錄對話歷史
config = {"configurable": {"thread_id": "1"}}
def stream_graph_updates(user_input: str):# 流式處理圖的事件for event in graph.stream({"messages": [{"role": "user", "content": user_input}]},config,):for value in event.values():# 打印助手的回復print("Assistant:", value["messages"][-1].content)while True:try:# 獲取用戶輸入user_input = input("User: ")# 檢查退出條件if user_input.lower() in ["quit", "exit", "q"]:print("再見!")break# 調用函數處理用戶輸入stream_graph_updates(user_input)except KeyboardInterrupt:# 處理用戶中斷輸入的情況print("\n輸入被中斷，再見!")breakexcept Exception as e:# 捕獲其他異常并打印錯誤信息print(f"發生錯誤: {e}")break

首先我們先問幾個簡單的問題，比如“你好”、“你是誰”，這些問題不足以讓大模型調用Tavily工具。

然后我們嘗試問大模型依據自身無法回答的問題，比如：”南京今天天氣如何？“，我們發現大模型自動調用了Tavily工具，接著我們繼續問“明天呢”，看看工作流是否具備聊天記憶。

到此為止，一個簡單的聊天機器人就完成了。當然 LangGraph 還有很多高級功能，比如人機交互（Humman-in-the-loop）、時間旅行（Time-travel）等，我在之后的文章中會介紹。

5. 參考文檔

• LangGraph Home