内容纲要
Human-in-the-loop（人机交互）
目前为止，现在能够实现的最复杂的一种 AI Agent 逻辑是：具备多外部工具支持的完全自主循环代理结构，同时具备多轮对话的上下文记忆能力。而具备这种能力的系统其实已经足够有效地应对大部分 AI 应用场景的需求，并且已经完全涵盖了 LangGraph 底层构建生产级别 Agent 的所必须掌握的知识点。
最后，要进行补充的是：这种完全自主循环代理构建起来的应用，在现实落地场景中，会衍生出什么迫切的需求？
一个有趣且现实的情况是：我们希望 AI Agent 能够帮助我们自主处理各种任务，但当它现在有能力做到时，我们又开始担心它可能做出不当的决策，尤其是涉及到高风险操作时。比如，Agent 可能会不小心删除生产环境中的数据库，或是转移账户中的余额等敏感数据。这些操作是无法容忍的。因此，在自主代理类架构下衍生出的非常明确的一类需求是：能否在保持 Agent 自主决策的基础上，对某些可以预见或计划的关键节点引入人工干预，暂时中止其自我决策过程，转而由人工介入进行审批和确认，再让其继续执行后续任务？比如以下场景：
删除数据库操作：当 Agent 决定删除数据库时，可以先中止操作，要求人工确认是否继续执行。经过人工确认后，Agent 才继续删除数据库并执行后续操作。
机票改签操作：当 Agent 决定更改机票时，系统会向用户发出通知，等待用户确认是否同意改签，用户点击确认后，Agent 才会继续执行改签操作。
……
通过这种方式，既能保持 Agent 的自主性，又能避免出现意外或不可控的风险，实现人机协作的平衡。而实现这种功能的技术，在 Agent 技术领域会被普遍称之为 Human-in-the-loop（HIL）。
一、LangGraph 中的 HIL
LangGraph 底层是通过图结构来进行构建，并由状态做消息的传递，那么对于这样的结构来说，如果我们想在这样的架构中加入人工的介入流程，能操作的大致思路应该是：通过 Router Agent 去做判断，如果生成的响应触发了某种条件，就在原本要正常进入的节点之前先停止，等待人工的确认，再决定要不要执行，或者执行什么逻辑。在这个过程中，有几种常见的用户交互模式，分别是：
- 批准（Approval）：在代理的执行过程中，人工暂停代理的自主工作流，向用户展示当前的状态，并批准或者不批准执行该操作。
- 编辑（Editing）：在需要的时候，人工可以暂停代理，向用户展示当前的状态，并允许用户对代理的状态进行编辑。
- 输入（Input）：专门设计一个图节点来收集用户的输入，并将这些输入直接用于更新代理的状态。
在 LangGraph 的设计思路下 HIL 通过战略性地放置断点（breakpoint）来实现的。这些断点会在关键点停止图的执行。在暂停期间，Agent 将等待用户输入，利用这段时间收集响应，将它们集成到图状态中，并顺利继续进行，从而实现用户和代理之间的协作和交互式体验 这个交互过程在 LangGraph 框架下的具体实现思路如下图所示：
LangGraph 能够在中断后继续运行的核心在于我们之前介绍的 checkpointer 组件。这个组件能够在一个独立的线程中保存图中每个节点的状态。由于这些信息被持久化保存（包括将内存用作持久存储），我们就可以随时提取并修改图产生的数据。更改完成后，再将这些数据重新传回到图中以继续运行流程。这一机制是 LangGraph 已成功实现的功能。由此可见 Human-in-the-loop（HIL）并不是一个全新的组件，而是基于 LangGraph 底层的构建组件延展出来的一种实现方法。那这里我们就需要清楚两个概念，其一是用于中断两个本应顺序执行节点的操作，在 LangGraph 中称其为 breakpoint（断点），其二是 breakpoint 是构建在 checkpointer 之上的。
因此，当我们需要在定义的图结构中加入人机交互，这个图必须具备的两个核心参数正如 compile() 方法中的源码所示:
链接：LangGraph Graph Compile : https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.compile
其中，checkpointer 参数可以接收任意类型的 checkpointer，用来保存图的状态。而 interrupt_before 和 interrupt_after 参数，接收图中某个节点的名称，将其作为 breakpoint，起到的作用是在该节点之前 / 之后中断图的继续运行。我们来看一个具体的应用案例。
二. 标准图结构中如何添加断点
如下代码实现的业务场景是一个自动化的人机交互流程，用于执行内容删除这种高度敏感性的操作。我们定义两个主要的功能节点：call_model 和 execute_users。call_model 节点处理大模型的调用和响应，如果用户输入包含 “删除” 的内容，则会触发需要人工审批的流程。execute_users 节点根据人工审核的结果来决定最终的响应内容。如下代码所示：
一、等待人工输入模式 4 个关键步骤
等待人工输入是 HIL 最常用交互模式，Agent 遇到高危操作会暂停并向用户确认，流程四步：
配置断点：图编译时通过 interrupt_before / interrupt_after 指定需要暂停的节点；
配置检查点：必须传入 checkpointer 持久化会话状态，中断后保存上下文；
更新状态：人工输入审批结果后，调用 graph.update_state() 把人工响应写入状态；
恢复执行：无新输入调用 stream(None, config)，从断点续跑完整图流程。
import getpass
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
import getpass
import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, StateGraph
import json
from langgraph.checkpoint.memory import MemorySaver
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"])
# 生成 Mermaid PNG 数据
image_data = graph.get_graph().draw_mermaid_png()
# 定义保存路径
file_path = "langgraph_structure.png"
# 写入文件
with open(file_path, "wb") as f:
f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")
# 在添加了 breakpoint（断点）后的图结构中，它的运行逻辑将变成：只要到达被设置为 breakpoint 的节点时，图就会中止运行。这里我们进行一个实际的调用测试来理解这个过程，代码如下所示:
# 创建一个线程
config = {"configurable": {"thread_id": "2"}}
# 运行图，直至到断点的节点
for chunk in graph.stream({"user_input": "我将在数据库中删除 id为 xigualaoshi 的所有信息"}, config, stream_mode="values"):
print(chunk)
"""
输出结果：
{'user_input': '我将在数据库中删除 id为 xigualaoshi 的所有信息'}
{'user_input': '我将在数据库中删除 id为 xigualaoshi 的所有信息', 'user_approval': '用户输入的指令是:我将在数据库中删除 id为 xigualaoshi 的所有信息, 请人工确认是否执行! '}
通过输出结果，我们可以观察到在执行到 execute_users 节点之前就暂停了，它将等待人工介入以决定是否继续执行。这就是 breakpoint（断点）的关键作用。
在这个阶段，我们就可以审查并调整图的状态，而关键的处理逻辑是：要修改或确认全局状态模式中 user_approval 字段的值，从而指导 execute_users 节点的行为。
"""
snapshot = graph.get_state(config)
print(snapshot)
"""
结果：
StateSnapshot(values={'user_input': '我将在数据库中删除 id为 xigualaoshi 的所有信息', 'user_approval': '用户输入的指令是:我将在数据库中删除 id为 xigualaoshi 的所有信息, 请人工确认是否执行! '}, next=('execute_users',), config={'configurable': {'thread_id': '2', 'checkpoint_ns': '', 'checkpoint_id': '1f179e82-e1ec-64b6-8001-98a80761772b'}}, metadata={'source': 'loop', 'step': 1, 'parents': {}}, created_at='2026-07-07T09:42:33.833797+00:00', parent_config={'configurable': {'thread_id': '2', 'checkpoint_ns': '', 'checkpoint_id': '1f179e82-e1e8-6f43-8000-c1b42c11280b'}}, tasks=(PregelTask(id='2d8452f5-1493-1586-1cbc-965a46950aa5', name='execute_users', path=('__pregel_pull', 'execute_users'), error=None, interrupts=(), state=None, result=None),), interrupts=())
我们先来看如何在图中止运行后，在图状态中添加用户的决策。
为实现这一点，我们可以手动设置 snapshot.values['user_approval'] 为是，用以说明在即将执行的 execute_users 节点，人工审批的状态被手动设定为同意。
接下来，使用 graph.update_state(config, snapshot.values) 来更新状态图中的状态。这个调用将状态图中的当前状态更新为包含了新的 user_approval 值的 snapshot.values。
"""
# 1. 修改快照里的人工审批状态
snapshot.values['user_approval']='是'
# 2. 将修改后的状态写回检查点
graph.update_state(config, snapshot.values)
# 修改完状态后，如果想让图基于 breakpoint 继续执行后续的操作，则只需要在 astream 方法中将 input 参数设置为 None，则会让图从上一次中断的地方继续。如下代码所示：
for chunk in graph.stream(None, config, stream_mode="values"):
print(chunk)
print("*" * 30,"人工修改审批标记为'否'，拒绝高危删除操作","*" * 50)
# 1. 获取当前会话断点快照
snapshot = graph.get_state(config)
# 2. 人工修改审批标记为"否"，拒绝高危删除操作
snapshot.values['user_approval']='否'
# 3. 将修改后的状态持久写入检查点
graph.update_state(config, snapshot.values)
# 4. input=None 从断点续跑完整图流程
for chunk in graph.stream(None, config, stream_mode="values"):
print(chunk)
# 以上是两种触发删除的高危风险审核的逻辑，而如果 Agent 接收的是普通的问答交互，不涉及高危险操作指令时，则可以直接将 input 参数设置为 None，运行断点之后的图节点逻辑。如下代码所示：
print("*" * 30,"普通的问答交互，不涉及高危险操作指令","*" * 30)
# 运行图，直至到断点的节点
for chunk in graph.stream({"user_input": "你好，请你介绍一下你自己"}, config, stream_mode="values"):
pass
for chunk in graph.stream(None, config, stream_mode="values"):
print(chunk)
# 创建一个函数来封装对话逻辑
def run_dialogue(graph, config, all_chunks=[]):
while True:
# 接收用户输入
user_input = input("请输入您的消息（输入'退出'结束对话）：")
if user_input.lower() == '退出':
break
# 运行图，直至到断点的节点
for chunk in graph.stream({"user_input": user_input}, config, stream_mode="values"):
all_chunks.append(chunk)
# 处理可能的审批请求
last_chunk = all_chunks[-1]
if last_chunk["user_approval"] == f"用户输入的指令是:{last_chunk['user_input']}, 请人工确认是否执行! ":
user_approval = input(f"当前用户的输入是：{last_chunk['user_input']}，请人工确认是否执行！请回复 是/否。")
graph.update_state(config, {"user_approval": user_approval})
# 继续执行图（从断点向后跑完剩余节点）
for chunk in graph.stream(None, config, stream_mode="values"):
all_chunks.append(chunk)
# 显示最终模型的响应
print("人工智能助理：", all_chunks[-1]["model_response"].content)
if __name__ == "__main__":
config = {"configurable": {"thread_id" : "3"}}
run_dialogue(graph, config)如上实现的功能所示，在图中的特定位置添加断点是启用人机协同的一种方法。在这种情况下，开发人员只要知道工作流中的哪个位置需要人工输入，只需在该特定图形节点之前或之后放置一个断点即可。LangGraph 的 breakpoint 和 HIL 功能提供了比较完善的实现机制来做交互式的图执行，允许用户输入来指导或中断流程。在这个案例中，我们展示的人机交互策略特别适用于有明确工作流的业务场景。开发人员可以根据业务逻辑精确地确定哪些环节可能涉及高风险操作，并通过人工决策介入来提升 Agent 应用程序的稳定性和可靠性。
接下来我们可以更近一步，看一下在具备工具调用的 Agent 架构中如何去做断点的人工交互逻辑，这包含 Tool Calling Agent 和 ReAct 两种模式。
三. 复杂代理架构中如何添加动态断点
上面案例中实现的人机交互模式，我们可以在不同的用户输入决策下完全自定义具体需要执行的操作逻辑。而这样的交互过程，我们可以手动实现来理解这个中间过程，当然也可以在 LangGraph 框架的封装下，借助 Router Agent 的机制从节点内部来进行动态管理，尤其是在涉及工具调用的 ReAct 框架下，其特征会更明显，且可以更加清晰的帮助我们理解 breakpoint 在图执行过程中的内部逻辑。其基本过程如下图所示：
四、一刀切，所有工具被调用前都需要人工审核
首先来考虑这样一个场景：对于 Tool Calling Agent 架构实现的应用程序，如果我们希望在执行任意一个工具之前，都需要对工具进行审核的话，那么这个过程就可以非常简单的通过如下代码来进行实现。这里我们先准备两个外部工具，一个用于实时检索互联网信息，一个用于检索实时天气。
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
from langgraph.prebuilt import ToolNode
import json
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
GOOGLE_SEARCH_KEY = os.environ["GOOGLE_SEARCH_KEY"]
class WeatherLoc(BaseModel):
location: str = Field(description="The location name of the city")
class SearchQuery(BaseModel):
query: str = Field(description="Questions for networking queries")
@tool(args_schema = SearchQuery)
def fetch_real_time_info(query):
"""Get real-time Internet information"""
url = "https://google.serper.dev/search"
payload = json.dumps({
"q": query,
"num": 1,
})
headers = {
'X-API-KEY': GOOGLE_SEARCH_KEY,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=payload)
data = json.loads(response.text) # 将返回的JSON字符串转换为字典
if 'organic' in data:
return json.dumps(data['organic'], ensure_ascii=False) # 返回'organic'部分的JSON字符串
else:
return json.dumps({"error": "No organic results found"}, ensure_ascii=False) # 如果没有'organic'键，返回错误信息
@tool(args_schema=WeatherLoc)
def get_weather(location):
"""
Function to query current weather.
:param loc: Required parameter, of type string, representing the specific city name for the weather query. \
Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
the loc parameter should be input as 'Beijing'.
:return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
"""
# Step 1. 构建请求
url = "https://api.openweathermap.org/data/2.5/weather"
# Step 2. 设置查询参数
params = {
"q": location,
"appid": weather_key, # 输入API key
"units": "metric", # 使用摄氏度而不是华氏度
"lang":"zh_cn" # 输出语言为简体中文
}
# Step 3. 发送GET请求
response = requests.get(url, params=params)
# Step 4. 解析响应
data = response.json()
return json.dumps(data)
"""
@tool 装饰器（来自 langchain_core.tools）会把原本的普通函数包装成一个 StructuredTool / Tool 对象，
而不再是一个普通的 Python 函数。这个对象暴露的是 LangChain 工具接口（.invoke()、.run()、.ainvoke() 等），
原始函数被存到了 .func 属性里。
正确调用：get_weather.invoke("beijing")
错误调用：get_weather("beijing") —— 会报错，因为 get_weather 已经不是函数了，是 Tool 对象，
直接 () 调用不符合它的签名（且它的调用协议和普通函数不同）。
"""
weather_result = get_weather.invoke("beijing")
# print(weather_result)
"""
该方法去掉装饰器之后，就变回普通方法，可以直接 方法名(参数) 调用
去掉 @tool(...) 后，get_weather 就是一个普通的 Python 函数，直接按位置/关键字参数调用即可
比如：get_weather("beijing")
"""
# weather_result = get_weather("beijing")
# print(weather_result)
web_search = fetch_real_time_info.invoke("小米汽车多少钱？")
# print(web_search)
tools = [get_weather, fetch_real_time_info]
tool_node = ToolNode(tools)
# 定义大模型实例，这里使用 GPT-4o 模型进行演示。
import getpass
import os
import json
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
llm = llm.bind_tools(tools)
# 定义一个 Router Function，用来根据大模型的实时响应判断是执行外部函数调用还是直接输出最终的响应。
def should_continue(state):
messages = state["messages"]
last_message = messages[-1]
# 如果没有工具调用，则输出至最终节点
if not last_message.tool_calls:
return "end"
# 如果还有子任务需要继续执行工具调用的话，则继续等待执行
else:
return "continue"
# 用于大模型交互的节点函数，其功能是接收用户的响应，使用 GPT-4o 模型生成具体的响应文本。
def call_model(state):
messages = state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
# 然后通过 LangGraph 基本构建图的方法，依次定义状态图 -> 向图中添加节点 -> 添加节点之间的边。
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.add_edge(START, "agent")
# 添加条件边
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"end": END,
},
)
workflow.add_edge("action", "agent")
# 最后，如果想要在执行任意工具前都由人工介入进行确认，只需要在编译图的时候，在调用 action 节点之前添加一个 breakpoint（断点）。如下所示:
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory, interrupt_before=["action"])
# 生成可视化图像结构
# 生成 Mermaid PNG 数据
image_data = graph.get_graph().draw_mermaid_png()
# 定义保存路径
file_path = "langgraph_structure.png"
# 写入文件
with open(file_path, "wb") as f:
f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")
# 编译后，则可以开始与基于该图构建出来的 Agent 进行对话交互，如下代码所示:
config = {"configurable": {"thread_id": "4"}}
for chunk in graph.stream({"messages": "请帮我查一下北京的天气"}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
# 输出示例
"""
================================ Human Message =================================
请帮我查一下beijing的天气
================================== Ai Message ==================================
Tool Calls:
get_weather (chatcmpl-tool-bac88512411dd27d)
Call ID: chatcmpl-tool-bac88512411dd27d
Args:
location: beijing
"""
print("*"*30,"断点中断后继续执行工具调用，input传None","*"*30)
# 断点中断后继续执行工具调用，input传None
for chunk in graph.stream(None, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================
Tool Calls:
get_weather (chatcmpl-tool-bac88512411dd27d)
Call ID: chatcmpl-tool-bac88512411dd27d
Args:
location: beijing
================================= Tool Message =================================
Name: get_weather
{"coord": {"lon": 116.3972, "lat": 39.9075}, "weather": [{"id": 802, "main": "Clouds", "description": "\u591a\u4e91", "icon": "03d"}], "base": "stations", "main": {"temp": 35.71, "feels_like": 36.21, "temp_min": 35.71, "temp_max": 35.71, "pressure": 996, "humidity": 32, "sea_level": 996, "grnd_level": 991}, "visibility": 10000, "wind": {"speed": 3.88, "deg": 201, "gust": 4.13}, "clouds": {"all": 32}, "dt": 1783496725, "sys": {"country": "CN", "sunrise": 1783457582, "sunset": 1783511123}, "timezone": 28800, "id": 1816670, "name": "Beijing", "cod": 200}
================================== Ai Message ==================================
根据查询结果，北京当前的天气情况如下：
**基本信息：**
- 城市：北京 (Beijing)
- 国家：中国 (CN)
- 时区：UTC+8
**天气状况：**
- 天气：多云 (Clouds)
- 温度：35.71°C
- 体感温度：36.21°C
- 最低/最高温度：35.71°C / 35.71°C
**其他信息：**
- 湿度：32%
- 风速：3.88 m/s，风向201度
- 能见度：10000米
- 云量：32%
**日出日落：**
- 日出时间：2025年9月6日 06:53:02
- 日落时间：2025年9月6日 18:45:23
总体来说，北京今天天气多云，气温较高（35.71°C），湿度适中（32%），风速较小，适合外出但需要注意防晒。
"""
"""
当 input 参数传入的值是 None 时，会传递状态模式中全部已产生的消息数据，所以它的执行过程看起来就像没有中断一样，按照我们构造的图结构顺利完成了所有节点的执行逻辑。对另外的一个 fetch_real_time_info 工具其实也是一样的，如果 Agent 判断需要调用的话，仍然会在执行前停止，等待人工批准。如下代码所示:
"""
config = {"configurable": {"thread_id": "5"}}
for chunk in graph.stream({"messages": "小米汽车"}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
# 执行输出示例
"""
================================ Human Message =================================
小米汽车
================================== Ai Message ==================================
Tool Calls:
fetch_real_time_info (chatcmpl-tool-a3bf319f58709d0d)
Call ID: chatcmpl-tool-a3bf319f58709d0d
Args:
query: 小米汽车最新消息 2024
"""
# 人工确认后继续执行工具调用
for chunk in graph.stream(None, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================
Tool Calls:
fetch_real_time_info (chatcmpl-tool-a3bf319f58709d0d)
Call ID: chatcmpl-tool-a3bf319f58709d0d
Args:
query: 小米汽车最新消息 2024
================================= Tool Message =================================
Name: fetch_real_time_info
[{"title": "新一代SU7 - 小米汽车", "link": "https://www.xiaomiev.com/su7", "snippet": "升级至700TOPS辅助驾驶芯片，相较于第一代小米SU7 Max，算力提升37.8%，支持更丰富场景的实时处理。数据来自小米汽车实验室。", "position": 1}]
================================== Ai Message ==================================
根据最新信息，小米汽车目前主要聚焦于**SU7系列**车型。以下是主要信息：
## 小米SU7系列
**最新动态：**
- **新一代SU7**已经发布，升级至**700TOPS辅助驾驶芯片**
- 相比第一代小米SU7 Max，**算力提升37.8%**
- 支持更丰富场景的实时处理
- 数据来自小米汽车实验室
## 小米汽车发展背景
小米于2021年宣布进军智能电动汽车领域，2024年正式发布首款车型SU7。小米汽车继承了小米在智能硬件和生态系统方面的优势，致力于打造智能化、网联化的新能源汽车。
## 产品特点
小米SU7定位于**中高端智能电动轿车**，主要特点包括：
- 智能化驾驶辅助系统
- 小米生态系统深度整合
- 高性能动力系统
- 先进的自动驾驶技术
目前小米汽车正在快速发展中，持续推出更新迭代的车型和技术。
您想了解小米汽车的哪个具体方面呢？比如价格、配置、预订信息等。
"""
"""
像上面案例中在普通节点加入 breakpoint 一样，在 Tool Calling 的架构中很明显呈现出来的特性就是“一刀切”：要么整个流程完全停止，要么全部通过。而考虑现实情况，通常一个 Agent 会包含多达数十甚至上百个工具，但其中可能只有少数涉及到高风险或敏感操作的工具。那么是否可以在保留自主循环的同时，仅在 Agent 自行判断必须要开始执行敏感工具或操作时，才触发人工介入机制？这种做法更符合实际应用需求，由此才有了 dynamic breakpoints（动态断点）的大规模应用。
动态断点指的是在图结构中，可以根据某些条件从给定节点内部动态的来中断图。其实在上面一个案例中已经有所涉及，即：
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并且在多轮对话 run_dialogue 的函数逻辑中，我们是通过代码的逻辑来对不同 user_approval 状态值采用不同的状态更新策略，即：
if last_chunk["user_approval"] == f"用户输入的指令是:{last_chunk['user_input']}, 请人工确认是否执行！":
user_approval = input(f"当前用户的输入是：{last_chunk['user_input']}, 请人工确认是否执行！请回复 是/否。")
graph.update_state(config, {"user_approval": user_approval})五. 实战：实现人机交互式信息管理系统应用
我们可以根据 dynamic breakpoints（动态断点）的思路进一步扩展 Agent 的复杂程度。
下面的案例中，我们实现一个灵活且自主的天气信息管理系统，这个系统通过四个主要的工具函数来操作天气数据，这些函数包括查询、插入、和删除天气信息，以及获取实时天气。其中我们将删除功能作为高危敏感工具，使用 dynamic breakpoints 的机制来控制，仅当用户的需求触发 Agent 判断需要执行该工具时，由人工介入决定是否执行删除数据操作，具体的工具函数描述如下：
获取实时天气（get_weather）：此工具允许用户输入城市名称，通过调用 OpenWeather API 获取该城市当前的天气情况，返回的数据以 JSON 格式呈现，包括温度、天气状况等信息。
- 插入天气信息到数据库（insert_weather_to_db）：这个工具用于将获取的天气数据存入数据库。
- 从数据库查询天气信息（query_weather_from_db）：此工具允许用户通过城市名查询已存储的天气信息。
- 从数据库删除天气信息（delete_weather_from_db）：允许用户删除指定城市的天气信息。
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, declarative_base
# 创建基类
Base = declarative_base()
# 定义 WeatherInfo 模型
class Weather(Base):
# 建表：weather
__tablename__ = 'weather'
# 表中的字段
city_id = Column(Integer, primary_key=True) # 城市ID
city_name = Column(String(50)) # 城市名称
main_weather = Column(String(50)) # 主要天气状况
description = Column(String(100)) # 描述
temperature = Column(Float) # 温度
feels_like = Column(Float) # 体感温度
temp_min = Column(Float) # 最低温度
temp_max = Column(Float) # 最高温度
# 数据库连接 URI，这里要替换成自己的Mysql 连接信息，以下是各个字段的对应解释：
# {acc}: MySQL 数据库的用户名。
# {passwrod}: MySQL 数据库的密码。
# localhost: MySQL 服务器的 IP 地址。
# langgraph: 要连接的数据库的名称。
# charset=utf8mb4: 设置数据库的字符集为 utf8mb4，支持更广泛的 Unicode 字符
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
# 如果表不存在，则创建表
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
import json
class WeatherLoc(BaseModel):
location: str = Field(description="The location name of the city")
class WeatherInfo(BaseModel):
"""Extracted weather information for a specific city."""
city_id: int = Field(..., description="The unique identifier for the city")
city_name: str = Field(..., description="The name of the city")
main_weather: str = Field(..., description="The main weather condition")
description: str = Field(..., description="A detailed description of the weather")
temperature: float = Field(..., description="Current temperature in Celsius")
feels_like: float = Field(..., description="Feels-like temperature in Celsius")
temp_min: float = Field(..., description="Minimum temperature in Celsius")
temp_max: float = Field(..., description="Maximum temperature in Celsius")
class QueryWeatherSchema(BaseModel):
"""Schema for querying weather information by city name."""
city_name: str = Field(..., description="The name of the city to query weather information")
class DeleteWeatherSchema(BaseModel):
"""Schema for deleting weather information by city name."""
city_name: str = Field(..., description="The name of the city to delete weather information")
from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]
## 第一个工具：获取天气信息
@tool(args_schema = WeatherLoc)
def get_weather(location):
"""
Function to query current weather.
:param loc: Required parameter, of type string, representing the specific city name for the weather query. \
Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
the loc parameter should be input as 'Beijing'.
:return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
"""
# Step 1.构建请求
url = "https://api.openweathermap.org/data/2.5/weather"
# Step 2.设置查询参数
params = {
"q": location,
"appid": weather_key, # 输入API key
"units": "metric", # 使用摄氏度而不是华氏度
"lang":"zh_cn" # 输出语言为简体中文
}
# Step 3.发送GET请求
response = requests.get(url, params=params)
# Step 4.解析响应
data = response.json()
return json.dumps(data)
## 第二个工具：往数据库插入天气信息
@tool(args_schema=WeatherInfo)
def insert_weather_to_db(city_id, city_name, main_weather, description, temperature, feels_like, temp_min, temp_max):
"""Insert weather information into the database."""
session = Session() # 确保为每次操作创建新的会话
try:
# 创建天气实例
weather = Weather(
city_id=city_id,
city_name=city_name,
main_weather=main_weather,
description=description,
temperature=temperature,
feels_like=feels_like,
temp_min=temp_min,
temp_max=temp_max
)
# 添加到会话
session.add(weather)
# 提交事务
session.commit()
return {"messages": [f"天气数据已成功存储至Mysql数据库。"]}
except Exception as e:
session.rollback() # 出错时回滚
return {"messages": [f"数据存储失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第三个工具：从数据库查询天气信息
@tool(args_schema=QueryWeatherSchema)
def query_weather_from_db(city_name: str):
"""Query weather information from the database by city name."""
session = Session()
try:
# 查询天气数据
weather_data = session.query(Weather).filter(Weather.city_name == city_name).first()
print(weather_data)
if weather_data:
return {
"city_id": weather_data.city_id,
"city_name": weather_data.city_name,
"main_weather": weather_data.main_weather,
"description": weather_data.description,
"temperature": weather_data.temperature,
"feels_like": weather_data.feels_like,
"temp_min": weather_data.temp_min,
"temp_max": weather_data.temp_max
}
else:
return {"messages": [f"未找到城市 '{city_name}' 的天气信息。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
## 第四个工具：从数据库删除天气信息
@tool(args_schema=DeleteWeatherSchema)
def delete_weather_from_db(city_name: str):
"""Delete weather information from the database by city name."""
session = Session()
try:
# 查询要删除的天气数据
weather_data = session.query(Weather).filter(Weather.city_name == city_name).first()
if weather_data:
# 删除记录
session.delete(weather_data)
session.commit()
return {"messages": [f"城市 '{city_name}' 的天气信息已成功删除。"]}
else:
return {"messages": [f"未找到城市 '{city_name}' 的天气信息。"]}
except Exception as e:
session.rollback() # 出错时回滚
return {"messages": [f"删除失败，错误原因：{e}"]}
finally:
session.close() # 关闭会话
# 使用 ToolNode 构建外部工具库，代码如下:
from langgraph.prebuilt import ToolNode
tools = [get_weather, insert_weather_to_db, query_weather_from_db, delete_weather_from_db]
tool_node = ToolNode(tools)
# 接下来定义用于 Agent 的基座模型，并绑定外部工具库。
import getpass
import json
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
llm = llm.bind_tools(tools)
# call_model 函数用来接收用户的输入请求，由大模型进行用户意图分析。
def call_model(state):
messages = state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
# should_continue 函数为 Router Function，当 Agent 判断用户的需求中需要触发 删除 的高危操作时，则需要进入到高危操作的具体处理策略中。
def should_continue(state):
messages = state["messages"]
last_message = messages[-1]
if not last_message.tool_calls:
return "end"
elif last_message.tool_calls[0]["name"] == "delete_weather_from_db":
return "run_tool"
else:
return "continue"
# run_tool 函数内的逻辑为高危操作工具的执行逻辑，需要人工批准后决定是否执行。
# def run_tool(state):
# new_messages = []
# tool_calls = state["messages"][-1].tool_calls
# # tools = [get_weather, insert_weather_to_db, query_weather_from_db, delete_weather_from_db]
# tools = [delete_weather_from_db]
# tools = {t.name: t for t in tools}
# for tool_call in tool_calls:
# result = tools[tool_call["name"]].invoke(tool_call["args"])
# new_messages.append(
# ToolMessage(content=str(result), name=tool_call["name"], tool_call_id=tool_call["id"])
# )
# return {"messages": new_messages}
# run_tool 函数内的逻辑为高危操作工具的执行逻辑，需要人工批准后决定是否执行。
def run_tool(state):
new_messages = []
tool_calls = state["messages"][-1].tool_calls
# tools = [get_weather, insert_weather_to_db, query_weather_from_db, delete_weather_from_db]
tools = [delete_weather_from_db]
tools = {t.name: t for t in tools}
for tool_call in tool_calls:
tool = tools[tool_call["name"]]
result = tool.invoke(tool_call["args"])
new_messages.append(
{
"role": "tool",
"name": tool_call["name"],
"content": result,
"tool_call_id": tool_call["id"],
}
)
return {"messages": new_messages}
# 节点函数和路由函数定义完毕后，开始构建图的完整结构。代码如下:
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.add_node("run_tool", run_tool)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"run_tool": "run_tool",
"end": END,
},
)
workflow.add_edge("action", "agent")
workflow.add_edge("run_tool", "agent")
# 最后，在编译图的阶段，添加 checkpointer 与具体的 breakpoint。
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory, interrupt_before=["run_tool"])
from IPython.display import Image, display
# 生成可视化图像结构
display(Image(graph.get_graph().draw_mermaid_png()))
"""
为什么我把insert_weather_to_db这个工具的提示词注释了，# "Insert weather information into the database."，
但是大模型还是理解了，然后往数据库中插入了指定数据。这是为什么？
因为这个工具函数上被注释掉的只是函数体内的 docstring（insert_weather_to_db 的说明），
但大模型判断"该调用哪个工具、传什么参数"依赖的信息不止这一处，还有几个关键点仍然完整保留：
1、@tool(args_schema=WeatherInfo) 里的 WeatherInfo 模型没有被注释，其中的 WeatherInfo 类：
类本身有 docstring："Extracted weather information for a specific city."
每个字段都有 Field(..., description="...")，比如 city_id、city_name、temperature 等都写清楚了含义。
LangChain 在把工具信息传给大模型（生成 function-calling schema）时，参数部分的描述是从 args_schema
（即 WeatherInfo）里提取的，而不是从函数体的 docstring 提取。所以即使函数自身没有描述，
模型依然能通过参数 schema 的 field description 知道每个参数是什么、该传什么值。
2、函数名 insert_weather_to_db 本身就是强语义线索
大模型（尤其是 GPT-4o 这类）对函数名、类名的语义非常敏感。insert_weather_to_db 这个命名已经清楚表达了"把天气数据插入数据库"
这个意图，即使没有 docstring，模型也能靠命名 + 参数 schema 推断出这是"存储天气数据"的工具。
3、工具描述缺失时 LangChain 不会报错，而是用其他信息兜底
@tool 装饰器生成 StructuredTool 时，如果函数没有 docstring，description 字段可能为空或退化处理，
但只要 args_schema 存在且有自己的 docstring/字段描述，模型在做 tool selection 时依然能拿到足够的结构化信息（工具名 + 参数 schema），而不是完全没有描述。
结论：注释掉的只是"工具的自然语言功能说明"这一层信息，但工具名称 + args_schema（WeatherInfo 类的 docstring 和各字段的 description）
仍然完整地把"这是做什么用的、需要哪些字段"传递给了大模型，所以它依然能正确判断要调用这个工具并插入数据。
如果想验证"提示词到底重不重要"，可以把 WeatherInfo 里的字段描述和类 docstring 也一并去掉，可以看到模型直接报错：
ValueError: Function must have a docstring if description not provided.
"""
config = {"configurable": {"thread_id": "100"}}
# for chunk in graph.stream({"messages": "天津的天气怎么样？并把数据插入数据库中"}, config, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
"""
在这个过程中，依次执行了 query_weather_from_db、get_weather 工具用来获取当前的天气信息，并且将查询到的数据实时的插入到了数据库中，我们可以进行数据验证：
继续加大问题的难度，如下所示：
"""
# 配置会话线程ID
# 流式执行工作流，查询上海、杭州天气并对比出游适宜度
# config = {"configurable": {"thread_id": "100"}}
# for chunk in graph.stream({"messages": "帮我同时查一下上海、杭州的天气，比较哪个城市更适合现在出游。然后把查询到的数据插入数据库中"}, config, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
# 从输出上，该系统同样很好的完成了我们提出的所有需求。而接下来，我们触发高危操作，让其删除数据库中的一些数据，请求如下:
# config = {"configurable": {"thread_id": "10"}}
# 看清楚是中文地名还是被翻译成英文地名。参考实际数据库
# for chunk in graph.stream({"messages": "帮我删除数据库中Tianjin的天气数据"}, config, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
# 在这样的请求中，当 Agent 判断需要执行 delete_weather_from_db 工具时，并没有自主执行，而是进入到 breakpoint 状态，等待人工介入。我们可以通过如下方式查看到当前的图状态：
# 打印图当前的状态：
# state = graph.get_state(config)
# print(f"图的下一个节点是:\n",state.next)
# print(f"图当前正在执行的任务是:\n",state.tasks)
# print(f"图当前正在执行的节点的内容是:\n",state.values)
# 首先还是同样，如果允许删除操作，仍然在 input 参数中填写 None，将上述全部状态信息传递到图中使其恢复中断的执行，如下所示：
# for chunk in graph.stream(None, config, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
"""
接下来，我们就可以按照这种思路，去构建完全自动化的具有人工介入的自主循环代理工作流。这个过程的关键是：如何去处理执行高危工具或者放弃执行高危工具的操作逻辑。
解决的思路是：当从用户那里获得自然语言反馈是不允许执行高危操作的时候，我们可以将这些反馈作为工具调用的模拟数据进行存入，代码如下所示：
"""
def human_review():
config = {"configurable": {"thread_id": "100"}}
for chunk in graph.stream({"messages": "帮我删除数据库中Shanghai的天气数据"}, config, stream_mode="values"):
state = graph.get_state(config)
# print(state.next)
# print(state.tasks)
# 检查是否有任务，如果没有则结束循环
if not state.tasks:
# print("所有任务都已完成。")
chunk["messages"][-1].pretty_print()
break
if state.tasks[0].name == 'run_tool':
while True:
user_input = input("是否允许执行删除操作？请输入'是'或'否'：")
if user_input in ["是", "否"]:
break
else:
print("输入错误，请输入'是'或'否'。")
if user_input == "是":
graph.update_state(config=config, values=chunk)
for event in graph.stream(None, config, stream_mode="values"):
event["messages"][-1].pretty_print()
elif user_input == "否":
state = graph.get_state(config)
tool_call_id = state.values["messages"][-1].tool_calls[0]["id"]
print(tool_call_id)
#我们现在需要构造一个替换工具调用，把参数改为"xxsd"，请注意，我们可以更改任意数量的参数或工具名称-它必须是一个有效的
new_message = {
"role": "tool",
# 这是得到的用户不允许操作的反馈
"content": "管理员不允许执行该操作！",
"name": "delete_weather_from_db",
"tool_call_id": tool_call_id,
}
graph.update_state(config, {"messages": [new_message]}, as_node="run_tool",)
for event in graph.stream(None, config, stream_mode="values"):
event["messages"][-1].pretty_print()
"""注意，这里的关键是需要更新 state。对于模拟的工具数据来说，需要传递一条消息，该消息的 ID 与要响应的工具调用 ID 相同。
接下来我们就可以基于这套逻辑
构建具有人机交互的多轮对话代理：
"""
def run_multi_round_dialogue(graph, config):
while True: # 开始多轮循环
# 询问用户输入操作，允许退出
user_input = input("请输入您的问题（例如：'帮我查询上海的天气数据'），输入'退出'结束对话：")
# 检查是否退出对话
if user_input.lower() == '退出':
print("对话已结束。")
break
# 启动对话，根据用户的输入进行处理
for chunk in graph.stream({"messages": user_input}, config, stream_mode="values"):
state = graph.get_state(config)
# 如果没有任务则结束这一轮循环
if not state.tasks:
if "messages" in chunk and len(chunk["messages"]) > 0:
print("人工智能助理：", chunk["messages"][-1].content)
break
# 处理动态断点任务
if state.tasks[0].name == 'run_tool':
user_approval = None
while True:
user_approval = input("是否允许执行删除操作？请输入'是'或'否'：")
if user_approval in ["是", "否"]:
break
else:
print("输入错误，请输入'是'或'否'。")
if user_approval == "是":
graph.update_state(config=config, values=chunk)
for event in graph.stream(None, config, stream_mode="values"):
if "messages" in event and len(event["messages"]) > 0:
print("人工智能助理：", event["messages"][-1].content)
# event["messages"][-1].pretty_print()
elif user_approval == "否":
state = graph.get_state(config)
tool_call_id = state.values["messages"][-1].tool_calls[0]["id"]
# 构造一个反馈消息来停止操作
new_message = {
"role": "tool",
"content": "管理员不允许执行该操作！",
"name": "delete_weather_from_db",
"tool_call_id": tool_call_id,
}
graph.update_state(config, {"messages": [new_message]}, as_node="run_tool")
for event in graph.stream(None, config, stream_mode="values"):
if "messages" in event and len(event["messages"]) > 0:
print("人工智能助理：", event["messages"][-1].content)
# event["messages"][-1].pretty_print()
if __name__ == "__main__":
config = {"configurable": {"thread_id": "102"}}
# 使用该函数启动对话
run_multi_round_dialogue(graph, config)