乘风的人 - LangChain Code
type: channel
link: https://blog.zhangbiechen.cn/index.php/tag/LangChain-Code/
description: ai
items: Code - LangGraph 基础
items: code - 本地智能问答助手(Ollama + RAG)
items: RAG Advanced (RAG 进阶) - code.py
items: RAG - 文档加载器
items: Validation & Retry.py
items: Structured Output.py
items: Middleware Basics -main.py
items: Middleware Basics.py
items: Checkpointing - main.py
items: check_point.py

Code - LangGraph 基础
type: item
date: 2026-05-19T14:20:01+08:00
description: LangGraph# 什么是 LangGraph？ # LangGraph 是一个用于构建状态化、多步骤 AI 应用的框架。它使用图（Graph） 的概念来组织工作流： # 节点（Nodes）：图中的处理单元，可以是 LLM 调用、工具执行或自定义函数 # 边（Edges）：连接节点的路径，定义执行顺序 # 状态（State）：在节点之间传递的数据结构 # LangGraph vs LangChain # 特性 LangChain LangGraph # 抽象级别 高级 低级 # 适用场景 快速构建标准 Agent 复杂自定义工作流 # 控制粒度 通过中间件 完全控制每个节点 # 状态管理 自动 手动但灵活 # Hello World Graph # 1. 定义状态 - 使用 TypedDict from typing import TypedDict, Annotated,Dict from langgraph.graph import StateGraph class AgentState(TypedDict): message: str def greeting_node(state: AgentState)-> AgentState: """ 一个向状态添加消息的简单节点 """ state['message'] = "Hello" + state["message"] return state graph = StateGraph(AgentState) graph.add_node("greeter",greeting_node) graph.set_entry_point("greeter") graph.set_finish_point("greeter") app = graph.compile() 打印图流程from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) 多个节点并添加边#多个节点并添加边 from typing import TypedDict, Annotated,Dict from langgraph.graph import StateGraph class AgentState(TypedDict): name :str age : str final:str def first_node(state: AgentState)-> AgentState: """这是第一个节点""" state["final"] = f"你好 {state['name']}" return state def second_node(state: AgentState)-> AgentState: """这是第二个节点""" state["final"] = state["final"] + f"你已经 {state['age']}了" return state graph = StateGraph(AgentState) graph.add_node("first_node",first_node) graph.add_node("second_node",second_node) graph.set_entry_point("first_node") graph.add_edge("first_node","second_node") graph.set_finish_point("second_node") app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) result = app.invoke({"name":"zhangsan","age":"18",}) print(result) 加模型顺序执行#加模型顺序执行 from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from typing import TypedDict, Annotated,Dict # ========================================== # 1. 初始化模型 # ========================================== model = ChatOllama( model="qwen3.5:2b", base_url="http:…

code - 本地智能问答助手(Ollama + RAG)
type: item
date: 2026-04-03T10:35:00+08:00
description: 数据向量化#读取 现有json文件，不需要分割的数据直接构建 import json from langchain_experimental.text_splitter import SemanticChunker from langchain_community.vectorstores import FAISS from langchain_core.documents import Document # 读取文件内容 with open("./iciba_data_multi.json", encoding="utf-8") as f: json_data = json.load(f) # 加载embedding模型 from langchain_huggingface import HuggingFaceEmbeddings MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" # mps mac芯片mps加速 embeddings = HuggingFaceEmbeddings( model_name=MODEL_PATH, model_kwargs={"device": "mps"}, encode_kwargs={"normalize_embeddings": True} ) # 2. 准备文档列表 # 这里我们不再使用 text_splitter，而是手动构建 Document 列表 documents = [] for item in json_data: # 拼接文本：英文+中文 full_text = f"{item['content']} {item['note']}" # 构建 LangChain 的 Document 对象 # page_content 是要向量化的内容 # metadata 是你想保留但不参与向量化的信息（比如日期） doc = Document( page_content=full_text, metadata={"date": item["date"]} ) documents.append(doc) # 从文档列表创建 FAISS 向量库 vector_store = FAISS.from_documents(documents=documents, embedding=embeddings) # 【可选】持久化存储：将索引保存到本地磁盘 vector_store.save_local("faiss_index") Ollama 运行 Qwen3.5-2B❯ ollama run qwen3.5:2b pulling manifest pulling b709d81508a0: 100% ▕██████████████████████████████████████▏ 2.7 GB pulling 9be69ef46306: 100% ▕██████████████████████████████████████▏ 11 KB pulling 9371364b27a5: 100% ▕██████████████████████████████████████▏ 65 B pulling ee043a99abe5: 100% ▕██████████████████████████████████████▏ 473 B verifying sha256 digest writing manifest success >>> 线阶段：RAG 问答# 在线阶段：RAG 问答 # 1. 核心思路 # 创建一个 “检索工具”，AI 回答问题前，先调用这个工具找相关文档，再结合文档回答（避免瞎编）。 import json from langchain.agents import create_agent from langchain_core.tools import tool from langchain_ollama import ChatOllama # ========================================== # 1. 初始化模型 # ========================================== model = ChatOllama( model="qwen3.5:2b", base_url="http://localhost:11434", temperature=0.7, ) # ==============…

RAG Advanced (RAG 进阶) - code.py
type: item
date: 2026-04-02T16:01:00+08:00
description: 加载文件进行分割向量化存储到 faissfrom langchain_experimental.text_splitter import SemanticChunker from langchain_community.vectorstores import FAISS # 读取文件内容 with open("./langchaintest.txt", encoding="utf-8") as f: state_of_the_union = f.read() #返回字符串 # 加载embedding模型 from langchain_huggingface import HuggingFaceEmbeddings MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" # mps mac芯片mps加速 embeddings = HuggingFaceEmbeddings( model_name=MODEL_PATH, model_kwargs={"device": "mps"}, encode_kwargs={"normalize_embeddings": True} ) # # 获取切割器 text_splitter = SemanticChunker( embeddings=embeddings, breakpoint_threshold_type="percentile",#断点阈值类型：字面值["百分位数", "标准差", "四分位距", "梯度"] 选其一 breakpoint_threshold_amount=15 #断点阈值数量 (极低阈值 → 高分割敏感度) ) # 切分文档 docs = text_splitter.create_documents(texts = [state_of_the_union]) # 从文档列表创建 FAISS 向量库 vector_store = FAISS.from_documents(documents=docs, embedding=embeddings) # 【可选】持久化存储：将索引保存到本地磁盘 vector_store.save_local("faiss_index") 读取 现有json文件，不需要分割的数据直接构建import json from langchain_experimental.text_splitter import SemanticChunker from langchain_community.vectorstores import FAISS from langchain_core.documents import Document # 读取文件内容 with open("./iciba_data_multi.json", encoding="utf-8") as f: json_data = json.load(f) # 加载embedding模型 from langchain_huggingface import HuggingFaceEmbeddings MODEL_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" # mps mac芯片mps加速 embeddings = HuggingFaceEmbeddings( model_name=MODEL_PATH, model_kwargs={"device": "mps"}, encode_kwargs={"normalize_embeddings": True} ) # 2. 准备文档列表 # 这里我们不再使用 text_splitter，而是手动构建 Document 列表 documents = [] for item in json_data: # 拼接文本：英文+中文 full_text = f"{item['content']} {item['note']}" # 构建 LangChain 的 Document 对象 # page_content 是要向量化的内容 # metadata 是你想保留但不参与向量化的信息（比如日期） doc = Document( page_content=full_text, metadata={"date": ite…

RAG - 文档加载器
type: item
date: 2026-03-19T17:08:00+08:00
description: 加载TXT的文件#安装langchain_community # pip install langchain_community # 读取普通的文件 from langchain_community.document_loaders import TextLoader text = TextLoader(file_path="asset/load/01-langchain-utf-8.txt",encoding="utf-8") docs = text.load() print(docs) # Documment对象中有两个重要的属性： # page_content：真正的文档内容 # metadata：文档内容的原数据 加载pdf#加载pdf #LangChain加载PDF文件使用的是pypdf # pip install pypdf from langchain_community.document_loaders import PyPDFLoader pdfLoader = PyPDFLoader(file_path="asset/load/02-load.pdf") docs = pdfLoader.load() # print(type(docs)) # print(docs[0]) for doc in docs: print(f"文档的内容：{doc.page_content}") # type(pages[0]) #langchain_core.documents.base.Document # pages[0].page_content #只获取本页内容 # pages[0].metadata # {...,'source': './asset/load/load.pdf',.., 'page': 0} load_and_split 加载并切分from langchain_community.document_loaders import PyPDFLoader pdfLoader = PyPDFLoader(file_path="asset/load/02-load.pdf") docs = pdfLoader.load_and_split() # print(docs) for doc in docs: print(f"❀{doc.page_content}") 加载CSVfrom langchain_community.document_loaders import CSVLoader loader = CSVLoader(file_path="asset/load/03-load.csv") data= loader.load() print(type(data[0].page_content)) #打印每一行 for doc in data: print(f"{doc.page_content}\n") 加载JSON#安装jq pip install jq from langchain_community.document_loaders import JSONLoader #情况1 打印完整json数据 loader = JSONLoader( file_path="asset/load/04-load.json", jq_schema=".", text_content=False # <--- 关键修改：允许 page_content 是 dict 类型 ) docs = loader.load() print(docs[0].page_content) # 情况2 # .messages[].content:遍历.messages[]中所有元素 从每一个元素中提取.content字段 json_loader = JSONLoader( file_path="asset/load/04-load.json", jq_schema=".messages[].content", ) # 3.加载 doc2 = json_loader.load() print("-" * 60) print(doc2[0].page_content) 提取JSON内容#json 示例2 提取04-response.json文件中嵌套在 data.items[].content 的文本 # 如果希望处理 JSON 中的 嵌套字段、数组元素提取，可以使用 content_key 配合is_content_key_jq_parsable=True ，通过 jq 语法精准定位目标数据。 # 通常，对api请求结果的采集 # "status": "success", # "data": { # "page": 2, # "p…

Validation & Retry.py
type: item
date: 2026-03-17T15:20:00+08:00
description: # 核心概念 # 验证和重试 = 确保 LLM 应用的可靠性和数据质量 # 在生产环境中，需要处理三类问题： # 网络错误 - 临时性连接问题（用 with_retry()） # 模型故障 - 主模型不可用（用 with_fallbacks()） # 输出质量 - LLM 输出不符合要求（用验证 + 重试循环） # 基本用法 # 1. with_retry() - 自动重试 # 处理临时性网络错误： import os import sys from langchain.chat_models import init_chat_model from pydantic import BaseModel, Field, ValidationError sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() def demo1(): # 添加重试机制 llm_with_retry = chat_model.with_retry( retry_if_exception_type=(ConnectionError, TimeoutError), wait_exponential_jitter=True, # 指数退避 + 随机抖动 stop_after_attempt=3, # 最多重试 3 次 ) response = llm_with_retry.invoke("你好") # 工作原理： # 第 1 次尝试 → 失败 (ConnectionError) # ↓ 等待 1s # 第 2 次尝试 → 失败 (ConnectionError) # ↓ 等待 2s # 第 3 次尝试 → 成功 ✓ # 2. with_fallbacks() - 降级方案 # 主模型失败时切换到备用模型： def demo2(): # 主模型 primary_model = init_chat_model("groq:llama-3.3-70b-versatile") # 备用模型（更可靠/更便宜） fallback_model = init_chat_model("groq:llama-3.1-8b-instant") # 配置降级 llm_with_fallbacks = primary_model.with_fallbacks([fallback_model]) response = llm_with_fallbacks.invoke("介绍 Python") # 主模型正常 → 使用主模型 # 主模型失败 → 自动切换到备用模型 # 3. Pydantic 验证 # 使用 Pydantic 约束确保数据质量： def demo3(): from pydantic import BaseModel, Field, field_validator class User(BaseModel): name: str = Field(min_length=2, max_length=20) age: int = Field(ge=0, le=150) # 0-150 岁 email: str @field_validator("email") @classmethod def validate_email(cls, v): if "@" not in v: raise ValueError("邮箱必须包含 @") return v # 使用 try: user = User(name="张三", age=200, email="invalid") # 失败 except ValidationError as e: print(e.errors()) # 查看错误详情 # 核心组件 # 1. with_retry() 参数 # 参数 说明 默认值 # retry_if_exception_type 哪些异常会触发重试 (Exception,) # stop_after_attempt 最大重试次数 3 # wait_exponential_jitter 是否使用指数退避 + 抖动 False # 常见重试异常： # # 网络相关 # (ConnectionError, TimeoutError, httpx.ConnectError) # # API 限流 # (RateLimitError, ) # # 所有异常（谨慎使用） # (Exception, ) # 2. with_fallbacks() 参数 # primary_mo…

Structured Output.py
type: item
date: 2026-03-17T11:53:00+08:00
description: # 核心概念 # Structured Output = 将 LLM 的自然语言输出转为结构化 Python 对象 # 在 LangChain 1.0 中，使用 with_structured_output() 方法结合 Pydantic 模型，可以确保 LLM 返回符合预定义模式的数据。 # 基本用法 # 定义 Pydantic 模型 from typing import List, Optional from pydantic import BaseModel, Field, model_serializer class Person(BaseModel): # 变量名是英文 (name)，描述可以是中文，帮助模型理解 name: str = Field(description="人的姓名") # 变量名是英文 (age) age: int = Field(description="人的年龄") # 变量名是英文 (job) job: Optional[str] = Field(default=None, description="人的职业") # 自定义序列化方法，消除警告 @model_serializer def serialize_model(self): return {"name": self.name, "age": self.age, "job": self.job} # email: Optional[str] = Field(default=None, description="人的邮箱") import os import sys from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.tools import tool from langchain.agents.middleware import AgentMiddleware from langgraph.checkpoint.memory import InMemorySaver sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() # 使用 with_structured_output() # 核心组件 # 1. Pydantic BaseModel # 所有结构化输出的数据模型都必须继承 BaseModel： # from pydantic import BaseModel # class MyModel(BaseModel): # field1: str # field2: int # 2. Field 描述 # 使用 Field() 添加字段描述，帮助 LLM 理解： from pydantic import Field # 高级特性 # 可选字段 class Person(BaseModel): name: str age: Optional[int] = None job: Optional[str] = None # 可以为 None def demo1(): # 创建结构化输出的 LLM structured_llm = chat_model.with_structured_output(Person, include_raw=False) response = structured_llm.invoke( "张三是一名 30 岁的软件工程师。请提取信息。按照要求返回JSON格式数据，字段名必须为英文（name/age/job）" ) # print(response) print(f"name={response.name} age={response.age} job={response.job}") # 工作原理 # 传统方式 vs 结构化输出 # 传统方式（繁琐）： # 1. 提示词要求 JSON # prompt = "以JSON格式返回：{name, age, occupation}" # response = chat_model.invoke(prompt) # # 2. 手动解析 # import json # data = json.loads(response.content) # # 3. 手动验证类型 # if not isinstance(da…

Middleware Basics -main.py
type: item
date: 2026-03-17T09:40:00+08:00
description: """ LangChain 1.0 - Middleware Basics (中间件基础) ============================================== 本模块重点讲解： 1. 什么是中间件（Middleware） 2. before_model 和 after_model 钩子 3. 自定义中间件的创建 4. 多个中间件的组合 5. 内置中间件的使用 """ import os import sys from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain.agents import create_agent from langchain_core.tools import tool from langchain.agents.middleware import AgentMiddleware from langgraph.checkpoint.memory import InMemorySaver sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from init_model import get_chat_model chat_model = get_chat_model() @tool def get_weather(city: str) -> str: """ 获取指定城市的天气信息 参数: city (str): 要查询天气的城市名称 例如 "北京" 返回: str: 包含天气信息的字符串，格式为 "城市 天气描述 温度" """ # 模拟天气数据 weather_data = { "北京": "晴天，温度 15°C，空气质量良好", "上海": "多云，温度 18°C，有轻微雾霾", "深圳": "阴天，温度 22°C，可能有小雨", "成都": "小雨，温度 12°C，湿度较高", } # 检查城市是否在数据中 if city not in weather_data: return f"城市 {city} 的天气信息未找到" # 返回天气信息 return weather_data.get(city, f"抱歉，暂时没有{city}的天气数据") # ============================================================================ # 示例 1：最简单的中间件 """ 关键规则 1. 必须继承 AgentMiddleware ← 这个固定 2. 方法名固定 (before_model, after_model) ← 这个固定 3. 类名随意 ← 这个不固定 LangGraph 只看： - 是否继承 AgentMiddleware？ - 是否有 before_model / after_model 方法？ """ import logging from typing import Any, Dict, Optional # 配置 logger logger = logging.getLogger(__name__) # ============================================================================ class LoggingMiddleware(AgentMiddleware): """ 日志中间件 - 记录每次模型调用 """ def before_model( self, state: Dict[str, Any], runtime: Any ) -> Optional[Dict[str, Any]]: """模型调用前""" messages = state.get("messages", []) logger.info(f"[Middleware] Before Model Call | Message Count: {len(messages)}") # 这里可以添加更多上下文信息，例如 runtime.request_id return None # 继续正常流程 def after_model( self, state: Dict[str, Any], runtime: Any ) -> Optional[Dict[str, Any]]: """模型响应后""" messages = state.get("messages", []) if not messages: logger.war…

Middleware Basics.py
type: item
date: 2026-03-17T09:39:00+08:00
description: # Middleware（中间件）= Agent 执行过程中的钩子函数 # 在 LangChain 1.0 中，中间件是处理 Agent 生命周期的标准方式。 # 基本用法 # 创建自定义中间件 import sys import os from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver from langchain.agents.middleware import AgentMiddleware chat_model = get_chat_model() class MyMiddleware(AgentMiddleware): def before_model(self, state, runtime): """模型调用前执行""" print("准备调用模型") return None # 返回 None 表示继续正常流程 def after_model(self, state, runtime): """模型响应后执行""" print("模型已响应") return None # 返回 None 表示不修改状态 def demo1(): # 使用中间件 agent = create_agent(model=chat_model, tools=[], middleware=[MyMiddleware()]) result = agent.invoke({"messages": [{"role": "user", "content": "你好"}]}) print(result["messages"][-1].content) # 准备调用模型 # 模型已响应 # 你好！👋 很高兴见到你！ # 有什么我可以帮你的吗？无论是回答问题、创作内容、查阅资料，还是随便聊 聊，我都在这里哦～ # 核心钩子方法 # 1. before_model（模型调用前） def before_model(self, state, runtime): """ 在模型调用前执行 返回值： - None: 继续正常流程 - dict: 更新状态（如 {"messages": [...]}） - {"jump_to": "..."}: 跳过正常流程 """ messages = state.get("messages", []) print(f"当前消息数: {len(messages)}") return None # 用途： # 消息修剪（trim messages） # PII 脱敏 # 输入验证 # 条件路由 # 1. before_model: 模型调用前的“守门员” # 执行时机： # 在 Agent 收集完所有上下文信息，准备将请求发送给大语言模型（LLM）之前立即执行。 # 核心参数： # state: 当前累积的对话状态（包含历史消息 messages、变量、用户输入等）。 # runtime: 运行时环境（包含配置、API Key、执行元数据等）。 # 主要功能场景： # 1. 动态提示词注入 (Prompt Injection)： # 根据当前状态，动态在 messages 列表头部插入 System Prompt 或 Few-shot 示例。 # 例：检测到用户情绪愤怒时，插入“请保持冷静和同理心”的系统指令。 # 2.上下文裁剪/压缩 (Context Pruning)： # 如果 state['messages'] 超过了模型的 Token 限制，在此处截断旧消息或调用摘要模型进行压缩。 # 3. 安全拦截 (Safety Guardrails)： # 检查用户输入是否包含敏感词、攻击性内容或越狱指令。如果违规，可以直接返回跳转指令，阻止请求发送给 LLM。 # 4.路由决策 (Routing)： # 分析用户意图，决定是否需要跳过当前模型，直接跳转到另一个特定的节点（Node）或工作流。 # 5.工具预填充： # 在某些架构中，可以预先计算某些工具的结果并放入上下文，让模型“看到”这些结果。 # 2. after_model（模型响应后） def…

Checkpointing - main.py
type: item
date: 2026-03-13T15:16:00+08:00
description: """ LangChain 1.0 - Checkpointing (检查点持久化) ========================================= 本模块重点讲解： 1. SqliteSaver - SQLite 持久化（LangGraph 提供） 2. 与 InMemorySaver 的区别 3. 跨进程、跨重启的对话持久化 4. 实际应用场景 1. invoke 前：LangGraph 自动调用 checkpointer.get(thread_id="user_123") - 查询数据库，读取该 thread_id 的历史消息 - 如果是第一次，返回空列表 2. invoke 中：Agent 处理时会看到完整历史(默认确实会全部读取，) state = { "messages": [历史消息1, 历史消息2, 新消息] # 自动合并 } 3. invoke 后：LangGraph 自动调用 checkpointer.put(thread_id, state) - 将新的完整状态写入数据库 - 数据库存储：(thread_id, timestamp, messages) """ import sys import os from openai import chat sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver chat_model = get_chat_model() # 持久化数据库路径 db_path = "checkpoints.sqlite" # ============================================================================ # SqliteSaver 持久化 # ============================================================================ def example_sqlite_saver(): """ 示例2：使用 SqliteSaver 实现持久化 关键： 1. SqliteSaver.from_conn_string("sqlite:///path/to/db.sqlite") 2. 对话持久化到 SQLite 文件 3. 程序重启后仍可恢复 """ with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_zhangsan"}} # 第一次运行 response = agent.invoke( {"messages": [{"role": "user", "content": "你好,我叫张三"}]}, config ) print(response["messages"][-1].content) def example_sqlite_saver_2(): """ 示例3：使用 SqliteSaver 实现持久化 关键： 1. SqliteSaver.from_conn_string("sqlite:///path/to/db.sqlite") 2. 对话持久化到 SQLite 文件 3. 程序重启后仍可恢复 """ with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_zhangsan"}} # 第一次运行 reponse = agent.…

check_point.py
type: item
date: 2026-03-13T15:15:00+08:00
description: # 核心概念 # Checkpointing = 将对话状态持久化到数据库 # InMemorySaver → 内存中（程序退出即丢失） # SqliteSaver/ → SQLite 数据库（持久化存储） # 基本用法 # InMemorySaver 的限制 import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from langchain.tools import tool from init_model import get_chat_model from langgraph.checkpoint.memory import InMemorySaver from langchain.agents import create_agent from langgraph.checkpoint.sqlite import SqliteSaver chat_model = get_chat_model() def demo1(): agent = create_agent(model=chat_model, tools=[], checkpointer=InMemorySaver()) # 限制： # ❌ 程序重启后丢失 # ❌ 无法跨进程共享 # ❌ 不适合生产环境 # SqliteSaver（推荐生产使用） def SqliteSaver_demo(): db_path = "checkpoints.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[], checkpointer=checkpointer # 使用 SQLite ) config = {"configurable": {"thread_id": "user_123"}} # 第一次运行 agent.invoke( {"messages": [{"role": "user", "content": "你好,我叫张三"}]}, config ) with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent(model=chat_model, checkpointer=checkpointer) result = agent.invoke( {"messages": [{"role": "user", "content": "我是谁"}]}, config ) print(result["messages"][-1].content) # 生成的checkpoints.sqlit文件 # ├── 09 - Checkpointing (检查点持久化)/ # │ └── check_point.py # ├── 📁 其他练习/ # ├── 📁 tools/ # ├── ⚙️ .env # ├── ️ checkpoints.sqlite # ├── 🗄️ checkpoints.sqlite-shm # ├── 🗄️ checkpoints.sqlite-wal # └── 📄 init_model.py # 运行结果 # 你是**张三**呀！👋 # 你之前已经跟我介绍过很多次啦（比如最开始你就说过“你好，我叫张三”）。 虽然你可能是在考我的记忆力，但我确实把你的名字记在心里了哦！😄 # 除了名字之外，我还挺好奇你的其他故事的。今天有什么特别想聊的话题，或 者需要我帮忙的地方吗？ # 跨进程访问 # 进程 A（Web 服务器） def demo2(): with SqliteSaver.from_conn_string("shared.sqlite") as checkpointer: agent_a = create_agent(model=model, checkpointer=checkpointer) agent_a.invoke({...}, config={"configurable": {"thread_id": "user_1"}}) # 进程 B（后台任务） with SqliteSaver.from_conn_string("shared.sqlite") as checkpointer: agent_b = create_agent(model=model, checkpointer=checkpointer) # 可以访问进程 A…