AI办公 生产环境部署指南｜附源码

随着大模型能力的快速成熟，“AI办公”已经从概念演示走向真实业务场景。无论是智能文档总结、会议纪要生成、合同审阅、知识库问答，还是自动写邮件、生成PPT大纲、数据报表解读，越来越多企业开始将AI能力接入日常办公系统。

但很多团队在落地过程中会遇到一个共同问题：Demo很好跑，生产环境很难上。

生产环境不仅要求功能可用，还要考虑稳定性、安全性、可扩展性、日志审计、权限控制、成本控制以及后续运维。本文将从工程实践角度，介绍一个可落地的AI办公系统生产环境部署方案，并附上核心源码示例，帮助你快速搭建一套可用于企业内部的AI办公平台。

一、AI办公系统的典型功能

在部署之前，我们先明确一个AI办公平台通常包含哪些能力。

常见功能包括：

1. 智能文档处理

   • Word、PDF、TXT、Markdown文档解析
   • 长文总结
   • 关键信息提取
   • 合同条款审阅
   • 文档翻译

2. 企业知识库问答

   • 上传企业制度、产品手册、培训材料
   • 自动切分文档并向量化
   • 用户基于自然语言提问
   • 返回引用来源，降低幻觉风险

3. 会议与邮件助手

   • 会议纪要生成
   • 待办事项提取
   • 邮件润色与回复建议
   • 工作周报自动生成

4. 办公自动化Agent

   • 调用企业内部API
   • 查询数据库
   • 生成报表
   • 自动创建任务或工单

5. 权限与审计

   • 不同部门访问不同知识库
   • 对用户提问、模型回答进行审计
   • 对敏感词、敏感数据做脱敏与拦截

本文以“企业知识库问答 + 文档总结 + AI写作助手”为例，构建一个生产环境部署方案。

二、生产环境架构设计

一个可用于生产的AI办公系统，不建议将所有功能写在一个单进程服务里。推荐采用分层架构：

用户浏览器
   │
   ▼
Nginx / HTTPS网关
   │
   ▼
前端 Web 应用
   │
   ▼
后端 API 服务 FastAPI
   │
   ├── PostgreSQL：业务数据、用户、权限、会话记录
   ├── Redis：缓存、限流、任务队列
   ├── Celery Worker：异步文档解析、向量化任务
   ├── Milvus / Qdrant / pgvector：向量数据库
   ├── MinIO / S3：文件存储
   └── LLM Provider：OpenAI、Azure OpenAI、本地大模型等

1. 为什么需要异步任务？

文档上传后通常需要执行以下流程：

1. 保存原始文件；
2. 解析文档内容；
3. 清洗文本；
4. 按长度切分；
5. 调用Embedding模型；
6. 写入向量数据库；
7. 更新任务状态。

这个过程可能需要几秒到几分钟，如果放在API请求中同步执行，用户体验差，也容易导致接口超时。因此生产环境中应使用任务队列，例如Celery + Redis。

2. 为什么需要向量数据库？

传统数据库适合精确查询，而知识库问答需要“语义检索”。例如用户问：

“公司年假可以跨年使用吗？”

文档里可能写的是：

“员工未休完的带薪休假原则上应于当年度内完成。”

二者文字不完全一样，但语义相关。向量数据库可以通过Embedding向量做相似度检索，找到相关片段，再交给大模型生成答案。

3. 为什么需要对象存储？

用户上传的PDF、Word、图片等文件不适合直接存入数据库。生产环境推荐使用MinIO、阿里云OSS、腾讯云COS或AWS S3存储原始文件，数据库只保存文件URL、元信息和解析状态。

三、技术选型建议

本文示例采用以下技术栈：

为什么选择pgvector？
对于中小型企业内部知识库，pgvector足够实用。它可以直接在PostgreSQL中存储向量，降低运维复杂度。如果数据规模达到千万级向量，再考虑Milvus、Qdrant、Weaviate等专业向量数据库。

四、项目目录结构

下面是一个适合生产部署的项目结构：

ai-office/
├── backend/
│   ├── app/
│   │   ├── api/
│   │   │   ├── chat.py
│   │   │   ├── document.py
│   │   │   └── health.py
│   │   ├── core/
│   │   │   ├── config.py
│   │   │   ├── database.py
│   │   │   ├── security.py
│   │   │   └── logging.py
│   │   ├── models/
│   │   │   ├── document.py
│   │   │   └── message.py
│   │   ├── services/
│   │   │   ├── llm.py
│   │   │   ├── embedding.py
│   │   │   ├── rag.py
│   │   │   └── file_parser.py
│   │   ├── tasks/
│   │   │   └── document_tasks.py
│   │   ├── main.py
│   │   └── worker.py
│   ├── requirements.txt
│   └── Dockerfile
├── nginx/
│   └── nginx.conf
├── docker-compose.prod.yml
├── .env.example
└── README.md

五、环境变量配置

生产环境不要把密钥写死在代码中，应统一通过环境变量注入。

.env.example 示例：

APP_ENV=production
APP_NAME=AI Office
API_PORT=8000

POSTGRES_DB=ai_office
POSTGRES_USER=ai_user
POSTGRES_PASSWORD=change_me_strong_password
POSTGRES_HOST=postgres
POSTGRES_PORT=5432

REDIS_HOST=redis
REDIS_PORT=6379

OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_API_KEY=sk-your-key
CHAT_MODEL=gpt-4o-mini
EMBEDDING_MODEL=text-embedding-3-small

MINIO_ENDPOINT=minio:9000
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=change_me_minio_password
MINIO_BUCKET=ai-office

JWT_SECRET_KEY=change_me_jwt_secret
JWT_EXPIRE_MINUTES=1440

MAX_UPLOAD_SIZE_MB=50

生产环境中，建议：

• 密码长度不少于16位；
• 不同环境使用不同密钥；
• .env文件不要提交到Git；
• 对密钥进行定期轮换；
• 云环境优先使用Secret Manager。

六、核心源码示例

下面给出一个简化版但具备生产思路的源码示例。

1. 配置文件 config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_env: str = "production"
    app_name: str = "AI Office"
    api_port: int = 8000

    postgres_db: str
    postgres_user: str
    postgres_password: str
    postgres_host: str = "postgres"
    postgres_port: int = 5432

    redis_host: str = "redis"
    redis_port: int = 6379

    openai_base_url: str
    openai_api_key: str
    chat_model: str = "gpt-4o-mini"
    embedding_model: str = "text-embedding-3-small"

    jwt_secret_key: str
    jwt_expire_minutes: int = 1440

    max_upload_size_mb: int = 50

    @property
    def database_url(self) -> str:
        return (
            f"postgresql+psycopg2://{self.postgres_user}:"
            f"{self.postgres_password}@{self.postgres_host}:"
            f"{self.postgres_port}/{self.postgres_db}"
        )

    @property
    def redis_url(self) -> str:
        return f"redis://{self.redis_host}:{self.redis_port}/0"

    class Config:
        env_file = ".env"


settings = Settings()

2. 数据库连接 database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from app.core.config import settings

engine = create_engine(
    settings.database_url,
    pool_size=20,
    max_overflow=30,
    pool_pre_ping=True,
    pool_recycle=3600,
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)

Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

生产环境中数据库连接池非常重要。pool_pre_ping=True可以避免数据库断连后连接池仍返回坏连接的问题。

3. 文档模型 document.py

from sqlalchemy import Column, Integer, String, Text, DateTime, func
from app.core.database import Base


class Document(Base):
    __tablename__ = "documents"

    id = Column(Integer, primary_key=True, index=True)
    filename = Column(String(255), nullable=False)
    file_url = Column(String(1024), nullable=False)
    status = Column(String(50), default="pending")
    error_message = Column(Text, nullable=True)
    created_by = Column(String(128), nullable=False)
    created_at = Column(DateTime, server_default=func.now())


class DocumentChunk(Base):
    __tablename__ = "document_chunks"

    id = Column(Integer, primary_key=True, index=True)
    document_id = Column(Integer, index=True, nullable=False)
    content = Column(Text, nullable=False)
    embedding = Column(Text, nullable=True)
    source = Column(String(512), nullable=True)
    created_at = Column(DateTime, server_default=func.now())

说明：为了让示例更易理解，这里将embedding字段简化为Text。真实生产环境中建议使用pgvector字段类型，例如：

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE document_chunks (
    id SERIAL PRIMARY KEY,
    document_id INTEGER NOT NULL,
    content TEXT NOT NULL,
    embedding vector(1536),
    source VARCHAR(512),
    created_at TIMESTAMP DEFAULT now()
);

CREATE INDEX idx_document_chunks_embedding
ON document_chunks USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

4. LLM服务 llm.py

from openai import OpenAI
from app.core.config import settings

client = OpenAI(
    api_key=settings.openai_api_key,
    base_url=settings.openai_base_url,
)


def chat_completion(system_prompt: str, user_prompt: str) -> str:
    response = client.chat.completions.create(
        model=settings.chat_model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.2,
    )

    return response.choices[0].message.content


def create_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model=settings.embedding_model,
        input=text,
    )
    return response.data[0].embedding

生产建议：

• 对LLM调用增加超时；
• 对限流错误增加重试；
• 对异常结果进行日志记录；
• 对提示词和输出做敏感信息过滤；
• 为每个用户设置调用额度。

5. 文档解析服务 file_parser.py

from pathlib import Path


def parse_text_file(file_path: str) -> str:
    path = Path(file_path)
    suffix = path.suffix.lower()

    if suffix in [".txt", ".md"]:
        return path.read_text(encoding="utf-8")

    raise ValueError(f"暂不支持的文件类型: {suffix}")


def split_text(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]:
    chunks = []
    start = 0
    length = len(text)

    while start < length:
        end = min(start + chunk_size, length)
        chunk = text[start:end].strip()
        if chunk:
            chunks.append(chunk)
        start += chunk_size - overlap

    return chunks

实际项目中，你可以接入：

• pypdf解析PDF；
• python-docx解析Word；
• unstructured处理复杂文档；
• OCR识别扫描件。

6. Celery任务 document_tasks.py

import json
from celery import Celery
from app.core.config import settings
from app.core.database import SessionLocal
from app.models.document import Document, DocumentChunk
from app.services.file_parser import parse_text_file, split_text
from app.services.llm import create_embedding

celery_app = Celery(
    "ai_office_worker",
    broker=settings.redis_url,
    backend=settings.redis_url,
)


@celery_app.task(name="process_document")
def process_document(document_id: int, local_path: str):
    db = SessionLocal()

    try:
        document = db.query(Document).filter(Document.id == document_id).first()
        if not document:
            return

        document.status = "processing"
        db.commit()

        text = parse_text_file(local_path)
        chunks = split_text(text)

        for chunk in chunks:
            embedding = create_embedding(chunk)

            item = DocumentChunk(
                document_id=document.id,
                content=chunk,
                embedding=json.dumps(embedding),
                source=document.filename,
            )
            db.add(item)

        document.status = "completed"
        db.commit()

    except Exception as e:
        db.rollback()
        document = db.query(Document).filter(Document.id == document_id).first()
        if document:
            document.status = "failed"
            document.error_message = str(e)
            db.commit()
        raise

    finally:
        db.close()

7. RAG问答服务 rag.py

import json
import math
from app.services.llm import create_embedding, chat_completion
from app.models.document import DocumentChunk


def cosine_similarity(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(x * x for x in b))

    if norm_a == 0 or norm_b == 0:
        return 0

    return dot / (norm_a * norm_b)


def retrieve_chunks(db, query: str, top_k: int = 5):
    query_embedding = create_embedding(query)

    chunks = db.query(DocumentChunk).limit(1000).all()

    scored = []
    for chunk in chunks:
        if not chunk.embedding:
            continue

        emb = json.loads(chunk.embedding)
        score = cosine_similarity(query_embedding, emb)
        scored.append((score, chunk))

    scored.sort(key=lambda x: x[0], reverse=True)

    return [chunk for score, chunk in scored[:top_k]]


def answer_with_rag(db, question: str) -> str:
    chunks = retrieve_chunks(db, question)

    context = "\n\n".join(
        [f"来源：{c.source}\n内容：{c.content}" for c in chunks]
    )

    system_prompt = """
你是企业内部AI办公助手。
请严格基于给定资料回答问题。
如果资料中没有相关信息，请回答“根据现有资料无法确认”。
回答要清晰、准确，并尽量给出依据来源。
"""

    user_prompt = f"""
资料如下：
{context}

用户问题：
{question}
"""

    return chat_completion(system_prompt, user_prompt)

注意：上面为了便于演示，采用了Python内存计算相似度。生产环境应该使用pgvector、Milvus或Qdrant进行向量检索，否则数据量大时性能会下降。

8. API接口 chat.py

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.services.rag import answer_with_rag

router = APIRouter(prefix="/api/chat", tags=["chat"])


class ChatRequest(BaseModel):
    question: str


class ChatResponse(BaseModel):
    answer: str


@router.post("/ask", response_model=ChatResponse)
def ask(req: ChatRequest, db: Session = Depends(get_db)):
    answer = answer_with_rag(db, req.question)
    return ChatResponse(answer=answer)

9. 文档上传接口 document.py

import os
import uuid
from fastapi import APIRouter, UploadFile, File, Depends
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.models.document import Document
from app.tasks.document_tasks import process_document

router = APIRouter(prefix="/api/documents", tags=["documents"])

UPLOAD_DIR = "/data/uploads"


@router.post("/upload")
async def upload_document(
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
):
    os.makedirs(UPLOAD_DIR, exist_ok=True)

    ext = os.path.splitext(file.filename)[1]
    saved_name = f"{uuid.uuid4().hex}{ext}"
    local_path = os.path.join(UPLOAD_DIR, saved_name)

    with open(local_path, "wb") as f:
        content = await file.read()
        f.write(content)

    document = Document(
        filename=file.filename,
        file_url=local_path,
        status="pending",
        created_by="system",
    )
    db.add(document)
    db.commit()
    db.refresh(document)

    process_document.delay(document.id, local_path)

    return {
        "document_id": document.id,
        "status": document.status,
    }

10. 应用入口 main.py

from fastapi import FastAPI
from app.api import chat, document, health
from app.core.config import settings

app = FastAPI(
    title=settings.app_name,
    docs_url="/docs" if settings.app_env != "production" else None,
    redoc_url=None,
)

app.include_router(health.router)
app.include_router(chat.router)
app.include_router(document.router)

生产环境中建议关闭Swagger文档，或者至少加认证保护。

七、Dockerfile

backend/Dockerfile：

FROM python:3.11-slim

WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

RUN apt-get update && apt-get install -y \
    gcc \
    curl \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["gunicorn", "app.main:app", \
     "-k", "uvicorn.workers.UvicornWorker", \
     "-w", "4", \
     "-b", "0.0.0.0:8000", \
     "--timeout", "120"]

requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
gunicorn==23.0.0
sqlalchemy==2.0.35
psycopg2-binary==2.9.9
pydantic-settings==2.5.2
python-multipart==0.0.9
celery==5.4.0
redis==5.0.8
openai==1.45.0

八、Docker Compose生产部署

docker-compose.prod.yml：

services:
  postgres:
    image: postgres:16
    container_name: ai-office-postgres
    restart: always
    environment:
      POSTGRES_DB: ${POSTGRES_DB}
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    networks:
      - ai_office_net

  redis:
    image: redis:7
    container_name: ai-office-redis
    restart: always
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    networks:
      - ai_office_net

  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    container_name: ai-office-backend
    restart: always
    env_file:
      - .env
    volumes:
      - upload_data:/data/uploads
    depends_on:
      - postgres
      - redis
    networks:
      - ai_office_net

  worker:
    build:
      context: ./backend
      dockerfile: Dockerfile
    container_name: ai-office-worker
    restart: always
    command: celery -A app.tasks.document_tasks.celery_app worker --loglevel=INFO --concurrency=2
    env_file:
      - .env
    volumes:
      - upload_data:/data/uploads
    depends_on:
      - postgres
      - redis
    networks:
      - ai_office_net

  nginx:
    image: nginx:1.27
    container_name: ai-office-nginx
    restart: always
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - backend
    networks:
      - ai_office_net

volumes:
  postgres_data:
  redis_data:
  upload_data:

networks:
  ai_office_net:
    driver: bridge

九、Nginx配置

nginx/nginx.conf：

events {}

http {
    client_max_body_size 50m;

    upstream backend {
        server backend:8000;
    }

    server {
        listen 80;
        server_name your-domain.com;

        location /api/ {
            proxy_pass http://backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_read_timeout 180s;
        }

        location /health {
            proxy_pass http://backend;
        }
    }
}

如果是正式对外环境，必须配置HTTPS。可以使用：

• Let’s Encrypt；
• 云厂商证书；
• 企业内部CA证书。

十、部署步骤

1. 准备服务器

推荐最低配置：

CPU：4核
内存：8GB
磁盘：100GB SSD
系统：Ubuntu 22.04 LTS

如果使用本地大模型，配置要求会更高，通常需要GPU。如果使用OpenAI兼容API，则普通CPU服务器即可。

2. 安装Docker

curl -fsSL https://get.docker.com | bash
systemctl enable docker
systemctl start docker

安装Docker Compose插件：

docker compose version

如果未安装，可根据Docker官方文档安装。

3. 上传代码

git clone https://example.com/ai-office.git
cd ai-office
cp .env.example .env

修改.env中的数据库密码、Redis配置、OpenAI Key等。

4. 启动服务

docker compose -f docker-compose.prod.yml up -d --build

查看容器状态：

docker ps

查看日志：

docker logs -f ai-office-backend
docker logs -f ai-office-worker

5. 初始化数据库

简单方式是在应用启动时创建表，也可以使用Alembic做迁移。生产环境推荐Alembic。

临时初始化示例：

from app.core.database import Base, engine
from app.models.document import Document, DocumentChunk

Base.metadata.create_all(bind=engine)

更规范的方式是：

alembic revision --autogenerate -m "init tables"
alembic upgrade head

十一、生产环境必须关注的问题

1. 权限控制

AI办公系统往往会接触企业内部资料，权限控制是核心。

至少需要做到：

• 用户登录认证；
• 部门级数据隔离；
• 文档所属者和可见范围控制；
• 管理员审核上传内容；
• API接口鉴权；
• 操作日志留痕。

不要让所有用户都能访问所有文档，否则知识库会变成企业内部的数据泄露入口。

2. 敏感信息保护

用户可能上传合同、薪资、客户资料、财务数据等敏感内容。因此需要：

• 对身份证号、手机号、银行卡号等做识别；
• 对敏感字段做脱敏；
• 对模型请求日志谨慎保存；
• 禁止将敏感数据发送到未经审批的第三方模型；
• 根据企业合规要求选择私有化模型或专有云模型。

如果企业对数据安全要求较高，建议部署本地模型，例如Qwen、DeepSeek、Llama等，并将Embedding模型也私有化。

3. 限流与成本控制

大模型调用通常按Token计费。如果没有限制，可能出现成本失控。

建议设计：

• 用户每日调用额度；
• 部门月度预算；
• 单次请求最大Token限制；
• 文档最大上传大小；
• 文档最大页数；
• 高频请求限流；
• 异常用户自动封禁。

Nginx和Redis都可以实现限流。对于业务精细化控制，推荐在后端中间件层实现。

4. 日志与审计

生产环境需要记录：

• 用户ID；
• 请求时间；
• 请求接口；
• 文档上传记录；
• 问答历史；
• 模型调用耗时；
• Token消耗；
• 异常堆栈；
• 敏感操作审计。

但要注意，不建议直接记录完整敏感原文。日志系统本身也需要权限控制。

5. 健康检查

建议提供健康检查接口。

health.py：

from fastapi import APIRouter

router = APIRouter()


@router.get("/health")
def health_check():
    return {"status": "ok"}

进一步可以检查数据库、Redis、模型接口是否正常：

@router.get("/health/full")
def full_health_check():
    return {
        "api": "ok",
        "database": "ok",
        "redis": "ok",
        "llm": "ok",
    }

6. 备份策略

生产环境一定要做备份：

• PostgreSQL每日全量备份；
• 上传文件定期备份；
• 向量数据可重新生成，但成本较高，也建议备份；
• Redis通常不作为核心数据源，但任务状态可开启持久化；
• 备份文件要异地保存；
• 定期做恢复演练。

PostgreSQL备份示例：

docker exec ai-office-postgres pg_dump \
  -U ai_user ai_office > backup_$(date +%F).sql

恢复示例：

cat backup_2025-01-01.sql | docker exec -i ai-office-postgres \
  psql -U ai_user ai_office

十二、性能优化建议

1. 文档切分优化

不要简单按字符长度切分所有文档。更好的方式是：

• 按标题、段落、章节切分；
• 保留上下文重叠；
• 对表格做结构化处理；
• 对合同、制度类文档保留条款编号；
• 对长文档建立摘要层级索引。

2. 检索优化

RAG质量很大程度取决于检索。

可采用：

• 向量检索；
• 关键词检索；
• 混合检索；
• 重排序模型Reranker；
• 元数据过滤；
• 根据用户权限过滤知识范围。

3. Prompt优化

一个稳定的企业知识库Prompt应明确：

• 只能根据资料回答；
• 不知道就说不知道；
• 给出引用来源；
• 不编造制度；
• 回答结构清晰；
• 对不确定内容提示人工确认。

4. 缓存优化

对于重复问题，可以缓存：

• Embedding结果；
• 检索结果；
• 模型回答；
• 用户会话上下文。

但涉及权限和敏感数据时，缓存必须带上用户、部门、知识库等维度，避免串数据。

十三、上线检查清单

上线前建议逐项检查：

[ ] .env密钥是否已更换
[ ] 是否关闭生产环境Swagger
[ ] 是否启用HTTPS
[ ] 是否配置文件上传大小限制
[ ] 是否配置用户认证
[ ] 是否配置接口限流
[ ] 是否配置数据库备份
[ ] 是否配置日志采集
[ ] 是否配置异常告警
[ ] 是否配置文档权限
[ ] 是否配置模型调用额度
[ ] 是否验证敏感数据处理策略
[ ] 是否完成压力测试
[ ] 是否完成恢复演练

十四、常见问题

1. 为什么上传文档后一直是pending？

可能原因：

• Celery Worker没有启动；
• Redis连接失败；
• Worker和API使用的数据库不是同一个；
• 文档解析报错；
• 模型Embedding接口调用失败。

排查命令：

docker logs -f ai-office-worker
docker logs -f ai-office-backend

2. 问答结果经常胡编怎么办？

可以从以下方向优化：

• 增加“必须基于资料回答”的系统提示词；
• 返回引用片段；
• 降低temperature；
• 优化文档切分；
• 使用Reranker；
• 对答案做事实校验；
• 没有检索到资料时直接拒答。

3. 数据量大后检索很慢怎么办？

不要使用内存遍历计算相似度，应迁移到：

• pgvector；
• Qdrant；
• Milvus；
• Elasticsearch混合检索。

同时增加索引、元数据过滤和分页策略。

十五、总结

AI办公系统真正落地，核心不只是“接一个大模型API”，而是构建一套安全、稳定、可维护的企业级应用体系。

本文给出了一套基于FastAPI、PostgreSQL、Redis、Celery、Nginx和Docker Compose的生产部署方案，覆盖了：

• 系统架构设计；
• 后端核心源码；
• 异步文档处理；
• RAG知识库问答；
• Docker容器化部署；
• Nginx反向代理；
• 环境变量配置；
• 生产安全注意事项；
• 备份、日志、限流与性能优化。

如果你的团队刚开始做AI办公平台，可以先用本文方案快速搭建MVP；当业务量增长后，再逐步引入专业向量数据库、权限系统、监控告警平台、私有化模型和多租户能力。

真正可用的AI办公平台，一定不是一次性Demo，而是持续迭代的工程系统。只要架构设计合理，后续无论扩展智能写作、合同审查、会议纪要、数据分析还是企业Agent，都可以在这套基础设施上平滑演进。