黔南布依族苗族自治州网站建设_网站建设公司_模板建站_seo优化

本文详细介绍了LangGraph框架的核心概念——State管理。通过计算器示例和聊天机器人实战，演示了如何使用字典和TypedDict定义State，以及Reducer机制控制状态更新。特别讲解了add_messages函数如何智能管理对话历史，并提供了完整的聊天机器人实现代码。读者可掌握State、Node、Edge等基础概念，理解LangGraph工作原理，并能独立开发大模型应用。

一、先把环境准备好

咱们先把开发环境搞定，后面代码都能直接跑起来。打开终端，依次执行：

# 安装核心包 pip install langgraph langchain-openai # 安装可视化工具（可选，但强烈推荐） pip install pyppeteer ipython

创建一个langgraph_demo.py文件，把API Key配置好：

import os import getpass # 设置OpenAI API Key if not os.environ.get("OPENAI_API_KEY"): os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入你的OpenAI API Key: ")

好了，准备工作完成。接下来进入正题。

二、State到底是个啥？

很多教程上来就讲概念，我换个思路。咱们先看一个实际场景：

假设你要做一个简单的计算器程序，流程是这样的：

1. 用户输入一个数字，比如10
2. 第一步：给这个数字加1
3. 第二步：用上一步的结果减2
4. 最后输出结果

传统写法可能是这样：

def calculator(x): x = x + 1 # 第一步 y = x - 2 # 第二步 return x, y result = calculator(10) print(result) # (11, 9)

这没问题，但如果流程复杂了呢？比如有10个步骤，每个步骤可能用到前面多个结果，代码就会很乱。

LangGraph的解决方案是：用一个共享的"状态"（State）来存储所有中间结果。每个处理步骤只需要：

• 读取State中需要的数据
• 处理完后，把结果写回State
• 不用管其他步骤的数据

这就是State的核心思想：一个在整个流程中共享的数据容器。

三、第一个例子：用字典做State

咱们把刚才的计算器用LangGraph实现一遍。完整代码如下，每行都加了注释：

from langgraph.graph import StateGraph, START, END # 定义第一个节点：加法 def addition(state): """ 这个函数会收到当前的state（一个字典） 从里面取出x的值，加1后返回 """ print(f"加法节点收到的state: {state}") current_x = state["x"] new_x = current_x + 1 # 只返回更新的部分，不用担心其他数据会丢 return {"x": new_x} # 定义第二个节点：减法 def subtraction(state): """ 这个函数也会收到state 注意：这里的state已经包含了上一个节点更新后的x值 """ print(f"减法节点收到的state: {state}") current_x = state["x"] y = current_x - 2 # 这里我们新增了一个y，x会保留 return {"y": y} # 开始构建图 # StateGraph(dict)表示我们的state是一个普通字典 builder = StateGraph(dict) # 添加两个节点 # 第一个参数是节点名称，第二个参数是对应的函数 builder.add_node("addition", addition) builder.add_node("subtraction", subtraction) # 定义节点之间的连接关系（边） # START -> addition -> subtraction -> END builder.add_edge(START, "addition") # 起点连到addition builder.add_edge("addition", "subtraction") # addition连到subtraction builder.add_edge("subtraction", END) # subtraction连到终点 # 编译成可执行的图 graph = builder.compile() # 运行！传入初始状态 initial_state = {"x": 10} final_state = graph.invoke(initial_state) print(f"\n最终状态: {final_state}")

运行这段代码，你会看到：

加法节点收到的state: {‘x’: 10}
减法节点收到的state: {‘x’: 11}

最终状态: {‘x’: 11, ‘y’: 9}

看到了吗？关键点来了：

1. addition节点只返回了{"x": 11}，但y没有丢
2. subtraction节点返回了{"y": 9}，x也还在
3. 最终状态包含了所有的更新

这就是LangGraph的魔法：节点只需要返回它要更新的部分，其他数据会自动保留。

看看图长什么样

代码能跑是一回事，理解结构又是另一回事。咱们把这个图可视化出来：

from IPython.display import Image, display # 生成图的可视化 # xray=True表示显示详细信息 display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

你会看到一个流程图，清楚地显示了节点的连接关系。开发的时候这个功能特别有用，能一眼看出逻辑对不对。

四、进阶：用TypedDict让代码更严谨

用字典虽然灵活，但有个大问题：如果你不小心写错了key的名字，程序运行到那里才会报错。

比如这样：

def buggy_function(state): # 假设state里没有"z"这个key return {"result": state["z"] + 1} # 💥运行时报错 为了避免这种问题，实际开发中我们用 TypedDict 来定义State的结构： from typing_extensions import TypedDict # 定义State的结构 class State(TypedDict): x: int # x是整数 y: int # y也是整数 # 现在用State来构建图 builder = StateGraph(State) # 节点函数保持不变 def addition(state: State): # 加上类型标注 return {"x": state["x"] + 1} def subtraction(state: State): return {"y": state["x"] - 2} # 后面的代码完全一样 builder.add_node("addition", addition) builder.add_node("subtraction", subtraction) builder.add_edge(START, "addition") builder.add_edge("addition", "subtraction") builder.add_edge("subtraction", END) graph = builder.compile() # 运行 result = graph.invoke({"x": 10}) print(result) # {'x': 11, 'y': 9}

用了TypedDict后，IDE会给你代码提示，写错了key名编辑器就会提醒你。虽然代码多了几行，但避免了很多低级错误。

五、Reducer：状态更新的幕后功臣

现在来解答一个关键问题：为什么节点只返回部分数据，State就能自动合并？

答案是Reducer函数。每个State的key背后都有一个Reducer，默认行为是"覆盖"。我画个图帮你理解：

当前State: {“x”: 10}
节点返回: {“x”: 11}
Reducer执行: 把x从10改成11
结果State: {“x”: 11}

当前State: {“x”: 11}
节点返回: {“y”: 9}
Reducer执行: x不变，新增y
结果State: {“x”: 11, “y”: 9}

默认的Reducer很简单，就是更新或新增。但我们可以自定义更复杂的行为。

六、实战：维护消息列表

聊天机器人需要记住对话历史，每次都要往消息列表里追加新消息。如果用默认的Reducer（覆盖），每次都会丢失之前的对话。

这时候就需要operator.add这个Reducer：

import operator from typing import Annotated, List from typing_extensions import TypedDict # 注意这里的写法 class State(TypedDict): # Annotated[类型, Reducer函数] messages: Annotated[List[dict], operator.add] # 测试一下 def node1(state): # 返回一个包含新消息的列表 return {"messages": [{"role": "user", "content": "你好"}]} def node2(state): print(f"node2收到的消息: {state['messages']}") # 再追加一条消息 return {"messages": [{"role": "assistant", "content": "你好啊"}]} builder = StateGraph(State) builder.add_node("node1", node1) builder.add_node("node2", node2) builder.add_edge(START, "node1") builder.add_edge("node1", "node2") builder.add_edge("node2", END) graph = builder.compile() # 初始状态包含一个空列表 result = graph.invoke({"messages": []}) print(f"\n最终消息列表: {result['messages']}")

运行结果：

node2收到的消息: [{'role': 'user', 'content': '你好'}] 最终消息列表: [ {'role': 'user', 'content': '你好'}, {'role': 'assistant', 'content': '你好啊'} ]

看到了吗？用了operator.add之后，每次返回的消息列表会追加到现有列表中，而不是覆盖。这就是Reducer的威力。

七、接入真实的AI模型

铺垫了这么多，终于到激动人心的部分了——接入GPT做一个真正能聊天的机器人！

完整代码如下，建议你创建一个新文件chatbot.py：

import os import getpass from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage from IPython.display import Image, display # 配置API Key if not os.environ.get("OPENAI_API_KEY"): os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入OpenAI API Key: ") # 定义State结构 class State(TypedDict): # add_messages是LangGraph提供的专门处理消息的Reducer messages: Annotated[list, add_messages] # 创建GPT模型实例 llm = ChatOpenAI(model="gpt-4o") # 定义聊天节点 def chatbot(state: State): """ 这个函数接收包含对话历史的state 把所有消息发给GPT，获取回复 """ # state["messages"]是一个消息列表 # 直接传给模型就行 response = llm.invoke(state["messages"]) # 返回模型的回复，add_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() # 可视化看看（可选） # display(Image(graph.get_graph(xray=True).draw_mermaid_png())) # 单次对话测试 def test_single(): user_message = HumanMessage(content="你好，请介绍一下你自己") result = graph.invoke({"messages": [user_message]}) print("用户:", user_message.content) print("AI:", result["messages"][-1].content) # 连续对话测试 def chat_loop(): print("聊天机器人已启动（输入'退出'结束对话）\n") while True: user_input = input("你: ").strip() if user_input == "退出": print("再见！") break if not user_input: continue # 每次都创建新的对话 # 如果要保持对话历史，需要把messages保存下来 for event in graph.stream({"messages": [HumanMessage(content=user_input)]}): for value in event.values(): print(f"AI: {value['messages'][-1].content}\n") # 运行 if __name__ == "__main__": # test_single() # 测试单次对话 chat_loop() # 开启连续对话

运行这段代码，你就有了一个能对话的AI助手！

不过这个版本有个问题：每次对话都是独立的，没有记忆。要实现带记忆的版本，需要这样改：

def chat_with_memory(): """带对话历史的聊天""" print("聊天机器人已启动（输入'退出'结束对话）\n") # 用这个列表保存整个对话历史 conversation_history = [] while True: user_input = input("你: ").strip() if user_input == "退出": print("再见！") break if not user_input: continue # 把用户消息加入历史 conversation_history.append(HumanMessage(content=user_input)) # 用完整的历史调用模型 result = graph.invoke({"messages": conversation_history}) # 获取AI回复 ai_message = result["messages"][-1] print(f"AI: {ai_message.content}\n") # 把AI回复也加入历史 conversation_history.append(ai_message) # 运行带记忆的版本 if __name__ == "__main__": chat_with_memory()

现在你可以这样聊：

你: 我叫小明
AI: 你好小明！很高兴认识你…

你: 我刚才说我叫什么？
AI: 你刚才说你叫小明。

看，AI能记住之前的对话了！

八、add_messages的秘密

可能你注意到了，我们用的是add_messages而不是operator.add。这两个有什么区别？

operator.add很简单粗暴：直接把两个列表拼起来。

add_messages更聪明：

1. 每条消息都有唯一ID
2. 如果新消息的ID和已有消息相同，就更新那条消息
3. 如果ID不同，就追加新消息

看个例子：

from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage # 场景1：追加新消息 msg1 = [HumanMessage(content="你好", id="1")] msg2 = [AIMessage(content="你好啊", id="2")] result = add_messages(msg1, msg2) print("场景1 - 追加:", [m.content for m in result]) # 输出: ['你好', '你好啊'] # 场景2：更新已有消息 msg1 = [HumanMessage(content="你好", id="1")] msg2 = [HumanMessage(content="你好呀", id="1")] # 注意ID相同 result = add_messages(msg1, msg2) print("场景2 - 更新:", [m.content for m in result]) # 输出: ['你好呀'] # 第一条消息被更新了

这个特性在人机协作场景特别有用。比如用户可以修改AI的回复，然后继续对话。

九、源码解析：add_messages是怎么实现的？

有些同学喜欢刨根问底，咱们快速过一下核心逻辑（不想看可以跳过）：

def add_messages(left, right): """ left: 已有的消息列表 right: 新来的消息列表 """ # 1. 确保都是列表 if not isinstance(left, list): left = [left] if not isinstance(right, list): right = [right] # 2. 给每条消息分配ID（如果没有的话） for m in left: if m.id is None: m.id = str(uuid.uuid4()) for m in right: if m.id is None: m.id = str(uuid.uuid4()) # 3. 建立ID到消息的映射 left_idx_by_id = {m.id: i for i, m in enumerate(left)} # 4. 合并：ID相同就更新，不同就追加 merged = left.copy() for m in right: if m.id in left_idx_by_id: # ID存在，更新 merged[left_idx_by_id[m.id]] = m else: # ID不存在，追加 merged.append(m) return merged

核心就是用ID来判断是更新还是追加。简单但很实用。

十、完整项目：支持多轮对话的智能助手

最后把所有知识点整合一下，做个功能完整的聊天助手：

import os import getpass from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # ===== 配置部分 ===== if not os.environ.get("OPENAI_API_KEY"): os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入OpenAI API Key: ") # ===== State定义 ===== class ChatState(TypedDict): messages: Annotated[list, add_messages] # ===== 创建模型 ===== llm = ChatOpenAI( model="gpt-4o", temperature=0.7, # 控制回复的创造性 ) # ===== 定义节点 ===== def chatbot_node(state: ChatState): """处理用户消息，生成回复""" messages = state["messages"] response = llm.invoke(messages) return {"messages": [response]} # ===== 构建图 ===== builder = StateGraph(ChatState) builder.add_node("chatbot", chatbot_node) builder.add_edge(START, "chatbot") builder.add_edge("chatbot", END) graph = builder.compile() # ===== 主程序 ===== def main(): print("=" * 50) print("智能聊天助手 v1.0") print("=" * 50) print("提示：输入'退出'结束对话\n") # 设置系统提示词 conversation = [ SystemMessage(content="你是一个友好、专业的AI助手。回答要简洁明了。") ] while True: # 获取用户输入 user_input = input("👤 你: ").strip() # 退出判断 if user_input.lower() in ["退出", "exit", "quit"]: print("\n👋 感谢使用，再见！") break # 跳过空输入 if not user_input: continue # 添加用户消息 conversation.append(HumanMessage(content=user_input)) try: # 调用图进行处理 result = graph.invoke({"messages": conversation}) # 获取AI回复 ai_response = result["messages"][-1] conversation.append(ai_response) # 显示回复 print(f"🤖 AI: {ai_response.content}\n") except Exception as e: print(f"❌ 出错了: {str(e)}\n") # 出错时移除最后添加的用户消息 conversation.pop() if __name__ == "__main__": main()

这个版本包含了：

• ✅ 完整的对话历史管理
• ✅ 系统提示词设置
• ✅ 异常处理
• ✅ 友好的交互界面

直接运行就能用！

十一、常见问题答疑

Q1: State可以存储什么类型的数据？

A: 理论上任何Python对象都行：字典、列表、自定义类、甚至函数。但建议用简单的数据结构，方便序列化和持久化。

Q2: 节点函数可以不返回任何东西吗？

A: 可以。如果节点只是做一些副作用操作（比如打印日志），可以返回空字典{}或者None。

Q3: 怎么调试State的变化？

A: 在每个节点函数里加print(state)，就能看到每一步State的变化情况。

Q4: TypedDict是必须的吗？

A: 不是必须的，但强烈推荐。它能帮你在开发阶段就发现很多类型错误。

Q5: 对话历史会越来越长，怎么办？

A: 这是个好问题。可以定期裁剪历史，只保留最近N条消息。或者用摘要技术压缩历史信息。后面的文章会详细讲。

十二、总结

今天我们深入学习了LangGraph的核心——State管理：

1. State是什么：在整个流程中共享的数据容器
2. 基本用法：节点读取State，处理后返回更新的部分
3. TypedDict：让代码更严谨，避免运行时错误
4. Reducer机制：控制State的更新方式（覆盖、追加等）
5. add_messages：智能管理消息列表的专用Reducer
6. 实战应用：构建带记忆的聊天机器人

掌握了这些，你就能灵活构建各种LangGraph应用了。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

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

为什么要学习大模型？

我国在A大模型领域面临人才短缺,数量与质量均落后于发达国家。2023年，人才缺口已超百万，凸显培养不足。随着AI技术飞速发展，预计到2025年,这一缺口将急剧扩大至400万,严重制约我国AI产业的创新步伐。加强人才培养,优化教育体系,国际合作并进是破解困局、推动AI发展的关键。

大模型入门到实战全套学习大礼包

1、大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

2、大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

3、AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

4、大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

5、大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

适用人群

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

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