LangGraph核心概念詳解節(jié)點(diǎn)、邊、狀態(tài)、條件邊上一篇用一個(gè)例子展示了LangGraph的能力。這一篇我們把核心概念拆開講清楚。LangGraph的四個(gè)核心概念是節(jié)點(diǎn)、邊、狀態(tài)、條件邊。理解了這四個(gè)LangGraph就入門了。狀態(tài)State狀態(tài)是LangGraph最核心的概念。整個(gè)圖的執(zhí)行過程中所有節(jié)點(diǎn)共享一個(gè)狀態(tài)對(duì)象。狀態(tài)通常用TypedDict來定義。每個(gè)字段代表一種信息。fromtypingimportTypedDict,List,OptionalclassAgentState(TypedDict):messages:List[str]# 對(duì)話歷史current_task:str# 當(dāng)前任務(wù)results:List[str]# 中間結(jié)果retry_count:int# 重試次數(shù)status:str# 當(dāng)前狀態(tài)每個(gè)節(jié)點(diǎn)函數(shù)接收狀態(tài)作為參數(shù)返回一個(gè)字典。返回的字典會(huì)合并到當(dāng)前狀態(tài)里。defsearch_node(state:AgentState)-dict:# 從狀態(tài)里讀取信息taskstate[current_task]# 執(zhí)行搜索resultf搜索{task}的結(jié)果# 返回要更新的字段return{results:state[results][result],status:searched,}這里有個(gè)重要的設(shè)計(jì)。節(jié)點(diǎn)返回的不是完整的新狀態(tài)只是要更新的字段。LangGraph會(huì)自動(dòng)把返回值合并到狀態(tài)里。沒返回的字段保持不變。這種設(shè)計(jì)讓節(jié)點(diǎn)之間的耦合度很低。每個(gè)節(jié)點(diǎn)只關(guān)心自己需要讀什么、寫什么不用管其他節(jié)點(diǎn)。節(jié)點(diǎn)Node節(jié)點(diǎn)是圖里的處理步驟。每個(gè)節(jié)點(diǎn)是一個(gè)函數(shù)接收狀態(tài)返回狀態(tài)更新。defanalyze_node(state:AgentState)-dict:resultsstate[results]# 分析結(jié)果analysisf分析了{(lán)len(results)}條結(jié)果return{messages:state[messages][analysis],status:analyzed,}節(jié)點(diǎn)的函數(shù)簽名是固定的。參數(shù)是當(dāng)前狀態(tài)返回值是狀態(tài)更新字典。添加節(jié)點(diǎn)用add_node方法。第一個(gè)參數(shù)是節(jié)點(diǎn)名字第二個(gè)參數(shù)是節(jié)點(diǎn)函數(shù)。fromlanggraph.graphimportStateGraph workflowStateGraph(AgentState)workflow.add_node(search,search_node)workflow.add_node(analyze,analyze_node)workflow.add_node(report,report_node)節(jié)點(diǎn)名字是字符串標(biāo)識(shí)后面添加邊的時(shí)候用這個(gè)名字來引用。邊Edge邊定義了節(jié)點(diǎn)之間的跳轉(zhuǎn)關(guān)系。從節(jié)點(diǎn)A執(zhí)行完后走向節(jié)點(diǎn)B。# search執(zhí)行完后走向analyzeworkflow.add_edge(search,analyze)# analyze執(zhí)行完后走向reportworkflow.add_edge(analyze,report)普通的邊是無條件的。A執(zhí)行完一定走向B。還有一個(gè)特殊的邊入口邊。用set_entry_point來設(shè)置圖的起始節(jié)點(diǎn)。# 圖從search節(jié)點(diǎn)開始workflow.set_entry_point(search)還有一個(gè)特殊的終止節(jié)點(diǎn)END。走到END就表示圖執(zhí)行結(jié)束。# report執(zhí)行完后結(jié)束fromlanggraph.graphimportEND workflow.add_edge(report,END)條件邊Conditional Edge條件邊是LangGraph最強(qiáng)大的功能。它允許根據(jù)當(dāng)前狀態(tài)動(dòng)態(tài)決定下一步走向哪個(gè)節(jié)點(diǎn)。defdecide_next(state:AgentState)-str:statusstate[status]ifstatusanalyzed:returnreportelifstatusneed_more_data:returnsearchelse:returnerror_handler# 添加條件邊workflow.add_conditional_edges(analyze,# 從哪個(gè)節(jié)點(diǎn)出發(fā)decide_next,# 判斷函數(shù))條件邊的判斷函數(shù)接收狀態(tài)返回一個(gè)字符串。這個(gè)字符串就是下一步要走向的節(jié)點(diǎn)名字。判斷函數(shù)里可以寫任意復(fù)雜的邏輯。讀狀態(tài)、做判斷、返回目標(biāo)節(jié)點(diǎn)。條件邊是實(shí)現(xiàn)分支和循環(huán)的關(guān)鍵。分支就是條件邊返回不同的節(jié)點(diǎn)名字。循環(huán)就是條件邊返回之前的節(jié)點(diǎn)。一個(gè)完整的例子把四個(gè)概念串起來做一個(gè)完整的Agent工作流。fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List# 1. 定義狀態(tài)classWorkState(TypedDict):question:strsearch_results:List[str]answer:strneeds_search:boolquality_ok:boolretry_count:int# 2. 定義節(jié)點(diǎn)defdecide_node(state:WorkState)-dict:判斷是否需要搜索questionstate[question]# 簡單問題不需要搜索iflen(question)10:return{needs_search:False}return{needs_search:True}defsearch_node(state:WorkState)-dict:搜索資料questionstate[question]resultf關(guān)于{question}的搜索結(jié)果resultsstate.get(search_results,[])[result]return{search_results:results}defanswer_node(state:WorkState)-dict:生成回答questionstate[question]resultsstate.get(search_results,[])ifresults:answerf根據(jù){len(results)}條資料回答{question}的答案是...else:answerf直接回答{question}的答案是...return{answer:answer}defcheck_node(state:WorkState)-dict:檢查回答質(zhì)量answerstate[answer]retrystate.get(retry_count,0)# 模擬質(zhì)量檢查iflen(answer)15orretry2:return{quality_ok:True}return{quality_ok:False,retry_count:retry1}# 3. 定義條件判斷函數(shù)defroute_after_decide(state:WorkState)-str:ifstate[needs_search]:returnsearchreturnanswerdefroute_after_check(state:WorkState)-str:ifstate[quality_ok]:returnENDreturnanswer# 質(zhì)量不行重新回答# 4. 構(gòu)建圖workflowStateGraph(WorkState)workflow.add_node(decide,decide_node)workflow.add_node(search,search_node)workflow.add_node(answer,answer_node)workflow.add_node(check,check_node)workflow.set_entry_point(decide)workflow.add_conditional_edges(decide,route_after_decide)workflow.add_edge(search,answer)workflow.add_edge(answer,check)workflow.add_conditional_edges(check,route_after_check)# 5. 編譯并運(yùn)行appworkflow.compile()resultapp.invoke({question:什么是RAG檢索增強(qiáng)生成技術(shù),search_results:[],answer:,needs_search:False,quality_ok:False,retry_count:0,})print(f回答:{result[answer]})print(f搜索次數(shù):{len(result.get(search_results,[]))})print(f重試次數(shù):{result.get(retry_count,0)})這個(gè)工作流的流程是。先判斷要不要搜索要搜就搜搜完生成回答。不要搜就直接生成回答。生成后檢查質(zhì)量不行就重來。編譯和運(yùn)行圖構(gòu)建好了以后需要編譯才能運(yùn)行。編譯會(huì)做一些優(yōu)化和驗(yàn)證。appworkflow.compile()運(yùn)行有兩種方式。invoke是同步運(yùn)行等全部執(zhí)行完返回最終狀態(tài)。resultapp.invoke(initial_state)stream是流式運(yùn)行每執(zhí)行一個(gè)節(jié)點(diǎn)就返回一次中間狀態(tài)。適合前端展示執(zhí)行進(jìn)度。foreventinapp.stream(initial_state):print(event)幾個(gè)實(shí)用技巧限制循環(huán)次數(shù)。條件邊可能導(dǎo)致無限循環(huán)。在狀態(tài)里加一個(gè)計(jì)數(shù)器超過次數(shù)就強(qiáng)制結(jié)束。defroute_after_check(state:WorkState)-str:ifstate[quality_ok]orstate.get(retry_count,0)3:returnENDreturnanswer用Annotated減少代碼。如果狀態(tài)字段是列表每次都要手動(dòng)拼接。用Annotated和reducer可以簡化。fromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassState(TypedDict):messages:Annotated[list,add_messages]# 自動(dòng)追加而不是覆蓋保存中間狀態(tài)。用stream可以看到每一步的中間狀態(tài)方便調(diào)試。下一篇講狀態(tài)管理實(shí)戰(zhàn)。在多輪交互中怎么維護(hù)Agent的狀態(tài)怎么處理復(fù)雜的對(duì)話上下文。