先理解：API 返回的不应该只有一句答案
客服 RAG 的输出至少包含三类信息：回答内容、引用来源、是否转人工。只返回一句自然语言，看起来像聊天机器人，但不适合生产排查。用户投诉时，你需要知道它引用了哪条知识、为什么没有转人工、当时检索分数是多少。
本篇先实现一个可调试的问答 API。它会从向量库取回候选 Chunk，再用阈值和风险规则决定回答还是转人工。暂时不用复杂的 LLM Prompt，是为了先把检索链路和业务控制跑通。
为什么要有转人工逻辑
客服系统不是考试系统，不知道答案时不能编。遇到低置信度、高风险承诺、知识缺失时，转人工是正确结果，不是失败结果。
你可以把 handoff=true 理解成系统的安全阀。早期宁可多转一点人工，也不要让机器人用不确定的信息回答用户。
调试顺序
如果回答不对，先看检索结果，不要马上改 Prompt。确认 top3 里有没有正确知识、来源是否正确、分数阈值是否合理。检索没拿到正确材料时，生成模型越强，越可能把错误包装得很自然。
本篇实现 /ask 接口。为了让教程可直接跑通，回答先用模板生成；接入 LLM 时只需替换 compose_answer。
检索器
src/retriever.py：
python
from chromadb import PersistentClient
from sentence_transformers import SentenceTransformer
from src.config import INDEX_DIR, COLLECTION_NAME, EMBEDDING_MODEL
model = SentenceTransformer(EMBEDDING_MODEL)
client = PersistentClient(path=str(INDEX_DIR))
collection = client.get_or_create_collection(
COLLECTION_NAME,
metadata={"hnsw:space": "cosine"},
)
def retrieve(query: str, product: str = "all", top_k: int = 4):
vector = model.encode([query], normalize_embeddings=True)[0].tolist()
result = collection.query(
query_embeddings=[vector],
n_results=top_k,
where={"$or": [{"product": product}, {"product": "all"}]},
)
return [
{"text": doc, "metadata": meta, "distance": distance}
for doc, meta, distance in zip(
result["documents"][0],
result["metadatas"][0],
result["distances"][0],
)
]
HIGH_RISK_WORDS = ["赔偿", "退款金额", "投诉", "律师", "起诉", "封号"]
MAX_COSINE_DISTANCE = 1.2
def should_handoff(question: str, docs: list[dict]) -> tuple[bool, str | None]:
if any(word in question for word in HIGH_RISK_WORDS):
return True, "问题涉及高风险承诺或投诉，需要人工处理"
if not docs:
return True, "没有检索到可用知识"
if docs[0]["distance"] > MAX_COSINE_DISTANCE:
return True, "检索相关性不足"
return False, None
def compose_answer(question: str, docs: list[dict]) -> str:
top = docs[0]
return (
f"根据当前资料，建议回复：\n\n{top['text']}\n\n"
f"如用户情况不符合上述资料，请转人工确认。\n"
f"引用来源：{top['metadata']['source']}"
)FastAPI
src/app.py：
python
from fastapi import FastAPI
from pydantic import BaseModel
from src.retriever import retrieve, should_handoff, compose_answer
from src.schema import Answer
app = FastAPI(title="Customer RAG")
class AskRequest(BaseModel):
question: str
product: str = "all"
@app.post("/ask", response_model=Answer)
def ask(req: AskRequest):
docs = retrieve(req.question, req.product)
handoff, reason = should_handoff(req.question, docs)
if handoff:
return Answer(answer="这个问题需要人工客服进一步确认。", sources=[], handoff=True, reason=reason)
return Answer(
answer=compose_answer(req.question, docs),
sources=[d["metadata"]["source"] for d in docs],
handoff=False,
)启动和测试
bash
uvicorn src.app:app --reloadbash
curl -X POST http://127.0.0.1:8000/ask \
-H 'Content-Type: application/json' \
-d '{"question":"耳机进水能保修吗","product":"headphone"}'再测转人工：
bash
curl -X POST http://127.0.0.1:8000/ask \
-H 'Content-Type: application/json' \
-d '{"question":"你们必须赔偿我 500 元"}'验收点
- 普通问题返回
handoff: false。 - 高风险问题返回
handoff: true。 - 普通回答包含
sources。 - 回答里有引用来源。
如果返回“检索相关性不足”
先不要急着改 Prompt。这个返回通常说明检索距离超过了阈值。Chroma 的 distance 是距离，不是相似度，数值越小越相关。本教程显式使用 cosine 距离后，可以先按这个范围理解：
- 小于
0.4：通常很相关。 0.4到0.9：可能相关，需要看命中文本。0.9到1.2：弱相关，小数据集早期可以先放行观察。- 大于
1.2：通常应转人工或拒答。
调试时可以临时打印 top 结果：
python
related_docs = retrieve("耳机进水能保修吗", "headphone")
for d in related_docs:
print(d["distance"], d["metadata"], d["text"][:80])python
from openai import OpenAI
client = OpenAI(api_key="你的API_KEY")
context = "\n".join(related_docs)
prompt = f"""
请根据以下内容回答问题：
{context}
问题：{query}
"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
print(response.choices[0].message.content)如果你已经按旧代码运行过入库脚本，必须删除 data/index 后重新执行 python -m src.ingest，否则旧 collection 仍可能不是 cosine 配置。
