綱要為什么要加入人工審核環(huán)節(jié)模型的不確定性高風險場景下的安全需求LangGraph 人機交互機制interrupt()在節(jié)點中暫停執(zhí)行Command(resume...)恢復執(zhí)行并傳遞人類反饋interrupt_before/interrupt_after在邊處打斷update_state直接修改圖的狀態(tài)實戰(zhàn)一等待用戶輸入構建帶反饋節(jié)點的簡單圖使用interrupt暫停Command恢復實戰(zhàn)二審查工具調用在工具執(zhí)行前插入審查節(jié)點支持批準、拒絕、修改參數實戰(zhàn)三編輯圖的狀態(tài)使用interrupt_before在特定節(jié)點前暫停通過graph.update_state()修改狀態(tài)后繼續(xù)完整可運行代碼總結與相關度說明為什么需要人工介入當前的大語言模型基于概率生成即使是最先進的模型也可能產生錯誤決策。在關鍵業(yè)務場景如金融交易、醫(yī)療建議、自動發(fā)布內容中一次錯誤的工具調用或回復可能造成嚴重影響。因此在生產級智能體系統(tǒng)中引入人機交互環(huán)節(jié)讓人類能夠審批、編輯甚至拒絕智能體的決策是保障系統(tǒng)可靠性的重要手段。LangGraph 為此提供了原語級的支持可以靈活地將人工審核嵌入到工作流的任意位置。核心機制interrupt與CommandLangGraph 的人機交互依賴于兩個核心 APIinterrupt()在節(jié)點內部調用暫停圖的執(zhí)行并拋出一個需要人類處理的中斷事件。Command(resume...)外部輸入反饋后通過Command恢復執(zhí)行同時可以將人類提供的數據注入到圖中。此外還可以在編譯圖時通過interrupt_before或interrupt_after參數在指定節(jié)點執(zhí)行前/后自動暫停無需在節(jié)點內部寫interrupt()。實戰(zhàn)一等待用戶輸入以下示例演示一個簡單的工作流step_one→human_feedback→step_three。在human_feedback節(jié)點中調用interrupt()讓人類提供反饋然后繼續(xù)執(zhí)行。importosfromtypingimportTypedDictfromdotenvimportload_dotenvfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterrupt,Command load_dotenv()# 定義狀態(tài)classFeedbackState(TypedDict):input:struser_feedback:str# 節(jié)點1不做具體處理只打印日志defstep_one(state:FeedbackState)-FeedbackState:print( 執(zhí)行 Step 1)returnstate# 節(jié)點2等待人類反饋defhuman_feedback(state:FeedbackState)-FeedbackState:print( 等待人類反饋...)# interrupt 暫停執(zhí)行提示信息會返回給調用方feedbackinterrupt(請?zhí)峁┠愕姆答?# 當外部通過 Command(resume...) 恢復時feedback 會被賦值為傳入的內容return{user_feedback:feedback}# 節(jié)點3使用反饋defstep_three(state:FeedbackState)-FeedbackState:print(f 執(zhí)行 Step 3收到的反饋{state[user_feedback]})returnstate# 構建圖builderStateGraph(FeedbackState)builder.add_node(step_one,step_one)builder.add_node(human_feedback,human_feedback)builder.add_node(step_three,step_three)builder.set_entry_point(step_one)builder.add_edge(step_one,human_feedback)builder.add_edge(human_feedback,step_three)builder.add_edge(step_three,END)# 激活短期記憶持久化用于中斷恢復memoryMemorySaver()appbuilder.compile(checkpointermemory)# 首次調用會中斷在 human_feedback 節(jié)點config{configurable:{thread_id:feedback-session-1}}input_data{input:你好}print( 首次調用會中斷)foreventinapp.stream(input_data,config):fornode_name,valueinevent.items():ifisinstance(value,dict)anduser_feedbackinvalue:print(f節(jié)點{node_name}: 收到反饋 {value[user_feedback]})# 此時執(zhí)行暫停我們需要獲取中斷事件并恢復# 實際上 stream 會拋出中斷信息這里為簡化演示直接使用 invoke 配合 Command 恢復# 方法使用 app.invoke 并傳入 Command(resume...) 作為 inputprint(\n 恢復執(zhí)行提供反饋)# 恢復時需要傳入同一個 thread_id并用 Command 包裹 human 的反饋resume_inputCommand(resume我覺得很好繼續(xù)吧)foreventinapp.stream(resume_input,config):fornode_name,valueinevent.items():ifisinstance(value,dict)anduser_feedbackinvalue:print(f節(jié)點{node_name}: 最終反饋 {value[user_feedback]})運行這段代碼你會看到第一次調用在human_feedback處暫停然后通過Command(resume...)繼續(xù)執(zhí)行并最終在step_three看到反饋內容。實戰(zhàn)二審查工具調用這是智能體最常用的場景在工具執(zhí)行前暫停讓人類審批工具的名稱和參數。如果審批通過執(zhí)行工具如果拒絕修改參數或終止流程。以下示例構建一個簡單的天氣查詢智能體工具調用前需要人類審查。importosfromtypingimportTypedDict,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage,ToolMessagefromlanggraph.graphimportStateGraph,END,MessagesStatefromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.typesimportinterrupt,Command load_dotenv()# 定義一個簡單的天氣工具模擬defget_weather(city:str)-str:查詢指定城市的天氣# 模擬返回固定天氣returnf{city}天氣晴朗25°Ctools[get_weather]# 綁定工具的LLMllmChatOpenAI(modelgpt-3.5-turbo,temperature0)llm_with_toolsllm.bind_tools(tools)# 定義狀態(tài)classAgentState(MessagesState):pass# 模型調用節(jié)點defcall_model(state:AgentState)-AgentState:responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 人類審查節(jié)點在工具執(zhí)行前調用defhuman_review_node(state:AgentState)-AgentState:# 獲取最后一條消息通常包含 tool_callslast_msgstate[messages][-1]ifnothasattr(last_msg,tool_calls)ornotlast_msg.tool_calls:returnstate# 無工具調用則跳過# 展示待審查的工具調用print( 需要審查的工具調用)forcallinlast_msg.tool_calls:print(f 工具:{call[name]}, 參數:{call[args]})# interrupt 等待人類決策返回字典格式的決策decisioninterrupt(請審批輸入 continue 批準或提供修改后的參數 JSON)# decision 可以是字符串也可以是包含了修改后參數的對象# 這里為了簡單只支持批準字符串 continueifdecisioncontinue:returnstateelse:# 其他情況可以拒絕或修改示例中我們簡單拒絕print( 審查未通過終止執(zhí)行)# 通過返回空消息終止流程實際可添加 ToolMessage 表示拒絕return{messages:[ToolMessage(content審查未通過,tool_call_idlast_msg.tool_calls[0][id])]}# 構建圖builderStateGraph(AgentState)builder.add_node(call_model,call_model)builder.add_node(review,human_review_node)builder.add_node(tools,ToolNode(tools))builder.set_entry_point(call_model)# 條件邊模型調用后如果有工具調用進入審查否則直接結束defshould_review(state:AgentState)-str:iftools_condition(state):returnreviewreturnEND builder.add_conditional_edges(call_model,should_review,{review:review,END:END})builder.add_edge(review,tools)builder.add_edge(tools,call_model)# 工具執(zhí)行后回到模型繼續(xù)思考memoryMemorySaver()appbuilder.compile(checkpointermemory,interrupt_before[review])# 在 review 節(jié)點前中斷# 測試詢問天氣會觸發(fā)工具調用和審查config{configurable:{thread_id:review-tools-1}}input_data{messages:[HumanMessage(content北京天氣怎么樣)]}print( 開始對話將中斷在審查前)# 第一次調用會暫停在 review 節(jié)點前foreventinapp.stream(input_data,config):pass# 恢復批準工具調用print(\n 恢復批準工具調用 )resume_cmdCommand(resumecontinue)foreventinapp.stream(resume_cmd,config):fornode_name,valueinevent.items():ifmessagesinvalueandvalue[messages]:print(f節(jié)點{node_name}:{value[messages][-1].content})運行后你會看到工具調用被暫停你可以選擇批準輸入continue或修改。這里為了簡化只支持批準。實際可以解析 JSON 來更新參數。實戰(zhàn)三編輯圖的狀態(tài)有時我們不希望僅僅在節(jié)點內部暫停而是想在圖的任意位置暫停并且可以修改狀態(tài)例如修改之前某個節(jié)點產生的數據后再繼續(xù)。使用interrupt_before在指定節(jié)點前暫停然后使用graph.update_state()直接修改狀態(tài)最后再用Command或stream繼續(xù)。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportCommandclassEditState(TypedDict):text:strdefstep_one(state:EditState)-EditState:# 將輸入大寫return{text:state[text].upper()}defstep_two(state:EditState)-EditState:# 添加后綴return{text:state[text] [processed]}defstep_three(state:EditState)-EditState:# 最終輸出return{text:最終結果: state[text]}builderStateGraph(EditState)builder.add_node(step_one,step_one)builder.add_node(step_two,step_two)builder.add_node(step_three,step_three)builder.set_entry_point(step_one)builder.add_edge(step_one,step_two)builder.add_edge(step_two,step_three)builder.add_edge(step_three,END)memoryMemorySaver()appbuilder.compile(checkpointermemory,interrupt_before[step_two])# 在 step_two 前暫停config{configurable:{thread_id:edit-state-1}}# 初始調用input1{text:hello world}print( 初始調用暫停在 step_two 前)eventslist(app.stream(input1,config))# 此時狀態(tài)為 step_one 已執(zhí)行text 為 HELLO WORLDprint(當前狀態(tài),app.get_state(config).values)# 我們想要修改這個狀態(tài)比如把 text 改成其他內容print(\n 更新狀態(tài) )app.update_state(config,{text:CUSTOM TEXT})print(更新后狀態(tài),app.get_state(config).values)# 繼續(xù)執(zhí)行print(\n 繼續(xù)執(zhí)行 )foreventinapp.stream(None,config):fornode_name,valueinevent.items():print(f節(jié)點{node_name}:{value})運行后你會發(fā)現step_two接收到的text已經被改成了CUSTOM TEXT之后的結果也是基于修改后的值。這種能力允許人類在流程中任意點修正數據再繼續(xù)執(zhí)行?？偨Y人機交互是 LangGraph 為生產級 AI 應用提供的關鍵能力。通過interrupt、Command、interrupt_before和update_state開發(fā)者可以靈活地在工作流中插入審批、修改狀態(tài)、拒絕工具調用等操作從而在不完全信任模型決策時確保系統(tǒng)安全可控。本文的三個實戰(zhàn)示例給出了最基礎的用法實際項目中可以組合出更復雜的審核流程。本文完整覆蓋了人機交互的概念、等待用戶輸入、審查工具調用、編輯圖狀態(tài)等所有演示場景并提供了可直接運行的代碼示例。代碼已可直接運行需安裝 langgraph, langchain-openai 等依賴并配置 API Key。