01 LangGraph循环图永不结束的对话想和智能体持续聊天, 怎么办?把END节点去掉, 把图改成循环——START到human_node,human_node到llm_node,llm_node再回到human_node……如此往复,图里没有END:builder.add_edge(START, human_node)builder.add_edge(human_node, llm_node)builder.add_edge(llm_node, human_node) # 不再到 END,而是回到 human_node效果是:图根本不会停下来。实际对话是这样的:问我是谁——模型答还不知道你的具体信息(它确实不知道); 继续聊;告诉它我是xxx——它回复很高兴认识你;再问我是谁——这次它知道你是xxx了。原因就在于:messages信息在不断地做数据累加。人工输入的数据、大模型返回的数据,都在循环中不断追加进数组,所以最后输入给大模型的内容会越来越长。代码里在llm_node加了一行打印, 可以直接观察输入给大模型的消息数量(len(state[messages]))多轮对话的问题:单用户与多用户搭建多轮对话这件事,整个智能体其实比较偏工程, 跟大模型本身的关联性没有那么强。比如用graph搭一张图, 希望实现多轮对话——可以不断地跟它聊天。让图不结束——START到human_node(人开始输入),human_node到大模型(回复),大模型回复又回到human_node,两个节点之间来回跳:这样就完成了多轮对话。如果只有一个用户,这没有任何问题。但如果来了多个用户——A 用户、B 用户同时访问——就出现了一个非常重要的概念:对话管理。我们需要对 A、B 两个用户的数据做隔离,两个人的数据绝不能串在一起——B的问题带着A的聊天记录,这肯定是不行的。因此LangGraph提出了一个概念叫做memory(记忆)短期记忆。图是无状态的:为什么需要专门保存数据假设把图搭成链式: START → human_node → llm_node → END,一条链直接走完; 在外面套一个死循环, 让这张图不断地被启动、不断地被启动——理论上也能达到多轮对话的效果这样做的问题在于没有数据保存。图本身不会保存数据的: 输入一个字典进来,输出一个字典出去,数据在图里流完一趟之后,没有任何东西把中间过程留下来。这个图本身就是一个无状态图——不保存历史数据,你可以把它理解成一个很健忘的东西。所以需要专门设置一个东西来保存这份数据。这就是memory登场的原因。02 InMemorySaver短期记忆与用户隔离接入方式:checkpointerLangGraph自带了一个包, 提供InMemorySaver。作用是把所有对话数据全部存在memory里, 哪怕对话结束了也没有问题。接入方式分两步:from langgraph.checkpoint.memory import InMemorySaver # 对话数据存在内存里 memory InMemorySaver() # 构建图的时候,编译时加 checkpointermemory # 编译时就会自动把这个记忆体放进图里 graph builder.compile(checkpointermemory)第一步,创建InMemorySaver();第二步,在builder.compile()的时候传入checkpointermemory,编译时框架就会自己把记忆体装进图里。除此之外,在invoke的时候还需要加一个东西:thread_id。thread_id:隔离数据的标识graph.invoke({}, config{configurable:{thread_id: user_id}})thread_id(线程 ID)这个字段是用来隔离用户数据的。可以这样理解 memory 的结构:memory 是一个存储空间,里面有很多格子,每个格子对应一个 thread_id:memory 存储空间的格子各自存放的内容thread_id 123用户 123 的全部对话记录thread_id 456用户 456 的全部对话记录不同的 thread_id,对应不同的内存空间。所以我们用 thread_id 作为隔离数据的标识。每次用户输完、图每次启动的时候,都把这个 thread_id 带上,框架就可以把该用户之前的记忆全部唤醒。踩坑:挂了checkpointer后,每次invoke都必须带thread_id第一轮输入user_id 100、对话你好一切正常;第二轮刚输完 user_id 200,程序立刻抛出:ValueError: Checkpointer requires one or more of the following configurable keys: thread_id, checkpoint_ns, checkpoint_id原因对照代码就清楚了:第一次invoke带了config,而循环里的正确写法被注释掉,实际执行的是裸调用:print (你可以和机器人聊天了)#对话数据存在内存里memoryInMemorySaver()graph builder.compile(checkpointermemory)user_id input(请输入user_id::: )#thread_id 用来隔离用户的graph.invoke({},config{configurable: {thread_id:user_id}})while True:user_id input(请输入user_id: )if user_id退出:breakgraph.invoke({},config{configurable: {thread_id:user_id}})# graph.invoke({}) #直接invoke 没有数据的保存, 会报错规则就一句话:只要编译时挂了checkpointer,之后每一次invoke都必须在config里带thread_id。为什么?挂上checkpointer之后,invoke的语义从跑一遍图变成了在某个会话(thread)上续跑:图启动时先按thread_id加载该用户的历史 state,跑完再把新state存回同一个key下。没有 thread_id,checkpointer就不知道该读写哪份存档。注意报错发生在图的启动阶段,所以现象是刚输完user_id、还没见到用户:提示就崩了。报错信息里的三个key,就是checkpointer定位数据的三个维度:key定位什么thread_id哪个会话(哪个用户)checkpoint_ns哪个子图的命名空间checkpoint_id哪个具体的存档点修复方法就是恢复被注释的那一行,循环里每次 invoke 都带上 config:whileTrue: user_id input(请输入user_id: ) if user_id 退出: break graph.invoke({}, config{configurable:{thread_id: user_id}})一次invoke的生命周期:图内流转与while循环再理清一个容易混淆的点:一次 invoke 图从 START 完整走到 END 一遍(这张图就是 human_node → llm_node 两个节点),走完图就结束了;多轮对话是 while 循环一次一次重新发起 invoke 实现的。你可以和机器人聊天了 ← print 请输入user_id: 100 ← input() 的提示参数 用户: 你好 │human_node: input(用户: ) 你好呀!有什么可以帮你? │llm_node: 调LLM 并打印回复 |走到END,图结束,state存入checkpointer ┌ while True 循环 ┐ 请输入user_id: ... │input 用户: ... (回复) │循环内 invoke:又完整走一遍图 └ 输入退出才 break ┘所以这里有两个不同层面的循环,不要混在一起:层面由谁驱动走到哪图内部invoke 一次,节点依次执行START → human_node → llm_node →END(图结束了)程序层面while True不断再 invoke图每次都重新从 START 跑一遍各轮之间的记忆衔接,靠的是 checkpointer按thread_id存取state——messages字段经add_messagesreducer一轮一轮往上累加,于是第二轮的 LLM 能看到第一轮说过的话。这才是图可以结束,程序不能结束的完整含义。代码里找不到print, 却能看到请输入user_id: 、用户: 这些提示——它们来自input()的提示参数: Python会先把input(...)的参数打印到终端(相当于print(..., end)),然后阻塞等待键盘输入, 回车后才把那一行作为返回值交给变量。提示语之后看到的多余字符只是终端等待输入时的光标和已敲入尚未回车的字符,并不是程序打印的内容。运行演示:两个用户互不串扰程序启动后先模拟输入 user_id(正常情况下不需要用户自己输入,这里因为还没有前端,所以用键盘输入模拟)。演示过程:轮次输入 user_id对话内容模型表现1123我是谁不知道(该用户无历史)2123我是aaa记录下来3123我是谁知道你是aaa4456我是谁不知道(换了新用户,数据隔离)5456我是bbb记录下来6456我是谁知道你是bbb7123我是谁仍然知道你是aaa关键在最后一行:切到456聊了几轮之后,再切回123,模型依然记得123用户是aaa——因为它之前的所有信息都被记录下来了。通过user_id作为线程ID就把用户隔离开来了。反过来,如果有需求让多个用户共享同一份记忆怎么办?很简单:把他们的 thread_id设成一样,他们就共享同一块内存了。在企业里user_id不需要用户自己输入,都是前端直接传过来。这个user_id,一般来说可能是手机设备号、产品的账号等——总之都是前端自动获取、由前端传过来的。memory的两个功能与局限总结一下memory实现了两个功能:数据存储:哪怕图结束了,数据仍然放在内存里;数据隔离:通过线程ID对这堆数据做了隔离,让不同用户的数据各自单独存放。但它还有一个很大的问题:数据是放在内存里的。所以如果代码结束了——程序停了,再重新启动——它就不知道了。比如重新启动后再输入 123问我是谁,它就不知道你是谁了。一句话概括这个局限:图可以结束, 程序不能结束。当然memory还有很多其他类型。03 SqliteSaver外挂数据库的持久化存储既然内存里的数据程序一停就没了,那我们的数据除了存在内存里,还可以直接外挂数据库。LangGraph对此自带支持:SqliteSaver——一个类似于MySQL的数据库。接入后图不变,只把memory换成连数据库。但这里有个地方需要注意:前面InMemorySaver是同步的, 而这里必须是异步连接数据库。因为要考虑这样一个事实:写内存是瞬间就能写完的; 而写数据库本身是要花时间的。所以框架把数据库操作做成了异步的。这样一来,对话启动图的时候,前面就得加个a(用异步版本ainvoke), 再配合await等待:async def main():print (你可以和机器人聊天了)#对话数据存在数据库里#异步连数据库async with AsyncSqliteSaver.from_conn_string(checkpoints.db) as memory:graph builder.compile(checkpointermemory)user_id input(请输入user_id: )await graph.ainvoke({},config{configurable: {thread_id:user_id}})while True:user_id input(请输入user_id: )if user_id退出:breakawait graph.ainvoke({},config{configurable: {thread_id:user_id}})# #不能这样直接启异步函数# main()#asyncio.runasyncio.run(main())有几个Python语法要点:异步函数不能直接调用启动,需要用asyncio.run()把它起起来——这就是固定套路; 异步流程里所有的await,含义都是我要等到这个操作结束了才往下走。持久化演示:程序停了数据还在直连数据库运行,可以看到:目录下多了一个checkpoints.db文件——数据库已经创建出来了;对话之后文件又发生变化,说明开始往库里写数据。对话演示与之前一致:123问我是谁——不知道;告诉它我是aaa;再问——它知道你是aaa了。关键的一步来了:把程序直接停掉,再重新运行,输入 123 问我是谁——它仍然知道你提到过自己是aaa。也就是说,这个存储是持久化存储:代码跑完了、程序停了,数据依然存在,下次启动仍然能读回来。多用户与更多存储后端数据库版本一样支持多用户:456 告诉它我是bbb之后再问,它知道你是bbb;再切回123 问,它依然知道你是aaa——功能与memory版本一模一样,区别只在于:数据库是持久化存储,而memory是程序一停就没了。LangGraph的checkpointer不光支持SQL数据库,Redis也都支持。数据库只做存储不做理解这里要建立一个正确的认知:数据库只做存储,不做理解。数据库可以无限大,你想让它多大就多大;真正限制你的,是模型的 TOKEN 数量(上下文窗口)——对话的长度是被模型限制住的,而不是被你的数据库或内存限制住的。另外一个设计边界:这里实现的是平层管理(一个 thread_id 一份数据的平面结构)。如果你要做多层管理这类非常复杂的业务逻辑,得在外面自己去设计,LangGraph 的 checkpointer 不负责这件事。Python异步与协程: async、await与asyncio.run同步: 正常情况下,函数是一步一步走的——必须把第一步走完,再走第二步,走完再走第三步。异步: 如果第一步特别耗时间(比如调用大模型、访问数据库),第二步也很耗时间,我们就可以把它改成异步。做法是:第一步启动之后,不等它执行完,直接走第二步;第二步也设成异步,启动之后也不等,直接来到第三步——这时你会发现,第一步和第二步是在同时执行的。打个比方:我是一个leader,我跟张三说你去写一个 LangGraph 的代码,说完我直接去找李四说你给我做个 PPT,安排完直接进入第三步。我不需要等张三做完、李四做完才走到第三步——我让这两个人完全并行起来,这就是所谓的异步。在 Python里,所有异步都用几个关键词表达:关键词位置含义async def函数定义前声明这是一个异步函数,你不需要等我走完就可以往后面走await调用处我必须等这个异步函数走完了,再往下走asyncio.run()最外层异步函数不能直接调用启动,要用它来起什么时候需要await?比如第三步必须用到第一、二步处理的结果,那在第三步处就要加await,等前两步全部走完再继续。这个概念在Python里叫做协程(coroutine)。架构澄清:大模型、LangGraph与数据库的关系第一,大模型是一个无状态的工具。它不存储任何历史信息——字符串进、字符串出,本身不保存任何中间信息,是个完全无状态的东西。LangGraph 调用大模型,大模型返回数据给LangGraph,仅此而已。那多轮对话能记住之前的东西是怎么实现的?是在传数据的时候,把历史信息一起传给大模型——也就是说,历史信息是在大模型之外保存的,每次调用时送进去。第二,LangGraph本身是半无状态的。如果图一直不结束(比如循环图 human → llm → human 永远转), 历史数据就一直存在state 里,一直可以发给大模型——这种情况其实一直有历史记录。但图一结束,历史记录就丢了: 链式图(START → human → llm → END)两个节点走完图就结束, 历史数据没有地方存放;程序一停,同样全没了。第三,为了防止这种丢失,才有了两级存储。存储方式特点不加 checkpointer图结束/程序停止,历史记录即丢失InMemorySaver(短期记忆)只要程序不停,历史记录一直在(存在内存中);程序停了,数据也就没了SqliteSaver(持久化存储)只要数据不误删,哪怕程序停了,数据依然存在数据库跟大模型本身不做任何交互。容量问题: 内存方案能存多少, 取决于内存有多大; 数据库能存多少, 取决于数据库支持多大——而对话这种纯文本数据, 基本上可以理解成用不完。上下文截断:记多长就传多长还有一个隐患要处理:大模型本身有上下文限制。如果一个人的聊天记录非常长,超过了大模型的 TOKEN 数量,是会报错的。所以如果历史记录非常长,数据进来的时候需要做一个截断。做法很简单,比如:state[messages][-20:]# 只取最近20轮聊天进大模型只把最近20轮的聊天记录送进大模型,而不是把所有历史数据都塞进去。每次做多轮对话,把多长的历史传进来,大模型的上下文就有多长——你希望它记多长,你就传多长。以上就是checkpointer的功能:它让你可以把节点中间信息全部存起来。04 Function Call让大模型调用工具工具就是函数:注释是写给大模型看的Function Call——现在一般叫Tool Call(工具调用):怎么样让大模型去调用工具?举两个工具的例子:一个加法、一个乘法。工具本身比较简单——所谓的工具,指的就是一个函数,这里可以是任何函数。这部分和讲 LangChain 时比较类似,因为 LangGraph 本身就是 LangChain 的升级版本。#大模型 使用外部的工具tooldef add(a: int, b: int) - int:Adds a and b.print (f调用工具计算{a}{b})return a btooldef multiply(a: int, b: int) - int:Multiplies a and b.print (f调用工具计算{a}*{b})return a * b函数上的注释(docstring)非常重要,不能乱写。很多人编程习惯不太好——注释不写,或者随便写两句——在别处也许无所谓,但在这里不行。因为这个注释是要给大模型自己去理解的东西:大模型靠它来判断这个工具是干什么的、什么时候该调用它。工具的注释一定要写得很清楚。另外,在工具函数里加了一行print(调用工具计算...),这是为了在演示时确认结果是通过调用工具算出来的,而不是大模型自己心算出来的。准备工作:工具数组、名称映射与绑定工具接着做三件准备工作:# 1. 把工具放在一个数组里:这里要调用的两个函数tools [add, multiply]# 2. 工具名 → 工具 的映射: 字符串 add 就代表要调 add 这个函数tools_by_name {tool.name: tool for tool in tools}# 3. 大模型绑定工具llm ChatOpenAI(model qwen3.7-plus,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,api_keyos.getenv(api_key))#大模型获取了工具的描述信息但是还不具备直接执行工具的能力llm_with_tools llm.bind_tools(tools)工具数组:把要暴露的工具放进一个数组;名称映射:做一个工具名到工具的映射字典——大模型返回的是字符串名称,通过这个映射找到真正要执行的函数;绑定工具:llm.bind_tools(tools)。注意这里的含义:大模型在使用的时候,会把这些工具的描述信息一并输入给它。绑定完之后,工具的名称、参数说明就都随着每次调用传给了大模型。但要记住一点:大模型获取了工具的描述信息,并不代表它具备了直接执行工具的能力——它只能决定调用哪个、传什么参数,真正的执行还是要靠我们的代码。构建工具调用图:大模型节点、工具节点与条件边图里有两个节点:def llm_node(state:MessagesState)-MessagesState:responsellm_with_tools.invoke(state[messages])return {messages: response}def execute_tools_node(state: MessagesState) - MessagesState:# 执行所有待处理的工具调用results[]messages state[messages]last_message messages[-1]for tool_call in last_message.tool_calls:tool_result tools_by_name[tool_call[name]].invoke(tool_call[args])results.append(ToolMessage(contenttool_result,tool_call_idtool_call[id]))return {messages:results}llm_node: 大模型节点,负责对话;工具节点: 负责执行工具。它是LangGraph里一个比较特殊的节点——最原生的写法是自己手写execute_tools_node; LangGraph还提供了封装好的ToolNode, 可以直接拿来用,把执行工具的步骤都封装起来了,代码更简单。加边的时候,关键是llm_node后面这一条:builder StateGraph(MessagesState)builder.add_node(llm_node, llm_node) # START到大模型节点builder.add_node(execute_tools_node, execute_tools_node)builder.add_edge(START, llm_node)# 工具节点执行完回到大模型节点builder.add_edge(execute_tools_node, llm_node)# 条件边:由should_continue 决定走向builder.add_conditional_edges(llm_node,should_continue,path_map{execute_tools: execute_tools_node,END: END})条件边(conditional edge)是这里最关键的概念, 而条件边里最关键的是这个路由函数:#条件边def should_continue(state:MessagesState):messages state[messages]last_message messages[-1]print (大模型的输出,last_message)#判断大模型是否调用了工具if not last_message.tool_calls:return END# Otherwise if there is, we continueelse:return execute_tools它的工作方式:大模型输出完成后,输出信息会进入条件边的这个函数里。函数检查最后一条消息(大模型的返回)是否带有tool_calls字段:如果是正常对话(比如你是谁),大模型的返回里没有tool_calls字段 → 返回 END → 直接结束。你问你是谁,它回答是谁,这个答案就是最终结果;如果用户输入刚好能触发工具调用(比如帮我算个加法),大模型的返回里就会有 tool_calls 字段(包含函数名称和函数参数)→ 返回 execute_tools → 走到工具节点,执行工具。最后那个path_map是对返回值的导向:函数返回END就走到结束符,返回 execute_tools 就走到工具节点——它描述返回值 → 去哪个节点的对应关系,决定图到底往哪走。运行演示:两种输入两种路径跑两个对比案例:案例一:问6×3 是多少。运行后会先打印出调用工具计算6*3——说明这个 18 是调用工具算出来的,不是大模型自己算出来的。案例二:问你是谁。它不会调用任何工具——条件边函数检查发现没有 tool_calls 字段,直接返回END,大模型答完就走完了。同一个图,两种输入,走了两条完全不同的路径——走哪条路,是大模型自己判断的,不是人指定的。完整流转分析:一次工具调用的全过程对着图把6×3是多少的完整流转走一遍。代码里把每一步都打印出来了:第一步: 用户输入进入 llm_node,大模型完成一次输入输出。注意观察大模型这次输出的是什么:content是空的!但tool_calls字段里有内容:函数名称(multiply)和函数参数(a6, b3)。也就是说,大模型这一轮并没有说话,它只表达了我要调用这个工具。第二步: 输出进入条件边函数 should_continue。大模型连的边就是条件边,所以大模型输出完之后,立刻调用条件边先做判断。函数发现最后一条消息带 tool_calls 字段 → 返回 execute_tools → 图走到工具节点。第三步: 工具节点执行工具。把工具的名称从 tool_calls 里取出来——注意一个细节:tool_calls 的数据形式虽然是数组,但实际上几乎都只有一个元素。理论上它支持多个元素,但就工作实践和查资料的情况来看,很少看到大模型一次并行调用多个工具,基本还是一个比较串行的过程。取出名称 multiply 后,通过 tools_by_name 的映射找到乘法函数,以a6, b3为参数 invoke 执行。第四步: 封装结果并返回。函数执行完,把结果(18)用ToolMessage封装(这是 LangChain 里的内容),返回到 state 里。注意:工具返回的就是函数的返回值本身,没有任何语言包装——你的函数 return 什么,它就是什么,直接返回来。第五步: 回到大模型节点,再调一次大模型。看工具节点连的是谁?连的是大模型节点。所以工具结果又会送回大模型——这一轮,是大模型对工具调用的结果进行总结:6 乘以 3 等于 18。第六步: 再走条件边,结束。大模型总结完再经过条件边, 这次没有工具调用, 返回END, 程序结束。整体结果(逐步打印):用户问6×3 是多少→ 大模型返回content为空,但有要调用工具的信息;工具返回18(函数返回值,无语言包装);第三轮把前面所有信息都给大模型,大模型总结:6乘以3等于18。这就是整套流程。例子虽然简单,但它恰恰体现了智能体和工作流的区别——理解这一步,是理解智能体非常关键的步骤。智能体与工作流的区别:控制权交给谁智能体和工作流的判断标准就一条:这张图的实际执行路径, 事先能不能预知。对上图, 它具体真实的路径:可能是START → LLM → END(直接答完结束);可能是START → LLM → 工具 → LLM → END;也可能是START → LLM → 工具 → LLM → 工具 → LLM……来回转好几圈。事先是不知道的,完全取决于大模型自身的判断结果。也就是说,我把判断权全部交给了大模型——图是我画的,但在这张图上到底怎么走,是大模型控制的,而不是我这个人控制的。这就叫做Agent(智能体)。打个比方:玩网游(MMORPG)时, 游戏设计者只设计了基本的游戏规则——你在里头打什么怪、跟谁组队、捡什么道具、买什么东西,设计者控制不了,完全由玩家自己决定。而横版过关那种很死板的RPG,第一关干什么、第二关干什么都是定死的,玩家没有什么自主选择权。智能体就是前者:把控制权交给了作为玩家的大模型。反过来看工作流的例子:RAG(召回 → 问答 → 答案)每一步的流程都是事先规定好的。它们在执行过程中也用大模型,但那是把大模型当工具、当函数用, 并没有把流程怎么走的权利交给大模型: 走完召回必然走问答, 完全由设计者控制。也正因为路径确定,它非常好debug。什么是智能体, 如果只回答智能体是个Function Call是没有任何亮点的——工作流一样能做工具调用。真正能说明问题的, 恰恰是例子中体现的东西:整个流程的自主权全部交给大模型。对照之前的RAG等例子: 大模型在其中只是作为一个语言工具、一个插件, 真正主导流程的还是设计者; 而对智能体来说,大模型起到的是决策作用。这就是二者的分水岭。两个框架的关系两个框架的关系:LangGraph是LangChain的全面升级版。LangChain 只提供了很基础的功能, 相当于给LangGraph提供了一些基本的插件(比如各种Message类型、工具调用的封装); 而LangGraph在这之上把图画起来,玩得更活更花。Function Call的实现有几种方法:最底层的实现: 自己写工具、自己写执行逻辑、自己写返回——怎么执行、怎么返回全部手写;ToolNode封装: 把执行工具的步骤封装起来直接搞一个ToolNode节点, add_node的时候把工具交给它, 执行工具全部封装起来,只需要写一个判断条件(条件边);create_react_agent: 高度封装的一行写法, 一句话图就出来了。内部等价于llm节点工具节点条件边那一整套。Function Call体会手写 → ToolNode → create_react_agent三层封装的演进。这三层体现的是一个通用规律:封装得越狠, 用起来就越方便——create_react_agent一句话直接出来;但越想做细微的修改就越困难。反过来,代码越偏向底层, 做定制化修改的时候就方便很多。 
长河君 · 2026/10/14 2:40:39 · 阅读 1.2k LangGraph多轮对话：用thread_id实现用户隔离
01 LangGraph循环图永不结束的对话想和智能体持续聊天, 怎么办?把END节点去掉, 把图改成循环——START到human_node,human_node到llm_node,llm_node再回到human_node……如此往复,图里没有END:builder.add_edge(START, human_node)builder.add_edge(human_node, llm_node)builder.add_edge(llm_node, human_node) # 不再到 END,而是回到 human_node效果是:图根本不会停下来。实际对话是这样的:问我是谁——模型答还不知道你的具体信息(它确实不知道); 继续聊;告诉它我是xxx——它回复很高兴认识你;再问我是谁——这次它知道你是xxx了。原因就在于:messages信息在不断地做数据累加。人工输入的数据、大模型返回的数据,都在循环中不断追加进数组,所以最后输入给大模型的内容会越来越长。代码里在llm_node加了一行打印, 可以直接观察输入给大模型的消息数量(len(state[messages]))多轮对话的问题:单用户与多用户搭建多轮对话这件事,整个智能体其实比较偏工程, 跟大模型本身的关联性没有那么强。比如用graph搭一张图, 希望实现多轮对话——可以不断地跟它聊天。让图不结束——START到human_node(人开始输入),human_node到大模型(回复),大模型回复又回到human_node,两个节点之间来回跳:这样就完成了多轮对话。如果只有一个用户,这没有任何问题。但如果来了多个用户——A 用户、B 用户同时访问——就出现了一个非常重要的概念:对话管理。我们需要对 A、B 两个用户的数据做隔离,两个人的数据绝不能串在一起——B的问题带着A的聊天记录,这肯定是不行的。因此LangGraph提出了一个概念叫做memory(记忆)短期记忆。图是无状态的:为什么需要专门保存数据假设把图搭成链式: START → human_node → llm_node → END,一条链直接走完; 在外面套一个死循环, 让这张图不断地被启动、不断地被启动——理论上也能达到多轮对话的效果这样做的问题在于没有数据保存。图本身不会保存数据的: 输入一个字典进来,输出一个字典出去,数据在图里流完一趟之后,没有任何东西把中间过程留下来。这个图本身就是一个无状态图——不保存历史数据,你可以把它理解成一个很健忘的东西。所以需要专门设置一个东西来保存这份数据。这就是memory登场的原因。02 InMemorySaver短期记忆与用户隔离接入方式:checkpointerLangGraph自带了一个包, 提供InMemorySaver。作用是把所有对话数据全部存在memory里, 哪怕对话结束了也没有问题。接入方式分两步:from langgraph.checkpoint.memory import InMemorySaver # 对话数据存在内存里 memory InMemorySaver() # 构建图的时候,编译时加 checkpointermemory # 编译时就会自动把这个记忆体放进图里 graph builder.compile(checkpointermemory)第一步,创建InMemorySaver();第二步,在builder.compile()的时候传入checkpointermemory,编译时框架就会自己把记忆体装进图里。除此之外,在invoke的时候还需要加一个东西:thread_id。thread_id:隔离数据的标识graph.invoke({}, config{configurable:{thread_id: user_id}})thread_id(线程 ID)这个字段是用来隔离用户数据的。可以这样理解 memory 的结构:memory 是一个存储空间,里面有很多格子,每个格子对应一个 thread_id:memory 存储空间的格子各自存放的内容thread_id 123用户 123 的全部对话记录thread_id 456用户 456 的全部对话记录不同的 thread_id,对应不同的内存空间。所以我们用 thread_id 作为隔离数据的标识。每次用户输完、图每次启动的时候,都把这个 thread_id 带上,框架就可以把该用户之前的记忆全部唤醒。踩坑:挂了checkpointer后,每次invoke都必须带thread_id第一轮输入user_id 100、对话你好一切正常;第二轮刚输完 user_id 200,程序立刻抛出:ValueError: Checkpointer requires one or more of the following configurable keys: thread_id, checkpoint_ns, checkpoint_id原因对照代码就清楚了:第一次invoke带了config,而循环里的正确写法被注释掉,实际执行的是裸调用:print (你可以和机器人聊天了)#对话数据存在内存里memoryInMemorySaver()graph builder.compile(checkpointermemory)user_id input(请输入user_id::: )#thread_id 用来隔离用户的graph.invoke({},config{configurable: {thread_id:user_id}})while True:user_id input(请输入user_id: )if user_id退出:breakgraph.invoke({},config{configurable: {thread_id:user_id}})# graph.invoke({}) #直接invoke 没有数据的保存, 会报错规则就一句话:只要编译时挂了checkpointer,之后每一次invoke都必须在config里带thread_id。为什么?挂上checkpointer之后,invoke的语义从跑一遍图变成了在某个会话(thread)上续跑:图启动时先按thread_id加载该用户的历史 state,跑完再把新state存回同一个key下。没有 thread_id,checkpointer就不知道该读写哪份存档。注意报错发生在图的启动阶段,所以现象是刚输完user_id、还没见到用户:提示就崩了。报错信息里的三个key,就是checkpointer定位数据的三个维度:key定位什么thread_id哪个会话(哪个用户)checkpoint_ns哪个子图的命名空间checkpoint_id哪个具体的存档点修复方法就是恢复被注释的那一行,循环里每次 invoke 都带上 config:whileTrue: user_id input(请输入user_id: ) if user_id 退出: break graph.invoke({}, config{configurable:{thread_id: user_id}})一次invoke的生命周期:图内流转与while循环再理清一个容易混淆的点:一次 invoke 图从 START 完整走到 END 一遍(这张图就是 human_node → llm_node 两个节点),走完图就结束了;多轮对话是 while 循环一次一次重新发起 invoke 实现的。你可以和机器人聊天了 ← print 请输入user_id: 100 ← input() 的提示参数 用户: 你好 │human_node: input(用户: ) 你好呀!有什么可以帮你? │llm_node: 调LLM 并打印回复 |走到END,图结束,state存入checkpointer ┌ while True 循环 ┐ 请输入user_id: ... │input 用户: ... (回复) │循环内 invoke:又完整走一遍图 └ 输入退出才 break ┘所以这里有两个不同层面的循环,不要混在一起:层面由谁驱动走到哪图内部invoke 一次,节点依次执行START → human_node → llm_node →END(图结束了)程序层面while True不断再 invoke图每次都重新从 START 跑一遍各轮之间的记忆衔接,靠的是 checkpointer按thread_id存取state——messages字段经add_messagesreducer一轮一轮往上累加,于是第二轮的 LLM 能看到第一轮说过的话。这才是图可以结束,程序不能结束的完整含义。代码里找不到print, 却能看到请输入user_id: 、用户: 这些提示——它们来自input()的提示参数: Python会先把input(...)的参数打印到终端(相当于print(..., end)),然后阻塞等待键盘输入, 回车后才把那一行作为返回值交给变量。提示语之后看到的多余字符只是终端等待输入时的光标和已敲入尚未回车的字符,并不是程序打印的内容。运行演示:两个用户互不串扰程序启动后先模拟输入 user_id(正常情况下不需要用户自己输入,这里因为还没有前端,所以用键盘输入模拟)。演示过程:轮次输入 user_id对话内容模型表现1123我是谁不知道(该用户无历史)2123我是aaa记录下来3123我是谁知道你是aaa4456我是谁不知道(换了新用户,数据隔离)5456我是bbb记录下来6456我是谁知道你是bbb7123我是谁仍然知道你是aaa关键在最后一行:切到456聊了几轮之后,再切回123,模型依然记得123用户是aaa——因为它之前的所有信息都被记录下来了。通过user_id作为线程ID就把用户隔离开来了。反过来,如果有需求让多个用户共享同一份记忆怎么办?很简单:把他们的 thread_id设成一样,他们就共享同一块内存了。在企业里user_id不需要用户自己输入,都是前端直接传过来。这个user_id,一般来说可能是手机设备号、产品的账号等——总之都是前端自动获取、由前端传过来的。memory的两个功能与局限总结一下memory实现了两个功能:数据存储:哪怕图结束了,数据仍然放在内存里;数据隔离:通过线程ID对这堆数据做了隔离,让不同用户的数据各自单独存放。但它还有一个很大的问题:数据是放在内存里的。所以如果代码结束了——程序停了,再重新启动——它就不知道了。比如重新启动后再输入 123问我是谁,它就不知道你是谁了。一句话概括这个局限:图可以结束, 程序不能结束。当然memory还有很多其他类型。03 SqliteSaver外挂数据库的持久化存储既然内存里的数据程序一停就没了,那我们的数据除了存在内存里,还可以直接外挂数据库。LangGraph对此自带支持:SqliteSaver——一个类似于MySQL的数据库。接入后图不变,只把memory换成连数据库。但这里有个地方需要注意:前面InMemorySaver是同步的, 而这里必须是异步连接数据库。因为要考虑这样一个事实:写内存是瞬间就能写完的; 而写数据库本身是要花时间的。所以框架把数据库操作做成了异步的。这样一来,对话启动图的时候,前面就得加个a(用异步版本ainvoke), 再配合await等待:async def main():print (你可以和机器人聊天了)#对话数据存在数据库里#异步连数据库async with AsyncSqliteSaver.from_conn_string(checkpoints.db) as memory:graph builder.compile(checkpointermemory)user_id input(请输入user_id: )await graph.ainvoke({},config{configurable: {thread_id:user_id}})while True:user_id input(请输入user_id: )if user_id退出:breakawait graph.ainvoke({},config{configurable: {thread_id:user_id}})# #不能这样直接启异步函数# main()#asyncio.runasyncio.run(main())有几个Python语法要点:异步函数不能直接调用启动,需要用asyncio.run()把它起起来——这就是固定套路; 异步流程里所有的await,含义都是我要等到这个操作结束了才往下走。持久化演示:程序停了数据还在直连数据库运行,可以看到:目录下多了一个checkpoints.db文件——数据库已经创建出来了;对话之后文件又发生变化,说明开始往库里写数据。对话演示与之前一致:123问我是谁——不知道;告诉它我是aaa;再问——它知道你是aaa了。关键的一步来了:把程序直接停掉,再重新运行,输入 123 问我是谁——它仍然知道你提到过自己是aaa。也就是说,这个存储是持久化存储:代码跑完了、程序停了,数据依然存在,下次启动仍然能读回来。多用户与更多存储后端数据库版本一样支持多用户:456 告诉它我是bbb之后再问,它知道你是bbb;再切回123 问,它依然知道你是aaa——功能与memory版本一模一样,区别只在于:数据库是持久化存储,而memory是程序一停就没了。LangGraph的checkpointer不光支持SQL数据库,Redis也都支持。数据库只做存储不做理解这里要建立一个正确的认知:数据库只做存储,不做理解。数据库可以无限大,你想让它多大就多大;真正限制你的,是模型的 TOKEN 数量(上下文窗口)——对话的长度是被模型限制住的,而不是被你的数据库或内存限制住的。另外一个设计边界:这里实现的是平层管理(一个 thread_id 一份数据的平面结构)。如果你要做多层管理这类非常复杂的业务逻辑,得在外面自己去设计,LangGraph 的 checkpointer 不负责这件事。Python异步与协程: async、await与asyncio.run同步: 正常情况下,函数是一步一步走的——必须把第一步走完,再走第二步,走完再走第三步。异步: 如果第一步特别耗时间(比如调用大模型、访问数据库),第二步也很耗时间,我们就可以把它改成异步。做法是:第一步启动之后,不等它执行完,直接走第二步;第二步也设成异步,启动之后也不等,直接来到第三步——这时你会发现,第一步和第二步是在同时执行的。打个比方:我是一个leader,我跟张三说你去写一个 LangGraph 的代码,说完我直接去找李四说你给我做个 PPT,安排完直接进入第三步。我不需要等张三做完、李四做完才走到第三步——我让这两个人完全并行起来,这就是所谓的异步。在 Python里,所有异步都用几个关键词表达:关键词位置含义async def函数定义前声明这是一个异步函数,你不需要等我走完就可以往后面走await调用处我必须等这个异步函数走完了,再往下走asyncio.run()最外层异步函数不能直接调用启动,要用它来起什么时候需要await?比如第三步必须用到第一、二步处理的结果,那在第三步处就要加await,等前两步全部走完再继续。这个概念在Python里叫做协程(coroutine)。架构澄清:大模型、LangGraph与数据库的关系第一,大模型是一个无状态的工具。它不存储任何历史信息——字符串进、字符串出,本身不保存任何中间信息,是个完全无状态的东西。LangGraph 调用大模型,大模型返回数据给LangGraph,仅此而已。那多轮对话能记住之前的东西是怎么实现的?是在传数据的时候,把历史信息一起传给大模型——也就是说,历史信息是在大模型之外保存的,每次调用时送进去。第二,LangGraph本身是半无状态的。如果图一直不结束(比如循环图 human → llm → human 永远转), 历史数据就一直存在state 里,一直可以发给大模型——这种情况其实一直有历史记录。但图一结束,历史记录就丢了: 链式图(START → human → llm → END)两个节点走完图就结束, 历史数据没有地方存放;程序一停,同样全没了。第三,为了防止这种丢失,才有了两级存储。存储方式特点不加 checkpointer图结束/程序停止,历史记录即丢失InMemorySaver(短期记忆)只要程序不停,历史记录一直在(存在内存中);程序停了,数据也就没了SqliteSaver(持久化存储)只要数据不误删,哪怕程序停了,数据依然存在数据库跟大模型本身不做任何交互。容量问题: 内存方案能存多少, 取决于内存有多大; 数据库能存多少, 取决于数据库支持多大——而对话这种纯文本数据, 基本上可以理解成用不完。上下文截断:记多长就传多长还有一个隐患要处理:大模型本身有上下文限制。如果一个人的聊天记录非常长,超过了大模型的 TOKEN 数量,是会报错的。所以如果历史记录非常长,数据进来的时候需要做一个截断。做法很简单,比如:state[messages][-20:]# 只取最近20轮聊天进大模型只把最近20轮的聊天记录送进大模型,而不是把所有历史数据都塞进去。每次做多轮对话,把多长的历史传进来,大模型的上下文就有多长——你希望它记多长,你就传多长。以上就是checkpointer的功能:它让你可以把节点中间信息全部存起来。04 Function Call让大模型调用工具工具就是函数:注释是写给大模型看的Function Call——现在一般叫Tool Call(工具调用):怎么样让大模型去调用工具?举两个工具的例子:一个加法、一个乘法。工具本身比较简单——所谓的工具,指的就是一个函数,这里可以是任何函数。这部分和讲 LangChain 时比较类似,因为 LangGraph 本身就是 LangChain 的升级版本。#大模型 使用外部的工具tooldef add(a: int, b: int) - int:Adds a and b.print (f调用工具计算{a}{b})return a btooldef multiply(a: int, b: int) - int:Multiplies a and b.print (f调用工具计算{a}*{b})return a * b函数上的注释(docstring)非常重要,不能乱写。很多人编程习惯不太好——注释不写,或者随便写两句——在别处也许无所谓,但在这里不行。因为这个注释是要给大模型自己去理解的东西:大模型靠它来判断这个工具是干什么的、什么时候该调用它。工具的注释一定要写得很清楚。另外,在工具函数里加了一行print(调用工具计算...),这是为了在演示时确认结果是通过调用工具算出来的,而不是大模型自己心算出来的。准备工作:工具数组、名称映射与绑定工具接着做三件准备工作:# 1. 把工具放在一个数组里:这里要调用的两个函数tools [add, multiply]# 2. 工具名 → 工具 的映射: 字符串 add 就代表要调 add 这个函数tools_by_name {tool.name: tool for tool in tools}# 3. 大模型绑定工具llm ChatOpenAI(model qwen3.7-plus,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,api_keyos.getenv(api_key))#大模型获取了工具的描述信息但是还不具备直接执行工具的能力llm_with_tools llm.bind_tools(tools)工具数组:把要暴露的工具放进一个数组;名称映射:做一个工具名到工具的映射字典——大模型返回的是字符串名称,通过这个映射找到真正要执行的函数;绑定工具:llm.bind_tools(tools)。注意这里的含义:大模型在使用的时候,会把这些工具的描述信息一并输入给它。绑定完之后,工具的名称、参数说明就都随着每次调用传给了大模型。但要记住一点:大模型获取了工具的描述信息,并不代表它具备了直接执行工具的能力——它只能决定调用哪个、传什么参数,真正的执行还是要靠我们的代码。构建工具调用图:大模型节点、工具节点与条件边图里有两个节点:def llm_node(state:MessagesState)-MessagesState:responsellm_with_tools.invoke(state[messages])return {messages: response}def execute_tools_node(state: MessagesState) - MessagesState:# 执行所有待处理的工具调用results[]messages state[messages]last_message messages[-1]for tool_call in last_message.tool_calls:tool_result tools_by_name[tool_call[name]].invoke(tool_call[args])results.append(ToolMessage(contenttool_result,tool_call_idtool_call[id]))return {messages:results}llm_node: 大模型节点,负责对话;工具节点: 负责执行工具。它是LangGraph里一个比较特殊的节点——最原生的写法是自己手写execute_tools_node; LangGraph还提供了封装好的ToolNode, 可以直接拿来用,把执行工具的步骤都封装起来了,代码更简单。加边的时候,关键是llm_node后面这一条:builder StateGraph(MessagesState)builder.add_node(llm_node, llm_node) # START到大模型节点builder.add_node(execute_tools_node, execute_tools_node)builder.add_edge(START, llm_node)# 工具节点执行完回到大模型节点builder.add_edge(execute_tools_node, llm_node)# 条件边:由should_continue 决定走向builder.add_conditional_edges(llm_node,should_continue,path_map{execute_tools: execute_tools_node,END: END})条件边(conditional edge)是这里最关键的概念, 而条件边里最关键的是这个路由函数:#条件边def should_continue(state:MessagesState):messages state[messages]last_message messages[-1]print (大模型的输出,last_message)#判断大模型是否调用了工具if not last_message.tool_calls:return END# Otherwise if there is, we continueelse:return execute_tools它的工作方式:大模型输出完成后,输出信息会进入条件边的这个函数里。函数检查最后一条消息(大模型的返回)是否带有tool_calls字段:如果是正常对话(比如你是谁),大模型的返回里没有tool_calls字段 → 返回 END → 直接结束。你问你是谁,它回答是谁,这个答案就是最终结果;如果用户输入刚好能触发工具调用(比如帮我算个加法),大模型的返回里就会有 tool_calls 字段(包含函数名称和函数参数)→ 返回 execute_tools → 走到工具节点,执行工具。最后那个path_map是对返回值的导向:函数返回END就走到结束符,返回 execute_tools 就走到工具节点——它描述返回值 → 去哪个节点的对应关系,决定图到底往哪走。运行演示:两种输入两种路径跑两个对比案例:案例一:问6×3 是多少。运行后会先打印出调用工具计算6*3——说明这个 18 是调用工具算出来的,不是大模型自己算出来的。案例二:问你是谁。它不会调用任何工具——条件边函数检查发现没有 tool_calls 字段,直接返回END,大模型答完就走完了。同一个图,两种输入,走了两条完全不同的路径——走哪条路,是大模型自己判断的,不是人指定的。完整流转分析:一次工具调用的全过程对着图把6×3是多少的完整流转走一遍。代码里把每一步都打印出来了:第一步: 用户输入进入 llm_node,大模型完成一次输入输出。注意观察大模型这次输出的是什么:content是空的!但tool_calls字段里有内容:函数名称(multiply)和函数参数(a6, b3)。也就是说,大模型这一轮并没有说话,它只表达了我要调用这个工具。第二步: 输出进入条件边函数 should_continue。大模型连的边就是条件边,所以大模型输出完之后,立刻调用条件边先做判断。函数发现最后一条消息带 tool_calls 字段 → 返回 execute_tools → 图走到工具节点。第三步: 工具节点执行工具。把工具的名称从 tool_calls 里取出来——注意一个细节:tool_calls 的数据形式虽然是数组,但实际上几乎都只有一个元素。理论上它支持多个元素,但就工作实践和查资料的情况来看,很少看到大模型一次并行调用多个工具,基本还是一个比较串行的过程。取出名称 multiply 后,通过 tools_by_name 的映射找到乘法函数,以a6, b3为参数 invoke 执行。第四步: 封装结果并返回。函数执行完,把结果(18)用ToolMessage封装(这是 LangChain 里的内容),返回到 state 里。注意:工具返回的就是函数的返回值本身,没有任何语言包装——你的函数 return 什么,它就是什么,直接返回来。第五步: 回到大模型节点,再调一次大模型。看工具节点连的是谁?连的是大模型节点。所以工具结果又会送回大模型——这一轮,是大模型对工具调用的结果进行总结:6 乘以 3 等于 18。第六步: 再走条件边,结束。大模型总结完再经过条件边, 这次没有工具调用, 返回END, 程序结束。整体结果(逐步打印):用户问6×3 是多少→ 大模型返回content为空,但有要调用工具的信息;工具返回18(函数返回值,无语言包装);第三轮把前面所有信息都给大模型,大模型总结:6乘以3等于18。这就是整套流程。例子虽然简单,但它恰恰体现了智能体和工作流的区别——理解这一步,是理解智能体非常关键的步骤。智能体与工作流的区别:控制权交给谁智能体和工作流的判断标准就一条:这张图的实际执行路径, 事先能不能预知。对上图, 它具体真实的路径:可能是START → LLM → END(直接答完结束);可能是START → LLM → 工具 → LLM → END;也可能是START → LLM → 工具 → LLM → 工具 → LLM……来回转好几圈。事先是不知道的,完全取决于大模型自身的判断结果。也就是说,我把判断权全部交给了大模型——图是我画的,但在这张图上到底怎么走,是大模型控制的,而不是我这个人控制的。这就叫做Agent(智能体)。打个比方:玩网游(MMORPG)时, 游戏设计者只设计了基本的游戏规则——你在里头打什么怪、跟谁组队、捡什么道具、买什么东西,设计者控制不了,完全由玩家自己决定。而横版过关那种很死板的RPG,第一关干什么、第二关干什么都是定死的,玩家没有什么自主选择权。智能体就是前者:把控制权交给了作为玩家的大模型。反过来看工作流的例子:RAG(召回 → 问答 → 答案)每一步的流程都是事先规定好的。它们在执行过程中也用大模型,但那是把大模型当工具、当函数用, 并没有把流程怎么走的权利交给大模型: 走完召回必然走问答, 完全由设计者控制。也正因为路径确定,它非常好debug。什么是智能体, 如果只回答智能体是个Function Call是没有任何亮点的——工作流一样能做工具调用。真正能说明问题的, 恰恰是例子中体现的东西:整个流程的自主权全部交给大模型。对照之前的RAG等例子: 大模型在其中只是作为一个语言工具、一个插件, 真正主导流程的还是设计者; 而对智能体来说,大模型起到的是决策作用。这就是二者的分水岭。两个框架的关系两个框架的关系:LangGraph是LangChain的全面升级版。LangChain 只提供了很基础的功能, 相当于给LangGraph提供了一些基本的插件(比如各种Message类型、工具调用的封装); 而LangGraph在这之上把图画起来,玩得更活更花。Function Call的实现有几种方法:最底层的实现: 自己写工具、自己写执行逻辑、自己写返回——怎么执行、怎么返回全部手写;ToolNode封装: 把执行工具的步骤封装起来直接搞一个ToolNode节点, add_node的时候把工具交给它, 执行工具全部封装起来,只需要写一个判断条件(条件边);create_react_agent: 高度封装的一行写法, 一句话图就出来了。内部等价于llm节点工具节点条件边那一整套。Function Call体会手写 → ToolNode → create_react_agent三层封装的演进。这三层体现的是一个通用规律:封装得越狠, 用起来就越方便——create_react_agent一句话直接出来;但越想做细微的修改就越困难。反过来,代码越偏向底层, 做定制化修改的时候就方便很多。