🤖 智净灵境 — 扫地/拖地机器人 AI 智能客服系统

Powered by LangChain RAG + MCP + ChromaDB + 阿里百炼 DashScope

---

📋 目录

1. 项目概述
2. 技术架构深度解析
   • STEP 1 — 整体架构设计
   • STEP 2 — RAG 检索增强生成
   • STEP 3 — LangChain Agent 智能体
   • STEP 4 — MCP 模型上下文协议集成
   • STEP 5 — ChromaDB 向量数据库
   • STEP 6 — Streamlit Web 界面
3. 项目结构
4. 快速开始
5. 知识延伸与学习路径

---

1. 项目概述

智净灵境 是一个面向扫地/拖地机器人领域的 AI 智能客服系统，提供双页面 Web 应用：

核心能力：

• 语义检索：基于向量相似度的知识库搜索，支持维护保养、故障排除、选购指南、常见问题
• 工具调用：LLM 自主决策何时查知识库、何时调高德地图 API 获取位置/天气
• 会话持久化：对话历史自动保存到本地 JSON 文件，刷新不丢失
• 去重入库：基于 MD5 的内容去重，避免重复文档片段污染检索结果

---

2. 技术架构深度解析

以下解析遵循 deep-teach 深度教学模式，每个关键决策都包含：所用技术、选择原因、底层原理、替代方案对比、优越性分析、知识迁移路径。

---

🎯 STEP 1 整体架构设计

📌 操作： 设计 Streamlit + LangChain Agent + RAG + MCP 混合智能客服系统架构

---

① 🛠️ 所用技术

---

② 💡 为什么选择 Agent-Tool 混合架构

项目约束：

• LLM 训练数据有截止日期，无法覆盖最新的扫地机器人型号和故障码
• 用户问题可能涉及实时信息（天气影响清洁建议、根据地区推荐购买渠道）
• 需要单一入口处理多种意图（产品咨询 / 故障排查 / 天气关联 / 购买建议）
• 团队规模小，无法为每种意图单独开发路由逻辑

匹配原因：

• Agent 模式让 LLM 自主选择调用哪个工具，无需人工编写意图分类器，天然覆盖长尾问题
• 工具可以是本地检索器（RAG）也可以是远程 API（MCP），统一接口，扩展方便
• LangChain 的 create_agent() 内置了 ReAct 循环（思考→行动→观察→再思考），处理多步推理

Trade-off：

✅ 选择了：LLM 自主决策的灵活性，无需硬编码路由规则
❌ 放弃了：确定性的执行路径，增加了 token 消耗和延迟
⚖️ 这个交换在本项目中是值得的，因为：客服场景的意图多样性远超规则能覆盖的范围，
   且每次对话多消耗的 token 成本远低于人工客服的成本

---

③ 📚 技术深度剖析

【核心原理】Agent-Tool 循环（ReAct 模式）

Agent 的核心是 ReAct（Reasoning + Acting）循环，它本质上是一个 while 循环：

循环开始
  │
  ├─→ LLM 分析当前状态：我已知什么？用户问了什么？
  │
  ├─→ LLM 决策：是否需要调用工具？
  │     ├─ 是 → 输出 ToolCall（工具名 + 参数）
  │     │        → 执行工具 → 观察结果 → 回到循环开头
  │     └─ 否 → 输出最终文本回复
  │
  └─→ 循环结束

LangChain create_agent() 将这个循环内置在 agent.ainvoke() 中，开发者无需手写循环逻辑。LLM 的每一次工具调用都是一次完整的 API 请求，所以 Agent 模式的 token 消耗 = 普通对话的 N 倍（N = 工具调用次数 + 1）。

【关键概念】

概念 1：Tool（工具） 工具是 LLM 的"手"。每个工具包含三个要素：名称（name）、描述（description，告诉 LLM 何时用）、输入 schema（parameters）。LLM 不执行工具，而是输出一个 ToolCall 消息；LangChain 框架拦截这个消息，执行对应函数，然后将结果以 ToolMessage 形式注入对话上下文。

# 工具定义示例（来自 rag/agent.py:74-89）
@tool  # ← 装饰器自动提取函数名和文档字符串作为 name/description
def search_knowledge_base(query: str) -> str:
    """搜索扫地机器人知识库。涵盖范围：产品维护保养、故障排除、选购指南、常见问题解答。
    :param query: 搜索关键词或问题，例如"日常如何保养"、"E13错误代码"、"适合宠物家庭的机型"
    """  # ← 这段 docstring 是 LLM 判断何时调用此工具的唯一依据
    docs = retriever.invoke(query)        # ← 实际执行：调用 ChromaDB 向量检索
    if not docs:
        return "[知识库] 未找到相关参考资料"  # ← 明确的失败反馈，LLM 据此调整策略
    parts = []
    for doc in docs:
        source = doc.metadata.get("source", "未知来源")
        parts.append(f"[来源: {source}]\n{doc.page_content}")  # ← 标注来源，可追溯
    return "\n\n---\n\n".join(parts)

概念 2：System Prompt（系统提示词） System Prompt 是 LLM 的"宪法"，定义了它的角色、可用工具、行为规则。在本项目中，System Prompt 被动态注入了用户 IP，使 LLM 在调用位置工具时可以直接使用：

# 动态 System Prompt（来自 rag/agent.py:56-62）
ip_hint = (
    f"当前用户的公网IP地址是: {user_ip}。"
    f"当用户要求查询位置或天气时，请直接使用此IP调用 maps_ip_location。"
    if user_ip
    else ""
)
system_prompt = SYSTEM_PROMPT.format(ip_hint=ip_hint)

概念 3：RAG（检索增强生成） RAG 是"给 LLM 外挂一个搜索引擎"。流程分两阶段：

离线阶段（建库）：
  文档 → 分割成chunks → 向量化(Embedding) → 存入向量数据库

在线阶段（问答）：
  用户问题 → 向量化 → 相似度检索 → 取 Top-K chunks → 拼入 LLM 上下文 → LLM 生成回答

⚠️ 常见陷阱

---

④ 🔄 可替代方案对比

对比总结： 本项目选择 Agent-Tool 架构是因为扫地机器人客服的意图高度分散（保养/故障/选购/天气关联），且需要整合异构数据源（文档库 + 地图 API）。意图分类路由无法覆盖长尾，纯 RAG 无法访问外部实时数据。Agent-Tool 虽然增加了延迟（2-5 次 API 调用），但每次调用的 token 开销与实际客服成本相比可忽略。

---

⑤ ⭐ 本选技术的优越性

1. 零规则维护

• 说明：无需编写任何 if "保养" in question: call_baoyang() 的硬编码路由。加一个新知识领域只需放入新文档，Agent 自动知道何时检索。加一个新外部工具只需注册到 Agent，LLM 会自动学习何时调用。
• 数据支撑：本项目的 5 个 TXT 文件覆盖了 5 个完全不同的知识领域（维护、选购、故障、FAQ、100 问），但 System Prompt 没有任何领域特定的分支逻辑。

2. 可扩展的工具生态

• 说明：MCP 协议提供了标准化的工具接入方式。当前接入高德地图（位置+天气），未来可接入物流查询、价格比较、售后工单等，Agent 自动学会编排多工具协作。
• 数据支撑：McpClientManager.get_tools_for(domain) 的领域设计（location/weather/poi）预留了扩展点，添加新领域只需在 tool_domains 字典新增一行。

3. 天然的多步推理

• 说明：用户问"今天下雨，我的扫地机器人要怎么保养？“时，Agent 自动规划：先查天气 → 得知下雨 → 再查知识库"潮湿环境保养” → 综合给出建议。这个过程无需任何人工编排。
• 数据支撑：System Prompt 中明确写了"当用户问题涉及’根据天气’、'这里的气候’等需要结合环境的场景，先查天气和位置，再查知识库给建议"。

📊 生态成熟度评估：

---

⑥ 🔗 知识延伸与迁移

🔄 思想迁移 — Agent-Tool 模式的本质思想还能用在：

• 场景 1：代码助手（Copilot） — LLM 作为 Agent，工具集 = 文件搜索 + 终端执行 + Git 操作 + Linter
• 场景 2：电商导购 — LLM 作为 Agent，工具集 = 商品搜索 + 价格比较 + 库存查询 + 用户画像
• 场景 3：运维助手 — LLM 作为 Agent，工具集 = 日志查询 + K8s 操作 + 告警查询 + Runbook 检索

📖 完整学习路径：

前置知识: [Python asyncio, Prompt Engineering 基础]
              ↓
          【Agent-Tool 架构】
              ↓
进阶方向: [Multi-Agent 协作, LangGraph 状态图, Agent 可观测性, 安全沙箱]

---

🎯 STEP 2 RAG 检索增强生成

📌 操作： 实现文档加载 → 文本分割 → 向量化 → 去重入库 → 语义检索的完整 RAG 管线

---

① 🛠️ 所用技术

---

② 💡 为什么选择 RecursiveCharacterTextSplitter + DashScope Embedding

项目约束：

• 文档为中文，分隔符和英文不同（句号是"。“不是”."）
• 文档长度不一（短的几百字，长的几万字），需要自适应分割
• Embedding 需要处理中文语义，不能依赖英文模型
• 需要在本地持久化，避免每次重启都重新计算向量

匹配原因：

• RecursiveCharacterTextSplitter 按优先级列表递归分割：先按段落 \n\n，再按句子 。！？，最后按字符，保证中文语义单元不被截断
• text-embedding-v4 是阿里百炼的中文优化 Embedding 模型，在中文语义相似度基准上优于 OpenAI text-embedding-3-small
• ChromaDB 内置持久化到磁盘，无需额外配置

Trade-off：

✅ 选择了：中文优化的分隔符 + 国产 Embedding 模型
❌ 放弃了：更通用的英文分隔符 + OpenAI 模型（需要外网访问）
⚖️ 值得是因为：本项目 100% 中文内容，中文 Embedding 模型在语义检索准确率上优势明显

---

③ 📚 技术深度剖析

【核心原理】文本分割的递归算法

RecursiveCharacterTextSplitter 不是简单的按字符数切分，而是按语义层级递归尝试：

给定一段文本，目标 chunk_size=500:

1. 尝试用 "\n\n" 分割 → 如果某段仍 > 500 → 对该段尝试 "\n"
2. 尝试用 "\n" 分割   → 如果某段仍 > 500 → 对该段尝试 "。" / "."  
3. 尝试按句子分割     → 如果某句仍 > 500 → 按 " " 分割
4. 最后按字符硬切      → 保证最终每段 ≤ 500

本项目自定义了中文分隔符列表（rag/config.py:29-31）：

separators: list = field(default_factory=lambda: [
    "\n\n", "\n", ".", "!", "?", "。", "！", "？", " ", ""
])
#                   ↑英文      ↑中文标点              ↑字符级兜底

关键设计： 小于 max_split_threshold（500 字符）的短文本直接保留完整，不做分割。这保证了类似"E13 错误代码怎么办"这样已经自成一体的短 FAQ 不会被切碎。

【关键概念】

概念 1：Embedding（向量嵌入）

Embedding 是将文本映射到高维空间中的坐标点。语义相近的文本，其向量在空间中距离也近。

"扫地机器人怎么保养" → [0.023, -0.451, 0.789, ..., 0.112]  ← 1536 维向量
"清洁机器人日常维护" → [0.019, -0.447, 0.793, ..., 0.108]  ← 向量距离很近（语义相似）
"今天天气怎么样"     → [0.891, 0.234, -0.567, ..., -0.432] ← 向量距离很远（语义不相关）

ChromaDB 使用余弦相似度计算查询向量与库中所有向量的距离，返回最接近的 Top-K 个文档片段。

概念 2：Chunk Overlap（片段重叠）

chunk_size: int = 500     # 每段 500 字符
chunk_overlap: int = 50   # 相邻段之间重叠 50 字符

重叠的目的：

chunk_1: "...清洁机器人需要定期[更换滤网和边刷]。建议..."
chunk_2:                    "[更换滤网和边刷]。建议每 3 个月..."

如果不重叠，[更换滤网和边刷] 恰好落在 chunk_1 末尾和 chunk_2 开头，语义被割裂。50 字符的重叠让关键信息在相邻 chunk 之间形成"桥梁"，保证检索时至少有一个 chunk 包含完整的语义片段。

【实现细节】

• 数据结构： DedupService._records 使用 Python set，O(1) 平均查找时间
• 算法复杂度： MD5 去重 O(n) / 向量检索 O(log n)（HNSW 近似最近邻）
• 关键机制： DedupService.mark() 采用 append-only 写入，不重写整个文件，避免大规模去重时的 IO 瓶颈
• 执行流程： 文件 → DocumentLoader.load() → DedupService.exists() → TextSplitter.split() → vector_store.add_texts() → DedupService.mark()

⚠️ 常见陷阱

---

④ 🔄 可替代方案对比

对比总结： 本项目知识库规模预计不超过 1 万条文档片段，ChromaDB 的单机架构完全够用。FAISS 的纯内存模式重启即丢失数据，需要手动序列化；Milvus 的运维负担对一个小型客服系统来说是过度设计。ChromaDB 的"pip install + 自动持久化"是最符合项目约束的选择。

---

⑤ ⭐ 本选技术的优越性

1. 零运维持久化

• 说明：ChromaDB 将向量数据和元数据存储在本地 ./rag/chroma_db/ 目录，重启后自动加载，不需要启动任何独立服务进程。
• 数据支撑：项目中 persist_directory 配置就是唯一的持久化设置，代码中无需任何 load() 或 restore() 调用。

2. 与 LangChain 深度集成

• 说明：ChromaDB 实现了 LangChain 的 VectorStore 接口，返回的 as_retriever() 可以直接传入 LangChain Chain 或 Agent。
• 数据支撑：knowledge_base.py:137-139 中 self.vector_store.as_retriever(search_kwargs={"k": self.config.retrieval_top_k}) 一行代码即完成检索器创建。

3. 内建元数据支持

• 说明：每次入库时附带 source 和 create_time 元数据，检索结果可以追踪到原始文件，可追溯性远超纯向量检索。
• 数据支撑：检索结果中包含 [来源: 维护保养.txt] 标记，用户在收到回答时可以知道信息来自哪个文档。

---

🎯 STEP 3 LangChain Agent 智能体

📌 操作： 使用 create_agent() 构建 LLM 驱动的工具调用智能体 + 异步对话管理 + 会话历史持久化

---

① 🛠️ 所用技术

---

② 💡 为什么选择 LangChain create_agent() + Qwen3-Max

项目约束：

• 需要一个支持工具调用（Function Calling）的国产大模型（网络限制）
• Agent 需要管理多轮对话历史并持久化
• 需要异步执行以支持 Streamlit 的 async 桥接

匹配原因：

• ChatTongyi 是阿里百炼的 LangChain 集成，直接对接 Qwen 系列模型。Qwen3-Max 原生支持 Function Calling，是目前国产模型中 Agent 能力最强的之一
• create_agent() 封装了 ReAct 循环、工具调用解析、错误重试，开发者只需提供 tools 列表和 system_prompt
• BaseChatMessageHistory 接口 + FileChatHistory 自定义实现提供了灵活的持久化方案

Trade-off：

✅ 选择了：LangChain 高层封装 + Qwen3-Max + 本地 JSON 文件存储
❌ 放弃了：更底层的手动 ReAct 循环（可控性）+ 数据库存储（并发能力）
⚖️ 值得是因为：项目当前是单用户单机环境，本地 JSON 的简单性远胜数据库的运维成本

---

③ 📚 技术深度剖析

【核心原理】create_agent() 内部机制

create_agent() 并不是什么魔法，它内部做了三件事：

1. 绑定工具到 LLM： 将 Python 函数转为 LLM 能理解的 JSON Schema，通过 model.bind_tools() 注入到每次 API 调用的参数中
2. 创建 Agent 图（Graph）： 构建一个循环执行的 LangGraph 图结构（LLM 节点 → 工具执行节点 → LLM 节点 → …）
3. 解析 ToolCall： 拦截 LLM 返回的 tool_call 消息，匹配本地函数，执行后将结果以 ToolMessage 格式反馈给 LLM

# create_agent() 等价于手动编写的以下流程（简化版）
while True:
    response = llm.invoke(messages)           # 1. LLM 决策
    if not response.tool_calls:               # 2. 没有工具调用 → 结束
        return response
    for tool_call in response.tool_calls:     # 3. 执行每个工具
        tool_result = execute_tool(tool_call.name, tool_call.args)
        messages.append(ToolMessage(content=tool_result))  # 4. 结果注入上下文
    # 5. 回到循环开头，LLM 根据工具结果继续推理

【关键概念】

概念 1：异步对话管理 + 历史持久化

本项目实现了完整的会话生命周期管理：

# rag/agent.py:92-107 — 一次完整的问答流程
async def query(self, question: str, session_id: str = "default") -> str:
    history = self.history_manager.get(session_id)      # ① 按 session 加载历史
    
    messages = history.messages + [HumanMessage(content=question)]  # ② 拼接历史 + 新问题
    
    result = await self.agent.ainvoke({"messages": messages})  # ③ Agent 执行
    
    all_messages = result["messages"]
    history.clear()                              # ④ 清空旧历史
    history.add_messages(all_messages)           # ⑤ 写入新历史（含工具调用中间结果）
    
    return all_messages[-1].content              # ⑥ 返回最后一条 AI 消息

概念 2：Streamlit Async 桥接

Streamlit 是同步框架，但 LangChain Agent 和 MCP 客户端都是异步的。run_async() 函数是它们之间的桥梁：

# app.py:79-89
def run_async(coro):
    """在同步上下文中安全执行异步协程"""
    try:
        loop = asyncio.get_running_loop()  # 检测是否已在事件循环中
    except RuntimeError:
        return asyncio.run(coro)          # 没有循环 → 新建一个执行
    new_loop = asyncio.new_event_loop()   # 已有循环（如嵌套调用）→ 新建独立循环
    try:
        return new_loop.run_until_complete(coro)
    finally:
        new_loop.close()                  # 确保清理，避免资源泄漏

概念 3：Session 隔离

每个用户 IP 对应独立的 Agent 实例和会话历史，数据完全隔离：

# app.py:131-143
def get_or_create_agent():
    user_ip = st.session_state.get("user_ip", "")
    cache_key = f"_agent_{user_ip}"         # IP 作为隔离键
    if cache_key not in st.session_state:
        st.session_state[cache_key] = RagAgent(
            config=config, mcp_tools=mcp_tools, user_ip=user_ip
        )                                   # 首次访问 → 创建新 Agent（含新聊天历史）
    return st.session_state[cache_key]

⚠️ 常见陷阱

---

④ 🔄 可替代方案对比

对比总结： 本项目的 Agent 需求是标准的：搜索知识库 + 调用外部 API。create_agent() 完美覆盖这个场景，不需要 LangGraph 的复杂状态管理，也不需要承受手写循环的维护成本。

---

🎯 STEP 4 MCP 模型上下文协议集成

📌 操作： 通过 MCP 协议接入阿里百炼高德地图服务，提供 IP 定位和天气查询工具

---

① 🛠️ 所用技术

---

② 💡 为什么选择 MCP 协议 + 单例模式

项目约束：

• 需要接入高德地图 API（位置 + 天气），但不想维护 API 调用的样板代码
• 多个页面和组件可能用到相同的 MCP 工具，需避免重复连接
• 网络请求有延迟和失败风险，需要合理的缓存策略

匹配原因：

• MCP 是 Anthropic 提出的标准化协议，阿里百炼已托管了高德地图 MCP Server，一行 URL 配置即可获取 maps_ip_location、maps_weather 等工具，省去数百行 API 封装代码
• 单例模式（__new__ + _instance）保证整个应用只有一个 MCP 连接和一份工具缓存
• _tools_cache 字典按领域（location/weather/poi）缓存工具列表，避免每次请求都去拉取远程工具定义

Trade-off：

✅ 选择了：MCP 托管服务（零维护 API 封装）+ 客户端单例
❌ 放弃了：直接调高德 API（更灵活参数）+ 工具级别的动态发现（每次拉取最新工具定义）
⚖️ 值得是因为：扫地机器人客服不需要高德 API 的全部功能，MCP 暴露的定位+天气已覆盖需求

---

③ 📚 技术深度剖析

【核心原理】MCP 协议交互流程

MCP 是客户端-服务器架构：

┌──────────────┐         ┌─────────────────────┐         ┌──────────────┐
│  LangChain   │  HTTP   │   阿里百炼 MCP       │  内部    │  高德地图     │
│  Agent       │ ←─────→ │   Server (托管)      │ ←─────→ │  API 服务     │
│  (客户端)    │  JSON   │                      │  API    │              │
└──────────────┘         └─────────────────────┘         └──────────────┘

步骤:
1. 客户端连接 MCP Server → 获取工具列表（tools/list）
2. Agent 决定调用某工具 → 客户端发送 tools/call（工具名 + 参数）
3. MCP Server 内部调用高德 API → 返回结果
4. 结果返回给 Agent → Agent 基于结果生成回复

【关键概念】

概念 1：单例模式（Singleton）

单例模式的核心思想：一个类在全局永远只有一个实例。实现方式是通过拦截 __new__（Python 中对象的创建方法）：

# mcp_service/mcp_client.py:29-42
_instance: "McpClientManager | None" = None  # ← 类变量，全局唯一

def __new__(cls) -> "McpClientManager":
    if cls._instance is None:               # ① 还没实例化过？
        cls._instance = super().__new__(cls) # ② 创建唯一实例
        cls._instance._initialized = False
    return cls._instance                     # ③ 返回同一个实例（无论调用多少次）

def __init__(self):
    if self._initialized:                    # ④ __init__ 会被多次调用
        return                               # ⑤ 但只有第一次执行初始化逻辑
    self._client = None
    self._tools_cache = {}                   # ⑥ 缓存只会创建一次
    self._initialized = True

生活类比：单例模式就像一家公司只有一个 CEO。无论你从哪个部门（页面/组件）找 CEO，找到的都是同一个人。

概念 2：懒加载（Lazy Loading）+ 缓存

# mcp_service/mcp_client.py:46-58
async def _get_client(self) -> MultiServerMCPClient:
    if self._client is None:               # ← 第一次调用才创建连接
        self._client = MultiServerMCPClient({
            "amap-server": {
                "transport": CONFIG.mcp_transport,  # "http"
                "url": CONFIG.mcp_url,              # 高德 MCP 服务地址
                "headers": {
                    "Authorization": f"Bearer {CONFIG.api_key}"
                }
            }
        })
    return self._client                    # ← 后续调用直接返回缓存的客户端

懒加载的优势：如果用户只访问知识库管理页面（不需要位置天气），MCP 连接根本不会创建，节省资源。

⚠️ 常见陷阱

---

🎯 STEP 5 ChromaDB 向量数据库

📌 操作： 使用 ChromaDB 作为向量存储后端，支持持久化、元数据关联和检索

---

① 🛠️ 所用技术

---

② 💡 为什么选择 ChromaDB

项目约束：

• 需要在 Windows 环境无额外依赖运行
• 数据量小（< 10 万向量），不需要分布式
• 需要与 LangChain 无缝集成
• 需要持久化到本地磁盘，重启不丢失

匹配原因：

• ChromaDB 是纯 Python 实现（基于 SQLite3 + DuckDB），Windows 开箱即用，无需 Docker
• 内建 HNSW 索引，在 1 万级数据量上检索延迟 < 10ms
• 实现了 LangChain 的 VectorStore 基类，as_retriever() 一行代码转检索器
• 数据存储在 ./rag/chroma_db/ 目录，重启自动加载，零运维

Trade-off：

✅ 选择了：嵌入式向量数据库（与应用同进程）
❌ 放弃了：独立的向量数据库服务（如 Milvus），无法横向扩展
⚖️ 值得是因为：项目数据量不会突破单机极限，嵌入式架构消除了网络延迟和运维负担

---

③ 📚 技术深度剖析

【核心原理】HNSW（Hierarchical Navigable Small World）索引

HNSW 是 ChromaDB 内部使用的近似最近邻搜索算法，核心思想来自现实世界的"六度分隔"理论：

• 构建一个多层图结构，上层节点少（高速公路），下层节点多（街道）
• 搜索时从最上层开始，快速定位大致区域，逐层向下细化
• 每一层只检查少数节点（候选集大小 = ef_search 参数）

时间复杂度：O(log N)，远优于暴力搜索的 O(N)

【关键概念】

概念 1：向量距离度量

ChromaDB 默认使用余弦相似度衡量向量距离：

余弦相似度 = cos(θ) = (A · B) / (||A|| × ||B||)

值域: [-1, 1]，1 表示完全相同方向（语义一致），0 表示正交（无关），-1 表示完全相反

对于 Embedding 模型，所有向量都位于单位超球面上，所以余弦相似度简化为向量的点积。

概念 2：Collection（集合）

ChromaDB 用 Collection 组织向量，类似关系数据库的"表"。本项目的 Collection 名为 agent_knowledge，所有文档片段的向量、文本、元数据都存储在这个 Collection 中。

⚠️ 常见陷阱

---

🎯 STEP 6 Streamlit Web 界面

📌 操作： 构建双页面 Streamlit 应用 — 智能客服（用户端）+ 知识库管理（商家端）

---

① 🛠️ 所用技术

---

② 💡 为什么选择 Streamlit

项目约束：

• 团队没有前端开发者，只有 Python 能力
• 需要快速原型交付（2-3 天完成 UI）
• 需要 Chat UI 组件（聊天气泡、输入框、历史消息展示）
• 不需要 SEO 或高并发

匹配原因：

• Streamlit 的 st.chat_message() 和 st.chat_input() 提供了开箱即用的聊天 UI
• 纯 Python 编写，不需要 HTML/JS/CSS（虽然本项目也注入了少量自定义 CSS）
• st.cache_resource 装饰器提供了跨用户共享的重量级对象缓存（MCP 连接、配置实例）
• st.rerun() 机制简化了状态更新后的 UI 刷新

Trade-off：

✅ 选择了：Python 原生的 Web 开发体验，极短的学习曲线
❌ 放弃了：React/Vue 的灵活性、SEO 能力、高并发性能
⚖️ 值得是因为：这是一个内部客服工具，并发量不超过 10 人，不需要 SEO，
   Streamlit 将开发效率提升了 5-10 倍

---

3. 项目结构

langchain-rag-agent/
│
├── app.py                       # 🔴 Streamlit Web 入口
│   ├── page_customer_service()  #   页面一：智能客服（用户端）
│   ├── page_knowledge_management() # 页面二：知识库管理（商家端）
│   └── main()                   #   主入口 + 导航 + 生命周期
│
├── config.py                    # 🟡 全局配置（模块级单例 CONFIG）
│   └── class Config             #   API Key、模型参数、MCP 连接、IP 检测
│
├── .env                         # 🟢 环境变量（API Key、高德 Key、IP）
│
├── rag/                         # 🔵 RAG 核心模块
│   ├── __init__.py              #   导出 RAGConfig, KnowledgeBaseBuilder, RagAgent
│   ├── config.py                #   RAG 专用配置（分割参数、检索参数、路径）
│   ├── agent.py                 #   RagAgent：Agent 创建 + 异步问答 + 历史管理
│   ├── knowledge_base.py        #   KnowledgeBaseBuilder + DedupService + TextSplitter
│   ├── loaders.py               #   DocumentLoader：工厂模式，支持 TXT/PDF
│   └── history.py               #   FileChatHistory + HistoryManager：会话持久化
│
├── mcp_service/                 # 🟣 MCP 客户端
│   ├── __init__.py
│   └── mcp_client.py            #   McpClientManager：单例模式 MCP 客户端
│
├── data/                        # 📁 知识库源文件
│   ├── 维护保养.txt
│   ├── 扫地机器人100问.txt
│   ├── 扫地机器人100问2.txt
│   ├── 扫拖一体机器人100问.txt
│   ├── 选购指南.txt
│   ├── 故障排除.txt
│   └── 扫地机器人100问.pdf
│
├── rag/chroma_db/               # 💾 ChromaDB 向量数据（自动生成）
├── rag/chat_history/            # 💬 会话历史文件（自动生成）
├── rag/md5.text                 # 🔑 MD5 去重记录（自动生成）
│
└── .venv/                       # 🐍 Python 虚拟环境

数据流图：

用户输入问题
     │
     ▼
┌─────────────┐     ┌──────────────────┐     ┌────────────┐
│  Streamlit  │────→│   RagAgent       │────→│   LLM      │
│  app.py     │     │   agent.py       │     │ Qwen3-Max  │
└─────────────┘     └──┬───────┬───────┘     └────────────┘
                       │       │                    │
                  Tool │  Tool │               ToolCall
                  ①    │   ②   │               Decision
                       │       │                    │
           ┌───────────┘       └──────────┐        │
           ▼                              ▼        ▼
   ┌───────────────┐          ┌──────────────┐ ┌──────┐
   │ ChromaDB      │          │ MCP Server   │ │ 直接 │
   │ 向量检索      │          │ 高德地图     │ │ 回复 │
   │ (语义搜索)    │          │ (位置/天气)  │ │      │
   └───────────────┘          └──────────────┘ └──────┘
           │                        │              │
           └────────────────────────┴──────────────┘
                            │
                            ▼
                   ┌────────────────┐
                   │ 合成最终回复    │
                   │ 返回给用户      │
                   └────────────────┘

---

4. 快速开始

环境要求

• Python 3.11+
• Windows / macOS / Linux

安装与运行

# 1. 克隆项目
cd langchain-rag-agent

# 2. 创建虚拟环境
python -m venv .venv
.venv\Scripts\activate     # Windows
# source .venv/bin/activate  # macOS/Linux

# 3. 安装依赖
pip install streamlit langchain langchain-community langchain-chroma \
    langchain-mcp-adapters langchain-text-splitters python-dotenv \
    dashscope httpx pypdf chromadb

# 4. 配置环境变量（.env 文件）
# DASHSCOPE_API_KEY=你的阿里百炼API密钥
# IP=你的公网IP（可选，留空自动检测）

# 5. 构建知识库（首次运行）
python -c "from rag import KnowledgeBaseBuilder; kb = KnowledgeBaseBuilder(); kb.build_from_directory()"

# 6. 启动应用
streamlit run app.py

使用指南

---

5. 知识延伸与学习路径

🗺️ 完整学习路径

前置知识                       本项目核心技术                  进阶方向
─────────────────────────────────────────────────────────────────────
Python 基础 (asyncio,       →  RAG (检索增强生成)       →  Advanced RAG
dataclass, typing)                                          (Self-RAG,
                               LangChain Agent               CRAG, GraphRAG)
Prompt Engineering          →  (ReAct 循环, 工具调用)
基础                                                     →  LangGraph
                               MCP (模型上下文协议)          (复杂多步工作流)
向量/Embedding 概念         →  (标准化工具接入)
                                                         →  Multi-Agent
                               ChromaDB (向量数据库)         (多智能体协作)
                                                            →  Agent 可观测性
                               Streamlit (Python Web)        (LangSmith/LangFuse)

📖 推荐学习资源

---

技术栈速查

---

智净灵境 © 2026 · 基于 LangChain RAG + MCP 的智能客服系统