最近在嘗試把一些單點(diǎn)任務(wù)升級(jí)成自動(dòng)化流程時(shí)我遇到了一個(gè)典型問(wèn)題用腳本或簡(jiǎn)單工具鏈跑通一次任務(wù)很容易但一旦想讓多個(gè)步驟、多個(gè)模型、多個(gè)判斷點(diǎn)串聯(lián)起來(lái)形成一個(gè)能穩(wěn)定運(yùn)行、狀態(tài)可控、還能處理異常的“工業(yè)級(jí)”流程立刻就變得棘手起來(lái)。腳本越寫(xiě)越長(zhǎng)狀態(tài)管理混亂錯(cuò)誤處理像打補(bǔ)丁想加個(gè)新功能或換個(gè)模型都得大動(dòng)干戈。這讓我開(kāi)始重新審視那些宣稱能構(gòu)建“智能體”的框架。很多工具確實(shí)能快速拼出一個(gè)原型演示起來(lái)很酷但離真正的“工業(yè)級(jí)”落地中間還隔著狀態(tài)管理、錯(cuò)誤恢復(fù)、多智能體協(xié)作、長(zhǎng)期記憶和可觀測(cè)性這幾座大山。直到我開(kāi)始深入使用 LangGraph并結(jié)合 LangChain 的生態(tài)才感覺(jué)找到了一個(gè)能把想法系統(tǒng)化落地的路徑。它提供的不是另一個(gè)花哨的演示而是一套用于構(gòu)建可控、可維護(hù)、可擴(kuò)展?fàn)顟B(tài)化工作流的“工程骨架”。今天我們就來(lái)深度拆解這套架構(gòu)。我不會(huì)只講概念而是會(huì)結(jié)合一個(gè)從零到一的完整實(shí)戰(zhàn)項(xiàng)目帶你走過(guò)這幾個(gè)關(guān)鍵階段如何用 StateGraph 設(shè)計(jì)并管控復(fù)雜狀態(tài)流、如何將單智能體擴(kuò)展為分工明確的多智能體系統(tǒng)、如何無(wú)縫集成 RAG 與多種大模型最終形成一個(gè)接近生產(chǎn)可用的智能體應(yīng)用。你會(huì)發(fā)現(xiàn)真正的價(jià)值不在于調(diào)用一兩個(gè) API而在于構(gòu)建一個(gè)可靠、透明且易于迭代的自動(dòng)化系統(tǒng)。1. 從“一次性腳本”到“狀態(tài)化工作流”為什么需要 LangGraph在開(kāi)始寫(xiě)代碼之前我們必須先想清楚一個(gè)問(wèn)題當(dāng)我們?cè)谡f(shuō)“構(gòu)建一個(gè)智能體”時(shí)我們到底在構(gòu)建什么是寫(xiě)一個(gè)調(diào)用了大模型 API 的函數(shù)嗎是串聯(lián)幾個(gè)工具調(diào)用嗎這些都很初級(jí)。一個(gè)工業(yè)級(jí)的智能體本質(zhì)上是一個(gè)有狀態(tài)、可分支、可循環(huán)、具備錯(cuò)誤恢復(fù)能力的工作流引擎。想象一個(gè)客服場(chǎng)景的智能體用戶輸入問(wèn)題 - 判斷意圖分類- 如果是查詢知識(shí)庫(kù)則檢索 - 組織答案 - 如果答案不完整可能觸發(fā)二次檢索或轉(zhuǎn)人工 - 最終回復(fù)。這個(gè)流程里有明確的“狀態(tài)”當(dāng)前進(jìn)行到哪一步、手里有什么數(shù)據(jù)有“分支”不同的意圖走向不同的處理節(jié)點(diǎn)有“循環(huán)”檢索-評(píng)估-再檢索還需要處理各種異常檢索失敗、模型超時(shí)。如果你用純腳本寫(xiě)很快就會(huì)變成“面條代碼”狀態(tài)用全局變量傳來(lái)傳去循環(huán)和分支靠一堆if-else硬編碼加個(gè)新步驟或者改個(gè)邏輯都心驚膽戰(zhàn)。而 LangGraph 的核心抽象——StateGraph就是為了解決這個(gè)問(wèn)題而生的。1.1 StateGraph把工作流畫(huà)成一張“狀態(tài)轉(zhuǎn)移圖”StateGraph 鼓勵(lì)你用“圖”的思維來(lái)建模工作流。圖中的節(jié)點(diǎn)Node是一個(gè)個(gè)執(zhí)行單元比如意圖判斷、知識(shí)檢索、答案生成邊Edge定義了節(jié)點(diǎn)之間的流轉(zhuǎn)條件。整個(gè)系統(tǒng)的“狀態(tài)”是一個(gè)共享的數(shù)據(jù)結(jié)構(gòu)在各個(gè)節(jié)點(diǎn)間傳遞和修改。這種設(shè)計(jì)帶來(lái)了幾個(gè)關(guān)鍵優(yōu)勢(shì)可視化與可理解性工作流不再是隱藏在代碼里的邏輯而是一張可以畫(huà)出來(lái)的圖。新人能快速理解業(yè)務(wù)邏輯。模塊化每個(gè)節(jié)點(diǎn)功能單一易于開(kāi)發(fā)、測(cè)試和復(fù)用。修改一個(gè)節(jié)點(diǎn)不會(huì)輕易影響全局。顯式狀態(tài)管理所有中間數(shù)據(jù)都存在于一個(gè)明確定義的“狀態(tài)”對(duì)象中避免了隱式依賴和全局變量的混亂。內(nèi)置的控制流LangGraph 原生支持條件分支conditional_edge和循環(huán)將邊指向之前的節(jié)點(diǎn)這讓實(shí)現(xiàn)復(fù)雜的業(yè)務(wù)邏輯變得直觀。1.2 LangGraph 與 LangChain 的關(guān)系生態(tài)與引擎很多人會(huì)混淆 LangGraph 和 LangChain。簡(jiǎn)單來(lái)說(shuō)LangChain是一個(gè)豐富的“生態(tài)工具箱”提供了連接大模型、向量數(shù)據(jù)庫(kù)、工具鏈、記憶模塊等各類組件的標(biāo)準(zhǔn)化接口和實(shí)現(xiàn)。它讓你能方便地“使用”各種AI能力。LangGraph則是建立在 LangChain 之上的“工作流引擎”或“編排框架”。它利用 LangChain 的組件作為節(jié)點(diǎn)的工作單元專注于解決如何將這些單元有機(jī)地、有狀態(tài)地組織成一個(gè)穩(wěn)健流程的問(wèn)題。你可以只用 LangChain 來(lái)構(gòu)建簡(jiǎn)單的鏈Chain但當(dāng)你需要復(fù)雜、有狀態(tài)、多步驟的交互時(shí)LangGraph 是更專業(yè)的選擇。它們不是替代關(guān)系而是互補(bǔ)LangGraph 負(fù)責(zé)調(diào)度和狀態(tài)LangChain 負(fù)責(zé)提供可調(diào)度的“零件”。2. 實(shí)戰(zhàn)第一步設(shè)計(jì)狀態(tài)與構(gòu)建基礎(chǔ) StateGraph理論說(shuō)再多不如動(dòng)手。我們以一個(gè)“智能研究助手”為項(xiàng)目目標(biāo)用戶輸入一個(gè)復(fù)雜問(wèn)題智能體需要自動(dòng)進(jìn)行網(wǎng)絡(luò)搜索、閱讀相關(guān)資料、總結(jié)分析并最終生成一份結(jié)構(gòu)化的報(bào)告。這個(gè)流程涉及多個(gè)步驟和決策點(diǎn)非常適合用 LangGraph 來(lái)構(gòu)建。2.1 定義核心狀態(tài)State狀態(tài)是整個(gè)工作流的“中央數(shù)據(jù)總線”。我們需要仔細(xì)設(shè)計(jì)它包含哪些信息。一個(gè)好的狀態(tài)設(shè)計(jì)應(yīng)該包含輸入、中間產(chǎn)物、最終輸出以及控制標(biāo)志。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class GraphState(TypedDict): 定義工作流的狀態(tài)結(jié)構(gòu)。 # 用戶輸入 question: str # 控制流程的標(biāo)志位例如當(dāng)前階段 current_step: str # 中間數(shù)據(jù)搜索查詢?cè)~、搜索結(jié)果、提取的文本 search_queries: List[str] search_results: List[dict] # 可能包含標(biāo)題、鏈接、摘要 extracted_texts: List[str] # 中間數(shù)據(jù)分析過(guò)程中的草稿、要點(diǎn) analysis_draft: str key_points: List[str] # 最終輸出 final_report: str # 消息歷史如果涉及多輪對(duì)話或智能體間通信 messages: Annotated[list, add_messages]這里我們使用了TypedDict來(lái)獲得類型提示。Annotated用于messages字段是 LangGraph 處理消息歷史的一種便捷方式。current_step是一個(gè)簡(jiǎn)單的控制標(biāo)志用于決定下一個(gè)節(jié)點(diǎn)。2.2 創(chuàng)建節(jié)點(diǎn)Node與邊Edge節(jié)點(diǎn)是執(zhí)行具體工作的函數(shù)。每個(gè)函數(shù)接收整個(gè)GraphState作為輸入修改其中部分內(nèi)容然后返回更新后的狀態(tài)。我們先創(chuàng)建兩個(gè)基礎(chǔ)節(jié)點(diǎn)search_web和extract_content。from langchain_community.tools import TavilySearchResults from langchain_community.document_loaders import AsyncHtmlLoader from langchain_community.document_transformers import Html2TextTransformer # 初始化工具需要TAVILY_API_KEY search_tool TavilySearchResults(max_results3) def search_web_node(state: GraphState) - GraphState: 節(jié)點(diǎn)根據(jù)問(wèn)題生成搜索詞并執(zhí)行搜索。 question state[question] # 1. 生成搜索查詢?cè)~這里簡(jiǎn)化直接使用問(wèn)題。進(jìn)階版可以用LLM優(yōu)化查詢?cè)~ queries [question] # 模擬可能生成多個(gè)查詢?cè)~ # queries llm.invoke(f針對(duì)問(wèn)題{question}生成3個(gè)不同的搜索關(guān)鍵詞。) # 2. 執(zhí)行搜索 all_results [] for query in queries: try: results search_tool.invoke(query) all_results.extend(results) except Exception as e: print(f搜索查詢 {query} 時(shí)出錯(cuò): {e}) # 可以在這里將錯(cuò)誤信息存入state供后續(xù)節(jié)點(diǎn)處理 # 更新?tīng)顟B(tài) return { search_queries: queries, search_results: all_results, current_step: content_extraction # 更新步驟標(biāo)志 } def extract_content_node(state: GraphState) - GraphState: 節(jié)點(diǎn)從搜索結(jié)果中加載并提取純文本內(nèi)容。 results state.get(search_results, []) urls [res.get(url) for res in results if res.get(url)] if not urls: return {extracted_texts: [], current_step: analyze} # 使用異步加載器提高效率 loader AsyncHtmlLoader(urls) docs loader.load() # 將HTML轉(zhuǎn)換為純凈文本 transformer Html2TextTransformer() transformed_docs transformer.transform_documents(docs) extracted_texts [doc.page_content[:5000] for doc in transformed_docs] # 截?cái)啾苊膺^(guò)長(zhǎng) return { extracted_texts: extracted_texts, current_step: analyze }有了節(jié)點(diǎn)我們開(kāi)始構(gòu)建圖。from langgraph.graph import StateGraph, END # 1. 創(chuàng)建圖構(gòu)建器 workflow StateGraph(GraphState) # 2. 添加節(jié)點(diǎn) workflow.add_node(search_web, search_web_node) workflow.add_node(extract_content, extract_content_node) # 3. 添加邊定義執(zhí)行順序 workflow.add_edge(search_web, extract_content) workflow.add_edge(extract_content, END) # 暫時(shí)直接結(jié)束 # 4. 設(shè)置入口點(diǎn) workflow.set_entry_point(search_web) # 5. 編譯圖 app workflow.compile()現(xiàn)在一個(gè)最簡(jiǎn)單的兩節(jié)點(diǎn)線性工作流就完成了。你可以運(yùn)行它initial_state GraphState(question什么是LangGraph它和LangChain有什么區(qū)別, current_stepstart) final_state app.invoke(initial_state) print(final_state[extracted_texts][0][:500]) # 打印部分提取的文本這只是一個(gè)開(kāi)始。目前的工作流是僵硬的直線。接下來(lái)我們要引入條件分支讓它具備決策能力。3. 引入決策與循環(huán)讓工作流“智能”起來(lái)一個(gè)只會(huì)機(jī)械執(zhí)行固定步驟的流程不是智能體。我們需要它能夠根據(jù)中間結(jié)果做出判斷。比如如果搜索返回的結(jié)果太少或質(zhì)量不高我們應(yīng)該觸發(fā)新一輪搜索如果提取的文本已經(jīng)足夠回答簡(jiǎn)單問(wèn)題或許可以跳過(guò)深度分析。3.1 使用條件邊Conditional Edge我們修改流程在extract_content節(jié)點(diǎn)后添加一個(gè)路由決策判斷提取的文本是否足夠進(jìn)行下一步分析。首先創(chuàng)建一個(gè)路由函數(shù)。這個(gè)函數(shù)不修改狀態(tài)只根據(jù)當(dāng)前狀態(tài)返回下一個(gè)節(jié)點(diǎn)的名稱。def route_after_extraction(state: GraphState) - str: 路由函數(shù)判斷內(nèi)容提取后是進(jìn)行分析還是重新搜索。 texts state.get(extracted_texts, []) # 簡(jiǎn)單的啟發(fā)式規(guī)則如果沒(méi)提取到文本或者文本總長(zhǎng)度太短則重新搜索 if not texts or sum(len(t) for t in texts) 500: # 可以在這里修改搜索策略例如更新查詢?cè)~ new_queries state.get(search_queries, []) [補(bǔ)充信息] state[search_queries] new_queries return search_web # 返回需要重新執(zhí)行的節(jié)點(diǎn)名 else: return analyze # 進(jìn)入分析節(jié)點(diǎn)然后修改我們的圖構(gòu)建邏輯用add_conditional_edges來(lái)替代之前直接指向END的邊。# 首先我們需要一個(gè)分析節(jié)點(diǎn)暫未實(shí)現(xiàn)先作為占位符 def analyze_node(state: GraphState) - GraphState: 節(jié)點(diǎn)分析提取的文本生成要點(diǎn)。 # 這里后續(xù)會(huì)集成LLM state[key_points] [要點(diǎn)1, 要點(diǎn)2] state[current_step] generate_report return state def generate_report_node(state: GraphState) - GraphState: 節(jié)點(diǎn)根據(jù)要點(diǎn)生成最終報(bào)告。 # 這里后續(xù)會(huì)集成LLM state[final_report] 這是一份基于分析的模擬報(bào)告。 return state # 重建圖 workflow StateGraph(GraphState) # 添加所有節(jié)點(diǎn) workflow.add_node(search_web, search_web_node) workflow.add_node(extract_content, extract_content_node) workflow.add_node(analyze, analyze_node) workflow.add_node(generate_report, generate_report_node) # 設(shè)置線性邊 workflow.add_edge(search_web, extract_content) workflow.add_edge(analyze, generate_report) workflow.add_edge(generate_report, END) # 關(guān)鍵為 extract_content 節(jié)點(diǎn)添加條件邊 workflow.add_conditional_edges( extract_content, route_after_extraction, # 路由函數(shù) { search_web: search_web, # 如果返回search_web則跳轉(zhuǎn)到search_web節(jié)點(diǎn) analyze: analyze # 如果返回analyze則跳轉(zhuǎn)到analyze節(jié)點(diǎn) } ) # 設(shè)置入口 workflow.set_entry_point(search_web) app workflow.compile()現(xiàn)在工作流具備了基本的“判斷-循環(huán)”能力。如果第一次搜索和提取的內(nèi)容不夠它會(huì)嘗試修改查詢?cè)~這里簡(jiǎn)單追加了“補(bǔ)充信息”并重新搜索直到內(nèi)容達(dá)標(biāo)或達(dá)到隱式循環(huán)上限需要額外機(jī)制避免無(wú)限循環(huán)。這就是狀態(tài)化工作流的威力狀態(tài)search_queries在循環(huán)中被修改并影響下一輪執(zhí)行。3.2 防止無(wú)限循環(huán)與超時(shí)處理在真實(shí)場(chǎng)景中必須避免無(wú)限循環(huán)。LangGraph 提供了interrupt和Timeout等機(jī)制但更常見(jiàn)的做法是在狀態(tài)中設(shè)置一個(gè)計(jì)數(shù)器并在路由函數(shù)中檢查。class GraphState(TypedDict): # ... 其他字段同上 ... search_attempt_count: int # 新增搜索嘗試計(jì)數(shù)器 def route_after_extraction(state: GraphState) - str: texts state.get(extracted_texts, []) attempt state.get(search_attempt_count, 0) # 規(guī)則1如果嘗試超過(guò)3次強(qiáng)制進(jìn)入分析即使內(nèi)容不足 if attempt 3: return analyze # 規(guī)則2內(nèi)容不足且嘗試次數(shù)未超限則重新搜索 if not texts or sum(len(t) for t in texts) 500: state[search_attempt_count] attempt 1 new_queries state.get(search_queries, []) [f補(bǔ)充信息嘗試{attempt1}] state[search_queries] new_queries return search_web else: return analyze4. 從單智能體到多智能體協(xié)作當(dāng)任務(wù)足夠復(fù)雜時(shí)讓一個(gè)“全能”的智能體做所有事情會(huì)導(dǎo)致其邏輯臃腫且容易出錯(cuò)。更好的模式是多智能體協(xié)作每個(gè)智能體對(duì)應(yīng)一個(gè)或多個(gè)節(jié)點(diǎn)職責(zé)單一通過(guò)狀態(tài)流進(jìn)行通信和協(xié)作。在我們的研究助手項(xiàng)目中可以設(shè)計(jì)三個(gè)智能體研究員Researcher負(fù)責(zé)搜索和獲取信息。對(duì)應(yīng)search_web和extract_content節(jié)點(diǎn)。分析師Analyst負(fù)責(zé)閱讀、理解和提煉信息要點(diǎn)。對(duì)應(yīng)analyze節(jié)點(diǎn)。撰稿人Writer負(fù)責(zé)根據(jù)要點(diǎn)組織語(yǔ)言生成格式優(yōu)美的最終報(bào)告。對(duì)應(yīng)generate_report節(jié)點(diǎn)。4.1 為智能體注入“大腦”集成大模型之前的節(jié)點(diǎn)都是基于規(guī)則或簡(jiǎn)單工具。現(xiàn)在我們?yōu)榉治鰩熀妥迦斯?jié)點(diǎn)集成大模型如 OpenAI GPT-4、 Anthropic Claude 或本地模型讓它們真正具備理解和生成能力。首先定義兩個(gè)不同角色的 LLM 調(diào)用。在實(shí)踐中可以為不同角色設(shè)定不同的系統(tǒng)提示詞System Prompt甚至使用不同能力的模型。from langchain_openai import ChatOpenAI # 假設(shè)使用 OpenAI API llm_analyst ChatOpenAI(modelgpt-4-turbo-preview, temperature0.1) llm_writer ChatOpenAI(modelgpt-4-turbo-preview, temperature0.7) # 撰稿人可以更有創(chuàng)造性 def analyze_node(state: GraphState) - GraphState: 分析師智能體閱讀文本提煉關(guān)鍵要點(diǎn)。 texts state.get(extracted_texts, []) question state[question] if not texts: state[key_points] [未找到相關(guān)信息。] return state # 構(gòu)建給分析師的提示詞 context \n\n---\n\n.join(texts[:3]) # 限制上下文長(zhǎng)度 prompt f 你是一位嚴(yán)謹(jǐn)?shù)难芯糠治鰩?。?qǐng)基于以下背景資料針對(duì)用戶問(wèn)題“{question}”提煉出3-5個(gè)最核心的要點(diǎn)。 要點(diǎn)要求客觀、簡(jiǎn)潔、基于資料。 背景資料 {context} 核心要點(diǎn)每條用‘- ’開(kāi)頭 response llm_analyst.invoke(prompt) points response.content.strip().split(\n) # 簡(jiǎn)單清理 points [p.strip(- ).strip() for p in points if p.strip().startswith(-)] state[key_points] points if points else [分析未能提取出明確要點(diǎn)。] state[current_step] generate_report return state def generate_report_node(state: GraphState) - GraphState: 撰稿人智能體根據(jù)要點(diǎn)撰寫(xiě)結(jié)構(gòu)化報(bào)告。 question state[question] key_points state.get(key_points, []) prompt f 你是一位專業(yè)的科技報(bào)告撰稿人。請(qǐng)根據(jù)以下要點(diǎn)圍繞問(wèn)題“{question}”撰寫(xiě)一份結(jié)構(gòu)清晰、語(yǔ)言流暢的簡(jiǎn)短報(bào)告約300字。 報(bào)告結(jié)構(gòu)建議 1. 引言重申問(wèn)題。 2. 核心發(fā)現(xiàn)基于要點(diǎn)展開(kāi)。 3. 總結(jié)與意義。 要點(diǎn) {chr(10).join(f- {p} for p in key_points)} 報(bào)告 response llm_writer.invoke(prompt) state[final_report] response.content return state現(xiàn)在我們的工作流擁有了兩個(gè)具備 LLM 能力的智能體節(jié)點(diǎn)。它們通過(guò)共享的GraphState特別是extracted_texts和key_points字段進(jìn)行協(xié)作。研究員為分析師準(zhǔn)備材料分析師為撰稿人提煉骨架撰稿人最終成文。這種基于狀態(tài)共享的協(xié)作模式比讓單個(gè) LLM 一次性完成所有步驟更加可控、可解釋也更容易針對(duì)單個(gè)環(huán)節(jié)進(jìn)行優(yōu)化或替換模型。5. 集成 RAG為智能體注入“長(zhǎng)期記憶”與“領(lǐng)域知識(shí)”多智能體協(xié)作解決了流程問(wèn)題但它們的知識(shí)仍然局限于單次會(huì)話和通用模型。對(duì)于專業(yè)領(lǐng)域問(wèn)題如公司內(nèi)部文檔、技術(shù)手冊(cè)、行業(yè)報(bào)告我們需要為智能體配備RAG檢索增強(qiáng)生成能力使其能訪問(wèn)外部知識(shí)庫(kù)。RAG 不是替代工作流而是增強(qiáng)其中某個(gè)或某幾個(gè)智能體的能力。例如可以讓“研究員”智能體不僅搜索公開(kāi)網(wǎng)絡(luò)還能檢索內(nèi)部知識(shí)庫(kù)或者讓“分析師”智能體在分析時(shí)結(jié)合檢索到的內(nèi)部資料進(jìn)行交叉驗(yàn)證。5.1 在狀態(tài)流中嵌入 RAG 檢索節(jié)點(diǎn)假設(shè)我們已經(jīng)有一個(gè)構(gòu)建好的向量數(shù)據(jù)庫(kù)例如使用 Chroma、Weaviate 或 Pinecone里面存儲(chǔ)了公司內(nèi)部技術(shù)文檔。我們創(chuàng)建一個(gè)新的節(jié)點(diǎn)retrieve_internal_knowledge并將其插入到工作流中合適的位置。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter # 注意以下為示例你需要預(yù)先準(zhǔn)備好向量庫(kù) # 初始化嵌入模型和向量庫(kù)連接 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(persist_directory./my_chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 檢索最相關(guān)的3條 def retrieve_knowledge_node(state: GraphState) - GraphState: 節(jié)點(diǎn)從內(nèi)部知識(shí)庫(kù)檢索相關(guān)信息。 question state[question] # 執(zhí)行檢索 relevant_docs retriever.invoke(question) # 將檢索到的文檔內(nèi)容存入狀態(tài)供后續(xù)節(jié)點(diǎn)使用 internal_knowledge [doc.page_content for doc in relevant_docs] # 可以單獨(dú)存放也可以合并到之前的 extracted_texts 中 state[internal_knowledge] internal_knowledge state[current_step] analyze # 假設(shè)檢索后直接進(jìn)入分析 return state然后修改工作流圖在搜索公開(kāi)信息后并行或串行地檢索內(nèi)部知識(shí)。# 修改圖在 extract_content 后analyze 前加入知識(shí)檢索 workflow StateGraph(GraphState) workflow.add_node(search_web, search_web_node) workflow.add_node(extract_content, extract_content_node) workflow.add_node(retrieve_knowledge, retrieve_knowledge_node) workflow.add_node(analyze, analyze_node) workflow.add_node(generate_report, generate_report_node) # 線性流程搜索 - 提取 - 檢索 - 分析 - 報(bào)告 workflow.add_edge(search_web, extract_content) workflow.add_edge(extract_content, retrieve_knowledge) workflow.add_edge(retrieve_knowledge, analyze) workflow.add_edge(analyze, generate_report) workflow.add_edge(generate_report, END) workflow.set_entry_point(search_web) app workflow.compile()現(xiàn)在分析師智能體在生成要點(diǎn)時(shí)就可以同時(shí)參考extracted_texts來(lái)自網(wǎng)絡(luò)和internal_knowledge來(lái)自內(nèi)部知識(shí)庫(kù)。我們需要修改analyze_node的提示詞將兩部分信息都整合進(jìn)去。def analyze_node(state: GraphState) - GraphState: texts state.get(extracted_texts, []) internal_knowledge state.get(internal_knowledge, []) question state[question] # 合并上下文 all_context_parts [] if texts: all_context_parts.append(【來(lái)自公開(kāi)網(wǎng)絡(luò)的信息】\n \n\n---\n\n.join(texts[:3])) if internal_knowledge: all_context_parts.append(【來(lái)自內(nèi)部知識(shí)庫(kù)的信息】\n \n\n---\n\n.join(internal_knowledge[:3])) if not all_context_parts: state[key_points] [未找到任何相關(guān)信息。] return state context \n\n.join(all_context_parts) prompt f 你是一位嚴(yán)謹(jǐn)?shù)难芯糠治鰩?。?qǐng)綜合以下來(lái)自公開(kāi)網(wǎng)絡(luò)和內(nèi)部知識(shí)庫(kù)的資料針對(duì)用戶問(wèn)題“{question}”提煉出3-5個(gè)最核心的要點(diǎn)。 注意比較和交叉驗(yàn)證不同來(lái)源的信息。 背景資料 {context} 核心要點(diǎn)每條用‘- ’開(kāi)頭 # ... 后續(xù)調(diào)用LLM的代碼不變 ...通過(guò)這種方式RAG 被無(wú)縫地編織進(jìn)了智能體的決策流程中成為其“長(zhǎng)期記憶”和“領(lǐng)域知識(shí)”的來(lái)源。這比單純用一個(gè) RAG 問(wèn)答鏈要強(qiáng)大得多因?yàn)闄z索到的知識(shí)是在一個(gè)更復(fù)雜、多步驟的推理流程中被使用的。5.2 多模型接入為不同任務(wù)選擇最合適的“大腦”在上面的例子中分析師和撰稿人都用了 GPT-4。但在實(shí)際生產(chǎn)中成本、速度和任務(wù)特性都需要考慮。LangChain/LangGraph 的另一個(gè)優(yōu)勢(shì)是模型無(wú)關(guān)性。你可以輕松地為不同節(jié)點(diǎn)切換不同的模型提供商。例如可以讓“研究員”使用快速且便宜的模型如 GPT-3.5-Turbo來(lái)生成搜索查詢?cè)~讓“分析師”使用能力強(qiáng)、上下文窗口大的模型如 Claude-3-Opus 或 GPT-4進(jìn)行深度分析讓“撰稿人”使用擅長(zhǎng)創(chuàng)意寫(xiě)作的模型比如特定調(diào)優(yōu)過(guò)的模型。只需在初始化節(jié)點(diǎn)時(shí)傳入不同的 LLM 實(shí)例即可。LangChain 的統(tǒng)一接口讓這變得非常簡(jiǎn)單。from langchain_anthropic import ChatAnthropic from langchain_google_genai import ChatGoogleGenerativeAI # 為不同智能體配置不同模型 llm_researcher ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 便宜用于生成查詢?cè)~ llm_analyst ChatAnthropic(modelclaude-3-opus-20240229, temperature0.1) # 能力強(qiáng)用于分析 llm_writer ChatGoogleGenerativeAI(modelgemini-pro, temperature0.7) # 流暢用于寫(xiě)作 # 然后在對(duì)應(yīng)的節(jié)點(diǎn)函數(shù)中使用各自的 llm 實(shí)例這種靈活性使得你可以構(gòu)建一個(gè)“混合模型”的智能體系統(tǒng)在效果、成本和速度之間取得最佳平衡。6. 工業(yè)級(jí)落地的關(guān)鍵可觀測(cè)性、持久化與錯(cuò)誤處理一個(gè)能在生產(chǎn)環(huán)境跑起來(lái)的智能體光有核心邏輯是不夠的。它必須健壯、可監(jiān)控、可調(diào)試。6.1 可觀測(cè)性記錄每一步的輸入輸出LangGraph 提供了Checkpointer和Message等機(jī)制來(lái)持久化狀態(tài)但對(duì)于日志我們通常需要更靈活的方式。一個(gè)簡(jiǎn)單有效的方法是在每個(gè)節(jié)點(diǎn)的函數(shù)內(nèi)部將關(guān)鍵操作和結(jié)果記錄到結(jié)構(gòu)化日志中。import logging import json logger logging.getLogger(__name__) def search_web_node(state: GraphState) - GraphState: question state[question] # 記錄輸入 logger.info(f[search_web_node] 開(kāi)始處理問(wèn)題: {question}, extra{state: state}) # ... 執(zhí)行搜索 ... # 記錄輸出和關(guān)鍵結(jié)果 logger.info(f[search_web_node] 生成查詢?cè)~: {queries}, extra{queries: queries}) logger.info(f[search_web_node] 獲得結(jié)果數(shù): {len(all_results)}, extra{sample_result: all_results[0] if all_results else None}) # 對(duì)于錯(cuò)誤記錄異常 except Exception as e: logger.error(f[search_web_node] 搜索失敗: {e}, exc_infoTrue) # 可以選擇將錯(cuò)誤信息存入state讓后續(xù)節(jié)點(diǎn)或路由函數(shù)處理 state[last_error] str(e) return new_state將這些日志收集到如 ELK、Loki 或 Datadog 等系統(tǒng)中你就能清晰地看到每個(gè)請(qǐng)求在智能體工作流中的完整生命周期便于排查問(wèn)題和分析性能瓶頸。6.2 持久化與斷點(diǎn)續(xù)跑使用 Checkpoints對(duì)于長(zhǎng)時(shí)間運(yùn)行或需要中斷后恢復(fù)的工作流LangGraph 的 Checkpoint 機(jī)制至關(guān)重要。它允許你在每個(gè)節(jié)點(diǎn)執(zhí)行后將完整的GraphState持久化到數(shù)據(jù)庫(kù)如 Redis、PostgreSQL。當(dāng)系統(tǒng)重啟或從故障中恢復(fù)時(shí)可以從上一個(gè)檢查點(diǎn)繼續(xù)執(zhí)行。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 初始化一個(gè) SQLite 檢查點(diǎn)存儲(chǔ)器 memory SqliteSaver.from_conn_string(:memory:) # 生產(chǎn)環(huán)境用文件或網(wǎng)絡(luò)數(shù)據(jù)庫(kù) workflow StateGraph(GraphState, config_schemaMyConfigSchema) # 需要定義ConfigSchema # ... 添加節(jié)點(diǎn)和邊 ... app workflow.compile(checkpointermemory) # 編譯時(shí)傳入 checkpointer # 調(diào)用時(shí)需要傳入一個(gè)線程IDthread_id用于標(biāo)識(shí)同一個(gè)會(huì)話或流程 config {configurable: {thread_id: user_123_session_456}} initial_state GraphState(...) # invoke 會(huì)返回最終狀態(tài)并在后臺(tái)自動(dòng)保存檢查點(diǎn) final_state app.invoke(initial_state, configconfig) # 如果中途中斷可以通過(guò) get_state 獲取最新?tīng)顟B(tài)并繼續(xù) invoke state app.get_state(config) # app.invoke(None, config) 可以從該狀態(tài)繼續(xù)執(zhí)行如果圖支持從中間開(kāi)始6.3 系統(tǒng)化的錯(cuò)誤處理錯(cuò)誤處理不能只靠try...except。需要在狀態(tài)圖和節(jié)點(diǎn)設(shè)計(jì)層面考慮節(jié)點(diǎn)級(jí)重試對(duì)于網(wǎng)絡(luò)請(qǐng)求如搜索、LLM調(diào)用等暫時(shí)性錯(cuò)誤可以在節(jié)點(diǎn)內(nèi)部實(shí)現(xiàn)重試邏輯。工作流級(jí)容錯(cuò)通過(guò)條件邊將錯(cuò)誤信息state[“l(fā)ast_error”]作為路由依據(jù)。例如如果搜索節(jié)點(diǎn)連續(xù)失敗可以路由到一個(gè)“人工降級(jí)”節(jié)點(diǎn)或直接生成一個(gè)包含錯(cuò)誤說(shuō)明的最終報(bào)告。超時(shí)控制為耗時(shí)長(zhǎng)的節(jié)點(diǎn)如 LLM 調(diào)用設(shè)置超時(shí)防止整個(gè)工作流卡死。這通常需要結(jié)合異步和外部超時(shí)機(jī)制。降級(jí)方案當(dāng)核心組件如某個(gè) LLM API不可用時(shí)是否有備選模型或規(guī)則兜底7. 總結(jié)從項(xiàng)目到平臺(tái)的思維轉(zhuǎn)變通過(guò)這個(gè)完整的實(shí)戰(zhàn)拆解我們可以看到使用 LangGraph LangChain 構(gòu)建工業(yè)級(jí) Agent遠(yuǎn)不止是調(diào)用 API 的簡(jiǎn)單疊加。它要求我們完成一次思維轉(zhuǎn)變從編寫(xiě)線性腳本到設(shè)計(jì)狀態(tài)化工作流思考的不再是一行行代碼而是由節(jié)點(diǎn)、邊和狀態(tài)構(gòu)成的可視化流程圖。從打造“全能模型”到構(gòu)建“協(xié)作團(tuán)隊(duì)”將復(fù)雜任務(wù)拆解由多個(gè)職責(zé)單一的智能體節(jié)點(diǎn)通過(guò)狀態(tài)共享協(xié)同完成系統(tǒng)更健壯也更易優(yōu)化。從單一模型調(diào)用到混合模型編排根據(jù)子任務(wù)的特點(diǎn)靈活選用不同能力、成本、速度的模型實(shí)現(xiàn)性價(jià)比最大化。從演示原型到可觀測(cè)系統(tǒng)將日志、檢查點(diǎn)、錯(cuò)誤處理、降級(jí)方案作為一等公民來(lái)設(shè)計(jì)確保系統(tǒng)在生產(chǎn)環(huán)境中可靠運(yùn)行。最終你構(gòu)建的不是一個(gè)“智能體”而是一個(gè)智能體框架或平臺(tái)。新的任務(wù)可以通過(guò)設(shè)計(jì)新的狀態(tài)圖和組裝不同的節(jié)點(diǎn)復(fù)用或新建來(lái)快速實(shí)現(xiàn)。這才是 LangGraph 帶來(lái)的長(zhǎng)期價(jià)值它提供了一套用于構(gòu)建復(fù)雜、可靠、可維護(hù)的 AI 應(yīng)用的標(biāo)準(zhǔn)范式和解耦架構(gòu)。當(dāng)你下次再面對(duì)一個(gè)需要多步驟、多判斷、多數(shù)據(jù)源的自動(dòng)化任務(wù)時(shí)不妨先拿起筆畫(huà)一畫(huà)它的狀態(tài)轉(zhuǎn)移圖。你會(huì)發(fā)現(xiàn)很多復(fù)雜性問(wèn)題在清晰的架構(gòu)面前都會(huì)迎刃而解。