AI办公 性能优化教程｜附源码

随着大语言模型、智能文档处理、自动化流程编排等技术快速发展，AI办公已经从“尝鲜工具”逐渐变成企业日常效率提升的重要组成部分。无论是自动生成会议纪要、智能总结文档、批量处理邮件，还是基于企业知识库进行问答，AI都能显著减少重复劳动。

但是，很多人在真正落地AI办公系统时会遇到一个共同问题：功能能跑，但速度慢、成本高、体验不稳定。例如：

• 用户上传一份文档后，等待几十秒才有结果；
• 多人同时使用时，系统响应明显变慢；
• 每次调用大模型都消耗大量Token，费用不断增加；
• 相同问题反复请求模型，造成重复计算；
• 文档内容过长，模型上下文不够，回答质量下降；
• API接口没有限流，容易被频繁请求拖垮。

因此，AI办公系统不仅要“能用”，更要“好用、快用、低成本使用”。本文将围绕AI办公场景，系统讲解性能优化思路，并提供一套可参考的源码示例，帮助你构建一个更加高效的AI办公应用。

一、AI办公系统常见性能瓶颈

在进行优化之前，首先要明确性能瓶颈来自哪里。AI办公系统通常由以下几个环节组成：

1. 用户输入：文本、文档、图片、表格等；
2. 数据预处理：格式转换、文本清洗、分段；
3. AI模型调用：调用大语言模型或本地模型；
4. 结果处理：格式化、摘要、结构化输出；
5. 数据存储：缓存、日志、知识库、数据库；
6. 前端展示：流式输出、状态提示、交互体验。

性能问题往往不是单点造成的，而是多个环节叠加产生。

1. 模型响应慢

大模型推理通常是整个链路中最耗时的部分。尤其是长文本输入、复杂提示词、多轮上下文时，请求延迟会明显增加。

2. Token消耗过高

Token数量直接影响模型响应速度和调用成本。提示词写得过长、上下文冗余、重复传入历史信息，都会造成Token浪费。

3. 文档处理效率低

AI办公经常涉及PDF、Word、Excel等文件。如果每次请求都重新解析全文，效率会很低。

4. 缺少缓存机制

很多用户会问类似的问题，例如“总结这份文档”“生成周报模板”“提炼重点”。如果每次都重新调用模型，就会造成重复开销。

5. 并发控制不足

当多个用户同时使用时，如果系统没有任务队列、限流、异步处理等机制，服务很容易出现阻塞甚至崩溃。

二、AI办公性能优化总体思路

AI办公性能优化可以从以下几个方向入手：

一个成熟的AI办公系统通常不是简单地“前端输入，后端调用模型”，而是需要在模型调用前后做大量工程优化。

三、项目示例：AI办公助手后端服务

下面我们使用 Python + FastAPI 编写一个简化版AI办公助手，主要实现以下功能：

• 文本总结接口；
• Prompt长度优化；
• 请求结果缓存；
• 异步调用模拟；
• 简单限流；
• 流式输出示例；
• 文档分块处理；
• 统计接口耗时。

说明：为了方便演示，源码中的大模型调用使用模拟函数。你可以替换为 OpenAI、通义千问、智谱、Moonshot、DeepSeek 或本地模型接口。

四、项目目录结构

ai-office-optimizer/
├── main.py
├── llm_client.py
├── cache.py
├── rate_limiter.py
├── text_splitter.py
├── prompt_builder.py
├── requirements.txt
└── README.md

五、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
pydantic==2.8.2

安装依赖：

pip install -r requirements.txt

启动服务：

uvicorn main:app --reload --host 0.0.0.0 --port 8000

六、核心源码

1. 缓存模块：cache.py

缓存是AI办公优化中非常重要的一环。对于相同输入、相同任务类型，可以直接返回历史结果，避免重复调用模型。

# cache.py
import hashlib
import time
from typing import Optional


class SimpleCache:
    def __init__(self, ttl: int = 3600):
        self.ttl = ttl
        self.store = {}

    def _make_key(self, text: str, task: str) -> str:
        raw = f"{task}:{text}"
        return hashlib.md5(raw.encode("utf-8")).hexdigest()

    def get(self, text: str, task: str) -> Optional[str]:
        key = self._make_key(text, task)
        item = self.store.get(key)

        if not item:
            return None

        value, expire_at = item
        if time.time() > expire_at:
            del self.store[key]
            return None

        return value

    def set(self, text: str, task: str, value: str):
        key = self._make_key(text, task)
        self.store[key] = (value, time.time() + self.ttl)

这个缓存模块使用内存字典实现，适合演示和小型项目。如果用于生产环境，可以替换为：

• Redis；
• Memcached；
• 数据库缓存；
• 对象存储 + 哈希索引。

2. 限流模块：rate_limiter.py

AI接口通常成本较高，因此必须限制用户请求频率，防止短时间大量调用。

# rate_limiter.py
import time
from collections import defaultdict


class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.user_requests = defaultdict(list)

    def allow(self, user_id: str) -> bool:
        now = time.time()
        requests = self.user_requests[user_id]

        self.user_requests[user_id] = [
            req_time for req_time in requests
            if now - req_time < self.window_seconds
        ]

        if len(self.user_requests[user_id]) >= self.max_requests:
            return False

        self.user_requests[user_id].append(now)
        return True

该模块实现了一个简单滑动窗口限流逻辑。例如每个用户每分钟最多请求10次。

3. 文本分块模块：text_splitter.py

AI办公经常需要处理长文档。如果直接把整篇文档塞给模型，不仅慢，还可能超过上下文限制。合理分块是非常关键的优化方式。

# text_splitter.py
from typing import List


def split_text(text: str, chunk_size: int = 800, overlap: int = 100) -> List[str]:
    """
    将长文本切分为多个片段。
    chunk_size: 每段最大字符数
    overlap: 相邻片段重叠字符数，避免语义断裂
    """
    if len(text) <= chunk_size:
        return [text]

    chunks = []
    start = 0

    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)

        start = end - overlap
        if start < 0:
            start = 0

        if start >= len(text):
            break

    return chunks

这里使用字符数切分，实际项目中可以进一步优化为：

• 按段落切分；
• 按标题切分；
• 按句子切分；
• 按Token数量切分；
• 结合语义相似度切分。

4. Prompt构建模块：prompt_builder.py

Prompt不是越长越好。办公场景中，Prompt应该明确、紧凑、结构化。

# prompt_builder.py

def build_summary_prompt(text: str) -> str:
    return f"""
你是一个专业的AI办公助手，请对以下内容进行总结。

要求：
1. 提炼核心观点；
2. 使用条目化表达；
3. 保留关键数字、时间、结论；
4. 不要编造原文没有的信息；
5. 输出中文。

待总结内容：
{text}
""".strip()


def build_meeting_minutes_prompt(text: str) -> str:
    return f"""
你是企业会议纪要助手，请根据会议内容生成纪要。

输出格式：
一、会议主题
二、参会人员
三、讨论重点
四、决策事项
五、待办任务
六、风险与问题

会议内容：
{text}
""".strip()

优化Prompt时要注意以下原则：

1. 指令明确，避免模糊描述；
2. 输出格式固定，减少模型自由发挥；
3. 删除不必要的寒暄；
4. 不重复传入无关上下文；
5. 对复杂任务拆成多个简单任务。

5. 模型调用模块：llm_client.py

这里使用模拟模型调用，实际项目中你可以替换成真实API。

# llm_client.py
import asyncio


async def call_llm(prompt: str) -> str:
    """
    模拟大模型调用。
    实际项目中可以在这里请求 OpenAI、DeepSeek、通义千问等模型接口。
    """
    await asyncio.sleep(1)

    return f"""以下是AI生成的结果：

- 已识别输入内容的核心信息；
- 已根据办公场景进行结构化整理；
- 已尽量压缩冗余表达；
- 已生成适合直接复制使用的结果。

提示词长度：{len(prompt)} 字符
"""

如果你使用真实模型接口，可以参考如下伪代码：

# 示例：伪代码，仅供参考
import httpx

async def call_real_llm(prompt: str) -> str:
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            "https://api.example.com/v1/chat/completions",
            headers={
                "Authorization": "Bearer YOUR_API_KEY"
            },
            json={
                "model": "your-model-name",
                "messages": [
                    {"role": "user", "content": prompt}
                ],
                "temperature": 0.3,
                "stream": False
            }
        )
        data = response.json()
        return data["choices"][0]["message"]["content"]

6. 主服务：main.py

# main.py
import time
import asyncio
from fastapi import FastAPI, HTTPException, Header
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from cache import SimpleCache
from rate_limiter import RateLimiter
from text_splitter import split_text
from prompt_builder import build_summary_prompt, build_meeting_minutes_prompt
from llm_client import call_llm


app = FastAPI(title="AI办公性能优化示例")

cache = SimpleCache(ttl=3600)
rate_limiter = RateLimiter(max_requests=10, window_seconds=60)


class TextRequest(BaseModel):
    text: str
    task: str = "summary"


@app.post("/ai/process")
async def process_text(
    request: TextRequest,
    user_id: str = Header(default="anonymous")
):
    start_time = time.time()

    if not rate_limiter.allow(user_id):
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

    cached_result = cache.get(request.text, request.task)
    if cached_result:
        return {
            "from_cache": True,
            "task": request.task,
            "cost_time": round(time.time() - start_time, 3),
            "result": cached_result
        }

    if request.task == "summary":
        prompt = build_summary_prompt(request.text)
    elif request.task == "meeting":
        prompt = build_meeting_minutes_prompt(request.text)
    else:
        raise HTTPException(status_code=400, detail="不支持的任务类型")

    result = await call_llm(prompt)
    cache.set(request.text, request.task, result)

    return {
        "from_cache": False,
        "task": request.task,
        "cost_time": round(time.time() - start_time, 3),
        "result": result
    }


@app.post("/ai/long-summary")
async def long_summary(
    request: TextRequest,
    user_id: str = Header(default="anonymous")
):
    start_time = time.time()

    if not rate_limiter.allow(user_id):
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

    chunks = split_text(request.text, chunk_size=800, overlap=100)

    partial_results = []
    for index, chunk in enumerate(chunks):
        prompt = build_summary_prompt(chunk)
        result = await call_llm(prompt)
        partial_results.append(f"第{index + 1}段总结：\n{result}")

    final_prompt = build_summary_prompt("\n".join(partial_results))
    final_result = await call_llm(final_prompt)

    return {
        "chunk_count": len(chunks),
        "cost_time": round(time.time() - start_time, 3),
        "result": final_result
    }


@app.post("/ai/stream")
async def stream_output(request: TextRequest):
    async def generator():
        prompt = build_summary_prompt(request.text)
        result = await call_llm(prompt)

        for char in result:
            yield char
            await asyncio.sleep(0.02)

    return StreamingResponse(generator(), media_type="text/plain")

七、接口测试示例

1. 普通总结接口

curl -X POST "http://127.0.0.1:8000/ai/process" \
-H "Content-Type: application/json" \
-H "user-id: user001" \
-d '{
  "task": "summary",
  "text": "本周销售部门完成了季度目标的85%，其中华东区域增长明显，同比增长18%。但华南区域因渠道调整，订单转化率下降。下周计划重点跟进重点客户，并优化渠道合作政策。"
}'

返回示例：

{
  "from_cache": false,
  "task": "summary",
  "cost_time": 1.003,
  "result": "以下是AI生成的结果：..."
}

再次请求相同内容时，会命中缓存：

{
  "from_cache": true,
  "task": "summary",
  "cost_time": 0.001,
  "result": "以下是AI生成的结果：..."
}

可以看到，缓存命中后响应时间会大幅降低。

2. 会议纪要接口

curl -X POST "http://127.0.0.1:8000/ai/process" \
-H "Content-Type: application/json" \
-H "user-id: user001" \
-d '{
  "task": "meeting",
  "text": "今天会议主要讨论了新品上线计划。产品部表示原型已经完成，研发部预计两周内完成开发，市场部需要提前准备推广方案。会议决定下周三进行阶段评审。"
}'

3. 长文档总结接口

curl -X POST "http://127.0.0.1:8000/ai/long-summary" \
-H "Content-Type: application/json" \
-H "user-id: user001" \
-d '{
  "task": "summary",
  "text": "这里放入一篇很长的文档内容..."
}'

八、关键性能优化策略详解

1. 使用缓存减少重复调用

缓存是最直接有效的优化手段。对于AI办公系统来说，以下结果都适合缓存：

• 文档摘要；
• 会议纪要；
• 邮件润色结果；
• 周报模板；
• 知识库问答结果；
• 文档分类结果；
• 文档关键词提取结果。

缓存的关键是设计合理的Key。例如：

cache_key = hash(任务类型 + 模型名称 + Prompt版本 + 输入内容)

如果Prompt升级了，缓存结果可能不再适用，因此建议加入Prompt版本号。

2. 控制Prompt长度

很多AI办公系统慢，是因为每次都传入大量无关内容。优化Prompt时可以遵循：

• 删除与任务无关的历史对话；
• 只保留最近几轮必要上下文；
• 长文档先检索相关片段；
• 输出格式用简洁模板；
• 系统提示词不要写成大段说明书。

例如，不推荐：

你是一个非常聪明、非常专业、非常认真、非常负责的办公助手……

推荐：

你是AI办公助手。请用中文总结以下内容，输出3-5条要点，不编造信息。

简洁Prompt不仅速度更快，还能降低成本。

3. 长文档采用“分块总结 + 汇总总结”

对于超长文档，可以使用两阶段处理：

1. 将文档拆分为多个片段；
2. 分别总结每个片段；
3. 再对所有片段总结进行总总结。

这种方式的好处是：

• 避免超过模型上下文；
• 降低单次请求失败风险；
• 结构更清晰；
• 便于并行处理。

如果进一步优化，可以把每个分块总结并发执行：

tasks = [call_llm(build_summary_prompt(chunk)) for chunk in chunks]
partial_results = await asyncio.gather(*tasks)

不过需要注意，并发越高，模型API压力越大，也更容易触发限流。

4. 使用流式输出提升体验

很多时候，模型总耗时无法完全避免，但可以通过流式输出改善用户体验。用户看到内容逐字生成，会觉得系统响应更快。

适合流式输出的场景包括：

• 写作助手；
• 邮件生成；
• 报告生成；
• 会议纪要生成；
• 文案润色；
• 长答案问答。

不适合流式输出的场景：

• 需要严格JSON格式结果；
• 后端要先完整校验结果；
• 需要一次性入库的结构化任务。

5. 简单任务不要全部交给大模型

不是所有办公任务都需要调用大模型。例如：

大模型应该用于语义理解、生成、总结、推理等复杂任务。能用规则完成的任务，优先用规则。

6. 异步化和队列化

AI办公中有些任务耗时较长，例如：

• 批量总结100份文档；
• 自动生成完整项目报告；
• 分析大量会议记录；
• 从知识库中构建向量索引。

这些任务不适合同步等待，可以使用异步任务队列，例如：

• Celery；
• RQ；
• Dramatiq；
• Kafka + Worker；
• FastAPI BackgroundTasks。

用户提交任务后，系统立即返回任务ID，前端轮询任务状态，或者通过WebSocket推送结果。

7. 向量检索优化知识库问答

企业AI办公常见需求是“基于内部资料问答”。如果每次都把整个知识库传给模型，显然不可行。正确做法是：

1. 文档切分；
2. 生成向量；
3. 存入向量数据库；
4. 用户提问时检索相关片段；
5. 只把最相关的内容交给模型回答。

常见向量数据库包括：

• Milvus；
• FAISS；
• Chroma；
• Weaviate；
• Qdrant；
• Elasticsearch Vector Search。

这种方式通常称为 RAG，即检索增强生成。

九、生产环境优化建议

上面的源码适合学习和原型开发。如果要部署到生产环境，建议继续优化：

1. 使用Redis替代内存缓存

内存缓存存在问题：

• 服务重启后缓存丢失；
• 多实例之间无法共享；
• 不适合集群部署。

生产环境建议使用Redis，并设置合理TTL。

2. 增加日志监控

至少记录以下指标：

• 请求耗时；
• 模型耗时；
• Token使用量；
• 缓存命中率；
• 错误率；
• 用户请求次数；
• 不同任务类型的调用量。

有了数据，才能判断优化是否有效。

3. 设置超时时间

调用模型API时必须设置超时，避免接口长时间阻塞。

timeout = 60

如果请求超过指定时间，应及时返回错误提示或进入异步任务队列。

4. 增加失败重试

模型服务可能偶发失败，可以针对以下情况重试：

• 网络超时；
• 502；
• 503；
• 504；
• 临时限流。

但重试次数不能过多，否则会放大系统压力。

5. 控制并发数量

可以通过信号量限制同时调用模型的数量：

semaphore = asyncio.Semaphore(5)

async with semaphore:
    result = await call_llm(prompt)

这样可以避免瞬间大量请求打满模型服务。

十、总结

AI办公应用的核心价值是提升效率，但如果系统响应慢、成本高、稳定性差，就很难真正落地。性能优化不是某一个技巧，而是一整套工程体系。

本文从AI办公常见瓶颈出发，介绍了缓存、限流、文本分块、Prompt优化、异步调用、流式输出、长文档总结等关键策略，并提供了一个基于 FastAPI 的完整示例源码。

如果你正在开发AI办公助手，可以优先做好以下五件事：

1. 缓存重复请求结果，减少不必要的模型调用；
2. 精简Prompt和上下文，降低Token成本；
3. 长文档分块处理，避免超过上下文限制；
4. 使用流式输出，提升用户体验；
5. 增加限流和监控，保障系统稳定运行。

AI办公的竞争不只是模型能力竞争，更是产品体验和工程优化能力的竞争。只有把模型能力与系统架构结合起来，才能真正做出高效、稳定、可持续的AI办公产品。