1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
| from langchain.messages import SystemMessage
from langchain_core.messages import ToolMessage
from langchain.tools import tool
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt, Command
llm = ChatOllama(model="qwen3.5:9b")
checkpointer = InMemorySaver()
@tool
def send_email(to: str, subject: str) -> str:
"""发送一封邮件（敏感操作，需要人工确认）。to=收件人，subject=主题。"""
return f"邮件已发送给 {to}，主题：{subject}"
@tool
def search_web(query: str) -> str:
"""搜索网页资料（普通操作，直接执行，无需人工确认）。"""
return f"关于 '{query}' 的搜索结果摘要……"
tools = [send_email, search_web]
llm_with_tools = llm.bind_tools(tools)
SENSITIVE_TOOLS = {"send_email"}
SYSTEM_PROMPT = (
"你是一个助手。当用户要发邮件时调用 send_email；当用户要查资料时调用 search_web。"
)
def agent(state: MessagesState) -> dict:
messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def human_approval(state: MessagesState) -> dict:
last = state["messages"][-1]
tool_call = last.tool_calls[0]
decision = interrupt({
"action": "请人工确认是否执行该敏感操作",
"tool": tool_call["name"],
"args": tool_call["args"],
})
if decision == "approve":
return ToolNode(tools).invoke(state)
else:
return {"messages": [ToolMessage(
content="用户拒绝了该敏感操作，未执行。",
tool_call_id=tool_call["id"],
)]}
def route_after_agent(state: MessagesState) -> str:
last = state["messages"][-1]
if getattr(last, "tool_calls", None):
name = last.tool_calls[0]["name"]
return "human_approval" if name in SENSITIVE_TOOLS else "tools"
return END
builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("human_approval", human_approval)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_after_agent)
builder.add_edge("tools", "agent")
builder.add_edge("human_approval", "agent")
graph = builder.compile(checkpointer=checkpointer)
def run_with_human(thread_id: str, user_req: str, human_decision: str) -> None:
config = {"configurable": {"thread_id": thread_id}}
result = graph.invoke({"messages": [{"role": "user", "content": user_req}]}, config)
if "__interrupt__" in result:
req = result["__interrupt__"][0].value
print(f"[线程 {thread_id}] ⏸ 人工审核请求: {req}")
print(f"[线程 {thread_id}] 人工决定: {human_decision}")
result = graph.invoke(Command(resume=human_decision), config)
else:
print(f"[线程 {thread_id}] (未触发人工审核)")
print(f"[线程 {thread_id}] 最终回答: {result['messages'][-1].content}\n")
if __name__ == "__main__":
print(graph.get_graph().draw_mermaid())
run_with_human(
"demo-approve",
"帮我给 lisi@company.com 发一封邮件，主题是：项目进度同步",
"approve",
)
run_with_human(
"demo-reject",
"帮我给 boss@company.com 发邮件，主题是：请假申请",
"reject",
)
|