实战案例1：智能客服机器人

┌─────────────────────────────────────────────────────────────────┐
│                           客户端层                               │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐  │
│  │ 用户聊天界面 │  │ 客服工作台  │  │      管理后台            │  │
│  │  (Vue3)     │  │   (Vue3)    │  │      (Vue3)              │  │
│  └──────┬──────┘  └──────┬──────┘  └───────────┬─────────────┘  │
└─────────┼────────────────┼────────────────────┼────────────────┘
          │                │                    │
          └────────────────┴────────────────────┘
                           │ HTTPS/WSS
┌──────────────────────────┼──────────────────────────────────────┐
│                          ▼                                      │
│                      Nginx 反向代理                              │
│                    (SSL终止/负载均衡)                             │
└──────────────────────────┬──────────────────────────────────────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
┌─────────────────┐ ┌─────────────┐ ┌─────────────────┐
│   前端静态资源   │ │  REST API   │ │  WebSocket服务  │
│   (Nginx托管)   │ │   FastAPI   │ │    FastAPI      │
└─────────────────┘ └──────┬──────┘ └─────────────────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
┌─────────────────┐ ┌─────────────┐ ┌─────────────────┐
│   RAG问答引擎    │ │  对话管理    │ │   人工接管服务   │
│  (LangChain)    │ │   Service   │ │    Service      │
└─────────────────┘ └──────┬──────┘ └─────────────────┘
                           │
                           ▼
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
┌─────────────────┐ ┌─────────────┐ ┌─────────────────┐
│   向量数据库     │ │   LLM API   │ │    关系数据库    │
│    (Chroma)     │ │ OpenAI/国产  │ │   (SQLite/PG)   │
└─────────────────┘ └─────────────┘ └─────────────────┘

# 1. 访问 https://www.python.org/downloads/ 下载 Python 3.10+
# 2. 安装时勾选 "Add Python to PATH"
# 3. 验证安装
python --version  # 应显示 Python 3.10.x 或更高版本

# 安装 pip 包管理器（通常已包含）
python -m ensurepip --upgrade

# 使用 Homebrew 安装
brew install python@3.10

# 验证安装
python3 --version

# 访问 https://nodejs.org/ 下载 LTS 版本
# 或使用 nvm 管理多版本

# Windows 使用 nvm-windows
nvm install 18.17.0
nvm use 18.17.0

# macOS/Linux 使用 nvm
curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh | bash
nvm install 18
nvm use 18

# 验证安装
node --version  # v18.x.x
npm --version   # 9.x.x

# 创建项目根目录
mkdir customer-service-bot
cd customer-service-bot

# 创建后端目录
mkdir backend
cd backend

# 创建Python虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS/Linux:
source venv/bin/activate

# 升级pip
pip install --upgrade pip

# 安装依赖包
pip install fastapi uvicorn langchain langchain-openai chromadb pydantic python-multipart python-dotenv websockets sqlalchemy alembic

# 生成依赖清单
pip freeze > requirements.txt

backend/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI主入口
│   ├── config.py            # 配置文件
│   ├── models.py            # 数据模型
│   ├── database.py          # 数据库连接
│   ├── dependencies.py      # 依赖注入
│   ├── services/
│   │   ├── __init__.py
│   │   ├── knowledge_base.py    # 知识库服务
│   │   ├── chat_service.py      # 对话服务
│   │   ├── llm_service.py       # LLM服务
│   │   └── handover.py          # 人工接管
│   └── routers/
│       ├── __init__.py
│       ├── chat.py          # 聊天接口
│       ├── admin.py         # 管理接口
│       └── websocket.py     # WebSocket接口
├── data/
│   ├── knowledge/           # 知识库文档存储
│   └── chroma_db/           # Chroma向量数据库
├── logs/                    # 日志目录
├── tests/                   # 测试目录
├── .env                     # 环境变量
├── requirements.txt         # Python依赖
└── Dockerfile               # Docker镜像配置

"""
配置文件模块 - 管理所有系统配置参数
使用 pydantic_settings 实现配置验证和环境变量加载
"""

import os
from typing import Optional
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """系统配置类 - 所有配置项均可通过环境变量覆盖"""
    
    # ========== 应用基础配置 ==========
    APP_NAME: str = "智能客服机器人系统"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False
    SECRET_KEY: str = "your-secret-key-change-in-production"
    
    # ========== 服务器配置 ==========
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    WORKERS: int = 1
    
    # ========== LLM API 配置 ==========
    OPENAI_API_KEY: Optional[str] = None
    OPENAI_BASE_URL: str = "https://api.openai.com/v1"
    OPENAI_MODEL: str = "gpt-4o-mini"
    
    ZHIPU_API_KEY: Optional[str] = None
    ZHIPU_MODEL: str = "glm-4"
    
    LLM_PROVIDER: str = "zhipu"  # 可选: openai, zhipu
    
    # ========== 向量数据库配置 ==========
    CHROMA_PERSIST_DIRECTORY: str = "./data/chroma_db"
    CHROMA_COLLECTION_NAME: str = "knowledge_base"
    EMBEDDING_MODEL: str = "BAAI/bge-large-zh"
    TOP_K: int = 5
    SIMILARITY_THRESHOLD: float = 0.6
    
    # ========== 对话系统配置 ==========
    MAX_HISTORY: int = 10
    MAX_TOKENS: int = 2000
    TEMPERATURE: float = 0.7
    SESSION_TIMEOUT: int = 3600
    
    # ========== 人工接管配置 ==========
    HANDOVER_KEYWORDS: list = ["人工", "客服", "投诉", "找人工", "转人工"]
    HANDOVER_CONFIDENCE_THRESHOLD: float = 0.3
    
    # ========== 数据库配置 ==========
    DATABASE_URL: str = "sqlite:///./data/app.db"
    
    # ========== 日志配置 ==========
    LOG_LEVEL: str = "INFO"
    LOG_FILE: str = "./logs/app.log"
    
    # ========== CORS配置 ==========
    CORS_ORIGINS: list = ["http://localhost:5173", "http://localhost:3000"]
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        case_sensitive = True


@lru_cache()
def get_settings():
    return Settings()


settings = get_settings()

"""数据模型定义 - Pydantic模型和SQLAlchemy ORM模型"""

from datetime import datetime
from typing import List, Optional, Dict, Any
from enum import Enum
from pydantic import BaseModel, Field
from sqlalchemy import Column, String, DateTime, Text, Integer, Float, Boolean, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import uuid

# SQLAlchemy ORM 模型
Base = declarative_base()


class ConversationDB(Base):
    """对话记录数据库模型"""
    __tablename__ = "conversations"
    
    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    session_id = Column(String(36), index=True, nullable=False)
    user_id = Column(String(50), nullable=True)
    role = Column(String(20), nullable=False)  # user, assistant, human, system
    content = Column(Text, nullable=False)
    message_type = Column(String(20), default="text")
    created_at = Column(DateTime, default=datetime.utcnow)
    is_deleted = Column(Boolean, default=False)


class KnowledgeDocDB(Base):
    """知识库文档数据库模型"""
    __tablename__ = "knowledge_docs"
    
    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    filename = Column(String(255), nullable=False)
    title = Column(String(255), nullable=True)
    file_type = Column(String(20), nullable=False)
    file_path = Column(String(500), nullable=False)
    file_size = Column(Integer, default=0)
    status = Column(String(20), default="pending")
    chunk_count = Column(Integer, default=0)
    uploaded_at = Column(DateTime, default=datetime.utcnow)


# Pydantic 模型（API请求/响应）
class ChatMessage(BaseModel):
    """单条聊天消息模型"""
    role: str
    content: str = Field(..., min_length=1, max_length=10000)
    message_type: str = Field(default="text")
    timestamp: Optional[datetime] = None


class ChatRequest(BaseModel):
    """聊天请求模型"""
    session_id: Optional[str] = None
    message: str = Field(..., min_length=1, max_length=5000)
    user_id: Optional[str] = None


class ChatResponse(BaseModel):
    """聊天响应模型"""
    session_id: str
    message: ChatMessage
    sources: Optional[List[Dict[str, Any]]] = None
    confidence: float = 0.0
    handover_triggered: bool = False
    handover_reason: Optional[str] = None


def init_database(database_url: str):
    """初始化数据库连接"""
    engine = create_engine(
        database_url,
        connect_args={"check_same_thread": False} if "sqlite" in database_url else {},
        echo=False
    )
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base.metadata.create_all(bind=engine)
    return engine, SessionLocal

"""知识库服务模块 - 负责文档处理、向量化和检索"""

import os
import logging
from typing import List, Dict, Any, Optional
from pathlib import Path

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

logger = logging.getLogger(__name__)


class KnowledgeBaseService:
    """知识库服务类 - 提供文档加载、切分、向量化和检索"""
    
    def __init__(self, persist_directory: str = "./data/chroma_db",
                 collection_name: str = "knowledge_base",
                 embedding_model: str = "BAAI/bge-large-zh"):
        self.persist_directory = persist_directory
        self.collection_name = collection_name
        self.embedding_model_name = embedding_model
        
        os.makedirs(persist_directory, exist_ok=True)
        
        self._embeddings = None
        self._vector_store = None
        
        # 文本分割器配置
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            length_function=len,
            separators=["\n\n", "\n", "。", ".", "!", "?", "；", " "],
        )
    
    @property
    def embeddings(self):
        """懒加载嵌入模型"""
        if self._embeddings is None:
            self._embeddings = HuggingFaceEmbeddings(
                model_name=self.embedding_model_name,
                model_kwargs={"device": "cpu"},
                encode_kwargs={"normalize_embeddings": True},
            )
        return self._embeddings
    
    @property
    def vector_store(self):
        """懒加载向量数据库"""
        if self._vector_store is None:
            self._vector_store = Chroma(
                collection_name=self.collection_name,
                embedding_function=self.embeddings,
                persist_directory=self.persist_directory,
            )
        return self._vector_store
    
    def load_document(self, file_path: str):
        """根据文件类型加载文档"""
        file_path = Path(file_path)
        suffix = file_path.suffix.lower()
        
        if suffix == ".pdf":
            loader = PyPDFLoader(str(file_path))
        elif suffix in [".txt", ".md"]:
            loader = TextLoader(str(file_path), encoding="utf-8")
        else:
            raise ValueError(f"不支持的文件类型: {suffix}")
        
        return loader.load()
    
    def split_documents(self, documents):
        """将文档切分为文本块"""
        for doc in documents:
            if not hasattr(doc, 'metadata'):
                doc.metadata = {}
        return self.text_splitter.split_documents(documents)
    
    def add_document(self, file_path: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None):
        """添加文档到知识库"""
        documents = self.load_document(file_path)
        chunks = self.split_documents(documents)
        
        for i, chunk in enumerate(chunks):
            chunk.metadata.update({
                "doc_id": doc_id or Path(file_path).stem,
                "chunk_index": i,
                "source_file": str(file_path),
                **(metadata or {})
            })
        
        chunk_ids = self.vector_store.add_documents(chunks)
        self.vector_store.persist()
        
        return len(chunks), chunk_ids
    
    def similarity_search(self, query: str, k: int = 5, threshold: float = 0.6):
        """相似度搜索 - RAG核心检索功能"""
        results = self.vector_store.similarity_search_with_score(query, k=k)
        
        filtered_results = []
        for doc, score in results:
            similarity = 1 - score  # 转换为相似度
            if similarity >= threshold:
                doc.metadata["similarity"] = similarity
                filtered_results.append((doc, similarity))
        
        return filtered_results


# 单例模式
_knowledge_base = None

def get_knowledge_base():
    global _knowledge_base
    if _knowledge_base is None:
        _knowledge_base = KnowledgeBaseService()
    return _knowledge_base

"""对话服务模块 - RAG问答核心逻辑"""

import logging
from typing import List, Dict, Optional
from datetime import datetime
import uuid

from app.config import settings
from app.services.knowledge_base import get_knowledge_base
from app.services.llm_service import get_llm_service

logger = logging.getLogger(__name__)


class ChatService:
    """对话服务类 - 处理用户输入到AI回复的完整流程"""
    
    def __init__(self):
        self.kb = get_knowledge_base()
        self.llm = get_llm_service()
        self.sessions: Dict[str, List[Dict]] = {}
        
        self.system_prompt = """你是智能客服助手，基于以下知识回答用户问题。

【知识库内容】
{context}

【对话历史】
{history}

请基于以上信息回答，如果不确定请诚实告知。"""
    
    def _get_session_history(self, session_id: str):
        return self.sessions.get(session_id, [])
    
    def _add_to_history(self, session_id: str, role: str, content: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        self.sessions[session_id].append({
            "role": role, "content": content,
            "timestamp": datetime.utcnow().isoformat()
        })
        # 保留最近N轮
        max_history = settings.MAX_HISTORY * 2
        if len(self.sessions[session_id]) > max_history:
            self.sessions[session_id] = self.sessions[session_id][-max_history:]
    
    def _format_context(self, search_results):
        if not search_results:
            return "（知识库中暂无相关信息）"
        parts = []
        for i, (doc, score) in enumerate(search_results, 1):
            parts.append(f"[{i}] {doc.page_content}")
        return "\n".join(parts)
    
    def _check_handover_keywords(self, message: str):
        message_lower = message.lower()
        for keyword in settings.HANDOVER_KEYWORDS:
            if keyword in message_lower:
                return True, f"触发关键词: {keyword}"
        return False, None
    
    async def chat(self, message: str, session_id: Optional[str] = None, user_id: Optional[str] = None):
        """处理用户对话请求"""
        session_id = session_id or str(uuid.uuid4())
        
        # 检查转人工关键词
        handover_triggered, handover_reason = self._check_handover_keywords(message)
        if handover_triggered:
            return {
                "session_id": session_id,
                "message": {"role": "assistant", "content": "正在为您转接人工客服..."},
                "handover_triggered": True,
                "handover_reason": handover_reason,
                "confidence": 0.0
            }
        
        # 检索相关知识
        search_results = self.kb.similarity_search(
            query=message, k=settings.TOP_K, threshold=settings.SIMILARITY_THRESHOLD
        )
        
        confidence = sum(s for _, s in search_results) / len(search_results) if search_results else 0.0
        
        # 组装提示词
        system_msg = self.system_prompt.format(
            context=self._format_context(search_results),
            history=str(self._get_session_history(session_id))
        )
        
        messages = [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": message}
        ]
        
        # 调用LLM
        try:
            response_text = await self.llm.chat(messages)
        except Exception as e:
            response_text = "抱歉，服务暂时不可用。"
            confidence = 0.0
        
        # 保存历史
        self._add_to_history(session_id, "user", message)
        self._add_to_history(session_id, "assistant", response_text)
        
        return {
            "session_id": session_id,
            "message": {"role": "assistant", "content": response_text},
            "confidence": round(confidence, 3),
            "handover_triggered": handover_triggered,
            "sources": [{"content": d.page_content[:100], "score": s} for d, s in search_results]
        }


# 单例
_chat_service = None

def get_chat_service():
    global _chat_service
    if _chat_service is None:
        _chat_service = ChatService()
    return _chat_service

"""FastAPI应用主入口"""

import os
import sys
import logging
from contextlib import asynccontextmanager

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from app.config import settings
from app.models import init_database
from app.routers import chat, admin, websocket

# 配置日志
os.makedirs("./logs", exist_ok=True)
logging.basicConfig(
    level=getattr(logging, settings.LOG_LEVEL),
    format=settings.LOG_FORMAT,
    handlers=[
        logging.FileHandler(settings.LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    logger.info(f"启动 {settings.APP_NAME} v{settings.APP_VERSION}")
    engine, SessionLocal = init_database(settings.DATABASE_URL)
    app.state.engine = engine
    app.state.SessionLocal = SessionLocal
    os.makedirs("./data/knowledge", exist_ok=True)
    os.makedirs("./data/chroma_db", exist_ok=True)
    yield
    logger.info("应用关闭")


app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    description="智能客服机器人系统API",
    lifespan=lifespan
)

# CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(chat.router, prefix="/api/chat", tags=["对话"])
app.include_router(admin.router, prefix="/api/admin", tags=["管理"])
app.include_router(websocket.router, prefix="/ws", tags=["WebSocket"])


@app.get("/health")
async def health_check():
    return {"status": "healthy", "app": settings.APP_NAME, "version": settings.APP_VERSION}


@app.get("/")
async def root():
    return {"message": f"欢迎使用 {settings.APP_NAME}", "docs": "/docs"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app.main:app", host=settings.HOST, port=settings.PORT, reload=settings.DEBUG)

# Web框架
fastapi==0.109.0
uvicorn[standard]==0.27.0
websockets==12.0
python-multipart==0.0.6
python-dotenv==1.0.0

# 数据库
sqlalchemy==2.0.25

# LangChain和LLM
langchain==0.1.4
langchain-community==0.0.16
langchain-openai==0.0.5

# 向量数据库
chromadb==0.4.22

# 文档处理
pypdf==4.0.0

# Embedding模型
sentence-transformers==2.2.2

# 工具库
numpy==1.26.3

# 创建项目
cd customer-service-bot
npm create vue@latest frontend
# 选择 Vue Router 和 Pinia

cd frontend
npm install
npm install element-plus axios marked @element-plus/icons-vue
npm run dev

import axios from 'axios'

const api = axios.create({
  baseURL: import.meta.env.VITE_API_URL || 'http://localhost:8000',
  timeout: 30000,
  headers: { 'Content-Type': 'application/json' }
})

api.interceptors.response.use(
  (response) => response.data,
  (error) => Promise.reject(new Error(error.response?.data?.detail || error.message))
)

export const chatApi = {
  sendMessage: (data) => api.post('/api/chat/send', data),
  clearSession: (id) => api.post(`/api/chat/session/${id}/clear`)
}

export default api

import { defineStore } from 'pinia'
import { ref, computed } from 'vue'
import { chatApi } from '@/api'

export const useChatStore = defineStore('chat', () => {
  const sessionId = ref(null)
  const messages = ref([])
  const isLoading = ref(false)

  const hasActiveSession = computed(() => !!sessionId.value && messages.value.length > 0)

  function initSession() {
    sessionId.value = 'sess_' + Math.random().toString(36).substr(2, 9)
    messages.value = []
  }

  function addMessage(msg) {
    messages.value.push({ id: Date.now(), timestamp: new Date().toISOString(), ...msg })
  }

  async function sendMessage(content) {
    if (!sessionId.value) initSession()
    addMessage({ role: 'user', content })
    isLoading.value = true
    
    try {
      const res = await chatApi.sendMessage({ session_id: sessionId.value, message: content })
      addMessage({ role: 'assistant', content: res.message.content, sources: res.sources })
      return res
    } finally {
      isLoading.value = false
    }
  }

  return { sessionId, messages, isLoading, hasActiveSession, sendMessage }
})

<template>
  <div class="chat-window">
    <div class="chat-header">
      <h3>智能客服助手</h3>
      <el-button size="small" @click="clear">清空</el-button>
    </div>
    
    <div class="messages">
      <div v-for="msg in messages" :key="msg.id" :class="msg.role">
        {{ msg.content }}
      </div>
    </div>
    
    <div class="input-area">
      <el-input v-model="input" @keyup.enter="send" />
      <el-button type="primary" @click="send" :loading="isLoading">发送</el-button>
    </div>
  </div>
</template>

<script setup>
import { ref } from 'vue'
import { useChatStore } from '@/stores/chat'

const store = useChatStore()
const input = ref('')
const { messages, isLoading, sendMessage, clearSession } = store

async function send() {
  if (!input.value.trim()) return
  await sendMessage(input.value)
  input.value = ''
}

function clear() {
  clearSession()
}
</script>

version: '3.8'
services:
  backend:
    build: ./backend
    ports:
      - "8000:8000"
    environment:
      - ZHIPU_API_KEY=${ZHIPU_API_KEY}
    volumes:
      - ./data:/app/data
  
  frontend:
    build: ./frontend
    ports:
      - "80:80"
    depends_on:
      - backend

server {
    listen 80;
    
    location / {
        root /var/www/frontend/dist;
        try_files $uri $uri/ /index.html;
    }
    
    location /api/ {
        proxy_pass http://backend:8000/;
    }
    
    location /ws/ {
        proxy_pass http://backend:8000/ws/;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}