图结构断点实现 Agent 与人进行交互

01. 人在环路与断点

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

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

流程图如下：

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

• 在图中设置 断点 ，这样图结构应用程序才只可以知道在哪个节点中断；

• 针对人类的交互进行判断，执行相应的操作，恢复执行，亦或者更换执行；

在 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)")



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:
    """聊天机器人函数"""
    
    ai_message = llm_with_tools.invoke(state["messages"])
    
    return {"messages": [ai_message]}


def route(state: State, config: dict) -> Literal["tools", "__end__"]:
    """动态选择工具执行亦或者结束"""
    
    ai_message = state["messages"][-1]
    
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    
    return END



graph_builder = StateGraph(State)


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


graph_builder.add_edge(START, "llm")
graph_builder.add_edge("tools", "llm")
graph_builder.add_conditional_edges("llm", route)


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


config = {"configurable": {"thread_id": 1}}
state = graph.invoke(
    {"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少")]},
    config=config,
)
print(state)


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 ，这个时候大语言模型会收到如下的消息列表：

• Human ：请帮我绘制一幅老爷爷爬山的图片。

• Ai ：调用 google_serper() 工具，并搜索 “2024年北京半程马拉松 前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)")



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:
    """聊天机器人函数"""
    
    ai_message = llm_with_tools.invoke(state["messages"])
    
    return {"messages": [ai_message]}


def route(state: State, config: dict) -> Literal["tools", "__end__"]:
    """动态选择工具执行亦或者结束"""
    
    ai_message = state["messages"][-1]
    
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    
    return END



graph_builder = StateGraph(State)


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


graph_builder.add_edge(START, "llm")
graph_builder.add_edge("tools", "llm")
graph_builder.add_conditional_edges("llm", route)


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


config = {"configurable": {"thread_id": 1}}
state = graph.invoke(
    {"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少")]},
    config,
)
print(state)


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]})


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

最新版 LangChain 用法提示

• 新版 LangChain 更推荐用 LCEL、Runnable、ChatModel.bind_tools()、结构化输出和 LangGraph 来组织复杂链路；老式 Chain、部分 AgentExecutor 写法可以读懂，但新项目应优先选择更清晰的图或 Runnable 编排。

• 工具调用相关代码要区分两层：模型是否原生支持 tool/function calling，以及业务侧如何定义工具 schema、参数校验、错误兜底和观测日志。

• 如果示例中的导入路径和你当前安装版本不同，优先查当前版本包内导出位置；常见迁移方向是从 langchain 拆到 langchain-core、langchain-community、langchain-openai、langgraph 等包。

拓展

• 工具或插件不要只看能不能调通，更要看是否可观测、可限流、可重试、可审计。联网类工具还要处理超时、空结果、搜索噪声和结果时效性。

• Agent 场景里，Prompt 只是调度策略的一部分；工具描述、参数 schema、历史状态、错误反馈、停止条件和人工介入点同样会影响最终稳定性。

常见问题

• 为什么模型没有调用工具？常见原因是工具描述不清晰、参数 schema 过宽或过窄、用户问题不需要工具、模型本身不支持工具调用，或者工具绑定位置不对。

• 为什么工具调用后回答仍然不准？先看工具返回是否正确，再看工具结果是否被放回模型上下文，最后检查输出解析、历史消息和异常兜底是否覆盖了真实错误。

面试题

• 解释函数调用、工具调用和 Agent 的区别。

• LangChain 中 tool schema 的作用是什么？为什么参数校验对生产环境很重要？

• ReACT Agent 和 tool-calling Agent 的核心差异是什么？分别适合什么场景？

• LangGraph 相比 LCEL 更适合解决哪些复杂编排问题？

生产问题排查

问题	常见原因	处理方式
工具没有被调用	工具描述弱、绑定失败、模型不支持	打印绑定后的模型配置，补充工具描述，换用支持工具调用的模型
参数格式错误	schema 设计不清晰，模型生成字段不稳定	使用 Pydantic/JSON Schema 校验，失败后把错误反馈给模型重试
联网结果不可用	搜索为空、接口超时、命中低质量页面	增加超时、重试、结果过滤、来源白名单和降级回答
Agent 循环不停止	缺少终止条件或工具返回被误判	设置最大迭代次数，记录每轮 thought/action/observation，增加停止规则
线上难以复现	缺少输入、工具请求和模型响应日志	给每次调用加 trace id，记录工具入参、出参、耗时和异常