LangGraph 进阶实战：从人工介入到时间旅行的智能交互系统构建

作者：佑瞻
在开发智能交互系统时，我们常常会遇到这样的挑战：当智能体做出错误决策时如何及时干预？复杂业务流程中如何实现状态的灵活定制？用户想要回溯历史对话路径时又该如何处理？今天我们就来聊聊 LangGraph 如何通过 "人在回路控制"、"自定义状态" 和 "时间旅行" 三大特性，帮助我们构建更可靠、更灵活的智能交互系统。
一、人在回路控制：让智能体接受人工指导

1. 为什么需要人工介入？

在实际应用中，智能体难免会遇到以下情况：

处理敏感信息时需要人工审核遇到复杂问题时决策质量下降执行关键操作前需要权限确认

LangGraph 通过interrupt机制实现了优雅的人工介入流程，让系统在需要时暂停并等待人工输入。
2. 实现人工介入工具

首先我们需要扩展工具集，添加human_assistance工具：
python

1. from langchain_core.tools import tool
2. from langgraph.types import Command, interrupt
4. @tool
5. def human_assistance(query: str) -> str:
6.     """请求人工协助的工具"""
7.     human_response = interrupt({"query": query})
8.     return human_response["data"]

复制代码

这个工具的核心是interrupt函数，它会暂停执行并等待人工输入，类似 Python 的input()但具备持久化能力。
3. 集成到状态图

接下来将工具集成到 StateGraph 中：
python

1. from langgraph.prebuilt import ToolNode, tools_condition
3. # 定义状态结构
4. class State(TypedDict):
5.     messages: list
7. graph_builder = StateGraph(State)
9. # 初始化工具和LLM
10. tool = TavilySearch(max_results=2)
11. tools = [tool, human_assistance]
12. llm_with_tools = llm.bind_tools(tools)
14. # 添加聊天节点
15. def chatbot(state: State):
16.     message = llm_with_tools.invoke(state["messages"])
17.     assert len(message.tool_calls) <= 1
18.     return {"messages": [message]}
19. graph_builder.add_node("chatbot", chatbot)
21. # 添加工具节点
22. tool_node = ToolNode(tools=tools)
23. graph_builder.add_node("tools", tool_node)
25. # 设置条件路由
26. graph_builder.add_conditional_edges("chatbot", tools_condition)
27. graph_builder.add_edge("tools", "chatbot")
28. graph_builder.add_edge(START, "chatbot")
30. # 编译时启用检查点
31. memory = MemorySaver()
32. graph = graph_builder.compile(checkpointer=memory)

复制代码

4. 人工介入实战

当用户请求专家指导时：
python

1. user_input = "我需要构建AI代理的专家指导，你能帮我请求协助吗？"
2. config = {"configurable": {"thread_id": "1"}}
4. # 触发工具调用
5. events = graph.stream({"messages": [{"role": "user", "content": user_input}]}, config)
6. for event in events:
7.     if "messages" in event:
8.         event["messages"][-1].pretty_print()

复制代码

系统会生成工具调用并暂停，此时可以通过 Command 恢复执行：
python

1. # 人工响应
2. human_response = "建议使用LangGraph构建代理，它比简单智能体更可靠"
3. human_command = Command(resume={"data": human_response})
5. # 恢复执行
6. events = graph.stream(human_command, config)
7. for event in events:
8.     if "messages" in event:
9.         event["messages"][-1].pretty_print()

复制代码

执行后智能体会结合人工建议生成回答，整个过程的状态变化可以通过 LangSmith 追踪查看。
二、自定义状态：构建复杂业务流程

1. 为什么需要自定义状态？

基础消息列表难以满足以下需求：

多轮对话中的结构化数据存储跨节点的状态共享持久化层的复杂数据管理

LangGraph 允许我们向状态中添加自定义字段，实现更灵活的业务逻辑。
2. 扩展状态结构

以查询实体生日为例，扩展状态定义：
python

1. class State(TypedDict):
2.     messages: list
3.     name: str
4.     birthday: str

复制代码

3. 实现状态审核流程

创建带状态验证的人工协助工具：
python

1. from langchain_core.messages import ToolMessage
2. from langchain_core.tools import InjectedToolCallId
4. @tool
5. def human_assistance(
6.     name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]
7. ) -> str:
8.     """带状态验证的人工协助工具"""
9.     human_response = interrupt({
10.         "question": "Is this correct?",
11.         "name": name,
12.         "birthday": birthday
13.     })
14.    
15.     # 处理人工反馈
16.     if human_response.get("correct", "").lower().startswith("y"):
17.         verified_name, verified_birthday = name, birthday
18.         response = "Correct"
19.     else:
20.         verified_name = human_response.get("name", name)
21.         verified_birthday = human_response.get("birthday", birthday)
22.         response = f"Made a correction: {human_response}"
23.    
24.     # 状态更新
25.     state_update = {
26.         "name": verified_name,
27.         "birthday": verified_birthday,
28.         "messages": [ToolMessage(response, tool_call_id=tool_call_id)]
29.     }
30.     return Command(update=state_update)

复制代码

4. 状态管理实战

当用户查询 LangGraph 发布日期时：
python

1. user_input = "你能查一下LangGraph什么时候发布的吗？查到后用人工协助工具审核结果"
2. config = {"configurable": {"thread_id": "1"}}
4. # 触发查询流程
5. events = graph.stream({"messages": [{"role": "user", "content": user_input}]}, config)
6. for event in events:
7.     if "messages" in event:
8.         event["messages"][-1].pretty_print()
10. # 人工修正状态
11. human_command = Command(resume={"name": "LangGraph", "birthday": "Jan 17, 2024"})
12. events = graph.stream(human_command, config)
13. for event in events:
14.     if "messages" in event:
15.         event["messages"][-1].pretty_print()
17. # 查看最终状态
18. snapshot = graph.get_state(config)
19. print({"name": snapshot.values["name"], "birthday": snapshot.values["birthday"]})

复制代码

5. 手动状态操作

LangGraph 还支持直接操作状态：
python

1. # 手动更新状态
2. graph.update_state(config, {"name": "LangGraph (library)"})
4. # 查看更新后状态
5. snapshot = graph.get_state(config)
6. print({"name": snapshot.values["name"], "birthday": snapshot.values["birthday"]})

复制代码

三、时间旅行：回溯历史对话路径

1. 时间旅行的应用场景

在以下场景中时间旅行功能尤为重要：

调试智能体决策错误重现用户问题场景支持 "如果 - 那么" 式的交互探索

LangGraph 通过状态历史记录实现了强大的时间旅行能力。
2. 状态历史获取

获取完整的状态历史：
python

1. # 记录对话历史
2. user_input1 = "我在学习LangGraph，你能帮我做些研究吗？"
3. events = graph.stream({"messages": [{"role": "user", "content": user_input1}]}, config)
5. user_input2 = "很有帮助！我想用它构建一个自主代理"
6. events = graph.stream({"messages": [{"role": "user", "content": user_input2}]}, config)
8. # 获取状态历史
9. to_replay = None
10. for state in graph.get_state_history(config):
11.     print(f"消息数: {len(state.values['messages'])}, 下一步: {state.next}")
12.     if len(state.values["messages"]) == 6:  # 选择特定状态
13.         to_replay = state

复制代码

3. 状态回溯实战

从历史状态恢复执行：
python

1. # 查看选择的状态信息
2. print(f"下一步节点: {to_replay.next}")
3. print(f"状态配置: {to_replay.config}")
5. # 从历史状态恢复
6. for event in graph.stream(None, to_replay.config):
7.     if "messages" in event:
8.         event["messages"][-1].pretty_print()

复制代码

执行后系统会从指定历史状态继续运行，实现了类似 "时光倒流" 的效果，这对于调试和用户体验优化非常有帮助。
四、总结与实践建议

通过 LangGraph 的三大特性，我们实现了：

人在回路控制：通过 interrupt 机制实现人工干预，提升系统可靠性自定义状态：扩展状态结构，满足复杂业务需求时间旅行：状态历史管理，支持回溯和重放

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148793745