3天改造一个烂尾智能客服：RAG双阶段检索与记忆压缩实战

# 双阶段检索核心逻辑
def hybrid_retrieve(query, top_k=3):
    # 第一阶段：向量召回20个候选
    candidates = vector_store.similarity_search(
        query, 
        k=20, 
        filter={"source": "knowledge_base"}
    )
    # 第二阶段：精排取前3
    reranked = reranker.rerank(
        query, 
        candidates, 
        top_n=top_k,
        return_scores=True
    )
    return reranked

# 记忆压缩实现
def compress_history(history, max_turns=3):
    if len(history) <= max_turns:
        return history
    
    recent = history[-max_turns:]  # 保留最近3轮
    old = history[:-max_turns]     # 较早的历史
    
    # 异步生成摘要（用更便宜的模型）
    summary = lightweight_llm.summarize(
        str(old), 
        max_tokens=200,
        prompt="请用100字以内概括以上对话的核心内容"
    )
    
    # 摘要作为system消息前置
    return [{"role": "system", "content": f"历史对话摘要：{summary}"}] + recent

# 封装LLM调用，自动记录指标
def call_llm_with_metrics(prompt):
    start = time.time()
    response = llm.invoke(prompt)
    cost = time.time() - start
    
    # 写入时序数据库，用于监控告警
    metrics.record(
        latency=cost,
        tokens=response.usage.total_tokens,
        model="glm-4"
    )
    return response

def get_response_with_fallback(query):
    try:
        return llm.invoke(query, timeout=3)
    except TimeoutError:
        # 降级为规则匹配 + 缓存命中
        return cache.get(query) or default_reply("系统繁忙，请稍后重试")