使用Pydanticai建立生产级LLM驱动的应用

在软件开发的快速发展的景观中，人工智能，数据验证和数据库管理的交集开辟了前所未有的可能性。这篇博客文章探讨了使用最新的Pydanticai Framework和Google的Gemini-1.5模型来探讨SQL代码生成和SQL代码说明的创新方法，以展示尖端AI技术如何简化和增强数据库查询开发。

对于开发人员，数据科学家和数据分析师，此探索从自然语言处理中瞥见了智能代码生成的未来，在这种情况下，可以轻松而准确地创建复杂的数据库查询。

学习目标

• 了解Pydantic和Pydanticai的基本面。
• 了解如何实现AI驱动的SQL代码生成系统。
• 在自然语言中探索双子座-1.5频率的功能，以进行SQL翻译。
• 洞悉用于构建数据库交互的智能AI代理。

本文作为数据科学博客马拉松的一部分发表。

目录

• 什么是pydanticai？
• Pydanticai的示例
• 什么是AI代理？
• 什么是代理工作流程？
• AI代理和代理工作流的现代用法
• 什么是Pydanticai框架？
• 开始您的项目
• 实施项目的逐步指南
• 结论
• 常见问题

什么是pydanticai？

Pydanticai是一个强大的Python库，可革新数据验证和类型检查。它为定义数据模型提供了一种声明性的方法，从而易于创建和验证复杂的数据结构。

Pydantic的重要特征包括：

定制

• 验证广泛的数据类型，包括原始类型和复杂的嵌套结构。
• 几乎支持任何python对象进行验证和序列化

灵活性

允许控制数据验证严格性：

• 将数据胁到预期类型
• 在需要时执行严格的类型检查

序列化

• 支持Pydantic对象，字典和JSON之间的无缝转换。
• 启用自我文献的API和与使用JSON模式的工具的兼容性。

表现

• 核心验证逻辑以生锈为特殊的速度和效率编写。
• 适用于可扩展REST API等高通量应用的理想选择。

生态系统

• 广泛用于流行的Python库中，例如Fastapi，Langchain，LlamainDex等。
• 如果没有pydantic，就无法实施现代的代理LLM。

Pydanticai的示例

Pydanticai简化了Python中的数据验证和类型检查，使其成为创建强大数据模型的强大工具。让我们探索一些展示其功能的实用示例。

基本数据验证

来自Pydantic Import Basemodel


类用户（基本模型）：
    名称：str
    年龄：int


＃有效数据
用户=用户（name =“ Alice”，年龄= 30）
打印（用户）

打印（“ ==================================
＃无效数据（年龄是字符串）
尝试：
    用户=用户（name =“ Alice”，age =“三十”）
除例外为E：
    打印（E）

上面的代码使用Pydantic的基本模型定义了用户模型，将名称作为字符串和年龄作为整数执行。它验证了正确的数据，但是当提供了无效的数据（年龄的字符串）时，会引起验证错误。

输出：

自动型强制

来自Pydantic Import Basemodel
类产品（基本模型）：
    价格：浮动
    数量：int

＃不匹配类型的数据
产品=产品（价格=“ 19.99”，数量=“ 5”）
印刷（产品）

打印（类型（product.price））
打印（类型（product.Quantity））

在这里，价格为浮动和数量的产品模型作为整数。 Pydantic自动将字符串输入（“ 19.99”和“ 5”）驱动到正确的类型（float and int）中，以演示其类型的转换功能。

输出：

嵌套模型

来自Pydantic Import Basemodel
班级地址（基本模型）：
    街：Str
    城市：Str
类用户（基本模型）：
    名称：str
    地址：地址

＃有效数据
user =用户（name =“ bob”，地址= {“街”：“ 123 Main St”，“ City”：“ Wonderland”}）
打印（用户）

＃访问嵌套属性
打印（user.address.city）

在这里，我们定义一个包含地址模型的嵌套用户模型。 Pydantic允许嵌套验证，并自动转换字典成模型。有效的数据初始化了用户对象，您可以直接访问诸如“ user.address.city”之类的嵌套属性。

输出：

使用自定义规则验证

来自Pydantic Import basemodel，field，field_validator

类用户（基本模型）：
    名称：str
    年龄：int = field（...，gt = 0，description =“年龄必须大于零”）

    @field_validator（“名称”）
    def name_must_be_non_empty（cls，value）：
        如果不是value.strip（）：
            提高价值Error（“名称不能为空”）
        返回值

＃有效数据
用户=用户（name =“ Charlie”，年龄= 25）
打印（用户）

＃无效数据
尝试：
    用户=用户（name =“”，年龄= -5）
除例外为E：
    打印（E）

在这里，我们定义具有验证规则的用户模型，年龄必须大于0，并且该名称不能为空（通过name_must_be_non_empty方法验证）。有效的数据创建了一个用户实例，而无效的数​​据（空名称或负年龄）会引发详细的验证错误，以证明Pydantic的验证功能。

输出：

这些是Pydantic的一些核心例子，我希望它们能帮助您了解数据验证的基本原理。

什么是AI代理？

AI代理是智能系统，旨在自主执行任务，做出决策并与其环境互动以实现特定目标。这些代理并不是新的，而是生成AI的近期快速发展，并将其与代理相结合，使代理软件在新时代开发。现在，代理可以处理输入，执行操作并动态适应。他们的行为模仿了类似人类的问题解决，使他们能够在人类干预的情况下在各个领域发挥作用。

什么是代理工作流程？

代理工作流程是指由一个或多个AI代理管理和执行的结构，目标驱动的任务序列。统一僵化的传统工作流程，代理工作流具有适应性，自主性和上下文意识。这些工作流程中的AI代理可以独立做出决策，委派子任务并从反馈中学习，从而导致有效且优化的结果。

AI代理和代理工作流的现代用法

AI代理和代理工作流程的整合通过使复杂的任务自动化，增强决策和推动效率来彻底改变行业。这些智能系统动态适应，从而使跨不同领域的更智能解决方案。

业务自动化

AI代理通过聊天机器人，电子邮件管理和销售管道优化的重复任务自动化重复任务。它们通过从高价值任务中释放人力资源来提高生产率。

软件开发

通过生成，测试和调试代码，通过AI驱动的代理加速软件生命周期，从而减少了开发时间和人为错误。

卫生保健

AI代理协助医疗诊断，患者监测和治疗个性化，改善医疗保健提供和操作效率。

金融

金融系统中的代理工作流程自动化欺诈检测，风险评估和投资分析，实现更快，更可靠的决策。

电子商务

情报机构可以增强购物体验中的个性化，优化产品建议和客户服务。

AI代理和代理工作流的兴起意味着向能够管理复杂过程的高度自主系统的转变。它们的适应性和学习能力使它们对于现代行业来说是必不可少的，推动了跨领域的创新，可扩展性和效率。随着AI的不断发展，AI代理将进一步集成到我们的日常工作流程中，从而改变了如何管理和执行任务。

什么是Pydanticai框架？

Pydanticai是由Pydantic，Fastapi的创建者开发的Python代理框架，用于简化利用生成AI的生产级应用的构建，它强调了类型的安全性，模型 - 敏锐的设计以及与大语言模型（LLMS）的无缝集成。

pydanticai的关键功能包括：

• 模型 - 不合Snostic的支持： Pydanticai与各种模型兼容，包括OpenAI，Antropic，Gemini，Groq，Mistral和Ollama，并具有直接的界面，以结合其他模型。
• 类型安全：利用Python的类型系统和Pydantic的验证，Pydanticai确保了稳健而可扩展的代理开发。
• 依赖注入系统：它引入了一种新颖的TY-SAFE依赖注入机制，增强了测试和评估驱动的发展。
• 结构化响应验证：利用Pydantic的验证功能，确保准确可靠的结构响应。
• LogFire集成：提供与Pydantic LogFire的集成，以增强对LLM功率应用程序的调试和监视。

这是Pydanticai的最小例子：

导入操作系统
来自Pydantic_ai进口代理
来自pydantic_ai.models.gemini导入geminimodel
来自dotenv import load_dotenv

load_dotenv（）

gemini_api_key = os.getenv（“ <google_api_key>”）

型号= geminimodel（
    “ Gemini-1.5-Flash”，
    api_key = gemini_api_key，
）

代理=代理（
    模型=模型，
    system_prompt =“要简洁，用一个句子回复。”，，
）

结果= agent.run_sync（'“ Hello World”来自哪里？'）
打印（result.data）</google_api_key>

输出：

现在是时候做一些真实的事情了。我们将使用Pydanticai代理框架构建Postgres SQL查询生成。

开始您的项目

通过逐步指南为您的项目奠定基础，以设置基本工具和环境。

设置环境

我们将为该项目创建一个Conda环境。

 ＃创建一个环境
$ conda create -name sql_gen python = 3.12

＃激活env
$ conda激活sql_gen

现在，创建一个项目文件夹

＃创建一个文件夹
$ mkdir sql_code_gen
＃更改为文件夹
$ CD SQL_CODE_GEN

安装Postgres和Load Database

要安装Postgres，PSQL-Command-Tools和Pgadmin-4，只需转到EDBDONDON for Systaller的系统，然后一次安装所有工具即可。

现在，从Hereand下载DVDRENALE数据库，将其加载到Postgres，请按照以下步骤

步骤1：打开终端

PSQL -U Postgres

＃它会要求密码放置

步骤2：创建数据库

＃在Postgres =＃中

创建数据库DVDRENTAL；

步骤3：终端的命令

现在，退出PSQL命令，然后输入终端

pg_restore -u postgres -d dvdrenaltal d：/sampledb/postgres/dvdrental.tar

步骤4：连接到PSQL

现在，连接到PSQL并检查数据库是否已加载。

 PSQL -U Postgres

＃与dvDretal
\ c dvDretal

＃让我们看看桌子

\ dt

输出：

如果您看到上表，则可以。我们都准备启动我们的主要项目。

现在，将必要的Python库安装到SQL_GEN CONDA ENV中。

 conda激活sql_gen

＃安装库
PIP安装pydantic asyncpg asyncio pydantic-ai 

PIP安装Python-Dotenv Fastapi Google-generativeai

PIP安装DevTools注释类型类型扩展

项目结构

我们的项目有4个文件，即主，模型，服务和架构。

 sql_query_gen/
|
| -main.py
|  -  models.py
| -schema.py
| -service.py
|  - 
| --__ init__.py
|  - 。吉蒂尼尔

实施项目的逐步指南

通过此全面的实施指南，深入研究详细的步骤和实用技术，以将您的项目从概念带到现实。

Pydantic模型

我们将首先在模型中创建数据模型。

从数据级导入数据级
从输入导入注释
导入异步
从Annotated_types进口minlen
来自Pydantic Import Basemodel，field

@DataClass
班级级别：
    conn：asyncpg.connection

班级成功（基本模型）：
    sql_query：注释[str，minlen（1）]
    说明：str = field（“”，description =“ SQL查询的说明，作为Markdown”）

类InvalidRequest（basemodel）：
    error_message：str

在上述代码中，

• DEPS类管理数据库连接依赖性。 @dataclass会自动生成__init__和__repr__等特殊方法。 conn键入为`asyncpg.connection` ，代表一个活动的后Ql连接。该设计遵循依赖注入模式，使代码更具测试和可维护。
• 成功类代表成功的SQL Query Generation， SQL_Query必须是一个非空字符串（ MinLen （1）），并使用注释来添加验证约束。说明是一个可选字段，带有默认的空字符串。
• InvalidRequest类是错误响应模型，代表了失败的SQL Query Generation尝试。

该代码为数据库连接管理，输入验证，结构化响应处理和错误处理建立了基础。

服务模块

现在，我们将在服务模块中为SQL生成实施Pydanticai服务。

导入库和配置

导入操作系统
从输入进口联盟
来自dotenv import load_dotenv
导入异步
从typing_extensions导入typealias
来自pydantic_ai进口代理，模型，runco​​ntext
来自pydantic_ai.models.gemini导入geminimodel
从模式导入db_schema
从模型导入深度，成功，无效

要配置，请在项目root中创建.ENV文件，然后将您的Gemini API密钥放在那里

＃.env

gemini_api_key =“ asgfhkdhjy457gthjhajblsd”

然后在service.py文件中：

 load_dotenv（）

gemini_api_key = os.getEnv（“ google_api_key”）

它将从`````'''加载Google API密钥。 env`文件。

创建模型和代理

响应：typealias = union [成功，无效]

型号= geminimodel（
    “ Gemini-1.5-Flash”，
    api_key = gemini_api_key，
）

代理=代理（
    模型，
    result_type =响应，＃类型：忽略
    dep_type = deps，
）

• 首先定义可以取得成功或无效的响应类型
• 用API键初始化Gemini 1.5 Flash模型
• 创建具有指定响应和依赖类型的Pydanticai代理

系统提示定义

现在，我们将定义SQL查询生成的系统提示。

 @agent.system_prompt
异步def system_prompt（） - > str：
    返回f“” \

鉴于以下邮政记录表，您的工作是
编写适合用户请求的SQL查询。

数据库架构：
{db_schema}

例子
    请求：查找所有租金大于$ 4.00的电影和“ PG”的评级
    响应：选择标题，lental_rate
    来自电影
    其中rental_rate> 4.00和等级='pg';
例子
    请求：查找最长长度的电影
    响应：选择标题，长度
    来自电影
    其中长度=（从胶片中选择最大（长度））;
例子
    请求：找到每个类别中电影的平均租金持续时间
    响应：选择C.NAME，AVG（F.Rental_duration）作为平均_rental_duration
    从类别c
    加入c.Category_id = fc.category_id on c.Category_id上
    加入FC.FILM_ID上的电影F = F.FILM_ID
    组由C.Name
    按平均订单_rental_duration desc;
”“”

在这里，我们定义了AI模型的基本上下文，并提供了示例查询以指导模型的响应。我们还将数据库架构信息包括在模型中，以便模型可以分析模式并产生更好的响应。

响应验证

为了使AI模型无错误的响应并符合项目要求，我们只是验证了响应。

 @Agent.Result_Validator
异步def validate_result（ctx：runco​​ntext [deps]，结果：响应） - >响应：
    如果IsInstance（结果，无效）：
        返回结果

    ＃双子座经常向SQL添加反弹
    result.sql_query = result.sql_query.replace（“ \\”，“”）
    如果不是结果。
        提高模型培训（“请创建一个选择查询”）

    尝试：
        等待ctx.deps.conn.execute（f“ divell {result.sql_query}”）
    除了asyncpg.exceptions.postgreserror为e：
        从e提出模型的（f“无效的sql：{e}”）
    别的：
        返回结果

在这里，我们将验证并处理生成的SQL查询

关键验证步骤：

• 如果结果是无效的，请立即返回，请清理额外的后斜线
• 确保查询是选择语句
• 使用PostgreSQL解释验证SQL语法
• 提高无效查询模型

数据库架构

要获取数据库架构，请打开您在Postgres设置期间安装的PGADMIN4，转到`` dvDrenaltal`数据库''，右键单击它，然后单击数据库`。
您将获得以下ERD图，现在从ERD生成SQL（请参阅图像上的圆形黑色标记）。

将架构复制到schema.py模块：

 ＃schema.py
db_schema =“”
开始;


创建表格，如果不存在public.actor
（（
    actor_id序列而不是null，
    first_name字符变化（45）caltrate pg_catalog。“默认”而不是null，
    last_name字符变化（45）整理pg_catalog。“默认”而不是null，
    last_update时间戳记没有时区，而不是null默认值（），
    约束actor_pkey主键（actor_id）
）；
。
。
。
  
。
。
。
”“”

上面的代码块被严重截断，要获得完整的代码，请访问项目回购。

现在，所有必要的模块都已完成，是时候实现主方法和测试了。

实施主要

我们将执行主要功能定义和提示处理。

导入异步
导入操作系统
导入系统
从输入进口联盟
来自dotenv import load_dotenv
导入异步
从DevTools导入调试
从typing_extensions导入typealias
来自Pydantic_ai进口代理
来自pydantic_ai.models.gemini导入geminimodel
从模型导入深度，成功，无效

load_dotenv（）

gemini_api_key = os.getEnv（“ google_api_key”）


响应：typealias = union [成功，无效]

型号= geminimodel（
    “ Gemini-1.5-Flash”，
    api_key = gemini_api_key，
）

代理=代理（
    模型，
    result_type =响应，＃类型：忽略
    dep_type = deps，
）


异步def main（）：
    如果Len（sys.argv）== 1：
        提示=“请创建一个选择查询”
    别的：
        提示= sys.argv [1]

    ＃连接到数据库
    conn =等待asyncpg.connect（
        用户=“ Postgres”，
        密码=“ avizyt”，
        主机=“ localhost”，
        端口= 5432，
        数据库=“ dvdretaltal”，
    ）
    尝试：
        deps = deps（conn）
        结果=等待代理。
        结果= debug（result.data）
        打印（“ ==========您的查询=========”）
        打印（debug（result.sql_query））
        打印（“ ===========解释=========”）
        打印（debug（结果。解释））

    最后：
        等待conn.close（）


如果__name__ ==“ __ -main __”：
    asyncio.run（main（））

首先，在这里定义异步主函数，然后检查客户端查询的命令行参数。如果没有提供ARG，请使用默认提示。

然后，我们将Postgres连接参数设置为与DVDRENTAR数据库服务连接。

在Try Block中，使用数据库连接创建一个DEPS实例，使用提示来运行AI代理，使用调试函数（ PIP INSTALS DEVTOOLS ）处理结果。然后打印格式的输出，包括生成的SQL查询和查询的说明。之后，我们终于关闭了数据库连接。

现在运行主模块，如下所示：

 ＃在终端中
python main.py“获取每个客户的租金总数”

输出：

测试PGADMIN4中的SQL查询后：

哇！它像我们想要的一样工作。测试更多这样的查询并享受学习。

结论

该项目代表了使数据库交互更加直观和访问的重要一步。通过将AI的功能与强大的软件工程原则相结合，我们创建了一个工具，不仅可以生成SQL查询，而且以安全，教育意义和实用的方式进行现实使用的方式。

该实施的成功表明了AI增强而不是取代传统数据库操作的潜力，从而为学习和生产力提供了有价值的工具。

项目回购- 此项目中使用的所有代码都可以在此处获得。

关键要点

• Pydanticai启用智能，上下文感知的代码生成。
• Gemini-1.5-Flash为技术任务提供了高级的自然语言理解。
• AI代理可以改变我们与数据库交互的方式并生成代码。
• 强大的验证在AI生成的代码系统中至关重要。

常见问题

Q 1。pydanticai在SQL Generation中的优势是什么？

A. Pydanticai提供了类型安全的，经过验证的代码生成，具有内置错误检查和上下文理解。

问2。Gemini-1.5-Flash如何为该项目做出贡献？

A. Gemini模型提供了先进的自然语言处理，将复杂的人类查询转化为精确的SQL语句。

问3。该项目可以扩展到其他AI应用程序吗？

答：绝对！该体系结构可以用于代码生成，数据转换和跨各个域的智能自动化。

本文所示的媒体不由Analytics Vidhya拥有，并由作者酌情使用。

以上是使用Pydanticai建立生产级LLM驱动的应用的详细内容。更多信息请关注PHP中文网其他相关文章！