企业官网建设流程全解析

用 LangGraph 的 Reducer，轻松搞定多轮对话记忆

做聊天机器人最头疼的是什么？是记不住上文。今天分享一个 LangGraph 的小技巧：用 Reducer 机制优雅地管理对话历史。

核心只有一行代码：

messages: Annotated[Sequence[AnyMessage], operator.add]

这里的 operator.add 就是 Reducer（归约器）。它的妙处在于：节点函数只需返回新增的消息，LangGraph 会自动把旧历史和新消息拼接起来，不用手动维护完整列表，代码瞬间清爽。

另一个容易踩坑的点：图里定义的 END 只是本次流程的终点，不是程序终点。多轮对话要靠外层 while 循环驱动，每次 invoke 后用返回的 result 更新 state，再传回下一轮，历史就自然延续了。

最后附上一个小实践：调用模型时只取最近 6 条消息作为上下文，既保证连贯性，又控制了 token 成本。

Reducer + 截断窗口 + 外层循环，三件套齐活，一个专业级多轮对话系统就搭好了。

代码

# -*- coding: utf-8 -*- """ 基于 LangGraph 构建的多轮对话系统示例 核心知识点： 1. 使用 Annotated + operator.add 作为 Reducer，实现消息历史的自动合并（累加） 2. 图的每次 invoke 是一次独立执行，END 只是本次流程的终点，外层 while 循环负责多轮对话 3. 通过不断用返回的 result 更新 state，实现跨轮次的消息历史保留 """ from typing import TypedDict, Annotated, Sequence from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, SystemMessage, AnyMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END import operator import os from dotenv import load_dotenv # 加载 .env 文件中的环境变量（如 API Key、Base URL） load_dotenv() # 初始化大语言模型客户端（兼容 OpenAI 接口协议，这里对接阿里云百炼的 deepseek-v3） model = ChatOpenAI( model_name="deepseek-v3", api_key=os.getenv("DASHSCOPE_API_KEY"), base_url=os.getenv("API_BASE_URL") ) # ================== 类型定义 ================== class ChatState(TypedDict): """ 对话状态定义 Annotated[Sequence[AnyMessage], operator.add] 中的 operator.add 是 Reducer（归约器）： - 节点返回 {"messages": [新消息]} 时，LangGraph 不会覆盖旧消息， 而是自动执行 旧消息 + 新消息 的拼接，实现历史记录的累加 - 这就是为什么节点函数只需返回"增量"消息，而不用手动拼接完整历史 """ messages: Annotated[Sequence[AnyMessage], operator.add] # 自动合并消息的历史记录 # ================== 节点函数1 ================== def handle_user_input(state: ChatState): """处理用户输入节点：从控制台读取用户输入，封装成 HumanMessage""" try: print(f"handle_user_input_node state:{state}") user_input = input("\n用户输入（输入'退出'结束）: ").strip() # 用户输入'退出'时返回 END，提前终止本次图流程（跳过 ai_response 节点） if user_input.lower() == "退出": print("正在结束对话...") return END # 注意：借助 Reducer（operator.add），这里只需返回"新增"的消息即可， # LangGraph 会自动将其与 state 中的历史消息合并。 # （若没有 Reducer，则需要手动写 state["messages"] + [HumanMessage(...)] 来覆盖整个列表） return {"messages": [HumanMessage(content=user_input)]} except KeyboardInterrupt: # 用户按 Ctrl+C 中断时，同样优雅退出本次图流程 return END # ================== 节点函数2 ================== def generate_ai_response(state: ChatState): """生成AI响应节点：取最近几条消息作为上下文，调用模型生成回复""" try: print(f"generate_ai_response_node state:{state}") # 只取最近 6 条消息作为上下文（截断窗口）： # 既保持对话连贯性，又控制 token 消耗、避免超出模型上下文长度 recent_history = state["messages"][-6:] response = model.invoke(recent_history) # 返回一个 AIMessage 对象 print("模型答复：",response) # 返回"增量"消息，由 Reducer 自动追加到历史中 return {"messages": [response]} except Exception as e: # 调用失败（网络异常、鉴权失败等）时不中断程序， # 而是生成一条兜底的错误提示消息，保证图流程能正常走完 error_msg = f"系统暂时无法响应，请稍后再试（错误代码：{str(e)[:30]})" return {"messages": [AIMessage(content=error_msg)]} # ================== 对话图构建 ================== builder = StateGraph(ChatState) # 节点注册：将节点函数挂载到图上，指定节点名称 builder.add_node("user_input", handle_user_input) builder.add_node("ai_response", generate_ai_response) # 流程设计：入口 -> 用户输入 -> AI响应 -> 结束 builder.set_entry_point("user_input") builder.add_edge("user_input", "ai_response") builder.add_edge("ai_response", END) # 编译图：得到可执行（invoke / stream）的对话应用 conversation = builder.compile() # ================== 运行逻辑 ================== if __name__ == "__main__": # 系统提示词：对话历史中的第一条消息，为模型设定角色 system_prompt = f""" 你是一个专业级中文智能助手！""" # 初始状态只包含系统消息，后续历史由 Reducer 自动累加 state = ChatState(messages=[SystemMessage(content=system_prompt)]) print("===== 智能对话系统已启动 =====") print("输入'退出'可随时结束对话\n") """ 关键机制说明： 1、每次调用 conversation.invoke(state) 都是一次独立的调用， 虽然图流程定义了 END，但图本身不会阻止下一次调用。 2、状态 state 会持续更新：每次执行完 conversation.invoke(state) 后， 返回的 result 会更新 state，从而保留了对话历史。 当下次调用 invoke 时，传入的是更新后的 state，因此对话可以继续。 3、END 只是图流程的结束点，不是程序的终止点： END 只是告诉 LangGraph 当前流程已结束，但并不会终止外部的 while 循环。 """ # 外层 while 循环负责"多轮对话"：每轮执行一次完整的图流程 while True: try: # 执行一次图流程（user_input -> ai_response），invoke 返回执行后的完整状态 result = conversation.invoke(state) print("图执行一次后：",result) # 异常防御：返回结果为空或缺少 messages 字段时退出循环 if result is None or "messages" not in result: break # 提取本轮新增的消息： # 由于 Reducer 是"追加"语义，result 中前面部分就是调用前的旧历史， # 从旧历史长度开始的切片即为本次流程新产生的消息（HumanMessage + AIMessage） new_messages = result["messages"][len(state["messages"]):] # 打印本轮 AI 响应（跳过用户新输入的 HumanMessage） for msg in new_messages: if isinstance(msg, AIMessage): print(f"\n【AI响应】\n{msg.content}\n") # 更新对话状态：用包含完整历史的 result 替换旧 state， # 下一轮循环将其传回 invoke，实现跨轮次的历史延续 state = result # 检查退出条件：历史中出现内容为'退出'的用户消息时结束循环 # （正常情况下输入'退出'时节点会直接返回 END 不会进入 ai_response， # 这里是双保险） if any(isinstance(m, HumanMessage) and m.content.lower() == "退出" for m in state["messages"]): break except Exception as e: # 捕获图执行过程中的其他异常，打印后退出，避免程序崩溃 print(f"\n系统异常：{str(e)}") break # ===== 对话结束后的收尾：打印完整对话记录 ===== print("\n===== 对话已结束 =====") print("\n【完整对话记录】") # state["messages"][1:] 跳过第一条系统提示，只展示用户与 AI 的往来消息 for i, msg in enumerate(state["messages"][1:]): prefix = "用户：" if isinstance(msg, HumanMessage) else "AI：" print(f"{i + 1}. {prefix}{msg.content}")

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】