智扫通：基于 RAG + Agent 的扫地机器人智能客服项目

GitHub 仓库：https://github.com/Fengj0n/smart-sweep-agent

一、项目背景

这个项目是一个面向扫地机器人场景的智能客服系统，核心目标不是做一个“能聊天”的界面，而是把实际业务中常见的几类问题拆开处理：

  • 产品选购
  • 故障排查
  • 日常保养
  • 实时天气查询
  • 设备使用月报
  • 知识库问答

这类问题有一个共同点：并不适合全部交给大模型直接生成。原因很简单，很多内容本身就依赖真实数据、固定规则或者已有知识文档。如果直接让模型回答，容易出现信息不准确、结论过泛、甚至编造数据的情况。

因此，这个项目采用了“规则分流 + 工具调用 + RAG 检索 + 模型总结”的方式，把不同问题交给不同链路处理。

二、项目效果

2.1 首页

在这里插入图片描述

2.2 知识库问答

在这里插入图片描述

2.3 实时天气

在这里插入图片描述

2.4 使用月报

在这里插入图片描述

三、功能范围

3.1 知识库问答

支持围绕扫地机器人场景进行问答，主要覆盖：

  • 选购建议
  • 清洁效果问题
  • 故障排查
  • 滚刷、尘盒、滤网等耗材维护
  • 地毯、瓷砖、木地板等地面适配
  • 宠物家庭、多人家庭等使用场景

3.2 实时天气查询

接入 Open-Meteo 的实时接口，可以查询指定城市当前天气。常见问题包括：

  • 今天西安天气怎么样
  • 北京现在多少度
  • 下雨天是否适合拖地

3.3 设备使用月报

支持按照用户和月份生成月报，读取的是演示用的 CSV 数据。可以输出：

  • 使用场景
  • 清洁表现
  • 耗材状态
  • 对比与建议
  • 最近多个月趋势

3.4 多工具组合

对于复合问题，系统会组合多个链路处理，例如：

  • 天气 + 清洁建议
  • 月报 + 保养建议
  • 知识库 + 模型总结

3.5 来源引用

知识库回答会附带来源信息，避免回答结果只是一段无法追溯的生成文本。

四、技术栈

模块 技术
前端 Streamlit
大模型 通义千问 qwen3-max
Embedding DashScope text-embedding-v4
Agent 自定义意图识别 + 工具编排
RAG LangChain
向量数据库 Chroma
文档处理 LangChain Text Splitters、PyPDF
配置 YAML
天气接口 Open-Meteo
运行环境 Python 3.8

五、系统架构

天气

月报

知识库

普通问答

用户输入

Streamlit 页面

ReactAgent

意图识别

天气工具

报告工具

RAG 工具

通义千问

Open-Meteo

CSV 使用数据

Chroma 向量库

本地知识文档

架构说明

  1. 用户在页面输入问题；
  2. ReactAgent 先判断问题类型；
  3. 不同类型进入不同工具链；
  4. 天气问题走天气接口；
  5. 月报问题读 CSV；
  6. 知识库问题先检索文档，再交给模型总结；
  7. 普通问答直接调用模型。

这样设计的目的，是把“检索、统计、调用、总结”拆开，避免所有问题都混在一个生成式链路里。

六、实现思路

6.1 意图识别

项目没有把所有问题都先发给模型判断，而是先用本地规则做分流。当前支持的意图包括：

  • 天气
  • 月报
  • 知识库问答
  • 普通问答
  • 天气 + 建议
  • 月报 + 建议

这样做的好处是路由清晰、调用成本低，也更容易排查问题。

6.2 工具优先

能直接用代码解决的部分，优先用工具而不是模型生成：

  • 天气 → 直接调接口
  • 月报 → 直接读 CSV
  • 知识库 → 先检索，再总结

这样能减少模型的自由发挥空间，也能提高结果稳定性。

6.3 模型负责总结

模型主要做三件事：

  • 组织自然语言
  • 综合多条信息
  • 给出更完整的表达

换句话说，模型在这个项目里更像“总结器”，而不是“数据源”。

6.4 知识库兜底

如果向量检索阶段出错，系统会退回到本地文本检索，避免整条链路直接失败。

七、核心代码

7.1 意图识别

agent/react_agent.py 的意图识别逻辑如下：

@staticmethod
def _intent(query: str) -> str:
    weather_words = ("天气", "气温", "温度", "多少度", "下雨", "降雨", "晴天", "阴天", "潮湿")
    if any(word in query for word in weather_words):
        if any(word in query for word in ("拖地", "清扫", "扫地", "湿拖")):
            return "weather_advice"
        return "weather"
    if any(word in query for word in ("报告", "月报", "统计", "趋势", "使用情况")):
        if any(word in query for word in ("保养", "耗材", "维护", "建议")):
            return "report_advice"
        return "report"
    robot_words = (
        "扫地机器人", "扫拖", "故障", "报错", "维护", "保养", "选购", "清洁", "吸力",
        "拖地", "避障", "基站", "尘盒", "电池", "滚刷", "边刷", "水箱", "耗材", "导航",
    )
    if any(word in query for word in robot_words):
        return "rag"
    return "general"

这段逻辑的作用是先判断问题类型，再走对应的处理路径。这样做的核心价值在于：简单、直接、可控。

7.2 实时天气工具

天气工具的实现思路是：先做城市地理编码，再请求实时天气接口，再格式化结果。

@tool
def get_weather(city: str) -> str:
    """查询指定城市当前实时天气，不使用模拟数据。"""
    city = (city or "").strip()
    if not city:
        return "请提供要查询的城市。"

    cache_seconds = int(agent_conf.get("weather_cache_seconds", 600))
    cached = _weather_cache.get(city)
    if cached and time.time() - float(cached["timestamp"]) < cache_seconds:
        return str(cached["value"])

    try:
        geo_url = (
            "https://geocoding-api.open-meteo.com/v1/search"
            f"?name={quote(city)}&count=5&language=zh&format=json&countryCode=CN"
        )
        results = (_fetch_json(geo_url) or {}).get("results") or []
        if not results:
            return f"未找到城市“{city}”，请检查城市名称。"

        location = results[0]
        weather_url = (
            "https://api.open-meteo.com/v1/forecast"
            f"?latitude={location['latitude']}&longitude={location['longitude']}"
            "&current=temperature_2m,relative_humidity_2m,apparent_temperature,"
            "weather_code,wind_speed_10m,wind_direction_10m,precipitation"
            "&timezone=Asia%2FShanghai"
        )
        current = (_fetch_json(weather_url) or {}).get("current") or {}
        desc = _wmo_description(int(current["weather_code"]))
        result = (
            f"{location.get('name', city)}实时天气：{desc}，{current['temperature_2m']}℃，"
            f"体感{current.get('apparent_temperature', '-')}℃，湿度{current['relative_humidity_2m']}%，"
            f"风速{current['wind_speed_10m']}km/h，降水{current.get('precipitation', 0)}mm。"
            f"更新时间：{str(current['time']).replace('T', ' ')}。"
        )
        _weather_cache[city] = {"timestamp": time.time(), "value": result}
        return result
    except Exception:
        return f"{city}实时天气暂时无法获取，请稍后重试。"

7.3 使用月报

月报功能先根据用户和月份读取数据，再决定是走单月报告还是趋势分析。

def _answer_report(self, query: str) -> str:
    user_id, month, trend_months = self._parse_report_query(query)
    user_id = user_id or self.context.get("user_id")
    month = month or self.context.get("report_month")
    self.context["user_id"] = user_id

    if trend_months > 1:
        records = get_external_records(user_id, trend_months)
        if not records:
            return f"用户{user_id}暂无可用月报数据。"
        # 组装趋势分析内容并交给模型总结

    if not month:
        return f"请选择月份后再生成用户{user_id}的月报。"
    record = get_external_record(user_id, month)
    if not record:
        return f"用户{user_id}在{month}没有使用数据，请更换用户或月份。"
    # 交给模型生成更自然的报告

这里没有把月报完全交给模型，而是先把真实数据组织好，再由模型负责表达。

7.4 RAG 兜底检索

def retriever_docs(self, query: str) -> List[Document]:
    try:
        docs = self.vector_store.similarity_search(query, k=6)
        docs = self._rerank(query, docs, limit=3)
        if docs:
            return docs
    except Exception as exc:
        logger.warning(f"[rag]向量检索失败，将使用本地文本兜底：{exc}")
    return self._rerank(query, self._load_fallback_docs(query, limit=6), limit=3)

这段代码的目标很明确：向量检索优先，失败时走本地文本兜底，避免知识库直接不可用。

7.5 RAG 总结

result = str(self.chain.invoke({"input": query, "context": context})).strip()
answer = result[:1600].rstrip()
if sources:
    answer += "\n\n来源：" + "、".join(sources[:2])
return answer

这里控制了输出长度，并附加来源，避免回答过长或没有依据。

八、目录结构

.
├── agent/
│   ├── react_agent.py
│   └── tools/
│       ├── agent_tools.py
│       └── middleware.py
├── config/
│   ├── agent.yml
│   ├── chroma.yml
│   ├── prompts.yml
│   └── rag.yml
├── data/
│   ├── external/records.csv
│   └── *.txt / *.pdf
├── model/
│   └── factory.py
├── prompts/
│   ├── main_prompt.txt
│   ├── rag_summarize.txt
│   └── report_prompt.txt
├── rag/
│   ├── rag_service.py
│   └── vector_store.py
├── utils/
│   ├── config_handler.py
│   ├── file_handler.py
│   ├── logger_handler.py
│   ├── path_tool.py
│   └── prompt_loader.py
├── app.py
├── evaluate.py
├── evaluation_cases.json
├── requirements.txt
└── README.md

九、环境准备

9.1 安装依赖

pip install -r requirements.txt

9.2 配置 DashScope API Key

Windows PowerShell：

$env:DASHSCOPE_API_KEY="你的 DashScope API Key"

Linux / macOS：

export DASHSCOPE_API_KEY="你的 DashScope API Key"

注意：不要把真实 Key 上传到公开仓库。

十、如何运行

10.1 构建知识库

python rag/vector_store.py

10.2 启动项目

streamlit run app.py

10.3 运行评估

python evaluate.py

十一、使用示例

11.1 知识库问答

扫地机器人选购怎么选？
滚刷被毛发缠绕怎么办？
拖地效果变差怎么处理？

11.2 实时天气

今天西安天气怎么样？
北京现在多少度？
西安下雨天适合拖地吗？

11.3 使用报告

生成用户1001在2025-12的使用报告
查看用户1010最近三个月趋势
生成用户1004在2025年1月的耗材建议报告

十二、常见问题

12.1 为什么知识库查询会失败？

常见原因是 embedding 服务请求受到网络或代理影响。可以检查网络环境，必要时重新构建知识库。

12.2 为什么天气查询不稳定？

请确认城市名是否正确，同时检查 Open-Meteo 是否可访问。

12.3 为什么月报显示没有数据？

请检查 data/external/records.csv 中是否存在对应用户与月份。

十三、后续可扩展方向

如果继续迭代，可以考虑：

  • 更强的意图识别
  • 会话持久化
  • 对话导出
  • PDF / HTML 报告导出
  • 更专业的知识库重排策略
  • 更完整的自动化评估
  • FastAPI 服务版接口

十四、项目地址

GitHub：https://github.com/Fengj0n/smart-sweep-agent

# python # 人工智能
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

智元机器人上半年出货约8400台

avatar DAMO开发者矩阵
cover

收藏！小白程序员快速上手大模型：AI赋能客服，效率翻倍！

avatar DAMO开发者矩阵
cover

基于具身交互智能数字人，我用Flask搭了一个AI教学平台：课代表AI全流程实战

avatar DAMO开发者矩阵