《实用笔记》：构建生产级本地RAG管道——完全免费，无需……
《实用笔记》操作指南：构建生产级本地RAG管道——完全免费，无需合同、审核流程，还为采用该方案的团队提供可直接使用的代码模板。
本指南将逐步构建从原材料到可运行系统的完整流程，主题为：打造生产级本地RAG处理管道——100%免费且无需使用云服务。重点在于可操作的步骤、明确的检查点，以及可直接放入代码仓库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏的状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。
为何这很重要
在完成“为何重要”这一阶段时，首先写下相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤出错时，错误应指向单一的责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难解决检索效果不佳的问题。
完整的技术栈
在完成“完整技术栈”阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。
先决条件
在完成前置条件阶段时，首先需写明接口规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词很难改善较差的检索效果。 在完成前置条件阶段时，首先需写明接口规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。
第一部分 — 环境搭建
将第一阶段的“环境搭建”视为可度量的工作面最为有效。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任模块，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
步骤1：创建项目
第一步“创建阶段”若被视为可度量的工作面，效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
mkdir local-rag
cd local-rag
uv init
第二步：创建并激活虚拟环境
将“第2步：创建并部署”视为可度量的工作面时，其效果最佳。在扩大范围之前，需记录一份理想的处理结果、一个故障案例以及回滚说明。 在功能测试结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应强制要求重新编写另一个。
uv venv
.venv\Scripts\activate
将“第2步：创建并部署”视为可度量的工作面时，其效果最佳。在扩大范围之前，需记录一份理想的处理结果、一个故障案例以及回滚说明。 需同时记录正常流程和恢复流程的相关文档。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
第3步：安装所有依赖项
在第三步“安装所有阶段”中，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任模块，而非复杂的流程链。必须引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是虚假信息还是索引缺失所致。
uv add google-genai pypdf chromadb rich python-dotenv huggingface_hub fpdf2
第四步：创建项目结构
在第四步“创建阶段”中，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。
mkdir pdfs
mkdir pdfs\versions
type nul > local_rag.ipynb
type nul > .env
type nul > .gitignore
local-rag/
├── .venv/ ← virtual environment (never commit)
├── pdfs/ ← drop your PDFs here
│ └── versions/ ← test PDFs for CDC testing
├── chroma_db/ ← auto-created on first ingest
├── memory_checkpoints/ ← auto-created on first memory session
├── staleness_registry.json ← auto-created
├── chunk_registry.json ← auto-created
├── local_rag.ipynb ← your notebook
├── .env ← API keys (never commit)
├── .gitignore
└── pyproject.toml
第五步：配置API密钥
在第五步“配置API”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。必须注明支撑答案的具体内容，若没有引用依据，操作人员就无法区分是虚假信息还是索引缺失导致的错误。在第五步“配置API”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。需同时记录正常流程和异常恢复流程，重试机制、人工审核环节以及错误处理方式都是产品本身的组成部分，而非后续需要补充的内容。
GEMINI_API_KEY=your_gemini_key_here
HF_API_KEY=your_huggingface_token_here
.env
chroma_db/
memory_checkpoints/
staleness_registry.json
chunk_registry.json
__pycache__/
.venv/
*.pyc
第6步：配置VS Code
在执行第6步的配置阶段时，首先需明确相关要求：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某一步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
Ctrl+Shift+P → Python: Select Interpreter → .venv\Scripts\python.exe
第二部分 —— 核心流程详解
在处理第二部分核心流程阶段时，首先需明确合同条款：所需的输入参数、成功信号以及出现部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
单元1 — 依赖项（位于笔记本内）
在处理阶段内的单元1依赖关系时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词往往无法改善较差的检索效果。
# Run once inside the notebook if uv add was not used externally
# %pip install google-genai pypdf chromadb rich python-dotenv huggingface_hub fpdf2
在处理阶段内的单元1依赖关系时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。
单元2——配置
将“单元2配置”阶段视为可测量的界面来处理效果最佳。在扩大范围之前，先记录一份理想的转录结果、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一责任主体，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
import os
from dotenv import load_dotenv
load_dotenv()
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
EMBED_DIM = 768
HF_API_KEY = os.environ.get("HF_API_KEY", "")
GEMINI_EMBED_MODEL = "gemini-embedding-001"
HF_LLM_MODEL = "openai/gpt-oss-20b:groq"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "local_rag"
CHUNK_SIZE = 800
CHUNK_OVERLAP = 120
TOP_K = 5
EMBED_BATCH_SIZE = 50
BATCH_SLEEP_SEC = 0.3
LLM_MAX_NEW_TOKENS = 1024
LLM_TEMPERATURE = 0.1
assert GEMINI_API_KEY, "❌ GEMINI_API_KEY not set"
assert HF_API_KEY, "❌ HF_API_KEY not set"
单元3——PDF文本提取
将“Cell 3 PDF文本处理阶段”视为可度量的工作面时，其效果最佳。在扩大范围之前，先记录一份理想的转录结果、一个失败案例以及回滚说明。把这一阶段视为输入与经过验证的输出之间的契约：为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。应将分块策略与检索策略分开处理，当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
from pypdf import PdfReader
def extract_text_from_pdf(pdf_path: str) -> tuple[str, int]:
reader = PdfReader(pdf_path)
pages = []
for i, page in enumerate(reader.pages):
text = page.extract_text()
if text and text.strip():
pages.append(f"[Page {i + 1}]\n{text.strip()}")
return "\n\n".join(pages), len(reader.pages)
Cell 4 — 滑动窗口分块
将“Cell 4滑动窗口”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想的处理结果、一个故障案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一个不应强制要求重新编写另一个。
def chunk_text(text: str) -> list[str]:
chunks, start = [], 0
while start < len(text):
end = start + CHUNK_SIZE
chunk = text[start:end].strip()
if len(chunk) >= 80:
chunks.append(chunk)
start += CHUNK_SIZE - CHUNK_OVERLAP
return chunks
将“Cell 4滑动窗口”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想的处理结果、一个故障案例以及回滚说明。 需同时记录正常处理路径和故障恢复路径。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。
Cell 5 — Gemini嵌入模型
在 Cell 5 Gemini 嵌入阶段，修改代码之前需先明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。 将客户端构建逻辑与消息处理循环分开，这样即便更换提供方，也无需重写对话状态机。
from google import genai
from google.genai import types
genai_client = genai.Client(api_key=GEMINI_API_KEY)
def embed_documents_batch(chunks: list[str]) -> list[list[float]]:
all_embeddings = []
for i, chunk in enumerate(chunks, 1):
result = genai_client.models.embed_content(
model=GEMINI_EMBED_MODEL,
contents=chunk,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBED_DIM,
),
)
all_embeddings.append(result.embeddings[0].values)
return all_embeddings
def embed_query(text: str) -> list[float]:
result = genai_client.models.embed_content(
model=GEMINI_EMBED_MODEL,
contents=text,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_QUERY",
output_dimensionality=EMBED_DIM,
),
)
return result.embeddings[0].values
Cell 6 — ChromaDB 存储与检索
对于 Cell 6 的 ChromaDB 存储阶段，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并杜绝无声的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失问题。
import uuid, chromadb
def get_collection():
client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
return client.get_or_create_collection(
name=COLLECTION_NAME,
metadata={"hnsw:space": "cosine"},
)
def store_in_chroma(chunks, embeddings, doc_name):
collection = get_collection()
ids = [str(uuid.uuid4()) for _ in chunks]
metadatas = [{"source": doc_name, "chunk_index": i}
for i in range(len(chunks))]
collection.add(ids=ids, embeddings=embeddings,
documents=chunks, metadatas=metadatas)
return len(chunks)
def retrieve_context(query: str) -> list[dict]:
collection = get_collection()
query_embedding = embed_query(query)
results = collection.query(
query_embeddings=[query_embedding],
n_results=TOP_K,
include=["documents", "metadatas", "distances"],
)
chunks = []
for doc, meta, dist in zip(results["documents"][0],
results["metadatas"][0],
results["distances"][0]):
chunks.append({
"text": doc,
"source": meta.get("source", "unknown"),
"chunk_index": meta.get("chunk_index", -1),
"score": round(1 - dist, 4),
})
return sorted(chunks, key=lambda x: x["score"], reverse=True)
Cell 7 — 执行数据摄取
在“单元7运行数据摄取”阶段，修改代码之前需明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。需注明实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。
PDF_PATH = "./pdfs/attention.pdf"
raw_text, page_count = extract_text_from_pdf(PDF_PATH)
chunks = chunk_text(raw_text)
embeddings = embed_documents_batch(chunks)
stored = store_in_chroma(chunks, embeddings,
os.path.basename(PDF_PATH))
在Cell 7的运行摄入阶段，修改代码之前需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
Cell 8 — LLM：通过HuggingFace使用的gpt-oss-20b模型
在处理 Cell 8 LLM gpt-oss-20b 阶段时，首先写下相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一责任点，而非复杂的流程链。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。
from huggingface_hub import InferenceClient
hf_client = InferenceClient(api_key=HF_API_KEY)
def build_messages(query: str, context_chunks: list[dict]) -> list[dict]:
context_str = "\n\n---\n\n".join([
f"[Source: {c['source']} | Chunk #{c['chunk_index']} | "
f"Relevance: {c['score']}]\n{c['text']}"
for c in context_chunks
])
return [
{"role": "system", "content": SYSTEM_MSG},
{"role": "user", "content":
f"CONTEXT:\n{context_str}\n\nQUESTION:\n{query}"},
]
def generate_answer(messages: list[dict]) -> str:
completion = hf_client.chat.completions.create(
model=HF_LLM_MODEL,
messages=messages,
max_tokens=LLM_MAX_NEW_TOKENS,
temperature=LLM_TEMPERATURE,
)
return completion.choices[0].message.content.strip()
Cell 9–11 — ask() 流程与 REPL
在处理“单元9 11：阶段”时，首先写下契约内容：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。
query → embed_query() → ChromaDB cosine search → top-5 chunks
→ build_messages() → generate_answer() → printed answer
1. attention.pdf chunk #34 [██████████████████████░░░░░░░░] 0.7335
第三部分 — 对话记忆
在完成第三阶段的对话记忆训练时，首先写下相关规范：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改始终符合要求。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回效果。仅仅更换提示词往往无法改善较差的检索性能。
问题所在
在问题分析阶段，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，这样操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
解决方案：ConversationMemory
在处理Solution ConversationMemory阶段时，首先需写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在调整提示词之前，先使用固定的问题集来测试信息的召回率。仅仅更换提示词往往无法解决信息检索能力薄弱的问题。
class ConversationMemory:
def __init__(self, session_id: str = None):
self.session_id = session_id or datetime.now().strftime("%Y%m%d_%H%M%S")
self.filepath = os.path.join(MEMORY_DIR, f"{self.session_id}.json")
self.history = []
# Auto-loads if resuming an existing session
if os.path.exists(self.filepath):
self._load()
def add_turn(self, question: str, answer: str, chunks: list[dict]):
# Append user + assistant turns, checkpoint immediately
...
self._save()
def get_messages_with_history(self, query, context_chunks, system_msg):
# Injects last 6 Q&A pairs into the message list before the current turn
...
# New session
memory = ConversationMemory()
# Resume yesterday's session
memory = ConversationMemory("20260413_104959")
Turn 4 question: "How does that compare to what you said about the BLEU score?"
Turn 4 answer: "The context also reports a BLEU score of 28.4 for the
Transformer (big) on WMT 2014 English-to-German. This matches
exactly what I previously stated."
第4部分 — 信息过时性跟踪、CDC与最新性加权
在处理第4阶段的陈旧性跟踪工作时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难解决检索效果不佳的问题。
现实世界中的问题
在处理“现实世界问题”阶段时，首先写下相关契约：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。
过时性跟踪（单元14A）
在处理“过时性跟踪单元14A”阶段时，首先需写下相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词往往无法改善较差的检索效果。 在处理“过时性跟踪单元14A”阶段时，首先需写下相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理方式都是产品的一部分，而非后续需要补充的内容。
def compute_file_hash(pdf_path: str) -> str:
sha = hashlib.sha256()
with open(pdf_path, "rb") as f:
for block in iter(lambda: f.read(65536), b""):
sha.update(block)
return sha.hexdigest()
CDC引擎（单元14B）
CDC Engine Cell 14B阶段在被视为可测量的表面时表现最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任主体，而非复杂的流程链。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
def compute_chunk_hash(text: str) -> str:
return hashlib.md5(text.encode("utf-8")).hexdigest()
def diff_chunks(old_registry: dict, new_chunks: list[str]) -> dict:
new_hash_map = {compute_chunk_hash(c): c for c in new_chunks}
old_hashes = set(old_registry.keys())
new_hashes = set(new_hash_map.keys())
return {
"added": {h: new_hash_map[h] for h in (new_hashes - old_hashes)},
"removed": {h: old_registry[h] for h in (old_hashes - new_hashes)},
"unchanged": {h: old_registry[h] for h in (old_hashes & new_hashes)},
}
v1 → v2 CDC result:
✅ Unchanged : 8 (kept — zero re-embedding cost)
➕ Added : 6 (embedded + inserted)
➖ Removed : 4 (deleted from ChromaDB)
💰 API calls saved: 8/14 (57% reuse)
v2 → v3 CDC result:
✅ Unchanged : 10 (kept - zero re-embedding cost)
➕ Added : 4 (embedded + inserted)
➖ Removed : 2 (deleted from ChromaDB)
💰 API calls saved: 10/14 (71% reuse)
近期权重检索（Cell 14C）
将“近期权重检索单元14C阶段”视为可测量的界面使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
def recency_decay(ingested_at_str: str, half_life_days: float = 30) -> float:
ingested = datetime.fromisoformat(ingested_at_str)
days_gone = (datetime.now() - ingested).total_seconds() / 86400
λ = math.log(2) / half_life_days
return round(math.exp(-λ * days_gone), 4)
blended = alpha * cosine_score + (1 - alpha) * recency_score
# Default: 0.85 * cosine + 0.15 * recency
第5部分 —— 使用多版本PDF进行测试
将第五部分测试与阶段环境视为可度量的对象时，效果最佳。在扩大测试范围之前，需记录一份理想响应示例、一个失败案例以及回滚说明。在功能测试结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，可避免在测试环境从演示版过渡到共享环境时出现意外费用。应将分块策略与检索策略分开处理，当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。将第五部分测试与阶段环境视为可度量的对象时，效果最佳。在扩大测试范围之前，需记录一份理想响应示例、一个失败案例以及回滚说明。需同时记录正常流程与故障恢复流程，重试机制、人工审核环节以及死信处理都属于产品功能的一部分，而非后续需要补充的内容。
TEST 1: Full ingest of v1
TEST 2: Staleness check — same file, correctly skipped
TEST 3: Baseline queries against v1
TEST 4: Copy v2 over active file → CDC kicks in
TEST 5: Same queries now return v2 content, newer chunks visible in recency scores
TEST 6: Copy v3 over active file → second CDC cycle
TEST 7: Recency verification — v3 chunks score highest across the board
TEST 8: Full stack test — weighted retrieval + conversation memory combined
结果与已验证答案
在“结果与验证答案”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型、可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。必须引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失所致。
效率总结
在效率总结阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分幻觉内容与索引缺失的问题。
已知限制
在“已知限制”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境转向共享环境时出现意外费用。 需注明实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。 在“已知限制”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理措施都是产品本身的组成部分，而非后续需要补充的内容。
您已构建的内容
在“您已构建的内容”阶段，首先需写下相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
PDF on disk
└► SHA256 hash check (staleness)
├► Unchanged → skip
└► Changed → CDC diff
├► Unchanged chunks → kept in ChromaDB (zero API cost)
├► Removed chunks → deleted from ChromaDB
└► Added chunks → embed (Gemini) → store (ChromaDB)
↓
User question
└► embed_query() [RETRIEVAL_QUERY task type]
└► ChromaDB cosine search (TOP_K × 3 candidates)
└► recency_decay() per chunk
└► blended score re-ranking
└► top-5 chunks as context
└► ConversationMemory.get_messages_with_history()
└► gpt-oss-20b via Groq/HuggingFace
└► grounded answer + checkpoint to disk
操作检查清单
将“操作检查清单”阶段视为可衡量的指标，效果会更好。在扩大范围之前，需记录一份最佳示例、一个失败案例以及回滚说明。
将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看全部架构即可进行审计。
将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
在预算允许的情况下，利用测试数据而非真实的付费 API，在持续集成过程中添加用于检测关键路径的冒烟测试。
同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的功能。
将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
在推广该技术栈之前，应先冻结版本，为关键流程记录标准输出日志，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及负责密钥轮换的明确责任人。与其展示花哨的一次性演示，不如注重扎实的可靠性。
关于8d172e929623的批处理说明：请将提供商密钥移出代码仓库，设定单会话令牌上限，并将日志存储在评估用示例文件旁，以便后续模型更换时保持数据可比性。