人机协作，Human-in-the-loop，自主决策的基础上，对某些可以预见或计划的关键节点引入人工干预，暂时中止其自我决策过程，转而由人工介入进行审批和确认，再让其继续执行后续任务？比如以下场景：
- 删除数据库操作：当
Agent决定删除数据库时，可以先中止操作，要求人工确认是否继续执行。经过人工确认后，Agent才继续删除数据库并执行后续操作。 - 机票改签操作：当
Agent决定更改机票时，系统会向用户发出通知，等待用户确认是否同意改签，用户点击确认后，Agent才会继续执行改签操作。
人机协作(由AIJoe生成)
通过这种方式，既能保持Agent的自主性，又能避免出现意外或不可控的风险，实现人机协作的平衡。而实现这种功能的技术，在Agent技术领域会被普遍称之为Human-in-the-loop（HIL）。
LangGraph底层是通过图结构来进行构建，并由状态做消息的传递，那么对于这样的结构来说，如果我们想在这样的架构中加入人工的介入流程，能操作的大致思路应该是：通过Router Agent去做判断，如果生成的响应触发了某种条件，就在原本要正常进入的节点之前先停止，等待人工的确认，再决定要不要执行，或者执行什么逻辑。在这个过程中，有几种常见的用户交互模式，分别是：
- 批准（Approval）：在代理的执行过程中，人工暂停代理的自主工作流，向用户展示当前的状态，并批准或者不批准执行该操作。
- 编辑（Editing）：在需要的时候，人工可以暂停代理，向用户展示当前的状态，并允许用户对代理的状态进行编辑。
- 输入（Input）：专门设计一个图节点来收集用户的输入，并将这些输入直接用于更新代理的状态。
在LangGraph的设计思路下 HIL通过战略性地放置断点（breakpoint）来实现的。这些断点会在关键点停止图的执行。在暂停期间，Agent将等待用户输入，利用这段时间收集响应，将它们集成到图状态中，并顺利继续进行，从而实现用户和代理之间的协作和交互式体验。
需要在定义的图结构中加入人机交互，这个图必须具备的两个核心参数正如compile()方法中的源码所示
def compile(
self,
checkpointer: Checkpointer = None,
*,
store: Optional[BaseStore] = None,
interrupt_before: Optional[Union[All, list[str]]] = None,
interrupt_after: Optional[Union[All, list[str]]] = None,
debug: bool = False,
) -> "CompiledStateGraph":
checkpointer参数可以接收任意类型的checkpointer，用来保存图的状态。而interrupt_before和interrupt_after参数，接收图中某个节点的名称，将其作为breakpoint，起到的作用是在该节点之前/之后中断图的继续运行
案例
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage
# 定义状态模式
class State(TypedDict):
user_input: str
model_response: str
user_approval: str
# 定义用于大模型交互的节点
def call_model(state):
messages = state["user_input"]
if '删除' in state["user_input"]:
state["user_approval"] = f"用户输入的指令是:{state['user_input']}, 请人工确认是否执行！"
else:
response = llm.invoke(messages)
state["user_approval"] = "直接运行！"
state["model_response"] = response
return state
# 定义人工介入的breakpoint内部的执行逻辑
def execute_users(state):
if state["user_approval"] == "是":
response = "您的删除请求已经获得管理员的批准并成功执行。如果您有其他问题或需要进一步的帮助，请随时联系我们。"
return {"model_response":AIMessage(response)}
elif state["user_approval"] == "否":
response = "对不起，您当前的请求是高风险操作，管理员不允许执行！"
return {"model_response":AIMessage(response)}
else:
return state
# 定义翻译节点
def translate_message(state: State):
system_prompt = """
Please translate the received text in any language into English as output
"""
messages = state['model_response']
messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
response = llm.invoke(messages)
return {"model_response": response}
# 构建状态图
builder = StateGraph(State)
# 向图中添加节点
builder.add_node("call_model", call_model)
builder.add_node("execute_users", execute_users)
builder.add_node("translate_message", translate_message)
# 构建边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "execute_users")
builder.add_edge("execute_users", "translate_message")
builder.add_edge("translate_message", END)
# 设置 checkpointer，使用内存存储
memory = MemorySaver()
# 在编译图的时候，添加短期记忆，并使用interrupt_before参数 设置 在 execute_users 节点之前中止图的运行，等待人工审核
graph = builder.compile(checkpointer=memory, interrupt_before=["execute_users"])
# 创建一个线程
config = {"configurable": {"thread_id": "2"}}
# 运行图，直至到断点的节点
async for chunk in graph.astream({"user_input": "我将在数据库中删除 id 为 user5625 的所有信息"}, config, stream_mode="values"):
print(chunk)