LangGraph核心组件-人机交互

张

张建站

2026/5/14 22:19:49

10分钟阅读

人机交互介绍在LangGraph框架中人机交互作为核心组件主要通过以下机制实现动态协作交互流程设计中断触发机制系统在以下场景自动暂停流程并请求人工介入关键决策节点例如置信度低于阈值异常状态检测如循环次数超过 max预设的人工审核点多模态输入接口支持人类通过# 伪代码示例 def human_feedback_handler(): input_modes [文本, 语音, 标注工具] return integrate_multimodal(input_modes)状态管理采用双向状态同步机制典型应用场景场景类型人类介入方式系统响应内容审核修正标注重新训练局部模型知识补充注入新数据更新知识图谱 $ G(V,E) $流程控制调整权重参数动态优化路径规划基本运用典型应用场景任务型对话系统例如订票、客服咨询通过 langgraph 逐步引导用户提供必要信息如时间、地点并实时验证完整性。多分支决策交互在游戏叙事、教育问答等场景中根据用户选择进入不同剧情分支或知识点讲解路径。混合主动对话系统可在用户未提问时主动推送信息如促销提醒再通过图中预设的“中断恢复节点”回到主流程。与传统方法的对比方法灵活性可维护性复杂度管理硬编码规则低差困难纯机器学习模型高中等依赖数据langgraph中高优秀可视化核心价值可视化开发图结构直观展示对话脉络降低设计门槛。模块化复用通用节点如身份验证可跨项目复用。可控性与扩展性新增分支仅需扩展子图无需重构整体逻辑。等待用户数据代码示例from pydantic import BaseModel from langgraph.graph import StateGraph ,START,END from langgraph.types import Command,interrupt from langgraph.checkpoint.memory import MemorySaver from IPython.display import Image,display class State(BaseModel): input:str user_feedback:str def step_1 (state:State): print(---step 1---) pass def human_feedback(state:State): print(---human_feedBack---) feedbackinterrupt(please provide feedback) return {user_feedback:feedback} def step_3 (state:State): print(---step 3---) pass builderStateGraph(State) builder.add_node(step_1) builder.add_node(step_3) builder.add_node(human_feedback,human_feedback) builder.add_edge(START,step_1) builder.add_edge(step_1,human_feedback) builder.add_edge(human_feedback,step_3) builder.add_edge(step_3,END) memoryMemorySaver() graphbuilder.compile(checkpointer memory) # display(Image(graph.get_graph().draw_mermaid_png())) init_inputState(input你好) config{configurable:{thread_id:1}} for e in graph.stream(init_input,config,stream_modeupdates): print(e) print(\n)示例输出添加人类组件 command 来让其继续执行 for event in graph.stream( Command(resumego to step_3), config, stream_modeupdates, ): print(event) print(\n)示例输出在智能体中引入人工介入代码示例from langchain_core.messages import HumanMessage from langgraph.graph import MessagesState, START from langgraph.types import Command,interrupt # 设置工具 # 我们将有一个真实工具 - 搜索工具 # 我们还将有一个假工具 - ask_human询问人类工具 # 这里我们定义任何实际工具 from langchain_core.tools import tool from langgraph.prebuilt import ToolNode tool def search(query: str): 调用此函数来浏览网络。 # 这只是实际实现的占位符 # 不要让大语言模型知道这一点哦 return f我查询了{query}。结果北京天气晴朗温度25度。 tools [search] tool_node ToolNode(tools) # 设置模型 from langchain_deepseek import ChatDeepSeek import os model ChatDeepSeek( modeldeepseek-chat, api_keyos.environ.get(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, temperature0.8 ) from pydantic import BaseModel # 我们将绑定所有工具到模型 # 我们有上面的实际工具但我们还需要一个模拟工具来询问人类 # 由于bind_tools接受工具但也接受工具定义 # 我们可以为ask_human定义一个工具定义 class AskHuman(BaseModel): 向人类提问 question: str model model.bind_tools(tools [AskHuman]) # 定义节点和条件边 # 定义确定是否继续的函数 def should_continue(state): messages state[messages] last_message messages[-1] # 如果没有函数调用则结束 if not last_message.tool_calls: return END # 如果工具调用是询问人类我们返回该节点 # 你也可以在这里添加逻辑让某些系统知道有需要人类输入的内容 elif last_message.tool_calls[0][name] AskHuman: return ask_human # 否则如果有函数调用我们继续 else: return action # 定义调用模型的函数 def call_model(state): messages state[messages] system_prompt 你是一个智能助手。当需要获取用户的位置信息时必须使用 AskHuman 工具来询问用户。不要直接用自然语言询问而要调用 AskHuman 工具。 # 在消息列表开头添加系统提示 enhanced_messages [{role: system, content: system_prompt}] messages response model.invoke(enhanced_messages) # 我们返回一个列表因为这将被添加到现有列表中 return {messages: [response]} # 我们定义一个假节点来询问人类 def ask_human(state): tool_call_id state[messages][-1].tool_calls[0][id] ask AskHuman.model_validate(state[messages][-1].tool_calls[0][args]) location interrupt(ask.question) from langchain_core.messages import ToolMessage tool_message ToolMessage(contentlocation, tool_call_idtool_call_id) return {messages: [tool_message]} # 构建图 from langgraph.graph import END, StateGraph # 定义一个新图 workflow StateGraph(MessagesState) workflow.add_node(agent,call_model) workflow.add_node(action,tool_node) workflow.add_node(ask_human,ask_human) workflow.add_edge(START, agent) workflow.add_conditional_edges(agent, should_continue,[action, ask_human, END]) workflow.add_edge(action, agent) workflow.add_edge(ask_human, agent) from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph workflow.compile(checkpointermemory) # display(Image(graph.get_graph().draw_mermaid_png())) # 1. 首次执行 - 会中断在 ask_human 节点 config {configurable: {thread_id: 2}} for event in graph.stream( MessagesState(messages[HumanMessage(content询问用户他们在哪里然后再去查询那里的天气)]), config, stream_modevalues, ): event[messages][-1].pretty_print() # 2. 恢复执行 - 提供用户反馈 for event in graph.stream( Command(resume北京), # 用户提供的答案 config, stream_modevalues, ): event[messages][-1].pretty_print()审查编写代码示例审查编写 from langchain_deepseek import ChatDeepSeek from typing import Literal from langgraph.types import Command, interrupt from langgraph.graph import END from dotenv import load_dotenv from langgraph.graph import MessagesState,StateGraph,START,END from langgraph.checkpoint.memory import MemorySaver import os from langchain_core.tools import tool from IPython.display import Image,display load_dotenv() model ChatDeepSeek( modeldeepseek-chat, api_keyos.environ.get(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com, temperature0.8 ) class State(MessagesState): input: str # - 这里导致了报错 def call_llm(state): model_with_tools model.bind_tools([weather_search]) response model_with_tools.invoke(state[messages]) return {messages: [response]} #定义天气查询函数 tool(description用于查询天气的函数。) def weather_search(city: str) - str: return f{city} 的天气晴朗温度25度。 # 1. 定义路由函数判断是否需要人工审核 def route_after_llm(state) - Literal[END, human_review_node]: 如果最后一条消息包含工具调用则路由到人工审核节点否则结束流程。 if len(state[messages][-1].tool_calls) 0: return END else: return human_review_node # 2. 定义人工审核节点 def human_review_node(state) - Command[Literal[call_llm, run_tool]]: last_message state[messages][-1] # 获取最新的工具调用请求 tool_call last_message.tool_calls[-1] # 这是我们将通过 Command(resumehuman_review) 提供的值 # 流程会在这里暂停直到人类输入反馈 human_review interrupt({ question: 这是正确的吗, # 显示工具调用以供审核 tool_call: tool_call, }) review_action human_review[action] review_data human_review.get(data) # 分支 1: 如果批准直接调用工具 if review_action continue: return Command(gotorun_tool) # 分支 2: 更新 AI 消息中的参数并调用工具 elif review_action update: updated_message { role: ai, content: last_message.content, tool_calls: [ { id: tool_call[id], name: tool_call[name], # 这是人类提供的更新后的参数 args: review_data, } ], # 这很重要 - 这需要与你替换的消息 ID 相同 # 否则它将显示为一个单独的消息追加而不是覆盖 id: last_message.id, } # 使用 update 参数覆盖旧消息并跳转到 run_tool return Command(gotorun_tool, update{messages: [updated_message]}) # 分支 3: 向 LLM 提供反馈不执行工具而是让 LLM 重新思考 elif review_action feedback: # 注意我们将反馈消息添加为 ToolMessage # 以保持消息历史中的正确顺序 # (带有工具调用的 AI 消息需要后跟工具调用消息) tool_message { role: tool, # 这是我们的自然语言反馈 content: review_data, name: tool_call[name], tool_call_id: tool_call[id], } # 跳转回 call_llm让模型看到反馈后重新生成 return Command(gotocall_llm, update{messages: [tool_message]}) # 3. 定义实际执行工具的节点 def run_tool(state): new_messages [] # 假设 weather_search 是已定义的工具函数 tools {weather_search: weather_search} tool_calls state[messages][-1].tool_calls for tool_call in tool_calls: tool tools[tool_call[name]] # 执行工具 result tool.invoke(tool_call[args]) new_messages.append({ role: tool, name: tool_call[name], content: result, tool_call_id: tool_call[id], }) return {messages: new_messages} # 1. 初始化状态图构建器 # 假设 State 已经在前面定义好例如继承自 MessagesState builder StateGraph(State) # 2. 添加节点 builder.add_node(call_llm) # LLM 调用节点 builder.add_node(run_tool) # 工具执行节点 builder.add_node(human_review_node) # 人工审核节点 # 3. 定义边流程控制 # 入口从 START 开始首先调用 LLM builder.add_edge(START, call_llm) # 条件路由在 LLM 调用后根据是否有工具调用以及是否需要人工介入来决定去向 # route_after_llm 函数会返回 run_tool, human_review_node 或 END builder.add_conditional_edges(call_llm, route_after_llm) # 闭环工具执行完成后回到 LLM 节点让模型根据工具返回的结果生成最终回复 builder.add_edge(run_tool, call_llm) # 4. 设置持久化必须步骤 # 要实现 interrupt 中断和 Command 恢复功能必须配置 checkpointer memory MemorySaver() # 5. 编译图 graph builder.compile(checkpointermemory) # 6. 可视化查看 # display(Image(graph.get_graph().draw_mermaid_png()))from langchain_core.messages import HumanMessage #不涉及工具调用 config {configurable: {thread_id: test_1}} initial_input {messages: [HumanMessage(content你好)]} # 使用 stream 观察每一步的输出 for event in graph.stream(initial_input, config): print(event)人机交互-对图状态进行编辑from pydantic import BaseModel from langgraph.graph import StateGraph ,START,END from langgraph.types import Command,interrupt from langgraph.checkpoint.memory import MemorySaver from IPython.display import Image,display class State(BaseModel): input:str user_feedback:str def step_1 (state:State): print(---step 1---) pass def human_feedback(state:State): print(---human_feedBack---) def step_3 (state:State): print(---step 3---) pass builderStateGraph(State) builder.add_node(step_1) builder.add_node(step_3) builder.add_node(human_feedback,human_feedback) builder.add_edge(START,step_1) builder.add_edge(step_1,human_feedback) builder.add_edge(human_feedback,step_3) builder.add_edge(step_3,END) memoryMemorySaver() graphbuilder.compile(checkpointer memory,interrupt_before[human_feedback]) # display(Image(graph.get_graph().draw_mermaid_png())) init_inputState(input你好) config{configurable:{thread_id:1}} for e in graph.stream(init_input,config, stream_modevalues): print(e) print(\n)