LangGraph实战案例之MessagesState类继承和能力扩展

LangGraph 的核心在于将复杂的 AI 应用逻辑解耦为可管理的状态图，但官方提供的 `MessagesState` 往往只能处理最基础的对话历史存储，无法满足实际工程中对于上下文检索、关键词提取或多模态数据流转的定制需求。这份实战资料聚焦于如何通过类继承机制对 `MessagesState`

单课价¥29新品上架

会员价 会员免费开通会员，全站课程按额度无限下载套餐详情

144 人关注更新于 2026-10-12 永久有效

发货

网盘链接 · 支付成功后自动发货

保障

永久有效 · 链接失效免费补发

客服

每日 9:00-21:00 在线客服联系客服

购买方式开通会员 · 全站免费下单课购买 · 仅本课

支付成功后自动显示网盘下载链接；链接失效可联系客服免费补链，永久有效。

正版保障极速发货售后无忧

LangGraph 的核心在于将复杂的 AI 应用逻辑解耦为可管理的状态图，但官方提供的 `MessagesState` 往往只能处理最基础的对话历史存储，无法满足实际工程中对于上下文检索、关键词提取或多模态数据流转的定制需求。这份实战资料聚焦于如何通过类继承机制对 `MessagesState` 进行深度扩展，解决当业务逻辑超出简单聊天机器人范畴时，状态管理僵化且难以复用的痛点。它不适合仅停留在调用现成 API 阶段的初学者，而是面向那些已经熟悉 LangChain 基础组件，并在使用 LangGraph 构建多智能体或复杂工作流时，遇到状态定义瓶颈的开发者。

建议首先关注状态定义的继承结构这一部分。代码演示了如何从基础状态类派生出新的状态容器，并引入如 `key_keywords` 和 `key_search` 等自定义字段。理解这一点至关重要，因为它展示了如何在保持与 LangGraph 核心兼容性的同时，注入业务特定的数据结构。接着，重点分析状态图（StateGraph）的编译与节点绑定过程。资料中展示了 `CompiledStateGraph` 的使用，这是连接静态状态定义与动态执行逻辑的关键桥梁。你需要仔细研读节点函数中如何读写这些扩展后的状态字段，特别是当 `AIMessage` 生成后，如何将提取的关键信息回写至状态中，供后续节点使用。这部分内容直接关联到数据在图节点间的流转机制，是理解 LangGraph 执行引擎内部逻辑的最佳切入点。

学完这部分内容，你应当能够独立设计一个具备记忆增强能力的 RAG 检索流程。例如，构建一个智能助手，它不仅能维持对话，还能在每一轮交互中自动提取用户意图关键词，并将这些关键词持久化到状态中，供后续的向量数据库检索节点使用。这种能力使得应用从“无状态问答”进化为“有状态推理”，显著提升了复杂任务的准确性。

在练习时，不要仅仅复制粘贴代码。建议先手动构建一个最小化的 `StateGraph`，仅包含两个节点：一个负责提取关键词，一个负责基于关键词生成回答。尝试修改继承类中的字段，观察状态在节点间传递时的数据变化。如果遇到状态更新冲突或字段丢失的问题，这通常是理解 `reducer` 机制的良好契机。将这份代码作为参考模板，替换其中的 LLM 调用逻辑和检索策略，逐步构建符合你具体业务场景的状态图。这种基于源码的阅读和重构，比观看泛泛而谈的视频教程更能让你掌握构建高可用 AI 应用的核心技巧。

import os
from copyreg import constructor from langchain_core.prompt_values import StringPromptValue
from langgraph.graph import StateGraph, MessagesState
from langgraph.graph.state import CompiledStateGraph, StateGraph # 重复了
from langgraph.constants import START,END
from langchain_core.messages import AIMessage
from app.bailian.common import llm key_keywords="key_keywords"
key_search="key_search"
key_answer="key_answer" class BaiduSearchSate(MessagesState): search_question:str search_keywords:str search_results:str def node_key_keywords(state:BaiduSearchSate): last_message = state['messages'][-1] content = last_message.content state['search_question'] = content prompt = StringPromptValue(text=f"请从下面的信息中提取需要在百度中搜索的关键词，并且直接返回结果:{content}") message = llm.invoke(input=prompt) state['search_keywords'] = message return state def node_key_search(state:BaiduSearchSate): state['search_results'] ="百度查询的天气结果是: 今天0度 有点冷" return state def node_key_answer(state:BaiduSearchSate): content = state['search_results'] question = state['search_question'] message = llm.invoke(input=f""" # 要求 请结合百度提供的结果 回答用户的问题 {question} {content} """) state['messages'].append(message) return state def output_graph_image(graph,file_name): png_data = graph.get_graph().draw_png() output_file_dir = os.path.dirname(__file__) output_file_path = os.path.join(output_file_dir,file_name+'.png') with open(output_file_path,'wb') as f: f.write(png_data) state_graph = StateGraph(MessagesState)
state_graph.add_node(key_keywords,node_key_keywords)
state_graph.add_node(key_search,node_key_search)
state_graph.add_node(key_answer,node_key_answer) state_graph.add_edge(START,key_keywords)
state_graph.add_edge(key_keywords,key_search)
state_graph.add_edge(key_search,key_answer)
state_graph.add_edge(key_answer,END) compiled_graph = state_graph.compile() #output_graph_image(compiled_graph,'output_graph_image') results = compiled_graph.stream({ "messages":[("user","请问北京今天的天气如何")]
}) for result in results: print(result)