基于PydanticAI的AI代理增删查改应用

又回到Skolo Online真是太好了！在我们2025年的第一个教程里，我们将深入了解PydanticAI。在这篇文章里，我们将涵盖以下几点：

• Pydantic AI 入门教程 — “hello world” 示例
• 设置 PostgreSQL 数据库，
• 一个更复杂的 PydanticAI 实现案例，其中两个代理协同工作来处理用户请求。代理 1 理解用户意图，代理 2 则执行该意图。

最后我们将在一个小时内构建一个笔记应用程序，你可以提供类似的自然语言提示：“请为我创建一个笔记，并添加以下内容”，应用程序将理解该指令并在数据库中执行相应的操作，添加一个新的笔记条目。你还能列出所有可用的笔记或查看详细的笔记。

请看分步视觉指导：https://www.youtube.com/watch?v=OAbyC1kMj5w。

这条 YouTube 视频 — Pydantic AI 与两个智能体

你可能已经听说过“生成式AI”这个词——但你知道AI代理是什么吗？本质上，它们是更高层次的工具，利用先进的语言模型自主处理任务。它们可以解析用户的查询，提取相关的信息，并以结构化的方式与外部服务互动。

从生成式AI技术到AI代理助手

上图展示了AI是如何从基本的生成模型进步到更高级的AI代理，能够与各种工具协作。

PydanticAI 是一个由开发了 Pydantic（一个在许多 Python 库和框架中不可或缺的验证层）的团队设计的 Python 代理框架，旨在简化利用生成式人工智能编写生产应用程序的过程。PydanticAI 侧重于类型安全，并且能够与 mypy 和 pyright 等静态类型检查器无缝集成。

PydanticAI的主要特点包括：

• 结构化响应：利用 Pydantic 对模型输出进行验证和结构化，确保每次运行的一致性。
• 依赖注入系统：提供一种可选的依赖注入系统，以向代理的系统提示、工具和结果验证器提供数据和服务，从而加强测试和迭代开发。
• 流式响应：支持连续流式传输的 LLM 输出，并立即验证，以确保快速准确的结果。

目前PydanticAI的API还在早期测试版阶段，可能会有所调整。开发团队欢迎各位提供反馈，以改进和完善其功能。

请在安装 pydantic-ai 之前确保您的 Python 版本为 3.9 或更高 。

在终端中输入以下命令：```
运行python --version

然后，创建并激活一个虚拟环境，接着安装 pydantic-ai。

创建一个名为skoloenv的虚拟环境
运行 skoloenv/bin/activate
使用pip安装pydantic-ai库

# 要了解PydanticAI的核心概念

在 PydanticAI 中，主要的工作马车是 `Agent` 类。通过使用它，你可以对各种模型进行查询。你可以在此 [官方文档页面](https://ai.pydantic.dev/api/agent/) 中查看兼容的模型列表。下面是一个快速示例，展示如何用 **OpenAI** 模型启动一个 Agent：

from pydantic_ai import Agent  # 代理
from pydantic_ai.models.openai import OpenAIModel  # 开放AI模型

model = OpenAIModel('gpt-4o', api_key='请在此处添加你的API密钥')
agent = Agent(model)
result = agent.run_sync("比特币是什么？")
print(result.data)

我们建议将您的 API 键设置为环境变量：

export OPENAI_API_KEY='你的API密钥'

这里列出了所有你可以用于PydanticAI的可用模型。

KnownModelName = Literal[  
    "openai:gpt-4o",  
    "openai:gpt-4o-mini",  
    "openai:gpt-4-turbo",  
    "openai:gpt-4",  
    "openai:o1-preview",  
    "openai:o1-mini",  
    "openai:o1",  
    "openai:gpt-3.5-turbo",  
    "groq:llama-3.3-70b-versatile",  
    "groq:llama-3.1-70b-versatile",  
    "groq:llama3-groq-70b-8192-tool-use-preview",  
    "groq:llama3-groq-8b-8192-tool-use-preview",  
    "groq:llama-3.1-70b-specdec",  
    "groq:llama-3.1-8b-instant",  
    "groq:llama-3.2-1b-preview",  
    "groq:llama-3.2-3b-preview",  
    "groq:llama-3.2-11b-vision-preview",  
    "groq:llama-3.2-90b-vision-preview",  
    "groq:llama3-70b-8192",  
    "groq:llama3-8b-8192",  
    "groq:mixtral-8x7b-32768",  
    "groq:gemma2-9b-it",  
    "groq:gemma-7b-it",  
    "gemini-1.5-flash",  
    "gemini-1.5-pro",  
    "gemini-2.0-flash-exp",  
    "vertexai:gemini-1.5-flash",  
    "vertexai:gemini-1.5-pro",  
    "mistral:mistral-small-latest",  
    "mistral:mistral-large-latest",  
    "mistral:codestral-latest",  
    "mistral:mistral-moderation-latest",  
    "ollama:codellama",  
    "ollama:gemma",  
    "ollama:gemma2",  
    "ollama:llama3",  
    "ollama:llama3.1",  
    "ollama:llama3.2",  
    "ollama:llama3.2-vision",  
    "ollama:llama3.3",  
    "ollama:mistral",  
    "ollama:mistral-nemo",  
    "ollama:mixtral",  
    "ollama:phi3",  
    "ollama:qwq",  
    "ollama:qwen",  
    "ollama:qwen2",  
    "ollama:qwen2.5",  
    "ollama:starcoder2",  
    "claude-3-5-haiku-latest",  
    "claude-3-5-sonnet-latest",  
    "claude-3-opus-latest",  
    "test",  
]

# 使用外部工具扩展 PydanticAI

我们将利用PostgreSQL数据库，并将其集成到我们的PydanticAI代理中，这样代理就可以执行数据库操作了。

# PostgreSQL 环境配置

首先，你需要一个正常运作且干净的PostgreSQL实例。在上述链接的视频中，我展示了如何在DigitalOcean上设置一个。

你可以通过这个推荐链接开始使用 DigitalOcean，<https://m.do.co/c/7d9a2c75356d>。

一旦你有了数据库（DB），拿到连接字符串——你下一步需要用到它。

然后安装以下：

使用pip安装psycopg2和asyncpg。
# 示例命令: pip install psycopg2 和 pip install asyncpg

我们将创建几个Python函数来创建一个“笔记”表并确认它是否已经存在:

import psycopg2  
数据库连接字符串 = "database-connection-string"

def create_notes_table():  
    """  
    如果不存在，创建一个包含 'id', 'title', 和 'text' 字段的 'notes' 表。  
    """  
    create_table_query = """  
    CREATE TABLE IF NOT EXISTS notes (  
        id SERIAL PRIMARY KEY,  
        title VARCHAR(200) UNIQUE NOT NULL,  
        text TEXT NOT NULL  
    );  
    """  
    try:  
        connection = psycopg2.connect(DB_DSN)  
        cursor = connection.cursor()  
        cursor.execute(create_table_query)  
        connection.commit()  
        print("成功创建或验证了 'notes' 表。")  
    except psycopg2.Error as e:  
        print(f"创建 'notes' 表时出错: {e}")  
    finally:  
        if connection:  
            cursor.close()  
            connection.close()  
def check_table_exists(table_name: str) -> bool:  
    """  
    检查指定的表是否存在于数据库中。  
    """  
    query = """  
    SELECT EXISTS (  
        SELECT 1  
        FROM information_schema.tables  
        WHERE table_schema = 'public'  
        AND table_name = %s  
    );  
    """  
    try:  
        connection = psycopg2.connect(DB_DSN)  
        cursor = connection.cursor()  
        cursor.execute(query, (table_name,))  
        exists = cursor.fetchone()[0]  
        return exists  
    except psycopg2.Error as e:  
        print(f"检查 {table_name} 表时出错: {e}")  
        return False  
    finally:  
        if connection:  
            cursor.close()  
            connection.close()

确保你可以运行**_check_table_exists("notes")_**函数，并且得到一个"True"的响应。这说明你的数据库连接正常，并且你已经成功创建了“notes”表。

下面我们要介绍一个异步类来管理笔记的管理操作，比如添加、查找笔记和列出标题。这个类将会被“Agent”用来进行操作。

import asyncpg # 引入异步 PostgreSQL 模块
from typing import Optional, List # 从 typing 导入 Optional 和 List 类型

class DatabaseConn:  
    def __init__(self):  
        """  
        初始化数据库连接，存储数据源名称（DSN）。  
        """  
        self.dsn = DB_DSN  

    async def _connect(self):  
        """  
        建立到PostgreSQL的异步连接。  
        """  
        return await asyncpg.connect(self.dsn)  

    async def add_note(self, title: str, text: str) -> bool:  
        """  
        插入具有给定标题和文本的笔记。如果已有相同标题的笔记，则不进行覆盖。  
        """  
        query = """  
        INSERT INTO notes (title, text)  
        VALUES ($1, $2)  
        ON CONFLICT (title) DO NOTHING;  
        """  
        conn = await self._connect()  
        try:  
            result = await conn.execute(query, title, text)  
            return result == "INSERT 0 1"  
        finally:  
            await conn.close()  

    async def get_note_by_title(self, title: str) -> Optional[dict]:  
        """  
        通过指定标题获取笔记。返回字典，若无匹配项则返回None。  
        """  
        query = "SELECT title, text FROM notes WHERE title = $1;"  
        conn = await self._connect()  
        try:  
            record = await conn.fetchrow(query, title)  
            if record:  
                return {"title": record["title"], "text": record["text"]}  
            return None  
        finally:  
            await conn.close()  

    async def list_all_titles(self) -> List[str]:  
        """  
        列出所有笔记标题。返回字典，若无匹配项则返回None。  
        """  
        query = "SELECT title FROM notes ORDER BY title;"  
        conn = await self._connect()  
        try:  
            results = await conn.fetch(query)  
            return [row["title"] for row in results]  
        finally:  
            await conn.close()

# 将笔记与 PydanticAI 集成

为了把这些组件粘合起来，我们将创建两个不同类型的_代理_：

1. 一个**意图识别代理**——判断用户想创建、列出或检索笔记。
2. 一个**操作执行代理**——实际通过我们的数据库代码来处理笔记。

以下是 **main.py** 示例结构：

from dataclasses import dataclass  # 从 dataclasses 导入 dataclass
from pydantic import BaseModel  # 从 pydantic 导入 BaseModel
from pydantic_ai import Agent, RunContext  # 从 pydantic_ai 导入 Agent 和 RunContext
from typing import Optional, List  # 从 typing 导入 Optional 和 List
from database import DatabaseConn  # 从 database 导入 DatabaseConn
from pydantic_ai.models.openai import OpenAIModel  # 从 pydantic_ai.models.openai 导入 OpenAIModel

OPENAI_API_KEY = "输入你的OpenAI API密钥"@dataclass  
class NoteIntent:  
    action: str  
    title: Optional[str] = None  
    text: Optional[str] = None@dataclass  
class NoteDependencies:  
    db: DatabaseConnclass NoteResponse(BaseModel):  
    message: str  
    note: Optional[dict] = None  
    titles: Optional[List[str]] = None# 1. Agent for parsing the user's intent  
intent_model = OpenAIModel('gpt-4o-mini', api_key=OPENAI_API_KEY)  
intent_agent = Agent(  
    intent_model,  
    result_type=NoteIntent,  
    system_prompt=(  
        "你是一个意图提取助手。理解用户想要做什么（例如，创建、检索、列出）并提取相关的数据，如标题和文本。输出格式必须是包含keys: action, title, text的类似JSON结构。"  
    )  
)# 2. Agent for executing the identified action  
action_model = OpenAIModel('gpt-4o-mini', api_key=OPENAI_API_KEY)  
action_agent = Agent(  
    action_model,  
    deps_type=NoteDependencies,  
    result_type=NoteResponse,  
    system_prompt=(  
        "根据识别的用户意图，执行笔记存储上的请求操作。操作可能包括：'create'（添加笔记）、'retrieve'（获取笔记）或 'list'（列出所有笔记）。"  
    )  
)# Tools for action_agent  
@action_agent.tool  
async def create_note_tool(ctx: RunContext[NoteDependencies], title: str, text: str) -> NoteResponse:  
    db = ctx.deps.db  
    success = await db.add_note(title, text)  
    return NoteResponse(message="创建：成功" if success else "创建：失败")@action_agent.tool  
async def retrieve_note_tool(ctx: RunContext[NoteDependencies], title: str) -> NoteResponse:  
    db = ctx.deps.db  
    note = await db.get_note_by_title(title)  
    return NoteResponse(message="获取：成功", note=note) if note else NoteResponse(message="获取：失败")@action_agent.tool  
async def list_notes_tool(ctx: RunContext[NoteDependencies]) -> NoteResponse:  
    db = ctx.deps.db  
    all_titles = await db.list_all_titles()  
    return NoteResponse(message="列表：成功", titles=all_titles)async def handle_user_query(user_input: str, deps: NoteDependencies) -> NoteResponse:  
    # Determine user intent  
    intent = await intent_agent.run(user_input)  
    print(intent.data)    if intent.data.action == "create":  
        query = f"创建一个名为 '{intent.data.title}' 的笔记，内容为 '{intent.data.text}'。"  
        response = await action_agent.run(query, deps=deps)  
        return response.data  
    elif intent.data.action == "retrieve":  
        query = f"获取标题为 '{intent.data.title}' 的笔记。"  
        response = await action_agent.run(query, deps=deps)  
        return response.data  
    elif intent.data.action == "list":  
        query = "列出所有笔记的标题。"  
        response = await action_agent.run(query, deps=deps)  
        return response.data  
    else:  
        return NoteResponse(message="未识别的操作。")async def ask(query: str):  
    db_conn = DatabaseConn()  
    note_deps = NoteDependencies(db=db_conn)  
    return await handle_user_query(query, note_deps)

这个设置负责主要工作。第一个代理识别用户意图；第二个代理知道该使用哪个工具（创建、检索或列出笔记）。

注释——这个类接收数据库连接，并被添加到第二个代理中，进而**_note_deps = NoteDependencies(db=db_conn)_**，以便它能够执行数据库查询。

# 构建一个Streamlit前端界面（一种使用Streamlit创建用户界面的方法）

最后一步是通过一个简单的 web UI 将所有内容变得可访问。安装 **Streamlit** 超级简单：

在终端中运行：`pip install streamlit`

首先，然后创建一个名为 **app.py** 的文件：  
（此处应去掉“然后”，使句子更加流畅自然。）

最终翻译：

首先创建一个名为 **app.py** 的文件：

import asyncio  
import streamlit as st  
from main import ask  # 来自main.py的ask

st.set_page_config(page_title="Note Manager", layout="centered")
st.title("我的笔记仪表板")
st.write("在下面输入指令来创建、检索或列出笔记。")
user_input = st.text_area("你想做什么？", placeholder="例如，'创建一个关于我周一会议的笔记。'")
if st.button("提交"):
    if not user_input.strip():
        st.error("请输入一些内容。")
    else:
        with st.spinner("处理中，请稍候..."):
            try:
                response = asyncio.run(ask(user_input))
                if response.note is not None:
                    st.success(response.message)
                    st.subheader(f"笔记标题：{response.note.get('title', '')}")
                    st.write(response.note.get('text', '没有内容。'))
                elif response.titles is not None:
                    st.success(response.message)
                    if response.titles:
                        st.subheader("当前标题：")
                        for t in response.titles:
                            st.write(f"- {t}")
                    else:
                        st.info("暂时还没有任何笔记。")
                else:
                    st.info(response.message)
            except Exception as e:
                st.error(f"错误：{e}")

你可以通过下面的方法启动它：

```shell
启动它，例如通过以下命令:
# 具体命令在这里

注意：请将 "# 具体命令在这里" 替换为实际的命令。

运行 Streamlit 应用程序的命令：streamlit run app.py

只需一点努力，我们已利用以下工具创建了一个强大的笔记管理工具。

• PydanticAI 用来解析用户的请求并结构化数据
• PostgreSQL 用于存储笔记内容
• Streamlit 用于提供一个简洁且交互性强的网页界面

更多详情请查看 https://github.com/skolo-online/ai-agents

如果你想更深入地了解，可以观看YouTube视频：
https://www.youtube.com/watch?v=OAbyC1kMj5w。