RAG在PostgreSQL上的实现:使用Azure Container Apps和OpenAI构建智能问答系统

随着人工智能技术的快速发展,越来越多的企业和开发者开始探索如何将AI能力整合到现有的应用系统中。本文将介绍一种基于检索增强生成(Retrieval Augmented Generation, RAG)的方案,通过结合Azure Container Apps、PostgreSQL和OpenAI,实现一个能够智能回答数据库中信息的问答系统。

RAG技术简介

检索增强生成(RAG)是一种结合了检索系统和生成模型的AI技术。它通过从知识库中检索相关信息,然后利用大语言模型生成最终答案,既保证了回答的准确性,又赋予了AI更强的推理和生成能力。在本项目中,RAG的实现主要包含以下几个关键步骤:

使用pgvector扩展对PostgreSQL数据库表进行向量化搜索
结合全文搜索,通过RRF(Reciprocal Rank Fusion)算法融合搜索结果
利用OpenAI的function calling将用户查询转换为SQL过滤条件
使用OpenAI的嵌入API将用户查询转换为向量

通过这些技术的结合,系统能够准确理解用户意图,快速检索相关信息,并生成符合上下文的自然语言回答。

系统架构概览

系统架构图

如上图所示,整个系统主要由以下几个部分组成:

前端:使用React和FluentUI构建的Web应用界面
后端:基于Python和FastAPI开发的API服务
数据库:Azure PostgreSQL灵活服务器
AI服务:Azure OpenAI

整个应用通过Azure Container Apps进行部署和托管,实现了高度的可扩展性和弹性。

核心功能实现

1. 向量化搜索

本项目使用了pgvector扩展来为PostgreSQL添加向量搜索能力。首先需要在数据库中创建表并添加向量字段:

from pgvector.django import VectorField

class Transcript(models.Model):
    content = models.TextField()
    embedding = VectorField(dimensions=1536)  # OpenAI ada-002模型的维度

然后使用OpenAI的Embedding API将文本转换为向量并存储:

import openai

def generate_embedding(text):
    response = openai.Embedding.create(
        input=text, 
        model="text-embedding-ada-002"
    )
    return response['data'][0]['embedding']

transcript = Transcript(content=text)
transcript.embedding = generate_embedding(text)
transcript.save()

2. 混合搜索实现

为了提高搜索的准确性,项目结合了向量搜索和全文搜索:

from django.contrib.postgres.search import SearchVector

def hybrid_search(query, top_k=5):
    # 向量搜索
    query_vector = generate_embedding(query)
    vector_results = Transcript.objects.order_by(models.F('embedding').cosine_distance(query_vector))[:top_k]
    
    # 全文搜索
    text_results = Transcript.objects.annotate(
        search=SearchVector('content')
    ).filter(search=query)[:top_k]
    
    # 使用RRF算法融合结果
    return combine_results(vector_results, text_results)

3. 查询转换

使用OpenAI的function calling功能,可以将自然语言查询转换为结构化的SQL条件:

import openai

def convert_to_sql(query):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-0613",
        messages=[{"role": "user", "content": query}],
        functions=[{
            "name": "generate_sql_condition",
            "description": "Generate SQL WHERE condition from natural language query",
            "parameters": {
                "type": "object",
                "properties": {
                    "condition": {"type": "string", "description": "SQL WHERE condition"}
                },
                "required": ["condition"]
            }
        }],
        function_call={"name": "generate_sql_condition"}
    )
    return response.choices[0].function_call.arguments