简述

随着大模型技术的发展，从单纯的对话交互到向具有自动化决策代理的大模型agent演进已经成为一个趋势，那如何在保证agent自动化进行决策的同时，提供安全的固定栅栏？答案是，在Agent进行决策的过程中，进行人工干预。
本文将基于LangGraph，详细剖析如何利用LangGraph中的状态机、检查点以及动态状态修改过程，最终构建一个需要人工参与审批的、时间旅行（time travel）以及状态编辑能力的智能Agent。

为啥使用LangGraph实现人机协作？

传统的链式框架（LangChain早期版本）通常难以精细化控制中途打断与状态回滚。而LangGraph 将 Agent 的执行逻辑建模为一个有向有环图，相对于之前的版本，其核心优势在于以下几点：

• 状态持久化（Checkpoints）： 自动记录每一次状态变更（State Transition）。
• 中断机制（Interrupts）： 支持在特定节点前强制挂起，等待外部干预。
• 状态分支与时间旅行： 可以从任意历史快照点继续执行或分叉。

最核心的实现：自定义Reducer的状态定义

在之前的博文AI Agents in LangGraph-2 中定义了智能体的状态类，在处理消息列表时，有用到简单的列表追加（使用operator.add将新的消息追加到列表中），在人机协作的场景下，无法满足“修改历史消息”或“替换未完成 Tool Call”的需求。为了满足该需求，我们需要定义一个自定义的消息融合逻辑：

from uuid import uuid4
from typing import TypedDict, Annotated
from langchain_core.messages import AnyMessage

def reduce_messages(left: list[AnyMessage], right: list[AnyMessage]) -> list[AnyMessage]:
    """
    根据 ID 动态更新或追加消息列表：
    - 如果新消息的 ID 已存在，则进行替换（Replace）
    - 如果新消息是全新的，则追加至末尾（Append）
    """
    for message in right:
        if not message.id:
            message.id = str(uuid4())
            
    merged = left.copy()
    for message in right:
        for i, existing in enumerate(merged):
            if existing.id == message.id:
                merged[i] = message
                break
        else:
            merged.append(message)
    return merged

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], reduce_messages]

构建支持人工中断的Agent

借助 interrupt_before=[“action”]，我们可以指定执行物理工具动作前（比如网络检索），等待人类的审批：

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import SystemMessage, ToolMessage

class Agent:
    def __init__(self, model, tools, system="", checkpointer=None):
        self.system = system
        graph = StateGraph(AgentState)
        
        # 1. 注册节点
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        
        # 2. 设置图边与条件逻辑
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        
        # 3. 编译图，并注入 Checkpointer 与中断点
        self.graph = graph.compile(
            checkpointer=checkpointer,
            interrupt_before=["action"]  # 关键：在工具执行前中断
        )
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        return {'messages': results}

人机协作（HITL）的三层实战维度

1. 手动审批(Manual Approve)

当用户提问后，Agent生成了调用工具的意图，但是受限于interrupt_before=[“action”]， 程序停止在了action节点之前。

memory = SqliteSaver.from_conn_string(":memory:")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

thread = {"configurable": {"thread_id": "1"}}
messages = [HumanMessage(content="Whats the weather in SF?")]

# 第一次流式运行：触发中断
for event in abot.graph.stream({"messages": messages}, thread):
    print(event)

# 检查当前状态：下一个可执行节点为 'action'
current_state = abot.graph.get_state(thread)
print(current_state.next) # 输出: ('action',)

# 人类审批后继续（传入 None 续接执行）
for event in abot.graph.stream(None, thread):
    print(event)

在实际的生产环境中，可以配合交互式控制台或者式前端按钮来完成干预机制：

while abot.graph.get_state(thread).next:
    _input = input("检测到敏感操作，是否继续执行？(y/n): ")
    if _input.lower() != "y":
        print("操作已取消。")
        break
    for event in abot.graph.stream(None, thread):
        print(event)

2. 运行时干预与状态篡改

在线上运行时，人类不仅可以决定是否继续，还可以决定“修改AI的决策”。比如说，AI在企图查询LA的天气，但是审批员希望将查询时的参数改成路易斯安娜

thread = {"configurable": {"thread_id": "3"}}
# 运行直到中断点
for event in abot.graph.stream({"messages": [HumanMessage("Whats the weather in LA?")]}, thread):
    pass

# 1. 读取当前最新快照
current_values = abot.graph.get_state(thread)
tool_call_id = current_values.values['messages'][-1].tool_calls[0]['id']

# 2. 修改 Tool Call 参数
current_values.values['messages'][-1].tool_calls = [{
    'name': 'tavily_search_results_json',
    'args': {'query': 'current weather in Louisiana'},
    'id': tool_call_id
}]

# 3. 将修改后的状态写回 Checkpointer
abot.graph.update_state(thread, current_values.values)

# 4. 恢复运行：Agent 将按人工修正后的参数搜索
for event in abot.graph.stream(None, thread):
    print(event)

3.时间旅行和分支演进

由于 LangGraph 保存了全量的 State History，开发者可以回退到历史上的任意快照节点，分叉出全新的执行路线。

# 获取历史状态栈（最新在前）
states = list(abot.graph.get_state_history(thread))

# 选择一个历史节点（例如倒数第三个历史节点）
to_replay = states[-3]

# 模拟人为直接“伪造” ToolMessage 结果返回，绕过模型本身的工具执行
state_update = {
    "messages": [
        ToolMessage(
            tool_call_id=to_replay.values['messages'][-1].tool_calls[0]['id'],
            name="tavily_search_results_json",
            content="54 degree celcius",
        )
    ]
}

# 以 'action' 节点的身份将状态注入该历史时刻，开启全新分支
branch_and_add = abot.graph.update_state(
    to_replay.config, 
    state_update, 
    as_node="action"
)

# 从新分支继续运行
for event in abot.graph.stream(None, branch_and_add):
    print(event)

状态转换与Reducer机制解析

理解 LangGraph 的 update_state 行为时，必须掌握 Reducer 的映射逻辑：

class AgentState(TypedDict):
    lnode: str
    scratch: str
    count: Annotated[int, operator.add] # 使用了加法累加器

对上述的状态进行更新时：

属性	定义方式	update_state传入值	更新后的实际状态值
scratch	普通字段	“hello”	“hello”（覆写）
count	operator.add	-3	Current_Count + (-3)（累加）
在使用 as_node=“NodeName” 时，LangGraph 会以该节点为视点，自动计算下一步的控制流走向，包括评估条件边。

总结

在 LangGraph 中，人机协作（Human-in-the-Loop） 不仅仅是一个中断开关，而是一套完整的状态管控体系：

• interrupt_before / interrupt_after 实现了精准的操作挂起。
• update_state() 结合自定义 Reducer 为人类提供了对 Agent 运行记忆的读写修正能力。
• get_state_history() 则为调试、审计以及复杂分支推演提供了可回滚的“时间旅行”机制。

掌握这些模式，是构建高可靠性、具备合规性与可控性的企业级 Agent 应用的关键一步。