# dynabook_kn_py

**Repository Path**: gaoerkai/dynabook_kn_py

## Basic Information

- **Project Name**: dynabook_kn_py
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-11-13
- **Last Updated**: 2025-11-13

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 企业知识库 AI 服务 (FastAPI + LangChain 1.0)

## 项目简介

这是一个基于 FastAPI 和 LangChain 1.0 的企业知识库 AI 服务，提供文档处理、向量搜索、AI 问答(RAG)和知识图谱等功能。

### 核心特性

- ✅ **统一的响应格式**: 所有 API 返回统一的 JSON 格式
- ✅ **统一的异常处理**: 全局异常捕获和处理中间件
- ✅ **LangChain 1.0 集成**: 基于最新的 LangChain 框架
- ✅ **RAG 问答系统**: 检索增强生成的智能问答
- ✅ **向量搜索**: 基于 Milvus 的语义搜索
- ✅ **文档处理**: 支持多种文档格式解析
- ✅ **知识图谱**: 实体关系提取和管理
- ✅ **请求日志**: 自动记录所有请求和响应
- ✅ **完整的文档**: Swagger/ReDoc 自动生成 API 文档

## 技术栈

- **Web 框架**: FastAPI 0.104+
- **AI 框架**: LangChain 1.0
- **LLM**: OpenAI GPT (可扩展)
- **向量数据库**: Milvus
- **嵌入模型**: Sentence Transformers
- **数据库**: MySQL (通过 SQLAlchemy)
- **文档解析**: PyPDF2, python-docx, openpyxl
- **NLP**: spaCy, jieba

## 项目结构

```
app/
├── api/                    # API 路由
│   ├── ai_qa.py           # AI 问答接口
│   ├── document_processing.py  # 文档处理接口
│   ├── vector_search.py   # 向量搜索接口
│   └── knowledge_graph.py # 知识图谱接口
├── core/                   # 核心模块
│   ├── exceptions.py      # 自定义异常
│   ├── middleware.py      # 中间件
│   └── logging.py         # 日志配置
├── schemas/                # 数据模型
│   └── response.py        # 统一响应模型
├── services/               # 业务逻辑
│   ├── langchain_rag_service.py  # LangChain RAG 服务
│   ├── embedding_service.py      # 向量嵌入服务
│   ├── vector_store.py           # 向量存储服务
│   ├── document_parser.py        # 文档解析服务
│   ├── chunk_service.py          # 文本分块服务
│   └── kg_service.py             # 知识图谱服务
├── models/                 # 数据库模型
├── config.py              # 配置管理
├── database.py            # 数据库连接
└── main.py                # 应用入口
```

## 快速开始

### 1. 环境准备

```bash
# 创建虚拟环境
python -m venv venv

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

# 安装依赖
pip install -r requirements_openai.txt.bak
```

### 2. 配置环境变量

创建 `.env` 文件：

```env
# 应用配置
APP_NAME=Knowledge Base AI Service
DEBUG=false

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=root
DB_NAME=knowledge_db

# Milvus 向量数据库配置
MILVUS_HOST=localhost
MILVUS_PORT=19530
MILVUS_COLLECTION_NAME=document_vectors

# Embedding 模型配置
EMBEDDING_MODEL=sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2
EMBEDDING_DEVICE=cpu

# LLM 配置
LLM_PROVIDER=openai
OPENAI_API_KEY=your-openai-api-key-here
OPENAI_MODEL=gpt-3.5-turbo
OPENAI_BASE_URL=

# 文档处理配置
CHUNK_SIZE=512
CHUNK_OVERLAP=50
```

### 3. 启动服务

```bash
# 开发模式（自动重载）
python app/main.py

# 或使用 uvicorn
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
```

### 4. 访问文档

- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
- 健康检查: http://localhost:8000/health

## API 使用示例

### 统一响应格式

所有接口都返回统一的 JSON 格式：

**成功响应**:
```json
{
  "code": 200,
  "message": "success",
  "data": {
    // 具体数据
  }
}
```

**错误响应**:
```json
{
  "code": 400,
  "message": "错误信息",
  "data": null
}
```

### 1. 文档处理

```bash
POST /api/v1/documents/process
Content-Type: application/json

{
  "document_id": 1,
  "file_path": "/path/to/document.pdf",
  "file_type": "pdf"
}
```

### 2. 向量搜索

```bash
POST /api/v1/search/vector
Content-Type: application/json

{
  "query": "如何使用知识库？",
  "top_k": 5,
  "document_ids": [1, 2, 3],
  "min_score": 0.7
}
```

### 3. AI 问答 (RAG)

```bash
POST /api/v1/qa/query
Content-Type: application/json

{
  "question": "知识库系统有哪些核心功能？",
  "document_ids": [1, 2],
  "top_k": 3,
  "max_tokens": 500
}
```

### 4. 文本摘要

```bash
POST /api/v1/qa/summarize
Content-Type: application/json

{
  "text": "很长的文档内容..."
}
```

### 5. 批量问答

```bash
POST /api/v1/qa/batch-query
Content-Type: application/json

{
  "questions": [
    "问题1？",
    "问题2？",
    "问题3？"
  ],
  "top_k": 3
}
```

### 6. 知识图谱

```bash
# 提取知识图谱
POST /api/v1/kg/extract
Content-Type: application/json

{
  "document_id": 1,
  "content": "文档内容..."
}

# 获取实体
GET /api/v1/kg/entities?document_id=1

# 获取关系
GET /api/v1/kg/relations?document_id=1

# 获取完整图谱
GET /api/v1/kg/graph/1
```

## 核心功能说明

### 1. 统一异常处理

框架提供了多种自定义异常类：

```python
from app.core.exceptions import (
    BusinessException,      # 业务逻辑异常
    ValidationException,    # 参数验证异常
    NotFoundException,      # 资源不存在异常
    AuthenticationException, # 认证异常
    AuthorizationException, # 授权异常
    LangChainException,     # LangChain 相关异常
    VectorStoreException    # 向量存储异常
)

# 使用示例
if not data:
    raise NotFoundException("数据不存在")
```

### 2. 统一响应格式

使用便捷函数生成响应：

```python
from app.schemas.response import success_response, error_response

# 成功响应
return success_response(
    data={"result": "数据"},
    message="操作成功"
)

# 错误响应（通常不需要手动调用，异常会自动处理）
return error_response(
    message="操作失败",
    code=400
)
```

### 3. LangChain RAG 服务

```python
from app.services.langchain_rag_service import LangChainRAGService

rag_service = LangChainRAGService()

# 问答
result = await rag_service.query(
    question="用户问题",
    document_ids=[1, 2],
    top_k=3
)

# 摘要
summary = await rag_service.summarize(text="文档内容")

# 批量问答
results = await rag_service.batch_query(
    questions=["问题1", "问题2"]
)
```

### 4. 日志系统

框架自动记录所有请求和错误：

```python
import logging

logger = logging.getLogger(__name__)

logger.info("信息日志")
logger.warning("警告日志")
logger.error("错误日志", exc_info=True)
```

日志会输出到：
- 控制台（彩色输出）
- 文件 `logs/app.log`（生产环境）

## 中间件

### 1. 异常处理中间件

自动捕获所有异常并转换为统一的响应格式。

### 2. 请求日志中间件

自动记录：
- 请求方法和路径
- 响应状态码
- 请求耗时
- 响应头添加 `X-Process-Time`

### 3. CORS 中间件

允许跨域请求（生产环境需要配置具体域名）。

## 扩展开发

### 添加新的 API 路由

1. 在 `app/api/` 创建新的路由文件
2. 使用统一的响应和异常处理
3. 在 `app/main.py` 中注册路由

```python
# app/api/my_api.py
from fastapi import APIRouter
from app.schemas.response import success_response
from app.core.exceptions import BusinessException

router = APIRouter()

@router.get("/example")
async def example():
    return success_response(data={"message": "Hello"})
```

```python
# app/main.py
from app.api import my_api

app.include_router(
    my_api.router,
    prefix="/api/v1/my",
    tags=["My API"]
)
```

### 添加新的服务

在 `app/services/` 创建新的服务类：

```python
# app/services/my_service.py
import logging
from app.core.exceptions import ServiceException

logger = logging.getLogger(__name__)

class MyService:
    def __init__(self):
        logger.info("MyService 初始化")
    
    async def process(self, data):
        try:
            # 处理逻辑
            return result
        except Exception as e:
            logger.error(f"处理失败: {str(e)}", exc_info=True)
            raise ServiceException(f"处理失败: {str(e)}")
```

## 部署

### Docker 部署

```bash
# 构建镜像
docker build -t knowledge-base-ai:latest .

# 运行容器
docker run -d \
  -p 8000:8000 \
  --env-file .env \
  --name kb-ai \
  knowledge-base-ai:latest
```

### 生产环境配置

1. 设置 `DEBUG=false`
2. 配置具体的 CORS 域名
3. 使用 Gunicorn 或 Uvicorn 多工作进程
4. 配置日志轮转
5. 使用 Nginx 反向代理

```bash
# 生产环境启动
gunicorn app.main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --access-logfile logs/access.log \
  --error-logfile logs/error.log
```

## 常见问题

### 1. LangChain 初始化失败

确保配置了正确的 `OPENAI_API_KEY`。

### 2. Milvus 连接失败

检查 Milvus 服务是否启动，端口是否正确。

### 3. 文档解析失败

确保文件路径正确，文件格式支持。

## 开发计划

- [ ] 支持更多 LLM 提供商（本地模型、其他云服务）
- [ ] 实现文档块的智能分割策略
- [ ] 增强知识图谱功能
- [ ] 添加用户认证和权限管理
- [ ] 实现 API 限流
- [ ] 添加缓存机制
- [ ] 支持流式响应

## 许可证

MIT License

## 贡献

欢迎提交 Issue 和 Pull Request！

## 联系方式

如有问题，请通过 Issue 联系。