LangGraph工作流人机交互 - AI创想

from typing import TypedDict
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):
"""图形状态。"""
some_text: str
def human_node(state: State):
value = interrupt(
# 任何可序列化为 JSON 的值，供人类查看。
# 例如，一个问题、一段文本或状态中的一组键
{
"text_to_revise": state["some_text"]
}
)
return {
# 使用人类的输入更新状态
"some_text": value
}
# 构建图形
graph_builder = StateGraph(State)
# 将人类节点添加到图形中
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")
# 使用中断功能需要一个检查点。
checkpointer = MemorySaver()
graph = graph_builder.compile(
checkpointer=checkpointer
)
thread_config = {"configurable": {"thread_id": "abc123"}}# 使用 stream() 直接展示 `__interrupt__` 信息。
for chunk in graph.stream({"some_text": "原始文本"}, config=thread_config):
print(chunk)
# 使用 Command 恢复
for chunk in graph.stream(Command(resume="编辑后的文本"), config=thread_config):
print(chunk)

复制代码

from __future__ import annotations
from typing import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from pyexpat.errors import messages
# 1. 状态定义
class State(TypedDict):
messages: list
customer_id: str | None
user_reply: str | None # 用于接住中断回复
# 2. 节点：未验证则中断
def verify_info(state: State):
if state["customer_id"] is not None:
return state
reply = interrupt({"user_reply": "请输入手机号"}) # key 必须和 State 对应
return {"messages":"需要输入手机号","user_reply": reply, "customer_id": reply}
# 3. 节点：根据手机号前缀决定回复
def supervisor(state: State):
if state["customer_id"].startswith(("138", "188")):
return {"messages": ["超级碗"]}
else:
return {"messages": ["啥也不是"]}
# 4. 构图
builder = StateGraph(State)
builder.add_node("verify", verify_info)
builder.add_node("supervisor", supervisor)
builder.add_edge(START, "verify")
builder.add_edge("verify", "supervisor")
builder.add_edge("supervisor", END)
graph = builder.compile(checkpointer=MemorySaver())
# 5. 第一次运行（中断前）
thread = {"configurable": {"thread_id": "1"}}
state1 = graph.invoke({"messages": [], "customer_id": None, "user_reply": None}, thread)
print("中断前状态 ->", state1)
# 6. 人工输入后续跑
resume_cmd = Command(resume="13800138000")
final_state = graph.invoke(resume_cmd, thread)
print("最终状态 ->", final_state)

复制代码