使用pydanticai和rag掌握AI代理：轻松构建智能系统的5个步骤
- Rifx.Online
- Generative AI , AI Applications , Programming
- 08 Mar, 2025
Generative AI 和 PydanticAI 的崛起
生成式 AI 的兴起改变了我们构建智能系统的方式，而像 PydanticAI 这样的框架使创建生产级 AI 代理变得前所未有的简单。在本教程中，我们将探讨如何使用 PydanticAI 构建具有检索增强生成 (RAG) 功能的 AI 代理。我们将逐步介绍核心概念、特性和一个实践示例，以帮助您入门。
什么是 PydanticAI？
PydanticAI 是一个基于流行的 Pydantic 库构建的 Python 框架。它通过提供以下工具简化了 AI 驱动的应用程序的开发：
- 结构化输出验证：确保 LLM 输出符合预定义的模式。
- 依赖注入：动态地为代理提供外部数据和服务。
- 工具集成：允许代理在运行时调用函数。
- 与模型无关的支持：适用于 OpenAI、Anthropic、Ollama 等。
- 类型安全：利用 Python 类型提示进行可靠的开发。
通过结合这些特性，PydanticAI 使得构建可靠、可扩展且可维护的 AI 系统变得更容易。
为什么将 RAG 与 PydanticAI 一起使用？
检索增强生成 (RAG) 通过集成外部知识源（如数据库或文档存储库）来增强 LLM。这种方法非常适合需要最新或特定领域信息的任务。使用 PydanticAI，您可以通过结合依赖注入、工具和结构化响应来无缝地实现 RAG。
设置您的环境
在深入研究代码之前，请确保您的环境已准备就绪：
-
安装 PydanticAI：
pip install pydantic-ai -
使用 pgvector 设置 PostgreSQL：
RAG 通常需要基于向量的搜索功能。使用
pgvector在 PostgreSQL 中启用此功能：mkdir postgres-data docker run --rm \ -e POSTGRES_PASSWORD=postgres \ -p 54320:5432 \ -v `pwd`/postgres-data:/var/lib/postgresql/data \ pgvector/pgvector:pg17 -
准备依赖项：
确保您拥有用于数据库连接的
asyncpg等工具。
构建 RAG 代理
让我们创建一个 PydanticAI 代理，该代理使用 RAG 根据存储在数据库中的文档来回答问题。
-
定义依赖项
依赖项为代理提供外部数据或服务。对于我们的 RAG 示例，我们将使用 PostgreSQL 连接池。
from dataclasses import dataclass from asyncpg import Pool @dataclass class Deps: pool: Pool -
创建数据库模式
数据库将文档部分存储为嵌入，以进行高效检索。
CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS doc_sections ( id SERIAL PRIMARY KEY, url TEXT NOT NULL UNIQUE, title TEXT NOT NULL, content TEXT NOT NULL, embedding VECTOR(1536) ); -
为 RAG 搜索定义工具
工具是代理在运行时可以调用的函数。以下是我们实现搜索工具的方式：
from pydantic_ai import Agent, RunContext async def search_docs(ctx: RunContext[Deps], query_embedding: list[float]) -> str: """Searches the documentation database using vector similarity.""" rows = await ctx.deps.pool.fetch( """ SELECT url, title, content FROM doc_sections ORDER BY embedding <-> $1 LIMIT 5 """, query_embedding ) return "\n\n".join( f"# {row['title']}\nDocumentation URL: {row['url']}\n\n{row['content']}" for row in rows ) -
构建代理
现在我们将所有内容组合成一个代理。
from pydantic_ai import Agent agent = Agent( model="openai:gpt-4o", deps_type=Deps, system_prompt="You are a helpful assistant answering questions using provided documentation.", ) @agent.tool async def search(ctx: RunContext[Deps], query: str) -> str: """Searches the documentation database.""" query_embedding = await ctx.deps.pool.fetchval("SELECT embedding_function($1)", query) return await search_docs(ctx, query_embedding) async def main(): async with database_connect() as pool: deps = Deps(pool=pool) question = "How do I configure logfire with FastAPI?" result = await agent.run(question, deps=deps) print(result.data)
运行示例
- 使用
pgvector启动您的 PostgreSQL 服务器。 - 使用您的文档的嵌入填充
doc_sections表。 - 运行上面的 Python 脚本与您的启用 RAG 的代理进行交互。
PydanticAI 的主要功能
- 结构化输出验证：确保响应符合使用 Pydantic 模型定义的模式。
- 动态工具集成：LLM 可以在运行时动态调用
search_docs等工具。 - 依赖注入：简化将外部资源（如数据库连接）传递到代理的过程。
使用 PydanticAI 和 RAG 构建一个全功能的聊天应用程序
我们将从之前的教程中获取 PydanticAI 和检索增强生成 (RAG) 的概念，并将它们扩展到一个功能齐全的聊天应用程序中。此应用程序将允许用户与 AI 代理交互，该代理可以从数据库中检索相关信息，就像知识助手一样。我们将受到 PydanticAI 聊天应用程序示例 的启发，但会通过 RAG 功能增强它，使响应更具上下文感知和数据驱动。
应用程序概述
该应用程序将包括：
- 后端：一个由 PydanticAI 驱动的 FastAPI 服务器，用于管理 AI 交互和数据库操作。
- 前端：一个用于用户交互的简单 HTML/JS/TS 界面。
- 数据库：SQLite 与
pgvector，用于存储和检索文档的嵌入。
设置环境
先决条件
- Python 3.9+
- Node.js（可选，用于前端开发）
- PostgreSQL 与
pgvector（用于 RAG 功能）
安装依赖项
pip install pydantic-ai fastapi uvicorn logfire asyncpg sqlite3
如果您使用 PostgreSQL 与 pgvector，请确保已安装并正在运行：
docker run --rm \
-e POSTGRES_PASSWORD=postgres \
-p 54320:5432 \
-v `pwd`/postgres-data:/var/lib/postgresql/data \
pgvector/pgvector:pg17
后端：FastAPI 服务器
对于 RAG，我们将把文档嵌入存储在 PostgreSQL 表中。
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS doc_sections (
id SERIAL PRIMARY KEY,
url TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
content TEXT NOT NULL,
embedding VECTOR(1536)
);
后端代码
这是完整的后端实现，结合了聊天功能和 RAG。
from fastapi import FastAPI, Request, Depends, Form
from fastapi.responses import FileResponse, StreamingResponse
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
from typing import Annotated
import asyncio
import sqlite3
from contextlib import asynccontextmanager
from datetime import datetime, timezone
import json
app = FastAPI()
@dataclass
class Deps:
db: sqlite3.Connection
agent = Agent(
model="openai:gpt-4o",
deps_type=Deps,
system_prompt="You are an AI assistant that retrieves and answers questions based on provided documentation."
)
@asynccontextmanager
async def lifespan(app: FastAPI):
con = sqlite3.connect("chat_app.db")
yield {"db": con}
con.close()
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def index():
return FileResponse("chat_app.html", media_type="text/html")
@app.get("/chat_app.ts")
async def main_ts():
return FileResponse("chat_app.ts", media_type="text/plain")
async def get_db(request: Request):
return request.state.db
@agent.tool
async def search_docs(ctx: RunContext[Deps], query_embedding: list[float]) -> str:
"""Searches the database for relevant documents."""
rows = ctx.deps.db.execute("""
SELECT url, title, content
FROM doc_sections
ORDER BY embedding <-> ?
LIMIT 5;
""", (query_embedding,))
results = "\n\n".join(
f"# {row[1]}\nURL: {row[0]}\n\n{row[2]}" for row in rows.fetchall()
)
return results or "No relevant documents found."
@app.post("/chat/")
async def post_chat(
prompt: Annotated[str, Form()], db=Depends(get_db)
) -> StreamingResponse:
async def stream_messages():
yield json.dumps({
"role": "user",
"timestamp": datetime.now(timezone.utc).isoformat(),
"content": prompt,
}).encode("utf-8") + b"\n"
messages = db.execute("SELECT message_list FROM messages").fetchall()
message_history = [json.loads(row[0]) for row in messages]
async with agent.run_stream(prompt, message_history=message_history) as result:
async for text in result.stream(debounce_by=0.01):
yield json.dumps({
"role": "model",
"timestamp": datetime.now(timezone.utc).isoformat(),
"content": text,
}).encode("utf-8") + b"\n"
db.execute(
"INSERT INTO messages (message_list) VALUES (?)",
(result.new_messages_json(),)
)
db.commit()
return StreamingResponse(stream_messages(), media_type="text/plain")
前端：HTML 和 TypeScript
前端将允许用户与聊天应用程序交互。
HTML 模板
将其保存为 chat_app.html：
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Chat App</title>
</head>
<body>
<main>
<h1>Chat Application</h1>
<div id="conversation"></div>
<form method="post">
<input id="prompt-input" name="prompt" placeholder="Type your question..." />
<button type="submit">Send</button>
</form>
</main>
</body>
<script src="/chat_app.ts" type="module"></script>
</html>
TypeScript 逻辑
将其保存为 chat_app.ts：
const convElement = document.getElementById('conversation')!;
const promptInput = document.getElementById('prompt-input') as HTMLInputElement;
document.querySelector('form')!.addEventListener('submit', async (e) => {
e.preventDefault();
const formData = new FormData(e.target as HTMLFormElement);
const response = await fetch('/chat/', { method: 'POST', body: formData });
const reader = response.body!.getReader();
let textDecoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const message = JSON.parse(textDecoder.decode(value));
const div = document.createElement('div');
div.textContent = `${message.role}: ${message.content}`;
convElement.appendChild(div);
}
});
运行应用程序
1- 启动后端服务器：
uvicorn app:app --reload --port 8000
2- 在浏览器中打开 http://localhost:8000
下一步
要扩展此示例：
- 添加更多工具以进行高级交互（例如，摘要或数据提取）。
- 使用多代理设置进行复杂的工作流程。
- 使用 Pydantic Logfire 等工具集成监控，以进行调试和性能跟踪。
有关更多详细信息，请查看 PydanticAI 的文档 和 RAG 示例。
