01. 人在环路与断点

人在环路（Human-in-the-loop，简称 HIL）交互对于 Agent 系统至关重要，特别是在一些特定领域的 Agent 中，需要经过人类的允许或者指示才能进入下一步（例如某些敏感或者重要操作），而 HIL 最重要的部分就是 断点。

断点 建立在 LangGraph 检查点之上，检查点在每个节点执行后保存图的状态，并且 检查点 可以使得图执行可以特定点暂停，等待人为批准，然后从最后一个检查点恢复执行。

流程图如下

通过流程图很容易可以知道，要想实现 断点 与 人机交互，其实要做的事情很简单：

1. 在图中设置 断点，这样图结构应用程序才只可以知道在哪个节点中断；
2. 针对人类的交互进行判断，执行相应的操作，恢复执行，亦或者更换执行；

在 LangGraph 中可以通过在 .compile() 编译的时候传递 interrupt_before(前置断点) 或者 interrupt_after(后置断点)，这样在图结构程序执行到 特定的节点 时就会暂停执行，等待其他操作（例如人类提示，修改状态等）。

如果需要恢复图执行，只需要再次调用 invoke/stream 等，并传递 inputs=None，传递输入为 None 意味着像中断没有发生一样继续执行，基于这个思路就可以实现让人类干预图的执行。

例如实现一个 ReACT 智能体，在需要调用工具时向人类发起提问，只有当人类输入 yes 的时候才继续执行工具，否则结束程序，示例

from typing import TypedDict, Annotated, Any, Literal

import dotenv

from langchain_community.tools import GoogleSerperRun

from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool

from langchain_community.utilities import GoogleSerperAPIWrapper

from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

from langgraph.checkpoint.memory import MemorySaver

from langgraph.graph import START, END, StateGraph

from langgraph.graph.message import add_messages

from langgraph.prebuilt import ToolNode

dotenv.load_dotenv()

class GoogleSerperArgsSchema(BaseModel):

    query: str = Field(description="执行谷歌搜索的查询语句")

class DallEArgsSchema(BaseModel):

    query: str = Field(description="输入应该是生成图像的文本提示(prompt)")

# 1.定义工具与工具列表

google_serper = GoogleSerperRun(

    name="google_serper",

    description=(

        "一个低成本的谷歌搜索API。"

        "当你需要回答有关时事的问题时，可以调用该工具。"

        "该工具的输入是搜索查询语句。"

    ),

    args_schema=GoogleSerperArgsSchema,

    api_wrapper=GoogleSerperAPIWrapper(),

)

dalle = OpenAIDALLEImageGenerationTool(

    name="openai_dalle",

    api_wrapper=DallEAPIWrapper(model="dall-e-3"),

    args_schema=DallEArgsSchema,

)

class State(TypedDict):

    """图状态数据结构，类型为字典"""

    messages: Annotated[list, add_messages]

tools = [google_serper, dalle]

llm = ChatOpenAI(model="gpt-4o-mini")

llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State, config: dict) -> Any:

    """聊天机器人函数"""

    # 1.获取状态里存储的消息列表数据并传递给LLM

    ai_message = llm_with_tools.invoke(state["messages"])

    # 2.返回更新/生成的状态

    return {"messages": [ai_message]}

def route(state: State, config: dict) -> Literal["tools", "__end__"]:

    """动态选择工具执行亦或者结束"""

    # 1.获取生成的最后一条消息

    ai_message = state["messages"][-1]

    # 2.检测消息是否存在tool_calls参数，如果是则执行`工具路由`

    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:

        return "tools"

    # 3.否则生成的内容是文本信息，则跳转到结束路由

    return END

# 1.创建状态图，并使用GraphState作为状态数据

graph_builder = StateGraph(State)

# 2.添加节点

graph_builder.add_node("llm", chatbot)

graph_builder.add_node("tools", ToolNode(tools=tools))

# 3.添加边

graph_builder.add_edge(START, "llm")

graph_builder.add_edge("tools", "llm")

graph_builder.add_conditional_edges("llm", route)

# 4.编译图为Runnable可运行组件

checkpointer = MemorySaver()

graph = graph_builder.compile(checkpointer=checkpointer, interrupt_before=["tools"])

# 5.调用图架构应用

config = {"configurable": {"thread_id": 1}}

state = graph.invoke(

    {"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少")]},

    config=config,

)

print(state)

# 6.获取人类的提示

if hasattr(state["messages"][-1], "tool_calls") and len(state["messages"][-1].tool_calls) > 0:

    tool_calls = state["messages"][-1].tool_calls

    print("准备调用工具: ", tool_calls)

    human_input = input("如需调用工具请回复yes，否则回复no: ")

    if human_input.lower() == "yes":

        print(graph.invoke(None, config))

    else:

        print("图程序执行结束")

交互内容如下：

{'messages': [HumanMessage(content='2024年北京半程马拉松的前3名成绩是多少', id='4c4cce54-4bf1-4e31-8b2c-ec28fb53be72'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_fV0JY28mOVdi9LoQzHCJzthk', 'function': {'arguments': '{"query":"2024 Beijing Half Marathon results top 3 finishers"}', 'name': 'google_serper'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 25, 'prompt_tokens': 168, 'total_tokens': 193}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-e3cabfd4-50aa-419b-acb5-e739ffcafbc0-0', tool_calls=[{'name': 'google_serper', 'args': {'query': '2024 Beijing Half Marathon results top 3 finishers'}, 'id': 'call_fV0JY28mOVdi9LoQzHCJzthk', 'type': 'tool_call'}], usage_metadata={'input_tokens': 168, 'output_tokens': 25, 'total_tokens': 193})]}

准备调用工具:  [{'name': 'google_serper', 'args': {'query': '2024 Beijing Half Marathon results top 3 finishers'}, 'id': 'call_fV0JY28mOVdi9LoQzHCJzthk', 'type': 'tool_call'}]

如需调用工具请回复yes，否则回复no: yes

{'messages': [HumanMessage(content='2024年北京半程马拉松的前3名成绩是多少', id='4c4cce54-4bf1-4e31-8b2c-ec28fb53be72'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_fV0JY28mOVdi9LoQzHCJzthk', 'function': {'arguments': '{"query":"2024 Beijing Half Marathon results top 3 finishers"}', 'name': 'google_serper'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 25, 'prompt_tokens': 168, 'total_tokens': 193}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-e3cabfd4-50aa-419b-acb5-e739ffcafbc0-0', tool_calls=[{'name': 'google_serper', 'args': {'query': '2024 Beijing Half Marathon results top 3 finishers'}, 'id': 'call_fV0JY28mOVdi9LoQzHCJzthk', 'type': 'tool_call'}], usage_metadata={'input_tokens': 168, 'output_tokens': 25, 'total_tokens': 193}), ToolMessage(content="Chinese runner He Jie, Ethiopian Dejene Hailu Bikila and Kenyans Robert Keter and Willy Mnangat at the finish line of the Beijing Half Marathon ... A controversial result finish at the Beijing Half Marathon is being investigated by event ... Duration: 0:42. Posted: Apr 15, 2024. Organizers are investigating a half marathon race in Beijing after three African runners appeared to let China's top long-distance runner ... The top three finishers of the Beijing half marathon are stripped of their medals after an investigation into the controversial result. Organisers have revoked He Jie's first place in the Beijing half marathon last weekend after an investigation confirmed that three other ... Missing: 3 | Show results with:3. China's marathon runner He Jie won the 2024 Beijing Half Marathon on Sunday with a time of 1:03:44. He was just one second ahead of Ethiopian ... The race's bizarre finish saw He Jie cross the line first after two runners from Kenya and one from Ethiopia seemed to deliberately allow him to ... Missing: finishers | Show results with:finishers. The top finishers in the Beijing Half Marathon have been disqualified after three runners from Africa appeared to step aside to let a ... Missing: results | Show results with:results. Beijing half marathon's bizarre finish saw He Jie cross the line first after the Kenyans Robert Keter and Willy Mnangat, and Dejene Hailu from ... Missing: finishers | Show results with:finishers. Video evidence from the final moments of the race showed He trailing behind Kenya's Robert Keter and Willy Mnangat and Ethiopia's Dejene Hailu. Missing: finishers | Show results with:finishers.", name='google_serper', id='429f485c-f6d4-423e-aa8f-347d84506a19', tool_call_id='call_fV0JY28mOVdi9LoQzHCJzthk'), AIMessage(content='2024年北京半程马拉松的前3名成绩如下：\n\n1. **何杰**（中国） - 1:03:44\n2. **德杰内·哈伊卢**（埃塞俄比亚） - 1:03:45\n3. **罗伯特·凯特尔**（肯尼亚）和**威利·姆南加特**（肯尼亚） - 具体时间未列出，但他们在何杰之后到达。\n\n不过，赛事组织者正在调查这场比赛，因为存在争议，可能会影响这些成绩的有效性。', response_metadata={'token_usage': {'completion_tokens': 131, 'prompt_tokens': 547, 'total_tokens': 678}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'stop', 'logprobs': None}, id='run-633ca834-a6de-4735-b81e-08a07a559d6e-0', usage_metadata={'input_tokens': 547, 'output_tokens': 131, 'total_tokens': 678})]}

输入 no 时的交互内容：

{'messages': [HumanMessage(content='2024年北京半程马拉松的前3名成绩是多少', id='92d280c0-382d-41e4-8b32-ff6abb9e112d'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_d9SKe8tbBLGFROjLwkuLQ0fL', 'function': {'arguments': '{"query":"2024年北京半程马拉松 前3名成绩"}', 'name': 'google_serper'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 27, 'prompt_tokens': 168, 'total_tokens': 195}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-84e98069-a553-4fa1-bc1f-10d9ee607ca5-0', tool_calls=[{'name': 'google_serper', 'args': {'query': '2024年北京半程马拉松 前3名成绩'}, 'id': 'call_d9SKe8tbBLGFROjLwkuLQ0fL', 'type': 'tool_call'}], usage_metadata={'input_tokens': 168, 'output_tokens': 27, 'total_tokens': 195})]}

准备调用工具:  [{'name': 'google_serper', 'args': {'query': '2024年北京半程马拉松 前3名成绩'}, 'id': 'call_d9SKe8tbBLGFROjLwkuLQ0fL', 'type': 'tool_call'}]

如需调用工具请回复yes，否则回复no: no

图程序执行结束

如果在二次执行 invoke/stream 的时候传递的并不是 None，而是对应的状态，则会对状态进行更新然后继续后续的操作，但是需要注意下，在这里额外更新状态要确保消息列表符合大语言模型的消息列表规范，避免发生错误，也尽可能不使用这种带有困惑性的方式。

02. 在图结构上更新对应状态

在 LangGraph 的图结构上，除了能通过 节点 更新 数据状态，还可以在图的外部通过调用图的 get_state() 与 update_state() 的方式来实现对数据状态的更新（特定线程下），并且 get_state() 和 update_state() 功能必须在 检查点 模式下才支持。

例如在上述的案例中，我们在 中断 程序后，修改 工具消息 的内容变成 2024年北京半程马拉松的第一名为好好01:59:40，第二名为好好好成绩为02:04:16，第三名为好好好好02:15:17，这个时候大语言模型会收到如下的消息列表：

1. Human：请帮我绘制一幅老爷爷爬山的图片。
2. Ai：调用 google_serper() 工具，并搜索 “2024年北京半程马拉松 前3名成绩”。
3. Tool：2024年北京半程马拉松的第一名为好好01:59:40，第二名为好好好成绩为02:04:16，第三名为好好好好02:15:17。

在上述示例中，我们认为篡改了 工具调用 返回的结果，让 ReACT 智能体去输出基于这个工具得到的答案。

代码示例如下

from typing import TypedDict, Annotated, Any, Literal

import dotenv

from langchain_community.tools import GoogleSerperRun

from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool

from langchain_community.utilities import GoogleSerperAPIWrapper

from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper

from langchain_core.messages import ToolMessage

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

from langgraph.checkpoint.memory import MemorySaver

from langgraph.graph import START, END, StateGraph

from langgraph.graph.message import add_messages

from langgraph.prebuilt import ToolNode

dotenv.load_dotenv()

class GoogleSerperArgsSchema(BaseModel):

    query: str = Field(description="执行谷歌搜索的查询语句")

class DallEArgsSchema(BaseModel):

    query: str = Field(description="输入应该是生成图像的文本提示(prompt)")

# 1.定义工具与工具列表

google_serper = GoogleSerperRun(

    name="google_serper",

    description=(

        "一个低成本的谷歌搜索API。"

        "当你需要回答有关时事的问题时，可以调用该工具。"

        "该工具的输入是搜索查询语句。"

    ),

    args_schema=GoogleSerperArgsSchema,

    api_wrapper=GoogleSerperAPIWrapper(),

)

dalle = OpenAIDALLEImageGenerationTool(

    name="openai_dalle",

    api_wrapper=DallEAPIWrapper(model="dall-e-3"),

    args_schema=DallEArgsSchema,

)

class State(TypedDict):

    """图状态数据结构，类型为字典"""

    messages: Annotated[list, add_messages]

tools = [google_serper, dalle]

llm = ChatOpenAI(model="gpt-4o-mini")

llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State, config: dict) -> Any:

    """聊天机器人函数"""

    # 1.获取状态里存储的消息列表数据并传递给LLM

    ai_message = llm_with_tools.invoke(state["messages"])

    # 2.返回更新/生成的状态

    return {"messages": [ai_message]}

def route(state: State, config: dict) -> Literal["tools", "__end__"]:

    """动态选择工具执行亦或者结束"""

    # 1.获取生成的最后一条消息

    ai_message = state["messages"][-1]

    # 2.检测消息是否存在tool_calls参数，如果是则执行`工具路由`

    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:

        return "tools"

    # 3.否则生成的内容是文本信息，则跳转到结束路由

    return END

# 1.创建状态图，并使用GraphState作为状态数据

graph_builder = StateGraph(State)

# 2.添加节点

graph_builder.add_node("llm", chatbot)

graph_builder.add_node("tools", ToolNode(tools=tools))

# 3.添加边

graph_builder.add_edge(START, "llm")

graph_builder.add_edge("tools", "llm")

graph_builder.add_conditional_edges("llm", route)

# 4.编译图为Runnable可运行组件

checkpointer = MemorySaver()

graph = graph_builder.compile(checkpointer=checkpointer, interrupt_after=["tools"])

# 5.调用图架构应用

config = {"configurable": {"thread_id": 1}}

state = graph.invoke(

    {"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少")]},

    config,

)

print(state)

# 6.修改图状态消息

graph_state = graph.get_state(config)

tool_message = ToolMessage(

    id=graph_state[0]["messages"][-1].id,

    tool_call_id=graph_state[0]["messages"][-2].tool_calls[0]["id"],

    name=graph_state[0]["messages"][-2].tool_calls[0]["name"],

    content="2024年北京半程马拉松的第一名为好好01:59:40，第二名为好好好成绩为02:04:16，第三名为好好好好02:15:17",

)

graph.update_state(config, {"messages": [tool_message]})

# 7.继续执行图

print(graph.invoke(None, config)["messages"][-1].content)

输出内容如下

{'messages': [HumanMessage(content='2024年北京半程马拉松的前3名成绩是多少', id='e7c7b73a-f1f3-4274-b716-4ce56337366b'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_aAXWa2UnaAD8CGkMSR6T1z2M', 'function': {'arguments': '{"query":"2024年北京半程马拉松 前3名成绩"}', 'name': 'google_serper'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 27, 'prompt_tokens': 168, 'total_tokens': 195}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_80a1bad4c7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-8dc3c1ca-94f3-40f9-b9db-22304088d874-0', tool_calls=[{'name': 'google_serper', 'args': {'query': '2024年北京半程马拉松 前3名成绩'}, 'id': 'call_aAXWa2UnaAD8CGkMSR6T1z2M', 'type': 'tool_call'}], usage_metadata={'input_tokens': 168, 'output_tokens': 27, 'total_tokens': 195}), ToolMessage(content='经过激烈角逐，男子组方面，中国选手何杰以1小时03分44秒的成绩夺得冠军，埃塞俄比亚选手DEJENE HAILU BIKILA以及来自肯尼亚的Robert Keter和WILLY MNANGAT三人以1小时03分45秒的成绩并列获得亚军，中国选手李春晖则以1小时06分58秒的成绩获得季军。', name='google_serper', id='57c0ca53-a272-4c00-9c0c-29fd2cdf5057', tool_call_id='call_aAXWa2UnaAD8CGkMSR6T1z2M')]}

2024年北京半程马拉松的前3名成绩如下：

1. 第一名：好好，成绩为 01:59:40

2. 第二名：好好好，成绩为 02:04:16

3. 第三名：好好好好，成绩为 02:15:17