1. 項(xiàng)目概述從 LangChain 到 LangGraphAgent 的范式躍遷如果你已經(jīng)用 LangChain 搭建過一些簡單的 AI 應(yīng)用比如一個(gè)能聯(lián)網(wǎng)搜索的問答機(jī)器人那你大概率已經(jīng)接觸過“Agent”這個(gè)概念了。在 LangChain 的早期版本里Agent 通常被理解為一個(gè)“工具調(diào)用者”你給它一個(gè)目標(biāo)比如“查一下今天上海的天氣”它就會(huì)規(guī)劃步驟“我需要調(diào)用天氣查詢工具”然后執(zhí)行調(diào)用 API最后給你結(jié)果。這個(gè)過程是線性的、一次性的。但當(dāng)我們開始構(gòu)建更復(fù)雜的應(yīng)用時(shí)比如一個(gè)能持續(xù)與用戶對(duì)話、管理長期任務(wù)、在多個(gè)工具間靈活跳轉(zhuǎn)的智能客服或者一個(gè)能自主分析數(shù)據(jù)、撰寫報(bào)告并發(fā)送郵件的自動(dòng)化助手這種簡單的“規(guī)劃-執(zhí)行”模式就顯得力不從心了。這時(shí)LangGraph就登場了。它不是要取代 LangChain而是 LangChain 生態(tài)系統(tǒng)中的一個(gè)專門用于構(gòu)建有狀態(tài)、多步驟、可循環(huán)的復(fù)雜 Agent 的框架。你可以把它想象成從“單次函數(shù)調(diào)用”升級(jí)到了“一個(gè)完整的應(yīng)用程序”。它的核心思想就是引入了計(jì)算機(jī)科學(xué)中一個(gè)經(jīng)典且強(qiáng)大的概念——狀態(tài)機(jī)。為什么狀態(tài)機(jī)如此重要因?yàn)楝F(xiàn)實(shí)世界中的任務(wù)很少是“一錘子買賣”。一個(gè)客服對(duì)話有多個(gè)回合每個(gè)回合的上下文用戶歷史、已查詢信息、用戶情緒都在變化一個(gè)數(shù)據(jù)分析任務(wù)可能需要先清洗數(shù)據(jù)再分析發(fā)現(xiàn)異常后再回頭重新清洗最后生成可視化圖表。這些流程都擁有明確的“狀態(tài)”比如“等待用戶輸入”、“數(shù)據(jù)清洗中”、“生成報(bào)告中”和狀態(tài)之間的“轉(zhuǎn)移條件”比如“用戶提問”觸發(fā)從“等待”到“處理”“清洗完成”觸發(fā)到“分析”。用狀態(tài)機(jī)來建模這些流程邏輯會(huì)變得異常清晰和健壯。所以當(dāng)你看到“高級(jí) AgentLangGraph 與狀態(tài)機(jī)”這個(gè)標(biāo)題時(shí)它指向的正是 AI 應(yīng)用開發(fā)的下一個(gè)階段如何構(gòu)建那些真正具備復(fù)雜邏輯、能夠處理非線性工作流、并且能維持長期記憶和上下文的智能體。這不再是玩具 demo而是邁向生產(chǎn)級(jí) AI 應(yīng)用的關(guān)鍵一步。接下來我將以一個(gè)“智能研究助手”Agent 為例帶你徹底拆解 LangGraph 的核心三要素并手把手實(shí)現(xiàn)一個(gè)具備循環(huán)、分支和人工審核能力的復(fù)雜狀態(tài)機(jī)。2. 核心三要素拆解State、Node、Edge要理解 LangGraph必須吃透它的三個(gè)核心抽象State、Node和Edge。這就像建房子的地基、磚塊和鋼筋三者結(jié)合才能構(gòu)筑起穩(wěn)固的架構(gòu)。2.1 State智能體的記憶與上下文在 LangGraph 中State 是一個(gè)字典它定義了整個(gè)工作流運(yùn)行過程中需要攜帶和更新的所有信息。你可以把它理解為 Agent 的“工作內(nèi)存”或“上下文白板”。與 LangChain 中每次調(diào)用都相對(duì)獨(dú)立的鏈不同LangGraph 的 State 會(huì)在整個(gè)圖執(zhí)行過程中持續(xù)存在并被修改。State 的定義與注解State 通常使用 Pydantic 的BaseModel來定義這能提供清晰的類型提示和驗(yàn)證。對(duì)于我們的研究助手State 可能包含from typing import List, Dict, Any, Optional, Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages import operator # 方式一使用 TypedDict更靈活兼容性好 class AgentState(TypedDict): # 對(duì)話消息歷史LangGraph 提供了專用注解來簡化消息列表的合并操作 messages: Annotated[List[Dict], add_messages] # 用戶輸入的研究主題 research_topic: str # 從網(wǎng)絡(luò)上搜集到的原始資料列表 gathered_sources: List[Dict[str, Any]] # 分析后的關(guān)鍵發(fā)現(xiàn) key_findings: List[str] # 生成的報(bào)告草稿 report_draft: Optional[str] # 一個(gè)控制流程的標(biāo)志位例如“是否需要人工審核” needs_human_review: bool # 人工審核的反饋意見 human_feedback: Optional[str]這里有幾個(gè)關(guān)鍵點(diǎn)Annotated[List[Dict], add_messages]這是 LangGraph 的一個(gè)“魔法”。add_messages是一個(gè)歸約器它定義了當(dāng)多個(gè)節(jié)點(diǎn)同時(shí)向state[‘messages’]字段寫入時(shí)如何合并這些值。對(duì)于消息列表最常見的操作就是追加。這確保了對(duì)話歷史能正確累積而不會(huì)被覆蓋。狀態(tài)即數(shù)據(jù)流圖中的每個(gè)節(jié)點(diǎn)都讀取和修改這個(gè)共享的 State。例如“搜索節(jié)點(diǎn)”會(huì)向gathered_sources添加數(shù)據(jù)“分析節(jié)點(diǎn)”會(huì)讀取gathered_sources并生成key_findings。設(shè)計(jì)原則State 應(yīng)該包含所有必要的上下文但也要保持精簡。避免將中間計(jì)算過程等臨時(shí)變量塞進(jìn)去專注于輸入、輸出和控制流數(shù)據(jù)。2.2 Node執(zhí)行具體任務(wù)的函數(shù)Node 是圖中的節(jié)點(diǎn)每個(gè)節(jié)點(diǎn)都是一個(gè)普通的 Python 函數(shù)或可調(diào)用對(duì)象。這個(gè)函數(shù)接收當(dāng)前的State作為參數(shù)執(zhí)行一些操作調(diào)用 LLM、使用工具、處理數(shù)據(jù)然后返回一個(gè)包含對(duì) State更新內(nèi)容的字典。節(jié)點(diǎn)的編寫范式一個(gè)典型的節(jié)點(diǎn)函數(shù)看起來是這樣的def search_node(state: AgentState) - Dict[str, Any]: 負(fù)責(zé)根據(jù)主題進(jìn)行網(wǎng)絡(luò)搜索的節(jié)點(diǎn)。 print(f“[搜索節(jié)點(diǎn)] 正在搜索主題{state[‘research_topic’]}”) # 1. 準(zhǔn)備搜索查詢這里可以加入查詢優(yōu)化邏輯 search_query f“{state[‘research_topic’]} latest research 2024” # 2. 調(diào)用搜索工具例如 Tavily Search API、Serper API 或 DuckDuckGo # 假設(shè)我們有一個(gè) search_web 函數(shù) search_results search_web(search_query, max_results5) # 3. 對(duì)結(jié)果進(jìn)行初步處理提取標(biāo)題、鏈接、摘要 processed_sources [] for result in search_results: processed_sources.append({ “title”: result.get(“title”), “url”: result.get(“url”), “snippet”: result.get(“snippet”)[:200] “...” # 截?cái)嗾?}) # 4. 返回要更新的 State 部分 # 注意我們返回的是 gathered_sources而不是整個(gè) state。 # LangGraph 會(huì)自動(dòng)將這個(gè)字典與當(dāng)前 state 合并。 return {“gathered_sources”: processed_sources}關(guān)鍵理解節(jié)點(diǎn)函數(shù)不直接修改傳入的state對(duì)象。它只是基于state進(jìn)行計(jì)算然后返回一個(gè)字典指明要更新哪些字段。返回的字典中的鍵必須與 State 中定義的字段名對(duì)應(yīng)。一個(gè)節(jié)點(diǎn)可以很復(fù)雜比如內(nèi)部封裝了一個(gè) LangChain Chain也可以很簡單只是一個(gè)邏輯判斷。2.3 Edge決定流程走向的規(guī)則Edge 定義了圖中節(jié)點(diǎn)之間的連接關(guān)系更重要的是它決定了在某個(gè)節(jié)點(diǎn)執(zhí)行完畢后下一個(gè)該執(zhí)行哪個(gè)節(jié)點(diǎn)。這是狀態(tài)機(jī)邏輯的核心。Edge 分為兩種普通邊直接從一個(gè)節(jié)點(diǎn)連接到另一個(gè)節(jié)點(diǎn)無條件執(zhí)行。條件邊根據(jù) State 中的某個(gè)條件動(dòng)態(tài)決定下一個(gè)節(jié)點(diǎn)。條件邊通過一個(gè)特殊的conditional_edge來創(chuàng)建它連接到一個(gè)“路由函數(shù)”。這個(gè)路由函數(shù)檢查 State并返回下一個(gè)要執(zhí)行的節(jié)點(diǎn)的名稱字符串。條件邊的實(shí)戰(zhàn)在我們的研究助手中在“生成報(bào)告”節(jié)點(diǎn)之后我們可能希望引入一個(gè)人工審核環(huán)節(jié)。但并非所有報(bào)告都需要審核只有當(dāng)內(nèi)容敏感或置信度低時(shí)才需要。我們可以這樣設(shè)計(jì)from langgraph.graph import END, START def should_review(state: AgentState) - str: 路由函數(shù)決定下一步是人工審核還是直接結(jié)束。 # 這里可以設(shè)計(jì)更復(fù)雜的邏輯例如 # - 檢查報(bào)告是否涉及特定關(guān)鍵詞如“醫(yī)療建議”、“投資” # - 調(diào)用一個(gè) LLM 來判斷報(bào)告內(nèi)容的置信度 # - 基于 state[‘needs_human_review’] 標(biāo)志位可能由之前的節(jié)點(diǎn)設(shè)置 sensitive_keywords [“醫(yī)療”, “法律”, “財(cái)務(wù)建議”, “投資”] report_text state.get(“report_draft”, “”).lower() if any(keyword in report_text for keyword in sensitive_keywords): return “human_review_node” # 前往人工審核節(jié)點(diǎn) elif state.get(“needs_human_review”, False): return “human_review_node” else: return END # 直接結(jié)束整個(gè)圖的工作流然后在構(gòu)建圖時(shí)你會(huì)將“生成報(bào)告節(jié)點(diǎn)”通過一個(gè)conditional_edge連接到這個(gè)路由函數(shù)而路由函數(shù)的結(jié)果“human_review_node”或END決定了真正的流向。注意START和END是 LangGraph 中兩個(gè)特殊的節(jié)點(diǎn)名分別代表圖的入口和出口。把 State、Node、Edge 組合起來你就得到了一個(gè)完整的、可定義復(fù)雜業(yè)務(wù)邏輯的“藍(lán)圖”。接下來我們就用這個(gè)藍(lán)圖搭建一個(gè)真實(shí)的研究助手。3. 構(gòu)建智能研究助手從藍(lán)圖到代碼讓我們實(shí)現(xiàn)一個(gè)具備完整流程的研究助手 Agent。它的工作流是接收主題 - 并行搜索與讀取本地文檔 - 綜合分析 - 生成報(bào)告 - 條件性人工審核 - 最終輸出。3.1 定義完整狀態(tài)與工具首先我們完善 State 并準(zhǔn)備一些工具函數(shù)。from typing import List, Dict, Any, Optional, Annotated, Literal from typing_extensions import TypedDict from langgraph.graph.message import add_messages import asyncio # 假設(shè)的工具函數(shù)和模型調(diào)用 from some_tool_module import tavily_search, read_pdf_text from some_llm_module import call_llm class ResearchAgentState(TypedDict): 研究助手 Agent 的完整狀態(tài)定義。 messages: Annotated[List[Dict], add_messages] research_topic: str # 來源分為網(wǎng)絡(luò)和本地 web_sources: List[Dict[str, Any]] local_sources: List[Dict[str, Any]] # 綜合后的資料 synthesized_info: Optional[str] key_findings: List[str] report_draft: Optional[str] needs_human_review: bool human_feedback: Optional[str] # 一個(gè)標(biāo)志記錄當(dāng)前流程步驟可用于調(diào)試或復(fù)雜路由 current_step: Literal[“start”, “gathering”, “analyzing”, “reporting”, “reviewing”, “end”]3.2 實(shí)現(xiàn)各個(gè)功能節(jié)點(diǎn)我們將創(chuàng)建多個(gè)節(jié)點(diǎn)每個(gè)負(fù)責(zé)一項(xiàng)具體工作。節(jié)點(diǎn)1任務(wù)規(guī)劃與初始化這個(gè)節(jié)點(diǎn)負(fù)責(zé)解析用戶輸入明確研究主題并初始化狀態(tài)。def planning_node(state: ResearchAgentState) - Dict[str, Any]: 規(guī)劃節(jié)點(diǎn)解析輸入初始化任務(wù)。 # 通常最后一條用戶消息是輸入 user_input state[“messages”][-1][“content”] if state[“messages”] else state.get(“research_topic”, “”) # 可以在這里調(diào)用一個(gè) LLM 來更好地提煉研究主題和子問題 # 例如用戶說“幫我研究一下太陽能電池的最新進(jìn)展”LLM 可以提煉出“鈣鈦礦太陽能電池效率”、“硅基異質(zhì)結(jié)技術(shù)”等子方向。 # 這里為了簡化我們直接使用輸入作為主題。 refined_topic user_input print(f“[規(guī)劃節(jié)點(diǎn)] 研究主題已確定{refined_topic}”) return { “research_topic”: refined_topic, “current_step”: “gathering”, “web_sources”: [], # 初始化空列表 “l(fā)ocal_sources”: [], “synthesized_info”: None, “key_findings”: [], “report_draft”: None, “needs_human_review”: False, “human_feedback”: None, }節(jié)點(diǎn)2與3并行信息搜集研究需要多來源。我們可以讓網(wǎng)絡(luò)搜索和本地文檔讀取同時(shí)進(jìn)行。這需要用到 LangGraph 的并行化能力。def web_search_node(state: ResearchAgentState) - Dict[str, Any]: 網(wǎng)絡(luò)搜索節(jié)點(diǎn)。 topic state[“research_topic”] print(f“[網(wǎng)絡(luò)搜索節(jié)點(diǎn)] 正在搜索{topic}”) try: # 使用 Tavily、Serper 或 DuckDuckGo 等搜索 API results tavily_search( queryf“{topic} site:.edu OR site:.gov OR site:.org recent”, max_results5, include_answerFalse, search_depth“basic” ) # 格式化結(jié)果 web_sources [] for r in results.get(“results”, []): web_sources.append({ “title”: r.get(“title”, “No Title”), “url”: r.get(“url”, “”), “content”: r.get(“content”, “”)[:500], # 取前500字符 “score”: r.get(“score”, 0.0) }) return {“web_sources”: web_sources} except Exception as e: print(f“網(wǎng)絡(luò)搜索失敗{e}”) return {“web_sources”: []} # 失敗時(shí)返回空不影響整體流程 def local_doc_node(state: ResearchAgentState) - Dict[str, Any]: 本地文檔讀取節(jié)點(diǎn)。 # 假設(shè)我們有一個(gè)已知的本地文檔路徑列表或者從 state 中獲取 doc_paths [“./data/research_paper1.pdf”, “./data/industry_report.pdf”] local_sources [] for path in doc_paths: try: text read_pdf_text(path) # 簡單提取前幾行作為“標(biāo)題”并截取部分內(nèi)容 local_sources.append({ “title”: path.split(“/”)[-1], “path”: path, “content”: text[:1000] “...” if len(text) 1000 else text }) except Exception as e: print(f“讀取文檔 {path} 失敗{e}”) print(f“[本地文檔節(jié)點(diǎn)] 已讀取 {len(local_sources)} 份文檔?！? return {“l(fā)ocal_sources”: local_sources}節(jié)點(diǎn)4信息綜合與分析這個(gè)節(jié)點(diǎn)接收并行搜集的結(jié)果調(diào)用 LLM 進(jìn)行總結(jié)、去重、提取關(guān)鍵發(fā)現(xiàn)。def analysis_node(state: ResearchAgentState) - Dict[str, Any]: 分析節(jié)點(diǎn)綜合所有信息提取關(guān)鍵發(fā)現(xiàn)。 print(“[分析節(jié)點(diǎn)] 開始綜合與分析信息...”) web_info “\n\n”.join([f“標(biāo)題{s[‘title’]}\n內(nèi)容{s[‘content’]}” for s in state[“web_sources”]]) local_info “\n\n”.join([f“文檔{s[‘title’]}\n內(nèi)容{s[‘content’]}” for s in state[“l(fā)ocal_sources”]]) all_info f“## 網(wǎng)絡(luò)信息\n{web_info}\n\n## 本地文檔信息\n{local_info}” # 構(gòu)造給 LLM 的提示詞 prompt f“”” 你是一個(gè)專業(yè)的研究分析員。請(qǐng)基于以下關(guān)于“{state[‘research_topic’]}”的資料完成以下任務(wù) 1. **綜合摘要**用一段話200字以內(nèi)概括所有資料的核心觀點(diǎn)。 2. **提取關(guān)鍵發(fā)現(xiàn)**列出3-5條最重要的發(fā)現(xiàn)或事實(shí)每條用短句陳述。 3. **評(píng)估信息質(zhì)量**整體上這些資料的可靠性和全面性如何高/中/低 請(qǐng)嚴(yán)格按照以下 JSON 格式輸出不要有任何其他文字 json {{ “summary”: “綜合摘要文本”, “key_findings”: [“發(fā)現(xiàn)一”, “發(fā)現(xiàn)二”, “...”], “reliability”: “高/中/低” }}資料如下 {all_info} “””try: response call_llm( model“gpt-4”, messages[{“role”: “user”, “content”: prompt}], temperature0.2, response_format{“type”: “json_object”} # 要求 JSON 輸出 ) import json analysis_result json.loads(response[“choices”][0][“message”][“content”]) # 根據(jù)可靠性評(píng)估決定是否需要人工審核 needs_review analysis_result.get(“reliability”, “中”) in [“低”] return { “synthesized_info”: analysis_result.get(“summary”, “”), “key_findings”: analysis_result.get(“key_findings”, []), “needs_human_review”: needs_review, “current_step”: “reporting” } except Exception as e: print(f“分析過程出錯(cuò){e}”) # 出錯(cuò)時(shí)標(biāo)記需要人工審核 return { “synthesized_info”: “分析過程出現(xiàn)錯(cuò)誤?！? “key_findings”: [], “needs_human_review”: True, “current_step”: “reporting” }**節(jié)點(diǎn)5報(bào)告生成** 基于分析結(jié)果生成一份結(jié)構(gòu)化的報(bào)告草稿。 python def report_generation_node(state: ResearchAgentState) - Dict[str, Any]: 報(bào)告生成節(jié)點(diǎn)。 print(“[報(bào)告生成節(jié)點(diǎn)] 正在撰寫報(bào)告草稿...”) prompt f“”” 你是一名技術(shù)文檔工程師。請(qǐng)基于以下關(guān)于“{state[‘research_topic’]}”的分析結(jié)果撰寫一份簡潔的專業(yè)報(bào)告草稿。 **分析摘要** {state[‘synthesized_info’]} **關(guān)鍵發(fā)現(xiàn)** {chr(10).join(‘- ‘ f for f in state[‘key_findings’])} **報(bào)告要求** 1. 標(biāo)題明確。 2. 包含“概述”、“主要發(fā)現(xiàn)”、“結(jié)論”三個(gè)部分。 3. 語言客觀、精煉避免主觀臆斷。 4. 如果某些發(fā)現(xiàn)存在不確定性或信息源可靠性為‘低’請(qǐng)?jiān)趫?bào)告中用‘[待核實(shí)]’標(biāo)注。 5. 報(bào)告總長度控制在500字以內(nèi)。 請(qǐng)直接輸出報(bào)告正文不需要額外的解釋。 “”” try: response call_llm( model“gpt-4”, messages[{“role”: “user”, “content”: prompt}], temperature0.3 ) draft response[“choices”][0][“message”][“content”] return { “report_draft”: draft, “current_step”: “reviewing” # 進(jìn)入審核階段 } except Exception as e: print(f“報(bào)告生成失敗{e}”) return { “report_draft”: “報(bào)告生成失敗?！? “current_step”: “reviewing”, “needs_human_review”: True # 生成失敗也需人工介入 }節(jié)點(diǎn)6人工審核交互節(jié)點(diǎn)這是一個(gè)特殊的節(jié)點(diǎn)它可能需要暫停工作流等待外部輸入如用戶在界面上點(diǎn)擊“通過”或輸入反饋。在 LangGraph 中這可以通過“暫?！被颉皺z查點(diǎn)”機(jī)制實(shí)現(xiàn)但為了簡化我們模擬一個(gè)自動(dòng)判斷。def human_review_node(state: ResearchAgentState) - Dict[str, Any]: 人工審核節(jié)點(diǎn)模擬。 print(“[人工審核節(jié)點(diǎn)] 報(bào)告已生成等待審核...”) # 在實(shí)際應(yīng)用中這里可能會(huì) # 1. 將報(bào)告發(fā)送到審核隊(duì)列如數(shù)據(jù)庫、消息隊(duì)列。 # 2. 調(diào)用 graph.checkpoint() 保存當(dāng)前狀態(tài)并暫停。 # 3. 等待一個(gè)外部事件如 HTTP 回調(diào)來恢復(fù)執(zhí)行并攜帶 human_feedback。 # 為了演示我們模擬一個(gè)自動(dòng)通過的邏輯或者根據(jù) needs_human_review 決定。 if state[“needs_human_review”]: # 模擬人工審核后給出了反饋 simulated_feedback “報(bào)告整體不錯(cuò)但第三點(diǎn)發(fā)現(xiàn)的數(shù)據(jù)來源請(qǐng)?jiān)俸藢?shí)一下?！?print(f“[模擬] 收到人工反饋{simulated_feedback}”) return { “human_feedback”: simulated_feedback, “current_step”: “revising” } else: print(“[模擬] 無需人工審核自動(dòng)通過。”) return {“current_step”: “end”} # 前往結(jié)束節(jié)點(diǎn)7報(bào)告修訂可選如果收到了人工反饋這個(gè)節(jié)點(diǎn)負(fù)責(zé)根據(jù)反饋修改報(bào)告。def revision_node(state: ResearchAgentState) - Dict[str, Any]: 根據(jù)人工反饋修訂報(bào)告。 feedback state[“human_feedback”] draft state[“report_draft”] print(f“[修訂節(jié)點(diǎn)] 正在根據(jù)反饋進(jìn)行修訂。反饋{feedback}”) prompt f“”” 以下是報(bào)告草稿和審核反饋請(qǐng)根據(jù)反饋修改報(bào)告草稿。 **原始報(bào)告草稿** {draft} **審核反饋** {feedback} 請(qǐng)輸出修改后的完整報(bào)告。如果反饋不涉及具體修改可以保持原樣。 “”” try: response call_llm( model“gpt-4”, messages[{“role”: “user”, “content”: prompt}], temperature0.1 ) revised_draft response[“choices”][0][“message”][“content”] return { “report_draft”: revised_draft, “current_step”: “end” } except Exception as e: print(f“修訂失敗{e}”) return {“current_step”: “end”} # 即使失敗也嘗試結(jié)束3.3 組裝圖并定義流程邏輯現(xiàn)在我們將所有節(jié)點(diǎn)和邊組裝起來形成完整的工作流。from langgraph.graph import StateGraph, END # 1. 創(chuàng)建圖構(gòu)建器并指定狀態(tài)結(jié)構(gòu) workflow StateGraph(ResearchAgentState) # 2. 添加節(jié)點(diǎn) workflow.add_node(“plan”, planning_node) workflow.add_node(“search_web”, web_search_node) workflow.add_node(“search_local”, local_doc_node) workflow.add_node(“analyze”, analysis_node) workflow.add_node(“generate_report”, report_generation_node) workflow.add_node(“human_review”, human_review_node) workflow.add_node(“revise”, revision_node) # 3. 設(shè)置入口點(diǎn) workflow.set_entry_point(“plan”) # 4. 添加邊定義流程 # 規(guī)劃之后并行執(zhí)行網(wǎng)絡(luò)搜索和本地搜索 workflow.add_edge(“plan”, “search_web”) workflow.add_edge(“plan”, “search_local”) # 兩個(gè)搜索節(jié)點(diǎn)都完成后再進(jìn)入分析節(jié)點(diǎn)。 # 這里需要用到 add_conditional_edges 或 add_edge 的聚合功能。 # 更常見的模式是使用一個(gè)“聚合節(jié)點(diǎn)”來等待并行任務(wù)但為簡化我們假設(shè)它們都完成后自動(dòng)進(jìn)入分析。 # 我們可以通過讓 analyze 節(jié)點(diǎn)在 state 中檢查兩個(gè)源是否都已存在非空來隱式實(shí)現(xiàn)或者使用更高級(jí)的構(gòu)造。 # 這里我們采用一個(gè)簡單方式從 plan 直接連到 analyze但在 analyze 節(jié)點(diǎn)內(nèi)等待/檢查數(shù)據(jù)。 # 實(shí)際上更規(guī)范的做法是使用 LangGraph 的 Pregel 的并發(fā)特性。為了清晰我們調(diào)整一下邏輯 # 讓 plan 之后先到一個(gè)“協(xié)調(diào)節(jié)點(diǎn)”由它來并發(fā)觸發(fā)搜索然后等待結(jié)果。但代碼會(huì)復(fù)雜很多。 # 作為教程我們簡化順序執(zhí)行 web - local - analyze這仍然是有效的狀態(tài)機(jī)只是沒并發(fā)。 workflow.add_edge(“search_web”, “search_local”) workflow.add_edge(“search_local”, “analyze”) # 分析之后生成報(bào)告 workflow.add_edge(“analyze”, “generate_report”) # 報(bào)告生成后根據(jù)條件決定是進(jìn)入人工審核還是結(jié)束 def after_report_route(state: ResearchAgentState) - str: if state[“needs_human_review”]: return “human_review” else: return END workflow.add_conditional_edges( “generate_report”, after_report_route, { “human_review”: “human_review”, END: END } ) # 人工審核后進(jìn)入修訂節(jié)點(diǎn) workflow.add_edge(“human_review”, “revise”) # 修訂后結(jié)束 workflow.add_edge(“revise”, END) # 5. 編譯圖 app workflow.compile() # 6. 可視化需要安裝 graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(“無法顯示圖形但圖已編譯完成?！?3.4 運(yùn)行與調(diào)試現(xiàn)在我們可以運(yùn)行這個(gè)研究助手了。# 初始化狀態(tài) initial_state: ResearchAgentState { “messages”: [{“role”: “user”, “content”: “請(qǐng)幫我研究一下大語言模型在代碼生成方面的最新進(jìn)展和主要挑戰(zhàn)。”}], “research_topic”: “”, “web_sources”: [], “l(fā)ocal_sources”: [], “synthesized_info”: None, “key_findings”: [], “report_draft”: None, “needs_human_review”: False, “human_feedback”: None, “current_step”: “start” } # 運(yùn)行圖 final_state app.invoke(initial_state) print(“\n” “”*50) print(“最終報(bào)告”) print(“”*50) print(final_state[“report_draft”]) print(“\n關(guān)鍵發(fā)現(xiàn)”, final_state[“key_findings”]) print(“當(dāng)前步驟”, final_state[“current_step”])運(yùn)行后你會(huì)在控制臺(tái)看到各個(gè)節(jié)點(diǎn)的執(zhí)行日志并最終得到一份生成的研究報(bào)告。這個(gè)流程清晰地展示了信息如何在不同節(jié)點(diǎn)間流動(dòng)狀態(tài)如何被逐步更新以及條件邏輯如何影響執(zhí)行路徑。4. 高級(jí)特性與生產(chǎn)級(jí)考量當(dāng)你掌握了基礎(chǔ)構(gòu)建方法后以下高級(jí)特性能讓你的 LangGraph Agent 更強(qiáng)大、更穩(wěn)健。4.1 持久化與檢查點(diǎn)讓 Agent 擁有“記憶”對(duì)于長時(shí)間運(yùn)行或需要中斷恢復(fù)的任務(wù)持久化至關(guān)重要。LangGraph 與 LangChain 生態(tài)深度集成支持將運(yùn)行狀態(tài)保存到數(shù)據(jù)庫如 PostgreSQL、MySQL或內(nèi)存中。from langgraph.checkpoint import MemorySaver from langgraph.graph import StateGraph, START, END # 在編譯圖時(shí)加入檢查點(diǎn)管理器 checkpointer MemorySaver() workflow StateGraph(ResearchAgentState, checkpointercheckpointer) # ... 添加節(jié)點(diǎn)和邊 ... app workflow.compile() # 運(yùn)行時(shí)會(huì)自動(dòng)創(chuàng)建檢查點(diǎn) config {“configurable”: {“thread_id”: “user_123_research_task”}} initial_state {…} # 第一次調(diào)用 result1 app.invoke(initial_state, configconfig) # 假設(shè)任務(wù)在這里因某種原因暫停了state 已被保存。 # 之后可以從最后一個(gè)檢查點(diǎn)恢復(fù)執(zhí)行 # 我們通過傳入一個(gè)空的輸入并指定從上一個(gè)線程繼續(xù) resumed_state app.invoke( {“messages”: [{“role”: “user”, “content”: “繼續(xù)”}]}, # 可以傳入新的消息來影響流程 configconfig )MemorySaver適用于開發(fā)和測試。在生產(chǎn)環(huán)境中你會(huì)使用PostgresSaver或MongoDBSaver將狀態(tài)持久化到外部存儲(chǔ)從而實(shí)現(xiàn)跨會(huì)話、跨服務(wù)器重啟的 Agent 狀態(tài)恢復(fù)。這對(duì)于構(gòu)建客服對(duì)話機(jī)器人等長周期應(yīng)用是必備功能。4.2 子圖與模塊化構(gòu)建復(fù)雜系統(tǒng)的基石當(dāng)你的 Agent 變得非常復(fù)雜時(shí)將所有邏輯塞進(jìn)一個(gè)圖里會(huì)難以維護(hù)。子圖允許你將一部分功能例如一個(gè)完整的“搜索-評(píng)估-過濾”流程封裝成一個(gè)獨(dú)立的、可復(fù)用的圖然后作為單個(gè)節(jié)點(diǎn)嵌入到主圖中。from langgraph.graph import StateGraph as SubStateGraph # 1. 定義一個(gè)“信息驗(yàn)證”子圖 def create_validation_subgraph(): sub_builder SubStateGraph(ResearchAgentState) def validate_source_node(state): # 驗(yàn)證單個(gè)信息來源的可信度 # ... return {“source_credibility_score”: 0.95} def aggregate_validation_node(state): # 聚合所有驗(yàn)證結(jié)果 # ... return {“overall_credibility”: “high”} sub_builder.add_node(“validate”, validate_source_node) sub_builder.add_node(“aggregate”, aggregate_validation_node) sub_builder.add_edge(“validate”, “aggregate”) sub_builder.set_entry_point(“validate”) sub_builder.set_finish_point(“aggregate”) # 子圖有明確的結(jié)束點(diǎn) return sub_builder.compile() # 2. 在主圖中將這個(gè)子圖作為一個(gè)節(jié)點(diǎn)添加 validation_subgraph create_validation_subgraph() workflow.add_node(“validate_sources”, validation_subgraph) # 3. 在主圖的適當(dāng)位置連接這個(gè)節(jié)點(diǎn) workflow.add_edge(“search_web”, “validate_sources”) workflow.add_edge(“validate_sources”, “analyze”)這樣做的好處是關(guān)注點(diǎn)分離主圖邏輯清晰子圖負(fù)責(zé)特定復(fù)雜功能?？蓮?fù)用性同一個(gè)驗(yàn)證子圖可以被用在多個(gè)不同的主圖中?？蓽y試性子圖可以獨(dú)立進(jìn)行單元測試。4.3 流式輸出與中斷對(duì)于需要實(shí)時(shí)向用戶反饋進(jìn)度的 Agent如一邊搜索一邊顯示結(jié)果或者需要支持用戶中途取消的任務(wù)流式輸出和中斷機(jī)制很重要。流式輸出LangGraph 的app.stream()方法可以讓你逐步獲取每個(gè)節(jié)點(diǎn)執(zhí)行后的狀態(tài)更新。inputs {“messages”: [{“role”: “user”, “content”: “研究主題”}]} config {“configurable”: {“thread_id”: “stream_demo”}} for event in app.stream(inputs, configconfig, stream_mode“values”): # event 是一個(gè)元組 (node_name, state_update) node, state_update event print(f“節(jié)點(diǎn) [{node}] 執(zhí)行完畢。”) if “report_draft” in state_update and state_update[“report_draft”]: print(“報(bào)告草稿已更新片段...”) # 你可以將 state_update 中的部分內(nèi)容如新的消息實(shí)時(shí)發(fā)送給前端中斷處理你可以在節(jié)點(diǎn)函數(shù)中檢查某個(gè)標(biāo)志位例如來自一個(gè)全局信號(hào)或數(shù)據(jù)庫或者通過app.invoke()的config傳入超時(shí)設(shè)置來實(shí)現(xiàn)優(yōu)雅的中斷。更精細(xì)的控制可能需要結(jié)合檢查點(diǎn)在每次節(jié)點(diǎn)執(zhí)行前檢查是否收到“取消”指令。4.4 錯(cuò)誤處理與韌性生產(chǎn)級(jí) Agent 必須能妥善處理失敗如 API 超時(shí)、網(wǎng)絡(luò)錯(cuò)誤、LLM 輸出格式錯(cuò)誤。節(jié)點(diǎn)級(jí)容錯(cuò)在每個(gè)節(jié)點(diǎn)函數(shù)內(nèi)部使用try...except并返回一個(gè)代表錯(cuò)誤的狀態(tài)如{“error”: “API timeout”, “step”: “retry”}。圖級(jí)容錯(cuò)路由利用條件邊根據(jù) State 中是否有error字段將流程路由到一個(gè)專門的“錯(cuò)誤處理節(jié)點(diǎn)”。這個(gè)節(jié)點(diǎn)可以嘗試重試、降級(jí)處理如使用備用模型、或通知用戶。超時(shí)設(shè)置在調(diào)用 LLM 或外部 API 時(shí)務(wù)必設(shè)置超時(shí)參數(shù)。驗(yàn)證與重試對(duì)于關(guān)鍵節(jié)點(diǎn)可以設(shè)計(jì)一個(gè)包裝器在失敗時(shí)自動(dòng)重試幾次。def robust_llm_call(prompt, max_retries3): for i in range(max_retries): try: return call_llm(prompt, timeout30) # 設(shè)置超時(shí) except TimeoutError: print(f“LLM 調(diào)用超時(shí)第 {i1} 次重試...”) if i max_retries - 1: raise except Exception as e: print(f“LLM 調(diào)用失敗{e}”) raise # 非超時(shí)錯(cuò)誤直接拋出將這些策略結(jié)合起來你的 Agent 就能在部分組件失效時(shí)依然提供有價(jià)值的服務(wù)或者至少能給出清晰的錯(cuò)誤報(bào)告而不是直接崩潰。5. 常見問題與避坑指南在實(shí)際開發(fā)中你會(huì)遇到各種各樣的問題。以下是我從多個(gè)項(xiàng)目中總結(jié)出的高頻問題和解決方案。5.1 狀態(tài)管理混亂問題State 字段設(shè)計(jì)不合理導(dǎo)致節(jié)點(diǎn)間數(shù)據(jù)污染或難以追蹤。例如多個(gè)節(jié)點(diǎn)都修改同一個(gè)列表但意圖不同。解決方案字段職責(zé)單一化為不同的數(shù)據(jù)階段設(shè)計(jì)不同的字段。例如不要只用sources而是分為raw_sources、filtered_sources、processed_sources。使用不可變數(shù)據(jù)結(jié)構(gòu)在返回更新字典時(shí)盡量創(chuàng)建新的列表或字典而不是修改傳入 state 中的對(duì)象。雖然 LangGraph 的合并機(jī)制會(huì)處理但清晰的意圖有助于調(diào)試。添加調(diào)試字段像我們例子中的current_step或者last_updated_by這樣的字段在復(fù)雜流程中非常有助于追蹤狀態(tài)變化軌跡。5.2 條件邊路由邏輯過于復(fù)雜問題路由函數(shù)should_do_something(state)里塞滿了大量的if-else判斷難以維護(hù)和測試。解決方案路由表模式將路由邏輯抽象成配置。例如定義一個(gè)字典鍵為條件名值為一個(gè)判斷函數(shù)和對(duì)應(yīng)的目標(biāo)節(jié)點(diǎn)。routing_rules [ {“name”: “needs_detail”, “condition”: lambda s: len(s[‘findings’]) 3, “target”: “detail_search_node”}, {“name”: “is_controversial”, “condition”: lambda s: “爭議” in s[‘summary’], “target”: “human_review_node”}, {“name”: “default”, “condition”: lambda s: True, “target”: END}, ]然后在路由函數(shù)中遍歷這個(gè)列表返回第一個(gè)滿足條件的target。專用路由節(jié)點(diǎn)如果路由邏輯極其復(fù)雜可以將其作為一個(gè)獨(dú)立的節(jié)點(diǎn)。這個(gè)節(jié)點(diǎn)不干別的只負(fù)責(zé)計(jì)算下一個(gè)節(jié)點(diǎn)名稱并返回。這樣可以將路由邏輯從conditional_edge中分離出來便于單獨(dú)測試和優(yōu)化。5.3 并行與同步的坑問題像我們例子中想并行執(zhí)行web_search和local_doc但簡單的add_edge無法實(shí)現(xiàn)真正的并發(fā)最終變成了順序執(zhí)行。解決方案使用Pregel的并發(fā)特性LangGraph 底層基于 Pregel 模型。要實(shí)現(xiàn)真正的并發(fā)需要將多個(gè)節(jié)點(diǎn)添加到同一個(gè)“層”并正確配置它們的依賴關(guān)系。這通常涉及更底層的StateGraph配置或者使用Channel的概念。對(duì)于大多數(shù)應(yīng)用如果并發(fā)不是性能瓶頸順序執(zhí)行簡化版也足夠清晰。明確聚合點(diǎn)如果必須并發(fā)一定要設(shè)計(jì)一個(gè)明確的“聚合節(jié)點(diǎn)”該節(jié)點(diǎn)等待所有并發(fā)分支的輸出都就緒后再執(zhí)行后續(xù)操作?？梢栽?State 中設(shè)置標(biāo)志位或者利用 LangGraph 更高級(jí)的Channel和Barrier原語。5.4 LLM 調(diào)用成本與延遲問題圖中每個(gè)節(jié)點(diǎn)都調(diào)用 LLM導(dǎo)致單次運(yùn)行成本高、速度慢。解決方案緩存對(duì)內(nèi)容變化不大的查詢?nèi)纭翱偨Y(jié)以下文本”這類提示詞固定、輸入變化不大的操作使用 LLM 調(diào)用緩存。LangChain/LangGraph 社區(qū)有一些緩存集成方案。批處理將多個(gè)小的、獨(dú)立的 LLM 調(diào)用合并成一個(gè)批處理提示。例如分析節(jié)點(diǎn)中需要評(píng)估多個(gè)信息來源的可信度可以設(shè)計(jì)一個(gè)提示詞讓 LLM 一次性對(duì)所有來源打分而不是循環(huán)調(diào)用 N 次。模型分級(jí)不是所有步驟都需要 GPT-4。對(duì)于信息提取、簡單分類等任務(wù)使用更便宜、更快的模型如 Claude Haiku, GPT-3.5-Turbo。只在需要深度推理、創(chuàng)意生成或關(guān)鍵決策時(shí)使用大模型。異步調(diào)用如果圖中有真正的并發(fā)節(jié)點(diǎn)確保使用異步的 LLM 客戶端如openai.AsyncOpenAI來并行執(zhí)行這些調(diào)用而不是同步等待。5.5 調(diào)試與可視化困難問題圖執(zhí)行到哪一步了State 變成什么樣了為什么卡住了解決方案善用日志在每個(gè)節(jié)點(diǎn)的開始和結(jié)束打印關(guān)鍵信息包括傳入的 State 片段和返回的更新。使用結(jié)構(gòu)化的日志格式如 JSON方便搜索和分析。利用檢查點(diǎn)不僅為了持久化檢查點(diǎn)也是強(qiáng)大的調(diào)試工具。你可以在任何步驟暫停檢查保存的 State 快照。圖形化調(diào)試app.get_graph().draw_mermaid_png()生成流程圖幫你宏觀理解邏輯。對(duì)于單次運(yùn)行可以手動(dòng)記錄每個(gè)節(jié)點(diǎn)的輸入輸出繪制出實(shí)際的執(zhí)行路徑圖。單元測試子圖將復(fù)雜的子圖單獨(dú)編譯和測試用固定的輸入驗(yàn)證其輸出是否符合預(yù)期。這比測試整個(gè)大圖要容易得多。構(gòu)建 LangGraph Agent 是一個(gè)迭代過程。從最簡單的線性流程開始逐步添加分支、循環(huán)、并行和錯(cuò)誤處理。時(shí)刻記住狀態(tài)機(jī)的基本思想當(dāng)前狀態(tài) 事件/條件 下一個(gè)狀態(tài)和動(dòng)作。用這個(gè)思維模型去設(shè)計(jì)你的節(jié)點(diǎn)和邊你會(huì)發(fā)現(xiàn)再復(fù)雜的業(yè)務(wù)邏輯也能被清晰地建模和實(shí)現(xiàn)。