添加人机协作环节¶

from typing import TypedDict
import uuid

from langgraph.checkpoint.memory import InMemorySaver
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( # (1)!
        {
            "text_to_revise": state["some_text"] # (2)!
        }
    )
    return {
        "some_text": value # (3)!
    }


# Build the graph
graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")

checkpointer = InMemorySaver() # (4)!

graph = graph_builder.compile(checkpointer=checkpointer)

# Pass a thread ID to the graph to run it.
config = {"configurable": {"thread_id": uuid.uuid4()}}

# Run the graph until the interrupt is hit.
result = graph.invoke({"some_text": "original text"}, config=config) # (5)!

print(result['__interrupt__']) # (6)!
# > [
# >    Interrupt(
# >       value={'text_to_revise': 'original text'}, 
# >       resumable=True,
# >       ns=['human_node:6ce9e64f-edef-fe5d-f7dc-511fa9526960']
# >    )
# > ] 

print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
# > {'some_text': 'Edited text'}

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 MemorySaver

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

# 模拟LLM输出节点
def generate_llm_output(state: State) -> State:
    return {"llm_output": "This is the generated output."}

# 人类批准节点
def human_approval(state: State) -> Command[Literal["approved_path", "rejected_path"]]:
    decision = interrupt({
        "question": "Do you approve the following output?",
        "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("✅ Approved path taken.")
    return state

# 拒绝后的替代路径
def rejected_node(state: State) -> State:
    print("❌ Rejected path taken.")
    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 = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 运行直到中断
config = {"configurable": {"thread_id": uuid.uuid4()}}
result = graph.invoke({}, config=config)
print(result["__interrupt__"])
# 输出：
# Interrupt(value={'question': 'Do you approve the following output?', 'llm_output': 'This is the generated 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 MemorySaver

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

# 模拟LLM摘要生成
def generate_summary(state: State) -> State:
    return {
        "summary": "The cat sat on the mat and looked at the stars."
    }

# 人类编辑节点
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"✅ Using edited summary: {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 = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

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

# 输出中断负载
print(result["__interrupt__"])
# 示例输出：
# Interrupt(
#   value={
#     'task': 'Please review and edit the generated summary if necessary.',
#     'generated_summary': 'The cat sat on the mat and looked at the stars.'
#   },
#   resumable=True,
#   ...
# )

# 使用人类编辑的输入恢复图
edited_summary = "The cat lay on the rug, gazing peacefully at the night sky."
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 MemorySaver

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

# 请求人类输入并验证它的节点
def get_valid_age(state: State) -> State:
    prompt = "Please enter your age (must be a non-negative integer)."

    while True:
        user_input = interrupt(prompt)

        # 验证输入
        try:
            age = int(user_input)
            if age < 0:
                raise ValueError("Age must be non-negative.")
            break  # 接收到有效输入
        except (ValueError, TypeError):
            prompt = f"'{user_input}' is not valid. Please enter a non-negative integer for age."

    return {"age": age}

# 使用有效输入的节点
def report_age(state: State) -> State:
    print(f"✅ Human is {state['age']} years old.")
    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 = MemorySaver()
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)  # 应包含有效的年龄

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 MemorySaver


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"总共进入了子图中的人类节点 {counter_human_node} 次")
    answer = interrupt("你的名字是什么？")
    print(f"得到了答案 {answer}")


checkpointer = MemorySaver()

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 = MemorySaver()
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)

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 MemorySaver


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 = MemorySaver()
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)