ChatGPT API实战：如何高效构建知识库问答系统

1次阅读

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

在实际应用中，ChatGPT API 虽然在生成自然语言回复方面表现出色，但当涉及到需要从特定知识库中检索信息时，直接使用 API 调用往往效果不佳。本文将带你一步步构建一个基于知识库的问答系统，解决长上下文管理和知识检索精度的问题。

上下文长度限制 ：ChatGPT API 对每次请求的 token 数量有限制（目前 GPT-3.5-turbo 是 4096 tokens），当知识库内容较多时，无法全部放入上下文。
检索精度问题 ：直接使用 API 进行问答时，模型可能会 ” 编造 ” 答案（hallucination），特别是当问题涉及专业知识时。
成本控制 ：将大量知识库内容直接放入 prompt 会导致高昂的 API 调用成本。

需要把所有相关知识放入 prompt
上下文 token 快速消耗
检索效率低下

分层处理架构 ：
知识库预处理层：将知识库内容向量化并建立索引
检索层：根据用户问题检索最相关的知识片段
生成层：将检索结果和问题一起发送给 ChatGPT 生成回答
优势对比 ：
减少每次 API 调用的 token 使用量
提高回答的准确性
降低 API 调用成本

import openai
from typing import List, Dict
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
import pickle
import os

class KnowledgeBase:
    def __init__(self, openai_key):
        self.openai_key = openai_key
        openai.api_key = openai_key
        self.knowledge_embeddings = {}

    def add_document(self, document_id: str, text: str):
        """将文档文本转化为 embedding 并存储"""
        response = openai.Embedding.create(
            input=text,
            model="text-embedding-ada-002"
        )
        self.knowledge_embeddings[document_id] = {
            'text': text,
            'embedding': response['data'][0]['embedding']
        }

    def save(self, filepath):
        """保存知识库到文件"""
        with open(filepath, 'wb') as f:
            pickle.dump(self.knowledge_embeddings, f)

    def load(self, filepath):
        """从文件加载知识库"""
        with open(filepath, 'rb') as f:
            self.knowledge_embeddings = pickle.load(f)

class ConversationManager:
    def __init__(self, max_context_length=3000):
        self.max_context_length = max_context_length
        self.context = []

    def add_message(self, role: str, content: str):
        """添加对话消息到上下文"""
        self.context.append({"role": role, "content": content})

    def get_relevant_context(self, question_embedding, knowledge_base, top_n=3):
        """根据问题检索最相关的知识片段"""
        similarities = {}
        for doc_id, doc_data in knowledge_base.knowledge_embeddings.items():
            sim = cosine_similarity([question_embedding],
                [doc_data['embedding']]
            )[0][0]
            similarities[doc_id] = sim

        sorted_docs = sorted(similarities.items(), key=lambda x: x[1], reverse=True)[:top_n]
        return [knowledge_base.knowledge_embeddings[doc_id]['text'] for doc_id, _ in sorted_docs]

    def prepare_prompt(self, question, relevant_knowledge):
        """准备发送给 ChatGPT 的 prompt"""
        knowledge_str = '\n'.join([f"知识片段 {i+1}: {text}" for i, text in enumerate(relevant_knowledge)])
        prompt = f""" 根据以下知识回答问题:
{knowledge_str}

问题: {question}
答案:"""

        # 确保不超过 token 限制
        estimated_tokens = len(prompt.split()) * 1.3  # 粗略估计
        if estimated_tokens > self.max_context_length:
            # 简化知识片段
            knowledge_str = '\n'.join([text[:500] for text in relevant_knowledge])
            prompt = f""" 根据以下知识回答问题:
{knowledge_str}

问题: {question}
答案:"""

        return prompt

class ChatGPTWrapper:
    def __init__(self, openai_key, model="gpt-3.5-turbo"):
        self.openai_key = openai_key
        self.model = model
        openai.api_key = openai_key

    def generate_response(self, prompt, max_retries=3):
        """调用 ChatGPT API 生成回答，包含重试逻辑"""
        for attempt in range(max_retries):
            try:
                response = openai.ChatCompletion.create(
                    model=self.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.7
                )
                return response.choices[0].message.content
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # 指数退避

缓存策略 ：
缓存常见问题的答案
缓存文档的 embeddings，避免重复计算
批量处理 embeddings：
当需要处理大量文档时，使用 OpenAI 的批量 embedding 接口

示例代码：

def batch_add_documents(self, documents: Dict[str, str], batch_size=20):
    """批量添加文档"""
    doc_ids = list(documents.keys())
    for i in range(0, len(doc_ids), batch_size):
        batch_ids = doc_ids[i:i+batch_size]
        batch_texts = [documents[doc_id] for doc_id in batch_ids]

        response = openai.Embedding.create(
            input=batch_texts,
            model="text-embedding-ada-002"
        )

        for j, doc_id in enumerate(batch_ids):
            self.knowledge_embeddings[doc_id] = {'text': documents[doc_id],
                'embedding': response['data'][j]['embedding']
            }

异步处理 ：
对于大规模知识库，使用异步方式处理 embedding 生成

知识库版本一致性 ：
当知识库更新时，确保所有服务节点同步更新
可以使用版本号或时间戳标记知识库版本
上下文 token 超限 ：
实时监控 token 使用量
当接近限制时，优先保留最相关的上下文

示例检查函数：

def estimate_token_usage(self, text):
    """粗略估计文本的 token 数量"""
    return len(text.split()) * 1.3  # 英语大约 1.3 tokens/word

敏感信息过滤 ：
在将用户问题和知识库内容发送给 API 前，进行敏感词过滤
可以使用正则表达式或专业过滤库

示例代码：

SENSITIVE_WORDS = [...]  # 定义敏感词列表

def filter_sensitive_content(self, text):
    for word in SENSITIVE_WORDS:
        text = text.replace(word, '[REDACTED]')
    return text