LangGraph 的 Checkpoint（检查点）机制是其内置持久化层的核心。简单来说，它像一个为你的图（Graph）状态提供的“Git”版本控制系统，会在图执行的每个“超级步”（Super-step）后，自动保存一份完整的状态快照。1. 核心概念要理解Checkpoint机制，首先需要掌握以下几个核心概念：Checkpoint (检查点)：它是图状态在某个特定时间点的完整快照。快照包含所有通道值、通道版本等信息。每个检查点都有一个全局唯一的、单调递增的ID。Super-step (超级步)：这是图执行的一个“节拍”。在一个超级步中，所有计划在该步执行的节点会（可能并行地）执行。LangGraph会在每个超级步的边界处创建一个检查点。Thread (线程)：这是将一系列检查点组织起来的逻辑单元，通过唯一的 thread_id 来标识。它代表了一次独立的对话或工作流会话。通过 thread_id，你可以隔离和管理不同会话的状态。2. Checkpointer 的作用与带来的能力Checkpointer（检查点器）是实现持久化的组件。当你用 checkpointer 参数编译图时，持久化层就被激活了。它能为你带来以下关键能力：记忆 (Memory)：在多轮对话中，Agent 能记住之前的交互内容。容错 (Fault-tolerance)：当执行因错误中断时，可以从最后一个成功的检查点恢复。人机协同 (Human-in-the-loop)：允许人类在检查点处审查、中断或修改状态，然后继续执行。时间旅行 (Time Travel)：你可以回溯到任何一个历史检查点，查看当时的状态，甚至从这个点“分叉”（Fork）出一条新的执行路径来探索其他可能性。3. 各种情况与场景详解Checkpoint 机制在不同场景下发挥着不同作用。3.1. 正常执行与自动保存这是最基础的情况。当你为图配置了 Checkpointer 并正常调用时，LangGraph 会自动在每一个 Super-step 结束后保存状态。3.2. 故障恢复与“待处理写入”(Pending Writes)当一个超级步中的部分节点成功、部分节点失败时。为了不重复执行已经成功的节点，LangGraph 会保存这些成功节点的写入操作作为“待处理写入”(Pending Writes)。当从该检查点恢复时，这些待处理的写入会被应用，而失败的节点会被重新执行。示例：在一个有“节点A”和“节点B”的超级步中，“节点A”成功写入了数据，但“节点B”崩溃了。Checkpoint 会记录下“节点A”的写入作为 pendingWrites。恢复执行时，系统会应用“节点A”的写入，并重新运行“节点B”。3.3. 时间旅行 (Time Travel)你可以回到过去的任何一个检查点，查看当时的状态，或从此处分叉。查看历史状态：通过 graph.getState(config) 获取最新状态，或通过 checkpointer.list(config) 列出某个线程的所有检查点。从检查点恢复/分叉：获取到特定的 checkpoint_id 后，可以在下一次调用时指定从该检查点恢复。示例：假设你的图执行了三个步骤，产生了 cp1, cp2, cp3 三个检查点。你发现 cp3 的结果不理想，想回到 cp2 时的状态重新尝试。你可以通过API获取 cp2 的ID，然后配置图从该检查点恢复执行。4. Checkpoint Saver 实现LangGraph 提供了多种 Checkpoint Saver，你可以根据场景选择：Saver 实现包名异步支持推荐场景InMemorySaverlanggraph-checkpoint是仅用于开发和测试SqliteSaverlanggraph-checkpoint-sqlite否轻量级演示和小型项目AsyncSqliteSaverlanggraph-checkpoint-sqlite是异步SQLite，不推荐用于生产PostgresSaverlanggraph-checkpoint-postgres否生产环境，需完整历史AsyncPostgresSaverlanggraph-checkpoint-postgres是异步生产环境此外，社区也提供了支持其他存储后端（如 DynamoDB, S3, MongoDB, Couchbase等）的实现。5. 代码示例5.1. 基本使用 (InMemorySaver)from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph # 1. 定义你的图 builder = StateGraph(YourState) # ... 添加节点和边 ... # 2. 创建 Checkpointer checkpointer = MemorySaver() # 3. 编译图时传入 Checkpointer graph = builder.compile(checkpointer=checkpointer) # 4. 执行图，必须指定 thread_id config = {"configurable": {"thread_id": "user_session_123"}} initial_state = {"messages": [("user", "你好！")]} # 图执行后，状态会自动保存 result = graph.invoke(initial_state, config)5.2. 生产环境 (PostgresSaver)from langgraph.checkpoint.postgres import PostgresSaver # 使用 PostgreSQL 作为持久化存储 with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db") as checkpointer: # 首次使用时需要执行 setup 创建表 checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) # ... 后续执行 ...5.3. 获取和恢复状态# 获取指定线程的最新状态 config = {"configurable": {"thread_id": "user_session_123"}} snapshot = await graph.aget_state(config) print(f"当前 checkpoint ID: {snapshot.checkpoint_id}") # 列出该线程的所有检查点 checkpoints = [] async for checkpoint in checkpointer.alist(config): checkpoints.append(checkpoint) # 从特定检查点恢复执行 (假设 checkpoint_id 为 'xyz') fork_config = { "configurable": { "thread_id": "user_session_123", "checkpoint_id": "xyz" # 指定要从哪个检查点恢复 } } # 后续的 invoke 会从该检查点继续6. 最佳实践与总结选择合适的 Checkpointer：开发测试用 MemorySaver，生产环境务必使用 PostgresSaver 等持久化方案。善用 thread_id：为每个独立的会话或工作流使用唯一的 thread_id 来隔离状态。数据序列化：Checkpoint 数据默认使用 JsonPlusSerializer 进行序列化。对于敏感数据，可以包装 EncryptedSerializer 进行加密。利用时间旅行调试：当 Agent 行为异常时，可以利用“时间旅行”功能回溯到历史检查点进行排查。
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://localhost:11434", temperature=0.7, ) class SimpleState(TypedDict): input_text:str processed_text :str llm_response :str final_output:str # 定义节点函数 def preprocess(state: SimpleState) -> dict: """预处理节点：清理和格式化输入""" text = state["input_text"].strip().lower() print(f" [预处理] 输入: '{state['input_text']}' -> '{text}'") return {"processed_text": text} def call_llm(state: SimpleState) -> dict: """LLM 节点：调用语言模型""" messages = [ {"role": "system", "content": "你是一个友好的助手，请简洁回答问题。"}, {"role": "user", "content": state["processed_text"]} ] response = model.invoke(messages) print(f" [LLM] 响应: {response.content[:50]}...") return {"llm_response": response.content} def postprocess(state: SimpleState) -> dict: """后处理节点：格式化输出""" final = f"✨ AI 回复：{state['llm_response']}" print(" [后处理] 完成格式化") return {"final_output": final} # 构建图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("preprocess", preprocess) graph.add_node("call_llm", call_llm) graph.add_node("postprocess", postprocess) # 添加边（定义执行顺序） graph.add_edge(START, "preprocess") graph.add_edge("preprocess", "call_llm") graph.add_edge("call_llm", "postprocess") graph.add_edge("postprocess", END) # 编译图 app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) result = app.invoke({"input_text": " 介绍一下你自己 "}) print(f"\n最终输出：\n{result['final_output']}") 条件分支工作流from typing import Literal #条件分支工作流 #加模型顺序执行 from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from typing import TypedDict, Annotated,Dict class AgentState(TypedDict): query: str query_type: str response: str def classify_query(state: AgentState) ->dict: """分类节点：判断查询类型""" query = state["query"].lower() if any(word in query for word in ["天气", "温度", "下雨"]): query_type = "weather" elif any(word in query for word in ["计算", "加", "减", "乘", "除", "等于"]): query_type = "math" else: query_type = "general" print(f" [分类] 查询类型: {query_type}") return {"query_type": query_type} def handle_weather(state: AgentState) -> dict: """处理天气查询""" print(" [天气处理] 执行天气查询逻辑...") # 实际应用中这里会调用天气 API response = "🌤️ 今天天气晴朗，温度 25°C，适合外出！" return {"response": response} def handle_math(state: AgentState) -> dict: """处理数学计算""" print(" [数学处理] 执行计算逻辑...") content = "计算器损坏，计算结果统一为 0。" return {"response": f"🔢 {content}"} def handle_general(state: AgentState) -> dict: """处理一般查询""" print(" [通用处理] 执行通用 LLM 调用...") messages = [ {"role": "system", "content": "你是一个友好的助手，请简洁回答问题。"}, {"role": "user", "content": state["query"]} ] result = model.invoke(messages) return {"response": f"💡 {result.content}"} def route_query(state: AgentState) -> Literal["weather", "math", "general"]: """路由函数：返回下一个节点名称""" return state["query_type"] # 构建图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("classify", classify_query) graph.add_node("weather", handle_weather) graph.add_node("math", handle_math) graph.add_node("general", handle_general) # 添加边 graph.add_edge(START, "classify") # 添加条件边 graph.add_conditional_edges( "classify", # 从哪个节点出发 route_query, # 路由函数 { # 路由映射 "weather": "weather", "math": "math", "general": "general" } ) # 所有处理节点都到 END graph.add_edge("weather", END) graph.add_edge("math", END) graph.add_edge("general", END) # 编译 app = graph.compile() from IPython.display import Image ,display display(Image(app.get_graph().draw_mermaid_png())) # 测试不同类型的查询 test_queries = [ "今天北京的天气怎么样？", "计算 123 加 456 等于多少？", "Python 是什么编程语言？" ] for query in test_queries: print(f"\n查询: {query}") result = app.invoke({"query": query}) print(f"响应: {result['response'][:100]}...") 带内存的对话工作流from typing import Literal from langgraph.graph.message import add_messages # 必须导入！ from langchain_core.messages import BaseMessage from langgraph.graph import StateGraph, START, END from langchain_ollama import ChatOllama from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated,Dict from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.checkpoint.memory import MemorySaver # 带内存的对话工作流 # 使用 add_messages 注解，消息会自动追加 class AgentState(TypedDict): messages: Annotated[list, add_messages] turn_count: int # model = ChatOllama( # model="qwen3.5:9b", # base_url="http://localhost:11434", # temperature=0.7, # # M5 专属优化 # num_ctx=1024, # 越小越快 # num_predict=256, # 限制回答长度 # use_metal=True, # ✅ M5 必开：调用GPU/NPU # # keep_alive="24h", # 模型常驻，不卸载 # ) model = ChatOpenAI( base_url="http://localhost:1234/v1", api_key="dummy-key", # 随便填 model="qwen/qwen3.5-9b", temperature=0.2, max_tokens=1024, # max_tokens=64, # ✅ 核心优化 2：关闭思考链（关键！） # 关键 2：强制关闭思考链（这个格式 100% 生效） # use_metal=True, # streaming=True, # 开启流式输出 # callbacks=[StreamingStdOutCallbackHandler()], # 实时打印 ) def chat_node(state: AgentState) -> dict: """对话节点：处理用户消息并生成回复""" # 添加系统提示（如果是第一轮） messages = state["messages"] if not any(isinstance(m, SystemMessage) for m in messages): messages = [ SystemMessage(content="你是友好助手，请直接回答问题，**绝对不要输出思考过程，不要写推理步骤**") ] + messages # 调用模型 response = model.invoke(messages) # 更新轮数 turn_count = state.get("turn_count", 0) + 1 print(f" [对话轮次 {turn_count}] AI: {response.content[:50]}...") # 返回新消息（会自动追加到 messages 列表） return { "messages": [response], # add_messages 会自动追加 "turn_count": turn_count } def should_continue(state: AgentState) -> Literal["continue", "end"]: """决定是否继续对话""" # 这里简化为总是返回 end，实际应用中可以检查用户意图 return "end" # 构建图 graph = StateGraph(AgentState) graph.add_node("chat", chat_node) graph.add_edge(START, "chat") graph.add_edge("chat", END) # 简化：每次调用处理一轮 # 使用内存保存器 memory = MemorySaver() app = graph.compile(checkpointer=memory) # 模拟多轮对话（使用相同的 thread_id） config = {"configurable": {"thread_id": "user_001"}} conversations = [ "你好！我叫小明。", "我最喜欢的编程语言是 Python。", "你还记得我的名字吗？", "我喜欢什么编程语言？" ] for user_input in conversations: print(f"\n用户: {user_input}") result = app.invoke( {"messages": [HumanMessage(content=user_input)]}, config=config ) print(f"AI: {result['messages'][-1].content}")
数据向量化#读取 现有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, ) # ========================================== # 2. 模拟你的知识库数据 (实际使用时请替换为你的向量库检索逻辑) # ========================================== # 这里为了演示，我直接把你的 JSON 数据放在这里，实际应替换为 vector_store.similarity_search(query) @tool def search_docs(query: str) -> str: """ 在文档库中搜索相关信息。 如果用户询问关于日期、格言或特定事件的内容，请使用此工具。 """ # --- 实际逻辑替换结束 --- formatted_results = [] results = ensemble_retriever.invoke(query) for doc in results: text = f"内容: {doc.page_content}" formatted_results.append(text) return "\n---\n".join(formatted_results) # ========================================== # 3. 创建 AI Agent # ========================================== agent = create_agent( model=model, tools=[search_docs], system_prompt="""你是每日一词的助手。 1. 回答用户问题前，必须先调用 search_docs 工具查找信息。 2. 只能根据搜索到的 [内容] 和 [备注] 进行回答。 3. 如果搜索不到相关信息，请直接说“我在日记中没找到相关内容”，不要编造。 4. 回答时直接输出工具查询的内容。 """ ) # ========================================== # 4. 问答循环 # ========================================== print("🤖 助手已就绪，请问关于这些日期的格言，你想知道什么？") while True: try: query = input("\n请输入你的问题 (输入 q 退出): ") if query.lower() == 'q': break # 调用 Agent # 注意：这里使用 invoke 传入字典 response = agent.invoke({ "messages": [{"role": "user", "content": query}] }) # 【关键优化】：正确提取回复内容 # response["messages"] 是一个列表，[-1] 是最后一条消息对象 # .content 是获取该对象中的文本内容 final_answer = response["messages"][-1].content print("\n📝 助手回答：") print(final_answer) print("-" * 30) except Exception as e: print(f"发生错误: {e}") 结果
加载文件进行分割向量化存储到 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": item["date"]} ) documents.append(doc) # 从文档列表创建 FAISS 向量库 vector_store = FAISS.from_documents(documents=documents, embedding=embeddings) # 【可选】持久化存储：将索引保存到本地磁盘 vector_store.save_local("faiss_index") 向量检索（语义搜索）fine_retriever = vector_store.as_retriever(search_kwargs={"k": 5}) final_docs = fine_retriever.invoke("光明") print(final_docs) BM25 检索（关键词搜索）from langchain_community.retrievers import BM25Retriever # 从分割好的文档创建BM25检索器（chunks和向量检索用的是同一个） bm25_retriever = BM25Retriever.from_documents(documents) bm25_retriever.k = 3 # 每次搜3个最相关的 # 实际查询 results = bm25_retriever.invoke("失败") print(results) # 你的搜索过程可能是这样的： # 你搜 "失败"。 # 如果你的 BM25 没有配置中文分词器（比如 jieba），它可能在找 "失败" 这个整体。 # 但文档里只有 "成功不会..."。 # 文档里没有 "失败" 这个词（只有“失”和“败”散落在不同地方，或者根本没有），所以 BM25 判定为 0 匹配。 # 它返回的那三个结果（潜力、冒险、成功），其实是因为它们包含了“我”、“你”、“的”、“是”这些高频通用词，在没有精准匹配的情况下，BM25 只能把这些“沾点边”的文档凑数返回给你。 print("-"*50) results = bm25_retriever.invoke("potential") print(results) # 英文： 句子是用空格隔开的。 # Success doesn't come to you -> 自动变成 ['Success', "doesn't", 'come', 'to', 'you']。BM25 很擅长处理这个。 怎么解决？（让 BM25 支持中文）import jieba # 定义一个专门切分中文的函数 def chinese_tokenizer(text): return list(jieba.cut(text)) # 创建检索器 bm25_retriever = BM25Retriever.from_documents( documents, # 关键在这里：告诉它用 jieba 来切分 preprocess_func=chinese_tokenizer ) bm25_retriever.k = 3 # 再次搜索 results = bm25_retriever.invoke("失败") print(results) RAG 进阶 - EnsembleRetriever 混合检索器（核心工具）from langchain_classic.retrievers import EnsembleRetriever # 创建混合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, fine_retriever], # 要组合的两个检索器 weights=[0.4, 0.6] # 各占50%权重（平衡配置） ) # 实际查询 query = "接受失败" results = ensemble_retriever.invoke(query) print(results) 实战 - 在线阶段：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, ) # ========================================== # 2. 模拟你的知识库数据 (实际使用时请替换为你的向量库检索逻辑) # ========================================== # 这里为了演示，我直接把你的 JSON 数据放在这里，实际应替换为 vector_store.similarity_search(query) @tool def search_docs(query: str) -> str: """ 在文档库中搜索相关信息。 如果用户询问关于日期、格言或特定事件的内容，请使用此工具。 """ # --- 实际逻辑替换结束 --- formatted_results = [] results = ensemble_retriever.invoke(query) for doc in results: text = f"内容: {doc.page_content}" formatted_results.append(text) return "\n---\n".join(formatted_results) # ========================================== # 3. 创建 AI Agent # ========================================== agent = create_agent( model=model, tools=[search_docs], system_prompt="""你是每日一词的助手。 1. 回答用户问题前，必须先调用 search_docs 工具查找信息。 2. 只能根据搜索到的 [内容] 和 [备注] 进行回答。 3. 如果搜索不到相关信息，请直接说“我在日记中没找到相关内容”，不要编造。 4. 回答时请尽量保持优雅和文学性，就像那些格言一样。 """ ) # ========================================== # 4. 问答循环 # ========================================== print("🤖 助手已就绪，请问关于这些日期的格言，你想知道什么？") while True: try: query = input("\n请输入你的问题 (输入 q 退出): ") if query.lower() == 'q': break # 调用 Agent # 注意：这里使用 invoke 传入字典 response = agent.invoke({ "messages": [{"role": "user", "content": query}] }) # 【关键优化】：正确提取回复内容 # response["messages"] 是一个列表，[-1] 是最后一条消息对象 # .content 是获取该对象中的文本内容 final_answer = response["messages"][-1].content print("\n📝 助手回答：") print(final_answer) print("-" * 30) except Exception as e: print(f"发生错误: {e}")
加载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, # "per_page": 3, # "total_pages": 5, # "total_items": 14, # "items": [ # { # "id": 101, # "title": "Understanding JSONLoader", # "content": "This article explains how to parse API responses...", # "author": { # "id": "user_1", # "name": "Alice" # }, # "created_at": "2023-10-05T08:12:33Z" # }, from langchain_community.document_loaders import JSONLoader file_path = 'asset/load/04-response.json' # 方式1： loader = JSONLoader( file_path=file_path, jq_schema=".data.items[].content", ) docs = loader.load() for doc in docs: print(f"文档内容:{doc.page_content}") # 方式2： loder2 = JSONLoader( file_path="asset/load/04-response.json", jq_schema=".data.items[]", # 先定位到数组条目 content_key=".content", # 再从条目中提取 content 字段 is_content_key_jq_parsable=True ) docs2 = loder2.load() print("-"*60) for doc in docs2: print(f"文档内容:{doc.page_content}") 深层提取json文本#提取04-response.json文件中嵌套在 data.items[] 里的 title、content 和 其文本 from langchain_community.document_loaders import JSONLoader from pprint import pprint loader = JSONLoader( file_path=file_path, # jq_schema=".data.items[] | {id, author, text: (.title + '\n' + .content)}", jq_schema=".data.items[]", content_key='(.title + "\n\n" + .content)', is_content_key_jq_parsable=True # 用jq解析content_key ) docs = loader.load() pprint(docs[0].page_content) #取出第二层的作者 loader2 = JSONLoader( file_path=file_path, jq_schema=".data.items[]", content_key='.author.name', is_content_key_jq_parsable=True ) docs2 = loader2.load() print("=== 取出第二层的作者 ===") print(docs2[0].page_content) 加载HTML# 加载HTML(了解) # 1.导入相关的依赖 from langchain_community.document_loaders import UnstructuredHTMLLoader # 2.定义UnstructuredHTMLLoader对象 # strategy: # "fast" 解析加载html文件速度是比较快（但可能丢失部分结构或元数据） # "hi_res": (高分辨率解析) 解析精准（速度慢一些） # "ocr_only" 强制使用ocr提取文本，仅仅适用于图像（对HTML无效） # mode ：one of `{'paged', 'elements', 'single'} # "elements" 按语义元素（标题、段落、列表、表格等）拆分成多个独立的小文档 html_loader = UnstructuredHTMLLoader( file_path="asset/load/05-load.html", mode="elements", strategy="fast" ) # 3.加载 docs = html_loader.load() print(len(docs)) # 16 # 4.打印 for doc in docs: print(doc) 加载Markdown# 2.6 加载Markdown(了解) # pip install markdown # pip install unstructured # 1.导入相关的依赖 from langchain_community.document_loaders import UnstructuredMarkdownLoader from pprint import pprint # 2.定义UnstructuredMarkdownLoader对象 md_loader = UnstructuredMarkdownLoader( file_path="asset/load/06-load.md", strategy="fast" ) # 3.加载 docs = md_loader.load() print(len(docs)) # 4.打印 for doc in docs: pprint(doc)
# 核心概念 # 验证和重试 = 确保 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_model.with_fallbacks( # fallbacks=[model1, model2], # 备用模型列表（按顺序尝试） # exceptions_to_handle=(Exception,) # 触发降级的异常 # ) # 工作流程： # 尝试 primary_model → 失败 # ↓ # 尝试 model1 → 失败 # ↓ # 尝试 model2 → 成功 ✓ # 3. Pydantic Field 约束 # 约束 用途 示例 # gt / ge 数值 > / >= Field(gt=0) # lt / le 数值 < / <= Field(le=100) # min_length / max_length 字符串长度 Field(min_length=2) # pattern 正则表达式 Field(pattern=r'^\d{11}$') # class Product(BaseModel): # name: str = Field(min_length=2, max_length=50) # price: float = Field(gt=0, description="价格必须 > 0") # stock: int = Field(ge=0, description="库存 >= 0") # 实际应用 # 1. LLM 输出验证 + 重试循环 # 当 LLM 输出不符合验证规则时，重新提示： def demo4(): from pydantic import ValidationError class Product(BaseModel): name: str = Field(min_length=2) price: float = Field(gt=0) structured_llm = chat_model.with_structured_output(Product) max_retries = 3 text = "产品价格是 -100 元" # 负价格会触发验证失败 for attempt in range(1, max_retries + 1): try: result = structured_llm.invoke(f"提取产品信息：{text}") # 验证通过 break except ValidationError as e: error_msg = e.errors()[0]["msg"] # 在提示中加入错误信息 text = f"{text}\n注意: {error_msg}。请确保价格 > 0" if attempt == max_retries: raise # 重试次数用完 # 第 1 次尝试: # LLM 输出: price = -100 # 验证: ✗ 失败 (price 必须 > 0) # 第 2 次尝试（修正提示）: # 提示: "...注意: price 必须 > 0" # LLM 输出: price = 100 # 验证: ✓ 通过
# 核心概念 # 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(data["age"], int): # raise ValueError("age must be int") # # 4. 手动创建对象 # person = Person(**data) # 结构化输出（简洁）： # 一步到位 # structured_llm = chat_model.with_structured_output(Person) # person = structured_llm.invoke("张三是一名 30 岁的软件工程师") # ✅ 自动解析、验证、创建对象 # 幕后流程 # 1. Pydantic 模型 → JSON Schema # Person → { # "type": "object", # "properties": { # "name": {"type": "string", "description": "姓名"}, # "age": {"type": "integer", "description": "年龄"} # } # } # 2. JSON Schema → LLM (函数调用) # LLM 被强制返回符合 schema 的 JSON # 3. JSON → Pydantic 对象 # 自动验证类型并创建 Person 实例 # 创建结构化输出的 LLM # 实际应用 # 1. 客户信息提取 class CustomerInfo(BaseModel): name: str = Field(description="客户姓名") phone: str = Field(description="电话号码") email: Optional[str] = Field(None, description="邮箱") issue: str = Field(description="问题描述") def demo2(): structured_llm = chat_model.with_structured_output(CustomerInfo) conversation = "客户: 我是李明，电话 138-1234-5678，问题是：订单没发货.仅返回JSON数据，不添加任何额外说明文字" info = structured_llm.invoke(f"提取客户信息：{conversation}") print(info) # return self.__pydantic_serializer__.to_python( # name='李明' phone='138-1234-5678' email=None issue='订单没发货' # 应用： # 自动填充 CRM 系统 # 工单自动分类 # 客服辅助 # 2. 产品评论分析 class Review(BaseModel): product: str rating: int = Field(description="评分 1-5") pros: List[str] = Field(description="优点列表") cons: List[str] = Field(description="缺点列表") def demo3(): structured_llm = chat_model.with_structured_output(Review) review = structured_llm.invoke( """ iPhone 15 很棒！摄像头强大，手感好。但是价格贵，没有充电器。4分。仅返回JSON数据，不添加任何额外说明文字 """ ) print(review) # return self.__pydantic_serializer__.to_python( # product='iPhone 15' rating=4 pros=['摄像头强大', '手感好'] cons=['价格贵', '没有充电器'] # 应用： # 批量处理用户评论 # 自动生成分析报告 # 发现产品改进点 # 3. 文档信息提取 嵌套模型 # 1. 定义子模型：商品明细 class InvoiceItem(BaseModel): name: str = Field(description="商品名称") amount: float = Field(description="商品单价或该项小计金额") # quantity: int = Field(default=1, description="数量，如果未提及默认为1") # 可选 # 2. 定义主模型：发票 class Invoice(BaseModel): invoice_number: str = Field(description="发票号") date: str = Field(description="日期，格式务必为 YYYY-MM-DD") total_amount: float = Field(description="发票总金额") items: List[InvoiceItem] = Field(description="商品明细列表，包含名称和价格") def demo4(): structured_llm = chat_model.with_structured_output(Invoice) document = """ 发票号：20230801001 日期：2023-08-01 总金额：1234.56 项目： 1. 商品A - 100.00 2. 商品B - 200.00 3. 服务费 - 934.56 """ # 优化后的 Prompt：更自然，不需要强调“仅返回JSON”，因为 structured_output 已经做了约束 prompt_text = f""" 请从以下文本中提取发票信息，并**以 JSON 格式返回**结果： {document} 注意： 1. 确保日期格式为 YYYY-MM-DD。 2. 将每个项目拆分为名称和对应的金额。 3. 如果总金额与各项之和不符，以文档中明确写出的"总金额"为准。 """ info: Invoice = structured_llm.invoke(prompt_text) print(info) # 应用： # 自动化财务处理 # OCR 后结构化 # 数据录入 # 最佳实践 # # 1. 使用清晰的字段描述 # class Good(BaseModel): # created_at: str = Field(description="创建时间，格式 YYYY-MM-DD") # class Bad(BaseModel): # created_at: str # 没有描述，LLM 可能格式错误 # # 2. 合理使用 Optional # class Good(BaseModel): # email: Optional[str] = None # 邮箱可能没有 # class Bad(BaseModel): # email: str # 强制必填，可能导致提取失败 # # 3. 使用枚举限制值 # class Good(BaseModel): # status: Status # 枚举 # class Bad(BaseModel): # status: str # 可能返回任意字符串 # # 4. 列表设置合理的描述 # class Good(BaseModel): # tags: List[str] = Field(description="产品标签，如 '电子产品', '手机'") # class Bad(BaseModel): # tags: List[str] # LLM 不知道该提取什么 # # 5. 嵌套模型保持简单 # class Good(BaseModel): # user: User # 1 层嵌套 # settings: dict # 复杂数据用 dict # class Bad(BaseModel): # user: User # company: Company # address: Address # country: Country # 4 层嵌套，容易出错 if __name__ == "__main__": # demo1() # demo2() # demo3() demo4()
""" 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.warning( "[Middleware] After Model Call | No messages found in state." ) return None last_message = messages[-1] msg_type = last_message.__class__.__name__ logger.info(f"[Middleware] After Model Call | Response Type: {msg_type}") # 如果是特定类型的消息，可以提取内容摘要记录 # if hasattr(last_message, 'content'): # logger.debug(f"Response Content Preview: {str(last_message.content)[:100]}...") return None # 不修改状态 # ============================================================================ # 示例 2：修改状态的中间件 # ============================================================================ class CallCounterMiddleware(AgentMiddleware): """ 计数中间件 - 统计模型调用次数 在中间件内部维护计数器（简单版本） """ def __init__(self): super().__init__() self.count = 0 # 简单计数器 def after_model(self, state, runtime): """模型响应后，增加计数""" self.count += 1 print(f"\n[计数器] 模型调用次数: {self.count}") return None # 不修改 state # ============================================================================ # 示例 3：消息修剪中间件 # ============================================================================ class MessageTrimmerMiddleware(AgentMiddleware): """ 消息修剪中间件 - 限制消息数量 before_model 修改消息列表 注意：需要配合无 checkpointer 使用，否则历史会被恢复 """ def __init__(self, max_messages=5): super().__init__() self.max_messages = max_messages self.trimmed_count = 0 # 统计修剪次数 def before_model(self, state, runtime): """模型调用前，修剪消息""" messages = state.get("messages", []) if len(messages) > self.max_messages: # 保留最近的 N 条消息 trimmed_messages = messages[-self.max_messages :] self.trimmed_count += 1 print( f"\n[修剪] 消息从 {len(messages)} 条减少到 {len(trimmed_messages)} 条 (第{self.trimmed_count}次修剪)" ) return {"messages": trimmed_messages} return None # ============================================================================ # 示例 4：输出验证中间件 # ============================================================================ class OutputValidationMiddleware(AgentMiddleware): """ 输出验证中间件 - 检查响应长度 after_model 验证输出 """ def __init__(self, max_length=100): super().__init__() self.max_length = max_length def after_model(self, state, runtime): """模型响应后，验证输出""" messages = state.get("messages", []) if not messages: return None last_message = messages[-1] content = getattr(last_message, "content", "") if len(content) > self.max_length: print( f"\n[警告] 响应���长 ({len(content)} 字符)，已截断到 {self.max_length}" ) # 这里可以实现截断或重试逻辑 return None # ============================================================================ # 示例 5：条件跳转（高级） # ============================================================================ class MaxCallsMiddleware(AgentMiddleware): """ 最大调用限制中间件 通过抛出异常来阻止模型调用（更可靠的方式） """ def __init__(self, max_calls=3): super().__init__() self.max_calls = max_calls self.count = 0 # 简单计数器 def before_model(self, state, runtime): """检查调用次数，超过限制则抛出异常""" if self.count >= self.max_calls: print(f"\n[限制] 已达到最大调用次数 {self.max_calls}，停止调用") # 抛出自定义异常来阻止继续执行 raise ValueError(f"已达到最大调用次数限制: {self.max_calls}") print(f"[限制] 当前调用次数: {self.count}/{self.max_calls}") return None def after_model(self, state, runtime): """增加计数""" self.count += 1 print("次数+1") return None # ============================================================================ # 示例 6：内置中间件使用 # ============================================================================ from langchain.agents.middleware import SummarizationMiddleware def demo1(): """ 示例 1：最简单的中间件 """ # 创建代理时添加中间件 agent = create_agent( model=chat_model, tools=[get_weather], checkpointer=InMemorySaver(), middleware=[ LoggingMiddleware(), CallCounterMiddleware(), MessageTrimmerMiddleware(), SummarizationMiddleware( # 内置中间件 model="groq:llama-3.3-70b-versatile", max_tokens_before_summary=200, # 超过 200 token 就摘要 ), ], # 添加日志中间件 ) # 2. 构造 config 对象 config = {"configurable": {"thread_id": "demo1"}} response = agent.invoke( {"messages": [{"role": "user", "content": "你好，今天北京的天气怎么样"}]}, config, ) print(f"Agent: {response['messages'][-1].content}") # 结果 # [中间件] before_model: 准备调用模型 # [中间件] 当前消息数: 1 # [中间件] after_model: 模型已响应 # [中间件] 响应类型: AIMessage # [中间件] before_model: 准备调用模型 # [中间件] 当前消息数: 3 # [中间件] after_model: 模型已响应 # [中间件] 响应类型: AIMessage # Agent: 今天北京的天气是晴天，温度 15°C，空气质量良好。 if __name__ == "__main__": demo1()
# 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 after_model(self, state, runtime): """ 在模型响应后执行 返回值： - None: 不修改状态 - dict: 更新状态 """ # 统计调用次数 count = state.get("call_count", 0) return {"call_count": count + 1} # 用途： # 输出验证 # 格式化响应 # 统计信息 # 状态更新 # 2. after_model: 模型调用后的“质检员” # 执行时机： # 在 LLM 返回生成结果（Completion），但尚未将该结果正式存入状态或返回给用户之后立即执行。 # 核心参数： # state: 包含模型刚刚生成的原始响应（通常在 messages 的最后一条，或者在单独的 response 字段中）。 # runtime: 运行时环境。 # (部分框架还会传入 response 对象，包含 token 用量、延迟等元数据) # 主要功能场景： # 1.输出格式化与清洗： # 移除模型生成的多余标记（如 Thought:, Action: 等中间思维链，如果不需要展示给用户）。 # 修复 JSON 格式错误（如果模型本应输出 JSON 但漏了括号）。 # 2.幻觉检测与事实核查 (Hallucination Check)： # 调用外部知识库或搜索工具，验证模型回答的事实准确性。如果不准确，可以标记为需要重试。 # 3.合规性审查 (Compliance)： # 检查模型输出是否包含偏见、泄露隐私或违反公司政策的内容。 # 4.结构化提取 (Structured Extraction)： # 如果模型输出了非结构化的文本，在此处将其解析为结构化数据（如提取订单号、日期），并存入 state 供后续节点使用。 # 5.自动重试逻辑 (Self-Correction)： # 如果检测到模型输出不符合预期（例如没有调用应有的工具），可以修改 state 并指示框架重新调用模型（通过返回特定的状态标志或抛出重试信号）。 # 返回值的作用 # 返回 None # def before_model(self, state, runtime): # print("日志记录") # return None # 不做任何修改，继续流程 # 返回字典（更新状态） # def after_model(self, state, runtime): # count = state.get("count", 0) # return {"count": count + 1} # 更新状态中的 count # 返回 jump_to（控制流程） # def before_model(self, state, runtime): # if state.get("count", 0) > 10: # return {"jump_to": "__end__"} # 跳过模型，直接结束 # return None # jump_to 目标： # "__end__" - 结束 Agent # "tools" - 跳到工具节点 # 其他自定义节点 # 执行顺序（重要！） # agent = create_agent( # model=model, # middleware=[Middleware1(), Middleware2(), Middleware3()] # ) # 执行流程： # 1. Middleware1.before_model ↓ 正序 # 2. Middleware2.before_model ↓ # 3. Middleware3.before_model ↓ # [模型调用] # 6. Middleware3.after_model ↑ 逆序 # 5. Middleware2.after_model ↑ # 4. Middleware1.after_model ↑ # 类似洋葱模型：外层先进后出 # 用户请求 # ↓ # [ Middleware1.before ] <-- 外层入口 # ↓ # [ Middleware2.before ] # ↓ # [ Middleware3.before ] <-- 内层入口 # ↓ # [ 🤖 模型调用 (Model) ] # ↓ # [ Middleware3.after ] <-- 内层出口 (最先处理模型结果) # ↑ # [ Middleware2.after ] # ↑ # [ Middleware1.after ] <-- 外层出口 (最后处理结果) # ↑ # 返回给用户 # 实际应用 class TestMiddleware: # before_model 示例：添加情感安抚指令 def before_model(self, state, runtime): last_msg = state["messages"][-1] # 验证用户身份：如果用户不是授权用户，直接返回跳转指令 if last_msg["role"] != "user": return {"jump_to": "unauthorized_node"} # 安全检查：如果包含攻击性词汇，直接返回跳转指令 if "混蛋" in last_msg["content"]: return {"jump_to": "safety_block_node"} # 2. 动态注入：如果检测到用户很着急，注入安抚指令 if "着急" in last_msg["content"]: system_prompt = { "role": "system", "content": "【动态指令】用户非常着急，请优先安抚情绪并快速给出方案。", } # 插入到 system prompt 之后，历史记录之前 new_messages = [state["messages"][0], system_prompt] + state["messages"][1:] return {"messages": new_messages} return None # 正常流程 # after_model 示例：确保输出包含工单号 def after_model(self, state, runtime): # 获取模型最后生成的消息 response_content = state["messages"][-1].content if "投诉" in state.get("intent", ""): if "工单号" not in response_content: response_content += " 请包含工单号，方便我们定位问题。" return None # 接受输出 # 1. 日志中间件 class LogMiddleware: def before_model(self, state, runtime): print("日志记录：开始模型调用") return None # 不做任何修改，继续流程 def after_model(self, state, runtime): print("日志记录：模型调用结束") return None # 不做任何修改，继续流程 # 2. 计数中间件 class CallCounterMiddleware(AgentMiddleware): def after_model(self, state, runtime): count = state.get("model_call_count", 0) return {"model_call_count": count + 1} # # 需要 checkpointer 来保存自定义状态 # agent = create_agent( # model=model, # middleware=[CallCounterMiddleware()], # checkpointer=InMemorySaver() # ) # 3. 消息修剪中间件 class MessageTrimmerMiddleware(AgentMiddleware): def __init__(self, max_messages=5): super().__init__() self.max_messages = max_messages def before_model(self, state, runtime): messages = state.get("messages", []) if len(messages) > self.max_messages: # 只保留最近的 N 条消息 return {"messages": messages[-self.max_messages :]} return None # 4. 输出验证中间件 class OutputValidatorMiddleware(AgentMiddleware): def after_model(self, state, runtime): messages = state["messages"] last_msg = messages[-1] if last_msg["role"] == "assistant": # 验证模型输出是否符合要求 if "工单号" not in last_msg["content"]: # 触发跳转指令 return {"jump_to": "missing_order_id_node"} return None # 5. 限流中间件 class MaxCallsMiddleware(AgentMiddleware): def __init__(self, max_calls=10): super().__init__() self.max_calls = max_calls def before_model(self, state, runtime): count = state.get("model_call_count", 0) if count >= self.max_calls: # 触发跳转指令 return {"jump_to": "rate_limit_node"} return None def after_model(self, state, runtime): count = state.get("call_count", 0) return {"call_count": count + 1} # 内置中间件 # SummarizationMiddleware（自动摘要） from langchain.agents.middleware import SummarizationMiddleware def demo1(): agent = create_agent( model=chat_model, middleware=[ SummarizationMiddleware( model="groq:llama-3.1-8b-instant", # 可用便宜模型 max_tokens_before_summary=500, ) ], checkpointer=InMemorySaver(), ) # 作用： # 消息超过 token 限制时自动摘要 # 保留最近消息 + 旧消息摘要 # 详见 08_context_management 章节 # HumanInTheLoopMiddleware（人工审核） from langchain.agents.middleware import HumanInTheLoopMiddleware def demo2(): agent = create_agent( model=chat_model, middleware=[ HumanInTheLoopMiddleware( interrupt_on={"send_email": True} # 调用此工具前暂停 ) ], checkpointer=InMemorySaver(), ) # PIIMiddleware（敏感信息处理） from langchain.agents.middleware import PIIMiddleware def demo3(): agent = create_agent( model=chat_model, middleware=[ PIIMiddleware("email", strategy="redact"), # 邮箱脱敏 PIIMiddleware("phone_number", strategy="block"), # 电话拦截 # 输出阶段：防止 LLM 幻觉生成假的或泄露真的身份证号 PIIMiddleware("id_number", strategy="redact", direction="output"), ], ) # graph LR # User[用户输入] --> PII_Check{PIIMiddleware 检查} # PII_Check -- 发现 Email --> Redact[执行脱敏 redact] # Redact --> LLM[发送给大模型] # PII_Check -- 发现 手机号 --> Block[执行拦截 block] # Block --> Error[返回错误/拒绝服务] # PII_Check -- 无敏感信息 --> LLM # 常见问题 # 不能直接访问。before_model 和 after_model 只在模型节点执行。 # 如果需要拦截工具调用，使用 wrap_tool_call（高级特性）。 # 2. 多个中间件的顺序重要吗？ # 非常重要！ # middleware=[ # TrimmerMiddleware(), # 1. 先修剪消息 # SummarizationMiddleware(), # 2. 再摘要 # LoggingMiddleware() # 3. 最后记录日志 # ] # before_model 按列表顺序执行 # after_model 按列表逆序执行 # 3. 修改状态需要 checkpointer 吗？ # 自定义状态需要，messages 不需要： # # 不需要 checkpointer（messages 自动保存） # def after_model(self, state, runtime): # return {"messages": [...]} # # 需要 checkpointer（自定义字段） # def after_model(self, state, runtime): # return {"my_custom_field": 123} # 4. 能在中间件里调用另一个模型吗 # 可以，但要小心： # class ValidationMiddleware(AgentMiddleware): # def __init__(self): # self.validator_model = init_chat_model(...) # def after_model(self, state, runtime): # # 用另一个模型验证输出 # last_msg = state['messages'][-1] # validation_result = self.validator_model.invoke(...) # return None # 最佳实践 # # 1. 生产环境推荐配置 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # MessageTrimmerMiddleware(max_messages=20), # 限制消息数 # SummarizationMiddleware(model=..., max_tokens=2000), # 自动摘要 # LoggingMiddleware(), # 日志记录 # ], # checkpointer=SqliteSaver.from_conn_string("...") # ) # # 2. 开发环境 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # LoggingMiddleware(), # 只要日志 # ] # ) # # 3. 测试环境 # agent = create_agent( # model=model, # tools=[...], # middleware=[ # MaxCallsMiddleware(max_calls=5), # 防止测试费用爆炸 # ] # ) if __name__ == "__main__": demo1()
""" 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.invoke( {"messages": [{"role": "user", "content": "你好,我是谁"}]}, config ) print(reponse["messages"][-1].content) # ============================================================================ # 带工具的持久化 Agent # ============================================================================ @tool def get_order_status(order_id: str) -> str: """查询订单状态""" orders = {"12345": "已发货，预计明天送达", "67890": "配送中，今天下午送达"} return orders.get(order_id, "订单不存在") """ 示例5：工具调用 + 持久化 Agent 记住工具调用历史 """ def example_3_tools_with_persistence(): """ 示例5：工具调用 + 持久化 Agent 记住工具调用历史 """ print("\n" + "=" * 70) print("示例 5：工具调用 + 持久化") print("=" * 70) db_path = "tools.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[get_order_status], checkpointer=checkpointer ) config = {"configurable": {"thread_id": "customer_001"}} print("\n第一轮：查询订单") print("客户: 查询订单 12345 的状态") response1 = agent.invoke( {"messages": [{"role": "user", "content": "查询订单 12345 的状态"}]}, config=config, ) print(f"Agent: {response1['messages'][-1].content}") print("\n第二轮：询问之前的查询结果") print("客户: 我的订单什么时候到？") response2 = agent.invoke( {"messages": [{"role": "user", "content": "我的订单什么时候到？"}]}, config=config, ) print(f"Agent: {response2['messages'][-1].content}") print("\n关键点：") print(" - Agent 记住了订单 12345 的查询结果") print(" - 工具调用历史也被持久化") print(" - 无需重复调用工具") # ============================================================================ # 实际应用 - 客服系统 # ============================================================================ def example_4_customer_service(): """ 示例6：实际应用 - 持久化客服系统 场景：客户可能分多次咨询，需要记住历史 """ print("\n" + "=" * 70) print("示例 6：实际应用 - 持久化客服系统") print("=" * 70) db_path = "customer_service.sqlite" with SqliteSaver.from_conn_string(db_path) as checkpointer: agent = create_agent( model=chat_model, tools=[get_order_status], system_prompt="""你是客服助手。 特点： - 记住客户之前的咨询 - 友好、耐心 - 使用工具查询订单""", checkpointer=checkpointer, ) customer_id = "customer_zhang" config = {"configurable": {"thread_id": customer_id}} print("\n第一次咨询（今天上午）：") conversations_morning = ["你好，我想查询订单", "订单号是 12345"] for msg in conversations_morning: print(f"\n客户: {msg}") response = agent.invoke( {"messages": [{"role": "user", "content": msg}]}, config=config ) print(f"客服: {response['messages'][-1].content}") print("\n" + "-" * 70) print("[几个小时后...]") print("-" * 70) print("\n第二次咨询（今天下午）：") print("\n客户: 我的订单到哪了？") response = agent.invoke( {"messages": [{"role": "user", "content": "我的订单到哪了？"}]}, config=config, ) print(f"客服: {response['messages'][-1].content}") print("\n关键点：") print(" - 客户无需重复订单号") print(" - 系统记住了上午的咨询") print(" - 即使客服系统重启也不影响") print(" - 生产级应用的标准做法") # ============================================================================ # SqliteSaver 参数说明 # ============================================================================ # SqliteSaver 创建方式： # 1. from_conn_string + with 语句（推荐） # with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: # agent = create_agent(model=model, checkpointer=checkpointer) # agent.invoke(...) # - 自动管理连接和资源 # - 支持相对路径和绝对路径 # - 最简单安全的方式 # - 确保正确释放数据库连接 # - 注意：直接传文件路径，不要加 sqlite:/// 前缀 # 2. 使用 sqlite3.connect（高级） # import sqlite3 # conn = sqlite3.connect("checkpoints.sqlite") # checkpointer = SqliteSaver(conn) # - 需要手动管理连接 # - 适合需要自定义连接参数的场景 # 数据库文件路径： # - 相对路径：checkpoints.sqlite（当前目录） # - 绝对路径：C:/Users/xxx/data/checkpoints.sqlite（Windows） # - 内存数据库：:memory:（测试用，程序退出即丢失） # 最佳实践： # ✅ 始终使用 with 语句管理 SqliteSaver # ✅ 直接传文件路径，不要加 sqlite:/// 前缀 # ✅ 生产环境：使用绝对路径 # ✅ 开发测试：使用相对路径 # ✅ 单元测试：使用 :memory: # ✅ 定期备份数据库文件 # """ if __name__ == "__main__": example_sqlite_saver() example_sqlite_saver_2()