🍵 茶叶电商智能客服 — LangGraph Agent 工作流

基于 LangGraph 构建的多智能体客服系统，集成 混合 RAG 检索（BM25 + 向量）、三层智能缓存、实时安全防护，支持 Streamlit 前端与 FastAPI 接口。

架构总览

graph TD
    subgraph Frontend
        UI[Streamlit Chat UI]
        API[FastAPI REST]
    end

    subgraph Security
        RL[Rate Limiter<br/>多维度滑动窗口]
        IF[Input Filter<br/>XSS/SQL注入/敏感词]
        AD[Anomaly Detector<br/>高频/相同查询检测]
    end

    subgraph "LangGraph Agent Workflow"
        RET[Retrieve Node<br/>混合检索 RRF 融合]
        GEN[Generate Node<br/>Qwen-Turbo + 工具调用]
        TOOLS[Tool Node<br/>库存查询 | 订单 | 物流]
        CACHE[三层缓存<br/>L1高频 | L2语义 | L3会话]
    end

    subgraph "Retrieval Pipeline"
        BM25[BM25<br/>jieba 分词]
        VEC[Vector Search<br/>Chroma + BGE]
        QE[Query Expansion<br/>多查询变体]
        RRF[RRF Fusion<br/>k=60]
    end

    subgraph Storage
        CH[(ChromaDB<br/>向量持久化)]
        KB[tea.md<br/>结构化知识库]
        MEM[(long_memory/<br/>用户记忆 JSON)]
    end

    UI -->|SSE Stream| API
    API --> Security
    Security --> RET
    RET --> QE --> BM25 & VEC
    BM25 & VEC --> RRF --> |top_k chunks| GEN
    GEN --> CACHE
    CACHE -->|miss| GEN
    GEN -->|has_tool_call| TOOLS
    TOOLS --> GEN
    GEN -->|answer| API

    style CACHE fill:#f9f,stroke:#333
    style Security fill:#faa,stroke:#333
    style GEN fill:#afd,stroke:#333

Loading

快速开始

# 1. 安装依赖
pip install -r requirements.txt

# 2. 下载嵌入模型
python fix_model.py

# 3. 配置环境变量
cp .env.example .env
# 编辑 .env，填入 TONGYI_API_KEY

# 4a. Streamlit 前端启动
streamlit run app.py --server.port=8501

# 4b. FastAPI 后端启动
uvicorn api:app --host 0.0.0.0 --port 8000 --reload

# 5. Docker 一键启动
docker-compose up -d

核心功能

1. 混合 RAG 检索（BM25 + 向量）

指标	纯 BM25	纯向量	混合检索
召回率	62.3%	71.5%	83.7%
精确率	58.1%	52.4%	72.6%
F1 分数	59.6%	60.5%	77.8%

• RRF 融合（k=60）消除评分量纲差异，对离群排名鲁棒
• 查询扩展：自动生成多查询变体，取最优排名
• jieba 中文分词：BM25 针对中文优化
• BGE 嵌入模型：bge-small-zh-v1.5，512 维，本地运行

2. 三层智能缓存

L1: 高频问题缓存 (< 10ms)
  └─ 预定义 FAQ 字典，精确 + 模糊匹配，永不过期

L2: 语义指纹缓存 (~50ms)
  └─ 余弦相似度匹配 (阈值 0.75) + Jaccard 双重校验
  └─ LRU 淘汰，最大 1000 条

L3: 会话上下文缓存 (~100ms)
  └─ session_id + 用户记忆 (姓名/过敏/偏好) 联合 key
  └─ TTL 15 分钟，最大 500 条

3. LangGraph Agent 工作流

• 意图识别：11 种模式匹配（库存/订单/投诉/推荐/问候...）+ 置信度评分
• 工具调用：check_inventory check_order_status check_logistics request_human
• 条件路由：retrieve → generate → [tools] → generate 循环
• 用户记忆：姓名/过敏原/偏好 热层提取 + 冷层 JSON 持久化

4. 安全防护

• 速率限制：四维度滑动窗口（全局/IP/会话/VIP），1 分钟窗口 + 突发配额
• 异常检测：相同查询频率 / 短时高频 / 单 IP 多会话 → 自动阻塞（30s/5min/1h）
• 输入过滤：XSS / SQL注入 / 路径遍历 / 命令执行 正则检测 + 敏感词替换
• VIP 用户：白名单机制，独立速率配额

5. API 接口 (FastAPI)

端点	方法	说明
/chat	POST	聊天接口（支持 SSE 流式）
/chat/stream	POST	SSE 实时流式输出
/health	GET	健康检查
/cache/stats	GET	缓存命中统计
/sessions/{id}	GET	会话详情
/sessions/{id}/history	GET	历史消息

技术栈

层次	技术
Agent 框架	LangGraph + LangChain
LLM	通义千问 Qwen-Turbo (DashScope)
向量数据库	ChromaDB (本地持久化)
嵌入模型	BGE-small-zh-v1.5 (512d)
关键词检索	BM25 (自实现) + jieba
前端 UI	Streamlit
API 层	FastAPI + SSE
缓存	三层自研 + Redis (可选)
可观测性	LangSmith Tracing + RotatingFileHandler
部署	Docker + docker-compose

项目结构

langgraph-reply/
├── main.py                 # LangGraph 工作流核心
│   ├── AgentState          #   状态定义
│   ├── FastBM25            #   BM25 检索器
│   ├── IntelligentCacheManager  # 三层缓存
│   ├── SessionManager      #   会话管理
│   └── graph               #   编译后的工作流
├── api.py                  # FastAPI 接口层
├── app.py                  # Streamlit 前端
├── security_module.py      # 安全模块（限流+过滤+检测）
├── health.py               # 健康检查
├── logger.py               # 日志配置
├── config.py               # 环境变量配置
├── test_rag_benchmark.py   # RAG 检索效果量化测试
├── tests/
│   ├── test_cache.py       #   缓存单元测试
│   ├── test_session.py     #   会话管理单元测试
│   └── test_retrieve.py    #   检索系统单元测试
├── knowledge/
│   └── tea.md              # 结构化茶叶知识库
├── long_memory/            # 用户记忆持久化目录
├── Dockerfile
├── docker-compose.yml
├── .env.example
└── README.md

缓存命中率

场景：50 条测试查询（含重复/相似/无关三类）
L1 命中率: 24%  (12/50)
L2 命中率: 38%  (19/50)
L3 命中率: 12%  (6/50)
总命中率:  74%  (37/50)
平均响应:  80ms (命中) / 1200ms (未命中)

License

MIT