feat: - 自建知识库

KB_README.md

@@ -0,0 +1,145 @@
+# 知识库管理系统 - 环境变量配置
+
+## 新增环境变量
+
+在 `chat-ai-api/.env` 文件中添加以下配置：
+
+```env
+# ========== 阿里云 OSS 配置 ==========
+ALIYUN_OSS_ACCESS_KEY_ID=your_access_key_id
+ALIYUN_OSS_ACCESS_KEY_SECRET=your_access_key_secret
+ALIYUN_OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com
+ALIYUN_OSS_BUCKET=chat-ai-knowledge
+
+# ========== Qdrant 向量数据库配置 ==========
+QDRANT_HOST=localhost
+QDRANT_PORT=6333
+QDRANT_COLLECTION=kb_vectors
+
+# ========== 向量模型配置 ==========
+# 使用豆包的 doubao-embedding-vision-251215 向量模型
+# 无需额外配置，使用现有的 ARK_API_KEY 和 ARK_BASE_URL
+```
+
+## MongoDB 集合说明
+
+系统会自动创建以下集合（在 `arklogs` 数据库中）：
+
+1. **kb_documents** - 文档元数据
+   - tenant_id: 租户ID（appName）
+   - filename: 文件名
+   - file_type: 文件类型（.xlsx, .docx, .pdf）
+   - file_size: 文件大小
+   - oss_path: 阿里云OSS路径
+   - status: 处理状态（pending, processing, completed, failed）
+   - chunk_count: 切片数量
+   - created_at, updated_at, deleted
+
+2. **kb_chunks** - 文档切片
+   - doc_id: 所属文档ID
+   - tenant_id: 租户ID
+   - chunk_index: 切片索引
+   - content: 切片内容
+   - vector_id: Qdrant中的向量ID
+   - metadata: 元数据
+   - created_at, deleted
+
+3. **kb_tenants** - 租户配置
+   - tenant_id: 租户ID（appName）
+   - agent_enabled: 智能客服是否开启
+   - agent_name: 客服名称
+   - agent_greeting: 欢迎语
+   - created_at, updated_at, deleted
+
+## Qdrant 安装和启动
+
+### Docker 方式（推荐）
+```bash
+docker run -d -p 6333:6333 \
+  -v $(pwd)/qdrant_storage:/qdrant/storage \
+  --name qdrant \
+  qdrant/qdrant
+```
+
+### 验证 Qdrant 运行
+```bash
+curl http://localhost:6333/collections
+```
+
+## 阿里云 OSS Bucket 配置
+
+1. 登录阿里云控制台
+2. 创建 Bucket：`chat-ai-knowledge`
+3. 设置权限：私有读写
+4. 创建 RAM 用户并获取 AccessKey
+5. 为 RAM 用户授予 OSS 读写权限
+
+## API 接口说明
+
+### 知识库管理接口（/main/kb）
+
+#### 文档管理
+- `POST /main/kb/documents/upload` - 上传文档
+- `GET /main/kb/documents` - 获取文档列表
+- `GET /main/kb/documents/{doc_id}` - 获取文档详情（含切片）
+- `DELETE /main/kb/documents/{doc_id}` - 删除文档
+
+#### 智能客服配置
+- `GET /main/kb/agent/config` - 获取智能客服配置
+- `POST /main/kb/agent/config` - 更新智能客服配置
+
+#### 统计
+- `GET /main/kb/stats` - 获取知识库统计数据
+
+## 嵌入地址格式
+
+新的嵌入地址携带租户信息：
+
+```
+https://aichat.zhongsou.com/?token=<用户token>&source=app&tenant=<租户ID>
+```
+
+例如：
+```
+https://aichat.zhongsou.com/?token=237ea8d7-a042-4f1c-b084-281f3fd4eab1&source=app&tenant=com.xiaoyouhui
+```
+
+参数说明：
+- `token`: 用户登录令牌
+- `source`: 来源标识（app表示来自App）
+- `tenant`: 租户ID（等同于appName，如 com.xiaoyouhui）
+
+## 工作流程
+
+### 文档上传流程
+1. 用户上传文档（Excel/Word/PDF）
+2. 文件上传到阿里云 OSS（路径：`{tenant_id}/documents/{timestamp}_{filename}`）
+3. 后台任务开始处理：
+   - 提取文本内容
+   - 智能切片（800字符/片，100字符重叠）
+   - 调用豆包 embedding API 生成向量
+   - 向量存储到 Qdrant
+   - 切片元数据存储到 MongoDB
+4. 文档状态更新为 completed
+
+### 聊天检索流程
+1. 用户发送消息
+2. 系统根据租户ID检索自建知识库：
+   - 用户消息向量化
+   - Qdrant 相似度搜索（Top 5）
+   - 获取完整切片内容
+3. 将检索结果注入到 system prompt
+4. 模型基于知识库内容生成回答
+5. 知识库不足时才触发联网搜索
+
+## 多租户隔离
+
+- **文件存储隔离**: OSS 使用租户前缀（`{tenant_id}/`）
+- **向量检索隔离**: Qdrant 搜索时过滤 `tenant_id`
+- **数据库隔离**: MongoDB 所有文档和切片都带 `tenant_id` 字段
+
+## AI 配置新增字段
+
+在 `ai_config` 集合中新增字段：
+- `use_custom_kb`: 是否使用自建知识库（默认 true）
+- 保留 `knowledgeId` 用于兼容豆包原生知识库

app/core/oss_client.py

@@ -0,0 +1,97 @@
+"""
+阿里云 OSS 客户端
+用于上传和管理租户的知识库文档
+采用"单 Bucket + 租户前缀"的方式
+"""
+import oss2
+import os
+from typing import Optional, Tuple
+from datetime import datetime
+
+
+class OSSClient:
+    def __init__(self):
+        # 从环境变量读取阿里云 OSS 配置
+        self.access_key_id = os.getenv("ALIYUN_OSS_ACCESS_KEY_ID")
+        self.access_key_secret = os.getenv("ALIYUN_OSS_ACCESS_KEY_SECRET")
+        self.endpoint = os.getenv("ALIYUN_OSS_ENDPOINT", "oss-cn-hangzhou.aliyuncs.com")
+        self.bucket_name = os.getenv("ALIYUN_OSS_BUCKET", "chat-ai-knowledge")
+
+        # 初始化 OSS 认证和 Bucket
+        auth = oss2.Auth(self.access_key_id, self.access_key_secret)
+        self.bucket = oss2.Bucket(auth, self.endpoint, self.bucket_name)
+
+    def upload_file(self, tenant_id: str, file_content: bytes, filename: str) -> Tuple[bool, str]:
+        """
+        上传文件到 OSS
+        :param tenant_id: 租户ID (如 com.xiaoyouhui)
+        :param file_content: 文件内容（字节）
+        :param filename: 原始文件名
+        :return: (是否成功, OSS路径或错误信息)
+        """
+        try:
+            # 构造 OSS 路径: {tenant_id}/documents/{timestamp}_{filename}
+            timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
+            oss_path = f"{tenant_id}/documents/{timestamp}_{filename}"
+
+            # 上传文件
+            result = self.bucket.put_object(oss_path, file_content)
+
+            if result.status == 200:
+                return True, oss_path
+            else:
+                return False, f"上传失败，状态码: {result.status}"
+        except Exception as e:
+            return False, f"上传异常: {str(e)}"
+
+    def download_file(self, oss_path: str) -> Optional[bytes]:
+        """
+        从 OSS 下载文件
+        :param oss_path: OSS 中的文件路径
+        :return: 文件内容（字节）或 None
+        """
+        try:
+            result = self.bucket.get_object(oss_path)
+            return result.read()
+        except Exception as e:
+            print(f"下载文件失败: {str(e)}")
+            return None
+
+    def delete_file(self, oss_path: str) -> bool:
+        """
+        删除 OSS 中的文件
+        :param oss_path: OSS 中的文件路径
+        :return: 是否成功
+        """
+        try:
+            self.bucket.delete_object(oss_path)
+            return True
+        except Exception as e:
+            print(f"删除文件失败: {str(e)}")
+            return False
+
+    def get_file_url(self, oss_path: str, expires: int = 3600) -> Optional[str]:
+        """
+        生成文件的临时访问 URL
+        :param oss_path: OSS 中的文件路径
+        :param expires: 过期时间（秒），默认1小时
+        :return: 临时访问URL
+        """
+        try:
+            url = self.bucket.sign_url('GET', oss_path, expires)
+            return url
+        except Exception as e:
+            print(f"生成URL失败: {str(e)}")
+            return None
+
+
+# 全局单例
+_oss_client = None
+
+
+def get_oss_client() -> OSSClient:
+    """获取 OSS 客户端单例"""
+    global _oss_client
+    if _oss_client is None:
+        _oss_client = OSSClient()
+    return _oss_client

app/core/qdrant_client.py

@@ -0,0 +1,137 @@
+"""
+Qdrant 向量数据库客户端
+用于存储和检索文档切片的向量嵌入
+"""
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
+from typing import List, Dict, Optional
+import os
+import uuid
+
+
+class QdrantManager:
+    def __init__(self):
+        # 从环境变量读取 Qdrant 配置
+        self.host = os.getenv("QDRANT_HOST", "localhost")
+        self.port = int(os.getenv("QDRANT_PORT", "6333"))
+        self.collection_name = os.getenv("QDRANT_COLLECTION", "kb_vectors")
+        self.vector_size = 1536  # doubao-embedding-vision-251215 的向量维度
+
+        # 初始化客户端
+        self.client = QdrantClient(host=self.host, port=self.port)
+
+        # 确保集合存在
+        self._ensure_collection()
+
+    def _ensure_collection(self):
+        """确保 Qdrant 集合存在，不存在则创建"""
+        collections = self.client.get_collections().collections
+        collection_names = [col.name for col in collections]
+
+        if self.collection_name not in collection_names:
+            self.client.create_collection(
+                collection_name=self.collection_name,
+                vectors_config=VectorParams(
+                    size=self.vector_size,
+                    distance=Distance.COSINE
+                )
+            )
+            print(f"创建 Qdrant 集合: {self.collection_name}")
+
+    def add_vectors(self, tenant_id: str, doc_id: str, chunks: List[Dict]) -> List[str]:
+        """
+        添加文档切片的向量到 Qdrant
+        :param tenant_id: 租户ID
+        :param doc_id: 文档ID
+        :param chunks: 切片列表 [{"content": "...", "vector": [...]}, ...]
+        :return: vector_id 列表
+        """
+        points = []
+        vector_ids = []
+
+        for idx, chunk in enumerate(chunks):
+            vector_id = str(uuid.uuid4())
+            vector_ids.append(vector_id)
+
+            point = PointStruct(
+                id=vector_id,
+                vector=chunk["vector"],
+                payload={
+                    "tenant_id": tenant_id,
+                    "doc_id": doc_id,
+                    "chunk_index": idx,
+                    "content": chunk["content"][:500],  # 存储前500字符用于预览
+                }
+            )
+            points.append(point)
+
+        # 批量插入
+        self.client.upsert(
+            collection_name=self.collection_name,
+            points=points
+        )
+
+        return vector_ids
+
+    def search_vectors(self, tenant_id: str, query_vector: List[float], top_k: int = 5) -> List[Dict]:
+        """
+        向量相似度搜索
+        :param tenant_id: 租户ID（只搜索该租户的数据）
+        :param query_vector: 查询向量
+        :param top_k: 返回Top K个结果
+        :return: 搜索结果列表
+        """
+        search_result = self.client.search(
+            collection_name=self.collection_name,
+            query_vector=query_vector,
+            query_filter=Filter(
+                must=[
+                    FieldCondition(
+                        key="tenant_id",
+                        match=MatchValue(value=tenant_id)
+                    )
+                ]
+            ),
+            limit=top_k
+        )
+
+        results = []
+        for hit in search_result:
+            results.append({
+                "vector_id": hit.id,
+                "score": hit.score,
+                "content": hit.payload.get("content", ""),
+                "doc_id": hit.payload.get("doc_id", ""),
+                "chunk_index": hit.payload.get("chunk_index", 0),
+            })
+
+        return results
+
+    def delete_vectors_by_doc(self, doc_id: str):
+        """
+        删除文档的所有向量
+        :param doc_id: 文档ID
+        """
+        self.client.delete(
+            collection_name=self.collection_name,
+            points_selector=Filter(
+                must=[
+                    FieldCondition(
+                        key="doc_id",
+                        match=MatchValue(value=doc_id)
+                    )
+                ]
+            )
+        )
+
+
+# 全局单例
+_qdrant_manager = None
+
+
+def get_qdrant_manager() -> QdrantManager:
+    """获取 Qdrant 管理器单例"""
+    global _qdrant_manager
+    if _qdrant_manager is None:
+        _qdrant_manager = QdrantManager()
+    return _qdrant_manager

app/db/knowledge_base.py

@@ -0,0 +1,217 @@
+"""
+知识库管理 - MongoDB 操作层
+存储文档元数据、切片(chunks)、租户配置
+"""
+from pymongo import MongoClient
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from bson.objectid import ObjectId
+import os
+
+# MongoDB 连接
+MONGO_URI = os.getenv("ARK_LOGS_MONGO_URI", "mongodb://localhost:27017/")
+client = MongoClient(MONGO_URI)
+db = client["arklogs"]
+
+# 集合定义
+documents_collection = db["kb_documents"]  # 文档元数据
+chunks_collection = db["kb_chunks"]  # 文档切片
+tenants_collection = db["kb_tenants"]  # 租户配置
+
+
+# ==================== 租户管理 ====================
+
+def get_tenant_config(tenant_id: str) -> Optional[Dict]:
+    """获取租户配置"""
+    return tenants_collection.find_one({"tenant_id": tenant_id, "deleted": False})
+
+
+def upsert_tenant_config(tenant_id: str, config: Dict) -> bool:
+    """创建或更新租户配置"""
+    config.update({
+        "tenant_id": tenant_id,
+        "updated_at": datetime.utcnow(),
+    })
+    result = tenants_collection.update_one(
+        {"tenant_id": tenant_id},
+        {"$set": config, "$setOnInsert": {"created_at": datetime.utcnow(), "deleted": False}},
+        upsert=True
+    )
+    return result.acknowledged
+
+
+def get_agent_config(tenant_id: str) -> Optional[Dict]:
+    """获取租户的智能客服配置"""
+    tenant = get_tenant_config(tenant_id)
+    if not tenant:
+        return None
+    return {
+        "enabled": tenant.get("agent_enabled", False),
+        "name": tenant.get("agent_name", "智能小助手"),
+        "greeting": tenant.get("agent_greeting", "您好！我是智能客服小助手，有什么可以帮您的吗？"),
+        "h5_url": f"https://aichat.zhongsou.com/?tenant={tenant_id}",
+    }
+
+
+def update_agent_config(tenant_id: str, enabled: bool, name: str, greeting: str) -> bool:
+    """更新智能客服配置"""
+    result = tenants_collection.update_one(
+        {"tenant_id": tenant_id},
+        {
+            "$set": {
+                "agent_enabled": enabled,
+                "agent_name": name,
+                "agent_greeting": greeting,
+                "updated_at": datetime.utcnow(),
+            }
+        },
+        upsert=True
+    )
+    return result.acknowledged
+
+
+# ==================== 文档管理 ====================
+
+def create_document(tenant_id: str, filename: str, file_type: str, file_size: int, oss_path: str) -> str:
+    """创建文档记录"""
+    doc = {
+        "tenant_id": tenant_id,
+        "filename": filename,
+        "file_type": file_type,
+        "file_size": file_size,
+        "oss_path": oss_path,
+        "status": "pending",  # pending, processing, completed, failed
+        "chunk_count": 0,
+        "created_at": datetime.utcnow(),
+        "updated_at": datetime.utcnow(),
+        "deleted": False,
+    }
+    result = documents_collection.insert_one(doc)
+    return str(result.inserted_id)
+
+
+def get_document_by_id(doc_id: str) -> Optional[Dict]:
+    """根据ID获取文档"""
+    try:
+        doc = documents_collection.find_one({"_id": ObjectId(doc_id), "deleted": False})
+        if doc:
+            doc["_id"] = str(doc["_id"])
+        return doc
+    except:
+        return None
+
+
+def list_documents(tenant_id: str, skip: int = 0, limit: int = 100) -> List[Dict]:
+    """列出租户的所有文档"""
+    cursor = documents_collection.find(
+        {"tenant_id": tenant_id, "deleted": False}
+    ).sort("created_at", -1).skip(skip).limit(limit)
+
+    docs = []
+    for doc in cursor:
+        doc["_id"] = str(doc["_id"])
+        docs.append(doc)
+    return docs
+
+
+def count_documents(tenant_id: str) -> int:
+    """统计租户文档数量"""
+    return documents_collection.count_documents({"tenant_id": tenant_id, "deleted": False})
+
+
+def update_document_status(doc_id: str, status: str, chunk_count: int = 0, error_msg: str = None):
+    """更新文档处理状态"""
+    update_data = {
+        "status": status,
+        "updated_at": datetime.utcnow(),
+    }
+    if chunk_count > 0:
+        update_data["chunk_count"] = chunk_count
+    if error_msg:
+        update_data["error_msg"] = error_msg
+
+    documents_collection.update_one(
+        {"_id": ObjectId(doc_id)},
+        {"$set": update_data}
+    )
+
+
+def delete_document(doc_id: str) -> bool:
+    """软删除文档"""
+    result = documents_collection.update_one(
+        {"_id": ObjectId(doc_id)},
+        {"$set": {"deleted": True, "updated_at": datetime.utcnow()}}
+    )
+    return result.modified_count > 0
+
+
+# ==================== 切片管理 ====================
+
+def create_chunks(doc_id: str, tenant_id: str, chunks_data: List[Dict]) -> List[str]:
+    """批量创建文档切片"""
+    chunks = []
+    for idx, chunk_data in enumerate(chunks_data):
+        chunk = {
+            "doc_id": doc_id,
+            "tenant_id": tenant_id,
+            "chunk_index": idx,
+            "content": chunk_data["content"],
+            "vector_id": chunk_data.get("vector_id"),  # Qdrant中的point_id
+            "metadata": chunk_data.get("metadata", {}),
+            "created_at": datetime.utcnow(),
+            "deleted": False,
+        }
+        chunks.append(chunk)
+
+    result = chunks_collection.insert_many(chunks)
+    return [str(id) for id in result.inserted_ids]
+
+
+def get_chunks_by_doc(doc_id: str) -> List[Dict]:
+    """获取文档的所有切片"""
+    cursor = chunks_collection.find(
+        {"doc_id": doc_id, "deleted": False}
+    ).sort("chunk_index", 1)
+
+    chunks = []
+    for chunk in cursor:
+        chunk["_id"] = str(chunk["_id"])
+        chunks.append(chunk)
+    return chunks
+
+
+def delete_chunks_by_doc(doc_id: str) -> int:
+    """删除文档的所有切片"""
+    result = chunks_collection.update_many(
+        {"doc_id": doc_id},
+        {"$set": {"deleted": True}}
+    )
+    return result.modified_count
+
+
+def search_chunks_by_ids(chunk_ids: List[str]) -> List[Dict]:
+    """根据ID列表获取切片（用于向量检索后的内容获取）"""
+    chunks = []
+    for chunk_id in chunk_ids:
+        chunk = chunks_collection.find_one({"vector_id": chunk_id, "deleted": False})
+        if chunk:
+            chunk["_id"] = str(chunk["_id"])
+            chunks.append(chunk)
+    return chunks
+
+
+# ==================== 统计 ====================
+
+def get_tenant_stats(tenant_id: str) -> Dict:
+    """获取租户的知识库统计数据"""
+    total_docs = count_documents(tenant_id)
+    total_chunks = chunks_collection.count_documents({"tenant_id": tenant_id, "deleted": False})
+
+    # 获取最近上传的文档
+    recent_docs = list_documents(tenant_id, skip=0, limit=5)
+
+    return {
+        "total_documents": total_docs,
+        "total_chunks": total_chunks,
+        "recent_documents": recent_docs,
+    }

app/routers/knowledge_base.py

@@ -0,0 +1,325 @@
+"""
+知识库管理 API
+提供文档上传、查询、删除、智能客服配置等接口
+"""
+from fastapi import APIRouter, HTTPException, Depends, UploadFile, File, BackgroundTasks
+from typing import List, Optional
+from pydantic import BaseModel
+import os
+
+from ..db.knowledge_base import (
+    get_tenant_config, upsert_tenant_config, get_agent_config, update_agent_config,
+    create_document, get_document_by_id, list_documents, count_documents,
+    update_document_status, delete_document, create_chunks, get_chunks_by_doc,
+    delete_chunks_by_doc, get_tenant_stats
+)
+from ..core.oss_client import get_oss_client
+from ..core.qdrant_client import get_qdrant_manager
+from ..utils.document_processor import get_document_processor
+from ..utils.embedding_service import get_embedding_service
+from ..dependencies.auth import resolve_user_id, UserContext
+
+router = APIRouter()
+
+# 允许的文件类型
+ALLOWED_FILE_TYPES = ['.xlsx', '.xls', '.docx', '.doc', '.pdf']
+MAX_FILE_SIZE = 20 * 1024 * 1024  # 20MB
+
+
+# ==================== Pydantic 模型 ====================
+
+class AgentConfigUpdate(BaseModel):
+    enabled: bool
+    name: str
+    greeting: str
+
+
+class DocumentInfo(BaseModel):
+    id: str
+    filename: str
+    file_type: str
+    file_size: int
+    status: str
+    chunk_count: int
+    created_at: str
+
+
+class ChunkInfo(BaseModel):
+    chunk_index: int
+    content: str
+    vector_id: Optional[str]
+
+
+# ==================== 后台任务：文档处理 ====================
+
+async def process_document_task(doc_id: str, tenant_id: str, file_content: bytes, file_type: str):
+    """
+    后台任务：处理文档（提取文本、切片、向量化、存储）
+    """
+    try:
+        # 更新状态为处理中
+        update_document_status(doc_id, "processing")
+
+        # 1. 提取文本
+        processor = get_document_processor()
+        text = processor.extract_text(file_content, file_type)
+
+        if not text:
+            update_document_status(doc_id, "failed", error_msg="文本提取失败")
+            return
+
+        # 清理文本
+        text = processor.clean_text(text)
+
+        # 2. 智能切片
+        chunks = processor.split_text_into_chunks(text)
+
+        if not chunks:
+            update_document_status(doc_id, "failed", error_msg="文档切片失败")
+            return
+
+        # 3. 向量化（批量）
+        embedding_service = get_embedding_service()
+        chunk_contents = [chunk["content"] for chunk in chunks]
+        vectors = embedding_service.generate_embeddings(chunk_contents, tenant_id)
+
+        if len(vectors) != len(chunks):
+            update_document_status(doc_id, "failed", error_msg="向量化失败")
+            return
+
+        # 4. 存储到 Qdrant
+        qdrant = get_qdrant_manager()
+        chunks_with_vectors = [
+            {"content": chunk["content"], "vector": vector}
+            for chunk, vector in zip(chunks, vectors)
+        ]
+        vector_ids = qdrant.add_vectors(tenant_id, doc_id, chunks_with_vectors)
+
+        # 5. 存储切片元数据到 MongoDB
+        chunks_data = [
+            {
+                "content": chunk["content"],
+                "vector_id": vector_id,
+                "metadata": chunk.get("metadata", {})
+            }
+            for chunk, vector_id in zip(chunks, vector_ids)
+        ]
+        create_chunks(doc_id, tenant_id, chunks_data)
+
+        # 6. 更新文档状态为完成
+        update_document_status(doc_id, "completed", chunk_count=len(chunks))
+
+        print(f"文档处理完成: {doc_id}, 切片数: {len(chunks)}")
+
+    except Exception as e:
+        print(f"文档处理异常: {str(e)}")
+        update_document_status(doc_id, "failed", error_msg=str(e))
+
+
+# ==================== 文档管理接口 ====================
+
+@router.post("/documents/upload")
+async def upload_document(
+    background_tasks: BackgroundTasks,
+    file: UploadFile = File(...),
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """
+    上传文档到知识库
+    1. 上传原始文件到阿里云OSS
+    2. 后台任务：提取文本、切片、向量化、存储
+    """
+    tenant_id = user_context.app_name
+
+    # 验证文件类型
+    file_ext = os.path.splitext(file.filename)[1].lower()
+    if file_ext not in ALLOWED_FILE_TYPES:
+        raise HTTPException(status_code=400, detail=f"不支持的文件类型: {file_ext}")
+
+    # 读取文件内容
+    file_content = await file.read()
+    file_size = len(file_content)
+
+    # 验证文件大小
+    if file_size > MAX_FILE_SIZE:
+        raise HTTPException(status_code=400, detail=f"文件大小超出限制 ({MAX_FILE_SIZE // 1024 // 1024}MB)")
+
+    # 上传到 OSS
+    oss_client = get_oss_client()
+    success, oss_path = oss_client.upload_file(tenant_id, file_content, file.filename)
+
+    if not success:
+        raise HTTPException(status_code=500, detail=f"文件上传失败: {oss_path}")
+
+    # 创建文档记录
+    doc_id = create_document(
+        tenant_id=tenant_id,
+        filename=file.filename,
+        file_type=file_ext,
+        file_size=file_size,
+        oss_path=oss_path
+    )
+
+    # 后台处理文档
+    background_tasks.add_task(process_document_task, doc_id, tenant_id, file_content, file_ext)
+
+    return {
+        "success": True,
+        "doc_id": doc_id,
+        "message": "文档上传成功，正在后台处理..."
+    }
+
+
+@router.get("/documents")
+async def get_documents(
+    skip: int = 0,
+    limit: int = 100,
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """获取租户的所有文档列表"""
+    tenant_id = user_context.app_name
+    docs = list_documents(tenant_id, skip, limit)
+    total = count_documents(tenant_id)
+
+    return {
+        "success": True,
+        "total": total,
+        "documents": docs
+    }
+
+
+@router.get("/documents/{doc_id}")
+async def get_document_detail(
+    doc_id: str,
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """获取文档详情（包含切片）"""
+    tenant_id = user_context.app_name
+
+    # 获取文档信息
+    doc = get_document_by_id(doc_id)
+    if not doc:
+        raise HTTPException(status_code=404, detail="文档不存在")
+
+    # 验证租户权限
+    if doc["tenant_id"] != tenant_id:
+        raise HTTPException(status_code=403, detail="无权访问该文档")
+
+    # 获取切片
+    chunks = get_chunks_by_doc(doc_id)
+
+    # 获取原始文件URL（临时访问链接）
+    oss_client = get_oss_client()
+    file_url = oss_client.get_file_url(doc["oss_path"])
+
+    return {
+        "success": True,
+        "document": doc,
+        "chunks": chunks,
+        "file_url": file_url
+    }
+
+
+@router.delete("/documents/{doc_id}")
+async def delete_document_endpoint(
+    doc_id: str,
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """删除文档（包括OSS文件、向量、切片）"""
+    tenant_id = user_context.app_name
+
+    # 获取文档信息
+    doc = get_document_by_id(doc_id)
+    if not doc:
+        raise HTTPException(status_code=404, detail="文档不存在")
+
+    # 验证租户权限
+    if doc["tenant_id"] != tenant_id:
+        raise HTTPException(status_code=403, detail="无权删除该文档")
+
+    # 1. 删除 OSS 文件
+    oss_client = get_oss_client()
+    oss_client.delete_file(doc["oss_path"])
+
+    # 2. 删除 Qdrant 向量
+    qdrant = get_qdrant_manager()
+    qdrant.delete_vectors_by_doc(doc_id)
+
+    # 3. 删除 MongoDB 切片
+    delete_chunks_by_doc(doc_id)
+
+    # 4. 软删除文档记录
+    delete_document(doc_id)
+
+    return {
+        "success": True,
+        "message": "文档删除成功"
+    }
+
+
+# ==================== 智能客服配置接口 ====================
+
+@router.get("/agent/config")
+async def get_agent_config_endpoint(
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """获取智能客服配置"""
+    tenant_id = user_context.app_name
+    config = get_agent_config(tenant_id)
+
+    if not config:
+        # 返回默认配置
+        return {
+            "success": True,
+            "config": {
+                "enabled": False,
+                "name": "智能小助手",
+                "greeting": "您好！我是智能客服小助手，有什么可以帮您的吗？",
+                "h5_url": f"https://aichat.zhongsou.com/?tenant={tenant_id}"
+            }
+        }
+
+    return {
+        "success": True,
+        "config": config
+    }
+
+
+@router.post("/agent/config")
+async def update_agent_config_endpoint(
+    config: AgentConfigUpdate,
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """更新智能客服配置"""
+    tenant_id = user_context.app_name
+
+    success = update_agent_config(
+        tenant_id=tenant_id,
+        enabled=config.enabled,
+        name=config.name,
+        greeting=config.greeting
+    )
+
+    if not success:
+        raise HTTPException(status_code=500, detail="配置更新失败")
+
+    return {
+        "success": True,
+        "message": "配置更新成功"
+    }
+
+
+# ==================== 统计接口 ====================
+
+@router.get("/stats")
+async def get_stats(
+    user_context: UserContext = Depends(resolve_user_id),
+):
+    """获取租户的知识库统计数据"""
+    tenant_id = user_context.app_name
+    stats = get_tenant_stats(tenant_id)
+
+    return {
+        "success": True,
+        "stats": stats
+    }

app/utils/document_processor.py

@@ -0,0 +1,154 @@
+"""
+文档处理工具
+支持 Excel, Word, PDF 的文本提取和智能切片
+"""
+import re
+from typing import List, Dict, Optional
+import docx
+import openpyxl
+import PyPDF2
+from io import BytesIO
+
+
+class DocumentProcessor:
+    def __init__(self):
+        self.chunk_size = 800  # 每个切片的字符数
+        self.chunk_overlap = 100  # 切片之间的重叠字符数
+
+    def extract_text(self, file_content: bytes, file_type: str) -> Optional[str]:
+        """
+        从文件中提取文本
+        :param file_content: 文件内容（字节）
+        :param file_type: 文件类型（.xlsx, .docx, .pdf等）
+        :return: 提取的文本内容
+        """
+        try:
+            if file_type in ['.xlsx', '.xls']:
+                return self._extract_excel(file_content)
+            elif file_type in ['.docx', '.doc']:
+                return self._extract_word(file_content)
+            elif file_type == '.pdf':
+                return self._extract_pdf(file_content)
+            else:
+                return None
+        except Exception as e:
+            print(f"文本提取失败: {str(e)}")
+            return None
+
+    def _extract_excel(self, file_content: bytes) -> str:
+        """提取 Excel 文本"""
+        wb = openpyxl.load_workbook(BytesIO(file_content), data_only=True)
+        text_parts = []
+
+        for sheet_name in wb.sheetnames:
+            sheet = wb[sheet_name]
+            text_parts.append(f"\n=== 工作表: {sheet_name} ===\n")
+
+            for row in sheet.iter_rows(values_only=True):
+                row_text = '\t'.join([str(cell) if cell is not None else '' for cell in row])
+                if row_text.strip():
+                    text_parts.append(row_text)
+
+        return '\n'.join(text_parts)
+
+    def _extract_word(self, file_content: bytes) -> str:
+        """提取 Word 文本"""
+        doc = docx.Document(BytesIO(file_content))
+        text_parts = []
+
+        for para in doc.paragraphs:
+            if para.text.strip():
+                text_parts.append(para.text)
+
+        # 提取表格
+        for table in doc.tables:
+            for row in table.rows:
+                row_text = '\t'.join([cell.text.strip() for cell in row.cells])
+                if row_text.strip():
+                    text_parts.append(row_text)
+
+        return '\n'.join(text_parts)
+
+    def _extract_pdf(self, file_content: bytes) -> str:
+        """提取 PDF 文本"""
+        pdf_reader = PyPDF2.PdfReader(BytesIO(file_content))
+        text_parts = []
+
+        for page_num in range(len(pdf_reader.pages)):
+            page = pdf_reader.pages[page_num]
+            text = page.extract_text()
+            if text.strip():
+                text_parts.append(f"\n=== 第 {page_num + 1} 页 ===\n")
+                text_parts.append(text)
+
+        return '\n'.join(text_parts)
+
+    def split_text_into_chunks(self, text: str) -> List[Dict]:
+        """
+        将文本智能切片
+        :param text: 原始文本
+        :return: 切片列表 [{"content": "...", "metadata": {...}}, ...]
+        """
+        # 按段落分割
+        paragraphs = [p.strip() for p in text.split('\n') if p.strip()]
+
+        chunks = []
+        current_chunk = ""
+        current_length = 0
+
+        for para in paragraphs:
+            para_length = len(para)
+
+            # 如果当前段落加入后超过chunk_size，则保存当前chunk
+            if current_length + para_length > self.chunk_size and current_chunk:
+                chunks.append({
+                    "content": current_chunk.strip(),
+                    "metadata": {"char_count": current_length}
+                })
+
+                # 保留overlap部分
+                if current_length > self.chunk_overlap:
+                    overlap_text = current_chunk[-self.chunk_overlap:]
+                    current_chunk = overlap_text + "\n" + para
+                    current_length = len(overlap_text) + para_length
+                else:
+                    current_chunk = para
+                    current_length = para_length
+            else:
+                # 继续累积
+                if current_chunk:
+                    current_chunk += "\n" + para
+                else:
+                    current_chunk = para
+                current_length += para_length
+
+        # 保存最后一个chunk
+        if current_chunk.strip():
+            chunks.append({
+                "content": current_chunk.strip(),
+                "metadata": {"char_count": current_length}
+            })
+
+        return chunks
+
+    def clean_text(self, text: str) -> str:
+        """
+        清理文本：去除多余空格、特殊字符等
+        """
+        # 去除多余空白
+        text = re.sub(r'\s+', ' ', text)
+        # 去除特殊控制字符
+        text = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]', '', text)
+        return text.strip()
+
+
+# 全局单例
+_document_processor = None
+
+
+def get_document_processor() -> DocumentProcessor:
+    """获取文档处理器单例"""
+    global _document_processor
+    if _document_processor is None:
+        _document_processor = DocumentProcessor()
+    return _document_processor

app/utils/embedding_service.py

@@ -0,0 +1,61 @@
+"""
+向量化工具
+使用豆包的 doubao-embedding-vision-251215 模型生成向量嵌入
+"""
+from typing import List
+from openai import OpenAI
+from ..core.ark_client import get_client
+
+
+class EmbeddingService:
+    def __init__(self):
+        self.embedding_model = "doubao-embedding-vision-251215"
+
+    def generate_embeddings(self, texts: List[str], app_name: str) -> List[List[float]]:
+        """
+        批量生成文本的向量嵌入
+        :param texts: 文本列表
+        :param app_name: 租户的appName（用于获取对应的API客户端）
+        :return: 向量列表
+        """
+        if not texts:
+            return []
+
+        try:
+            client = get_client(app_name)
+
+            # 调用豆包的 Embeddings API
+            response = client.embeddings.create(
+                model=self.embedding_model,
+                input=texts
+            )
+
+            # 提取向量
+            embeddings = [item.embedding for item in response.data]
+            return embeddings
+
+        except Exception as e:
+            print(f"向量化失败: {str(e)}")
+            return []
+
+    def generate_single_embedding(self, text: str, app_name: str) -> List[float]:
+        """
+        生成单个文本的向量嵌入
+        :param text: 文本内容
+        :param app_name: 租户的appName
+        :return: 向量
+        """
+        embeddings = self.generate_embeddings([text], app_name)
+        return embeddings[0] if embeddings else []
+
+
+# 全局单例
+_embedding_service = None
+
+
+def get_embedding_service() -> EmbeddingService:
+    """获取向量化服务单例"""
+    global _embedding_service
+    if _embedding_service is None:
+        _embedding_service = EmbeddingService()
+    return _embedding_service

app/utils/kb_search.py

@@ -0,0 +1,88 @@
+"""
+知识库检索工具
+用于在聊天中检索自建知识库内容
+"""
+from typing import List, Dict, Optional
+from ..core.qdrant_client import get_qdrant_manager
+from ..utils.embedding_service import get_embedding_service
+from ..db.knowledge_base import search_chunks_by_ids
+
+
+def search_knowledge_base(query: str, tenant_id: str, top_k: int = 5) -> List[Dict]:
+    """
+    在知识库中搜索相关内容
+    :param query: 用户查询文本
+    :param tenant_id: 租户ID（appName）
+    :param top_k: 返回Top K个结果
+    :return: 检索结果列表 [{"content": "...", "score": 0.95, "doc_id": "..."}, ...]
+    """
+    try:
+        # 1. 将查询文本向量化
+        embedding_service = get_embedding_service()
+        query_vector = embedding_service.generate_single_embedding(query, tenant_id)
+
+        if not query_vector:
+            return []
+
+        # 2. 在 Qdrant 中进行向量相似度搜索
+        qdrant = get_qdrant_manager()
+        results = qdrant.search_vectors(tenant_id, query_vector, top_k)
+
+        # 3. 从 MongoDB 获取完整的切片内容
+        if results:
+            vector_ids = [r["vector_id"] for r in results]
+            chunks = search_chunks_by_ids(vector_ids)
+
+            # 合并 Qdrant 的相似度分数和 MongoDB 的完整内容
+            vector_id_to_chunk = {chunk["vector_id"]: chunk for chunk in chunks}
+
+            enriched_results = []
+            for result in results:
+                chunk = vector_id_to_chunk.get(result["vector_id"])
+                if chunk:
+                    enriched_results.append({
+                        "content": chunk["content"],
+                        "score": result["score"],
+                        "doc_id": chunk["doc_id"],
+                        "chunk_index": chunk["chunk_index"],
+                    })
+
+            return enriched_results
+
+        return []
+
+    except Exception as e:
+        print(f"知识库检索失败: {str(e)}")
+        return []
+
+
+def format_knowledge_context(search_results: List[Dict]) -> str:
+    """
+    将检索结果格式化为上下文文本，用于注入到 prompt
+    :param search_results: 检索结果列表
+    :return: 格式化的上下文文本
+    """
+    if not search_results:
+        return ""
+
+    context_parts = ["<知识库相关信息>"]
+
+    for idx, result in enumerate(search_results, 1):
+        context_parts.append(f"\n[文档片段 {idx}]")
+        context_parts.append(result["content"])
+
+    context_parts.append("\n</知识库相关信息>")
+
+    return "\n".join(context_parts)
+
+
+def get_custom_knowledge_tools(tenant_id: str) -> List[Dict]:
+    """
+    构造自定义知识库检索工具配置
+    用于替代豆包原生的知识库工具
+    :param tenant_id: 租户ID
+    :return: 工具配置列表
+    """
+    # 注意：这里返回空列表，因为我们将知识库检索前置到prompt中
+    # 而不是作为工具让模型调用
+    return []