Add multiple LangGraph examples including basic agents, chatbots, conditional edges, checkpointers, tool calls, and human-in-the-loop approval processes. Update .gitignore to include local environment files.

.gitignore

@@ -16,6 +16,9 @@ __pycache__/
 venv/
 env/
 
+# Local uv environment/version files
+.python-version
+
 # IDE and OS files
 .vscode/
 .idea/

01_basic_agent_no_llm.py

@@ -0,0 +1,38 @@
+from typing import TypedDict
+
+from langgraph.graph import StateGraph, START, END
+
+
+class State(TypedDict):
+    text: str
+    length: int
+
+
+def normalize_text(state: State):
+    """清理文本。"""
+    normalized = state["text"].strip().lower()
+    return {"text": normalized}
+
+
+def calculate_length(state: State):
+    """计算文本长度。"""
+    return {"length": len(state["text"])}
+
+
+builder = StateGraph(State)
+
+builder.add_node("normalize_text", normalize_text)
+builder.add_node("calculate_length", calculate_length)
+
+builder.add_edge(START, "normalize_text")
+builder.add_edge("normalize_text", "calculate_length")
+builder.add_edge("calculate_length", END)
+
+graph = builder.compile()
+
+result = graph.invoke({
+    "text": "  Hello LangGraph!  ",
+    "length": 0,
+})
+
+print(result)

02_chat_graph.py

@@ -0,0 +1,41 @@
+from typing import Annotated, TypedDict
+
+from langchain_openai import ChatOpenAI
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+
+
+class State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+model = ChatOpenAI(
+    model="deepseek-v4-flash",
+    temperature=0,
+    base_url="https://api.deepseek.com/v1",
+)
+
+
+def chatbot(state: State):
+    response = model.invoke(state["messages"])
+    return {"messages": [response]}
+
+
+builder = StateGraph(State)
+
+builder.add_node("chatbot", chatbot)
+builder.add_edge(START, "chatbot")
+builder.add_edge("chatbot", END)
+
+graph = builder.compile()
+
+result = graph.invoke({
+    "messages": [
+        {
+            "role": "user",
+            "content": "请用一句话解释 LangGraph。",
+        }
+    ]
+})
+
+print(result["messages"][-1].content)

03_conditional_edge.py

@@ -0,0 +1,54 @@
+from typing import Literal, TypedDict
+
+from langgraph.graph import StateGraph, START, END
+
+
+class State(TypedDict):
+    score: int
+    result: str
+
+
+def evaluate(state: State):
+    return {}
+
+
+def success(state: State):
+    return {"result": "通过"}
+
+
+def retry(state: State):
+    return {"result": "需要重试"}
+
+
+def route_after_evaluate(
+    state: State,
+) -> Literal["success", "retry"]:
+    if state["score"] >= 60:
+        return "success"
+    return "retry"
+
+
+builder = StateGraph(State)
+
+builder.add_node("evaluate", evaluate)
+builder.add_node("success", success)
+builder.add_node("retry", retry)
+
+builder.add_edge(START, "evaluate")
+
+builder.add_conditional_edges(
+    "evaluate",
+    route_after_evaluate,
+    {
+        "success": "success",
+        "retry": "retry",
+    },
+)
+
+builder.add_edge("success", END)
+builder.add_edge("retry", END)
+
+graph = builder.compile()
+
+print(graph.invoke({"score": 80, "result": ""}))
+print(graph.invoke({"score": 40, "result": ""}))

04_checkpointer.py

@@ -0,0 +1,64 @@
+from typing import Annotated, TypedDict
+
+from langchain_openai import ChatOpenAI
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+
+
+class State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+model = ChatOpenAI(
+    model="deepseek-v4-flash",
+    temperature=0,
+    base_url="https://api.deepseek.com/v1"
+)
+
+
+def chatbot(state: State):
+    return {
+        "messages": [model.invoke(state["messages"])]
+    }
+
+
+builder = StateGraph(State)
+builder.add_node("chatbot", chatbot)
+builder.add_edge(START, "chatbot")
+builder.add_edge("chatbot", END)
+
+checkpointer = MemorySaver()
+graph = builder.compile(checkpointer=checkpointer)
+
+config = {
+    "configurable": {
+        "thread_id": "user-001",
+    }
+}
+
+first = graph.invoke(
+    {
+        "messages": [
+            {
+                "role": "user",
+                "content": "我叫小明，请记住。",
+            }
+        ]
+    },
+    config=config,
+)
+
+second = graph.invoke(
+    {
+        "messages": [
+            {
+                "role": "user",
+                "content": "我叫什么名字？",
+            }
+        ]
+    },
+    config=config,
+)
+
+print(second["messages"][-1].content)

05_tool_call.py

@@ -0,0 +1,143 @@
+"""LangGraph 工具调用示例：天气查询 + 计算器。
+
+图结构：
+
+    START -> model
+               | 有 tool_calls
+               v
+             tools
+               |
+               +---------> model
+
+             model -- 无 tool_calls --> END
+
+天气数据是本地模拟数据，重点是学习 LangGraph 的工具调用循环。
+"""
+
+import os
+from typing import Literal
+
+from dotenv import load_dotenv
+from langchain_core.tools import tool
+from langchain_openai import ChatOpenAI
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.prebuilt import ToolNode
+
+load_dotenv()
+
+
+# 1. 定义工具。类型注解和 docstring 会变成提供给模型的工具 schema。
+@tool
+def get_weather(city: str) -> str:
+    """查询指定城市当前的天气。城市名称使用中文，例如北京、上海。"""
+    weather_data = {
+        "北京": "晴，25°C，北风 2 级",
+        "上海": "多云，28°C，东南风 3 级",
+        "广州": "雷阵雨，30°C，南风 2 级",
+        "深圳": "阴，29°C，东风 2 级",
+    }
+    return weather_data.get(city, f"暂时没有 {city} 的天气数据")
+
+
+@tool
+def calculator(a: float, b: float, operation: Literal["add", "subtract", "multiply", "divide"]) -> float:
+    """对两个数字进行加、减、乘、除运算。
+
+    Args:
+        a: 第一个数字。
+        b: 第二个数字。
+        operation: 运算类型：add、subtract、multiply 或 divide。
+    """
+    if operation == "add":
+        return a + b
+    if operation == "subtract":
+        return a - b
+    if operation == "multiply":
+        return a * b
+    if b == 0:
+        raise ValueError("除数不能为 0")
+    return a / b
+
+
+# 2. 将工具绑定给模型。绑定后，模型可以在 AIMessage.tool_calls 中请求工具。
+tools = [get_weather, calculator]
+model = ChatOpenAI(
+    model=os.getenv("OPENAI_MODEL", "openai/gpt-5.4"),
+    api_key=os.getenv("OPENAI_API_KEY"),
+    base_url=os.getenv("OPENAI_BASE_URL", "https://api.qnaigc.com/v1"),
+    temperature=0,
+    max_retries=2,
+)
+model_with_tools = model.bind_tools(tools)
+
+
+# 3. 模型节点：读取完整消息历史，返回一条新的 AIMessage。
+def call_model(state: MessagesState) -> dict:
+    response = model_with_tools.invoke(state["messages"])
+    return {"messages": [response]}
+
+
+# 4. 路由函数：检查模型是否发起了工具调用。
+def route_after_model(state: MessagesState) -> Literal["tools", "__end__"]:
+    last_message = state["messages"][-1]
+    if last_message.tool_calls:
+        return "tools"
+    return END
+
+
+# 5. ToolNode 自动完成：找到工具、传入参数、执行工具、生成 ToolMessage。
+tool_node = ToolNode(tools)
+
+builder = StateGraph(MessagesState)
+builder.add_node("model", call_model)
+builder.add_node("tools", tool_node)
+
+builder.add_edge(START, "model")
+builder.add_conditional_edges(
+    "model",
+    route_after_model,
+    {
+        "tools": "tools",
+        END: END,
+    },
+)
+builder.add_edge("tools", "model")
+
+graph = builder.compile()
+
+
+def ask(question: str) -> str:
+    """调用助手，并打印每个节点产生的更新。"""
+    final_answer = ""
+
+    for event in graph.stream(
+        {"messages": [{"role": "user", "content": question}]},
+        stream_mode="updates",
+    ):
+        for node_name, update in event.items():
+            print(f"\n[{node_name}]")
+            for message in update.get("messages", []):
+                message.pretty_print()
+                if node_name == "model" and not message.tool_calls:
+                    final_answer = str(message.content)
+
+    return final_answer
+
+
+def main() -> None:
+    print("=== 天气 + 计算器工具助手 ===")
+    print("输入 quit 退出。")
+
+    while True:
+        question = input("\n用户：").strip()
+        if question.lower() in {"quit", "exit", "q"}:
+            break
+        if not question:
+            continue
+
+        answer = ask(question)
+        print(f"\n最终回答：{answer}")
+
+
+if __name__ == "__main__":
+    main()

06_human_in_the_loop.py

@@ -0,0 +1,315 @@
+"""LangGraph 人工审批示例：普通工具直接执行，敏感工具先暂停审批。
+
+图结构：
+
+    START -> model
+               | 无 tool_calls ------------------------> END
+               | 普通工具调用 -> tools ----------------> model
+               | 敏感工具调用 -> approval
+                                    | 批准 -> tools ----> model
+                                    | 拒绝 -> model
+
+运行：
+    uv run --with langgraph --with langchain-openai --with python-dotenv \
+        python 06_human_in_the_loop.py
+
+说明：send_email 是模拟工具，不会真正发送邮件。
+"""
+
+import os
+import uuid
+from typing import Literal
+
+from dotenv import load_dotenv
+from langchain_core.messages import ToolMessage
+from langchain_core.tools import tool
+from langchain_openai import ChatOpenAI
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.prebuilt import ToolNode
+from langgraph.types import Command, interrupt
+
+load_dotenv()
+
+
+# -----------------------------------------------------------------------------
+# 1. 定义工具
+# -----------------------------------------------------------------------------
+
+
+@tool
+def get_weather(city: str) -> str:
+    """查询指定城市当前的天气。城市名称使用中文，例如北京、上海。"""
+    weather_data = {
+        "北京": "晴，25°C，北风 2 级",
+        "上海": "多云，28°C，东南风 3 级",
+        "广州": "雷阵雨，30°C，南风 2 级",
+        "深圳": "阴，29°C，东风 2 级",
+    }
+    return weather_data.get(city, f"暂时没有 {city} 的天气数据")
+
+
+@tool
+def calculator(
+    a: float,
+    b: float,
+    operation: Literal["add", "subtract", "multiply", "divide"],
+) -> float:
+    """对两个数字进行加、减、乘、除运算。
+
+    Args:
+        a: 第一个数字。
+        b: 第二个数字。
+        operation: 运算类型：add、subtract、multiply 或 divide。
+    """
+    if operation == "add":
+        return a + b
+    if operation == "subtract":
+        return a - b
+    if operation == "multiply":
+        return a * b
+    if b == 0:
+        raise ValueError("除数不能为 0")
+    return a / b
+
+
+@tool
+def send_email(recipient: str, subject: str, body: str) -> str:
+    """向指定邮箱发送邮件。这是有外部副作用的敏感操作，执行前必须审批。
+
+    Args:
+        recipient: 收件人的邮箱地址。
+        subject: 邮件主题。
+        body: 邮件正文。
+    """
+    # 教学示例：这里只模拟发送，不连接真实邮件服务。
+    print("\n[模拟发送邮件]")
+    print(f"收件人：{recipient}")
+    print(f"主题：{subject}")
+    print(f"正文：{body}")
+    return f"邮件已成功发送给 {recipient}"
+
+
+TOOLS = [get_weather, calculator, send_email]
+SENSITIVE_TOOL_NAMES = {"send_email"}
+
+
+# -----------------------------------------------------------------------------
+# 2. 创建模型并绑定工具
+# -----------------------------------------------------------------------------
+
+model = ChatOpenAI(
+    model=os.getenv("OPENAI_MODEL", "openai/gpt-5.4"),
+    api_key=os.getenv("OPENAI_API_KEY"),
+    base_url=os.getenv("OPENAI_BASE_URL", "https://api.qnaigc.com/v1"),
+    temperature=0,
+    max_retries=2,
+)
+model_with_tools = model.bind_tools(TOOLS)
+
+
+# -----------------------------------------------------------------------------
+# 3. 定义节点与路由
+# -----------------------------------------------------------------------------
+
+
+def call_model(state: MessagesState) -> dict:
+    """调用模型；模型可以直接回答，也可以生成一个或多个 tool_calls。"""
+    response = model_with_tools.invoke(state["messages"])
+    return {"messages": [response]}
+
+
+def route_after_model(
+    state: MessagesState,
+) -> Literal["tools", "approval", "__end__"]:
+    """无工具调用则结束；敏感工具去审批；其余工具直接执行。"""
+    tool_calls = state["messages"][-1].tool_calls
+
+    if not tool_calls:
+        return END
+
+    if any(call["name"] in SENSITIVE_TOOL_NAMES for call in tool_calls):
+        return "approval"
+
+    return "tools"
+
+
+def approval_node(
+    state: MessagesState,
+) -> Command[Literal["tools", "model"]]:
+    """暂停图执行，等待用户批准或拒绝当前批次的工具调用。"""
+    tool_calls = state["messages"][-1].tool_calls
+
+    # interrupt() 应放在副作用之前。恢复时节点可能从函数开头重新执行。
+    decision = interrupt(
+        {
+            "type": "tool_approval",
+            "message": "模型请求执行敏感工具，是否批准？",
+            "tool_calls": tool_calls,
+        }
+    )
+
+    approved = (
+        bool(decision.get("approved"))
+        if isinstance(decision, dict)
+        else bool(decision)
+    )
+
+    if approved:
+        # ToolNode 将执行这一批工具调用，并生成对应的 ToolMessage。
+        return Command(goto="tools")
+
+    # 每个 tool_call 都必须得到对应的 ToolMessage，否则消息协议不完整，
+    # 部分模型服务会拒绝下一次请求。整个批次在审批拒绝后均不执行。
+    rejection_messages = [
+        ToolMessage(
+            content="用户拒绝了本批次工具调用，工具未执行。请告知用户并询问下一步。",
+            tool_call_id=call["id"],
+            name=call["name"],
+        )
+        for call in tool_calls
+    ]
+    return Command(
+        update={"messages": rejection_messages},
+        goto="model",
+    )
+
+
+# ToolNode 自动查找并执行工具，再把结果包装为 ToolMessage。
+tool_node = ToolNode(TOOLS)
+
+
+# -----------------------------------------------------------------------------
+# 4. 构建图并启用 Checkpointer
+# -----------------------------------------------------------------------------
+
+builder = StateGraph(MessagesState)
+builder.add_node("model", call_model)
+builder.add_node("approval", approval_node)
+builder.add_node("tools", tool_node)
+
+builder.add_edge(START, "model")
+builder.add_conditional_edges(
+    "model",
+    route_after_model,
+    {
+        "tools": "tools",
+        "approval": "approval",
+        END: END,
+    },
+)
+builder.add_edge("tools", "model")
+
+# interrupt() 必须配合 checkpointer 使用。InMemorySaver 只适合本地学习，
+# 进程退出后 checkpoint 会消失。
+checkpointer = InMemorySaver()
+graph = builder.compile(checkpointer=checkpointer)
+
+
+# -----------------------------------------------------------------------------
+# 5. 命令行交互
+# -----------------------------------------------------------------------------
+
+
+def print_update(event: dict) -> None:
+    """打印 model/tools 节点产生的消息更新。"""
+    for node_name, update in event.items():
+        if node_name == "__interrupt__" or not isinstance(update, dict):
+            continue
+
+        messages = update.get("messages", [])
+        if not messages:
+            continue
+
+        print(f"\n[{node_name}]")
+        for message in messages:
+            message.pretty_print()
+
+
+def run_until_pause_or_end(graph_input: dict | Command, config: dict):
+    """运行图，返回本轮产生的 interrupt；正常结束则返回 None。"""
+    pending_interrupt = None
+
+    for event in graph.stream(
+        graph_input,
+        config=config,
+        stream_mode="updates",
+    ):
+        print_update(event)
+
+        interrupts = event.get("__interrupt__", ())
+        if interrupts:
+            pending_interrupt = interrupts[0]
+
+    return pending_interrupt
+
+
+def ask_for_approval(interrupt_value) -> bool:
+    """在命令行展示待执行工具，并读取审批结果。"""
+    payload = interrupt_value.value
+
+    print("\n=== 需要人工审批 ===")
+    print(payload.get("message", "是否批准工具调用？"))
+
+    for index, call in enumerate(payload.get("tool_calls", []), start=1):
+        print(f"{index}. 工具：{call['name']}")
+        print(f"   参数：{call['args']}")
+
+    while True:
+        answer = input("批准执行吗？[y/n]：").strip().lower()
+        if answer in {"y", "yes"}:
+            return True
+        if answer in {"n", "no"}:
+            return False
+        print("请输入 y 或 n。")
+
+
+def run_turn(question: str, config: dict) -> None:
+    """执行一轮对话；如果暂停，则审批后使用 Command(resume=...) 恢复。"""
+    pending = run_until_pause_or_end(
+        {"messages": [{"role": "user", "content": question}]},
+        config,
+    )
+
+    # 理论上一次模型响应可能导致多次暂停，因此使用循环处理。
+    while pending is not None:
+        approved = ask_for_approval(pending)
+        pending = run_until_pause_or_end(
+            Command(resume={"approved": approved}),
+            config,
+        )
+
+
+def main() -> None:
+    print("=== LangGraph 工具助手（带人工审批）===")
+    print("天气和计算器直接执行；发送邮件需要人工批准。")
+    print("输入 /new 创建新会话，输入 quit 退出。")
+
+    # 同一 thread_id 的多轮输入共享 checkpoint 和消息历史。
+    thread_id = str(uuid.uuid4())
+
+    while True:
+        question = input(f"\n用户[{thread_id[:8]}]：").strip()
+
+        if question.lower() in {"quit", "exit", "q"}:
+            break
+        if question == "/new":
+            thread_id = str(uuid.uuid4())
+            print(f"已创建新会话：{thread_id}")
+            continue
+        if not question:
+            continue
+
+        config = {
+            "configurable": {"thread_id": thread_id},
+            "recursion_limit": 20,
+        }
+
+        try:
+            run_turn(question, config)
+        except Exception as exc:
+            print(f"\n执行失败：{type(exc).__name__}: {exc}")
+
+
+if __name__ == "__main__":
+    main()

07_command.py

@@ -0,0 +1,51 @@
+from typing import Literal, TypedDict
+
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import Command
+
+
+class State(TypedDict):
+    number: int
+    result: str
+
+
+def check_number(
+    state: State,
+) -> Command[Literal["positive", "negative"]]:
+    if state["number"] >= 0:
+        return Command(
+            update={"result": "这是非负数"},
+            goto="positive",
+        )
+
+    return Command(
+        update={"result": "这是负数"},
+        goto="negative",
+    )
+
+
+def positive(state: State):
+    print(state["result"])
+    return {}
+
+
+def negative(state: State):
+    print(state["result"])
+    return {}
+
+
+builder = StateGraph(State)
+builder.add_node("check", check_number)
+builder.add_node("positive", positive)
+builder.add_node("negative", negative)
+
+builder.add_edge(START, "check")
+builder.add_edge("positive", END)
+builder.add_edge("negative", END)
+
+graph = builder.compile()
+
+graph.invoke({
+    "number": -5,
+    "result": "",
+})