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