第70集：实战：将你的知识库问答智能体打包上线，分享给朋友

章节标题：实战：将你的知识库问答智能体打包上线，分享给朋友

核心知识点讲解

项目规划

在将知识库问答智能体上线之前，我们需要进行项目规划：

功能规划：
- 知识库文档上传和管理
- 基于文档的问答功能
- 用户友好的界面
- 响应式设计，支持多设备访问
技术栈选择：
- 后端：Python + FastAPI
- 前端：Gradio（快速构建界面）
- 知识库：LangChain + FAISS
- 部署：Docker + Railway/Fly.io
架构设计：
- 前端：Gradio界面，负责用户交互
- 后端：FastAPI服务，处理核心逻辑
- 存储：本地文件系统存储上传的文档，FAISS存储向量索引

项目结构

一个合理的项目结构有助于代码管理和维护：

knowledge-base-agent/
├── app/
│   ├── api/            # API路由
│   ├── services/       # 核心服务
│   │   ├── knowledge_base.py  # 知识库管理
│   │   └── qa_agent.py        # 问答智能体
│   ├── utils/          # 工具函数
│   └── main.py         # 应用入口
├── frontend/           # 前端代码（如果不使用Gradio）
├── Dockerfile          # Docker构建文件
├── docker-compose.yml  # Docker Compose配置
├── requirements.txt    # Python依赖
└── .env.example        # 环境变量示例

部署流程

本地开发和测试：
- 实现核心功能
- 测试问答效果
- 优化用户界面
容器化：
- 编写Dockerfile
- 构建Docker镜像
- 本地测试容器
云部署：
- 选择云平台
- 配置环境变量
- 部署应用
- 测试线上访问
监控和维护：
- 设置日志收集
- 监控应用状态
- 定期更新和维护

实用案例分析

案例：构建和部署知识库问答智能体

1. 项目初始化

步骤：

创建项目目录：

mkdir knowledge-base-agent
cd knowledge-base-agent

创建requirements.txt：

fastapi
uvicorn
langchain
langchain-openai
faiss-cpu
pypdf
python-dotenv
gradio

创建.env.example：
```
OPENAI_API_KEY=your-openai-api-key
```

2. 实现核心功能

创建app/services/knowledge_base.py：

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os

class KnowledgeBase:
    def __init__(self, api_key, index_path="vectorstore"):
        self.api_key = api_key
        self.index_path = index_path
        self.embeddings = OpenAIEmbeddings(api_key=api_key)
        self.vectorstore = None
        
        # 加载现有索引
        if os.path.exists(self.index_path):
            self.vectorstore = FAISS.load_local(
                self.index_path, 
                self.embeddings,
                allow_dangerous_deserialization=True
            )
    
    def add_document(self, file_path):
        """添加文档到知识库"""
        # 加载文档
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        
        # 分割文档
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        splits = text_splitter.split_documents(documents)
        
        # 添加到向量存储
        if self.vectorstore:
            self.vectorstore.add_documents(splits)
        else:
            self.vectorstore = FAISS.from_documents(splits, self.embeddings)
        
        # 保存索引
        self.vectorstore.save_local(self.index_path)
        
        return len(splits)
    
    def query(self, question, k=3):
        """基于知识库回答问题"""
        if not self.vectorstore:
            return "知识库为空，请先上传文档。"
        
        # 检索相关文档
        docs = self.vectorstore.similarity_search(question, k=k)
        context = "\n".join([doc.page_content for doc in docs])
        
        return context

创建app/services/qa_agent.py：

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

class QAAgent:
    def __init__(self, api_key):
        self.api_key = api_key
        self.llm = ChatOpenAI(
            api_key=api_key,
            model="gpt-3.5-turbo",
            temperature=0.7
        )
        
        # 定义提示词模板
        self.prompt = ChatPromptTemplate.from_template(
            "你是一个基于文档的问答助手。请根据以下文档内容回答问题，不要添加文档中没有的信息。\n\n"  
            "文档内容：\n{context}\n\n"  
            "问题：{question}\n\n"  
            "回答："
        )
        
        # 创建链
        self.chain = LLMChain(
            llm=self.llm,
            prompt=self.prompt
        )
    
    def answer(self, question, context):
        """根据上下文回答问题"""
        response = self.chain.run(
            context=context,
            question=question
        )
        return response

创建app/main.py：

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import os
import tempfile
from dotenv import load_dotenv
from app.services.knowledge_base import KnowledgeBase
from app.services.qa_agent import QAAgent
import gradio as gr

# 加载环境变量
load_dotenv()

# 初始化应用
app = FastAPI(
    title="知识库问答智能体",
    description="基于文档的智能问答系统",
    version="1.0.0"
)

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 初始化知识库和问答智能体
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise HTTPException(status_code=500, detail="OPENAI_API_KEY not set")

knowledge_base = KnowledgeBase(api_key)
qa_agent = QAAgent(api_key)

# 创建临时目录存储上传的文件
UPLOAD_DIR = tempfile.mkdtemp()
os.makedirs(UPLOAD_DIR, exist_ok=True)

# API路由
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """上传文档到知识库"""
    try:
        # 保存上传的文件
        file_path = os.path.join(UPLOAD_DIR, file.filename)
        with open(file_path, "wb") as f:
            f.write(await file.read())
        
        # 添加到知识库
        chunks_count = knowledge_base.add_document(file_path)
        
        return {
            "filename": file.filename,
            "chunks_count": chunks_count,
            "message": "文档上传成功并添加到知识库"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/qa")
async def question_answer(question: str):
    """基于知识库回答问题"""
    try:
        # 从知识库检索相关文档
        context = knowledge_base.query(question)
        
        # 生成回答
        answer = qa_agent.answer(question, context)
        
        return {
            "question": question,
            "answer": answer
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Gradio界面
def create_interface():
    def upload_file_interface(file):
        """Gradio文件上传处理"""
        try:
            # 添加到知识库
            chunks_count = knowledge_base.add_document(file.name)
            return f"文档上传成功！共分割为{chunks_count}个片段并添加到知识库。"
        except Exception as e:
            return f"上传失败：{str(e)}"
    
    def qa_interface(question):
        """Gradio问答处理"""
        try:
            # 从知识库检索相关文档
            context = knowledge_base.query(question)
            
            # 生成回答
            answer = qa_agent.answer(question, context)
            
            return answer
        except Exception as e:
            return f"回答失败：{str(e)}"
    
    # 创建Gradio界面
    with gr.Blocks() as demo:
        gr.Markdown("# 知识库问答智能体")
        gr.Markdown("上传文档到知识库，然后基于文档内容进行问答。")
        
        with gr.Tab("上传文档"):
            file_input = gr.File(label="选择PDF文件")
            upload_button = gr.Button("上传并添加到知识库")
            upload_output = gr.Textbox(label="上传结果")
            
            upload_button.click(
                fn=upload_file_interface,
                inputs=file_input,
                outputs=upload_output
            )
        
        with gr.Tab("问答"):
            question_input = gr.Textbox(label="输入问题", placeholder="请输入你的问题...")
            qa_button = gr.Button("获取答案")
            answer_output = gr.Textbox(label="回答", lines=5)
            
            qa_button.click(
                fn=qa_interface,
                inputs=question_input,
                outputs=answer_output
            )
    
    return demo

# 启动Gradio界面
interface = create_interface()

# 导出Gradio应用
app = interface

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. 容器化

创建Dockerfile：

# 使用官方Python镜像作为基础
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建上传目录
RUN mkdir -p /app/uploads

# 暴露端口
EXPOSE 8000

# 运行应用
CMD ["python", "app/main.py"]

创建docker-compose.yml：

version: '3.8'

services:
  knowledge-base-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./vectorstore:/app/vectorstore
      - ./uploads:/app/uploads

volumes:
  vectorstore:
  uploads:

4. 部署到云平台

部署到Railway：

创建GitHub仓库：
- 初始化git仓库
- 添加所有文件
- 提交并推送到GitHub
部署到Railway：
- 登录Railway
- 创建新项目，选择从GitHub仓库部署
- 连接你的GitHub仓库
- 在环境变量设置中添加OPENAI_API_KEY
- 部署应用

部署到Fly.io：

安装Fly CLI：
```
curl -L https://fly.io/install.sh | sh
```
登录Fly：
```
fly auth login
```
初始化项目：
```
fly launch
```

设置环境变量：

fly secrets set OPENAI_API_KEY=your-api-key

部署应用：
```
fly deploy
```

5. 测试和优化

功能测试：
- 上传文档到知识库
- 基于文档内容提问
- 测试不同类型的问题
性能优化：
- 调整文档分割策略
- 优化向量检索参数
- 考虑使用更高效的向量数据库
用户体验优化：
- 改进界面设计
- 添加加载状态指示
- 优化错误处理和提示

代码示例解释

知识库管理

KnowledgeBase类：负责文档的加载、分割、向量化和存储
FAISS：轻量级向量数据库，适合存储和检索向量嵌入
文档处理：使用LangChain的PyPDFLoader加载PDF文档，RecursiveCharacterTextSplitter分割文档

问答智能体

QAAgent类：负责基于检索到的文档生成回答
ChatOpenAI：使用OpenAI的GPT模型生成回答
提示词设计：设计了一个专注于基于文档内容回答问题的提示词

前端界面

Gradio：快速构建用户友好的界面
双标签设计：分别用于文档上传和问答
响应式布局：适配不同设备尺寸

容器化和部署

Dockerfile：定义了应用的运行环境
Docker Compose：简化了本地开发和测试
云部署：支持部署到Railway和Fly.io等云平台

常见问题与解决方案

1. 文档上传失败

问题：上传文档时出现错误

解决方案：

检查文件格式是否为PDF
检查文件大小是否过大
查看应用日志，定位具体错误
确保服务器有足够的存储空间

2. 问答质量不高

问题：智能体回答的质量不高，或与文档内容不符

解决方案：

调整文档分割策略，尝试不同的chunk_size
增加检索的文档数量（k值）
优化提示词设计
考虑使用更高级的模型（如gpt-4）

3. 应用响应缓慢

问题：应用响应时间较长

解决方案：

优化文档处理流程
考虑使用异步处理
对于较大的知识库，考虑使用更强大的向量数据库
选择离用户更近的云服务器区域

4. 部署后无法访问

问题：部署到云平台后无法访问应用

解决方案：

检查云平台的部署状态
查看应用日志，定位错误
确保环境变量正确设置
检查网络配置和防火墙规则

最佳实践

数据安全：
- 不要在日志中记录敏感信息
- 定期清理临时文件
- 考虑对上传的文档进行访问控制
性能优化：
- 实现文档的增量更新，避免重复处理
- 考虑使用批处理优化大量文档的处理
- 实现缓存机制，减少重复计算
可靠性：
- 实现完善的错误处理
- 添加健康检查端点
- 定期备份知识库
可扩展性：
- 设计模块化的代码结构
- 考虑支持更多类型的文档
- 预留接口支持其他LLM模型
用户体验：
- 提供清晰的用户引导
- 添加加载状态和进度指示
- 设计直观的错误提示

总结

通过本文的学习，你已经完成了一个完整的知识库问答智能体的构建和部署：

项目规划：明确了功能需求和技术栈选择
核心功能实现：
- 知识库管理：文档上传、分割、向量化和存储
- 问答智能体：基于检索到的文档生成回答
- 用户界面：使用Gradio构建友好的交互界面
容器化：编写Dockerfile和docker-compose.yml，实现应用的容器化
云部署：部署到Railway和Fly.io等云平台
测试和优化：功能测试、性能优化和用户体验优化

这个知识库问答智能体可以帮助用户基于上传的文档进行问答，适用于多种场景，如：

企业内部知识库：员工可以基于公司文档快速获取信息
个人学习助手：基于学习资料进行问答，辅助学习
客户支持：基于产品文档回答客户问题

通过这个实战项目，你不仅掌握了AI智能体的开发和部署技能，还学会了如何将多个技术组件整合在一起，构建一个完整的应用。这为你未来开发更复杂的AI应用奠定了基础。