淄博网站制作公司定制广东深圳职业技术学院

淄博网站制作公司定制,广东深圳职业技术学院,河南省漯河建设局网站,建设互联网教育网站文章详细介绍了AI智能体记忆机制的技术实现#xff0c;涵盖Agent Memory基本概念、工作原理和分类方式。重点讲解了LangGraph框架下的长短期记忆实现技术#xff0c;包括短期记忆存储管理和长期记忆语义检索方法。通过引入MCP协议的实战案例#xff0c;展示了如何构建融合长…文章详细介绍了AI智能体记忆机制的技术实现涵盖Agent Memory基本概念、工作原理和分类方式。重点讲解了LangGraph框架下的长短期记忆实现技术包括短期记忆存储管理和长期记忆语义检索方法。通过引入MCP协议的实战案例展示了如何构建融合长记忆机制的Multi-Agent系统完整呈现中断、记忆与协作的融合过程为开发者提供从理论到实践的全面指导。如何让AI智能体Agent像人类一样拥有持久的记忆从而在复杂的连续任务中保持上下文感知和深度理解这已成为构建高级智能体的核心挑战。本文将深入探讨Agent Memory的核心概念并聚焦于LangGraph框架下的长短期记忆实现详解短期会话与长期知识的存储、管理、语义检索等技巧。更进一步地我们将通过一个引入MCP协议的实战案例手把手带你构建一个真实的融合长记忆机制的Multi-Agent系统直观展示中断、记忆与协作的融合。基于大语言模型LLM的智能体Agent系统中记忆机制是实现持续、连贯和个性化交互的核心基石通过记忆可以让Agent记住过往的交互保持上下文的一致性并能从反馈中学习适应用户的偏好。本文核心要点概述1.介绍Agent Memory的基本情况2.LangGraph长短期记忆详解及案例说明包含短期记忆实现、管理方法长期记忆的实现方法以及搭建了融合postgres数据库、集成Embedding服务进行语义搜索等可用于生产环境的真实案例。3.引入MCP协议构建真实的Agent长记忆应用搭建一个基于supervisor架构集成中断机制、长短期记忆机制的multi-agent系统。记忆机制介绍Agent Memory是什么上图中来源于Mem0[1]左边是没有Memory的agent右边是有Memory的agent后者可以根据用户的过往信息素食主义者、不喜欢乳制品给出更合理的响应不含乳制品的素食菜单而前者的回答显然是不合适的。简单来说Memory是赋予Agent记忆能力的技术和架构能够让Agent像人一样记住过去的交互、学到的知识、执行过的任务及未来的计划是将一个LLM转变为能够执行复杂、长期任务的真正”智能体“的核心所在。关于Agent Memory我们需要考虑什么如何获取记忆通过和用户交互、环境交互…怎么组织记忆模型参数、模型上下文、数据库怎么利用记忆RAG、Few-shot…有哪些Memory类型关于Memory的分类有许多种分类体系这里我们只讨论最常见及最易于理解的。正如人类利用长短期记忆进行有效的交互和学习一样Agent的记忆机制通常划分为短期记忆short-term memory和长期记忆(long-term memory)短期记忆决定了Agent在微观任务上的即时表现而长期记忆则作为持久知识库决定了Agent在宏观时间尺度上的智能深度和个性化水平通过两者配合Agent才能表现出连贯性、上下文感知能力才会显得更智能。Agent Memory如何工作Agent通常通过以下几步来有效地管理记忆使得每次于用户的交互都更加精准智能记忆存储通过设计一系列策略来存储重要的交互信息这些信息可能来源于对话内容、历史数据或任务要求等等。记忆更新记忆会随着交互的发生不断地进行更新例如用户的偏好、最新的近况等等。记忆更新使得Agent能够不断优化其响应。记忆检索Agent根据当下的需求去记忆中检索需要的记忆内容从而提供更加智能的回复。Agent Memory怎么实现物理外挂即外置数据库和 RAG需要检索当前query相关的内容例如Mem0、ACE。好处是即插即用坏处是不够end-to-endMemory as Reasoning / Tool通过训练Reasoning或Tool的方式动态更新context例如MemAgent、memory-R1。好处是更接近end-to-end但不是很灵活。参数更新LLM本身就是一个Memory体所有参数都是它的Memory通过更新参数来更新记忆这种方式是最本质的但也是最难实现的。LangGraph中的记忆管理LangGraph[2]作为一款面向多智能体协作与状态管理的框架其设计了巧妙的记忆管理系统旨在为Agent提供在不同交互中存储、检索和利用信息的能力。它区分了两种主要的记忆类型短期记忆和长期记忆。在实际使用中通过这两种记忆协同既能保障实时任务的高效执行又支持了跨任务、跨周期的经验复用。● short-term memory通过Checkpointer实现针对单个对话线程核心价值在于保障对话的临时性使得Agent能够跟踪会话中的多轮对话可以在该线程内的任何时刻被回忆。● long-term memory通过Store实现可以跨对话线程共享可以在任何时间任何线程中被回忆而不像短期记忆局限于单个对话。通过下表可以更清晰的看到两者的区别LangGraph记忆的架构基础要想更好的理解LangGraph中的记忆机制首先需要理解其支持双轨记忆系统的核心概念。CheckpointerLangGraph有一个内置的持久化Persistence层通过checkpointer实现能够持久化存储图状态这使得开发记忆功能和人类干预功能成为可能。当使用检查点编译一个图时检查点会在每个super-step保存图状态的checkpoint这些checkpoint被保存到一个thread中可以在图执行后访问。因为threads允许在执行后访问图的状态所以可以实现记忆、人机协作、时间旅行、容错等多种强大的功能。工作流程用户输入 → [节点 1] → 保存状态 → [节点 2] → 保存状态 → 输出↓ ↓Checkpoint 1 Checkpoint 2Thread为了管理多个独立的对话LangGraph使用了thread的概念。thread_id是由checkpointer保存的每个checkpoint的唯一id是激活和区分不同对话线程的唯一key。在调用图的invoke或stream方法时通过configurable字典传入一个thread_id就代表这次操作属于thread_id这个特定的对话。Store如上所述图状态可以由checkpointer在每个super-step写入线程从而实现状态的持久化。但是如果想在多个线程之间保留一些信息的话那么就需要用到Store。Store本质上是一个暴露给图节点和工具的键值数据库与checkpointer的自动化快照不同Store需要显式和主动的进行操作。NamespaceStore中的数据通常通过更持久的标识来组织。user_id是最常见的用于关联用户的所有信息此外namespace提供了一种数据隔离机制例如使用使用 (“memories”, user_id) 这样的元组作为命名空间可以将用户的记忆与其他类型的数据如用户偏好 (“preferences”, user_id)清晰地分离开来避免数据冲突保持知识库的整洁有序。短期记忆详解InMemorySaver内存会话临时存储对于开发、原型设计和测试阶段最简单快捷的方式是使用InMemorySaver。它将所有的对话状态存储在内存中的一个Python字典里。1.设置记忆管理检查点from langchain_openai import ChatOpenAIfrom langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.prebuilt import create_react_agent # 初始化检查点保存器checkpointer InMemorySaver()2.定义大模型并创建agentBASE_URL TOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)agent create_react_agent( modelmodel, tools[], # 传入检查点是将持久化能力“注入”图的关键步骤。编译后的graph对象现在具备了状态管理的能力。 checkpointercheckpointer)如果是底层自定义api在图构建阶段传入检查点的代码是graph builder.compile(checkpointercheckpointer)。3.短期记忆-内存后端config {configurable: {thread_id: 1}} # 激活记忆机制的核心。如果没有提供thread_id每次invoke调用都将是无状态的只要使用相同的thread_idLangGraph就会在多次调用之间维持对话状态response agent.invoke( {messages: [{role: user, content: 你好我叫ada}]}, config)print(fthread1_bot_answer{response[messages][-1].content})response agent.invoke( {messages: [{role: user, content: 你好请问你还记得我叫什么名字么}]}, config)print(------------线程1------------------)print(fthread1_bot_answer{response[messages][-1].content})new_config {configurable: {thread_id: 2}}response agent.invoke( {messages: [{role: user, content: 你好请问你还记得我叫什么名字么}]}, new_config)print(------------线程2------------------)print(fthread2_bot_answer{response[messages][-1].content})执行上面代码可以看到输出如下thread1_bot_answer你好Ada很高兴认识你 这是一个很美的名字呢有什么我可以帮助你的吗无论是想聊聊天还是有任何问题需要解答我都很乐意为你提供帮助。------------线程1------------------thread1_bot_answer当然记得你刚才告诉我你叫 Ada很高兴再次和你打招呼------------线程2------------------thread2_bot_answer你好很抱歉我无法记住之前对话中的个人信息比如你的名字。这是为了保护你的隐私所以我不会保留这类数据。你可以告诉我你的名字或者任何你想让我称呼你的方式我会很乐意在这次的对话中使用它短期记忆与线程相关在对话时需要在配置中传入thread_id。通过上面的结果我们可以看到当我们传入相同的thread_id时agent就可以记住用户的名字然而当我们更换thread_id时agent就不记得用户的名字了。需要注意的是InMemorySaver将所有状态都保存在内存中一旦程序终止那么所有对话历史都会消失。数据库持久化存储可以发现上面一小节的代码在应用程序结束后再启动记忆就又消失了。这是因为InMemorySaver仅仅是把记忆保存在内存中应用程序结束后释放内存记忆就消失了。在生产环境中常常使用数据库支持的检查点记录器持久化保存记忆以保证数据的可靠性和服务的连续性。这里我们以postgres数据库为例来说明怎么持久化地保存记忆数据。1.首先安装以下依赖pip install -U psycopg[binary,pool] langgraph-checkpoint-postgres2.安装postgres数据库具体的安装方法可以参考[Linux下安装PostgreSQL_linux安装postgresq下载选择最新的postgresql-18.0.tar.gz。3.安装数据库成功后编码如下代码。DB_URI是数据库连接的URL。想要自动保存在数据库中的State需要在PostgresSaver.from_conn_string(DB_URI)上下文中操作。from langchain.chat_models import init_chat_modelfrom langgraph.graph import StateGraph, MessagesState, STARTfrom langgraph.checkpoint.postgres import PostgresSaverBASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)DB_URI postgresql://postgres:postgreslocalhost:5432/postgres?sslmodedisablewith PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 第一次调用时必须要setup() def call_model(state: MessagesState): response model.invoke(state[messages]) return {messages: response} builder StateGraph(MessagesState) builder.add_node(call_model) builder.add_edge(START, call_model) graph builder.compile(checkpointercheckpointer) config { configurable: {thread_id: 1 } } response graph.invoke( {messages: [{role: user, content: 你好我叫ada}]}, config ) print(response[messages][-1].content) response graph.invoke( {messages: [{role: user, content: 你好请问你还记得我叫什么名字么}]}, config ) print(response[messages][-1].content)运行一次上述代码后关闭应用程序后重启再次运行上述代码print结果如下bot_answer_1你好Ada很高兴再次见到你 你的名字真动听今天有什么我可以帮你解答或者想聊的话题吗bot_answer_2当然记得你告诉我你叫 **Ada**。很高兴再次和你打招呼可以看到记忆已经被保存了。我们检查数据库可以发现postgres数据库中出现了四个表上述表中checkpoints表是”状态快照“表每当程序执行一个step时它就会在这张表中创建一条新记录这条记录就是一个检查点的快照。查询该表可以得到如下结果接下来我们来分析每一列的含义理解了上面checkpoints表后那么不禁会问真正的消息内容被存到了哪里呢真正的消息内容存储在checkpoint_writes表中如下除了PostgreSQL之外LangGraph还支持MongoDB、Redis等数据库。子图中的记忆当构建复杂的、由多个子图嵌套而成的应用时需要更灵活的记忆管理策略。● 记忆继承默认默认情况下子图会继承其父图的checkpointer。这意味着整个嵌套图共享同一个对话状态数据可以在父子图之间无缝流动。这对于将一个大型任务分解为多个模块化子任务非常有用。● 记忆隔离在某些场景下例如构建多智能体系统希望每个智能体由一个子图表示拥有自己独立的内存空间互不干扰。此时可以在编译子图时设置checkpointerTrue。如下代码可以在子图中直接使用父图的短期记忆from langgraph.graph import START, StateGraphfrom langgraph.checkpoint.memory import InMemorySaverfrom typing import TypedDictclass State(TypedDict): foo: str# 子图def subgraph_node_1(state: State): return {foo: state[foo] bar}subgraph_builder StateGraph(State)subgraph_builder.add_node(subgraph_node_1)subgraph_builder.add_edge(START, subgraph_node_1)subgraph subgraph_builder.compile() # 父图builder StateGraph(State)builder.add_node(node_1, subgraph) builder.add_edge(START, node_1)checkpointer InMemorySaver()graph builder.compile(checkpointercheckpointer)如果子图希望使用自己的短期记忆那么需要在编译子图时显示传入子图的检查点subgraph_builder StateGraph(...)subgraph subgraph_builder.compile(checkpointerTrue)工具中的记忆交互在工具中读取状态LangGraph允许工具直接访问和读取当前的图状态使其具备上下文感知能力。核心机制state: Annotated[CustomState, InjectedState]InjectedState的作用是在调用这个工具时将当前的完整状态对象作为第一个参数传递到工具中使得这个工具能根据当前状态来执行更智能的操作。from typing import Annotatedfrom langchain.chat_models import init_chat_modelfrom langgraph.prebuilt import InjectedState, create_react_agentfrom langgraph.prebuilt.chat_agent_executor import AgentStateBASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)class CustomState(AgentState): user_id: strdef get_user_info( state: Annotated[CustomState, InjectedState]) - str: 查询用户信息 user_id state[user_id] returnadaif state[user_id] user_adaelseUnknownagent create_react_agent( modelmodel, tools[get_user_info], state_schemaCustomState,)agent.invoke({ messages: 查询用户信息, user_id: user_ada})返回结果如下根据查询结果您的用户信息显示用户名为**ada**\n\n这是您的用户信息查询结果。如果您需要了解其他信息或有其他需求请告诉我。在工具中写入状态如果要在工具执行期间修改图的记忆那么可以直接从工具返回状态更新。这对于持久化中间结果、传递信息给后续工具等非常有用。核心机制工具返回Command对象。此时LangGraph会将其返回值解释为对状态的直接修改指令。Command(update{…})中的字典定义了要更新的状态字段及其新值。这允许工具在完成其主要任务的同时将结果写回智能体的短期记忆中从而影响后续的决策。from typing import Annotatedfrom langchain_core.tools import InjectedToolCallIdfrom langchain_core.runnables import RunnableConfigfrom langchain_core.messages import ToolMessagefrom langgraph.prebuilt import InjectedState, create_react_agentfrom langgraph.prebuilt.chat_agent_executor import AgentStatefrom langgraph.types import Commandclass CustomState(AgentState): user_name: strdef update_user_info( tool_call_id: Annotated[str, InjectedToolCallId], config: RunnableConfig) - Command: 查询并更新用户信息 user_id config[configurable].get(user_id) name adaif user_id user_123elseUnknown user return Command(update{ user_name: name, # 更新消息历史记录 messages: [ToolMessage( 成功查询到用户信息, tool_call_idtool_call_id) ] })def greet( state: Annotated[CustomState, InjectedState]) - str: 找到用户信息后使用此方式向用户问好。 user_name state[user_name] return f你好 {user_name}agent create_react_agent( modelmodel, tools[update_user_info, greet], state_schemaCustomState)agent.invoke( {messages: [{role: user, content: 向用户打招呼}]}, config{configurable: {user_id: user_123}})输出结果如下长期记忆详解LangGraph中的长期记忆允许系统在不同对话中保留信息是跨对话线程共享的可以在任何时间、任何线程中被回忆。与短期记忆不同长期记忆保存在自定义的命名空间中每个记忆都组织在一个自定义的namespace和一个唯一的key下。记忆存储LangGraph将长期记忆存储为JSON文档使用Store进行管理允许存储结构化和非结构化的数据。记忆更新时机● 热路径Hot Path在应用程序逻辑运行时实时创建记忆store.put()优点是实时更新但可能增加程序复杂性、延迟等问题。● 后台Background作为单独的异步任务创建记忆store.put()优点是避免主应用延迟、逻辑分离难点在于确定更新频率和触发时机。记忆检索● store.get()根据命名空间和键精确获取记忆。● store.search()在指定命名空间内实现灵活记忆检索不但可以通过命名空间和标识符更可以通过语义检索到记忆内容。通常需要Store配置一个embed来支持语义搜索。记忆的应用● 语义记忆存储事实和概念。分为以下两种情况Profile将关于用户、组织或代理自身的特定信息存储为一个持续更新的JSON文档需要模型来生成新的Profile或更新已有JSON档案Collection将记忆存储为一组独立的文档易于生成但检索和更新较为复杂且可能难以捕获记忆间的完整上下文。在应用时可以将检索到的记忆作为上下文或系统指令的一部分传递给LLM用于个性化响应和回答事实性问题。● 情景记忆存储过去的事件或行为经验。通常通过few-shot example prompt来实现以指导模型完成任务。● 程序记忆存储执行任务的规则或指令。通常通过修改代码自身的prompt来实现将其应用于LLM。InMemoryStore与Checkpointer类似InMemoryStore用于快速开发和原型验证。它将所有数据存储在内存中。from langchain.chat_models import init_chat_modelfrom langchain_core.runnables import RunnableConfigfrom langgraph.store.memory import InMemoryStorefrom langgraph.config import get_storefrom langgraph.prebuilt import create_react_agentstore InMemoryStore()BASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)store.put( (users,), # 命名空间元组类型类比文件系统中的文件夹支持分层组织结构 user_123, # 键: 字符串是命名空间内的唯一标识符一般推荐使用uuid库生成唯一标识符 { name: ada, language: 中文, } # 值Python字典类型比如保存公共角色资料时可以是包含姓名、偏好等键值对的字典)def get_user_info(config: RunnableConfig) - str: 查找用户信息的函数可以查看长期记忆中储存的用户信息 store get_store() # 获取上下文中可用的store实例 user_id config[configurable].get(user_id) user_info store.get((users,), user_id) # 输入命名空间和键进行精确查询 return str(user_info.value) if user_info elseUnknown useragent create_react_agent( modelmodel, tools[get_user_info], # 传入store storestore)response agent.invoke( {messages: [{role: user, content: 帮我查找长期记忆中储存的用户信息}]}, config{configurable: {user_id: user_123}})print(response[messages])输出结果如下可以看到在工具函数中成功调用store查找到了保存的用户信息数据库持久化存储为了让记忆真正”长期“生产环境必须使用数据库支持的StoreLangGraph目前主要支持PostgresStore和RedisStore。我们以PostgresStore为例来进行说明。首先先安装对应的包pip install -U psycopg[binary,pool] langgraph-checkpoint-postgresPostgres数据库的安装请参考上文。接下来进行示例说明。整体过程与Checkpointer类似关键区别在于Store是怎样在节点内部被访问和使用的。import uuidfrom langchain.chat_models import init_chat_modelfrom langchain_core.runnables import RunnableConfigfrom langgraph.graph import StateGraph, MessagesState, STARTfrom langgraph.store.base import BaseStorefrom langgraph.store.postgres import PostgresStorefrom langgraph.checkpoint.postgres import PostgresSaverBASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)DB_URI postgresql://postgres:postgreslocalhost:5432/postgres?sslmodedisablewith ( PostgresStore.from_conn_string(DB_URI) as store, PostgresSaver.from_conn_string(DB_URI) as checkpointer,): store.setup() # 第一次调用时必须要setup()#checkpointer.setup() # 声明store参数 def call_model( state: MessagesState, config: RunnableConfig, *, store: BaseStore, # 在节点中访问store的标准方式需要在函数签名上加一个store ): # 从store中读取记忆 user_id config[configurable][user_id] namespace (memories, user_id) memories store.search(namespace, querystr(state[messages][-1].content)) info \n.join([d.value[data] for d in memories]) system_msg f你是一个与人类交流的小助手用户信息: {info} # 向store中写入记忆 last_message state[messages][-1] if记住in last_message.content.lower():memory 用户名字是adastore.put(namespace, str(uuid.uuid4()), {data: memory}) response model.invoke([{role: system, content: system_msg}] state[messages]) return {messages: response} builder StateGraph(MessagesState) builder.add_node(call_model) builder.add_edge(START, call_model) graph builder.compile(checkpointercheckpointer, storestore) # agent同时配备了短期记忆和长期记忆能力 # 第一次对话告诉agent用户的名字 config { configurable: {thread_id: 3,user_id: 1, } } response graph.invoke( {messages: [{role: user, content: 你好我叫ada记住这个名字呦~}]}, config ) print(response[messages][-1].content) # 第二次对话新线程询问agent记不记得用户的名字 config { configurable: {thread_id: 4,user_id: 1, } } response graph.invoke( {messages: [{role: user, content: 我的名字是什么?}]}, config ) print(response[messages][-1].content)输出结果如下# ------------------第一次对话----------------------你好ada很高兴认识你我已经记住你的名字啦✨ 有什么我可以帮你的吗# ------------------第二次对话----------------------你的名字是ada在第一次对话时对话线程id为3agent被要求记住用户的名字并且根据代码逻辑用户名字信息是通过store.put()写入数据库的。第二次对话时线程id为4当被问起用户的名字时agent通过store.search()办法从数据库中检索到了这个信息并成功回答这展示了Store的跨记忆存储能力。长期知识赋能工具在工具中读取长期记忆参考上文中长期记忆-InMemoryStore中的示例。其中核心在于store get_store() 这个函数是一个上下文感知的辅助函数能够在工具执行时自动获取并返回compile或create_react_agent中传入的store实例。在工具中写入长期记忆from typing_extensions import TypedDict from langchain.chat_models import init_chat_modelfrom langgraph.config import get_storefrom langchain_core.runnables import RunnableConfigfrom langgraph.prebuilt import create_react_agentfrom langgraph.store.memory import InMemoryStorestore InMemoryStore() BASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)class UserInfo(TypedDict): name: strdef save_user_info(user_info: UserInfo, config: RunnableConfig) - str: 将用户信息保存到store store get_store() user_id config[configurable].get(user_id) store.put((users,), user_id, user_info) return成功保存了用户信息agent create_react_agent( modelmodel, tools[save_user_info], storestore)agent.invoke( {messages: [{role: user, content: 我叫ada请你记住我的名字}]}, config{configurable: {user_id: user_123}} )首先需要定义要存储的数据内容即UserInfo它为LLM提供了一个清晰的结构化输出格式当LLM决定调用save_user_info的工具时会自动生成一个包含name字段的字典。然后调用store.put()方法将数据存储下来。plaintextstore.get((“users”,), “user_123”).value # 输出{‘name’: ‘ada’}##### **语义搜索** Store最强大的功能之一是支持语义搜索这能将Store从一个简单的键值数据库转变为一个功能完备的向量数据库。智能体不再只能通过精确的关键词来检索记忆而是能够根据概念的相似性来查找相关信息。**这实际就是一套RAG流程**。 下面我们将基于自定义部署的Embedding服务来演示如何进行长期记忆语义搜索。特别说明的是代码仅供演示使用实际使用可以参考下面代码编写更规范的代码。 1.首先我们需要自己**部署一个Embedding服务**这里我们以Qwen3-Embedding-4B为例。 2.**创建自定义Embedding类**这个类需要继承自langchain.embeddings.base.Embeddings这个类的作用是负责与Embedding服务进行通信。 plaintext import requestsfrom typing import List, Optional, Dictfrom langchain.embeddings.base import Embeddingsclass SelfAPIEmbeddings(Embeddings): 一个自定义的 Embedding 类用于调用自部署的 embedding API 服务。 def __init__(self): 初始化函数。 self.token self.url self.model_id def _call_api(self, texts: List[str]) - List[List[float]]: 内部方法用于调用 API。 *** 您需要根据您自己服务的实际 API 格式来修改这部分 *** try:payload { model: self.model_id, input: texts}headers { Content-Type: application/json, Authorization: fBearer {self.token}}response requests.post(self.url, headersheaders, datajson.dumps(payload))# 判断是否异常if response.status_code ! 200: print(response.json()) exit()res response.json()return res[data] except requests.exceptions.RequestException as e:print(fError calling embedding API: {e})# 可以选择返回空列表或重新抛出异常raise e def embed_documents(self, texts: List[str]) - List[List[float]]: 为文档列表生成 embeddings。 if not texts:return [] # 为了避免请求体过大可以分批处理 # 这里为了简单起见一次性发送所有文本 new_texts [] for i in texts:i eval(i)new_texts.append(i[text]) res self._call_api(new_texts) new_res [] for i in res:new_res.append(i[embedding]) return new_res def embed_query(self, text: str) - List[float]: 为单个查询文本生成 embedding。 if not text:return [] result self._call_api([text]) return result[0][embedding]3.将自定义的Embedding类集成到工作流中通过在Store中配置index来启用语义搜索。from langchain.embeddings import init_embeddingsfrom langgraph.store.memory import InMemoryStore custom_embeddings SelfAPIEmbeddings()store InMemoryStore( index{ embed: custom_embeddings, dims: 2560, })4.语义查询测试。store.put((user_123, memories), 1, {text: 我喜欢吃披萨})store.put((user_123, memories), 2, {text: 我是一名程序员}) items store.search( (user_123, memories), query我肚子饿了, limit1)输出如下尽管查询没有”披萨“这个词但是通过Embedding模型计算知道披萨和饿了是相近的语义因此成功检索出了相关的记忆。[Item(namespace[user_123, memories], key1, value{text: 我喜欢吃披萨}, created_at2025-11-12T09:59:55.09793100:00, updated_at2025-11-12T09:59:55.09793700:00, score0.6804530799409887)]短期记忆管理策略随着对话的进行短期记忆对话历史会不断增长可能会超出LLM的上下文窗口导致请求调用失败或者使LLM反应变慢、变差。这时就需要对记忆进行管理了。常见的解决办法有● 修剪消息trim messages移除前 N 条或后 N 条消息在调用 LLM 之前。最简单直接但信息丢失严重适合短期任务、无状态问答机器人、近期上下文最重要的应用。● 删除消息delete messages从LangGraph状态中永久删除消息。可以精确的控制移除内容但需要自定义逻辑来判断哪些消息需要删除适合用于移除不再需要的冗余系统消息、工具输出或错误信息。● 总结消息summarize messages汇总历史记录中的早期消息并将其替换为摘要。保留了核心语义信息但计算成本高实现相对复杂适合用于长期连续对话、需要维持深度长期上下文的智能体。● 自定义策略例如消息过滤等。修剪消息管理对话历史的一个重要概念是限制传递给模型的消息数量trim_messages就是LangChain提供的一个实用函数它根据指定的策略、token限制、模型要求以及是否包含系统消息等来裁剪消息列表它的主要目的是确保对话历史不会超出模型的上下文窗口大小。它的解决策略是当消息历史过长时从开头或结尾丢弃一部分消息以确保总长度符合限制。from langchain_core.messages.utils import ( trim_messages, count_tokens_approximately)from langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import StateGraph, START, MessagesStateBASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)summarization_model model.bind(max_tokens128)def call_model(state: MessagesState): # 保留最近消息总 token ≤ 128 messages trim_messages( state[messages], strategylast, # 保留最后的消息 token_countercount_tokens_approximately, max_tokens128, start_onhuman, # 确保第一条消息不包括系统消息是从human消息开始保留 end_on(human, tool), # 保留到human或tool消息为止 allow_partialFalse, # 不允许分割消息内容 include_systemTrue # 保留system prompt ) # --- 在这里打印传入模型的内容 --- print(- * 20) print(fMessages being sent to the model (trimmed to 128 tokens): {len(messages)}) for msg in messages: print(f [{msg.type.upper()}]: {msg.content}) print(- * 20) response model.invoke(messages) return {messages: [response]}checkpointer InMemorySaver()builder StateGraph(MessagesState)builder.add_node(call_model)builder.add_edge(START, call_model)graph builder.compile(checkpointercheckpointer)# 如果需要在agent中修剪那么需要将pre_model_hook和trim_messages结合使用def call_model(state) messages trim_messages(...) agent create_react_agent( model, tools, pre_model_hookcall_model, checkpointercheckpointer,)发起请求如下config {configurable: {thread_id: 1}} graph.invoke({messages: 你好我叫ada}, config)graph.invoke({messages: 请写一首诗关于小狗的}, config)graph.invoke({messages: 再写一首关于小猫的}, config)final_response graph.invoke({messages: 我叫什么名字呢}, config)输出打印如下--------------------Messages being sent to the model (trimmed to 128 tokens): 1 [HUMAN]: 你好我叫ada----------------------------------------Messages being sent to the model (trimmed to 128 tokens): 3 [HUMAN]: 你好我叫ada [AI]: 你好Ada很高兴认识你这是一个很美的名字呢有什么我可以帮助你的吗无论是想聊天、有问题需要解答还是需要任何形式的帮助我都很乐意为你服务。 [HUMAN]: 请写一首诗关于小狗的----------------------------------------Messages being sent to the model (trimmed to 128 tokens): 5 [HUMAN]: 你好我叫ada [AI]: 你好Ada很高兴认识你这是一个很美的名字呢有什么我可以帮助你的吗无论是想聊天、有问题需要解答还是需要任何形式的帮助我都很乐意为你服务。 [HUMAN]: 请写一首诗关于小狗的 [AI]: 好的Ada为你写一首关于小狗的可爱小诗希望你喜欢**《小狗的约定》**..... 此处省略诗的内容这首诗捕捉了小狗的活泼、忠诚和它们带给我们的温暖。你觉得怎么样 [HUMAN]: 再写一首关于小猫的----------------------------------------Messages being sent to the model (trimmed to 128 tokens): 3 [HUMAN]: 再写一首关于小猫的 [AI]: 好的Ada这首关于小猫的诗希望同样能带给你一丝轻盈与温柔**《小猫的遐想》**..... 此处省略诗的内容希望你喜欢这首小诗 [HUMAN]: 我叫什么名字呢--------------------模型最终的输出为可以看出传递给模型的消息内容已经被裁剪修剪的过程为1.保留系统消息include_systemTrue2.strategy“last”反转消息列表以便从最新的消息开始处理3.累积token数量当达到max_tokens限制那么进行修剪4.修剪时由于allow_partialFalse因此保留的消息都是完整的且start_on“human”所以修剪后第一条非system prompt是用户消息虽然对传递给模型的历史消息进行了裁剪但是查询state可以发现历史记录仍被完整的保留在内存中没有被删除。print(\n *30)print( 查看 thread_id1 的完整对话历史)print(*30) current_state graph.get_state(config)conversation_history current_state.values[messages] for message in conversation_history: print(f[{message.type.upper()}]: {message.content})删除消息这种方法允许从状态中永久移除特定的消息。要删除消息不能直接从状态的messages列表中移除而是使用RemoveMessage函数从graph state中直接删除消息来管理对话历史。为了让RemoveMessage生效需要使用带有add_messages reducer的状态键例如MessagesState。删除特定消息from langchain_core.messages import RemoveMessagefrom langchain.chat_models import init_chat_modelfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.graph import StateGraph, START, MessagesStateBASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)def delete_messages(state): messages state[messages] if len(messages) 2: # 删除最早的两条消息 return {messages: [RemoveMessage(idm.id) for m in messages[:2]]} def call_model(state: MessagesState): response model.invoke(state[messages]) return {messages: response}builder StateGraph(MessagesState)builder.add_sequence([call_model, delete_messages])builder.add_edge(START, call_model)checkpointer InMemorySaver()app builder.compile(checkpointercheckpointer)config {configurable: {thread_id: 1}}for event in app.stream( {messages: [{role: user, content: 你好我是ada}]}, config, stream_modevalues): print([(message.type, message.content) for message in event[messages]])for event in app.stream( {messages: [{role: user, content: 我叫什么名字}]}, config, stream_modevalues): print([(message.type, message.content) for message in event[messages]])输出如下当请求完成时如果消息数量2那么最早的两条消息会被删除。清空所有消息from langgraph.graph.message import REMOVE_ALL_MESSAGES def clear_messages(state): return {messages: [RemoveMessage(idREMOVE_ALL_MESSAGES)]}builder StateGraph(MessagesState)builder.add_sequence([call_model, clear_messages])builder.add_edge(START, call_model)checkpointer InMemorySaver()app builder.compile(checkpointercheckpointer)config {configurable: {thread_id: 1}}for event in app.stream( {messages: [{role: user, content: 你好我是ada}]}, config, stream_modevalues): print([(message.type, message.content) for message in event[messages]])for event in app.stream( {messages: [{role: user, content: 我叫什么名字}]}, config, stream_modevalues): print([(message.type, message.content) for message in event[messages]])输出如下请求完成后会立即删除所有消息记录。总结消息通过修剪、删除来管理历史消息会有丢失信息的问题。为了避免这个问题可以进行消息总结也就是通过调用LLM对历史对话进行摘要并将摘要作为新的上下文传入以在减少消息数量的同时保留关键信息。1.首先安装LangMem这是一个由LangChain维护的库提供了用于在agent中管理记忆的工具。pip install -U langmem2.langmem库提供了一个预构建的SummarizationNode可以极大地简化实现过程import tiktokenfrom typing import Any, TypedDictfrom langchain.chat_models import init_chat_modelfrom langchain_core.messages import AnyMessage, BaseMessage, SystemMessagefrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langgraph.graph import StateGraph, START, MessagesStatefrom langgraph.checkpoint.memory import InMemorySaverfrom langmem.short_term import SummarizationNode, RunningSummarysummary_prompt ChatPromptTemplate.from_messages( [ MessagesPlaceholder(variable_namemessages), # 使用 HumanMessage 模拟用户在最后发出总结指令 (human, 请根据以上对话生成一段简洁、连贯的中文摘要注意不要丢失细节), ])update_summary_prompt ChatPromptTemplate.from_messages( [ MessagesPlaceholder(variable_namemessages), # 使用 HumanMessage 模拟用户在最后发出总结指令 (human, 以下是目前为止的对话摘要{existing_summary}\n\n请根据以上新消息扩展此摘要), ])BASE_URLTOKENMODEL_NAMEmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)summarization_model model.bind(max_tokens128)# count_tokens_approximately更适合英文分词中文这里使用tiktoken来计算token数量encoding tiktoken.get_encoding(cl100k_base)def count_tokens_accurately(messages: list[BaseMessage]) - int: 使用 tiktoken 精确计算消息列表的 token 总数 # 注意langmem 的 token_counter 期望的输入是消息列表 text_content .join([msg.content for msg in messages if isinstance(msg.content, str)]) return len(encoding.encode(text_content))class State(MessagesState): context: dict[str, RunningSummary] class LLMInputState(TypedDict): summarized_messages: list[AnyMessage] context: dict[str, RunningSummary]summarization_node SummarizationNode( #token_countercount_tokens_approximately, token_countercount_tokens_accurately, # 更换为自定义的token计算工具 modelsummarization_model, max_tokens256, max_tokens_before_summary256, max_summary_tokens128, initial_summary_promptsummary_prompt, # 使用自定义prompt默认为英文 existing_summary_promptupdate_summary_prompt)def call_model(state: LLMInputState): response model.invoke(state[summarized_messages]) return {messages: [response]}checkpointer InMemorySaver()builder StateGraph(State)builder.add_node(call_model)builder.add_node(summarize, summarization_node) builder.add_edge(START, summarize)builder.add_edge(summarize, call_model)graph builder.compile(checkpointercheckpointer)# Invoke the graphconfig {configurable: {thread_id: 1}}graph.invoke({messages: 你好我叫ada}, config)graph.invoke({messages: 请写一首诗关于小狗的}, config)graph.invoke({messages: 再写一首关于小猫的}, config)final_response graph.invoke({messages: 我叫什么名字}, config)final_response[messages][-1].pretty_print()# 检查摘要是否生成ifrunning_summaryin final_response[context]: print(\n生成的摘要:, final_response[context][running_summary].summary)else: print(\n对话较短尚未生成摘要。)输出如下如果需要在agent中实现的话那么将SummarizationNode传入pre_model_hook即可。SummarizationNode会自动检查历史消息的长度当token数量超过阈值时触发一次总结然后将”摘要 最新消息“的组合传递给model。其中initial_summary_prompt用于第一次生成摘要时的prompt模板existing_summary_prompt用于更新现有摘要的prompt模板final_prompt是将摘要与剩余的消息合并后的prompt模板。State中的context字段用于存储运行中的摘要信息 避免在每次调用时都重复总结。检查点管理对有状态的agent的记忆进行检查、管理和重置对于监控agent和提高用户使用体验都必不可少LangGraph提供了以下的一些工具用来对检查点进行管理。查看最近的短期记忆也就是最近的检查点的状态graph.get_state(configconfig)查看线程的所有短期记忆会按时间顺序返回这个线程所有的历史检查点graph.get_state_history(configconfig)删除一个线程的所有短期记忆一般用于重启对话场景checkpointer.delete_thread(thread_id)引入MCP协议构建真实的Agent长记忆应用本节将介绍如何基于Model Context ProtocolMCP协议使用LangGraph-Supervisor框架构建一个实用的、集成中断机制、有长记忆的多Agent系统。MCP是一种社区共建的AI开放协议它标准化了应用向AI提供上下文的方式极大简化了工具集成过程。接下来我们从头开始搭建multi-agent系统模拟一个用户去进行旅游信息查询并进行酒店预定然后酒店管理侧可以查询用户的预定信息。整个Demo我们将展示Supervisor框架的搭建、人工介入、长短期记忆的应用等。我们将逐步构建multi-agent工作流的每个组件它包含三个子智能体三个专门的 ReAct推理和行动子智能体然后它们将组合起来创建一个包含额外步骤的多智能体工作流。我们的工作流从以下开始1.human_input:用户输入;admin_input:管理员输入2.supervisor协调三个子agent根据input内容选择合适的agent进行工作3.当supervisor选择调用search_assistant的时候那么查询信息,并将结果返回4.当supervisor选择调用hotel_assistant的时候那么把用户的预定信息,更新到Store中5.当supervisor选择调用booking_info_assistant会先进行verify_info中断图的执行以请求管理员ID,当输入管理员ID后接着判断管理员ID是否符合要求如果不符合,那么不进行记忆查询如果符合则查询记忆,并返回。步骤一环境准备与安装pip install langchain-mcp-adapters步骤二模型初始化BASE_URL TOKENMODEL_NAMEfrom langchain_openai import ChatOpenAIfrom langchain.chat_models import init_chat_modelmodel init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0,)步骤三初始化长短期记忆from langgraph.store.memory import InMemoryStorefrom langgraph.checkpoint.memory import InMemorySaver store InMemoryStore()checkpointer InMemorySaver()步骤四工具与助手配置搜索助手from typing import Listfrom typing_extensions import TypedDictfrom langchain_core.messages import BaseMessagefrom langchain_core.runnables import RunnableConfigfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langgraph.prebuilt import create_react_agentfrom langgraph.config import get_storefrom langgraph.graph import StateGraph, ENDfrom langgraph.types import interruptfrom langchain_core.messages import AIMessage, ToolMessage, HumanMessage# 搜索功能url TOKEN search_client MultiServerMCPClient( { other_search: {url: url,headers: { Authorization: fBearer {TOKEN}},transport: sse } })search_tools await search_client.get_tools()search_agent create_react_agent( model, search_tools, namesearch_assistant, prompt你是一个能搜索各种信息的助手。)酒店预定助手定义图节点之间流动的共享数据结构将需要存储的记忆格式化。class UserInfo(TypedDict): user_id: str hotel_name: str date: str num_guests: int定义预定酒店子智能体并将用户预定历史存储下来。def book_hotel(user_info: UserInfo, config: RunnableConfig): 处理酒店预订并更新长期记忆 user_id config[configurable].get(user_id) print(user_info) hotel_name user_info.get(hotel_name) date user_info.get(date) num_guests user_info.get(num_guests) # 存储用户个人预订历史 namespace (user_bookings,) user_bookings store.get(namespace, user_id) or [] user_bookings.append(user_info) store.put(namespace, user_id, user_bookings) # 更新总预订计数 namespace (total_hotel_bookings,) total_bookings store.get(namespace, total_bookings_num) or 0 store.put(namespace, total_bookings_num, total_bookings 1) return f成功为用户 {user_id} 预订了 {hotel_name}入住日期{date}入住人数{num_guests}book_hotel_agent create_react_agent( modelmodel, tools[book_hotel], storestore, namehotel_assistant, prompt你是一个酒店预定助手。不需要用户ID、身份证号码、姓名和联系方式就可以预定请直接预定)查询助手用户的预定信息都是需要保密的只有特定的管理员才可以进行查询因此需要设计中断机制审核请求查询用户的权限是否符号要求即在正式查询前先中断一下要求输入管理员ID信息等输入后接着执行图再去判断管理员ID信息是否符合要求只有符合才能正常进行用户信息查询。1.查询工具定义def query_booking_from_store(config: RunnableConfig) - str: 根据用户ID从存储中查询酒店预订信息。 store get_store() user_id config[configurable].get(user_id) booking_info store.get((user_bookings,), user_id) if booking_info and booking_info.value: return f已找到预订信息{str(booking_info.value)} else: return 未找到该用户的预订信息2.定义子图的状态class SubgraphState(TypedDict): messages: List[BaseMessage]3.创建新的图节点负责中断和验证def authentication_and_query_node(state: SubgraphState, config: RunnableConfig): 这个节点首先中断图的执行以请求管理员ID 然后在恢复后验证ID并调用工具查询信息。 # 核心调用 interrupt() 来暂停图的执行 admin_input interrupt(请输入管理员id如需退出查询请输入exit) # 当图被恢复时admin_input 将会获得传入的值 if admin_input exit: result 用户已退出查询。 elif admin_input admin_123: # 验证成功调用真正的查询工具 result query_booking_from_store(config) else: # 验证失败 result f没有权限查询admin_id 不匹配 (输入为: {admin_input}) return {messages: [AIMessage(contentresult)]}4.构建包含中断节点的子agentquery_workflow StateGraph(SubgraphState)query_workflow.add_node(auth_and_query, authentication_and_query_node)query_workflow.set_entry_point(auth_and_query)query_workflow.add_edge(auth_and_query, END) booking_query_subgraph query_workflow.compile(checkpointercheckpointer, storestore) # 为子图命名以便Supervisor可以调用它booking_query_subgraph.name booking_info_assistant步骤五Supervisor架构构建基于上述组件我们将构建一个完整的Supervisor架构工作流。from langgraph_supervisor import create_supervisor workflow create_supervisor( [search_agent, book_hotel_agent, booking_query_subgraph], modelmodel, prompt( 您是团队主管负责管理信息搜索助手、酒店预订助手、以及用户信息查询助手。 如需搜索各种信息请交由 search_assistant 处理。 如需预定酒店请交由 hotel_assistant 处理。 如需查询用户的酒店预定信息请交由 booking_info_assistant 处理。 **注意**你每次只能调用一个助理agent ),)supervisor workflow.compile(checkpointercheckpointer, storestore)步骤六系统运行第一次交互 查询北京火锅店config {configurable: {thread_id: 1, user_id: user_123}} print(--- 第一次交互 查询北京火锅店---)async for chunk in supervisor.astream( {messages: [(user, 北京最出名的老北京火锅是哪家)]}, config): # 打印每个数据块的内容 for key, value in chunk.items(): print(fNode: {key}) if value:print( value:)print(value) print(----)输出北京最出名的老北京火锅有很多家其中比较有名的包括\n\n1. **东来顺**东来顺是北京最著名的老北京涮羊肉火锅店之一历史悠久以其独特的涮羊肉和秘制的调料而闻名。\n\n2. **南门涮肉**南门涮肉也是一家老字号的火锅店以其传统的涮羊肉和地道的北京风味而受到欢迎。\n\n3. **老北京涮肉馆**这是一家专注于传统老北京涮肉的火锅店以其正宗的口味和优质的服务而受到食客的喜爱。\n\n4. **聚宝源**聚宝源是一家以涮羊肉为主的火锅店以其新鲜的食材和独特的调料而受到欢迎。\n\n这些火锅店都有各自的特色和忠实的顾客群体您可以根据自己的口味和需求选择合适的火锅店。第二次交互根据上一步推荐的火锅店查询酒店print(--- 第二次交互 根据上一步推荐的火锅店查询酒店---)async for chunk in supervisor.astream( {messages: [(user, 那第一个推荐的火锅店附近有哪些酒店呀)]}, config): # 打印每个数据块的内容 for key, value in chunk.items(): print(fNode: {key}) if value:print( value:)print(value) print(----)输出东来顺火锅店位于北京市东城区东华门大街。以下是东来顺火锅店附近的一些酒店推荐\n\n1. **北京王府井希尔顿酒店**这是一家豪华酒店距离东来顺火锅店步行约10分钟提供高品质的住宿和服务。\n\n2. **北京东方君悦大酒店**这家五星级酒店位于王府井大街距离东来顺火锅店步行约15分钟设施齐全服务优质。\n\n3. **北京华尔道夫酒店**这是一家高端酒店距离东来顺火锅店步行约10分钟提供豪华的住宿体验和优质的服务。\n\n4. **北京诺富特和平宾馆**这家四星级酒店位于王府井大街距离东来顺火锅店步行约15分钟性价比较高适合商务和休闲旅行。\n\n5. **北京天伦王朝酒店**这家四星级酒店距离东来顺火锅店步行约10分钟提供舒适的住宿环境和便捷的交通。\n\n这些酒店都位于东来顺火锅店附近您可以根据自己的需求和预算选择合适的酒店。可以看到supervisor记得上文中提到过的第一个推荐的火锅店这是短期记忆的典型应用。第三次交互预定酒店print(--- 第三次交互 预定酒店---) async for chunk in supervisor.astream( {messages: [(user, 帮我预定北京王府井希尔顿酒店酒店预定日期2025-11-13到2025-11-14入住人数1)]}, config): for key, value in chunk.items(): print(fNode: {key}) if value:print( value:)print(value) print(----)输出已成功为您预订了北京王府井希尔顿酒店入住日期为2025年11月13日至2025年11月14日入住人数为1人。祝您旅途愉快成功调用预定酒店助手并完成酒店预定。第四次交互管理员查询预定信息print(--- 第四次交互 管理员查询预定信息---)config {configurable: {thread_id: 2, user_id: user_123}} # 更换管理员操作线程interrupt_data Noneinterrupt_input Noneprint(--- 第一次运行将会触发中断 ---)async for chunk in supervisor.astream( {messages: [(user, 查询用户预定酒店信息)]}, config,): for key, value in chunk.items(): print(fNode: {key}) if value:print( value:)print(value) if key __interrupt__:print(\n 图已成功中断)interrupt_data value[0] print(f中断信息: {interrupt_data.value})break if interrupt_data: break print(----)if interrupt_data: # 模拟管理员输入正确的密码 interrupt_input Command(resumeadmin_123) # 如果想测试错误的密码可以使用下面这行#interrupt_input Command(resumewrong_password) print(f\n--- 接收到中断输入 {interrupt_input}继续执行图 ---) # 恢复图的执行 async for chunk in supervisor.astream( interrupt_input, config, ): for key, value in chunk.items():print(fNode: {key})if value: print( value:) print(value) print(----)第一次运行触发中断并输出请输入管理员id如需退出查询请输入exit接着程序模拟管理员输入正确的ID后继续执行图booking_info_assistant查询到长期记忆并输出已找到预订信息[{user_id: user_123, hotel_name: 北京王府井希尔顿酒店, date: 2025-11-13到2025-11-14, num_guests: 1}]最终supervisor返回最终的结果用户预定的酒店信息如下\n- 用户ID: user_123\n- 酒店名称: 北京王府井希尔顿酒店\n- 预定日期: 2025-11-13到2025-11-14\n- 客人数: 1未来工作更智能的记忆管理策略在上述的示例预定酒店中我们仅将用户预定信息直接进行存储、检索但对于supervisor来说记住和不同用户的过往交互是非常重要的在后续的交互中才能针对不同的用户给出更合适的回应。让Agent能自主决定记忆的存储、遗忘、更新和检索优先级才能真正模拟人类的记忆过程。因此未来需要改进设计更智能的记忆管理策略。记忆驱动的多智能体系统架构选择在上述示例预定酒店中我们选择了Supervisor架构进行实现但这显然存在缺陷管理员系统不应该和用户系统使用同一个中央智能体当系统功能越来越完善时这样的设计会使得supervisor非常繁杂、且难以维护Supervisor架构更适合需要明确控制流程和集中决策的场景。融合记忆功能的Multi-Agent系统可以根据应用场景选择更合适的架构例如Hierarchical架构可以用于不同层级的记忆服务于不同目的个体、团队、全局的场景Custom架构预先定义好各个Agent的记忆走向构建更灵活的系统。AI时代未来的就业机会在哪里答案就藏在大模型的浪潮里。从ChatGPT、DeepSeek等日常工具到自然语言处理、计算机视觉、多模态等核心领域技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。掌握大模型技能就是把握高薪未来。那么普通人如何抓住大模型风口AI技术的普及对个人能力提出了新的要求在AI时代持续学习和适应新技术变得尤为重要。无论是企业还是个人都需要不断更新知识体系提升与AI协作的能力以适应不断变化的工作环境。因此这里给大家整理了一份《2025最新大模型全套学习资源》包括2025最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题等带你从零基础入门到精通快速掌握大模型技术由于篇幅有限有需要的小伙伴可以扫码获取1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。5. 大模型行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。为什么大家都在学AI大模型随着AI技术的发展企业对人才的需求从“单一技术”转向 “AI行业”双背景。企业对人才的需求从“单一技术”转向 “AI行业”双背景。金融AI、制造AI、医疗AI等跨界岗位薪资涨幅达30%-50%。同时很多人面临优化裁员近期科技巨头英特尔裁员2万人传统岗位不断缩减因此转行AI势在必行这些资料有用吗这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。大模型全套学习资料已整理打包有需要的小伙伴可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】