AI开发之LangGraph教程2
概述
这篇教程主打零基础上手,带大家用 LangGraph 从零搭建一款自带上下文记忆、支持自定义工具调用的智能聊天机器人。
我们都知道:原生大语言模型 LLM 天生无状态、无记忆,单独只能做简单单次问答,既记不住多轮聊天上下文,也没法自主完成数学计算、数据查询这类实操任务。而 LangGraph 就是专门解决这两个痛点的框架:它可以给 LLM 搭建标准化工作流程、持久保存对话记忆、自动判断什么时候该调用外部工具、工具执行完再把结果交给大模型整理成自然语言回复。
本教程会把所有底层关键知识点用大白话讲透,包含:状态、归约函数、状态图、节点、实线普通边、虚线条件边、工具绑定、条件分支判断、多轮对话记忆原理,同时提供完整可运行代码,学完就能理解 LangGraph 智能体的核心底层逻辑。
一、前置必懂基础知识点
1. LLM 核心短板
大模型本身没有任何记忆存储能力,每一次问答都是独立全新的一次运算。它不会主动保存聊天记录、记不住上一轮对话,想要实现连贯聊天,必须每次请求都把完整历史消息全部发给大模型。
2. LangGraph 核心五大基础概念
-
State 状态可以理解成机器人的专属记事本,专门用来存放全局数据,本项目里主要存所有轮次的聊天消息。
-
Reducer 归约函数用来规定状态数据的更新规则。本项目用的
add_messages是官方内置归约器,规则是新消息追加到列表末尾,绝不覆盖旧聊天记录,这是实现多轮记忆的核心。 -
StateGraph 状态图整个机器人的流程总设计师,负责把一个个功能节点串联起来,规定先执行谁、后执行谁、遇到分支该怎么走。
-
Node 节点独立的功能执行单元,相当于机器人的一个个 “工种”:
chatbot节点:负责调用大模型生成思考和回复tools节点:负责执行具体的工具函数(加法、乘法计算)
- Edge 边(分两种,对应流程图实线、虚线)
-
实线 = 普通无条件边代码里
add_edge()创建,是固定必走流程,没有选择余地,走完这个节点必然跳到下一个指定节点。比如:程序启动必然进聊天节点、工具执行完必然跳回聊天节点。 -
虚线 = 条件分支边代码里
add_conditional_edges()创建,是二选一判断流程。由自定义判断函数should_continue决定走向:大模型需要调用工具就走工具节点,只是普通闲聊就直接结束对话。
3. 工具调用是什么
大模型擅长语言理解,但不擅长精准数学计算、实时数据查询等硬性任务。我们可以自定义工具函数交给大模型调用,大模型会自动识别用户需求、自动触发工具、框架自动执行运算,最后把结果整理成人话回复用户。本教程内置两个工具:三数连加、三数连乘。
二、项目整体运行流程
完整流程图逻辑:START 程序入口 → 【实线】→ chatbot 聊天节点 → 【虚线条件分支】
- 分支一:无需工具 → 直达
END流程结束 - 分支二:需要工具 → 【虚线】→
tools 工具节点→ 【实线】→ 回流到chatbot聊天节点,由大模型整理工具结果后,再结束流程

三、安装依赖
运行代码前先安装必备库:
pip install -U langgraph langchain-openai typing-extensions
四、完整可运行代码
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# ==================== 自行替换个人API配置 ====================
API_KEY = "请填入自己的大模型API密钥"
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL_NAME = "qwen3.6-plus"
# ==================== 自定义工具函数 ====================
@tool
def add_three_numbers(
a: Annotated[int, "第一个数"],
b: Annotated[int, "第二个数"],
c: Annotated[int, "第三个数"]
) -> int:
"""将三个数相加并返回结果。"""
result = a + b + c
print(f"[Tool] 计算 {a} + {b} + {c} = {result}")
return result
@tool
def multiply_three_numbers(
a: Annotated[int, "第一个数"],
b: Annotated[int, "第二个数"],
c: Annotated[int, "第三个数"]
) -> int:
"""将三个数相乘并返回结果。"""
result = a * b * c
print(f"[Tool] 计算 {a} * {b} * {c} = {result}")
return result
# ==================== 定义全局状态 ====================
class State(TypedDict):
"""
定义图的状态结构。
messages: 消息列表,使用 add_messages 归约器
- 作用:新消息追加到列表,不覆盖历史对话
- 解决LLM无状态短板,实现多轮聊天记忆
"""
messages: Annotated[list, add_messages]
# ==================== 定义节点功能函数 ====================
def chatbot(state: State):
"""
聊天机器人节点:
把完整的历史对话消息全部传给LLM,获取模型响应
"""
return {"messages": [llm.invoke(state["messages"])]}
def should_continue(state: State) -> str:
"""
条件判断函数:控制虚线分支走向
检测最后一条模型回复是否携带工具调用指令
- 有工具调用:返回 tools,走工具节点
- 无工具调用:返回 END,直接结束对话
"""
messages = state["messages"]
last_message = messages[-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
# ==================== 搭建状态图工作流 ====================
# 注册可用工具列表
tools = [add_three_numbers, multiply_three_numbers]
# 初始化大模型,并绑定自定义工具
llm = ChatOpenAI(
model=MODEL_NAME,
base_url=BASE_URL,
api_key=API_KEY,
temperature=0,
timeout=60,
request_timeout=60
).bind_tools(tools)
# 创建状态图构建器
graph_builder = StateGraph(State)
# 向图中添加两个核心节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools))
# 添加实线普通边:固定流程跳转
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("tools", "chatbot")
# 添加虚线条件边:二一分支判断
graph_builder.add_conditional_edges("chatbot", should_continue, {"tools": "tools", END: END})
# 编译状态图,生成可执行程序
graph = graph_builder.compile()
# ==================== 辅助工具函数 ====================
def display_graph():
"""控制台打印ASCII版流程图,直观查看实线虚线结构"""
try:
print("\n图结构:")
print(graph.get_graph().draw_ascii())
except Exception:
pass
def stream_graph_updates(user_input: str):
"""流式接收用户输入,执行图流程并输出回复"""
for event in graph.stream({"messages": [("user", user_input)]}):
for node_name, value in event.items():
if node_name == "chatbot":
if "messages" in value:
last_msg = value["messages"][-1]
if hasattr(last_msg, "content") and last_msg.content:
print(f"助手: {last_msg.content}")
# ==================== 主交互程序 ====================
def main():
print("=" * 60)
print("LangGraph 入门实战:带工具调用的智能聊天机器人")
print("模型: 阿里云通义千问")
print("=" * 60)
display_graph()
print("\n可用工具:")
print(" - add_three_numbers: 三个数连加计算")
print(" - multiply_three_numbers: 三个数连乘计算")
print("\n输入 quit / exit / q 即可退出程序")
print("=" * 60 + "\n")
while True:
try:
user_input = input("用户: ").strip()
if user_input.lower() in ["quit", "exit", "q"]:
print("再见!")
break
if not user_input:
continue
stream_graph_updates(user_input)
print()
except KeyboardInterrupt:
print("\n再见!")
break
except Exception as e:
print(f"发生错误: {e}")
if __name__ == "__main__":
main()
五、逐模块核心知识点精讲
1. 工具函数模块
@tool装饰器:把普通 Python 函数标记为大模型可识别的工具,大模型能自动读取函数说明、参数含义;Annotated注解:给参数加文字描述,方便大模型理解每个参数该填什么内容;- 工具内部实现具体业务逻辑,专门处理大模型不擅长的精准计算。
2. 状态 State 模块
TypedDict:固定状态的数据结构,规定只能有messages这一个字段;messages: Annotated[list, add_messages]第一层list:对话消息用列表存储;第二层add_messages:归约函数,强制新消息追加不覆盖,保住全程聊天记忆。
3. chatbot 聊天节点
每次执行都会把 state["messages"] 完整历史消息一次性传给 LLM,正因为 LLM 无状态,只有传全部历史,模型才能看懂上下文、连贯回复。
4. should_continue 条件判断函数
专门用来匹配流程图虚线分支:读取模型最后一条回复,判断有没有 tool_calls 工具调用标记;有标记就走工具节点,没标记直接结束对话,是整个分支流程的核心控制器。
5. 状态图搭建
bind_tools(tools):把自定义工具绑定给大模型,让模型知道自己有哪些技能可以调用;ToolNode:LangGraph 内置工具节点,不用自己写执行逻辑,自动解析工具调用、执行函数、返回结果;- 实线边固定流转,虚线边动态判断流转,组合成完整智能体工作流。
六、运行实测三种场景
-
普通闲聊(走虚线直达 END)用户:你好,介绍一下自己助手直接文字回复,不触发任何工具,流程直接结束。
-
三数乘法(触发工具分支)用户:计算 4 乘以 6 乘以 7框架自动走虚线跳到工具节点,执行乘法运算,再实线回流到聊天节点,大模型整理结果后回复。
-
多轮带记忆对话第一轮问加法,第二轮接着问 “再算一遍”,机器人能记住上一轮数字,依靠 State + add_messages 实现上下文关联。
七、核心知识点总结
- LangGraph 用 State + add_messages 归约器 补齐 LLM 无状态缺陷,实现多轮对话记忆;
- 实线普通边是固定流程,虚线条件边是智能判断分支;
- 工具调用完整闭环:用户提问→模型识别需求→触发工具→执行运算→返回模型→整理自然语言回复;
- 所有节点、流程、分支都由 StateGraph 统一编排,结构清晰,方便后期扩展更多工具、更多分支逻辑。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)