启用人工干预

from typing import Literal, TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

# 定义共享图状态
class State(TypedDict):
    llm_output: str
    decision: str

# 模拟 LLM 输出节点
def generate_llm_output(state: State) -> State:
    return {"llm_output": "这是生成的输出。"}

# 人工批准节点
def human_approval(state: State) -> Command[Literal["approved_path", "rejected_path"]]:
    decision = interrupt({
        "question": "您是否批准以下输出？",
        "llm_output": state["llm_output"]
    })

    if decision == "approve":
        return Command(goto="approved_path", update={"decision": "approved"})
    else:
        return Command(goto="rejected_path", update={"decision": "rejected"})

# 批准后的后续步骤
def approved_node(state: State) -> State:
    print("✅ 采取了批准路径。")
    return state

# 拒绝后的替代路径
def rejected_node(state: State) -> State:
    print("❌ 采取了拒绝路径。")
    return state

# 构建图
builder = StateGraph(State)
builder.add_node("generate_llm_output", generate_llm_output)
builder.add_node("human_approval", human_approval)
builder.add_node("approved_path", approved_node)
builder.add_node("rejected_path", rejected_node)

builder.set_entry_point("generate_llm_output")
builder.add_edge("generate_llm_output", "human_approval")
builder.add_edge("approved_path", END)
builder.add_edge("rejected_path", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 运行直到中断
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)
print(result["__interrupt__"])
# 输出：
# Interrupt(value={'question': '您是否批准以下输出？', 'llm_output': '这是生成的输出。'}, ...)

# 模拟使用人工输入恢复
# 要测试拒绝，请将 resume="approve" 替换为 resume="reject"
final_result = graph.invoke(Command(resume="approve"), config=config)
print(final_result)

from typing import TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

# 定义图状态
class State(TypedDict):
    summary: str

# 模拟 LLM 摘要生成
def generate_summary(state: State) -> State:
    return {
        "summary": "猫坐在垫子上，望着星星。"
    }

# 人工审查编辑节点
def human_review_edit(state: State) -> State:
    result = interrupt({
        "task": "请审查并根据需要编辑生成的摘要。",
        "generated_summary": state["summary"]
    })
    return {
        "summary": result["edited_summary"]
    }

# 模拟下游对编辑后摘要的使用
def downstream_use(state: State) -> State:
    print(f"✅ 使用编辑后的摘要：{state['summary']}")
    return state

# 构建图
builder = StateGraph(State)
builder.add_node("generate_summary", generate_summary)
builder.add_node("human_review_edit", human_review_edit)
builder.add_node("downstream_use", downstream_use)

builder.set_entry_point("generate_summary")
builder.add_edge("generate_summary", "human_review_edit")
builder.add_edge("human_review_edit", "downstream_use")
builder.add_edge("downstream_use", END)

# 设置内存检查点以支持中断
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 调用图直到触发中断
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)

# 输出中断有效载荷
print(result["__interrupt__"])
# 示例输出：
# > [
# >     Interrupt(
# >         value={
# >             'task': '请审查并根据需要编辑生成的摘要。',
# >             'generated_summary': '猫坐在垫子上，望着星星。'
# >         },
# >         id='...'
# >     )
# > ]

# 使用人工编辑的输入恢复图
edited_summary = "猫躺在地毯上，平静地凝视着夜空。"
resumed_result = graph.invoke(
    Command(resume={"edited_summary": edited_summary}),
    config=config
)
print(resumed_result)

from typing import TypedDict
import uuid

from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

# 定义图状态
class State(TypedDict):
    age: int

# 请求人类输入并验证它的节点
def get_valid_age(state: State) -> State:
    prompt = "请输入您的年龄（必须是非负整数）。"

    while True:
        user_input = interrupt(prompt)

        # 验证输入
        try:
            age = int(user_input)
            if age < 0:
                raise ValueError("年龄必须是非负数。")
            break  # 收到有效输入
        except (ValueError, TypeError):
            prompt = f"'{user_input}' 无效。请输入一个非负整数作为年龄。"

    return {"age": age}

# 使用有效输入的节点
def report_age(state: State) -> State:
    print(f"✅ 人类年龄为 {state['age']} 岁。")
    return state

# 构建图
builder = StateGraph(State)
builder.add_node("get_valid_age", get_valid_age)
builder.add_node("report_age", report_age)

builder.set_entry_point("get_valid_age")
builder.add_edge("get_valid_age", "report_age")
builder.add_edge("report_age", END)

# 使用内存检查点创建图
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 运行图直到第一次中断
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)
print(result["__interrupt__"])  # 第一次提示：“请输入您的年龄...”

# 模拟无效输入（例如，字符串而不是整数）
result = graph.invoke(Command(resume="not a number"), config=config)
print(result["__interrupt__"])  # 带验证消息的后续提示

# 模拟第二次无效输入（例如，负数）
result = graph.invoke(Command(resume="-10"), config=config)
print(result["__interrupt__"])  # 另一次重试

# 提供有效输入
final_result = graph.invoke(Command(resume="25"), config=config)
print(final_result)  # 应包含有效年龄

from IPython.display import Image, display
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    input: str


def step_1(state):
    print("---Step 1---")
    pass


def step_2(state):
    print("---Step 2---")
    pass


def step_3(state):
    print("---Step 3---")
    pass


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# 设置检查点
checkpointer = InMemorySaver() # (1)!

graph = builder.compile(
    checkpointer=checkpointer, # (2)!
    interrupt_before=["step_3"] # (3)!
)

# 查看
display(Image(graph.get_graph().draw_mermaid_png()))


# 输入
initial_input = {"input": "hello world"}

# 线程
thread = {"configurable": {"thread_id": "1"}}

# 运行图直到第一次中断
for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)

# 这将运行直到断点
# 此时你可以获取图的状态
print(graph.get_state(config))

# 你可以通过传入 `None` 作为输入来继续图的执行
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

import uuid
from typing import TypedDict

from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver


class State(TypedDict):
    """图状态。"""
    state_counter: int


counter_node_in_subgraph = 0

def node_in_subgraph(state: State):
    """子图中的节点。"""
    global counter_node_in_subgraph
    counter_node_in_subgraph += 1  # 这段代码将**不会**再次运行！
    print(f"总共进入了 `node_in_subgraph` {counter_node_in_subgraph} 次")

counter_human_node = 0

def human_node(state: State):
    global counter_human_node
    counter_human_node += 1 # 这段代码将再次运行！
    print(f"总共进入了子图中的 human_node {counter_human_node} 次")
    answer = interrupt("你叫什么名字？")
    print(f"得到了答案 {answer}")


checkpointer = InMemorySaver()

subgraph_builder = StateGraph(State)
subgraph_builder.add_node("some_node", node_in_subgraph)
subgraph_builder.add_node("human_node", human_node)
subgraph_builder.add_edge(START, "some_node")
subgraph_builder.add_edge("some_node", "human_node")
subgraph = subgraph_builder.compile(checkpointer=checkpointer)


counter_parent_node = 0

def parent_node(state: State):
    """这个父节点将调用子图。"""
    global counter_parent_node

    counter_parent_node += 1 # 在恢复时这段代码将再次运行！
    print(f"总共进入了 `parent_node` {counter_parent_node} 次")

    # 请注意，我们有意增加图状态中的状态计数器
    # 以证明子图对相同键的更新不会与父图冲突（直到
    subgraph_state = subgraph.invoke(state)
    return subgraph_state


builder = StateGraph(State)
builder.add_node("parent_node", parent_node)
builder.add_edge(START, "parent_node")

# 必须启用检查点才能使中断正常工作！
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
      "thread_id": uuid.uuid4(),
    }
}

for chunk in graph.stream({"state_counter": 1}, config):
    print(chunk)

print('--- 恢复 ---')

for chunk in graph.stream(Command(resume="35"), config):
    print(chunk)

总共进入了 `parent_node` 1 次
总共进入了 `node_in_subgraph` 1 次
总共进入了子图中的 human_node 1 次
{'__interrupt__': (Interrupt(value='你叫什么名字？', id='...'),)}
--- 恢复 ---
总共进入了 `parent_node` 2 次
总共进入了子图中的 human_node 2 次
得到了答案 35
{'parent_node': {'state_counter': 1}}

import uuid
from typing import TypedDict, Optional

from langgraph.graph import StateGraph
from langgraph.constants import START
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver


class State(TypedDict):
    """图状态。"""

    age: Optional[str]
    name: Optional[str]


def human_node(state: State):
    if not state.get('name'):
        name = interrupt("你叫什么名字？")
    else:
        name = "N/A"

    if not state.get('age'):
        age = interrupt("你的年龄是多少？")
    else:
        age = "N/A"

    print(f"姓名: {name}. 年龄: {age}")

    return {
        "age": age,
        "name": name,
    }


builder = StateGraph(State)
builder.add_node("human_node", human_node)
builder.add_edge(START, "human_node")

# 必须启用检查点才能使中断正常工作！
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": uuid.uuid4(),
    }
}

for chunk in graph.stream({"age": None, "name": None}, config):
    print(chunk)

for chunk in graph.stream(Command(resume="John", update={"name": "foo"}), config):
    print(chunk)

{'__interrupt__': (Interrupt(value='你叫什么名字？', id='...'),)}
姓名: N/A. 年龄: John
{'human_node': {'age': 'John', 'name': 'N/A'}}