搜索运行手册，而非直接粘贴：LangChain中一个简化的引用型RAG实现方案
在 LangChain 中构建两步检索路径：加载运行手册，将其分块，使用精准查询进行搜索，保留元数据，要求提供引用来源的 ID，并在无需付费嵌入模型的情况下进行测试。
搜索运行手册而非在提示词中填塞内容
此时，帮助台应用会根据工单类型进行分类处理。数据库相关工单会同时调取服务目录、近期事件以及最新部署信息，而访问权限相关的工单则走另一条处理路径。不过，建议背后详细的操作步骤仍未呈现。
你可以将所有运行手册的内容都粘贴到提示词中。对于数量较少的文件来说这可行，但随着文档量增加就不再适用了。过长的提示词不仅成本更高、处理速度更慢，还难以确定是哪段内容支撑了答案。此外，它们还可能让模型接触到当前用户根本无权阅读的文档。解决办法就是搜索。
这里的处理流程分为两个阶段。在索引阶段，Markdown格式的运行手册会转化为包含源代码元数据的文档对象，随后这些文档对象会被进一步拆分成更小的片段：
indexing time
Markdown runbooks
-> Document objects with source metadata
-> smaller chunks
-> embeddings
-> vector store
在发起请求时，票证分类会生成一个精准的搜索查询，该查询会被发送给检索系统：
request time
ticket classification
-> focused search query
-> retriever
-> a few runbook passages with source IDs
-> recommendation chain
-> source-ID check
人们通常将这种方法称为两步检索增强生成，或两步RAG。标签的重要性不如流程顺序：首先进行搜索，然后生成模型仅接收搜索结果中的段落。模型不会自行决定是否进行搜索，也无法随意选择数据来源，因此不需要任何智能体。
贯穿全文的案例
所使用的示例是一个工单内容：在14:05版本部署后，欧盟地区的结账请求失败，且结账API的日志显示“数据库连接被拒绝”。其结构化分类如下：
{
"ticket_id": "INC-2048",
"summary": "EU checkout fails after a deployment.",
"category": "database",
"severity": "SEV2",
"service": "checkout-api",
"confidence": 0.91,
"evidence": ["database connection refused"],
"missing_information": ["Affected request percentage"]
}
一次有效的搜索应返回结账数据库操作手册，而非支付延迟指南或身份访问指南。该结果还应包含足够的元数据，以便回答四个问题：文本来自哪个文件、由哪项服务拥有、该片段在文件中的起始位置，以及推荐内容引用了哪些来源ID。没有来源元数据的文本难以处理。
检索的六个组成部分
在使用每个部分之前都需要先明确其定义。
文档。可以将Document视为带有标签的文本块。这些文本可能来自Markdown文件、PDF、网页、维基页面，甚至是表格中的一行，而标签则可用于记录来源ID、标题、服务类型、路径、所有者、版本信息以及允许查看该内容的人员。
分块。文档会被拆分成可独立检索的片段。一份操作手册可能会同时描述诊断、恢复和回滚流程，但每当遇到数据库凭证相关问题时，并无需每次都返回回滚相关的资料。
嵌入。嵌入模型会将文本转换为数字列表，根据所选的比较方法，含义相似的文本会出现在彼此附近。但这并非事实核查手段——相似的文本仍可能错误、过时或不可使用。
向量存储。向量存储用于保存文档向量并查找相近的向量。有些向量仅存在于内存中，而另一些则存储在数据库或托管服务上。向量的存储位置并不能替代访问控制机制。
检索。将自由形式的查询文本交给检索系统，它就会返回Document对象。这一概念比向量搜索更为宽泛：在相同的接口背后，可以集成关键词匹配、向量存储、混合搜索服务或自研的搜索API。
生成。最终阶段会获取选定的段落并生成结构化的答案。无论输出什么内容，系统都不能假装搜索结果包含其实并不存在的资料。
检索并非内存
一系列操作手册构成知识库，对它们进行查询即属于检索行为。聊天记录是一系列消息的集合，保存的用户信息属于普通应用数据，而工作流检查点则用于记录执行状态。这些数据的保留策略和权限规则各不相同，正因如此，后续阶段会在代码中将它们区分开来。
项目结构与配置
此阶段的快照结构如下：
langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│ ├── checkout-database.md
│ ├── identity-access.md
│ └── payment-latency.md
└── tests/
└── test_retrieval.py
安装这些软件包：
python -m pip install -U \
langchain \
langchain-openai \
langchain-text-splitters \
pydantic \
python-dotenv \
pytest
该应用会调用提供商的嵌入内容。测试时会用一个小的本地替代物替换，因此不会发生对提供商的调用。
步骤1：编写可识别的运行手册
结账流程的运行手册以标题行开头：
# Checkout database connection failures
接着是服务、所有者和主体：
Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
previous successful deployment.
2. Confirm that the application secret points to the expected
credential version.
3. Test network reachability from the application environment without
printing the credential.
4. Compare connection-pool limits with the database connection limit.
标题、服务名称及所有者信息均为纯文本，加载器还会将选定的值复制到元数据中。在企业系统中，绝不能让标题决定访问权限。访问标签必须来自可信系统，并且在每次数据检索时都需进行验证。
步骤2：将每个文件加载到文档中
将load_runbooks函数放入retrieval.py文件中。首先需要路径处理工具：
from pathlib import Path
接下来是加载器本身：
from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
documents: list[Document] = [] for path in sorted(runbook_dir.glob("*.md")):
text = path.read_text(encoding="utf-8")
title = text.splitlines()[0].removeprefix("# ").strip()
service_line = next(
line for line in text.splitlines()
if line.startswith("Service:")
)
service = service_line.split(":", 1)[1].strip() documents.append(
Document(
page_content=text,
metadata={
"source_id": path.stem,
"title": title,
"service": service,
"path": str(path),
},
)
) if not documents:
raise ValueError(f"No Markdown runbooks found in {runbook_dir}") return documents
处理流程：
glob("*.md")用于选取某个已知目录中的 Markdown 文件。sorted(...)可确保本地索引顺序的一致性。搜索质量不应依赖于文件系统的排序方式，但一致的输入有助于调试。- 文件会以 UTF-8 格式解码。
- 提取标题和服务信息的方式刻意保持简单，因为示例文件遵循固定的格式。真正的加载器应验证元数据并报告格式错误的文件，而非凭猜测处理。
page_content用于存储可搜索的文本。metadata用于存储后续所有数据块都应包含的字段。- 最后的检查可避免空索引被误认为是处理成功的结果。
步骤 3：分割文档
使用 RecursiveCharacterTextSplitter：
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_runbooks(documents: list[Document]) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=420,
chunk_overlap=60,
add_start_index=True,
)
return splitter.split_documents(documents)
chunk_size=420 表示每个数据块约包含420个字符，这很适合这个小型示例。chunk_overlap=60 会让相邻数据块之间重复少量文本，这样当分割边界恰好位于句子中间时，该句子仍能与对应的解释保持在一起。不过过大的重叠度并非总是更好：它会导致存储的文本增多、嵌入计算量增加，还会出现更多近似重复的结果。
使用add_start_index=True时，每个数据块都会记录其起始的字符偏移量，这样就可以追溯到原始文件中的对应位置。split_documents会返回新的Document对象，且每个数据块都会继承其父对象的元数据。
选择数据块大小
从教程中复制数字并非明智之举。首先应了解文档的组织方式以及人们实际提出的问题。某个故障排除步骤可能需要将标题、条件及警告集中放在一处，而长长的政策表格则可能需要使用具备结构感知能力的拆分工具。之后再通过带标签的问题来测试检索效果。
第4步：创建向量存储检索器
为保持示例的简洁性，该应用使用了InMemoryVectorStore：
from langchain_core.vectorstores import InMemoryVectorStore
def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
chunks = split_runbooks(load_runbooks(runbook_dir))
store = InMemoryVectorStore.from_documents(
chunks,
embedding=embeddings,
)
return store.as_retriever(search_kwargs={"k": k})
from_documents 用于嵌入并存储这些片段，而 as_retriever 则提供标准的检索接口。k=3 表示请求三份文档。但这并不意味着这三份文档都相关，也无法保证其中一定包含所需的段落。
该应用会生成提供者嵌入向量：
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
Path(__file__).parent / "runbooks",
embeddings,
)
可用的模型及其名称取决于您的服务提供商账户，因此请在配置中保留模型名称。此外，InMemoryVectorStore在进程结束时会导致索引丢失。这对于教学示例和小型测试来说尚可，但无法作为共享的生产环境索引使用。
第5步：从可信字段构建精准查询
不要使用整个应用状态的庞大数据集来进行搜索，而应选择特定的分类字段：
def build_query(classification: TicketClassification) -> str:
values = [
classification.service or "unknown service",
classification.category,
classification.summary,
*classification.evidence,
]
return " | ".join(values)
对于联程票，查询结果大致如下：
checkout-api | database | EU checkout fails after a deployment. |
database connection refused
该字符串会包含可能的服务名称、问题类别、简短摘要以及从工单中提取的证据。机密信息、隐藏指令和完整的消息历史则不会被包含在内。
第6步：在转换过程中保留元数据
检索工具返回的是Document对象，每个对象都应转化为带类型的文本段落：
class RetrievedPassage(BaseModel):
source_id: str
title: str
service: str
start_index: int
text: str
然后将 Document 映射到它上：
def to_passage(document: Document) -> RetrievedPassage:
return RetrievedPassage(
source_id=str(document.metadata["source_id"]),
title=str(document.metadata["title"]),
service=str(document.metadata["service"]),
start_index=int(document.metadata.get("start_index", 0)),
text=document.page_content.strip(),
)
必填字段会用方括号标出，因此缺少 source_id、title 或 service 的数据块会立即引发错误。start_index 有默认值，因为较早存储的数据块可能没有该字段，实际迁移时需判断是否可使用此默认值。
第7步：构建精简的推荐输入
面向模型的写入程序会收到如下结构：
class RecommendationInput(BaseModel):
ticket_id: str
ticket_summary: str
category: Category
severity: Severity
service: str | None
evidence: list[str]
passages: list[RetrievedPassage] = Field(
min_length=1,
max_length=4,
)
当没有相关段落返回时，就无法给出有依据的推荐：
def prepare_recommendation_input(
classification: TicketClassification,
passages: list[RetrievedPassage],
) -> RecommendationInput:
if not passages:
raise ValueError("At least one runbook passage is required")
return RecommendationInput(
ticket_id=classification.ticket_id,
ticket_summary=classification.summary,
category=classification.category,
severity=classification.severity,
service=classification.service,
evidence=classification.evidence,
passages=passages[:4],
)
如果搜索没有找到结果，就不应让模型凭其一般知识即兴发挥。应将工单转交给相关人员，或使用已获批准的备用方案。
第8步：在结构化输出中请求源ID
结果架构包含引用信息：
class GroundedRecommendation(BaseModel):
explanation: str = Field(min_length=20, max_length=500)
next_checks: list[str] = Field(min_length=1, max_length=4)
source_ids: list[str] = Field(min_length=1, max_length=4)
needs_human_review: bool
提示内容如下：
Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.
提示说明会有帮助，但应用层的检查依然重要。
第9步：拒绝那些检索未返回的引用
作者运行程序后，需将其来源ID与检索到的段落进行比对：
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError(
"Recommendation cited sources that retrieval did not return: "
+ ", ".join(sorted(unknown))
)
这只能检测到伪造的来源ID，仅此而已。模型即便引用了实际返回的文档，仍可能误解其含义，因此仍需进行评估和人工审核。
第10步：连接固定工作流
请求时的代码非常简单：
def run_retrieval_workflow(
classification: TicketClassification,
retriever,
recommendation_chain,
) -> RetrievalResult:
query = build_query(classification)
passages = retrieve_passages(retriever, query)
values = prepare_recommendation_input(classification, passages)
recommendation = recommendation_chain.invoke(values)
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError("Recommendation cited an unavailable source") return RetrievalResult(
query=query,
recommendation=recommendation,
passages=passages,
)
没有隐藏的循环。搜索执行一次，写入操作执行一次，源代码检查也执行一次。
输出结果
结果如下所示：
{
"query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
"recommendation": {
"explanation": "The error matches the checkout database connection runbook.",
"next_checks": [
"Compare the active database host, port and database name with the previous successful deployment.",
"Confirm that the application secret points to the expected credential version."
],
"source_ids": ["checkout-database"],
"needs_human_review": false
},
"passages": [
{
"source_id": "checkout-database",
"title": "Checkout database connection failures",
"service": "checkout-api",
"start_index": 0
}
]
}
该 API 可以将每个源 ID 转换为链接，但前提是必须先验证当前调用者有权打开该链接。
在不支付嵌入服务费用的情况下测试检索功能
配套的快照中包含了 KeywordEmbeddings，这是一种用于测试的小型替代方案，它使用固定的词汇表进行计算：
VOCABULARY = (
"database",
"connection",
"checkout",
"deployment",
"credential",
"pool",
"payment",
"latency",
"timeout",
"identity",
"access",
"role",
"claim",
)
不要将其误认为是正式的嵌入工具。它的唯一作用就是让测试快速、离线且结果可预测。第一个测试会检查正在处理的工单：
passages = retrieve_passages(
retriever,
build_query(classification()),
)
assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0
在三个标记过的查询中，第二个度量标准在两个位置出现了数值：
score = hit_at_k(
retriever,
[
("checkout database connection refused", "checkout-database"),
("payment provider latency timeout", "payment-latency"),
("identity role claim access denied", "identity-access"),
],
)
assert score == 1.0
这两种测试方式都引出了一个简单的问题：预期的来源是否位于前两个结果中？一个真正的测试集需要包含大量问题、难以处理的负例、相似文档以及当前的访问规则。仅在三个示例上获得满分只能证明这三个样本通过了测试。
第三个测试会输入一个标明为not-returned的虚假推荐，工作流必须将其拒绝。请使用以下方式运行所有测试：
pytest -q
预期结果：
3 passed
检索失败类型的命名
每种失败类型都应拥有专属名称，这样便于确定排查方向。
- 索引构建失败：文件无法读取、元数据缺失或嵌入请求失败。切勿将未完全构建的索引当作完整版本发布。
- 数据新鲜度失败：策略发生了变化，但搜索系统仍使用旧有的规则。需记录版本信息与时间戳，并确定修改和删除内容如何同步到索引中。
- 权限失败：索引返回了当前用户无权阅读的内容。应在任何文本传递给模型之前就应用访问过滤器，因为事后隐藏链接为时已晚。
- 排序失败：正确的文档已被索引，但却未出现在顶部结果中。需调整查询条件、过滤器、分块方式、嵌入模型或排序规则，然后再进行测试。
检索到的文本不可信
操作手册中存在错误，有时还会出现类似对模型下达指令的表述。应将检索到的段落视为引用的资料，而绝不能将其视为具有更高优先级的应用规则。应在系统提示中保持行为稳定，对检索到的文本进行界定，并在应用代码中对工具的使用加以限制。任何操作手册都不得自行授权触发用于写入内容的工具。
何时向量搜索不是合适工具
有些查询并不需要向量。精确的票证编号只需通过普通数据库查询即可获取，而已知的服务名称则可直接对应到目录条目中。精确的错误代码和罕见的标识符更适合通过关键词搜索来处理。如果公司已经拥有能够索引相关文档并管控权限的搜索服务，那就使用它。向量搜索只是众多工具中的一种而已。
投入生产还缺什么
在一家公司能够依赖这种方案之前，还有几项要素缺失：
- 需使用持久化存储而非内存
- 必须明确指定每个文档的负责人
- 在查询时需应用访问过滤机制
- 需要处理文档版本和删除问题
- 需制定嵌入模型变更时的应对方案
- 需限制返回的文本量
- 需要用于评估检索效果的标记测试数据
该框架将各部分相互连接，而应用程序负责做出相关决策。
帮助台应用程序的现状
它现在可以搜索受控的运行手册集并保留其来源：
classification
-> focused query
-> retriever
-> typed passages with metadata
-> recommendation
-> source-ID check
接下来要做的是将消息历史、保存的用户信息、实时工作流状态以及检查点分开处理。明确这些界限有助于判断是普通的 LangChain 链路就足够了，还是需要使用 LangGraph 更为合适。
延伸阅读：
- https://docs.langchain.com/oss/python/langchain/retrieval
- https://docs.langchain.com/oss/python/langchain/knowledge-base
- https://docs.langchain.com/oss/python/integrations/splitters/index
- https://docs.langchain.com/oss/python/integrations/retrievers