前言
上一个项目做 AI 深度调研时，发现一个核心痛点：LLM 收到检索结果后容易"编造来源"——给不存在的 source_id 打分、虚构 URL、把广告当权威来源。
于是设计了一套完整的检索-清洗-评分-校验管线，覆盖网络搜索和本地知识库两个数据源。本文记录具体方案。
整体流程
搜索计划 → 网络检索(Bocha API) ──→ 去重 → 相关性过滤 → 证据整理(LLM) → prune → enrich
└── 本地检索(Milvus) ────→ 去重 → 相关性过滤 → 证据整理(LLM) → prune → enrich
↓
EvidenceJudge 评分审计
↓
证据池 (evidence_pool)
一、网络检索：Bocha API 集成
def bocha_web_search_records(query: str, count: int = 4) -> list[dict]:
api_key = os.getenv("BOCHA_API_KEY", "").strip()
if not api_key:
return []
payload = {
"query": query,
"summary": True,
"freshness": "noLimit",
"count": count, # 控制返回量，减少无用 Token 消耗
}
request = urllib.request.Request(
url="https://api.bocha.cn/v1/web-search",
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
method="POST",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
)
with urllib.request.urlopen(request, timeout=30) as response:
result = json.loads(response.read().decode("utf-8"))
pages = result.get("data", {}).get("webPages", [])
records = []
for idx, page in enumerate(pages[:count], 1):
url = page.get("url", "")
domain = url.split("://", 1)[1].split("/", 1)[0] if "://" in url else ""
records.append({
"source_id": f"WEB-{idx}",
"title": page.get("name"),
"url": url,
"snippet": page.get("summary", ""),
"domain": domain,
"source_type": "web",
})
return records
注意 count 设为 4 而非默认 8——每个搜索结果后续都会进 LLM 上下文，控制量就是控制 Token 成本。
二、本地检索：Milvus RAG
class RAGSystem:
def __init__(self, api_key, config):
self.embeddings = DashScopeEmbeddings(
model="text-embedding-v1",
dashscope_api_key=api_key,
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\\n\\n", "\\n", "。", "！", "？", "；", "，", " ", ""],
)
self.vectorstore = Milvus(
embedding_function=self.embeddings,
collection_name=config.collection_name,
connection_args={"uri": f"http://{config.milvus_host}:{config.milvus_port}"},
auto_id=True,
)
def search_records(self, query: str, k: int = 4) -> list[dict]:
docs = self.vectorstore.similarity_search(query, k=k)
records = []
for idx, doc in enumerate(docs, 1):
records.append({
"source_id": f"LOC-{idx}",
"doc_id": doc.metadata.get("source", ""),
"title": Path(doc.metadata.get("source", "")).name,
"snippet": doc.page_content,
"source_type": "local",
})
return records
中文检索的分隔符很重要——RecursiveCharacterTextSplitter 按 \\n\\n → \\n → 句号 → 感叹号 → 逗号的优先级逐级切分，避免在中文字符中间截断。
三、数据清洗管线
1. 去重
def _dedupe_sources(items: list[dict], key_fields: list[str]) -> list[dict]:
seen = set()
results = []
for item in items:
key = tuple(str(item.get(f, "")).strip() for f in key_fields)
if key in seen:
continue
seen.add(key)
results.append(item)
return results
# 网页按 url+title 去重，本地文档按 doc_id+snippet 去重
raw_records = _dedupe_sources(raw_records, ["url", "title"]) # web
raw_records = _dedupe_sources(raw_records, ["doc_id", "snippet"]) # local
2. 低质来源过滤
def _is_bad_web_domain(domain: str) -> bool:
blocked = ["datasheet", "bdtic", "doc88", "elecfans", "down"]
return any(item in domain.lower() for item in blocked)
不是所有网页都值得进证据链。广告站、文档搬运站、下载站直接拦截。
3. 相关性过滤
def _estimate_relevance(query: str, text: str) -> float:
terms = _extract_query_terms(query)
if not terms:
return 0.0
haystack = text.lower()
hits = sum(1 for term in terms if term in haystack)
return hits / max(len(terms), 1)
关键词命中率低于阈值的记录丢弃。阈值设得比较宽容（0.2），宁愿多留也不要误杀——反正后续还有 LLM 裁判环节。
四、证据整理与幻觉控制
检索到的原始记录交给 WebScout / LocalRAGScout 两个 Agent 做结构化整理：
payload, content, _ = _invoke_json_agent(
state,
"请基于以下网页证据整理结构化 JSON。\\n"
f"原始网页证据：\\n{_format_raw_records(raw_records, 'web')}",
agent,
…
)
evidence = payload.get("evidence", [])
两个关键安全措施：
1. Prune：裁剪幻觉 source_id
allowed_source_ids = {str(item.get("source_id")) for item in raw_records}
evidence = _prune_evidence_to_allowed_sources(evidence, allowed_source_ids)
LLM 有时候会"编造"输入中不存在的 source_id。直接把不在白名单里的条目干掉。
2. Enrich：补充丢失字段
def _enrich_evidence_from_raw(evidence, raw_records):
raw_lookup = {r.get("source_id"): r for r in raw_records}
for ev in evidence:
sid = ev.get("source_id")
raw = raw_lookup.get(sid, {})
if not ev.get("url") and raw.get("url"):
ev["url"] = raw["url"]
if not ev.get("domain") and raw.get("domain"):
ev["domain"] = raw["domain"]
return evidence
LLM 整理 JSON 时可能丢失 url、domain 等字段，从原始记录补回来。
五、证据评分
EvidenceJudge 对每一条证据打分：
def _score_evidence(record: dict) -> tuple[float, str]:
source_type = record.get("source_type")
if source_type == "local":
return 0.92, "企业内部知识库证据，默认高可信"
domain = record.get("domain", "").lower()
if _is_official_domain(domain): # .gov.cn, .edu 等
return 0.88, "官方或权威机构域名"
if any(w in domain for w in ["news", "finance", "reuters", "people", "xinhuanet"]):
return 0.72, "主流媒体域名"
if domain:
return 0.58, "普通互联网来源，需要交叉验证"
return 0.45, "来源信息不完整"
评分逻辑很朴素：知识库 > 官方域名 > 主流媒体 > 普通网页 > 匿名来源。
六、引用校验
Writer 生成报告时，正文中引用了一个 [WEB1_1-3]，这个 ID 真的存在吗？校验逻辑：
def _validate_and_fix_citations(content: str, valid_source_ids: set[str]) -> tuple[str, list[str]]:
pattern = r'\\[([A-Z]+\\d+_\\d+-\\d+)\\]'
def replace_citation(match):
cid = match.group(1)
return f"[{cid}]" if cid in valid_source_ids else "" # 非法引用直接删除
return re.sub(pattern, replace_citation, content), used_ids
总结
检索增强生成（RAG）的核心不只是"查到了什么"，更关键的是"查到的东西能不能用"。这条管线六个环节环环相扣：
去重 — 别让重复内容浪费 LLM 上下文
域过滤 — 垃圾站的数据不进证据链
相关性打分 — 宁多勿少，后续 LLM 还会二审
结构化整理 — LLM 把原始结果转成标准证据格式
幻觉裁剪 — 白名单机制，编造的 source_id 直接剔除
引用校验 — 报告中的引用必须真实存在
