🎯 课程主题

项目第八阶段（最终阶段）：封装 generate_answer() 函数，串联检索与 Agent 推理——格式化检索结果、拼接上下文 prompt、调用 Agent 生成回答，完成 RAG + Agent 客服知识库的完整闭环；同时包含一个实际 Bug 的调试修复过程。

📝 核心知识点

  1. generate_answer() 函数设计：

    • 入参：query: str（用户问题）
    • 流程：调用 retrieve() 获取命中文档 → 格式化搜索结果为上下文文本 → 构造 user_prompt（包含 query + context）→ agent.invoke() 得到回答 → 解析并打印最终回答

  2. 检索结果格式化：

    • 遍历 hits 列表，每个 hit 包含 entity（实体数据）和 distance（相似度）
    • 从 entity 中提取 text（文本内容）、source（来源文件）、chunk_id（切片序号），缺失字段用 "UNKNOWN" 填充
    • 拼接所有命中结果为格式化的上下文段落，便于 Agent 理解

  3. Prompt 构造：

    • 使用 f-string 将用户问题与检索上下文拼入 user_prompt 模板
    • 以字典格式传入 agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
    • Agent 返回的 result["messages"] 列表中，[-1] 是最终回答消息

  4. Agent 响应解析：

    • result["messages"] 包含多轮消息：HumanMessage → （工具调用 ToolMessage，若有）→ AIMessage
    • 最终回答在 messages[-1]，通过 .pretty_print() 输出格式化内容

  5. Bug 调试实例：向量化时报 object cannot be converted to py string 错误，根因是 embed_query() 底层调用 embed_documents() 时代码路径异常。解决方案：显式用 str(query) 强制转换传入的查询字符串。

🏗️ 架构与工作流

[阶段8] generate_answer(query):
       user_query
         ↓
       retrieve(query, limit=5)  →  hits (命中文档列表)
         ↓
       ┌─ 遍历 hits ──────────────────────────────┐
       │  entity → {text, source, chunk_id}         │
       │  distance → 相似度值                        │
       │  拼接 → formatted_context (段落文本)         │
       └───────────────────────────────────────────┘
         ↓
       构造 user_prompt:
          f"问题: {query}\n上下文:\n{formatted_context}"
         ↓
       agent.invoke({
           "messages": [{"role": "user", "content": user_prompt}]
       })  →  result
         ↓
       result["messages"][-1].pretty_print()
         ↓
       输出最终回答

💻 代码实战

# ============ 阶段8: 生产与回答生成 ============
def generate_answer(query: str):
    """
    串联检索与 Agent，生成最终回答
    """
    # 8.1 调用检索函数获取命中文档
    hits = retrieve(query=str(query), limit=5)  # 注意 str() 强转

    # 8.2 格式化检索结果为上下文字符串
    context_parts = []
    for i, hit in enumerate(hits):
        entity = hit["entity"]
        text = entity.get("text", "UNKNOWN")
        source = entity.get("source", "UNKNOWN")
        chunk_id = entity.get("chunk_id", "UNKNOWN")
        distance = hit.get("distance", 0.0)

        context_parts.append(
            f"[文档{i+1}] 来源: {source} | chunk_id: {chunk_id} | 相似度: {distance:.4f}\n"
            f"内容: {text}\n"
        )

        print(f"检索命中[{i+1}]: source={source}, chunk_id={chunk_id}, distance={distance:.4f}")

    context = "\n".join(context_parts)

    # 8.3 构造 user_prompt
    user_prompt = f"""问题: {query}

上下文:
{context}

请基于以上上下文回答问题。如果上下文信息不足，请明确回复"我不知道"。""

    # 8.4 调用 Agent 生成回答
    result = agent.invoke({
        "messages": [
            {"role": "user", "content": user_prompt}
        ]
    })

    # 8.5 解析并输出最终回答
    final_message = result["messages"][-1]
    print("\n===== 最终回答 =====")
    final_message.pretty_print()

# ============ 测试 ============
if __name__ == "__main__":
    generate_answer(query="试用期结束后多久可以转正？")

⚠️ 常见问题与避坑指南

  1. str(query) 强转问题：某些情况下 embed_query() 底层编码器无法正确识别传入的字符串类型，报 cannot convert to py string 错误。用 str(query) 显式转换可规避。该问题可能与 Python 版本或 LangChain 内部类型检查相关。

  2. messages[-1] 索引：Agent 返回的 result["messages"] 包含完整对话历史（用户消息 → 可能的工具消息 → AI 回复），[-1] 确保始终取到最后一条 AI 回复。若未来引入工具调用，中间会多出 ToolMessage，用 [-1] 比硬编码索引更稳健。

  3. 检索命中数为 0 的处理：当前代码未显式处理检索无结果的情况。实际生产中应检查 len(hits) == 0，此时可直接让 Agent 回复"暂无相关数据"，避免传入空上下文。

  4. 上下文过长问题：检索结果过多时，拼接的 context 可能超出模型 token 限制。需控制 limit 值，或对文本做截断处理。

💡 个人总结与延伸

本节将 RAG 架构的所有组件（加载、切分、向量化、存储、检索、Agent 推理）串联为完整的端到端应用。至此，"尚硅谷 Assistant 客服知识库"项目的 8 个阶段全部完成，形成一个可运行的 RAG + Agent 知识库系统。

完整项目流程回顾：

阶段内容核心组件
1全局配置常量定义（URL/DB/Collection/模型）
2初始化 MilvusMilvusClient → 数据库/Collection 创建
3嵌入模型OpenAIEmbeddings
4文档切分TextLoader + RecursiveCharacterTextSplitter
5向量化写入embed_documents → upsert → flush
6创建 Agentcreate_agent(model, system_prompt)
7检索函数embed_query → search → 返回命中
8生产回答retrieve + prompt 构造 + agent.invoke

后续优化方向：引入 Re-ranking 提升检索精度、增加工具调用能力（如让 Agent 可以查询更多数据源）、接入 MCP（Model Context Protocol）和 Skills 拓展 Agent 能力边界——这也是课程末尾预告的进阶内容。

最近更新:
Contributors: yangwenguang