用LangGraph打造高可用生產(chǎn)級(jí)AI Agent 原創(chuàng)
人工智能（AI）已經(jīng)不再只是個(gè)時(shí)髦詞，它正在改變我們解決實(shí)際問題的方式。從聊天機(jī)器人到自動(dòng)化工作流，AI 智能體是這些創(chuàng)新的核心。但要打造一個(gè)可靠、可擴(kuò)展、隨時(shí)能上線的 AI 智能體可不是件容易事。這時(shí)候，LangGraph 登場了！它是 LangChain 的一個(gè)強(qiáng)大框架，能幫你輕松構(gòu)建復(fù)雜、狀態(tài)化的 AI 智能體，處理各種棘手的任務(wù)。這篇文章咱們就來聊聊如何用 LangGraph 打造生產(chǎn)就緒的 AI 智能體，還會(huì)通過一個(gè)實(shí)際案例，配上代碼、工作流和圖表，把一切講得明明白白。
什么是 LangGraph？
LangGraph 是 LangChain 框架的擴(kuò)展，專門用來創(chuàng)建狀態(tài)化、多角色的 AI 智能體，支持循環(huán)工作流。跟傳統(tǒng)的線性工作流（LangChain 里的“鏈”）不同，LangGraph 允許你構(gòu)建圖結(jié)構(gòu)，讓任務(wù)可以循環(huán)、分支，動(dòng)態(tài)適應(yīng)各種條件。這讓它特別適合處理現(xiàn)實(shí)世界中那些復(fù)雜、需要靈活性的任務(wù)。
LangGraph 的核心功能
- 狀態(tài)化工作流：每一步自動(dòng)保存狀態(tài)，智能體可以暫停、恢復(fù)，或者從錯(cuò)誤中恢復(fù)。
- 人工介入（Human-in-the-Loop）：支持人類介入審批或編輯操作，確保關(guān)鍵任務(wù)有人把關(guān)。
- 循環(huán)圖：支持循環(huán)和條件邏輯，實(shí)現(xiàn)動(dòng)態(tài)決策。
- 與 LangChain 無縫集成：可以跟 LangChain 的工具和 LangSmith 配合，方便調(diào)試和監(jiān)控。
- 可擴(kuò)展性：專為生產(chǎn)環(huán)境設(shè)計(jì)，支持持久執(zhí)行和錯(cuò)誤處理。
你可以把 LangGraph 想象成 AI 智能體的 GPS——它不只是沿著直路走，還能繞道、回頭、根據(jù)“路況”（用戶需求）靈活調(diào)整。
為什么用 LangGraph 打造生產(chǎn)就緒的 AI 智能體？
弄個(gè)能在 demo 里跑的 AI 智能體是一回事，但讓它在生產(chǎn)環(huán)境里穩(wěn)如老狗又是另一回事。生產(chǎn)系統(tǒng)得面對(duì)這些挑戰(zhàn)：
- 雜亂輸入：用戶給的信息不一定清楚或完整。
- 邊緣情況：意想不到的場景可能讓系統(tǒng)崩掉。
- 擴(kuò)展性：得能處理成千上萬的請(qǐng)求還不宕機(jī)。
- 調(diào)試：出錯(cuò)時(shí)得知道問題出在哪兒。
LangGraph 幫你解決這些問題，提供了：
- 細(xì)粒度控制：把任務(wù)拆成小塊（節(jié)點(diǎn)），通過清晰的連接（邊）管理。
- 狀態(tài)管理：在交互中保持上下文，智能體能“記住”正在發(fā)生的事。
- 錯(cuò)誤恢復(fù)：失敗后能從斷點(diǎn)繼續(xù)。
- 可觀察性：搭配 LangSmith，追蹤執(zhí)行過程，輕松調(diào)試。
實(shí)際案例：旅行規(guī)劃助手
咱們來看個(gè)實(shí)際例子：一個(gè)旅行規(guī)劃助手，幫用戶規(guī)劃行程，實(shí)時(shí)獲取航班和酒店信息，展示在網(wǎng)頁上，還能根據(jù)用戶需求通過郵件發(fā)送行程。這是個(gè)很適合用 LangGraph 的場景，因?yàn)樗婕埃?/p>
- 多步驟操作（搜索航班、酒店，發(fā)送郵件）。
- 條件邏輯（比如只有用戶要求時(shí)才發(fā)送郵件）。
- 外部工具集成（航班和酒店的 API）。
- 人工介入（讓用戶確認(rèn)行程）。
工作原理
用戶輸入需求（比如“下周末從紐約去巴黎的行程”），助手會(huì)：
- 處理請(qǐng)求，提取關(guān)鍵信息（目的地、日期等）。
- 調(diào)用 API 獲取航班和酒店信息。
- 把結(jié)果整理成用戶友好的格式。
- 在網(wǎng)頁上展示行程，詢問用戶是否需要郵件發(fā)送。
- 如果用戶要求，就通過郵件發(fā)送行程。
LangGraph 的核心組件
在看代碼之前，先搞懂 LangGraph 的幾個(gè)核心部分：
- 節(jié)點(diǎn)（Nodes）：單個(gè)任務(wù)或功能（比如調(diào)用 API、處理用戶輸入）。
- 邊（Edges）：節(jié)點(diǎn)之間的連接，定義工作流的走向。邊可以是直接的（總?cè)ハ乱粋€(gè)節(jié)點(diǎn)）或條件的（根據(jù)邏輯選擇下一個(gè)節(jié)點(diǎn)）。
- 狀態(tài)（State）：一個(gè)共享的數(shù)據(jù)結(jié)構(gòu)，保存智能體的當(dāng)前狀態(tài)，比如用戶輸入、API 結(jié)果或?qū)υ挌v史。
- 圖（Graph）：把節(jié)點(diǎn)和邊連起來的整體結(jié)構(gòu)。
簡單打個(gè)比方：節(jié)點(diǎn)就像工廠里的工人，邊是傳送帶，狀態(tài)是記錄工作進(jìn)度的剪貼板。
一步步實(shí)現(xiàn)
咱們用 Python 和 LangGraph 來實(shí)現(xiàn)這個(gè)旅行規(guī)劃助手。假設(shè)你對(duì) Python 有一定了解，也知道 LangChain 的基本用法。
第一步：配置環(huán)境
先安裝所需庫：
pip install langgraph langchain langchain-openai requests設(shè)置 OpenAI（用于 LLM）和旅行 API（比如 Google Flights、Hotels）的 API 密鑰，安全存儲(chǔ)在環(huán)境變量中。
第二步：定義狀態(tài)
狀態(tài)是個(gè)共享數(shù)據(jù)結(jié)構(gòu)，保存智能體運(yùn)行所需的所有信息，比如用戶輸入、API 響應(yīng)和生成的行程。我們用 ??typing_extensions??? 的 ??TypedDict?? 定義一個(gè)清晰、類型安全的結(jié)構(gòu)。
from typing_extensions import TypedDict
class AgentState(TypedDict):
user_request: str
flight_options: list
hotel_options: list
itinerary: str
send_email: bool第三步：創(chuàng)建節(jié)點(diǎn)
每個(gè)節(jié)點(diǎn)執(zhí)行一個(gè)特定任務(wù)。以下是我們助手的節(jié)點(diǎn)：
- 處理請(qǐng)求：從用戶輸入中提取關(guān)鍵信息。
- 獲取航班：調(diào)用航班 API。
- 獲取酒店：調(diào)用酒店 API。
- 整理行程：格式化結(jié)果。
- 詢問郵件：詢問用戶是否需要郵件發(fā)送行程。
- 發(fā)送郵件：通過郵件發(fā)送行程。
節(jié)點(diǎn)代碼如下：
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
import requests
from typing importDict
llm = ChatOpenAI(model="gpt-4", api_key="your-openai-api-key")
defprocess_request(state: AgentState) -> Dict:
prompt = PromptTemplate(
input_variables=["user_request"],
template="從以下內(nèi)容提取目的地、出發(fā)地和日期：{user_request}"
)
response = llm.invoke(prompt.format(user_request=state["user_request"]))
# 假設(shè)響應(yīng)包含解析后的數(shù)據(jù)（為簡化示例）
return {"user_request": response.content}
deffetch_flights(state: AgentState) -> Dict:
# 模擬 API 調(diào)用（用真實(shí)的 Google Flights API 替換）
flight_data = [{"airline": "Air France", "price": "$500", "time": "10:00 AM"}]
return {"flight_options": flight_data}
deffetch_hotels(state: AgentState) -> Dict:
# 模擬 API 調(diào)用（用真實(shí)的 Google Hotels API 替換）
hotel_data = [{"hotel": "Paris Inn", "price": "$150/night"}]
return {"hotel_options": hotel_data}
defcompile_itinerary(state: AgentState) -> Dict:
itinerary = f"航班信息：\n{state['flight_options']}\n酒店信息：\n{state['hotel_options']}"
return {"itinerary": itinerary}
defask_email(state: AgentState) -> str:
# 模擬人工介入（生產(chǎn)環(huán)境中用 UI 或輸入框）
return"send_email"ifinput("是否通過郵件發(fā)送行程？(y/n): ") == "y"else"end"
defsend_email(state: AgentState) -> Dict:
# 模擬郵件發(fā)送（用 SendGrid 或類似服務(wù)替換）
print(f"正在發(fā)送行程郵件：{state['itinerary']}")
return {"send_email": True}第四步：構(gòu)建工作流
現(xiàn)在定義圖，連接節(jié)點(diǎn)和邊。我們用條件邊來決定是發(fā)送郵件還是結(jié)束工作流。
from langgraph.graph import StateGraph, START, END
workflow = StateGraph(AgentState)
# 添加節(jié)點(diǎn)
workflow.add_node("process_request", process_request)
workflow.add_node("fetch_flights", fetch_flights)
workflow.add_node("fetch_hotels", fetch_hotels)
workflow.add_node("compile_itinerary", compile_itinerary)
workflow.add_node("ask_email", ask_email)
workflow.add_node("send_email", send_email)
# 添加邊
workflow.add_edge(START, "process_request")
workflow.add_edge("process_request", "fetch_flights")
workflow.add_edge("fetch_flights", "fetch_hotels")
workflow.add_edge("fetch_hotels", "compile_itinerary")
workflow.add_edge("compile_itinerary", "ask_email")
workflow.add_conditional_edges(
"ask_email",
lambda state: state.get("send_email", "end"),
{"send_email": "send_email", "end": END}
)
workflow.add_edge("send_email", END)
# 編譯圖
agent = workflow.compile()第五步：運(yùn)行智能體
用一個(gè)示例輸入測(cè)試智能體：
initial_state = {"user_request": "計(jì)劃下周末從紐約去巴黎的行程"}
result = agent.invoke(initial_state)
print("最終行程：")
print(result["itinerary"])示例輸出
假設(shè)用戶在詢問郵件時(shí)輸入“y”：
最終行程：
航班信息：
[{'airline': 'Air France', 'price': '$500', 'time': '10:00 AM'}]
酒店信息：
[{'hotel': 'Paris Inn', 'price': '$150/night'}]
正在發(fā)送行程郵件：航班信息：[{'airline': 'Air France', 'price': '$500', 'time': '10:00 AM'}] 酒店信息：[{'hotel': 'Paris Inn', 'price': '$150/night'}]工作流圖
工作流的可視化如下：
（此處為文字描述，實(shí)際生產(chǎn)中會(huì)有圖表展示從處理用戶請(qǐng)求到發(fā)送郵件或結(jié)束工作流的流程，基于用戶選擇）
為什么這個(gè)智能體是生產(chǎn)就緒的？
這個(gè)旅行規(guī)劃助手之所以能達(dá)到生產(chǎn)就緒的標(biāo)準(zhǔn)，是因?yàn)椋?/p>
- 錯(cuò)誤處理：LangGraph 的狀態(tài)管理確保工作流在失敗后能恢復(fù)。
- 支持人工介入：詢問郵件的節(jié)點(diǎn)允許用戶干預(yù)。
- 模塊化：每個(gè)節(jié)點(diǎn)都是小而專注的任務(wù)，方便調(diào)試或替換組件（比如換用其他 API）。
- 可擴(kuò)展：LangGraph 的基礎(chǔ)設(shè)施支持通過 LangGraph Platform 部署，處理大量用戶。
你可以進(jìn)一步優(yōu)化，比如：
- 用 LangSmith 進(jìn)行調(diào)試和監(jiān)控。
- 通過 LangGraph Platform 部署以提高擴(kuò)展性。
- 集成真實(shí) API（比如 SendGrid 發(fā)郵件，Google Flights 獲取實(shí)時(shí)數(shù)據(jù)）。
其他現(xiàn)實(shí)世界的應(yīng)用場景
LangGraph 在很多場景都能大放異彩：
- 客服智能體：解答問題，檢索知識(shí)庫，必要時(shí)轉(zhuǎn)交人工。
- 研究助手：搜索網(wǎng)絡(luò)，總結(jié)信息，生成報(bào)告。
- 財(cái)務(wù)顧問：分析市場數(shù)據(jù)，推薦投資，監(jiān)控投資組合。
比如，LinkedIn 用 LangGraph 驅(qū)動(dòng)內(nèi)部 SQL Bot，把自然語言查詢翻譯成 SQL，大幅提升團(tuán)隊(duì)效率。
打造生產(chǎn)就緒智能體的建議
- 從簡單開始：先搞個(gè)基礎(chǔ)工作流，再逐步加復(fù)雜功能。
- 用條件邏輯：利用條件邊實(shí)現(xiàn)動(dòng)態(tài)決策。
- 測(cè)試邊緣情況：模擬雜亂輸入和失敗場景，確保系統(tǒng)穩(wěn)健。
- 用 LangSmith 監(jiān)控：追蹤 token 使用、錯(cuò)誤和性能。
- 引入人工介入：對(duì)敏感任務(wù)使用 human-in-the-loop，避免出錯(cuò)。
總結(jié)
LangGraph 讓打造超越 demo 的可靠 AI 智能體變得簡單。通過把任務(wù)拆成小步驟（節(jié)點(diǎn)），用清晰的路徑（邊）連接，跟蹤信息（狀態(tài)），LangGraph 讓復(fù)雜工作流變得易于管理。我們的旅行規(guī)劃助手展示了如何把這些理念用在實(shí)際場景中，配上代碼、結(jié)果和清晰的工作流圖。
鏈接
- Scribd: Building Production-Ready AI Agents with LangGraph (https://www.scribd.com/document/887012226/Building-Production-Ready-AI-Agents-with-LangGraph)
- Slideshare: Building Production-Ready AI Agents with LangGraph (https://www.slideshare.net/slideshow/building-production-ready-ai-agents-with-langgraph-pdf/281497850)
本文轉(zhuǎn)載自??PyTorch研習(xí)社??，作者：AI研究生
