DuckDB 教程：构建 AI 项目

最近，DuckDB 正式脱离测试阶段发布稳定版，并随着各类数据框架将其集成进生态而迅速走红。现在正是学习 DuckDB 的好时机，帮助您跟上数据与 AI 领域瞬息万变的节奏。

在本教程中，我们将通过代码示例学习 DuckDB 及其关键特性。我们的主要关注点是如何将其与当前的 AI 框架集成。为此，我们将实践两个项目。首先，我们将构建一个基于检索增强生成（RAG）的应用，使用 DuckDB 作为向量数据库。然后，我们将把 DuckDB 用作 AI 查询引擎，使用自然语言而非 SQL 来分析数据。

什么是 DuckDB？

DuckDB 是一款现代、高性能的内存型分析型数据库管理系统（DBMS），用于支持复杂的分析查询。它是关系型（面向表）的 DBMS，并支持结构化查询语言（SQL）。

DuckDB 将 SQLite 的简洁易用与分析型工作负载所需的高性能能力相结合，是数据科学家与分析师的理想选择。

主要特性

操作简单：DuckDB 为无服务器架构，无外部依赖，并嵌入于宿主进程中。这使其安装与部署十分容易，构建时仅需 C++11 编译器。
功能丰富：支持广泛的 SQL 数据管理特性。DuckDB 也与 Python 与 R 深度集成，适用于数据科学与交互式数据分析。
分析查询高速：DuckDB 采用面向列的向量化查询执行引擎，针对分析优化，支持并行查询处理并高效处理大数据集。
免费且开源：基于宽松的 MIT 许可发布，可免费使用并开放源代码。
可移植性：DuckDB 无外部依赖，具备高度可移植性，可运行于多种操作系统（Linux、macOS、Windows）与 CPU 架构（x86、ARM）。通过 DuckDB-Wasm 甚至可在浏览器中运行。
可扩展性：DuckDB 支持灵活的扩展机制，可新增数据类型、函数、文件格式与 SQL 语法。
测试完备：通过持续集成进行高强度测试，测试套件包含数百万条查询，确保在不同平台与编译器上的稳定性与可靠性。

DuckDB 入门

在本节中，我们将学习如何设置 DuckDB、加载 CSV 文件、执行数据分析，并了解关系与查询函数。

我们先安装 DuckDB 的 Python 包。

pip install duckdb --upgrade技术支持

创建 DuckDB 数据库

要创建持久化数据库，使用 connect 函数并提供数据库名称即可。

import duckdb
con = duckdb.connect("datacamp.duckdb")技术支持

它会在您的本地目录中创建一个数据库文件。

我们将加载一个 CSV 文件并创建一个名为 "bank" 的表。我们使用的数据集在 DataLab 上，名为Bank Marketing。它包含葡萄牙某银行机构通过电话进行的直销活动。

要加载 CSV 文件，您需要先用 SQL 创建一个表，然后在 SQL 脚本中使用 read_csv() 函数加载文件。就这么简单。

随后，我们将通过执行显示数据库内所有表的 SQL 脚本，并使用 fetchdf 函数以 pandas DataFrame 展示结果，来验证我们的数据表。

注意：我们使用 DataCamp 的 DataLab 作为代码编辑器。DataLab 是一个云端 Jupyter Notebook，只要拥有 DataCamp 账号即可免费使用。

con.execute("""
    CREATE TABLE IF NOT EXISTS bank AS 
    SELECT * FROM read_csv('bank-marketing.csv')
""")
con.execute("SHOW ALL TABLES").fetchdf()技术支持

成功创建第一张表后，我们将运行一个入门级查询来分析数据，并将结果以 DataFrame 形式展示。

con.execute("SELECT * FROM bank WHERE duration < 100 LIMIT 5").fetchdf()技术支持

DuckDB 已原生集成到新版 DataCamp DataLab。通过阅读博文“DuckDB Makes SQL a First-Class Citizen on DataLab”并使用交互式 SQL 单元格来分析数据，了解更多信息。

DuckDB 关系（Relations）

DuckDB 的 relation 本质上是可通过关系型 API 查询的表。该 API 允许对诸如 Pandas DataFrame 等数据源串联多种查询操作。与其编写 SQL，您将通过串联多个 Python 函数来分析数据。

例如，我们将加载一个 CSV 文件来创建 DuckDB relation。要分析该表，您可以串联使用 filter 与 limit 函数。

bank_duck = duckdb.read_csv("bank-marketing.csv",sep=";")
bank_duck.filter("duration < 100").limit(3).df()技术支持

我们也可以通过从 DuckDB 数据库加载表来创建 relations。

rel = con.table("bank")
rel.columns技术支持

['age',
 'job',
 'marital',
 'education',
 'default',
 'housing',
 'loan',
 'contact',
 'month',
 'day_of_week',
 'duration',
 'campaign',
 'pdays',
 'previous',
 'poutcome',
 'emp.var.rate',
 'cons.price.idx',
 'cons.conf.idx',
 'euribor3m',
 'nr.employed',
 'y']技术支持

让我们编写一个使用多个函数分析数据的 relation。

rel.filter("duration < 100").project("job,education,loan").order("job").limit(3).df()技术支持

我们得到了三行记录，按 job 列排序，并对 duration 列进行了筛选。

DuckDB 查询函数

DuckDB 的查询函数允许在数据库内执行 SQL 查询，并返回可转换为多种格式以便进一步分析的结果。

在代码示例中，我们运行 SQL 查询来找出年龄大于 30 岁的客户的职业名称，统计每个职业被联系的客户数，并计算活动的平均持续时间。

参加SQL Fundamentals 技能路径，学习如何管理关系型数据库并执行简单数据分析查询。

res = duckdb.query("""SELECT 
                            job,
                            COUNT(*) AS total_clients_contacted,
                            AVG(duration) AS avg_campaign_duration,
                        FROM 
                            'bank-marketing.csv'
                        WHERE 
                            age > 30
                        GROUP BY 
                            job
                        ORDER BY 
                            total_clients_contacted DESC;""")
res.df()技术支持

现在我们将关闭与数据库的连接并释放相关资源，防止潜在的内存与文件句柄泄漏。

con.close()技术支持

如果您在运行上述代码时遇到问题，请查看Getting Started with DuckDB 工作空间。

使用 DuckDB 构建 RAG 应用

在第一个项目中，我们将学习使用 LlamaIndex 构建 RAG 应用，并用 DuckDB 作为向量数据库与检索器。

环境配置

安装创建与检索索引所需的全部 Python 包。

%%capture
%pip install duckdb
%pip install llama-index
%pip install llama-index-vector-stores-duckdb技术支持

导入所需的 Python 包与函数。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.vector_stores.duckdb import DuckDBVectorStore
from llama_index.core import StorageContext

from IPython.display import Markdown, display技术支持

配置 GPT-4o 与嵌入模型

语言模型方面，我们将使用最新的 GPT4o 模型与 OpenAI API。要创建大语言模型（LLM）客户端，您只需提供模型名称与 API key。

import os
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o",api_key=os.environ["OPENAI_API_KEY"])技术支持

接着，我们将使用 OpenAI 的 text-embedding-3-small 模型创建嵌入模型客户端。

注意：如果您的开发环境中已设置名为“OPENAI_API_KEY”的环境变量，则可无需显式提供 OpenAI API Key。

from llama_index.embeddings.openai import OpenAIEmbedding
embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
)技术支持

我们将把 OpenAI 的 LLM 与嵌入模型设置为全局默认，供所有 LlamaIndex 函数使用。

from llama_index.core import Settings

Settings.llm = llm
Settings.embed_model = embed_model技术支持

将 DuckDB 用作向量数据库

在本项目中，我们将从 data 文件夹加载 PDF 文件。这些 PDF 文件是来自 DataCamp 的教程，通过浏览器的打印功能保存为 PDF。

将文件夹路径传给 SimpleDirectoryReader 函数并加载数据。

documents = SimpleDirectoryReader("Data").load_data()技术支持

随后，使用现有数据库 “datacamp.duckdb” 创建名为“blog”的向量存储。然后将 PDF 中的数据转换为嵌入并存入该向量存储。

vector_store = DuckDBVectorStore(database_name = "datacamp.duckdb",table_name = "blog",persist_dir="./", embed_dim=1536)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)技术支持

为检查向量存储是否创建成功，我们将通过 DuckDB 的 Python API 连接数据库，并运行 SQL 查询展示数据库中的所有表。

import duckdb
con = duckdb.connect("datacamp.duckdb")

con.execute("SHOW ALL TABLES").fetchdf()技术支持

我们现在有两张表：一张是“bank”推广表，另一张是“blog”，即向量存储。“blog” 表中有一个 “embedding” 列存放所有嵌入。

创建一个简单的 RAG 应用

将索引转换为查询引擎，查询时会先在向量数据库中检索相似文档，并利用附加上下文生成回答。

为测试 RAG 查询引擎，我们将就教程提出一个问题。

query_engine = index.as_query_engine()
response = query_engine.query("Who wrote 'GitHub Actions and MakeFile: A Hands-on Introduction'?")
display(Markdown(f"<b>{response}</b>"))技术支持

答案正确。

The author of "GitHub Actions and MakeFile: A Hands-on Introduction" is Abid Ali Awan.技术支持

创建带记忆的 RAG 聊天机器人

现在，让我们创建一个进阶版的 RAG 应用，它会利用对话历史来生成回答。为此，我们需要创建一个聊天记忆缓冲区，然后构建包含记忆、LLM 与向量存储检索器的聊天引擎。

from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.chat_engine import CondensePlusContextChatEngine

memory = ChatMemoryBuffer.from_defaults(token_limit=3900)

chat_engine = CondensePlusContextChatEngine.from_defaults(
    index.as_retriever(),
    memory=memory,
    llm=llm
)

response = chat_engine.chat(
    "What is the easiest way of finetuning the Llama 3 model? Please provide step-by-step instructions."
)

display(Markdown(response.response))技术支持

我们询问了如何微调 Llama 3 模型，聊天引擎利用向量存储给出了高准确度的回答。

为验证记忆缓冲是否正常工作，我们将提出一个追问。

response = chat_engine.chat(
    "Could you please provide more details about the Post Fine-Tuning Steps?"
)
display(Markdown(response.response))技术支持

聊天引擎记住了先前的对话，并做出了相应的回答。

如果您在运行上述代码时遇到问题，请查看Building a RAG application with DuckDB 工作空间。

使用 LLM 构建 DuckDB SQL 查询引擎

在第二个项目中，我们将把 DuckDB 用作 SQL 查询引擎。这需要将数据库引擎与 GPT-4o 模型集成，以自然语言回答有关数据库的问题。

安装 duckdb-engine，以便通过 SQLAlchemy 创建数据库引擎。

%pip install duckdb-engine -q技术支持

加载 DuckDB 数据库

我们将使用 create_engine 函数加载 DuckDB 数据库，随后编写一个简单的 SQL 查询以检查是否加载成功。

from sqlalchemy import create_engine

engine = create_engine("duckdb:///datacamp.duckdb")
with engine.connect() as connection:
    cursor = connection.exec_driver_sql("SELECT * FROM bank LIMIT 3")
    print(cursor.fetchall())技术支持

完美。我们的 DuckDB 数据库引擎已就绪。

[(56, 'housemaid', 'married', 'basic.4y', 'no', 'no', 'no', 'telephone', 'may', 'mon', 261, 1, 999, 0, 'nonexistent', 1.1, 93.994, -36.4, 4.857, 5191.0, 'no'), (57, 'services', 'married', 'high.school', 'unknown', 'no', 'no', 'telephone', 'may', 'mon', 149, 1, 999, 0, 'nonexistent', 1.1, 93.994, -36.4, 4.857, 5191.0, 'no'), (37, 'services', 'married', 'high.school', 'no', 'yes', 'no', 'telephone', 'may', 'mon', 226, 1, 999, 0, 'nonexistent', 1.1, 93.994, -36.4, 4.857, 5191.0, 'no')]技术支持

接下来，我们需要使用 SQLDatabase 函数创建一个数据库工具。向其提供 engine 对象与表名。

from llama_index.core import SQLDatabase
sql_database = SQLDatabase(engine, include_tables=["bank"])技术支持

构建 SQL 查询引擎

使用 NLSQLTableQueryEngine 函数，并传入 LlamaIndex 的 SQL 数据库对象，来创建 SQL 查询引擎。

from llama_index.core.query_engine import NLSQLTableQueryEngine

query_engine = NLSQLTableQueryEngine(sql_database)技术支持

用自然语言向查询引擎询问有关 “bank” 表的问题。

response = query_engine.query("Which is the longest running campaign?")


print(response.response)技术支持

响应将以自然语言给出答案。是不是很棒？

The longest running campaign in the database has a duration of 4918 days.技术支持

我们再问一个复杂的问题。

response = query_engine.query("Which type of job has the most housing loan?")
print(response.response)技术支持

答案准确，并附带了更多信息。

The job type with the most housing loans is 'admin.' with 5559 housing loans. This is followed by 'blue-collar' with 4710 housing loans and 'technician' with 3616 housing loans. Other job types with significant housing loans include 'services', 'management', 'retired', 'entrepreneur', and 'self-employed'.技术支持

为了解后端发生了什么，我们将打印元数据。

print(response.metadata)技术支持

如我们所见，GPT-4o 先生成 SQL 查询，运行查询获得结果，然后基于结果生成回答。这个多步骤流程仅用两行代码即可完成。

{'d4ddf03c-337e-4ee6-957a-5fd2cfaa4b1c': {}, 'sql_query': "SELECT job, COUNT(housing) AS housing_loan_count\nFROM bank\nWHERE housing = 'yes'\nGROUP BY job\nORDER BY housing_loan_count DESC;", 'result': [('admin.', 5559), ('blue-collar', 4710), ('technician', 3616), ('services', 2050), ('management', 1490), ('retired', 892), ('entrepreneur', 779), ('self-employed', 740), ('unemployed', 557), ('housemaid', 540), ('student', 471), ('unknown', 172)], 'col_keys': ['job', 'housing_loan_count']}技术支持

项目完成后请关闭引擎。

engine.close()技术支持

如果您在运行上述代码时遇到问题，请查看DuckDB SQL Query Engine 工作空间。

结语

DuckDB 速度快、易用，并能与众多数据与 AI 框架无缝集成。作为数据科学家，您会发现只需几分钟就能熟悉其 API，并像使用其他 Python 包一样开始上手。DuckDB 的一大优势是无依赖，这意味着几乎在任何地方都能使用，无需担心托管或额外配置。

在本教程中，我们了解了 DuckDB 及其关键特性，并探索了 DuckDB 的 Python API，用其创建数据表并进行简单数据分析。教程下半部分包含两个项目：其一是以 DuckDB 为向量数据库的检索增强生成（RAG）应用，其二演示了将 DuckDB 作为 SQL 查询引擎的用法。

在使用 SQL 查询引擎或将数据库与 AI 集成之前，您需要具备 SQL 与数据分析的基础知识。您也许会写查询，但不一定知道该问什么问题？这正是基础数据分析与 SQL 知识的用武之地。您可以通过完成Associate Data Analyst in SQL 职业路径来获得这些知识。

作者

Abid Ali Awan

LinkedIn Twitter

作为一名持证的数据科学家，我热衷于利用前沿技术打造创新的机器学习应用。凭借在语音识别、数据分析与报告、MLOps、对话式人工智能以及自然语言处理方面的扎实背景，我不断打磨构建智能系统的能力，力求带来切实影响。除技术专长外，我也擅长沟通，能够将复杂概念提炼为清晰、简明的表述。因此，我成为数据科学领域备受关注的博主，与不断壮大的数据专业人士社区分享洞见与实践经验。目前，我专注于内容创作与编辑，借助大语言模型打造有力且吸引人的内容，帮助企业与个人更好地发挥数据价值。

主题

人工智能

主题

技术