概述

这篇教程主打零基础上手,带大家用 LangGraph 从零搭建一款自带上下文记忆、支持自定义工具调用的智能聊天机器人。

我们都知道:原生大语言模型 LLM 天生无状态、无记忆,单独只能做简单单次问答,既记不住多轮聊天上下文,也没法自主完成数学计算、数据查询这类实操任务。而 LangGraph 就是专门解决这两个痛点的框架:它可以给 LLM 搭建标准化工作流程、持久保存对话记忆、自动判断什么时候该调用外部工具、工具执行完再把结果交给大模型整理成自然语言回复。

本教程会把所有底层关键知识点用大白话讲透,包含:状态、归约函数、状态图、节点、实线普通边、虚线条件边、工具绑定、条件分支判断、多轮对话记忆原理,同时提供完整可运行代码,学完就能理解 LangGraph 智能体的核心底层逻辑。


一、前置必懂基础知识点

1. LLM 核心短板

大模型本身没有任何记忆存储能力,每一次问答都是独立全新的一次运算。它不会主动保存聊天记录、记不住上一轮对话,想要实现连贯聊天,必须每次请求都把完整历史消息全部发给大模型

2. LangGraph 核心五大基础概念

  1. State 状态可以理解成机器人的专属记事本,专门用来存放全局数据,本项目里主要存所有轮次的聊天消息。

  2. Reducer 归约函数用来规定状态数据的更新规则。本项目用的 add_messages 是官方内置归约器,规则是新消息追加到列表末尾,绝不覆盖旧聊天记录,这是实现多轮记忆的核心。

  3. StateGraph 状态图整个机器人的流程总设计师,负责把一个个功能节点串联起来,规定先执行谁、后执行谁、遇到分支该怎么走。

  4. Node 节点独立的功能执行单元,相当于机器人的一个个 “工种”:

  • chatbot 节点:负责调用大模型生成思考和回复
  • tools 节点:负责执行具体的工具函数(加法、乘法计算)
  1. 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 内置工具节点,不用自己写执行逻辑,自动解析工具调用、执行函数、返回结果;
  • 实线边固定流转,虚线边动态判断流转,组合成完整智能体工作流。

六、运行实测三种场景

  1. 普通闲聊(走虚线直达 END)用户:你好,介绍一下自己助手直接文字回复,不触发任何工具,流程直接结束。

  2. 三数乘法(触发工具分支)用户:计算 4 乘以 6 乘以 7框架自动走虚线跳到工具节点,执行乘法运算,再实线回流到聊天节点,大模型整理结果后回复。

  3. 多轮带记忆对话第一轮问加法,第二轮接着问 “再算一遍”,机器人能记住上一轮数字,依靠 State + add_messages 实现上下文关联。


七、核心知识点总结

  1. LangGraph 用 State + add_messages 归约器 补齐 LLM 无状态缺陷,实现多轮对话记忆;
  2. 实线普通边是固定流程,虚线条件边是智能判断分支;
  3. 工具调用完整闭环:用户提问→模型识别需求→触发工具→执行运算→返回模型→整理自然语言回复;
  4. 所有节点、流程、分支都由 StateGraph 统一编排,结构清晰,方便后期扩展更多工具、更多分支逻辑。
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐