在现在强大的 LLM 应用程序中，最有用的无外乎就是人机交互工作流，它将人工输入集成到自动化流程中，允许在关键阶段做出决策、验证或更正，因为底层模型可能会偶尔产生不准确之处，在合规性、决策或内容生成等低容错场景中，人工参与可通过审查、更正或覆盖模型输出来确保可靠性。
我么的应用程序在什么时候会使用人机交互工作流程呢？

• 1. 审查工具调用：人类可以在工具执行之前审查、编辑或批准 LLM 请求的工具调用。
• 2. 验证 LLM 输出：人类可以审查、编辑或批准 LLM 生成的内容。
• 3. 提供背景信息：使 LLM 能够明确请求人工输入以澄清或其他详细信息或支持多轮对话。

在开始我们正文前，我想给大家讲述在LangGraph里面非常奇妙的一个机制—— interrupt 。
看过之前文章的小伙伴肯定很熟悉，它就是LangGraph 中的中断功能，它通过在 graph 中的特定节点暂停，向人类展示信息，并在获取人类输入后继续执行，从而实现了人机协同工作流。这个功能适用于需要审批、编辑或收集额外输入等任务。中断功能通常与 Command 对象结合使用，以便在人类提供输入值后继续执行图。
举个例子：

from typing import TypedDict
import uuid

from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

class State(TypedDict):
   messages: str

def human_node(state: State):
   value = interrupt(
      # 任何序列化的值都可以展示出来，比如一个问题，一段文本
      {
         "text_to_revise": state["messages"]
      }
   )
   return {
      # 根据人类的输入更新状态，或根据输入决定图的执行路径。
      "messages": value
   }


graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")

# 要使 interrupt 功能正常工作，必须配置一个检查点（checkpointer）。
checkpointer = MemorySaver()
graph = graph_builder.compile(
   checkpointer=checkpointer
)
# 传递一个线程id运行graph
thread_config = {"configurable": {"thread_id": uuid.uuid4()}}

# 使用 stream() 方法直接展示 __interrupt__ 的信息。
for chunk in graph.stream({"messages": "hello，可以告诉我未来rag的发展趋势吗？"}, config=thread_config):
   print(chunk)

# 使用 Command 恢复
for chunk in graph.stream(Command(resume="怎么提高rag的检索能力"), config=thread_config):
   print(chunk)

langgraph的中断功能既强大又符合人体工程学。它们在开发体验上与 Python 的 input() 函数相似，但需要注意的是，中断并不会自动从中断点恢复执行。相反，它们会重新运行使用了中断的整个节点。因此，中断通常最好放置在节点的开头或一个专用的节点中。
总结一下要在 graph 中使用中断功能，指定一个检查点（checkpointer），在适当的位置调用 interrupt()，使用线程 ID 运行图：直到遇到中断为止，使用 invoke/ainvoke/stream/astream 恢复执行。
接下来开始我们的正文介绍几种设计模式：

一 、批准或拒绝设计模式

在人机协同工作流中，可以执行以下三种不同的操作：

• 批准或拒绝：
  在关键步骤（例如 API 调用）之前暂停图，以审查并批准操作。如果操作被拒绝，可以阻止图执行该步骤，并可能采取替代操作。这种模式通常基于人类的输入来决定图的执行路径。
• 编辑 graph 状态：
  暂停图以审查并编辑图的状态。这对于纠正错误或用额外信息更新状态非常有用。这种模式通常涉及根据人类的输入更新状态。
• 获取输入：
  在 graph 的特定步骤显式请求人类输入。这对于收集额外信息或上下文以支持 agent 的决策过程，或支持多轮对话非常有用。
  根据我们人类的批准或拒绝，graph 可以继续执行操作或采取替代路径。如下图：
  
  在我们执行关键步骤之前暂停 graph，我们人类介入审查并批准操作。如果操作被拒绝，可以阻止 graph 执行该步骤，并可能采取替代操作。代码如下：

from typing import Literal
from langgraph.types import interrupt, Command


def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]:
   is_approved = interrupt(
      {
         "question": "工作流引擎的设计方案",
         # 由人类审查和批准的输出。
         "llm_output": state["llm_output"]
      }
   )

   if is_approved:
      return Command(goto="one_node")
   else:
      return Command(goto="two_node")


# 将节点添加到 graph 中的适当位置，并将其连接到相关节点
graph_builder.add_node("human_approval", human_approval)
graph = graph_builder.compile(checkpointer=checkpointer)

# 在运行 graph 并触发中断后，graph 将暂停。通过批准或拒绝来恢复其执行。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(Command(resume=True), config=thread_config)

二 、审查并编辑状态设计模式

我们人类可以审查并编辑 graph 的状态。这对于纠正错误或用额外信息更新状态非常有用。

代码如下：

from langgraph.types import interrupt

def human_editing(state: State):
    ...
    result = interrupt(
        {
            "task": "审查大模型的输出，来进行必要的修改.",
            "llm_generated_summary": state["messages"]
        }
    )

    # 使用编辑后的文本更新状态。
    return {
        "messages": result["编辑后的文本"] 
    }

# 将节点添加到 graph 中的适当位置，并将其连接到相关节点
graph_builder.add_node("human_editing", human_editing)
graph = graph_builder.compile(checkpointer=checkpointer)

...
# 在运行 graph 并触发中断后，graph 将暂停。通过批准或拒绝来恢复其执行。
thread_config = {"configurable": {"thread_id": "some_id"}}
graph.invoke(
    Command(resume={"edited_text": "The edited text"}), 
    config=thread_config
)

三、审查调用工具设计模式

在继续执行之前，人类可以审查并编辑来自 LLM 的输出。加假如 LLM 请求的工具调用比较敏感则需要人工监督。如下图

代码如下面所示：

def human_review_node(state) -> Command[Literal["call_llm", "run_tool"]]:
    # 这是我们将通过 Command(resume=<human_review>) 提供的值。
    human_review = interrupt(
        {
            "question": "Is this correct?",
            # 为了审查显示调用的工具
            "tool_call": tool_call
        }
    )

    review_action, review_data = human_review

    # 批准工具调用
    if review_action == "continue":
        return Command(goto="run_tool")

    # 手动修改工具调用，然后继续执行。
    elif review_action == "update":
        updated_msg = get_updated_msg(review_data)
        # 要修改现有消息，我们需要传递具有匹配 ID 的消息。
        return Command(goto="run_tool", update={"messages": [updated_message]})

    # 提供自然语言反馈，然后将其传回给代理。
    elif review_action == "feedback":
        feedback_msg = get_feedback_msg(review_data)
        return Command(goto="call_llm", update={"messages": [feedback_msg]})

四、多轮对话设计模式

一种多轮对话设计，其中代理和人类节点来回循环，直到代理决定将对话移交给另一个代理或系统的另一部分。

多轮对话的设计涉及 agent 与人类之间的多次来回交互，这使得 agent 能够以对话的方式从人类那里收集更多信息。
这种设计模式在由多个 agent 组成的 LLM 应用程序中非常有用。一个或多个 agent 可能需要与人类进行多轮对话，而人类在对话的不同阶段提供输入或反馈。为了简化，下面的代理实现被展示为单个节点，但实际上它可能是由多个节点组成的更大图的一部分，并且可能包含条件边。
下面关于多轮对话又有两种设计方式：
第一种：

from langgraph.types import interrupt

def human_input(state: State):
    human_message = interrupt("human_input")
    return {
        "messages": [
            {
                "role": "human",
                "content": human_message
            }
        ]
    }

def agent(state: State):
   

graph_builder.add_node("human_input", human_input)
graph_builder.add_edge("human_input", "agent")
graph = graph_builder.compile(checkpointer=checkpointer)

graph.invoke(
    Command(resume="hello!"),
    config=thread_config
)

在这种模式中，每个代理都有自己的用于收集用户输入的人类节点。这可以通过为人类节点指定唯一名称（例如，“代理 1 的人类节点”、“代理 2 的人类节点”）或使用子图来实现，其中子图包含一个人类节点和一个代理节点。
第二种：

from langgraph.types import interrupt

def human_node(state: MessagesState) -> Command[Literal["agent_1", "agent_2", ...]]:
    """A node for collecting user input."""
    user_input = interrupt(value="Ready for user input.")

    # 从状态中确定活动代理，以便在收集输入后路由到正确的代理。例如，向状态添加字段或使用最后一个活动代理，或者填充代理生成的 AI 消息的 name 属性。
    active_agent = ... 

    return Command(
        update={
            "messages": [{
                "role": "human",
                "content": user_input,
            }]
        },
        goto=active_agent,
    )

在这种模式中，一个单独的人类节点用于为多个代理收集用户输入。活动代理从状态中确定，因此在收集人类输入后，图可以路由到正确的代理。

五、总结

从上面4中设计模式可以看到，LangGraph 的中断功能为 LLM 应用程序提供了强大的人机交互能力，使得在关键步骤中引入人工审查和干预成为可能。通过合理设计工作流，开发者可以确保LLM生成的输出和工具调用在敏感或关键场景中的准确性和可靠性。无论是批准、编辑还是多轮对话，LangGraph 的设计模式都为构建复杂的人机协同系统提供了灵活且强大的工具。