LangGraph中人机协作的方法 / ai #62

人机协作，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)