ChatGPT API 知识库集成实战：从架构设计到生产环境避坑指南

1次阅读

没有评论

共计 2242 个字符，预计需要花费 6 分钟才能阅读完成。

当我们尝试直接将企业知识库内容通过 ChatGPT API 进行查询时，会遇到几个关键问题：

Token 限制问题 ：GPT 模型有严格的 token 数量限制（如 gpt-3.5-turbo 的 4096 tokens），而知识库文档往往较大，无法完整放入上下文。
语义漂移现象 ：当简单截取文档片段时，模型容易丢失整体上下文，导致回答偏离预期。
冷启动延迟 ：每次查询都需要重新处理整个知识库，响应时间难以满足实时交互需求。

主流选择有两种路径：

OpenAI 官方模型 （text-embedding-ada-002）：
优点：开箱即用，效果稳定
缺点：按调用次数计费，数据需传输到云端
开源模型 （如 all-MiniLM-L6-v2）：
优点：可本地部署，数据不出域
缺点：需要自行维护模型服务

固定窗口分块 ：简单按字符数切分，实现容易但可能破坏语义
语义分割 ：使用 NLP 模型识别段落边界（如 pytextrank），保持内容完整性

推荐采用重叠滑动窗口（overlap=20%）来缓解边界信息丢失问题。

结合两种检索方式提升召回率：

Embedding 语义搜索 ：计算查询与文档片段的余弦相似度
关键词召回 ：针对专有名词等精确匹配场景

from typing import List, Dict
from pydantic import BaseModel
import tiktoken

class DocumentChunk(BaseModel):
    text: str
    doc_id: str
    chunk_id: int
    embedding: List[float] = None

def split_with_overlap(text: str, chunk_size: int = 500, overlap: int = 100) -> List[str]:
    """滑动窗口分块实现"""
    if len(text) <= chunk_size:
        return [text]
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunks.append(text[start:end])
        start += (chunk_size - overlap)
    return chunks

import aiohttp
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
async def get_embeddings(texts: List[str], model: str = "text-embedding-ada-002") -> List[List[float]]:
    """批量获取 embedding（带自动重试）"""
    async with aiohttp.ClientSession() as session:
        resp = await session.post(
            "https://api.openai.com/v1/embeddings",
            json={"input": texts, "model": model},
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        result = await resp.json()
        return [item["embedding"] for item in result["data"]]

建议测试不同 chunk size 对以下指标的影响：

回答准确率（需人工评估）
API 调用延迟
Token 消耗量

from datetime import datetime
import hashlib

class EmbeddingCache:
    """带时效的本地缓存"""
    def __init__(self, ttl_hours=24):
        self.cache = {}
        self.ttl = ttl_hours * 3600

    def get_key(self, text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()

    def get(self, text: str) -> Optional[List[float]]:
        key = self.get_key(text)
        if key in self.cache and (datetime.now() - self.cache[key]["time"]).seconds < self.ttl:
            return self.cache[key]["embedding"]
        return None