使用 Azure Database for PostgreSQL 和 Python 生成 RAG 应用程序

7 分钟

现在，数据库已准备好嵌入索引和向量索引，现在可以将检索转换为工作应用程序了。目标是简单的：回答用户问题、从 PostgreSQL 检索最相关的区块，并通过 LangChain生成基于这些区块的答案。

在本单元中，你将了解在 Python 中生成 RAG 应用程序的基础知识。应用程序连接到数据库，对数据库运行相似性搜索，并将搜索结果传递给具有明确说明的模型，以避免幻觉。由于模型根据数据库数据接收定义良好的上下文，因此它可以生成更准确且更相关的答案，这些答案基于该上下文。

使用 PostgreSQL 和 LangChain 实现检索增强生成

回顾之前的单元，RAG 流程是一系列结合检索和生成的步骤。让我们来详细讲解：

用户提出问题。
数据库计算问题的嵌入，并使用向量索引查找数据库中最近的匹配项（顶部区块）。
顶级分块会与提示一起传递给模型，提示内容如下：“仅根据此上下文回答。如果不知道，就直接说明。”
模型根据提供的上下文返回自然语言答案。

LangChain 提供了一个框架，用于使用语言模型生成应用程序。它简化了连接到各种数据源、管理提示和处理响应的过程。通过将 Azure Database for PostgreSQL 与 Python 和 LangChain 相结合，可以创建一个功能强大的 RAG 应用程序，用于检索相关信息并生成准确的响应。虽然 RAG 应用程序可能会演变，但这些核心原则指导其开发。

在此应用程序中，将使用 LangChain AzureChatOpenAI 包装器与 Azure OpenAI 服务进行交互。

对于以下 Python 示例，请使用下表结构：

CREATE TABLE company_policies (
    id SERIAL PRIMARY KEY,
    title TEXT,
    policy_text TEXT,
    embedding VECTOR(1536)
);

在此表中，假设在列上 embedding 创建了矢量索引，以启用高效的相似性搜索。

让我们查看每个步骤的 Python 代码片段。大多数 RAG 应用程序都遵循类似的结构。

连接到数据库

保持连接的持续时间短且安全。在此示例中，机密位于环境变量中，并传递给连接。

import os, psycopg2
from contextlib import contextmanager

@contextmanager
def get_conn():
    conn = psycopg2.connect(
        host=os.getenv("PGHOST"),
        user=os.getenv("PGUSER"),
        password=os.getenv("PGPASSWORD"),
        dbname=os.getenv("PGDATABASE"),
        connect_timeout=10
    )
    try:
        yield conn
    finally:
        conn.close()

此脚本为数据库连接设置上下文管理器，确保它们在使用后正确关闭。现在是进行实际检索的时候了。

检索相关的区块

由于用户正在提出问题，因此你需要一个函数来查询具有该问题的数据库。该函数应基于问题返回相关区块。此示例假定矢量索引使用 余弦距离进行排名，因此使用相应的运算符类（<=>）进行查询。

def retrieve_chunks(question, top_k=5):
    sql = """
    WITH q AS (
        SELECT azure_openai.create_embeddings(%s, %s)::vector AS qvec
    )
    SELECT id, title, policy_text
    FROM company_policies, q
    ORDER BY embedding <=> q.qvec
    LIMIT %s;
    """
    params = (os.getenv("OPENAI_EMBED_DEPLOYMENT"), question, top_k)
    with get_conn() as conn, conn.cursor() as cur:
        cur.execute(sql, params)
        rows = cur.fetchall()
    return [{"id": r[0], "title": r[1], "text": r[2]} for r in rows]

因此，此函数根据用户的问题从数据库中检索相关区块。然后，这些区块用于在后续步骤中生成上下文感知答案。

使用 LangChain 生成答案

现在，你有一个函数来检索相关的区块，你需要使用这些区块生成答案。模型应仅使用检索到的上下文并引用策略标题。创建答案生成函数的时间。

from langchain_openai import AzureChatOpenAI

SYSTEM_PROMPT = """
You are a helpful assistant. Answer using ONLY the provided context.
If the answer is not in the context, say you don’t have enough information.
Cite policy titles in square brackets, e.g., [Vacation policy].
"""

def format_context(chunks):
    return "\n\n".join([f"[{c['title']}] {c['text']}" for c in chunks])

def generate_answer(question, chunks):
    llm = AzureChatOpenAI(
        azure_deployment=os.getenv("OPENAI_CHAT_DEPLOYMENT"),
        api_key=os.getenv("AZURE_OPENAI_API_KEY"),
        azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        temperature=0
    )
    context = format_context(chunks)
    messages = [
        ("system", SYSTEM_PROMPT),
        ("human", f"Question: {question}\nContext:\n{context}")
    ]
    return llm.invoke(messages).content

注释

使用 AzureChatOpenAI：

将消息作为（角色，内容）元组列表（或消息对象）传递。
通过 env vars 和构造函数提供 Azure 设置：AZURE_OPENAI_API_KEY、AZURE_OPENAI_ENDPOINT、azure_deployment 和api_version。
保留 temperature=0 用于事实性答案。较大的值会增加创造力，但可能会降低准确性。

此函数是 RAG 应用程序的核心，用于处理与语言模型的交互。请注意检索的区块的格式和包含在上下文中的方式。此外，系统提示旨在确保模型遵循提供的上下文，并减少可能不正确的 AI 生成的响应。最后，由语言模型处理消息，以使用 LangChain调用方法生成响应。该方法 invoke 使用格式化的消息调用，模型响应将作为自然语言文本返回。

将其绑在一起

最后需要的是一个简单的函数来运行完整流：

def answer_question(question):
    chunks = retrieve_chunks(question)
    if not chunks:
        return "I couldn’t find relevant content in the policy store."
    return generate_answer(question, chunks)

# Quick test
print(answer_question("How many vacation days do employees get?"))

在此示例中，检索数据库中的相关区块，并使用它们生成上下文感知答案。此函数调用演示从问题到答案生成的完整流程。首先，它调用 retrieve_chunks 函数来获取相关上下文（基本上是从数据库返回的行）。然后，此函数将该上下文 generate_answer 传递给与语言模型交互的函数，以生成最终答案（将行用作上下文的一部分）作为自然语言响应。完整流可确保答案以检索的数据为基础，从而提供更准确、更可靠的响应。

关键结论

实际的 RAG 应用程序会提出用户问题，在 SQL 中创建嵌入内容，使用向量索引提取数据库中最近的段落，并将该上下文只交给模型。为了减少幻觉，模型还被指示仅在提供的上下文中回答，并在信息缺失时予以说明。使连接保持生存期较短，参数化查询，并通过环境变量传递机密。在回答事实性问题时使用较低的温度，并包括简单的引用（标题或 ID），以便能够追溯响应。

现在，你应该深入了解如何使用 Azure Database for PostgreSQL 和 Python 生成检索扩充生成（RAG）应用程序。尽管 RAG 应用程序在实际方案中可能更为复杂，但核心原则保持不变。

反馈

此页面是否有帮助？