这是这是我们《AI开发实践》系列的第18篇我们继续学习Langchain。前面两节中我们学习了langchain中是如何解决LLM失忆问题的研究了如何压缩数据不超过LLM的上下文窗口大小。但是实际上问题仍旧没有结束因为在实际的业务中我们还需要使用自定义数据还要从对话中提取出来有用的数据存储下来并在后续使用。因此在这一节中我们就来学习这两点自定义持久化内容如何在对话中提取信息并在后续工具和提示词中访问 。对于LLM交互中的数据大致可以分成三类#数据类型典型内容存储位置 方案生命周期1本次请求临时数据token、ip、request_id、租户 ID存放CustomContext仅单次传入、不持久化单次请求有效2短期多轮聊天记忆对话中提及的预算、出行日期、临时需求存放CustomState由Checkpointer 自动持久化托管跟随会话 thread清空对话即失效3长期业务资料手机号、收货地址、订单、资产、会员档案State临时缓存 业务数据库MySQL/Redis双落地永久保存跨会话、跨设备生效我们主要是探讨第1种和第2种对于长期业务资料可以在2的基础上再同时按业务需要落库就行。在本文中先通过第一个示例来看看langchain中是怎么实现自定义持久化内容的包括上下文CustomContext和状态数据CustomState然后再通过两个更复杂的示例学习如何在用户对话中提取结构化数据如何在Tools和提示词中访问自定义的数据。一、自定义持久化内容在langchain中 自定义持久化内容是通过继承并实现AgentState来实现的。 自定义的内容也会自动的随着checkpoint的存档数据保存下来。fromlanggraph.checkpoint.postgresimportPostgresSaverfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.chat_modelsimportinit_chat_modelfromlangchain_deepseekimportChatDeepSeekfromlangchain.messagesimportHumanMessagefromrichimportprint# 初始化模型modelinit_chat_model(modeldeepseek-chat,temperature0.1,top_p0.3,max_tokens200,## stop [\n\n])#除了message, 我们还要记下来用用户IDclassCustomAgentState(AgentState):user_name:strtooldefget_user_info(runtime:ToolRuntime)-str:获取当前用户的名称user_nameruntime.state[user_name]returnuser_name#创建 checkpointerDB_URIpostgresql://postgres:Test_1234localhost:5432/postgres?sslmodedisablewithPostgresSaver.from_conn_string(DB_URI)ascheckpointer:checkpointer.setup()# auto create tables in PostgreSQLagentcreate_agent(modelmodel,# tools[get_user_info],state_schemaCustomAgentState,#将agent的state_schema 替换为自定义的statecheckpointercheckpointer,)#Sean, favorite color is green, 调用是关联到thread_id:1questionHumanMessage(content请查询并显示当前用户的名称)config{configurable:{thread_id:1}}responseagent.invoke({messages:[question_sean],user_name:小刚},config,)print(response)可以看到记录的历史信息中增加了user_name: 小刚{ messages: [ HumanMessage( content请使用工具查询并显示当前用户的名称, ... ), AIMessage( content, ... }, idlc_run--019e0574-0ce5-7031-b7bd-bf4eb4171377-0, tool_calls[ { name: get_user_info, args: {}, id: call_00_yukmYR51ggmci8lE2rAZ2304, type: tool_call } ], ... ), ToolMessage( content小刚, nameget_user_info, ... ), AIMessage( content当前用户的名称是 **小刚**。, ... ) ], user_name: 小刚 }这些自定义内容保存在 数据库的checkpoint表中channel_values字段。以上述代码的checkpoint 一条记录为例4, id: 1f14a871-9f7d-6c0e-8003-adf470de64a1, ts: 2026-05-08T02:39:14.19101500:00, versions_seen: { model: { branch:to:model: 00000000000000000000000000000004.0.4444557217455798 }, tools: {}, __input__: {}, __start__: { __start__: 00000000000000000000000000000001.0.9832353077932918 } }, channel_values: { user_name: 小刚 }, channel_versions: { messages: 00000000000000000000000000000005.0.3893909274255245, __start__: 00000000000000000000000000000002.0.8868211186730097, user_name: 00000000000000000000000000000002.0.8868211186730097, __pregel_tasks: 00000000000000000000000000000004.0.4444557217455798, branch:to:model: 00000000000000000000000000000005.0.3893909274255245 }, updated_channels: [messages] | {step: 3, source: loop, parents: {}}可以看到其中的channel_values字段中保存了自定义内容user_name: 小刚。二、Access memory 访问记忆上述的示例中对于用户名称是我们在agent.invoke()时主动的传入进去的。在实际的工作中我们也可以从对话中提取出来结构化的数据存储在State中并在后续进行访问使用。根据访问的方式不同可以分为在工具中进行访问记忆数据也可以在提示词中访问记忆数据。1、 Tools 工具访问在这个示例中我们先通过小模型从用户对话中提取出来用户名进行简单的合法性验证后保存到State中去。然后在后续对话中历史信息被裁剪之后再通过Tool从State中获取存储下来的结构化数据。通过这个过程我们模拟了在实际工作中如何从对话提取数据校验并保存数据工具访问数据的过程。fromlangchain.toolsimporttool,ToolRuntimefromlangchain_core.runnablesimportRunnableConfigfromlangchain.messagesimportToolMessage,HumanMessage,SystemMessagefromlangchain.agentsimportcreate_agent,AgentStatefromlanggraph.typesimportCommandfrompydanticimportBaseModelfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.chat_modelsimportinit_chat_modelfromrichimportprintfromlangchain_ollamaimportOllamaLLMfromdotenvimportload_dotenvfromutilsimportprint_all_checkpoints,find_keep_start_index,set_debug_mode,LLMDebugCallbackfromlangchain.agents.middlewareimportbefore_modelfromlanggraph.runtimeimportRuntimefromtypingimportAnyfromlangchain.messagesimportRemoveMessageimportre# 加载环境变量load_dotenv()# 调测开关set_debug_mode(True)# 多轮对话记忆classCustomState(AgentState):user_name:str无名称# 单次invoke上下文classCustomContext(BaseModel):user_id:str# 本地轻量模型低成本快速用于总结摘要local_llmOllamaLLM(modelllama3.2:3b,temperature0.1,top_p0.3,callbacks[LLMDebugCallback()])# 验证 清理合法中文名defis_valid_chinese_name(name:str)-bool: 判断提到出来的用户名称是否为【合法中文名】 规则2-4个纯中文字符无任何其他字符 ifnotisinstance(name,str):returnFalse# 新增判断字符串不等于 Noneifname.strip()None:returnFalse# 正则严格匹配 2-4 个中文字符无任何额外处理returnbool(re.fullmatch(r^[\u4e00-\u9fa5]{2,4}$,name))before_modeldefextract_userName_from_userInput(state:AgentState,runtime:Runtime): 利用本地小模型从当前对话消息中提取当前用户名并存入到State中去 old_messagesstate[messages]#如果不是HumanMessage直接返回ifnotisinstance(old_messages[-1],HumanMessage):return# 新增对话tate中去new_user_inputstate[messages][-1].content.strip()# 系统提示词system_promptf 任务从对话中提取当前用户名 规则 1.仅从对话原文中提取 2.没有用户名时返回None 3.除用户名外不输出任何字符 4.输出格式为输出:用户名 # 用户问题user_promptf 对话为“{new_user_input}” # 构造消息列表extract_prompt[SystemMessage(contentsystem_prompt),# 系统规则HumanMessage(contentuser_prompt)# 用户问题]# 从LLM输出中提取输出的名称llm_extract_resultlocal_llm.invoke(extract_prompt)# 正则提取“输出”后的内容matchre.search(r输出[:]\s*(.),llm_extract_result.strip())ifnotmatch:return# 获取名称并校验potention_namematch.group(1).strip()ifnotpotention_nameornotis_valid_chinese_name(potention_name):returnprint(fpotention_name {potention_name})# 更新状态returnCommand(update{user_name:potention_name})before_modeldeftrim_messages(state:AgentState,runtime:Runtime)-dict[str,Any]|None:Keep only the last few messages to fit context window.messagesstate[messages]iflen(messages)2:returnNone# 消息太少就不处理了keep_startfind_keep_start_index(messages,keep_rounds2)return{messages:[RemoveMessage(idm.id)forminmessages[0:keep_start]]}tooldefget_user_name(runtime:ToolRuntime[CustomContext,CustomState])-str|Command:使用该方法获取当前对话的用户名.user_nameruntime.state.get(user_name,None)returnuser_name# 初始化模型modelinit_chat_model(modeldeepseek-chat,temperature0.1,top_p0.3,max_tokens500,## stop [\n\n])memory_checkpointerInMemorySaver()agentcreate_agent(modelmodel,middleware[extract_userName_from_userInput,trim_messages],#从用户消息中提取用户名称只保留2对完整对话tools[get_user_name],#获取当前用户名称的工具checkpointermemory_checkpointer,state_schemaCustomState,#自定义State, 持久化用户名context_schemaCustomContext,#定制上下文存放user_idsystem_prompt这是一个短小的对话每次回答不超过50字。如果需要答复用户名称必须调用工具get_user_name来获取信息,)config:RunnableConfig{configurable:{thread_id:1}}agent.invoke({messages:你好我的名字叫小明},contextCustomContext(user_iduser_123),configconfig)agent.invoke({messages:其实我的名字叫小刚},contextCustomContext(user_iduser_123),configconfig)agent.invoke({messages:写一首关于猫的短诗},config)agent.invoke({messages:现在用同样的风格写一首关于狗的},config)agent.invoke({messages:现在用同样的风格写一首关于牛的},config)agent.invoke({messages:现在用同样的风格写一首关于羊的},config)agent.invoke({messages:现在用同样的风格写一首关于树的},config)agent.invoke({messages:现在用同样的风格写一首关于草的},config)final_responseagent.invoke({messages:我叫什么名子?},contextCustomContext(user_iduser_123),configconfig)# print(final_response[messages][-1].content)print(final_response)# print_all_checkpoints(memory_checkpointer,config)在示例中我们在before_model方法 extract_userName_from_userInput中通过本地小模型OllamaLLM llama3.2:3b 尝试从每句对话中提取用户名存入State在用户询问名称时通过在线大模型调用工具通过tool 方法get_user_name从State中取数据。在这段信息中用户先声称自己叫“小刚”后又更正为“小刚”。如果仅仅是让模型自己来根据历史信息直接回答很容易出错而在历史信息被裁剪的情况下根本就无法获得正确的名称。通过本地LLM主动信息提取的方法则可以准确的将有用的信息主动的结构化保存下来。其中的注意点是1使用的本地LLM需要注意比较比如qwen3:1.7bqwen3.5:2bqwen3:4bllama3.2:3b需要实际对比选择一个准确度和效率都比较高的模型。本例中实际对比后选用的是llama3.2:3b,速度堪比qwen3:1.7b准确率堪比qwen3:4b。2小模型的提示词要简洁简单小短句两三句话把目标和规则讲清楚太复杂的提示词小模型理解不了会迷乱掉胡乱输出3要对输出的格式进行约束方便输出后通过正则表达式进行结果提取4格式约束的方式要选择小模型擅长的键值对比如上文中对本地LLM输出要求“ 4.输出格式为输出:用户名”引导词为一个固定字符不能为目标结果的同近义词防止被模型替换掉不要尝试用JSON小模型理解不了这么复杂的格式。5即使是一样的提示词类似的输入每次输出时模型仍旧可能会添加各种多余的输出所以仍旧要通过正则表达式进行提取6小模型给出的结果必然会有错误的因此 对于提取的结果一定要进行尽力而为的严格验证凡是存疑的都要转人工去处理。2、Prompt 提示如果需要根据运行时上下文或智能体状态修改系统提示词时可以dynamic_prompt来直接修改request: ModelRequest中的系统提示词。fromlangchain.agentsimportcreate_agentfromtypingimportTypedDictfromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestfromlangchain.agents.middlewareimportbefore_modelfromlanggraph.typesimportCommandfromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain_core.runnablesimportRunnableConfigfromutilsimportprint_all_checkpoints,set_debug_mode,LLMDebugCallbackfromdotenvimportload_dotenv# 加载环境变量load_dotenv()# 调测开关set_debug_mode(True)classCustomContext(TypedDict):user_name:str# 系统提示词SYSTEM_PROMPT_TEMPLATE 请帮忙写一段{style}。 defget_user_style(user_id:str)-str: 根据用户名返回诗歌风格古诗 / 现代诗 ifuser_iduser_001:return古诗ifuser_iduser_002:return现代诗# 默认风格return现代诗dynamic_promptdefdynamic_system_prompt(request:ModelRequest)-str:user_idrequest.runtime.context.get(user_id,invalid_id)styleget_user_style(user_id)updated_promptSYSTEM_PROMPT_TEMPLATE.format(user_iduser_id,stylestyle)returnupdated_prompt# 初始化模型modelinit_chat_model(modeldeepseek-chat,temperature0.1,top_p0.3,max_tokens500,## stop [\n\n]callbacks[LLMDebugCallback()])memory_checkpointerInMemorySaver()agentcreate_agent(modelmodel,middleware[dynamic_system_prompt],checkpointermemory_checkpointer,context_schemaCustomContext,)responseagent.invoke({messages:[{role:user,content:写一首关于猫的短诗}]},context{user_id:user_001},config{configurable:{thread_id:1}})print(response)responseagent.invoke({messages:[{role:user,content:写一首关于猫的短诗}]},context{user_id:user_002},config{configurable:{thread_id:2}})print(response)dynamic_prompt是在before_model之后执行的此时已经形成了请求结构体ModelRequest因此可以直接获取修改。但是对于动态提示词也完全可以通过before_model中通过获取提示词进行参数注入。三、总结本文主要通过两个实例展示了如何在langchain中自定义数据如何从对话中提取结构化数据如何通过langchain的自定义State进行持久化如何在运行期间通过Tools和提示词为获取持久化数据。结合这些基础方法可以对模型使用过程中进行数据的治理打下一个基础。