本文目录15 个章节
把演示资料换成可检索资料库
Tool Registry 已经统一处理参数、超时、错误和结果上限,但 search_local_sources 与 get_document 仍然只读取演示数据。把 fixtures/sources 变成小型资料库后,Agent 可以先搜索摘要,再按 source_id 读取正文。
第一版坚持使用关键词检索。算法简单,命中理由能解释,测试结果也稳定。向量检索会在扩展练习出现;没有基线就直接换 Embedding,很难判断质量变化来自切块、召回还是模型。
一条 RAG 管线包含哪些步骤
原始 Markdown
│
▼
清洗元数据 ──▶ 按标题与长度切块 ──▶ 建立 source_id
│
用户问题 ──▶ 规范化查询 ──▶ 召回 ──▶ 排序与去重
│
▼
返回少量证据摘要RAG 只处理“找到资料并放进上下文”。最终结论仍由模型生成,引用是否有效还要由应用校验。
设计可追踪的 Chunk
每个 chunk 保留文档 ID、块 ID、标题、位置和正文。source_id 一旦进入发布产物就应保持稳定,不能每次索引重建都随机变化。
from dataclasses import dataclass
@dataclass(frozen=True)
class Chunk:
source_id: str
document_id: str
title: str
position: int
text: str
def make_chunks(document_id: str, title: str, paragraphs: list[str],
max_chars: int = 900) -> list[Chunk]:
chunks = []
buffer = ""
for paragraph in paragraphs:
candidate = f"{buffer}\n\n{paragraph}".strip()
if buffer and len(candidate) > max_chars:
position = len(chunks)
chunks.append(Chunk(f"{document_id}#{position}", document_id, title, position, buffer))
buffer = paragraph
else:
buffer = candidate
if buffer:
position = len(chunks)
chunks.append(Chunk(f"{document_id}#{position}", document_id, title, position, buffer))
return chunks按段落边界切块可以保留语义完整性。标题和列表项不应与说明正文分离。生产资料还要记录版本、更新时间和访问级别,当前版本只保留检索所需字段。
建立关键词召回基线
import re
def tokens(text: str) -> set[str]:
return set(re.findall(r"[a-z0-9]+", text.lower()))
def search_chunks(query: str, chunks: list[Chunk], limit: int = 4) -> list[dict]:
query_tokens = tokens(query)
scored = []
for chunk in chunks:
score = len(query_tokens & tokens(f"{chunk.title} {chunk.text}"))
if score:
scored.append((score, chunk.source_id, chunk))
scored.sort(key=lambda item: (-item[0], item[1]))
return [
{"source_id": chunk.source_id, "title": chunk.title, "preview": chunk.text[:280]}
for _, _, chunk in scored[:limit]
]同分结果按 source_id 排序,测试不会因字典遍历顺序抖动。搜索只返回 preview,模型选中后再调用 get_document(source_id),避免一次把所有正文塞进上下文。
用固定查询集调试召回
准备至少六份资料和十个查询,为每个查询写出期望来源。例如:
| 查询 | 期望命中 | 用途 |
|---|---|---|
| agent loop stopping conditions | agent-loop#1 | 精确术语 |
| fixed workflow routing | workflow#0 | 多词组合 |
| quantum networking | 空 | 无证据路径 |
先算 Recall@4:期望来源是否出现在前四条。再看最终简报质量。检索没有命中时,修改 Prompt 无法补回缺失资料;检索命中而回答错误时,问题才落到上下文组织或模型。
接入现有 Agent Loop
注册两个只读工具:search_local_sources(query) 返回摘要列表,get_document(source_id) 返回一块正文。工具描述要求先搜索后读取,并限制每次结果大小。
运行日志应出现清楚链路:
topic
└─▶ search_local_sources("agent loop stopping")
└─▶ [agent-loop#1, agent-loop#0]
└─▶ get_document("agent-loop#1")
└─▶ draft未知 source_id 返回 source_not_found,不把所有可用 ID 泄露给模型。空搜索返回 matches: [],由后面的输出协议表达信息不足。
检索部分的测试
- 同一资料重建索引后,
source_id保持不变。 - 十个查询的期望结果全部冻结,报告 Recall@4。
- 重复 chunk 只保留一份,结果总数不超过四条。
- 搜索与读取都能脱离模型独立测试。
检索接通后,下一步是把任务指令、检索证据和历史装进清楚的上下文,并让最终简报中的每条结论带着可校验来源。
检索命中不等于回答可靠
前面的检索代码能找到正确 chunk,模型仍可能忽略来源、混用两段资料,或在没有命中时补写听起来合理的结论。RAG 的召回结果需要经过一层上下文编排,最终输出也要能被代码检查。
上下文不应成为一条不断增长的字符串。稳定规则、当前任务、外部证据和对话历史拥有不同信任等级与保留周期。
一次推理所需的四层上下文
┌────────────────────────────────────┐
│ 1. 系统与安全规则 稳定、最高优先级 │
├────────────────────────────────────┤
│ 2. 当前任务规格 主题、受众、预算 │
├────────────────────────────────────┤
│ 3. 检索证据 不可信、带 source_id│
├────────────────────────────────────┤
│ 4. 必要历史 最近决策与工具结果 │
└────────────────────────────────────┘检索资料属于不可信数据。即使文档中出现“忽略系统指令”,它也只能作为引用材料,不能改变工具权限和任务目标。第 9 篇会在工具权限和攻击测试中继续加固这条边界。
构建上下文包
from dataclasses import dataclass
@dataclass(frozen=True)
class Evidence:
source_id: str
title: str
text: str
def build_context(task: dict, evidence: list[Evidence], max_chars: int = 6000) -> str:
blocks = [
"[SYSTEM_RULES]\nUse only supplied evidence. Preserve source IDs. Report gaps.",
f"[TASK]\nTopic: {task['topic']}\nAudience: {task['audience']}",
]
used = sum(len(block) for block in blocks)
for item in evidence:
block = f"[EVIDENCE source_id={item.source_id} title={item.title}]\n{item.text}"
if used + len(block) > max_chars:
break
blocks.append(block)
used += len(block)
return "\n\n".join(blocks)预算按完整 block 截断,避免保留半句话却丢掉来源标签。生产版可以为各层单独分配 token,当前版本用字符数建立可观察基线,并在日志中记录省略了哪些 source_id。
输出使用 claim,而不是一整段 Markdown
模型先返回结构化结果,再由呈现层渲染 Markdown。
{
"status": "succeeded",
"claims": [
{
"text": "Agent loop 会在工具结果返回后重新决策。",
"source_ids": ["agent-loop#1"]
}
],
"evidence_gaps": []
}结构化 claim 让应用逐条检查引用。写作风格可以变化,来源关系仍然稳定。
用代码校验来源关系
def validate_brief(result: dict, known_source_ids: set[str]) -> list[str]:
errors = []
status = result.get("status")
claims = result.get("claims", [])
if status == "succeeded" and not claims:
errors.append("claims_required")
for index, claim in enumerate(claims):
source_ids = claim.get("source_ids", [])
if not source_ids:
errors.append(f"claim_{index}_missing_source")
elif not set(source_ids) <= known_source_ids:
errors.append(f"claim_{index}_unknown_source")
return errors这个检查只能证明来源 ID 存在,不能证明证据支持结论。评测集还要抽样做 entailment 检查,查看数字、否定词和限定条件是否与原文一致。
无证据时怎样结束
搜索返回空数组,或者所有资料都超出时间范围时,输出状态改为 insufficient_evidence。
{
"status": "insufficient_evidence",
"claims": [],
"evidence_gaps": ["资料库没有 2026 年后的官方版本说明"]
}这条路径不算运行失败。它说明 Agent 正常完成检索,并按照任务规则拒绝无来源结论。产品界面可以提示用户补充资料或允许远程搜索。
第一版完整简报怎样跑
一次成功运行形成下面的链路。
任务规格
└─▶ search_local_sources
└─▶ get_document
└─▶ build_context
└─▶ structured brief
└─▶ validate_brief
└─▶ Markdown renderer测试同时覆盖:命中一个来源、命中多个冲突来源、无证据、上下文截断、模型返回未知来源。每条失败都能定位到检索、上下文、生成或校验中的一层。
第二个里程碑
本地资料已经可以生成带出处的技术简报。第 6 篇的工具边界继续保护检索调用,本篇新增的引用校验则把 source_id 约束延伸到最终输出。
第 8 篇把稳定的操作流程、领域规则和配套资源封装成 Skill,使 Agent 能按任务加载方法,避免所有说明长期堆在系统 Prompt 中。
REFERENCES
参考链接
所属系列
AI AGENT 入门教程