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