多文档问答代理
为文档语料库建立索引，并回答跨越所有文档的问题。
多文档问答代理 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
多文档问答概览
多文档问答智能体会从包含 N 份文档的集合中检索相关内容，综合生成答案，并将每个结论归因于其来源。
与单文档问答不同，多文档智能体必须处理不同来源之间相互冲突的信息，并判断哪些文档与问题最相关。
为多个文档建立索引
在回答任何问题之前，必须为所有文档建立索引：解析、分块、嵌入，并存储到向量数据库中。每个文本块都会与元数据一起存储，元数据会将它关联回源文档。
import chromadb
from chromadb.utils import embedding_functions
import os
client = chromadb.PersistentClient(path='./doc_index')
ef = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv('OPENAI_API_KEY'),
model_name='text-embedding-3-small'
)
collection = client.get_or_create_collection('documents', embedding_function=ef)
def index_document(doc_id, doc_path, doc_title):
# Parse and chunk
chunks = pdf_to_chunks(doc_path, chunk_size=800, overlap=150)
for i, chunk in enumerate(chunks):
chunk_id = f'{doc_id}_chunk_{i}'
collection.add(
ids=[chunk_id],
documents=[chunk['text']],
metadatas=[{
'doc_id': doc_id,
'title': doc_title,
'page': chunk['page'],
'source_file': doc_path
}]
)
print(f'Indexed {len(chunks)} chunks from: {doc_title}')检索相关文本块
收到问题后，向量存储会从所有已建立索引的文档中检索语义上最相似的文本块。n_results 参数控制要检索的文本块数量。
def retrieve_relevant_chunks(question, n_results=8):
results = collection.query(
query_texts=[question],
n_results=n_results,
include=['documents', 'metadatas', 'distances']
)
chunks = []
for i in range(len(results['documents'][0])):
chunks.append({
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'distance': results['distances'][0][i],
'relevance': 1 - results['distances'][0][i] # cosine similarity proxy
})
# Sort by relevance
chunks.sort(key=lambda x: x['relevance'], reverse=True)
return chunks在提示中标注来源
将检索到的文本块传给 LLM 时，请为每个文本块标注其文档来源。这样，LLM 就可以在答案中使用编号引用来标明来源。
def format_chunks_for_prompt(chunks, max_chars=4000):
sections = []
used_chars = 0
for i, chunk in enumerate(chunks, 1):
meta = chunk['metadata']
header = f"[Source {i}: {meta['title']}, page {meta.get('page', '?')}]"
content = chunk['text'][:600]
entry = f'{header}\n{content}'
if used_chars + len(entry) > max_chars:
break
sections.append(entry)
used_chars += len(entry)
return '\n\n'.join(sections)
QA_PROMPT = '''Answer the question based on the provided document excerpts.
Cite sources as [Source N]. If sources conflict, mention both views.
{context}
Question: {question}
Answer:'''
def answer_question(question):
chunks = retrieve_relevant_chunks(question, n_results=6)
context = format_chunks_for_prompt(chunks)
return llm_call(QA_PROMPT.format(context=context, question=question))跨文档推理
有些问题需要综合多个文档中的信息，而不只是找到一个匹配的文本块。例如：“三份合同中，哪一份的违约金条款最低？”
请采用两步法：先从每份文档中检索相关文本块，再让 LLM 对这些内容进行比较和综合。
def cross_document_compare(question, doc_ids):
# Retrieve best chunks per document
per_doc_chunks = {}
for doc_id in doc_ids:
results = collection.query(
query_texts=[question],
n_results=3,
where={'doc_id': {'$eq': doc_id}} # filter by document
)
if results['documents'][0]:
per_doc_chunks[doc_id] = results['documents'][0]
# Format with document labels
context_parts = []
for doc_id, texts in per_doc_chunks.items():
doc_label = f'Document {doc_id}'
combined = ' '.join(texts[:2])[:600]
context_parts.append(f'== {doc_label} ==\n{combined}')
comparison_context = '\n\n'.join(context_parts)
return llm_call(f'Compare these documents to answer: {question}\n\n{comparison_context}')处理相互冲突的信息
不同文档可能会陈述相互冲突的事实——一份合同规定应在 30 天内付款，另一份则规定应在 60 天内付款。智能体必须检测并呈现这些冲突，而不是悄悄选择其中一种说法。
CONFLICT_PROMPT = '''You are analyzing multiple document sources.
Some may contain conflicting information.
For each factual claim you make:
1. Cite the source document
2. If another source contradicts it, explicitly note the conflict
3. Indicate which source you believe is more authoritative, if possible
Document excerpts:
{context}
Question: {question}
Answer (with conflict notes where applicable):'''
def answer_with_conflict_detection(question):
chunks = retrieve_relevant_chunks(question, n_results=8)
context = format_chunks_for_prompt(chunks)
return llm_call(CONFLICT_PROMPT.format(
context=context, question=question
))按相关性阈值过滤
检索到的文本块并不一定都真正相关——向量相似度需要在召回率和精确率之间进行权衡。请设置最低相关性阈值，排除匹配程度较低、可能误导 LLM 的文本块。
MIN_RELEVANCE = 0.72 # cosine similarity threshold
def retrieve_above_threshold(question, n_results=10, threshold=MIN_RELEVANCE):
chunks = retrieve_relevant_chunks(question, n_results=n_results)
relevant = [c for c in chunks if c['relevance'] >= threshold]
print(f'Retrieved: {len(chunks)}, Above threshold: {len(relevant)}')
if not relevant:
# Fallback: use top 3 even if below threshold
return chunks[:3]
return relevant
def answer_with_threshold(question):
chunks = retrieve_above_threshold(question)
if not chunks:
return 'I could not find relevant information in the indexed documents.'
context = format_chunks_for_prompt(chunks)
return llm_call(QA_PROMPT.format(context=context, question=question))生成来源引用
生成答案后，提取其中引用的来源文档，并以结构化列表返回。这样可以帮助用户找到原始文档进行核验。
import re
def extract_citations(answer_text, chunks):
# Find all [Source N] references in the answer
cited_nums = set(int(m) for m in re.findall(r'\[Source (\d+)\]', answer_text))
citations = []
for num in sorted(cited_nums):
idx = num - 1
if idx < len(chunks):
meta = chunks[idx]['metadata']
citations.append({
'source_num': num,
'title': meta.get('title', 'Unknown'),
'page': meta.get('page', 'N/A'),
'file': meta.get('source_file', '')
})
return citations
def answer_with_citations(question):
chunks = retrieve_above_threshold(question)
context = format_chunks_for_prompt(chunks)
answer = llm_call(QA_PROMPT.format(context=context, question=question))
citations = extract_citations(answer, chunks)
return {'answer': answer, 'citations': citations}使用交叉编码器重新排序
初始向量检索使用双编码器（速度快，但结果近似）。交叉编码器会将每个（查询、文本块）组合在一起进行评分，从而对顶部结果重新排序——准确度更高，但速度更慢。这种两阶段方法可以提升最终答案的质量。
# pip install sentence-transformers
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def rerank_chunks(question, chunks, top_k=4):
# Score each chunk against the question
pairs = [(question, c['text']) for c in chunks]
scores = reranker.predict(pairs)
# Attach scores and re-sort
scored_chunks = list(zip(scores, chunks))
scored_chunks.sort(key=lambda x: x[0], reverse=True)
top_chunks = [chunk for _, chunk in scored_chunks[:top_k]]
print(f'Re-ranked {len(chunks)} chunks -> kept top {top_k}')
return top_chunks
def answer_with_reranking(question):
# Retrieve more initially
initial_chunks = retrieve_relevant_chunks(question, n_results=12)
# Re-rank for precision
top_chunks = rerank_chunks(question, initial_chunks, top_k=4)
context = format_chunks_for_prompt(top_chunks)
return llm_call(QA_PROMPT.format(context=context, question=question))按文档级元数据过滤
当用户指定某份文档或日期范围时，请在进行嵌入搜索之前先在元数据层面进行过滤。这样可以防止不相关的文档影响检索结果。
def retrieve_filtered(question, filters=None, n_results=8):
query_kwargs = {
'query_texts': [question],
'n_results': n_results,
'include': ['documents', 'metadatas', 'distances']
}
# ChromaDB metadata filters
# Example: {'doc_id': 'contract_2024', 'year': {'$gte': 2023}}
if filters:
query_kwargs['where'] = filters
results = collection.query(**query_kwargs)
return [
{'text': t, 'metadata': m, 'relevance': 1 - d}
for t, m, d in zip(
results['documents'][0],
results['metadatas'][0],
results['distances'][0]
)
]
# Example usage
chunks = retrieve_filtered(
'What are the payment terms?',
filters={'doc_id': {'$in': ['contract_a', 'contract_b']}}
)使用新文档更新索引
文档集合会随着时间发生变化——会添加新文件，也会更新旧文件。建立索引的流程必须支持增量更新：添加新文档、为已更新的文档重新建立索引，并移除已删除的文档。
import os
import hashlib
# Track indexed documents by file hash
index_registry = {} # {filepath: {hash, doc_id, indexed_at}}
def file_hash(filepath):
with open(filepath, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
def index_if_new_or_changed(filepath, title):
fhash = file_hash(filepath)
existing = index_registry.get(filepath)
if existing and existing['hash'] == fhash:
print(f'Skipping unchanged: {title}')
return existing['doc_id']
if existing:
# Remove old chunks from vector store
collection.delete(where={'doc_id': {'': existing['doc_id']}})
print(f'Re-indexing updated: {title}')
else:
print(f'Indexing new: {title}')
doc_id = hashlib.md5(filepath.encode()).hexdigest()[:8]
index_document(doc_id, filepath, title)
index_registry[filepath] = {'hash': fhash, 'doc_id': doc_id}
return doc_id
def sync_document_directory(directory):
pdf_files = [f for f in os.listdir(directory) if f.endswith('.pdf')]
for fname in pdf_files:
fpath = os.path.join(directory, fname)
title = fname.replace('.pdf', '').replace('_', ' ').title()
index_if_new_or_changed(fpath, title)
print(f'Sync complete: {len(pdf_files)} files processed')知识检查
在使用检索增强生成的多文档问答系统中，相关性阈值用于防止什么问题？
回顾：多文档问答智能体
多文档问答的流程是：为所有文档建立索引（解析 → 分块 → 嵌入 → 与元数据一起存储）→ 在所有文档中检索相关文本块 → 添加来源标签进行格式化 → 生成带引用的综合答案。
高级技术包括：使用跨文档比较回答比较类问题；通过提示词检测冲突；按相关性阈值过滤；使用交叉编码器重新排序以提高精确率；以及使用元数据过滤，将查询范围限定到特定文档或日期范围。
常见问题解答
「多文档问答代理」课时是免费的吗？
是的 — 「多文档问答代理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「多文档问答代理」这节课中我会学到什么？
为文档语料库建立索引，并回答跨越所有文档的问题。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「多文档问答代理」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。