用PydanticAI寫AI Agent太簡單：全流程CRUD
AI代理簡介
你可能聽說過“生成式AI”這個術語——但什么是AI代理？本質上，它們是利用高級語言模型來更自主地處理任務的下一代工具。它們能夠解析用戶查詢，挑選相關信息，并以結構化的方式與外部服務交互。
上圖展示了AI如何從基本的生成模型發展到能夠與各種工具協作的復雜AI代理。
PydanticAI簡介
PydanticAI是一個Python代理框架，旨在簡化利用生成式AI開發生產級應用的流程。由Pydantic背后的團隊開發——Pydantic是許多Python庫和框架中不可或缺的驗證層——PydanticAI強調類型安全，并與mypy和pyright等靜態類型檢查器無縫集成。
詳細內容可以閱讀上一篇文章：??全新代理構建器與框架 - PydanticAI??
PydanticAI的關鍵特性包括：
?結構化響應：利用Pydantic驗證和結構化模型輸出，確保跨運行的一致性。?依賴注入系統：提供可選系統，為代理的系統提示、工具和結果驗證器提供數據和服務，增強測試和迭代開發。?流式響應：支持LLM輸出的持續流式傳輸，并進行即時驗證，以獲得快速且準確的結果。
目前PydanticAI處于早期測試階段，其API可能會發生變化，開發團隊歡迎反饋以優化和增強其功能。
環境設置
在安裝pydantic-ai之前，請確認你的Python版本為3.9或更高：
python --version然后，創建并激活虛擬環境，接著安裝pydantic-ai：
virtualenv skoloenv
source skoloenv/bin/activate
pip install pydantic-ai理解PydanticAI的核心
在PydanticAI中，核心工作組件是??Agent??類。通過它，你可以在多種模型上運行查詢。你可以在官方文檔中查看支持的模型完整列表。以下是一個使用OpenAI模型初始化Agent的簡單代碼片段：
from pydantic_ai importAgent
from pydantic_ai.models.openai importOpenAIModel
# 引用你所需的模型
model =OpenAIModel('gpt-4o', api_key='add-your-api-key-here')
agent =Agent(model)
result = agent.run_sync("什么是比特幣？")
print(result.data)我們建議將API密鑰存儲為環境變量：
export OPENAI_API_KEY='your-api-key'以下是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上設置一個PostgreSQL數據庫（注意：你可以使用任何PostgreSQL數據庫）。
你可以使用以下DigitalOcean的推薦鏈接開始：https://m.do.co/c/7d9a2c75356d
一旦你有了數據庫，獲取連接字符串——你將在下一步中使用它。
然后安裝以下內容：
pip install psycopg2
pip install asyncpg我們將創建幾個Python函數來設置一個“notes”表并檢查其是否存在：
import psycopg2
DB_DSN ="database-connection-string"
def create_notes_table():
"""
如果“notes”表不存在，則創建它，包含“id”、“title”和“text”字段。
"""
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.Erroras e:
print(f"創建表時出錯：{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.Erroras e:
print(f"檢查表時出錯：{e}")
returnFalse
finally:
if connection:
cursor.close()
connection.close()確保你可以運行??check_table_exists("notes")??函數并得到“True”響應。這表明你的數據庫連接正常工作，并且“notes”表已成功創建。
接下來，我們將引入一個異步類來管理筆記操作，例如添加筆記、檢索筆記和列出標題。這個類將由“Agent”使用。
import asyncpg
from typing importOptional,List
classDatabaseConn:
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"]}
returnNone
finally:
await conn.close()
async def list_all_titles(self)->List[str]:
"""
獲取并返回所有筆記標題。
"""
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集成
為了將這些組件整合在一起，我們將創建兩個不同的代理：
?意圖提取代理：判斷用戶是想創建、列出還是檢索筆記。?動作處理代理：使用我們的數據庫代碼實際處理數據。
以下是??main.py??的示例結構：
from dataclasses import dataclass
from pydantic importBaseModel
from pydantic_ai importAgent,RunContext
from typing importOptional,List
from database importDatabaseConn
from pydantic_ai.models.openai importOpenAIModel
OPENAI_API_KEY ="enter-your-openai-api-key-here"
@dataclass
classNoteIntent:
action: str
title:Optional[str]=None
text:Optional[str]=None
@dataclass
classNoteDependencies:
db:DatabaseConn
classNoteResponse(BaseModel):
message: str
note:Optional[dict]=None
titles:Optional[List[str]]=None
# 1. 用于解析用戶意圖的代理
intent_model =OpenAIModel('gpt-4o-mini', api_key=OPENAI_API_KEY)
intent_agent =Agent(
intent_model,
result_type=NoteIntent,
system_prompt=(
"你是一個意圖提取助手。理解用戶想要做什么（例如創建、檢索、列出），并提取相關數據，如標題和文本。"
"你的輸出格式必須是JSON-like結構，包含鍵：action、title、text。"
)
)
# 2. 用于執行識別動作的代理
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’（列出所有筆記）。"
)
)
# 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)
returnNoteResponse(message="CREATED:SUCCESS"if success else"CREATED:FAILED")
@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)
returnNoteResponse(message="GET:SUCCESS", note=note)if note elseNoteResponse(message="GET:FAILED")
@action_agent.tool
async def list_notes_tool(ctx:RunContext[NoteDependencies])->NoteResponse:
db = ctx.deps.db
all_titles = await db.list_all_titles()
returnNoteResponse(message="LIST:SUCCESS", titles=all_titles)
async def handle_user_query(user_input: str, deps:NoteDependencies)->NoteResponse:
# 確定用戶意圖
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:
returnNoteResponse(message="無法識別的操作。")
async def ask(query: str):
db_conn =DatabaseConn()
note_deps =NoteDependencies(db=db_conn)
return await handle_user_query(query, note_deps)構建Streamlit前端
最后一步是通過一個簡單的Web界面使所有功能可訪問。安裝Streamlit非常簡單：
pip install streamlit然后創建一個??app.py??文件：
import asyncio
import streamlit as st
from main import ask # 從main.py導入ask函數
st.set_page_config(page_title="筆記管理器", layout="centered")
st.title("我的筆記儀表板")
st.write("在下方輸入指令以創建、檢索或列出筆記。")
user_input = st.text_area("你想做什么？", placeholder="例如，‘創建一篇關于我周一會議的筆記。’")
if st.button("提交"):
ifnot user_input.strip():
st.error("請輸入內容。")
else:
with st.spinner("正在處理..."):
try:
response = asyncio.run(ask(user_input))
if response.note isnotNone:
st.success(response.message)
st.subheader(f"筆記標題：{response.note.get('title', '')}")
st.write(response.note.get('text','未找到內容。'))
elif response.titles isnotNone:
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)
exceptExceptionas e:
st.error(f"錯誤：{e}")然后可以通過以下命令啟動：
streamlit run app.py總結
通過一點努力，我們構建了一個強大的筆記管理工具，使用了以下技術：
?PydanticAI：用于解析用戶請求和結構化數據?PostgreSQL：用于存儲筆記?Streamlit：提供一個流暢、交互式的Web界面
本文轉載自??????PyTorch研習社???，作者：AI研究生
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
