乘风的人 - 大模型代码
type: channel
link: https://blog.zhangbiechen.cn/index.php/category/llm-code/
items: langgraph 图的 checkpoint 机制
items: 贷款审批业务 LangGraph 示例图
items: LangGraph 的 常见状态图（StateGraph）
items: llamaindex rag示例项目
items: Code - LangGraph 基础
items: code - 本地智能问答助手(Ollama + RAG)
items: RAG Advanced (RAG 进阶) - code.py
items: Mac 使用transformers 库运行Qwen3-Embedding-0.6B模型使用全记录
items: RAG - 文档加载器
items: Validation & Retry.py

langgraph 图的 checkpoint 机制
type: item
date: 2026-07-10T10:13:43+08:00
description: 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) # ...…

贷款审批业务 LangGraph 示例图
type: item
date: 2026-07-09T14:59:00+08:00
description: 📊 核心流程图（Mermaid 示意）核心在于 review 审核不通过时会指向 revise（修改），而 revise 执行完又会指回 review，形成闭环。graph TD; START([开始]) --> apply[提交贷款申请]; apply --> review[风控审核]; review -->|"✅ 批准"| approve[放款成功] --> END([结束]); review -->|"❌ 拒绝（评分过低）"| reject[直接拒绝] --> END; %% 核心循环：返回上一层（退回修改） review -->|"🔄 补充材料/信息有误"| revise[申请人修改补充]; revise -->|"修改完成，再次提交"| review; style revise fill:#f9f,stroke:#333,stroke-width:2px; style review fill:#bbf,stroke:#333,stroke-width:2px;💻 完整可运行代码（带循环返回机制）这个代码模拟了贷款审批的完整闭环，你复制运行就能看到状态如何在 审核 和 修改 之间反复横跳。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 定义状态（记录贷款信息和循环次数） class LoanState(TypedDict): applicant: str income: int # 月收入（万） credit_score: int # 信用分 (0-100) missing_docs: str # 缺失材料描述 attempts: int # 当前是第几次提交审核 status: str # "pending", "approved", "rejected" # 2. 节点 A：提交申请（初始化或修改后重新提交） def submit_application(state: LoanState): print(f"📄 [{state['applicant']}] 提交申请，收入={state['income']}万，信用分={state['credit_score']}") # 每次提交，尝试次数 +1 return {"attempts": state.get("attempts", 0) + 1} # 3. 节点 B：风控审核（核心判断节点） def review_application(state: LoanState): income = state["income"] score = state["credit_score"] attempts = state["attempts"] missing = state.get("missing_docs", "") print(f"🔍 第 {attempts} 次审核 {state['applicant']}...") # 业务规则判断 if score < 50: return {"status": "rejected"} # 信用太差，直接拒绝，不循环 if income < 5: # 【核心循环触发点】：收入证明不足，退回修改 return { "status": "pending", "missing_docs": "请补充近6个月银行流水（当前收入低于5万）" } if missing != "": # 如果之前有缺失材料，审核再次发现没补全，继续退回 return { "status": "pending", "missing_docs": "材料仍未补全，请重新上传身份证复印件" } # 全部通过 return {"status": "approved"} # 4. 节点 C：修改补充材料（返回上一层后的处理动作） def revise_documents(state: LoanState): print(f"✏️ {state['applicant']} 正在修改材料：{state['missing_docs']}") # 模拟申请人修改：提高收入（编造数据模拟修正），并清空缺失提示 # 注意：这里修改状态，使得再次进入 review 时能通过 return { "income": state["income"] + 3, # 补交材料证明实际收入更高 "missing_docs": "", # 清空错误标记 "attempts" : state["attempts"] + 1 } # 5. 【核心】条件路由…

LangGraph 的 常见状态图（StateGraph）
type: item
date: 2026-07-08T11:52:57+08:00
description: 示例 1：基础顺序链（Linear Chain）像工厂流水线，A -> B -> C 一步步往下走。📊 Mermaid 流程图graph TD; START([开始]) --> clean[清洗数据]; clean --> format[格式化输出]; format --> END([结束]);🖥️ 运行生成 ASCII 图from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): text: str def clean(state): return {"text": state["text"].strip()} def format(state): return {"text": f"Result: {state['text']}"} builder = StateGraph(State) builder.add_node("clean", clean).add_node("format", format) builder.add_edge(START, "clean").add_edge("clean", "format").add_edge("format", END) graph = builder.compile() # 打印控制台流程图 print(graph.get_graph().draw_ascii()) # 输出： # +-----------+ +----------+ +------+ # | __start__ | --> | clean | --> | format | --> | __end__ | # +-----------+ +----------+ +------+ # 生成 Mermaid 代码（可复制到支持 Mermaid 的工具查看） print(graph.get_graph().draw_mermaid())示例 2：条件路由（Conditional Branching）类似“智能分拣”，根据业务规则（如关键词、分数）走不同的分支。📊 Mermaid 流程图graph TD; START([开始]) --> classify[分类器]; classify -->|"类别: 数学"| math_node[数学处理器]; classify -->|"类别: 闲聊"| chat_node[闲聊回复]; math_node --> END([结束]); chat_node --> END([结束]);🖥️ 完整带图代码from typing import Literal, TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): query: str category: str def classify(state): cat = "math" if "计算" in state["query"] else "chat" return {"category": cat} def do_math(state): return {"response": "数学答案: 42"} def do_chat(state): return {"response": "闲聊: 今天天气不错"} def route_condition(state) -> Literal["math_node", "chat_node"]: return "math_node" if state["category"] == "math" else "chat_node" builder = StateGraph(State) builder.add_node("classify", classify) builder.add_node("math_node", do_math) builder.add_node("chat_node", do_chat) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route_condition) # 条件分支 builder.add_edge("math_node", END) builder.add_edge("chat_node", END) graph = builder.compile() # 打印出带有分支箭头的 ASCII 图 print(graph.get_graph().draw_ascii()) # 输出…

llamaindex rag示例项目
type: item
date: 2026-05-25T15:15:00+08:00
description: 项目结构llamaindex-rag-example/├── data/ # 存放待处理的原始文档（如PDF/TXT/Markdown等）├── storage_exp/ # 持久化存储向量索引的目录（自动生成）├── 01_build.py # 构建并持久化向量索引的脚本└── 02_query.py # 加载索引并进行问答查询的脚本PDF文件转Markdown#如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整 官网：https://mineru.net/ 01_build.py 构建索引import os import time import torch from pathlib import Path from llama_index.core import VectorStoreIndex, Settings, Document from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core.node_parser import SimpleNodeParser # 配置 device = "cpu" #如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整https://mineru.net/ markdown_path = "./data/test_manual.md" persist_dir = "./storage_exp" # 创建目录 os.makedirs(persist_dir, exist_ok=True) os.makedirs("./models", exist_ok=True) # 2. 配置全局 Embedding（本地 Qwen3 模型） EMBEDDING_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" Settings.embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_PATH, trust_remote_code=True ) # 读取 Markdown 文件 print(f"读取 Markdown: {markdown_path}") with open(markdown_path, "r", encoding="utf-8") as f: markdown_content = f.read() print(f"文档长度: {len(markdown_content)} 字符") # 创建 Document doc = Document( text=markdown_content, metadata={ "source": Path(markdown_path).name, "file_path": str(Path(markdown_path).absolute()), } ) # 切分文档 print("切分文档...") parser = SimpleNodeParser(chunk_size=512, chunk_overlap=50) nodes = parser.get_nodes_from_documents([doc]) print(f"生成 {len(nodes)} 个节点") # 构建索引 print("构建索引...") start_time = time.time() index = VectorStoreIndex(nodes) build_time = time.time() - start_time print(f"索引构建完成，耗时 {build_time:.2f} 秒") # 保存索引 print(f"保存索引到 {persist_dir}...") index.storage_context.persist(persist_dir=persist_dir) # 统计 index_size = 0 for root, dirs, files in os.walk(persist_dir): for file in files: index_size += os.path.getsize(os.path.join(root, file)) print("\n" + …

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…

Mac 使用transformers 库运行Qwen3-Embedding-0.6B模型使用全记录
type: item
date: 2026-03-30T14:10:00+08:00
description: Mac 使用transformers 库运行本地大模型使用全记录下载大模型到本地//创建本地运行虚拟环境 conda create --name vllm_dev python=3.12 conda activate vllm_dev //刚开始想用 vllm 运行模型，但是发现这个模型很小，改用transformers库直接运行，更省资源 //huggingface下载Qwen3-Embedding-0.6B模型 pip install huggingface_hub export HF_ENDPOINT=https://hf-mirror.com hf download Qwen/Qwen3-Embedding-0.6B >Fetching 12 files: 100%|████████████████████████████████████████████| 12/12 [01:14<00:00, 6.20s/it] Download complete: : 1.21GB [01:14, 19.4MB/s] /Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418 本地运行Qwen3-Embedding-0.6B模型//安装相关包pip install torch transformers fastapi uvicorn tiktoken sentencepiece //包解释torch —— 模型推理 transformers —— 加载 Qwen3 模型 fastapi —— API 服务 uvicorn —— 运行服务器 tiktoken —— Qwen3 分词器必须 sentencepiece —— 兼容依赖 开启本地服务emb_serve.py 启动服务：❯ python emb_server.py 🚀 当前计算设备: MPS 📥 正在加载模型: /Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418 ... `torch_dtype` is deprecated! Use `dtype` instead! Loading weights: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 310/310 [00:00<00:00, 359.07it/s] ✅ 模型加载完成！服务准备就绪。 INFO: Started server process [7296] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8001 (Press CTRL+C to quit) INFO: 127.0.0.1:53982 - "GET /docs HTTP/1.1" 200 OK INFO: 127.0.0.1:53983 - "GET /redoc HTTP/1.1" 200 OK INFO: 127.0.0.1:54022 - "GET /openapi.json HTTP/1.1" 200 OK INFO: 127.0.0.1:54121 - "POST /v1/embeddings HTTP/1.1" 200 OK emb_server.pyimport torch import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Union, Optional from transformers import AutoTokenizer, AutoModel # ========================================== # 1. 配置区域 (请在此处修改本地模型路径) # ==…

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…