From f0b0eb13fe191fefff771d69ac77d1cab30df2b2 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Sat, 25 Jul 2026 17:31:26 +0800 Subject: [PATCH] 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 | 3 + 01_basic_agent_no_llm.py | 38 +++++ 02_chat_graph.py | 41 +++++ 03_conditional_edge.py | 54 +++++++ 04_checkpointer.py | 64 ++++++++ 05_tool_call.py | 143 ++++++++++++++++++ 06_human_in_the_loop.py | 315 +++++++++++++++++++++++++++++++++++++++ 07_command.py | 51 +++++++ 8 files changed, 709 insertions(+) create mode 100644 01_basic_agent_no_llm.py create mode 100644 02_chat_graph.py create mode 100644 03_conditional_edge.py create mode 100644 04_checkpointer.py create mode 100644 05_tool_call.py create mode 100644 06_human_in_the_loop.py create mode 100644 07_command.py diff --git a/.gitignore b/.gitignore index e0c7f39..a971e16 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,9 @@ __pycache__/ venv/ env/ +# Local uv environment/version files +.python-version + # IDE and OS files .vscode/ .idea/ diff --git a/01_basic_agent_no_llm.py b/01_basic_agent_no_llm.py new file mode 100644 index 0000000..ca4c4ec --- /dev/null +++ b/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) \ No newline at end of file diff --git a/02_chat_graph.py b/02_chat_graph.py new file mode 100644 index 0000000..0e7ff57 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/03_conditional_edge.py b/03_conditional_edge.py new file mode 100644 index 0000000..8f2f697 --- /dev/null +++ b/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": ""})) \ No newline at end of file diff --git a/04_checkpointer.py b/04_checkpointer.py new file mode 100644 index 0000000..126f7ee --- /dev/null +++ b/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) \ No newline at end of file diff --git a/05_tool_call.py b/05_tool_call.py new file mode 100644 index 0000000..8090795 --- /dev/null +++ b/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() diff --git a/06_human_in_the_loop.py b/06_human_in_the_loop.py new file mode 100644 index 0000000..c7912eb --- /dev/null +++ b/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() diff --git a/07_command.py b/07_command.py new file mode 100644 index 0000000..f43ea8e --- /dev/null +++ b/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": "", +}) \ No newline at end of file