ChatGPT API 知识库与上下文保存实战：从原理到最佳实践

1次阅读

没有评论

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

在使用 ChatGPT API 构建对话系统时，开发者常遇到两个核心问题：

知识库管理困难 ：ChatGPT 本身不具备长期记忆功能，每次对话都是独立的。对于需要专业领域知识的场景，每次都要重新注入相关知识，效率低下且成本高。
上下文断裂 ：虽然可以通过在请求中传递历史对话来维持上下文，但随着对话轮次增加，Token 数量迅速膨胀，不仅增加 API 调用成本，还可能遇到模型的最大 Token 限制（如 GPT-3.5-turbo 的 4096 Token 限制）。

这些痛点导致对话系统显得不够智能，用户体验大打折扣。

针对上述问题，目前主流解决方案有以下几种：

优点：数据持久化，可扩展性强，支持复杂查询
缺点：需要额外维护数据库，增加系统复杂度
适用场景 ：需要长期保存大量知识或对话历史的场景

优点：支持语义搜索，能处理大规模知识库
缺点：实现复杂，需要额外向量数据库（如 Pinecone、Milvus）
适用场景 ：需要根据用户问题动态检索相关知识的场景

优点：实现简单，无需额外基础设施
缺点：上下文长度仍受 Token 限制
适用场景 ：短对话场景，对上下文长度要求不高的应用

下面我们展示一个完整的 Python 实现，使用 PostgreSQL 作为外部存储：

import psycopg2
from psycopg2 import sql
import openai

# 初始化数据库连接
def init_db():
    conn = psycopg2.connect(
        dbname="chatgpt_db",
        user="postgres",
        password="yourpassword",
        host="localhost"
    )
    cur = conn.cursor()

    # 创建知识库表
    cur.execute("""
        CREATE TABLE IF NOT EXISTS knowledge_base (
            id SERIAL PRIMARY KEY,
            topic TEXT NOT NULL,
            content TEXT NOT NULL,
            embedding vector(1536)
        )
    """)

    # 创建对话历史表
    cur.execute("""
        CREATE TABLE IF NOT EXISTS conversation_history (
            session_id TEXT PRIMARY KEY,
            history JSONB NOT NULL,
            last_updated TIMESTAMP NOT NULL
        )
    """)

    conn.commit()
    return conn

# 保存知识到数据库
def save_knowledge(conn, topic, content):
    # 生成嵌入向量
    response = openai.Embedding.create(
        input=content,
        model="text-embedding-ada-002"
    )
    embedding = response['data'][0]['embedding']

    cur = conn.cursor()
    cur.execute(
        """
        INSERT INTO knowledge_base (topic, content, embedding)
        VALUES (%s, %s, %s)
        """,
        (topic, content, embedding)
    )
    conn.commit()

# 检索相关知识
def retrieve_knowledge(conn, query, top_k=3):
    # 生成查询的嵌入向量
    response = openai.Embedding.create(
        input=query,
        model="text-embedding-ada-002"
    )
    query_embedding = response['data'][0]['embedding']

    cur = conn.cursor()
    cur.execute(
        """
        SELECT content, 1 - (embedding <=> %s) as similarity
        FROM knowledge_base
        ORDER BY similarity DESC
        LIMIT %s
        """,
        (query_embedding, top_k)
    )

    return [row[0] for row in cur.fetchall()]

# 保存对话历史
def save_conversation(conn, session_id, history):
    cur = conn.cursor()
    cur.execute(
        """
        INSERT INTO conversation_history (session_id, history, last_updated)
        VALUES (%s, %s, NOW())
        ON CONFLICT (session_id) DO UPDATE
        SET history = EXCLUDED.history,
            last_updated = NOW()
        """,
        (session_id, json.dumps(history))
    )
    conn.commit()

# 加载对话历史
def load_conversation(conn, session_id):
    cur = conn.cursor()
    cur.execute(
        "SELECT history FROM conversation_history WHERE session_id = %s",
        (session_id,)
    )
    result = cur.fetchone()
    return json.loads(result[0]) if result else []